diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index 85c987d20..8c660f70a 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -236,11 +236,20 @@ func processDeleteEvent( glog.V(2).Infof("delete: %+v", resp) dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(resp.Directory, message.OldEntry.Name), remoteStorageMountLocation) if !message.OldEntry.IsDirectory { + current, err := currentEntry(filerSource, resp.Directory, message.OldEntry.Name) + if err != nil { + return err + } + if isRemoteOnly(current) { + glog.V(0).Infof("keep %s: a remote-only entry has taken its path since", remote_storage.FormatLocation(dest)) + return nil + } // The object at the key may have been replaced since the delete: a // rename copies another entry's content onto it, a rewrite uploads // over it. Match the recorded etag before deleting so only the - // object the delete meant goes away. - if oldETag := message.OldEntry.GetRemoteEntry().GetRemoteETag(); oldETag != "" { + // object the delete meant goes away. A changed object with no entry + // at the path is an upload that finished after the delete, and goes. + if oldETag := message.OldEntry.GetRemoteEntry().GetRemoteETag(); oldETag != "" && current != nil { if live, err := client.StatFile(dest); err == nil { if live.RemoteETag != "" && live.RemoteETag != oldETag { glog.V(0).Infof("keep %s: the object changed since the delete", remote_storage.FormatLocation(dest)) @@ -250,14 +259,6 @@ func processDeleteEvent( return err } } - current, err := currentEntry(filerSource, resp.Directory, message.OldEntry.Name) - if err != nil { - return err - } - if isRemoteOnly(current) { - glog.V(0).Infof("keep %s: a remote-only entry has taken its path since", remote_storage.FormatLocation(dest)) - return nil - } } if message.OldEntry.IsDirectory { glog.V(0).Infof("rmdir %s", remote_storage.FormatLocation(dest)) diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index 8267b9a2e..ba1e09122 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -1044,7 +1044,8 @@ func TestDeleteEventAbsentRemoteObjectIsSuccess(t *testing.T) { deleted := proto.Clone(resp).(*filer_pb.SubscribeMetadataResponse) deleted.EventNotification.OldEntry.RemoteEntry = &filer_pb.RemoteEntry{RemoteETag: `"orig"`, RemoteSize: 10} remote := &recordingRemote{objects: map[string][]byte{"/b/vol/state.db.tmp": []byte("copied")}, statETag: `"copy"`} - if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, deleted); err != nil { + filerClient := &stubFilerClient{entry: &filer_pb.Entry{Name: "state.db.tmp", Content: []byte("copied")}} + if err := processDeleteEvent(filerClient, remote, mountedDir, mountLoc, deleted); err != nil { t.Fatalf("err = %v", err) } if len(remote.deletes) != 0 { @@ -1052,6 +1053,20 @@ func TestDeleteEventAbsentRemoteObjectIsSuccess(t *testing.T) { } }) + t.Run("a changed object with no entry at the path still goes", func(t *testing.T) { + // an upload that finished after the delete leaves a newer object that + // no entry describes; keeping it would leave it on the remote forever + deleted := proto.Clone(resp).(*filer_pb.SubscribeMetadataResponse) + deleted.EventNotification.OldEntry.RemoteEntry = &filer_pb.RemoteEntry{RemoteETag: `"orig"`, RemoteSize: 10} + remote := &recordingRemote{objects: map[string][]byte{"/b/vol/state.db.tmp": []byte("late")}, statETag: `"late"`} + if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, deleted); err != nil { + t.Fatalf("err = %v", err) + } + if len(remote.deletes) != 1 { + t.Errorf("deletes = %+v, want one", remote.deletes) + } + }) + t.Run("an unchanged object still goes", func(t *testing.T) { deleted := proto.Clone(resp).(*filer_pb.SubscribeMetadataResponse) deleted.EventNotification.OldEntry.RemoteEntry = &filer_pb.RemoteEntry{RemoteETag: `"orig"`, RemoteSize: 10} diff --git a/weed/server/filer_grpc_server_rename.go b/weed/server/filer_grpc_server_rename.go index af5123d14..418605e77 100644 --- a/weed/server/filer_grpc_server_rename.go +++ b/weed/server/filer_grpc_server_rename.go @@ -82,13 +82,8 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom fs.filer.DeleteChunksNotRecursive(pendingChunkDeletes) } fs.finishRemoteCopies(afterCommitCtx, pendingRemoteCopies) - for i := range metadataEvents { - event := &metadataEvents[i] - for _, pending := range pendingRemoteCopies { - if pending.failed && event.oldEntry.FullPath == pending.oldPath { - event.deleteOldRemote = false - } - } + applyRemoteCopies(metadataEvents, pendingRemoteCopies) + for _, event := range metadataEvents { event.notify(fs.filer, afterCommitCtx, req.Signatures) } @@ -154,13 +149,8 @@ func (fs *FilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest, fs.filer.DeleteChunksNotRecursive(pendingChunkDeletes) } fs.finishRemoteCopies(afterCommitCtx, pendingRemoteCopies) - for i := range metadataEvents { - event := &metadataEvents[i] - for _, pending := range pendingRemoteCopies { - if pending.failed && event.oldEntry.FullPath == pending.oldPath { - event.deleteOldRemote = false - } - } + applyRemoteCopies(metadataEvents, pendingRemoteCopies) + for _, event := range metadataEvents { event.notify(fs.filer, afterCommitCtx, req.Signatures) } @@ -184,6 +174,7 @@ type pendingRemoteCopy struct { oldPath util.FullPath newPath util.FullPath failed bool + copied *filer_pb.RemoteEntry } // finishRemoteCopies runs the deferred remote copies after the commit. A @@ -208,6 +199,30 @@ func (fs *FilerServer) finishRemoteCopies(ctx context.Context, pendings []*pendi if updateErr := fs.filer.UpdateEntry(ctx, entry, entry, false); updateErr != nil { glog.WarningfCtx(ctx, "stamp remote rename copy of %s: %v", pending.newPath, updateErr) pending.failed = true + continue + } + pending.copied = remoteEntry + } +} + +// applyRemoteCopies brings the queued rename events in line with the deferred +// copies. The stamp is written with a plain UpdateEntry, which publishes no +// event, so the rename event carries it to peer filers and subscribers. A +// failed copy keeps the old object, which the entry still describes. +func applyRemoteCopies(events []metadataEvent, pendings []*pendingRemoteCopy) { + for i := range events { + event := &events[i] + for _, pending := range pendings { + if event.oldEntry.FullPath != pending.oldPath { + continue + } + if pending.failed { + event.deleteOldRemote = false + } else if pending.copied != nil && event.newEntry != nil { + newEntry := event.newEntry.ShallowClone() + newEntry.Remote = pending.copied + event.newEntry = newEntry + } } } } diff --git a/weed/server/filer_grpc_server_rename_remote_test.go b/weed/server/filer_grpc_server_rename_remote_test.go index e25d33610..ebbdf4e6a 100644 --- a/weed/server/filer_grpc_server_rename_remote_test.go +++ b/weed/server/filer_grpc_server_rename_remote_test.go @@ -147,6 +147,8 @@ func TestRenameRemoteOnlyEntryOverTargetKeepsTheCopy(t *testing.T) { require.Len(t, events, 2) assert.True(t, filer.IsMetadataOnlyDelete(events[0].notification.OldEntry), "the target's delete event must leave the remote object alone") assert.False(t, filer.IsMetadataOnlyDelete(events[1].notification.OldEntry)) + // the copy lands after the commit; the rename event still carries its stamp + assert.Equal(t, `"copy"`, events[1].notification.NewEntry.RemoteEntry.GetRemoteETag()) dst, err := store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg") require.NoError(t, err) assert.Equal(t, uint64(101), dst.Attr.Inode)