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) + } + }) + } +}