fix: write the new key when a remote-synced file is renamed (#11270)

* refactor: extract update event handling into processUpdateEvent Pull the OldEntry/NewEntry update branch of the remote sync event processor into its own function so the rename skip logic can be exercised by tests with stub clients. No behavior change. * test: reproduce remote sync rename dropping the new key A rename under a remote mount arrives as an update whose NewEntry inherits the source RemoteEntry. shouldSendToRemote returns false for it, so processUpdateEvent skipped the event without writing the new key, while the filer had already deleted the old object. The test runs such an event through processUpdateEvent and expects both a delete of the old key and a write of the new one. Fails before the fix. See #11261. * fix: write the new key when a remote-synced file is renamed A rename under a remote mount arrives as an update whose NewEntry inherits the source RemoteEntry, so shouldSendToRemote returns false (RemoteMtime >= Mtime) and processUpdateEvent skipped the event. That skip is only valid when the destination key is unchanged; a path change always needs a write, and the delete-old/write-new handling below the early return is exactly what a rename needs. Guard the skip with proto.Equal(oldDest, dest) so a rename falls through to it. Fixes #11261. * fix: skip empty upload when renaming a remote-only entry A remote-only entry (no local chunks or content, data lives only on the remote object) carries a positive RemoteSize but nothing for NewFileReader to read. After the previous commit lets a rename fall through to the delete-old/write-new path, such a rename would upload EOF and create a zero-byte object at the new key, then stamp it as synced. Guard the write so a path change on a remote-only entry skips the upload instead of replacing the file with zero bytes. The filer has already deleted the old object, so the data is gone regardless; this avoids leaving a misleading empty object behind. * fix: propagate old-key delete errors except already-deleted When deleting the old key on a rename fails for a non-multipart entry, the error was swallowed and the write proceeded, which could leave both remote keys. Return the error so MetadataProcessor retries the event. The filer deletes the source remote object synchronously during the rename, so the sync delete is redundant and the object may already be gone. GCS reports that as ErrRemoteObjectNotFound (unlike S3/Azure, whose deletes are idempotent), so treat it as a successful deletion and continue to retriedWriteFile rather than pinning the sync offset.
2026-09-15 11:00:51 +02:00 · 2026-09-11 10:47:37 -07:00
parent 5ff49909a0
commit 80dae68dbf
2 changed files with 247 additions and 48 deletions
@@ -213,54 +213,7 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem
 			return client.DeleteFile(dest)
 		}
 		if message.OldEntry != nil && message.NewEntry != nil {
-			if isMultipartUploadFile(message.NewParentPath, message.NewEntry.Name) {
-				return nil
-			}
-			// Skip updates to internal version paths
-			if isVersionedPath(message.NewParentPath, message.NewEntry.Name, message.NewEntry.IsDirectory) {
-				glog.V(2).Infof("skipping update of internal version path: %s/%s", message.NewParentPath, message.NewEntry.Name)
-				return nil
-			}
-			oldDest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(resp.Directory, message.OldEntry.Name), remoteStorageMountLocation)
-			dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(message.NewParentPath, message.NewEntry.Name), remoteStorageMountLocation)
-			if !shouldSendToRemote(message.NewEntry) {
-				glog.V(2).Infof("skipping updating: %+v", resp)
-				return nil
-			}
-			if message.NewEntry.IsDirectory {
-				return client.WriteDirectory(dest, message.NewEntry)
-			}
-			if isMetadataOnlyUpdate(resp.Directory, message) {
-				remoteEntry, err := liveRemoteEntry(option, message.NewParentPath, message.NewEntry)
-				if errors.Is(err, filer_pb.ErrNotFound) {
-					glog.V(2).Infof("skipping updating deleted entry: %+v", resp)
-					return nil
-				}
-				if err != nil {
-					return err
-				}
-				if remoteEntry != nil {
-					glog.V(2).Infof("update meta: %+v", resp)
-					return client.UpdateFileMetadata(dest, message.OldEntry, message.NewEntry)
-				}
-				glog.V(0).Infof("never replicated, uploading %s", remote_storage.FormatLocation(dest))
-			}
-			glog.V(2).Infof("update: %+v", resp)
-			if !proto.Equal(oldDest, dest) {
-				glog.V(0).Infof("delete %s", remote_storage.FormatLocation(oldDest))
-				if err := client.DeleteFile(oldDest); err != nil && isMultipartUploadFile(resp.Directory, message.OldEntry.Name) {
-					return nil
-				}
-			}
-			remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, message.NewEntry, dest)
-			if errors.Is(writeErr, errSuperseded) {
-				glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(dest), writeErr)
-				return nil
-			}
-			if writeErr != nil {
-				return writeErr
-			}
-			return updateLocalEntry(option, message.NewParentPath, message.NewEntry, remoteEntry)
+			return processUpdateEvent(option, filerSource, client, mountedDir, remoteStorageMountLocation, resp)
 		}

 		return nil
@@ -268,6 +221,73 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem
 	return eachEntryFunc, nil
 }

+func processUpdateEvent(
+	filerClient filer_pb.FilerClient,
+	filerSource filer_pb.FilerClient,
+	client remote_storage.RemoteStorageClient,
+	mountedDir string,
+	remoteStorageMountLocation *remote_pb.RemoteStorageLocation,
+	resp *filer_pb.SubscribeMetadataResponse,
+) error {
+	message := resp.EventNotification
+	if isMultipartUploadFile(message.NewParentPath, message.NewEntry.Name) {
+		return nil
+	}
+	if isVersionedPath(message.NewParentPath, message.NewEntry.Name, message.NewEntry.IsDirectory) {
+		glog.V(2).Infof("skipping update of internal version path: %s/%s", message.NewParentPath, message.NewEntry.Name)
+		return nil
+	}
+	oldDest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(resp.Directory, message.OldEntry.Name), remoteStorageMountLocation)
+	dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(message.NewParentPath, message.NewEntry.Name), remoteStorageMountLocation)
+	if proto.Equal(oldDest, dest) && !shouldSendToRemote(message.NewEntry) {
+		glog.V(2).Infof("skipping updating: %+v", resp)
+		return nil
+	}
+	if message.NewEntry.IsDirectory {
+		return client.WriteDirectory(dest, message.NewEntry)
+	}
+	if isMetadataOnlyUpdate(resp.Directory, message) {
+		remoteEntry, err := liveRemoteEntry(filerClient, message.NewParentPath, message.NewEntry)
+		if errors.Is(err, filer_pb.ErrNotFound) {
+			glog.V(2).Infof("skipping updating deleted entry: %+v", resp)
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+		if remoteEntry != nil {
+			glog.V(2).Infof("update meta: %+v", resp)
+			return client.UpdateFileMetadata(dest, message.OldEntry, message.NewEntry)
+		}
+		glog.V(0).Infof("never replicated, uploading %s", remote_storage.FormatLocation(dest))
+	}
+	if !proto.Equal(oldDest, dest) && !filer.HasData(message.NewEntry) && message.NewEntry.IsInRemoteOnly() {
+		glog.V(0).Infof("skip uploading renamed remote-only entry %s: content is only on the deleted remote object", remote_storage.FormatLocation(dest))
+		return nil
+	}
+	glog.V(2).Infof("update: %+v", resp)
+	if !proto.Equal(oldDest, dest) {
+		glog.V(0).Infof("delete %s", remote_storage.FormatLocation(oldDest))
+		if err := client.DeleteFile(oldDest); err != nil {
+			if isMultipartUploadFile(resp.Directory, message.OldEntry.Name) {
+				return nil
+			}
+			if !errors.Is(err, remote_storage.ErrRemoteObjectNotFound) {
+				return err
+			}
+		}
+	}
+	remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, message.NewEntry, dest)
+	if errors.Is(writeErr, errSuperseded) {
+		glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(dest), writeErr)
+		return nil
+	}
+	if writeErr != nil {
+		return writeErr
+	}
+	return updateLocalEntry(filerClient, message.NewParentPath, message.NewEntry, remoteEntry)
+}
+
 // isSuperseded reports whether the filer has moved past the entry an event
 // described: it is deleted, or it no longer references every chunk the event
 // named. Those are the chunks the filer deletes when an entry is updated, so
@@ -16,6 +16,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"google.golang.org/grpc"
+	"google.golang.org/protobuf/proto"
 )

 // TestVersionedFilePathRewrittenForRemote verifies that the fix for
@@ -425,6 +426,10 @@ func (c *stubFilerClient) LookupDirectoryEntry(context.Context, *filer_pb.Lookup
 	return &filer_pb.LookupDirectoryEntryResponse{Entry: c.entry}, nil
 }

+func (c *stubFilerClient) UpdateEntry(context.Context, *filer_pb.UpdateEntryRequest, ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) {
+	return &filer_pb.UpdateEntryResponse{}, nil
+}
+
 func (c *stubFilerClient) WithFilerClient(_ bool, fn func(filer_pb.SeaweedFilerClient) error) error {
 	return fn(c)
 }
@@ -629,3 +634,177 @@ func TestRetriedWriteFileStopsWhenSuperseded(t *testing.T) {
 		}
 	})
 }
+
+type recordingRemote struct {
+	remote_storage.RemoteStorageClient
+	deletes   []*remote_pb.RemoteStorageLocation
+	writes    []*remote_pb.RemoteStorageLocation
+	deleteErr error
+}
+
+func (r *recordingRemote) WriteFile(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry, _ io.Reader) (*filer_pb.RemoteEntry, error) {
+	r.writes = append(r.writes, loc)
+	return &filer_pb.RemoteEntry{StorageName: loc.Name, RemoteETag: "etag", RemoteSize: int64(len(entry.Content)), RemoteMtime: entry.Attributes.GetMtime()}, nil
+}
+
+func (r *recordingRemote) DeleteFile(loc *remote_pb.RemoteStorageLocation) error {
+	r.deletes = append(r.deletes, loc)
+	return r.deleteErr
+}
+
+// TestRenameWithInheritedRemoteEntryWritesNewKey reproduces #11261: a rename
+// arrives as an update whose NewEntry carries the RemoteEntry inherited from
+// the source. shouldSendToRemote returns false for it (RemoteMtime >= Mtime),
+// so the old code skipped the event and never wrote the new key, while the
+// filer had already deleted the old object. A path change must always write.
+func TestRenameWithInheritedRemoteEntryWritesNewKey(t *testing.T) {
+	const mountedDir = "/buckets"
+	mountLoc := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/"}
+
+	replicated := &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: "abc", RemoteSize: 2048, RemoteMtime: 1786096669}
+	oldEntry := &filer_pb.Entry{Name: "probe.bin", Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669}, RemoteEntry: replicated}
+	newEntry := &filer_pb.Entry{
+		Name:        "probe.bin",
+		Content:     []byte("payload"),
+		Attributes:  &filer_pb.FuseAttributes{Mtime: 1786096669},
+		RemoteEntry: replicated,
+	}
+	resp := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/buckets/b/src",
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry:      oldEntry,
+			NewParentPath: "/buckets/b/dst",
+			NewEntry:      newEntry,
+		},
+	}
+
+	if shouldSendToRemote(newEntry) {
+		t.Fatal("precondition: inherited RemoteEntry should make shouldSendToRemote false, the bug's trigger")
+	}
+
+	remote := &recordingRemote{}
+	filerClient := &stubFilerClient{}
+	if err := processUpdateEvent(filerClient, filerClient, remote, mountedDir, mountLoc, resp); err != nil {
+		t.Fatal(err)
+	}
+
+	wantDelete := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/b/src/probe.bin"}
+	if len(remote.deletes) != 1 || !proto.Equal(remote.deletes[0], wantDelete) {
+		t.Errorf("deletes = %+v, want the old key %s deleted", remote.deletes, remote_storage.FormatLocation(wantDelete))
+	}
+	wantWrite := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/b/dst/probe.bin"}
+	if len(remote.writes) != 1 || !proto.Equal(remote.writes[0], wantWrite) {
+		t.Errorf("writes = %+v, want the new key %s written", remote.writes, remote_storage.FormatLocation(wantWrite))
+	}
+}
+
+// TestRenameRemoteOnlyEntrySkipsEmptyUpload guards the edge case from the
+// review of #11261: a remote-only entry (no local chunks, content lives only
+// on the remote object the filer already deleted) must not be re-uploaded,
+// since NewFileReader would supply EOF and create a zero-byte object.
+func TestRenameRemoteOnlyEntrySkipsEmptyUpload(t *testing.T) {
+	const mountedDir = "/buckets"
+	mountLoc := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/"}
+
+	remoteOnly := &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: "abc", RemoteSize: 20971520, RemoteMtime: 1786096669}
+	oldEntry := &filer_pb.Entry{Name: "video.mp4", Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669}, RemoteEntry: remoteOnly}
+	newEntry := &filer_pb.Entry{
+		Name:        "video.mp4",
+		Attributes:  &filer_pb.FuseAttributes{Mtime: 1786096669},
+		RemoteEntry: remoteOnly,
+	}
+	if !newEntry.IsInRemoteOnly() {
+		t.Fatal("precondition: entry should be remote-only")
+	}
+	if filer.HasData(newEntry) {
+		t.Fatal("precondition: remote-only entry should have no local data")
+	}
+
+	resp := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/buckets/b/src",
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry:      oldEntry,
+			NewParentPath: "/buckets/b/dst",
+			NewEntry:      newEntry,
+		},
+	}
+
+	remote := &recordingRemote{}
+	filerClient := &stubFilerClient{}
+	if err := processUpdateEvent(filerClient, filerClient, remote, mountedDir, mountLoc, resp); err != nil {
+		t.Fatal(err)
+	}
+	if len(remote.writes) != 0 {
+		t.Errorf("writes = %+v, want none: a remote-only rename must not upload an empty object", remote.writes)
+	}
+}
+
+// TestRenameDeleteOldKeyFailureReturnsError checks that a failed delete of the
+// old key on a rename is returned so MetadataProcessor retries the event,
+// instead of silently continuing and leaving both keys on the remote.
+func TestRenameDeleteOldKeyFailureReturnsError(t *testing.T) {
+	const mountedDir = "/buckets"
+	mountLoc := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/"}
+
+	newEntry := &filer_pb.Entry{
+		Name:       "probe.bin",
+		Content:    []byte("payload"),
+		Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669},
+	}
+	oldEntry := &filer_pb.Entry{Name: "probe.bin", Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669}}
+	resp := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/buckets/b/src",
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry:      oldEntry,
+			NewParentPath: "/buckets/b/dst",
+			NewEntry:      newEntry,
+		},
+	}
+
+	deleteErr := errors.New("AccessDenied: Access Denied")
+	remote := &recordingRemote{deleteErr: deleteErr}
+	filerClient := &stubFilerClient{}
+	err := processUpdateEvent(filerClient, filerClient, remote, mountedDir, mountLoc, resp)
+	if !errors.Is(err, deleteErr) {
+		t.Errorf("err = %v, want the delete failure returned so the event is retried", err)
+	}
+	if len(remote.writes) != 0 {
+		t.Errorf("writes = %+v, want none: must not write the new key when the old key delete failed", remote.writes)
+	}
+}
+
+// TestRenameDeleteOldKeyNotFoundStillWrites checks that an already-deleted old
+// key (the filer deletes the source object synchronously during rename) does
+// not block the destination write. GCS reports a missing object as
+// ErrRemoteObjectNotFound, unlike S3/Azure whose deletes are idempotent, so
+// treating it as a real error would pin the sync offset and never write the
+// new key.
+func TestRenameDeleteOldKeyNotFoundStillWrites(t *testing.T) {
+	const mountedDir = "/buckets"
+	mountLoc := &remote_pb.RemoteStorageLocation{Name: "gcs", Bucket: "bucket", Path: "/"}
+
+	newEntry := &filer_pb.Entry{
+		Name:       "probe.bin",
+		Content:    []byte("payload"),
+		Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669},
+	}
+	oldEntry := &filer_pb.Entry{Name: "probe.bin", Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669}}
+	resp := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/buckets/b/src",
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry:      oldEntry,
+			NewParentPath: "/buckets/b/dst",
+			NewEntry:      newEntry,
+		},
+	}
+
+	remote := &recordingRemote{deleteErr: remote_storage.ErrRemoteObjectNotFound}
+	filerClient := &stubFilerClient{}
+	if err := processUpdateEvent(filerClient, filerClient, remote, mountedDir, mountLoc, resp); err != nil {
+		t.Fatalf("err = %v, want nil: an already-deleted old key must not block the write", err)
+	}
+	wantWrite := &remote_pb.RemoteStorageLocation{Name: "gcs", Bucket: "bucket", Path: "/b/dst/probe.bin"}
+	if len(remote.writes) != 1 || !proto.Equal(remote.writes[0], wantWrite) {
+		t.Errorf("writes = %+v, want the new key %s written", remote.writes, remote_storage.FormatLocation(wantWrite))
+	}
+}