From 67240f538d46622a6bed063b8d7a674738d2cc2f Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 26 Oct 2026 18:03:12 -0700 Subject: [PATCH] filer.remote.gateway: stop zero-byte uploads of remote-only entries (#12110) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * filer.remote.gateway: skip uploading a dropped local copy on update The bucketed update path uploaded the chunkless entry whenever the update was not metadata-only. remote.uncache, eviction, and remote.meta.sync drop a synced entry's chunks and clear its sync stamp, which reads as dirty when the remote mtime shares the entry's second; the upload then replaced the object that is the entry's only content with zero bytes. Skip the upload the same way the mounted-directory sync does. * filer.remote.gateway: copy a remote-only entry's object on rename A rename deleted the source object and uploaded the chunkless entry, replacing the entry's only content with zero bytes — the same loss remote.uncache caused on plain updates. Complete the rename on the remote the way the mounted-directory sync does: copy the source object over (or take the destination when it already describes the entry), restamp so reads stop resolving through the pending-source marker, and remove the source only while it is still the object the stamp describes. A remote-only entry renamed out of the gateway's coverage keeps its object, and the plain rename's old-key delete only removes the object the event's stamp still describes. * filer.remote.gateway: scope the sync-stamp skip to same-path updates shouldSendToRemote reads the entry's sync stamp, which still covers a rename's unchanged mtime, so a synced or pending-stamped rename event was skipped before its copy or upload could run. Only updates in place may skip on the stamp. * filer.remote.gateway: keep the source a pending rename moved on from A rename event whose pending destination has since been renamed again still names the source a later event reads through its own marker; deleting the old key here loses the only copy. Mirror the remote.sync check: leave the source when the event's entry carries the pending-source marker and no entry holds the destination any more. * filer.remote.gateway: release a rename's pending source once complete A rename whose old entry read through a pending-source marker leaves that earlier source behind: the destination is released by its own key, never the marker's. Mirror releaseRenamedSource — once the renamed entry no longer reads through a marker, delete the pending source guarded by the old entry's stamp, keeping it while a remote-only entry holds its path. * filer.remote.gateway: finalize a complete rename before failing its source A rename source outside the gateway's coverage errored before the destination was checked, and the recorded failure held the event offset. Stat the destination first: when it already holds the content the entry describes, refresh the stamp and leave the unmanaged source untouched; fail only when nothing holds the content. * filer.remote.sync: document the source-of-truth protocol both processors share --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/command/filer_remote_gateway_buckets.go | 214 +++++++++++-- .../filer_remote_gateway_buckets_test.go | 289 ++++++++++++++++++ weed/command/filer_remote_sync_dir.go | 21 ++ 3 files changed, 500 insertions(+), 24 deletions(-) diff --git a/weed/command/filer_remote_gateway_buckets.go b/weed/command/filer_remote_gateway_buckets.go index 050916bea7..e07f70e877 100644 --- a/weed/command/filer_remote_gateway_buckets.go +++ b/weed/command/filer_remote_gateway_buckets.go @@ -73,6 +73,10 @@ func (option *RemoteGatewayOptions) followBucketUpdatesAndUploadToRemote(filerSo return pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, metadataFollowOption, processEventFnWithOffset) } +// makeBucketedEventProcessor is the gateway half of the sync protocol +// documented atop filer_remote_sync_dir.go: the same source-of-truth rules +// applied per bucket, with each entry's object resolved through its own +// bucket's mount. Guards added to processUpdateEvent belong here too. func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *source.FilerSource) (pb.ProcessMetadataFunc, error) { handleCreateBucket := func(entry *filer_pb.Entry) error { @@ -360,16 +364,18 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour oldBucket, oldRemoteStorageMountLocation, oldRemoteStorage, oldOk := option.detectBucketInfo(resp.Directory) newBucket, newRemoteStorageMountLocation, newRemoteStorage, newOk := option.detectBucketInfo(message.NewParentPath) if oldOk && newOk { - if !shouldSendToRemote(message.NewEntry) { - glog.V(2).Infof("skipping updating: %s/%s ts_ns=%d", filer_pb.MetadataEventTargetDirectory(resp), message.GetNewEntry().GetName(), resp.TsNs) - return nil - } client, err := remote_storage.GetRemoteStorage(oldRemoteStorage) if err != nil { return err } if resp.Directory == message.NewParentPath && message.OldEntry.Name == message.NewEntry.Name { - // update the same entry + // update the same entry; the sync-stamp check holds only + // for updates in place: a rename keeps the mtime, so a + // synced or pending-stamped rename must still run + if !shouldSendToRemote(message.NewEntry) { + glog.V(2).Infof("skipping updating: %s/%s ts_ns=%d", filer_pb.MetadataEventTargetDirectory(resp), message.GetNewEntry().GetName(), resp.TsNs) + return nil + } if message.NewEntry.IsDirectory { // update directory property return nil @@ -393,41 +399,76 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour remoteEntry = updated } return updateLocalEntry(option, message.NewParentPath, message.NewEntry, remoteEntry) - } else { - newDest := toRemoteStorageLocation(newBucket, util.NewFullPath(message.NewParentPath, message.NewEntry.Name), newRemoteStorageMountLocation) - remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, message.NewEntry, newDest) - if errors.Is(writeErr, errSuperseded) { - glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(newDest), writeErr) - return nil - } - if writeErr != nil { - return writeErr - } - return updateLocalEntry(option, message.NewParentPath, message.NewEntry, remoteEntry) } + if isDroppedLocalCopy(message.NewEntry) { + // remote.uncache, eviction, or remote.meta.sync dropped + // the local copy: the object at this key is the entry's + // content, and uploading the chunkless entry replaces + // it with zero bytes + glog.V(2).Infof("skipping update of remote-only entry: %s/%s ts_ns=%d", filer_pb.MetadataEventTargetDirectory(resp), message.GetNewEntry().GetName(), resp.TsNs) + return nil + } + newDest := toRemoteStorageLocation(newBucket, util.NewFullPath(message.NewParentPath, message.NewEntry.Name), newRemoteStorageMountLocation) + remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, message.NewEntry, newDest) + if errors.Is(writeErr, errSuperseded) { + glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(newDest), writeErr) + return nil + } + if writeErr != nil { + return writeErr + } + return updateLocalEntry(option, message.NewParentPath, message.NewEntry, remoteEntry) } } // the following is entry rename + var oldDest *remote_pb.RemoteStorageLocation + if oldOk { + oldDest = toRemoteStorageLocation(oldBucket, util.NewFullPath(resp.Directory, message.OldEntry.Name), oldRemoteStorageMountLocation) + } + if newOk && !message.NewEntry.IsDirectory { + // A remote-only entry's content is the object its + // renamed-from path maps to; copying it on the remote + // completes the rename. Deleting the source and uploading + // the chunkless entry would lose the only copy. + current, err := currentEntry(filerSource, message.NewParentPath, message.NewEntry.Name) + if err != nil { + return err + } + if isRemoteOnly(current) { + if err := gatewayRemoteOnlyRename(option, option, resp, current); err != nil { + return err + } + // a pending entry the filer copied on this rename + // leaves the source its marker named + return releaseGatewayRenamedSource(filerSource, option, message) + } + if current == nil && len(message.NewEntry.Extended[filer.ExtRemoteRenameSourceKey]) > 0 { + // the rename's copy was pending and the entry has moved + // on: a later rename carries the source in its own + // marker and finishes or releases it, so this leaves it + glog.V(0).Infof("keep %s: the pending rename moved on", util.NewFullPath(resp.Directory, message.OldEntry.Name)) + return nil + } + } if oldOk { client, err := remote_storage.GetRemoteStorage(oldRemoteStorage) if err != nil { return err } - oldDest := toRemoteStorageLocation(oldBucket, util.NewFullPath(resp.Directory, message.OldEntry.Name), oldRemoteStorageMountLocation) if message.OldEntry.IsDirectory { return client.RemoveDirectory(oldDest) } + if !newOk && isRemoteOnly(message.OldEntry) { + glog.V(0).Infof("keep %s: the renamed entry's only content", remote_storage.FormatLocation(oldDest)) + return nil + } glog.V(0).Infof("delete %s", remote_storage.FormatLocation(oldDest)) - if err := client.DeleteFile(oldDest); err != nil { + if err := deleteRemoteFileIfUnchanged(client, oldDest, message.OldEntry.GetRemoteEntry()); err != nil { return err } } if newOk { - if !shouldSendToRemote(message.NewEntry) { - glog.V(2).Infof("skipping updating: %s/%s ts_ns=%d", filer_pb.MetadataEventTargetDirectory(resp), message.GetNewEntry().GetName(), resp.TsNs) - return nil - } client, err := remote_storage.GetRemoteStorage(newRemoteStorage) if err != nil { return err @@ -439,12 +480,15 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, message.NewEntry, newDest) if errors.Is(writeErr, errSuperseded) { glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(newDest), writeErr) - return nil + return releaseGatewayRenamedSource(filerSource, option, message) } if writeErr != nil { return writeErr } - return updateLocalEntry(option, message.NewParentPath, message.NewEntry, remoteEntry) + if err := updateLocalEntry(option, message.NewParentPath, message.NewEntry, remoteEntry); err != nil { + return err + } + return releaseGatewayRenamedSource(filerSource, option, message) } } @@ -453,6 +497,128 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour return eachEntryFunc, nil } +// gatewayRemoteOnlyRename completes a rename whose content exists only as +// the object the renamed-from path maps to: the chunkless entry has nothing +// to upload. When the destination already holds what the entry's stamp +// describes (an earlier run copied it), only the stamp is refreshed. +// Otherwise the source object is copied over, the entry is stamped with the +// copy so reads stop resolving through the pending-source marker, and the +// source goes only while it is still the object the stamp describes. A source +// outside the gateway's coverage is never touched; the rename then completes +// only out of what the destination already holds. With neither object present +// the content is lost: the event fails and holds the offset for recovery. +func gatewayRemoteOnlyRename(filerClient filer_pb.FilerClient, option *RemoteGatewayOptions, resp *filer_pb.SubscribeMetadataResponse, current *filer_pb.Entry) error { + message := resp.EventNotification + srcPath := util.FullPath(current.Extended[filer.ExtRemoteRenameSourceKey]) + if srcPath == "" { + srcPath = util.NewFullPath(resp.Directory, message.OldEntry.Name) + } + newBucket, newMountLocation, newStorage, ok := option.detectBucketInfo(message.NewParentPath) + if !ok { + return nil + } + client, err := remote_storage.GetRemoteStorage(newStorage) + if err != nil { + return err + } + dest := toRemoteStorageLocation(newBucket, util.NewFullPath(message.NewParentPath, current.Name), newMountLocation) + stamp := message.OldEntry.GetRemoteEntry() + if stamp == nil { + stamp = proto.Clone(current.RemoteEntry).(*filer_pb.RemoteEntry) + } + + // a source outside the gateway's coverage stays untouched; the rename + // may only complete out of what the destination already holds + srcDir, _ := srcPath.DirAndName() + var srcClient remote_storage.RemoteStorageClient + var source renameSource + if srcBucket, srcMountLocation, srcStorage, ok := option.detectBucketInfo(srcDir); ok { + if srcClient, err = remote_storage.GetRemoteStorage(srcStorage); err != nil { + return err + } + source = renameSource{path: srcPath, loc: toRemoteStorageLocation(srcBucket, srcPath, srcMountLocation), stamp: stamp} + } + if existing, statErr := client.StatFile(dest); statErr == nil { + if remote_storage.Describes(current.RemoteEntry, existing) { + glog.V(0).Infof("%s already holds the renamed content", remote_storage.FormatLocation(dest)) + if len(current.Extended[filer.ExtRemoteRenameSourceKey]) > 0 { + if err := updateLocalEntry(filerClient, message.NewParentPath, current, existing); err != nil { + return err + } + } + if srcClient == nil { + return nil + } + return deleteRenameSource(filerClient, srcClient, message.NewParentPath, current.Name, source, stamp) + } + glog.V(0).Infof("%s holds an object the entry does not describe (size %d, want %d); replacing it from %s", remote_storage.FormatLocation(dest), existing.RemoteSize, current.RemoteEntry.RemoteSize, srcPath) + } else if !errors.Is(statErr, remote_storage.ErrRemoteObjectNotFound) { + return statErr + } + if srcClient == nil { + return fmt.Errorf("rename source %s is not under a mounted bucket and %s does not hold the content", srcPath, remote_storage.FormatLocation(dest)) + } + stat, err := srcClient.StatFile(source.loc) + if errors.Is(err, remote_storage.ErrRemoteObjectNotFound) { + return fmt.Errorf("%s: content is on neither %s nor %s", util.NewFullPath(message.NewParentPath, current.Name), remote_storage.FormatLocation(source.loc), remote_storage.FormatLocation(dest)) + } + if err != nil { + return err + } + reader, err := openRemoteObject(srcClient, source.loc, stat.RemoteSize) + if err != nil { + return err + } + defer reader.Close() + glog.V(0).Infof("copy %s -> %s", remote_storage.FormatLocation(source.loc), remote_storage.FormatLocation(dest)) + remoteEntry, err := client.WriteFile(dest, current, reader) + if err != nil { + return err + } + if err := updateLocalEntry(filerClient, message.NewParentPath, current, remoteEntry); err != nil { + return err + } + return deleteRenameSource(filerClient, srcClient, message.NewParentPath, current.Name, source, stamp) +} + +// releaseGatewayRenamedSource releases the pending source a rename's old +// entry read, once the renamed entry no longer reads through a marker: the +// gateway-side releaseRenamedSource, with the source resolved through its +// own bucket's mount. +func releaseGatewayRenamedSource(filerSource filer_pb.FilerClient, option *RemoteGatewayOptions, message *filer_pb.EventNotification) error { + srcPath := util.FullPath(message.OldEntry.GetExtended()[filer.ExtRemoteRenameSourceKey]) + if srcPath == "" { + return nil + } + current, err := currentEntry(filerSource, message.NewParentPath, message.NewEntry.Name) + if err != nil { + return err + } + if current != nil && len(current.Extended[filer.ExtRemoteRenameSourceKey]) > 0 { + return nil + } + srcDir, srcName := srcPath.DirAndName() + srcBucket, srcMountLocation, srcStorage, ok := option.detectBucketInfo(srcDir) + if !ok { + return nil + } + current, err = currentEntry(filerSource, srcDir, srcName) + if err != nil { + return err + } + if isRemoteOnly(current) { + glog.V(0).Infof("keep %s: a remote-only entry holds the pending rename's source", srcPath) + return nil + } + srcClient, err := remote_storage.GetRemoteStorage(srcStorage) + if err != nil { + return err + } + srcLoc := toRemoteStorageLocation(srcBucket, srcPath, srcMountLocation) + glog.V(0).Infof("delete %s, the source of a pending rename", remote_storage.FormatLocation(srcLoc)) + return deleteRemoteFileIfUnchanged(srcClient, srcLoc, message.OldEntry.GetRemoteEntry()) +} + func (option *RemoteGatewayOptions) findRemoteStorageClient(bucketName string) (client remote_storage.RemoteStorageClient, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, err error) { bucket := util.FullPath(option.bucketsDir).Child(bucketName) diff --git a/weed/command/filer_remote_gateway_buckets_test.go b/weed/command/filer_remote_gateway_buckets_test.go index d36a21865a..b6d36decd9 100644 --- a/weed/command/filer_remote_gateway_buckets_test.go +++ b/weed/command/filer_remote_gateway_buckets_test.go @@ -1,12 +1,14 @@ package command import ( + "slices" "testing" "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" "github.com/seaweedfs/seaweedfs/weed/remote_storage" + "github.com/seaweedfs/seaweedfs/weed/replication/source" ) type recordingRemoteMaker struct{ client *recordingRemote } @@ -16,6 +18,293 @@ func (m recordingRemoteMaker) Make(*remote_pb.RemoteConf) (remote_storage.Remote } func (m recordingRemoteMaker) HasBucket() bool { return true } +func gatewayTestOption(storageType string) (*RemoteGatewayOptions, *recordingRemote) { + remote := &recordingRemote{} + remote_storage.RemoteStorageClientMakers[storageType] = recordingRemoteMaker{client: remote} + filerAddress := "localhost:1" + return &RemoteGatewayOptions{ + filerAddress: &filerAddress, + bucketsDir: "/buckets", + mappings: &remote_pb.RemoteStorageMapping{Mappings: map[string]*remote_pb.RemoteStorageLocation{ + "/buckets/gold": {Name: storageType, Bucket: "gold", Path: "/"}, + }}, + remoteConfs: map[string]*remote_pb.RemoteConf{storageType: {Name: storageType, Type: storageType}}, + }, remote +} + +// TestGatewayDroppedLocalCopyIsNotUploaded: remote.uncache drops a synced +// entry's chunks and clears its sync stamp; shouldSendToRemote reads the +// cleared stamp in the remote mtime's second as dirty, so the update reaches +// the upload. The object at the key is the entry's only content and the +// chunkless upload would replace it with zero bytes, so it is skipped. +func TestGatewayDroppedLocalCopyIsNotUploaded(t *testing.T) { + option, remote := gatewayTestOption("gatewaytest-uncache") + t.Cleanup(func() { delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-uncache") }) + process, err := option.makeBucketedEventProcessor(&source.FilerSource{}) + if err != nil { + t.Fatal(err) + } + + resp := droppedCopyUpdate( + droppedCopyEntry(droppedCopySynced, chunk("3,01", "e1")), + droppedCopyEntry(droppedCopyStamp()), + ) + if !shouldSendToRemote(resp.EventNotification.NewEntry) { + t.Fatal("precondition: the cleared stamp reads the entry as dirty") + } + if err := process(resp); err != nil { + t.Fatal(err) + } + if len(remote.writes) != 0 { + t.Errorf("writes = %+v, want none: the remote object is the entry's content and an upload would zero it", remote.writes) + } + if len(remote.deletes) != 0 || len(remote.metadataUpdates) != 0 { + t.Errorf("deletes = %+v, metadata updates = %+v, want the remote object left alone", remote.deletes, remote.metadataUpdates) + } +} + +// TestGatewayUncachedRewriteIsStillUploaded: a rewrite of an uncached entry +// holds local data, so the dropped-copy skip does not apply and the rewrite +// uploads. The stamp cannot reach a filer in the test, so the call errors; +// the upload is what is asserted. +func TestGatewayUncachedRewriteIsStillUploaded(t *testing.T) { + option, remote := gatewayTestOption("gatewaytest-rewrite") + t.Cleanup(func() { delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-rewrite") }) + process, err := option.makeBucketedEventProcessor(&source.FilerSource{}) + if err != nil { + t.Fatal(err) + } + + rewrite := entryWith(droppedCopyName, droppedCopyStamp(), chunk("4,02", "e2")) + resp := droppedCopyUpdate(droppedCopyEntry(droppedCopyStamp()), rewrite) + if err := process(resp); err != nil { + t.Logf("stamp against the test filer fails as expected: %v", err) + } + if len(remote.writes) != 1 { + t.Fatalf("writes = %+v, want the rewrite uploaded once", remote.writes) + } + if !slices.Equal(remote.writtenChunks[0], []string{"4,02"}) { + t.Errorf("uploaded chunks %v, want the rewrite's [4,02]", remote.writtenChunks[0]) + } +} + +// TestGatewayRemoteOnlyRenameCopiesTheObject: a remote-only entry holds its +// content only as the object the renamed-from path maps to. The rename copies +// that object to the destination key, stamps the entry so reads stop going +// through the pending-source marker, and removes the source. Uploading the +// chunkless entry would put a zero-byte object over it, and deleting the +// source first would lose the only copy. +func TestGatewayRemoteOnlyRenameCopiesTheObject(t *testing.T) { + option, remote := gatewayTestOption("gatewaytest-rename") + t.Cleanup(func() { delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-rename") }) + + oldPath := "/buckets/gold/dir/old.txt" + stamp := &filer_pb.RemoteEntry{StorageName: "gatewaytest-rename", RemoteETag: `"abc"`, RemoteSize: 5, RemoteMtime: 100} + oldEntry := &filer_pb.Entry{Name: "old.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 100, FileSize: 5}, RemoteEntry: stamp} + current := &filer_pb.Entry{ + Name: "new.txt", + Attributes: &filer_pb.FuseAttributes{Mtime: 200, FileSize: 5}, + RemoteEntry: stamp, + Extended: map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte(oldPath)}, + } + resp := &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/gold/dir", + EventNotification: &filer_pb.EventNotification{ + OldEntry: oldEntry, NewParentPath: "/buckets/gold/dir", NewEntry: current, + }, + } + filerClient := &stubFilerClient{entries: map[string]*filer_pb.Entry{"/buckets/gold/dir/new.txt": current}} + + remote.objects = map[string][]byte{"/dir/old.txt": []byte("hello")} + remote.statETag = `"abc"` + remote.statMtime = 100 + + if err := gatewayRemoteOnlyRename(filerClient, option, resp, current); err != nil { + t.Fatal(err) + } + if len(remote.writes) != 1 || remote.writes[0].Path != "/dir/new.txt" { + t.Fatalf("writes = %+v, want the source object copied to /dir/new.txt", remote.writes) + } + if string(remote.written[0]) != "hello" { + t.Errorf("copied %q, want the source object's content", remote.written[0]) + } + if filerClient.updates != 1 { + t.Errorf("stamped the entry %d times, want 1 to clear the pending-source marker", filerClient.updates) + } + if len(remote.deletes) != 1 || remote.deletes[0].Path != "/dir/old.txt" { + t.Errorf("deletes = %+v, want the source object removed", remote.deletes) + } +} + +// TestGatewayRemoteOnlyRenameAlreadyComplete: when the destination already +// holds what the stamp describes (an earlier run copied it), the rename only +// refreshes the stamp and removes the source. +func TestGatewayRemoteOnlyRenameAlreadyComplete(t *testing.T) { + option, remote := gatewayTestOption("gatewaytest-rename-complete") + t.Cleanup(func() { delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-rename-complete") }) + + oldPath := "/buckets/gold/dir/old.txt" + stamp := &filer_pb.RemoteEntry{StorageName: "gatewaytest-rename-complete", RemoteETag: `"abc"`, RemoteSize: 5, RemoteMtime: 100} + oldEntry := &filer_pb.Entry{Name: "old.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 100, FileSize: 5}, RemoteEntry: stamp} + current := &filer_pb.Entry{ + Name: "new.txt", + Attributes: &filer_pb.FuseAttributes{Mtime: 200, FileSize: 5}, + RemoteEntry: stamp, + Extended: map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte(oldPath)}, + } + resp := &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/gold/dir", + EventNotification: &filer_pb.EventNotification{ + OldEntry: oldEntry, NewParentPath: "/buckets/gold/dir", NewEntry: current, + }, + } + filerClient := &stubFilerClient{entries: map[string]*filer_pb.Entry{"/buckets/gold/dir/new.txt": current}} + + remote.objects = map[string][]byte{"/dir/old.txt": []byte("hello"), "/dir/new.txt": []byte("hello")} + remote.statETag = `"abc"` + remote.statMtime = 100 + + if err := gatewayRemoteOnlyRename(filerClient, option, resp, current); err != nil { + t.Fatal(err) + } + if len(remote.writes) != 0 { + t.Errorf("writes = %+v, want none: the destination already holds the content", remote.writes) + } + if len(remote.deletes) != 1 || remote.deletes[0].Path != "/dir/old.txt" { + t.Errorf("deletes = %+v, want the source object removed", remote.deletes) + } +} + +// TestGatewayRenameReleasesPendingSource: a rename whose old entry read a +// pending source (an earlier rename's deferred copy had not finished) leaves +// that source behind once the renamed entry no longer reads through a +// marker. The gateway deletes it, guarded by the old entry's stamp. +func TestGatewayRenameReleasesPendingSource(t *testing.T) { + option, remote := gatewayTestOption("gatewaytest-release") + t.Cleanup(func() { delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-release") }) + + stamp := &filer_pb.RemoteEntry{StorageName: "gatewaytest-release", RemoteETag: `"abc"`, RemoteSize: 5, RemoteMtime: 100} + oldEntry := &filer_pb.Entry{ + Name: "mid.txt", + Attributes: &filer_pb.FuseAttributes{Mtime: 100, FileSize: 5}, + RemoteEntry: stamp, + Extended: map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte("/buckets/gold/dir/orig.txt")}, + } + newEntry := &filer_pb.Entry{Name: "new.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 200, FileSize: 5}, RemoteEntry: stamp} + message := &filer_pb.EventNotification{OldEntry: oldEntry, NewParentPath: "/buckets/gold/dir", NewEntry: newEntry} + filerClient := &stubFilerClient{entries: map[string]*filer_pb.Entry{ + "/buckets/gold/dir/new.txt": newEntry, + }} + remote.objects = map[string][]byte{"/dir/orig.txt": []byte("hello")} + remote.statETag = `"abc"` + remote.statMtime = 100 + + if err := releaseGatewayRenamedSource(filerClient, option, message); err != nil { + t.Fatal(err) + } + if len(remote.deletes) != 1 || remote.deletes[0].Path != "/dir/orig.txt" { + t.Errorf("deletes = %+v, want the pending source released", remote.deletes) + } +} + +// TestGatewayRenameKeepsHeldPendingSource: the pending source stays while a +// remote-only entry still holds its path. +func TestGatewayRenameKeepsHeldPendingSource(t *testing.T) { + option, remote := gatewayTestOption("gatewaytest-release") + t.Cleanup(func() { delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-release") }) + + stamp := &filer_pb.RemoteEntry{StorageName: "gatewaytest-release", RemoteETag: `"abc"`, RemoteSize: 5, RemoteMtime: 100} + oldEntry := &filer_pb.Entry{ + Name: "mid.txt", + Attributes: &filer_pb.FuseAttributes{Mtime: 100, FileSize: 5}, + RemoteEntry: stamp, + Extended: map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte("/buckets/gold/dir/orig.txt")}, + } + newEntry := &filer_pb.Entry{Name: "new.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 200, FileSize: 5}, RemoteEntry: stamp} + heldSource := &filer_pb.Entry{Name: "orig.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 300, FileSize: 5}, RemoteEntry: stamp} + message := &filer_pb.EventNotification{OldEntry: oldEntry, NewParentPath: "/buckets/gold/dir", NewEntry: newEntry} + filerClient := &stubFilerClient{entries: map[string]*filer_pb.Entry{ + "/buckets/gold/dir/new.txt": newEntry, + "/buckets/gold/dir/orig.txt": heldSource, + }} + remote.objects = map[string][]byte{"/dir/orig.txt": []byte("hello")} + remote.statETag = `"abc"` + remote.statMtime = 100 + + if err := releaseGatewayRenamedSource(filerClient, option, message); err != nil { + t.Fatal(err) + } + if len(remote.deletes) != 0 { + t.Errorf("deletes = %+v, want the pending source kept while an entry holds its path", remote.deletes) + } +} + +// TestGatewayRemoteOnlyRenameSourceOutOfCoverage: a rename source outside the +// gateway's buckets is left untouched; when the destination already holds +// the content the rename only refreshes the stamp instead of failing. +func TestGatewayRemoteOnlyRenameSourceOutOfCoverage(t *testing.T) { + option, remote := gatewayTestOption("gatewaytest-outsrc") + t.Cleanup(func() { delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-outsrc") }) + + stamp := &filer_pb.RemoteEntry{StorageName: "gatewaytest-outsrc", RemoteETag: `"abc"`, RemoteSize: 5, RemoteMtime: 100} + oldEntry := &filer_pb.Entry{Name: "old.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 100, FileSize: 5}, RemoteEntry: stamp} + current := &filer_pb.Entry{ + Name: "new.txt", + Attributes: &filer_pb.FuseAttributes{Mtime: 200, FileSize: 5}, + RemoteEntry: stamp, + Extended: map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte("/unmanaged/dir/old.txt")}, + } + resp := &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/gold/dir", + EventNotification: &filer_pb.EventNotification{ + OldEntry: oldEntry, NewParentPath: "/buckets/gold/dir", NewEntry: current, + }, + } + filerClient := &stubFilerClient{entries: map[string]*filer_pb.Entry{"/buckets/gold/dir/new.txt": current}} + + remote.objects = map[string][]byte{"/dir/new.txt": []byte("hello")} + remote.statETag = `"abc"` + remote.statMtime = 100 + + if err := gatewayRemoteOnlyRename(filerClient, option, resp, current); err != nil { + t.Fatalf("the completed destination should finalize without touching the out-of-coverage source: %v", err) + } + if filerClient.updates != 1 { + t.Errorf("stamped the entry %d times, want 1 to clear the pending-source marker", filerClient.updates) + } + if len(remote.deletes) != 0 || len(remote.writes) != 0 { + t.Errorf("deletes = %+v, writes = %+v, want the unmanaged source and destination left alone", remote.deletes, remote.writes) + } +} + +// TestGatewayRenameOutKeepsRemoteOnlyObject: a remote-only entry renamed out +// of the gateway's coverage keeps the remote object: it is the entry's only +// content. +func TestGatewayRenameOutKeepsRemoteOnlyObject(t *testing.T) { + option, remote := gatewayTestOption("gatewaytest-out") + t.Cleanup(func() { delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-out") }) + process, err := option.makeBucketedEventProcessor(&source.FilerSource{}) + if err != nil { + t.Fatal(err) + } + + stamp := &filer_pb.RemoteEntry{StorageName: "gatewaytest-out", RemoteETag: `"abc"`, RemoteSize: 5, RemoteMtime: 100} + resp := &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/gold/dir", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "old.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 100, FileSize: 5}, RemoteEntry: stamp}, + NewParentPath: "/outside/dir", + NewEntry: &filer_pb.Entry{Name: "new.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 200, FileSize: 5}, RemoteEntry: stamp}, + }, + } + if err := process(resp); err != nil { + t.Fatal(err) + } + if len(remote.deletes) != 0 { + t.Errorf("deletes = %+v, want the remote-only object kept", remote.deletes) + } +} + func TestGatewayMetadataOnlyDeleteKeepsTheRemoteObject(t *testing.T) { remote := &recordingRemote{} remote_storage.RemoteStorageClientMakers["gatewaytest"] = recordingRemoteMaker{client: remote} diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index 3d4cfdad68..1942950f08 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -26,6 +26,27 @@ import ( "google.golang.org/protobuf/proto" ) +// Which side holds an entry's content, and the two processors that keep the +// remote in step with it. +// +// An entry with local data (chunks or inline content) is authoritative: the +// update uploads it. A chunkless entry stamped with a RemoteEntry is +// remote-only: its content is the remote object its stamp describes, so the +// chunkless entry is never uploaded, since it would write zero bytes over +// the only copy, and the object goes only while the stamp still describes it +// (remote_storage.StillCurrent). While the entry carries a +// filer.ExtRemoteRenameSourceKey marker the content is the object at the +// renamed-from path, until the rename's copy or upload restamps the entry +// and clears it. A dropped local copy (remote.uncache, eviction, +// remote.meta.sync) keeps the remote object's size; a size that moved off +// the stamp (a truncate) is new content and uploads. +// +// processUpdateEvent serves filer remote.sync over one mounted directory; +// makeBucketedEventProcessor in filer_remote_gateway_buckets.go serves filer +// remote.gateway over the bucketed space. They implement the same protocol +// and must change together: a guard added to one and not the other is how +// the zero-byte re-upload on remote.uncache happened. + func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *source.FilerSource, mountedDir string) error { // read filer remote storage mount mappings