diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index d682a7e2e..bdc6bc1ad 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -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 diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index de08100d9..b2f682cc2 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -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)) + } +}