From 728b476ba2c0d7d45a1c9f77cb72f6039bd83c21 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 6 Nov 2026 13:37:08 -0800 Subject: [PATCH] filer.remote.gateway: keep the remote object while a version under a delete marker is locked (#12459) * filer.remote.gateway: keep the remote object while a version under a delete marker is locked Every version of a versioned bucket uploads to the object's key, so on a remote without versioning that object is the only remote copy of the latest version. A delete marker deleted it even when a version still held a legal hold or unexpired retention. Keep it while any version of the key is locked, the same rule the filer uses before pruning a remote entry. Co-Authored-By: Claude Opus 5.5 * filer.remote.gateway, filer.remote.sync: say that earlier versions need remote versioning Co-Authored-By: Claude Opus 5.5 * filer.remote.gateway: delete a kept delete marker's object once the version lock is released A delete marker kept for a locked version stays unstamped. When an update or delete leaves a version that carried a legal hold or retention without an active lock, look up the key's latest version and, if it is still that unstamped marker, run the delete again. Only changes to versions that carried a lock trigger it, so the noncurrent mark and tag updates of unlocked versions do not send repeat remote deletes. Co-Authored-By: Claude Opus 5.5 * filer.remote.gateway: test the kept delete marker retry through the bucket mount The retry reads the key's versions through a filer client parameter, as the rename helpers do, so a test can check the mapped object key and the marker stamp, and that lock-releasing updates and deletes reach it. Co-Authored-By: Claude Opus 5.5 * filer.remote.gateway: check the lock release lookup goes to the gateway's filer Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/command/filer_remote_gateway.go | 5 + weed/command/filer_remote_gateway_buckets.go | 25 +++ .../filer_remote_gateway_buckets_test.go | 92 +++++++++ weed/command/filer_remote_sync.go | 5 + weed/command/filer_remote_sync_dir.go | 64 ++++++ weed/command/filer_remote_sync_dir_test.go | 193 ++++++++++++++++++ 6 files changed, 384 insertions(+) diff --git a/weed/command/filer_remote_gateway.go b/weed/command/filer_remote_gateway.go index 040e5b0555..828248ac91 100644 --- a/weed/command/filer_remote_gateway.go +++ b/weed/command/filer_remote_gateway.go @@ -75,6 +75,11 @@ var cmdFilerRemoteGateway = &Command{ weed filer.remote.gateway -createBucketAt=cloud1 + In a versioned bucket every version is written to the object's key, so the + remote keeps earlier versions only when versioning is enabled on it. A delete + marker over a version with a legal hold or retention keeps the remote object + until a later change to that version releases the lock. + `, } diff --git a/weed/command/filer_remote_gateway_buckets.go b/weed/command/filer_remote_gateway_buckets.go index 7a9453acfb..a7369848ce 100644 --- a/weed/command/filer_remote_gateway_buckets.go +++ b/weed/command/filer_remote_gateway_buckets.go @@ -326,6 +326,9 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour // Skip deletion of internal version files; individual version // deletes should not propagate to the remote object if isVersionedPath(resp.Directory, message.OldEntry.Name, message.OldEntry.IsDirectory) { + if releasesVersionLock(message.OldEntry, nil) { + return syncGatewayKeptDeleteMarker(option, option, resp.Directory, message.OldEntry.Name) + } glog.V(2).Infof("skipping delete of internal version path: %s/%s", resp.Directory, message.OldEntry.Name) return nil } @@ -371,6 +374,9 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour } // Skip updates to internal version paths if isVersionedPath(message.NewParentPath, message.NewEntry.Name, message.NewEntry.IsDirectory) { + if releasesVersionLock(message.OldEntry, message.NewEntry) { + return syncGatewayKeptDeleteMarker(option, option, message.NewParentPath, message.NewEntry.Name) + } glog.V(2).Infof("skipping update of internal version path: %s/%s", message.NewParentPath, message.NewEntry.Name) return nil } @@ -597,6 +603,25 @@ func gatewayRemoteOnlyRename(filerClient filer_pb.FilerClient, option *RemoteGat return deleteRenameSource(filerClient, srcClient, message.NewParentPath, current.Name, source, stamp) } +// syncGatewayKeptDeleteMarker runs syncKeptDeleteMarker against the object +// key that the version file's bucket mount maps to. +func syncGatewayKeptDeleteMarker(filerClient filer_pb.FilerClient, option *RemoteGatewayOptions, versionsDir, versionName string) error { + newParent, newName, ok := rewriteVersionedSourcePath(versionsDir, versionName) + if !ok { + return nil + } + bucket, remoteStorageMountLocation, remoteStorage, ok := option.detectBucketInfo(newParent) + if !ok { + return nil + } + client, err := remote_storage.GetRemoteStorage(remoteStorage) + if err != nil { + return err + } + dest := toRemoteStorageLocation(bucket, util.NewFullPath(newParent, newName), remoteStorageMountLocation) + return syncKeptDeleteMarker(client, filerClient, versionsDir, dest) +} + // 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 diff --git a/weed/command/filer_remote_gateway_buckets_test.go b/weed/command/filer_remote_gateway_buckets_test.go index 32d0b8aed2..2c5bca9d23 100644 --- a/weed/command/filer_remote_gateway_buckets_test.go +++ b/weed/command/filer_remote_gateway_buckets_test.go @@ -6,10 +6,12 @@ import ( "testing" "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb" "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" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" ) type recordingRemoteMaker struct{ client *recordingRemote } @@ -483,3 +485,93 @@ func TestGatewayKeepsTheDefaultStorageAcrossMappingChanges(t *testing.T) { }) } } + +// TestGatewayKeptDeleteMarkerRetry: once a lock is released, the kept marker's +// delete goes to the object key the bucket mount maps the version to, and the +// marker is stamped with the mount's storage. +func TestGatewayKeptDeleteMarkerRetry(t *testing.T) { + const versionsDir = "/buckets/gold/dir/held.bin.versions" + option, remote := gatewayTestOption("gatewaytest-kept-marker") + t.Cleanup(func() { + delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-kept-marker") + remote_storage.ResetRemoteStorageClients() + }) + filerClient := &stubFilerClient{entries: map[string]*filer_pb.Entry{ + versionsDir: {Name: "held.bin.versions", IsDirectory: true, Extended: map[string][]byte{ + s3_constants.ExtLatestVersionFileNameKey: []byte("v_marker"), + s3_constants.ExtLatestVersionIsDeleteMarker: []byte("true"), + }}, + versionsDir + "/v_marker": { + Name: "v_marker", + Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669}, + Extended: map[string][]byte{s3_constants.ExtDeleteMarkerKey: []byte("true")}, + }, + }} + + if err := syncGatewayKeptDeleteMarker(filerClient, option, versionsDir, "v_old"); err != nil { + t.Fatal(err) + } + if len(remote.deletes) != 1 || remote.deletes[0].Bucket != "gold" || remote.deletes[0].Path != "/dir/held.bin" { + t.Errorf("deletes = %+v, want gold:/dir/held.bin", remote.deletes) + } + stamped := filerClient.updated.GetEntry() + if filerClient.updates != 1 || stamped.GetName() != "v_marker" || stamped.GetRemoteEntry().GetStorageName() != "gatewaytest-kept-marker" { + t.Errorf("stamps = %d (%v), want v_marker stamped with the mount's storage", filerClient.updates, stamped) + } + + if err := syncGatewayKeptDeleteMarker(filerClient, option, "/buckets/unmounted/held.bin.versions", "v_old"); err != nil { + t.Fatal(err) + } + if len(remote.deletes) != 1 || filerClient.updates != 1 { + t.Errorf("deletes = %+v, stamps = %d; want nothing more for an unmounted bucket", remote.deletes, filerClient.updates) + } +} + +// TestGatewayLockReleaseLooksUpTheKeptMarker: an update or delete that +// releases a version's lock reads the key's versions from the gateway's filer, +// which the test cannot reach, so the event fails with that filer's address +// instead of being skipped. A change to an unlocked version is skipped +// without a lookup. +func TestGatewayLockReleaseLooksUpTheKeptMarker(t *testing.T) { + const versionsDir = "/buckets/gold/dir/held.bin.versions" + option, remote := gatewayTestOption("gatewaytest-lock-release") + t.Cleanup(func() { + delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-lock-release") + remote_storage.ResetRemoteStorageClients() + }) + filerGrpcAddress := pb.ServerAddress(*option.filerAddress).ToGrpcAddress() + process, err := option.makeBucketedEventProcessor(nil) + if err != nil { + t.Fatal(err) + } + version := func(extended map[string][]byte) *filer_pb.Entry { + return &filer_pb.Entry{Name: "v_old", Attributes: &filer_pb.FuseAttributes{FileSize: 8192}, Extended: extended} + } + held := map[string][]byte{s3_constants.ExtLegalHoldKey: []byte(s3_constants.LegalHoldOn)} + released := map[string][]byte{s3_constants.ExtLegalHoldKey: []byte(s3_constants.LegalHoldOff)} + + for _, tc := range []struct { + name string + event *filer_pb.EventNotification + wantLookup bool + }{ + {"hold released", &filer_pb.EventNotification{OldEntry: version(held), NewParentPath: versionsDir, NewEntry: version(released)}, true}, + {"held version deleted", &filer_pb.EventNotification{OldEntry: version(held)}, true}, + {"unlocked version updated", &filer_pb.EventNotification{OldEntry: version(nil), NewParentPath: versionsDir, NewEntry: version(released)}, false}, + {"unlocked version deleted", &filer_pb.EventNotification{OldEntry: version(nil)}, false}, + } { + t.Run(tc.name, func(t *testing.T) { + err := process(&filer_pb.SubscribeMetadataResponse{Directory: versionsDir, EventNotification: tc.event}) + if tc.wantLookup { + if err == nil || !strings.Contains(err.Error(), filerGrpcAddress) { + t.Errorf("process = %v, want an unanswered lookup on %s", err, filerGrpcAddress) + } + } else if err != nil { + t.Errorf("process = %v, want the change skipped", err) + } + if len(remote.deletes) != 0 || len(remote.writes) != 0 { + t.Errorf("deletes = %+v, writes = %+v; want the remote untouched", remote.deletes, remote.writes) + } + }) + } +} diff --git a/weed/command/filer_remote_sync.go b/weed/command/filer_remote_sync.go index 60e683454d..7ab94ad40d 100644 --- a/weed/command/filer_remote_sync.go +++ b/weed/command/filer_remote_sync.go @@ -65,6 +65,11 @@ var cmdFilerRemoteSynchronize = &Command{ weed filer.remote.sync -dir=/mount/s3_on_cloud + In a versioned bucket every version is written to the object's key, so the + remote keeps earlier versions only when versioning is enabled on it. A delete + marker over a version with a legal hold or retention keeps the remote object + until a later change to that version releases the lock. + The metadata sync starting time is determined with the following priority order: 1. specified by timeAgo 2. last sync timestamp for this directory diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index e7629157b2..8c9e89a5e8 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -12,6 +12,7 @@ import ( "time" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_objectlock" "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" @@ -367,6 +368,10 @@ func processDeleteEvent( // Skip deletion of internal version files; individual version // deletes should not propagate to the remote object if isVersionedPath(resp.Directory, message.OldEntry.Name, message.OldEntry.IsDirectory) { + if newParent, newName, ok := rewriteVersionedSourcePath(resp.Directory, message.OldEntry.Name); ok && releasesVersionLock(message.OldEntry, nil) { + dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(newParent, newName), remoteStorageMountLocation) + return syncKeptDeleteMarker(client, filerSource, resp.Directory, dest) + } glog.V(2).Infof("skipping delete of internal version path: %s/%s", resp.Directory, message.OldEntry.Name) return nil } @@ -481,6 +486,10 @@ func processUpdateEvent( return nil } if isVersionedPath(message.NewParentPath, message.NewEntry.Name, message.NewEntry.IsDirectory) { + if newParent, newName, ok := rewriteVersionedSourcePath(message.NewParentPath, message.NewEntry.Name); ok && releasesVersionLock(message.OldEntry, message.NewEntry) { + dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(newParent, newName), remoteStorageMountLocation) + return syncKeptDeleteMarker(client, filerClient, message.NewParentPath, dest) + } glog.V(2).Infof("skipping update of internal version path: %s/%s", message.NewParentPath, message.NewEntry.Name) return nil } @@ -1369,12 +1378,30 @@ func isDeleteMarker(entry *filer_pb.Entry) bool { // syncDeleteMarker propagates a delete marker to the remote storage and // persists a local sync marker so that replaying the same event is a no-op. +// Every version uploads to the object's key, so while a version under the +// marker holds a legal hold or retention the remote object can be its only +// remote copy: it is kept, and the marker stays unstamped for +// syncKeptDeleteMarker to retry. func syncDeleteMarker( client remote_storage.RemoteStorageClient, filerClient filer_pb.FilerClient, message *filer_pb.EventNotification, dest *remote_pb.RemoteStorageLocation, ) error { + var locked bool + if err := filerClient.WithFilerClient(true, func(c filer_pb.SeaweedFilerClient) error { + ctx, cancel := lookupContext() + defer cancel() + var err error + locked, err = s3_objectlock.HasObjectsWithActiveLocks(ctx, c, message.NewParentPath) + return err + }); err != nil { + return err + } + if locked { + glog.V(0).Infof("keep %s: a version under the delete marker is locked", remote_storage.FormatLocation(dest)) + return nil + } glog.V(0).Infof("delete (marker) %s", remote_storage.FormatLocation(dest)) if err := deleteRemoteFile(client, dest); err != nil { return err @@ -1385,6 +1412,43 @@ func syncDeleteMarker( }) } +// syncKeptDeleteMarker retries the delete of the latest delete marker in +// versionsDir that syncDeleteMarker kept for a locked version. +func syncKeptDeleteMarker( + client remote_storage.RemoteStorageClient, + filerClient filer_pb.FilerClient, + versionsDir string, + dest *remote_pb.RemoteStorageLocation, +) error { + dir, name := util.FullPath(versionsDir).DirAndName() + versionsEntry, err := currentEntry(filerClient, dir, name) + if err != nil || versionsEntry == nil { + return err + } + markerName := string(versionsEntry.Extended[s3_constants.ExtLatestVersionFileNameKey]) + if string(versionsEntry.Extended[s3_constants.ExtLatestVersionIsDeleteMarker]) != "true" || markerName == "" { + return nil + } + marker, err := currentEntry(filerClient, versionsDir, markerName) + if err != nil || !isDeleteMarker(marker) || marker.RemoteEntry != nil { + return err + } + return syncDeleteMarker(client, filerClient, &filer_pb.EventNotification{NewParentPath: versionsDir, NewEntry: marker}, dest) +} + +// releasesVersionLock reports an update or delete of a version that carried a +// legal hold or retention and holds no active lock afterwards. +func releasesVersionLock(oldEntry, newEntry *filer_pb.Entry) bool { + if oldEntry == nil || oldEntry.IsDirectory || isDeleteMarker(oldEntry) { + return false + } + hold := strings.TrimSpace(strings.ToUpper(string(oldEntry.Extended[s3_constants.ExtLegalHoldKey]))) + if hold != s3_constants.LegalHoldOn && len(oldEntry.Extended[s3_constants.ExtObjectLockModeKey]) == 0 { + return false + } + return !s3_objectlock.EntryHasActiveLock(newEntry, time.Now()) +} + // isVersionedPath returns true if the dir/name refers to an internal // versioning path (.versions directory or a version file inside it). // These paths are SeaweedFS-internal and must not be synced to remote diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index c4f17b6a73..a8e0a3f7bb 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -464,6 +464,13 @@ func (c *stubFilerClient) UpdateEntry(_ context.Context, req *filer_pb.UpdateEnt return &filer_pb.UpdateEntryResponse{}, nil } +func (c *stubFilerClient) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) { + if c.err != nil { + return nil, c.err + } + return (&lookupEntrySeaweedClient{entries: c.entries}).ListEntries(ctx, in, opts...) +} + func (c *stubFilerClient) WithFilerClient(_ bool, fn func(filer_pb.SeaweedFilerClient) error) error { return fn(c) } @@ -2965,3 +2972,189 @@ func TestRemoteSyncSkipsSystemFolder(t *testing.T) { t.Fatalf("expected 0 remote deletes for system path, got %d deletes", len(remote.deletes)) } } + +// TestSyncDeleteMarkerKeepsALockedVersion: every version uploads to the +// object's key, so deleting that object for a delete marker can remove the +// only remote copy of a version under legal hold or retention. +func TestSyncDeleteMarkerKeepsALockedVersion(t *testing.T) { + const versionsDir = "/buckets/b/held.bin.versions" + dest := &remote_pb.RemoteStorageLocation{Name: "azure", Bucket: "b", Path: "/held.bin"} + marker := &filer_pb.Entry{ + Name: "v_marker", + Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669}, + Extended: map[string][]byte{s3_constants.ExtDeleteMarkerKey: []byte("true")}, + } + future := []byte(fmt.Sprint(time.Now().Add(time.Hour).Unix())) + past := []byte(fmt.Sprint(time.Now().Add(-time.Hour).Unix())) + for _, tc := range []struct { + name string + lock map[string][]byte + wantDelete bool + }{ + {"legal hold", map[string][]byte{s3_constants.ExtLegalHoldKey: []byte(s3_constants.LegalHoldOn)}, false}, + {"retention", map[string][]byte{s3_constants.ExtObjectLockModeKey: []byte(s3_constants.RetentionModeGovernance), s3_constants.ExtRetentionUntilDateKey: future}, false}, + {"released hold", map[string][]byte{s3_constants.ExtLegalHoldKey: []byte(s3_constants.LegalHoldOff)}, true}, + {"expired retention", map[string][]byte{s3_constants.ExtObjectLockModeKey: []byte(s3_constants.RetentionModeCompliance), s3_constants.ExtRetentionUntilDateKey: past}, true}, + } { + t.Run(tc.name, func(t *testing.T) { + filerClient := &stubFilerClient{entries: map[string]*filer_pb.Entry{ + versionsDir + "/v_old": {Name: "v_old", Attributes: &filer_pb.FuseAttributes{FileSize: 8192}, Extended: tc.lock}, + versionsDir + "/v_new": {Name: "v_new", Attributes: &filer_pb.FuseAttributes{FileSize: 16384}}, + versionsDir + "/v_marker": marker, + }} + remote := &recordingRemote{} + message := &filer_pb.EventNotification{NewParentPath: versionsDir, NewEntry: marker} + if err := syncDeleteMarker(remote, filerClient, message, dest); err != nil { + t.Fatalf("err = %v", err) + } + if deleted := len(remote.deletes) == 1; deleted != tc.wantDelete || len(remote.deletes) > 1 { + t.Errorf("deletes = %+v, want deleted=%v", remote.deletes, tc.wantDelete) + } + if stamped := filerClient.updates == 1; stamped != tc.wantDelete || filerClient.updates > 1 { + t.Errorf("stamped %d times, want the marker stamped only once deleted (a kept one is retried)", filerClient.updates) + } + }) + } + + t.Run("listing failure", func(t *testing.T) { + remote := &recordingRemote{} + filerClient := &stubFilerClient{err: status.Error(codes.Unavailable, "filer is shutting down")} + message := &filer_pb.EventNotification{NewParentPath: versionsDir, NewEntry: marker} + if err := syncDeleteMarker(remote, filerClient, message, dest); err == nil { + t.Fatal("err = nil, want the listing failure") + } + if len(remote.deletes) != 0 || filerClient.updates != 0 { + t.Errorf("deletes = %+v, stamps = %d; want neither before the lock check answers", remote.deletes, filerClient.updates) + } + }) +} + +// TestKeptDeleteMarkerRetriesWhenTheLockIsReleased: a delete marker kept for a +// locked version is deleted once an update or delete of a version that held +// the lock leaves no lock on the key. +func TestKeptDeleteMarkerRetriesWhenTheLockIsReleased(t *testing.T) { + const ( + mountedDir = "/buckets/b" + versionsDir = "/buckets/b/held.bin.versions" + ) + mountLoc := &remote_pb.RemoteStorageLocation{Name: "azure", Bucket: "b", Path: "/"} + held := map[string][]byte{s3_constants.ExtLegalHoldKey: []byte(s3_constants.LegalHoldOn)} + released := map[string][]byte{s3_constants.ExtLegalHoldKey: []byte(s3_constants.LegalHoldOff)} + expired := map[string][]byte{ + s3_constants.ExtObjectLockModeKey: []byte(s3_constants.RetentionModeGovernance), + s3_constants.ExtRetentionUntilDateKey: []byte(fmt.Sprint(time.Now().Add(-time.Hour).Unix())), + } + version := func(name string, extended map[string][]byte) *filer_pb.Entry { + return &filer_pb.Entry{Name: name, Attributes: &filer_pb.FuseAttributes{FileSize: 8192}, Extended: extended} + } + marker := func(stamp *filer_pb.RemoteEntry) *filer_pb.Entry { + return &filer_pb.Entry{ + Name: "v_marker", + Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669}, + Extended: map[string][]byte{s3_constants.ExtDeleteMarkerKey: []byte("true")}, + RemoteEntry: stamp, + } + } + latest := func(isMarker string) *filer_pb.Entry { + return &filer_pb.Entry{Name: "held.bin.versions", IsDirectory: true, Extended: map[string][]byte{ + s3_constants.ExtLatestVersionFileNameKey: []byte("v_marker"), + s3_constants.ExtLatestVersionIsDeleteMarker: []byte(isMarker), + }} + } + update := func(oldEntry, newEntry *filer_pb.Entry) *filer_pb.SubscribeMetadataResponse { + return &filer_pb.SubscribeMetadataResponse{Directory: versionsDir, EventNotification: &filer_pb.EventNotification{ + OldEntry: oldEntry, NewEntry: newEntry, NewParentPath: versionsDir, + }} + } + remove := func(oldEntry *filer_pb.Entry) *filer_pb.SubscribeMetadataResponse { + return &filer_pb.SubscribeMetadataResponse{Directory: versionsDir, EventNotification: &filer_pb.EventNotification{OldEntry: oldEntry}} + } + + for _, tc := range []struct { + name string + entries map[string]*filer_pb.Entry + event *filer_pb.SubscribeMetadataResponse + wantDelete bool + }{ + { + name: "hold released", + entries: map[string]*filer_pb.Entry{ + mountedDir + "/held.bin.versions": latest("true"), + versionsDir + "/v_old": version("v_old", released), + versionsDir + "/v_marker": marker(nil), + }, + event: update(version("v_old", held), version("v_old", released)), + wantDelete: true, + }, + { + name: "expired version deleted", + entries: map[string]*filer_pb.Entry{ + mountedDir + "/held.bin.versions": latest("true"), + versionsDir + "/v_marker": marker(nil), + }, + event: remove(version("v_old", expired)), + wantDelete: true, + }, + { + name: "another version still held", + entries: map[string]*filer_pb.Entry{ + mountedDir + "/held.bin.versions": latest("true"), + versionsDir + "/v_old": version("v_old", released), + versionsDir + "/v_other": version("v_other", held), + versionsDir + "/v_marker": marker(nil), + }, + event: update(version("v_old", held), version("v_old", released)), + }, + { + name: "marker already synced", + entries: map[string]*filer_pb.Entry{ + mountedDir + "/held.bin.versions": latest("true"), + versionsDir + "/v_old": version("v_old", released), + versionsDir + "/v_marker": marker(&filer_pb.RemoteEntry{StorageName: "azure"}), + }, + event: update(version("v_old", held), version("v_old", released)), + }, + { + name: "a newer version is latest", + entries: map[string]*filer_pb.Entry{ + mountedDir + "/held.bin.versions": latest("false"), + versionsDir + "/v_old": version("v_old", released), + versionsDir + "/v_marker": marker(nil), + }, + event: update(version("v_old", held), version("v_old", released)), + }, + { + name: "unlocked version updated", + entries: map[string]*filer_pb.Entry{ + mountedDir + "/held.bin.versions": latest("true"), + versionsDir + "/v_old": version("v_old", nil), + versionsDir + "/v_marker": marker(nil), + }, + event: update(version("v_old", nil), version("v_old", map[string][]byte{s3_constants.ExtNoncurrentSinceNsKey: []byte("1")})), + }, + } { + t.Run(tc.name, func(t *testing.T) { + filerClient := &stubFilerClient{entries: tc.entries} + remote := &recordingRemote{} + var err error + if tc.event.EventNotification.NewEntry != nil { + err = processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, tc.event) + } else { + err = processDeleteEvent(filerClient, remote, mountedDir, mountLoc, tc.event) + } + if err != nil { + t.Fatalf("err = %v", err) + } + if tc.wantDelete { + if len(remote.deletes) != 1 || remote.deletes[0].Path != "/held.bin" { + t.Errorf("deletes = %+v, want /held.bin", remote.deletes) + } + if filerClient.updates != 1 || filerClient.updated.GetEntry().GetName() != "v_marker" { + t.Errorf("stamps = %d (%v), want the marker stamped", filerClient.updates, filerClient.updated) + } + } else if len(remote.deletes) != 0 || filerClient.updates != 0 { + t.Errorf("deletes = %+v, stamps = %d; want neither", remote.deletes, filerClient.updates) + } + }) + } +}