objstorage/objstorageprovider: save custom object name to catalog

Previously, we weren't adding the custom object name to the
catalog. As a result, upon re-opening, various parts of pebble would
treat this external object as a local object.

objstorage/objstorage.go

@@ -141,7 +141,7 @@ func (meta *ObjectMetadata) AssertValid() {
 			panic(errors.AssertionFailedf("meta.Remote not empty: %#v", meta.Remote))
 		}
 	} else {
-		if meta.Remote.CustomObjectName != "" {
+		if meta.Remote.CustomObjectName == "" {
 			if meta.Remote.CreatorID == 0 {
 				panic(errors.AssertionFailedf("CreatorID not set"))
 			}

objstorage/objstorageprovider/provider.go

@@ -464,12 +464,13 @@ func (p *provider) addMetadata(meta objstorage.ObjectMetadata) {
 	p.mu.knownObjects[meta.DiskFileNum] = meta
 	if meta.IsRemote() {
 		p.mu.remote.catalogBatch.AddObject(remoteobjcat.RemoteObjectMetadata{
-			FileNum:        meta.DiskFileNum,
-			FileType:       meta.FileType,
-			CreatorID:      meta.Remote.CreatorID,
-			CreatorFileNum: meta.Remote.CreatorFileNum,
-			Locator:        meta.Remote.Locator,
-			CleanupMethod:  meta.Remote.CleanupMethod,
+			FileNum:          meta.DiskFileNum,
+			FileType:         meta.FileType,
+			CreatorID:        meta.Remote.CreatorID,
+			CreatorFileNum:   meta.Remote.CreatorFileNum,
+			Locator:          meta.Remote.Locator,
+			CleanupMethod:    meta.Remote.CleanupMethod,
+			CustomObjectName: meta.Remote.CustomObjectName,
 		})
 	} else {
 		p.mu.localObjectsChanged = true

objstorage/objstorageprovider/provider_test.go

@@ -593,7 +593,7 @@ func TestParallelSync(t *testing.T) {
 							}
 						}
 					}
-				}(numOps*n, shared)
+				}(numOps*(n+1), shared)
 			}
 			wg.Wait()
 		})

objstorage/objstorageprovider/remote_backing.go

@@ -276,12 +276,13 @@ func (p *provider) AttachRemoteObjects(
 		defer p.mu.Unlock()
 		for _, d := range decoded {
 			p.mu.remote.catalogBatch.AddObject(remoteobjcat.RemoteObjectMetadata{
-				FileNum:        d.meta.DiskFileNum,
-				FileType:       d.meta.FileType,
-				CreatorID:      d.meta.Remote.CreatorID,
-				CreatorFileNum: d.meta.Remote.CreatorFileNum,
-				CleanupMethod:  d.meta.Remote.CleanupMethod,
-				Locator:        d.meta.Remote.Locator,
+				FileNum:          d.meta.DiskFileNum,
+				FileType:         d.meta.FileType,
+				CreatorID:        d.meta.Remote.CreatorID,
+				CreatorFileNum:   d.meta.Remote.CreatorFileNum,
+				CleanupMethod:    d.meta.Remote.CleanupMethod,
+				Locator:          d.meta.Remote.Locator,
+				CustomObjectName: d.meta.Remote.CustomObjectName,
 			})
 		}
 	}()

objstorage/objstorageprovider/remote_backing_test.go

@@ -119,3 +119,40 @@ func TestCreateSharedObjectBacking(t *testing.T) {
 	require.Equal(t, "custom-obj-name", d.meta.Remote.CustomObjectName)
 	require.Equal(t, objstorage.SharedNoCleanup, d.meta.Remote.CleanupMethod)
 }
+
+func TestAttachRemoteObjects(t *testing.T) {
+	st := DefaultSettings(vfs.NewMem(), "")
+	sharedStorage := remote.NewInMem()
+	st.Remote.StorageFactory = remote.MakeSimpleFactory(map[remote.Locator]remote.Storage{
+		"foo": sharedStorage,
+	})
+	p, err := Open(st)
+	require.NoError(t, err)
+	defer p.Close()
+	require.NoError(t, p.SetCreatorID(1))
+	backing, err := p.CreateExternalObjectBacking("foo", "custom-obj-name")
+	require.NoError(t, err)
+	_, err = p.AttachRemoteObjects([]objstorage.RemoteObjectToAttach{{
+		FileType: base.FileTypeTable,
+		FileNum:  base.FileNum(100).DiskFileNum(),
+		Backing:  backing,
+	}})
+	require.NoError(t, err)
+
+	// Sync, close, and reopen the provider and expect that we see
+	// our object.
+	require.NoError(t, p.Sync())
+	require.NoError(t, p.Close())
+
+	p, err = Open(st)
+	require.NoError(t, err)
+	defer p.Close()
+	require.NoError(t, p.SetCreatorID(1))
+	objs := p.List()
+	require.Len(t, objs, 1)
+	o := objs[0]
+	require.Equal(t, remote.Locator("foo"), o.Remote.Locator)
+	require.Equal(t, "custom-obj-name", o.Remote.CustomObjectName)
+	require.Equal(t, uint64(100), uint64(o.DiskFileNum.FileNum()))
+	require.Equal(t, base.FileTypeTable, o.FileType)
+}