diff --git a/weed/command/filer_remote_gateway_buckets.go b/weed/command/filer_remote_gateway_buckets.go index 13f17a21b..7e2254931 100644 --- a/weed/command/filer_remote_gateway_buckets.go +++ b/weed/command/filer_remote_gateway_buckets.go @@ -359,13 +359,21 @@ func (option *RemoteGatewayOptions) makeBucketedEventProcessor(filerSource *sour if message.OldEntry.RemoteEntry != nil && filer.IsSameData(message.OldEntry, message.NewEntry) { glog.V(2).Infof("update meta: %+v", resp) oldDest := toRemoteStorageLocation(oldBucket, util.NewFullPath(resp.Directory, message.OldEntry.Name), oldRemoteStorageMountLocation) - if err := client.UpdateFileMetadata(oldDest, message.OldEntry, message.NewEntry); err != nil { + updated, err := client.UpdateFileMetadata(oldDest, message.OldEntry, message.NewEntry) + if err != nil { return err } remoteEntry := message.NewEntry.RemoteEntry if remoteEntry == nil { remoteEntry = message.OldEntry.RemoteEntry } + // the metadata write can rewrite the object (S3 copies + // it in place for a Content-Encoding change, minting a + // new ETag): restamp from the object the update + // produced when the backend reports it + if updated != nil { + remoteEntry = updated + } return updateLocalEntry(option, message.NewParentPath, message.NewEntry, remoteEntry) } else { newDest := toRemoteStorageLocation(newBucket, util.NewFullPath(message.NewParentPath, message.NewEntry.Name), newRemoteStorageMountLocation) diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index 81b5f916c..8827d2f09 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -242,20 +242,6 @@ func processDeleteEvent( if err := releasePendingSource(filerSource, client, mountedDir, remoteStorageMountLocation, message.OldEntry); err != nil { return err } - // 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, and a copy can land before its entry is visible. Delete - // only an object the event's stamp still describes. - if live, statErr := client.StatFile(dest); statErr == nil { - if oldRemote := message.OldEntry.GetRemoteEntry(); oldRemote == nil || - !describes(oldRemote, live) || - oldRemote.RemoteETag != "" && live.RemoteETag == "" { - glog.V(0).Infof("keep %s: the object changed since the delete", remote_storage.FormatLocation(dest)) - return nil - } - } else if !errors.Is(statErr, remote_storage.ErrRemoteObjectNotFound) { - return statErr - } current, err := currentEntry(filerSource, resp.Directory, message.OldEntry.Name) if err != nil { return err @@ -269,8 +255,12 @@ func processDeleteEvent( glog.V(0).Infof("rmdir %s", remote_storage.FormatLocation(dest)) return client.RemoveDirectory(dest) } + // 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, and a + // copy can land before its entry is visible. Delete only an object the + // event's stamp still describes. glog.V(0).Infof("delete %s", remote_storage.FormatLocation(dest)) - return deleteRemoteFile(client, dest) + return deleteRemoteFileIfUnchanged(client, dest, message.OldEntry.GetRemoteEntry()) } // releasePendingSource deletes the source a pending-rename marker on entry @@ -293,19 +283,8 @@ func releasePendingSource(filerSource filer_pb.FilerClient, client remote_storag glog.V(0).Infof("keep %s: a remote-only entry holds the pending rename's source", remote_storage.FormatLocation(srcDest)) return nil } - live, err := client.StatFile(srcDest) - if errors.Is(err, remote_storage.ErrRemoteObjectNotFound) { - return nil - } - if err != nil { - return err - } - if stamp := entry.GetRemoteEntry(); !describes(stamp, live) || stamp.RemoteETag != "" && live.RemoteETag == "" { - glog.V(0).Infof("keep %s: the pending rename's source changed since", remote_storage.FormatLocation(srcDest)) - return nil - } glog.V(0).Infof("delete %s, the source of a pending rename", remote_storage.FormatLocation(srcDest)) - return deleteRemoteFile(client, srcDest) + return deleteRemoteFileIfUnchanged(client, srcDest, entry.GetRemoteEntry()) } // deleteRemoteFile deletes the object and treats an already-absent object as @@ -317,6 +296,32 @@ func deleteRemoteFile(client remote_storage.RemoteStorageClient, dest *remote_pb return nil } +// deleteRemoteFileIfUnchanged deletes the object only while it is still the +// one stamp describes. Backends offering a conditional delete bind the check +// and the delete into one preconditioned request; the rest fall back to a +// guarded stat-then-delete. +func deleteRemoteFileIfUnchanged(client remote_storage.RemoteStorageClient, dest *remote_pb.RemoteStorageLocation, stamp *filer_pb.RemoteEntry) error { + if deleter, ok := client.(remote_storage.RemoteStorageConditionalDeleter); ok { + deleted, err := deleter.DeleteFileIfUnchanged(dest, stamp) + if err != nil { + return err + } + if !deleted { + glog.V(0).Infof("keep %s: the object changed since the delete", remote_storage.FormatLocation(dest)) + } + return nil + } + live, err := client.StatFile(dest) + if err != nil && !errors.Is(err, remote_storage.ErrRemoteObjectNotFound) { + return err + } + if err == nil && !remote_storage.StillCurrent(stamp, live) { + glog.V(0).Infof("keep %s: the object changed since the delete", remote_storage.FormatLocation(dest)) + return nil + } + return deleteRemoteFile(client, dest) +} + func processUpdateEvent( filerClient filer_pb.FilerClient, filerSource filer_pb.FilerClient, @@ -365,9 +370,16 @@ func processUpdateEvent( } if remoteEntry != nil { glog.V(2).Infof("update meta: %+v", resp) - if err := client.UpdateFileMetadata(dest, message.OldEntry, remoteWriteEntry(message.NewEntry, storageClass)); err != nil { + updated, err := client.UpdateFileMetadata(dest, message.OldEntry, remoteWriteEntry(message.NewEntry, storageClass)) + if err != nil { return err } + // a metadata write can rewrite the object (S3 copies it in place + // for a Content-Encoding change, minting a new ETag): restamp from + // the object the update produced when the backend reports it + if updated != nil { + remoteEntry = updated + } return updateLocalEntry(filerClient, message.NewParentPath, message.NewEntry, remoteEntry) } glog.V(0).Infof("never replicated, uploading %s", remote_storage.FormatLocation(dest)) @@ -492,7 +504,7 @@ func completeRemoteOnlyRename(filerClient filer_pb.FilerClient, client remote_st sourceStamp = proto.Clone(current.RemoteEntry).(*filer_pb.RemoteEntry) } if existing, err := client.StatFile(dest); err == nil { - if describes(current.RemoteEntry, existing) { + 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 { // reads still go to the source until the stamp clears the marker @@ -550,29 +562,10 @@ func deleteRenameSource(filerClient filer_pb.FilerClient, client remote_storage. if current != nil && len(current.Extended[filer.ExtRemoteRenameSourceKey]) > 0 { return fmt.Errorf("%s still reads %s", util.NewFullPath(dir, name), remote_storage.FormatLocation(source.loc)) } - live, err := client.StatFile(source.loc) - if errors.Is(err, remote_storage.ErrRemoteObjectNotFound) { - return nil - } - if err != nil { + if err := deleteRemoteFileIfUnchanged(client, source.loc, stamp); err != nil { return err } - if !describes(stamp, live) || stamp.GetRemoteETag() != "" && live.RemoteETag == "" { - glog.V(0).Infof("keep %s: it is no longer the renamed content", remote_storage.FormatLocation(source.loc)) - return nil - } - return deleteRemoteFile(client, source.loc) -} - -// describes reports whether the object a stat returned is the one the entry's -// stamp describes: same size, and the same ETag when both sides carry one (a -// copy written as one stream can legitimately carry a different ETag from a -// multipart original). -func describes(stamp, object *filer_pb.RemoteEntry) bool { - if stamp == nil || object == nil || stamp.RemoteSize != object.RemoteSize { - return false - } - return stamp.RemoteETag == "" || object.RemoteETag == "" || stamp.RemoteETag == object.RemoteETag + return nil } // openRemoteObject streams the object when the client can, and reads it whole diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index 0a33aeb30..3930df536 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -695,6 +695,10 @@ type recordingRemote struct { // statETags names one for its path. statETag string statETags map[string]string + statMtime int64 + // metadataUpdateResult is the RemoteEntry UpdateFileMetadata reports; nil + // means the update left the object untouched + metadataUpdateResult *filer_pb.RemoteEntry } func (r *recordingRemote) WriteFile(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry, reader io.Reader) (*filer_pb.RemoteEntry, error) { @@ -727,7 +731,7 @@ func (r *recordingRemote) StatFile(loc *remote_pb.RemoteStorageLocation) (*filer if e, ok := r.statETags[loc.Path]; ok { etag = e } - return &filer_pb.RemoteEntry{StorageName: loc.Name, RemoteSize: int64(len(body)), RemoteETag: etag}, nil + return &filer_pb.RemoteEntry{StorageName: loc.Name, RemoteSize: int64(len(body)), RemoteETag: etag, RemoteMtime: r.statMtime}, nil } func (r *recordingRemote) ReadFile(loc *remote_pb.RemoteStorageLocation, offset int64, size int64) ([]byte, error) { @@ -743,9 +747,9 @@ func (r *recordingRemote) DeleteFile(loc *remote_pb.RemoteStorageLocation) error return r.deleteErr } -func (r *recordingRemote) UpdateFileMetadata(loc *remote_pb.RemoteStorageLocation, oldEntry, newEntry *filer_pb.Entry) error { +func (r *recordingRemote) UpdateFileMetadata(loc *remote_pb.RemoteStorageLocation, oldEntry, newEntry *filer_pb.Entry) (*filer_pb.RemoteEntry, error) { r.metadataUpdates = append(r.metadataUpdates, loc) - return nil + return r.metadataUpdateResult, nil } // TestRenameWithInheritedRemoteEntryWritesNewKey reproduces #11261: a rename @@ -1751,3 +1755,64 @@ func TestCompleteRemoteOnlyRenameRetryUsesTheSourceStamp(t *testing.T) { t.Errorf("deletes = %+v, want the unchanged source deleted on the retry", remote.deletes) } } + +// TestMetadataUpdateRestampsTheUpdatedObject: a metadata-only update can +// rewrite the remote object itself (S3 changes Content-Encoding through an +// in-place copy, minting a new ETag and mtime). The stamp saved back to the +// filer must be the one the update produced — a stat instead could fail the +// freshness comparison on the legitimately new ETag and keep the stale +// stamp, so a later delete would keep the object. +func TestMetadataUpdateRestampsTheUpdatedObject(t *testing.T) { + const mountedDir = "/buckets" + mountLoc := &remote_pb.RemoteStorageLocation{Name: "gcs", Bucket: "bucket", Path: "/"} + stamp := &filer_pb.RemoteEntry{StorageName: "gcs", RemoteETag: `"orig"`, RemoteSize: 3, RemoteMtime: 100} + resp := &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/b/vol", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "f", Attributes: &filer_pb.FuseAttributes{Mtime: 100}, RemoteEntry: stamp}, + NewParentPath: "/buckets/b/vol", + NewEntry: &filer_pb.Entry{Name: "f", Attributes: &filer_pb.FuseAttributes{Mtime: 100}, RemoteEntry: stamp, + Extended: map[string][]byte{"Content-Encoding": []byte("gzip")}}, + }, + } + remote := &recordingRemote{ + objects: map[string][]byte{"/b/vol/f": []byte("old")}, + metadataUpdateResult: &filer_pb.RemoteEntry{StorageName: "gcs", RemoteETag: `"new"`, RemoteSize: 3, RemoteMtime: 200}, + } + filerClient := &stubFilerClient{} + if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, resp); err != nil { + t.Fatal(err) + } + if len(remote.metadataUpdates) != 1 { + t.Fatalf("metadataUpdates = %+v, want one", remote.metadataUpdates) + } + got := filerClient.updated.Entry.RemoteEntry + if filerClient.updates != 1 || got.RemoteMtime != 200 || got.RemoteETag != `"new"` { + t.Fatalf("stamped RemoteEntry = %+v, want the update's result (mtime 200, etag \"new\")", got) + } +} + +// TestMetadataUpdateKeepsTheStoredStampOnNilResult: a backend that did not +// rewrite the object reports no stamp; the entry keeps its own. +func TestMetadataUpdateKeepsTheStoredStampOnNilResult(t *testing.T) { + const mountedDir = "/buckets" + mountLoc := &remote_pb.RemoteStorageLocation{Name: "gcs", Bucket: "bucket", Path: "/"} + stamp := &filer_pb.RemoteEntry{StorageName: "gcs", RemoteETag: `"orig"`, RemoteSize: 3, RemoteMtime: 100} + resp := &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/b/vol", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "f", Attributes: &filer_pb.FuseAttributes{Mtime: 100}, RemoteEntry: stamp}, + NewParentPath: "/buckets/b/vol", + NewEntry: &filer_pb.Entry{Name: "f", Attributes: &filer_pb.FuseAttributes{Mtime: 100}, RemoteEntry: stamp, + Extended: map[string][]byte{"Content-Encoding": []byte("gzip")}}, + }, + } + remote := &recordingRemote{objects: map[string][]byte{"/b/vol/f": []byte("old")}} + filerClient := &stubFilerClient{} + if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, resp); err != nil { + t.Fatal(err) + } + if filerClient.updates != 1 || filerClient.updated.Entry.RemoteEntry.RemoteMtime != 100 { + t.Fatalf("stamped RemoteEntry = %+v, want the entry's own stamp kept (mtime 100)", filerClient.updated.Entry.RemoteEntry) + } +} diff --git a/weed/filer/arangodb/arangodb_store_kv.go b/weed/filer/arangodb/arangodb_store_kv.go index ae768c7fb..e9add4644 100644 --- a/weed/filer/arangodb/arangodb_store_kv.go +++ b/weed/filer/arangodb/arangodb_store_kv.go @@ -39,7 +39,7 @@ func (store *ArangodbStore) KvGet(ctx context.Context, key []byte) (value []byte } if err != nil { glog.ErrorfCtx(ctx, "kv get: %s %v", string(key), err) - return nil, filer.ErrKvNotFound + return nil, err } return arrayToBytes(model.Meta), nil } diff --git a/weed/filer/elastic/v7/elastic_store_kv.go b/weed/filer/elastic/v7/elastic_store_kv.go index f662639ff..d6b9c844a 100644 --- a/weed/filer/elastic/v7/elastic_store_kv.go +++ b/weed/filer/elastic/v7/elastic_store_kv.go @@ -37,13 +37,16 @@ func (store *ElasticStore) KvGet(ctx context.Context, key []byte) (value []byte, if elastic.IsNotFound(err) { return value, filer.ErrKvNotFound } + if err != nil { + glog.ErrorfCtx(ctx, "find key(%s),%v.", string(key), err) + return value, err + } if searchResult != nil && searchResult.Found { esEntry := &ESKVEntry{} if err := jsoniter.Unmarshal(searchResult.Source, esEntry); err == nil { return esEntry.Value, nil } } - glog.ErrorfCtx(ctx, "find key(%s),%v.", string(key), err) return value, filer.ErrKvNotFound } diff --git a/weed/filer/filer_delete_entry.go b/weed/filer/filer_delete_entry.go index bec4d7f68..6f7949a61 100644 --- a/weed/filer/filer_delete_entry.go +++ b/weed/filer/filer_delete_entry.go @@ -247,6 +247,9 @@ func (f *Filer) doBatchDeleteFolderMetaAndData(ctx context.Context, entry *Entry if len(sub.HardLinkId) != 0 { // hard link chunk data are deleted separately err = onHardLinkIdsFn([]HardLinkId{sub.HardLinkId}) + if shouldDeleteChunks && f.hardLinkIsOrphaned(ctx, sub.HardLinkId) { + chunksToDelete = append(chunksToDelete, sub.GetChunks()...) + } } else { if shouldDeleteChunks { chunksToDelete = append(chunksToDelete, sub.GetChunks()...) @@ -411,3 +414,10 @@ func (f *Filer) maybeDeleteHardLinks(ctx context.Context, hardLinkIds []HardLink } } } + +// hardLinkIsOrphaned reports whether the shared inode record for hardLinkId +// is gone, meaning the removed link was the last owner of its chunks. +func (f *Filer) hardLinkIsOrphaned(ctx context.Context, hardLinkId HardLinkId) bool { + _, err := f.Store.KvGet(ctx, hardLinkId) + return errors.Is(err, ErrKvNotFound) +} diff --git a/weed/filer/filer_delete_entry_test.go b/weed/filer/filer_delete_entry_test.go index 20834609a..7a049fade 100644 --- a/weed/filer/filer_delete_entry_test.go +++ b/weed/filer/filer_delete_entry_test.go @@ -6,7 +6,10 @@ import ( "fmt" "os" "testing" + "time" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" "github.com/stretchr/testify/require" ) @@ -61,3 +64,56 @@ func TestKeepRemoteObjectMarkerOnlyLivesOnDeleteEvents(t *testing.T) { require.Equal(t, "dir", keep.Last().EventNotification.OldEntry.Name) require.True(t, IsMetadataOnlyDelete(keep.Last().EventNotification.OldEntry)) } + +// A recursive delete that removes the last link to an inode must free the +// shared chunks: the record that owned them is gone, so nothing else will. +func TestDeleteFolderFreesLastHardLinkChunks(t *testing.T) { + store := newStubFilerStore() + f := newTestFiler(t, store, NewFilerRemoteStorage()) + ctx := context.Background() + + hardLinkId := NewHardLinkId() + require.NoError(t, f.CreateEntry(ctx, &Entry{FullPath: "/d", Attr: Attr{Mode: os.ModeDir | 0755}}, nil, false, false, nil, true, 255)) + linkEntry := &Entry{ + FullPath: "/d/a.txt", + Attr: Attr{Mode: 0644, Mtime: time.Now()}, + HardLinkId: hardLinkId, + HardLinkCounter: 1, + Chunks: []*filer_pb.FileChunk{{FileId: "3,01637037d6"}}, + } + require.NoError(t, f.CreateEntry(ctx, linkEntry, nil, false, false, nil, true, 255)) + + require.NoError(t, f.DeleteEntryMetaAndData(ctx, "/d", true, false, true, false, nil, 0)) + + _, err := store.KvGet(ctx, hardLinkId) + require.ErrorIs(t, err, ErrKvNotFound) + require.Equal(t, 1, f.pendingDeletionCount()) +} + +// The same delete keeps the shared chunks while another link outside the +// folder still owns them. +func TestDeleteFolderKeepsSharedHardLinkChunks(t *testing.T) { + store := newStubFilerStore() + f := newTestFiler(t, store, NewFilerRemoteStorage()) + ctx := context.Background() + + hardLinkId := NewHardLinkId() + for _, dir := range []string{"/d", "/keep"} { + require.NoError(t, f.CreateEntry(ctx, &Entry{FullPath: util.FullPath(dir), Attr: Attr{Mode: os.ModeDir | 0755}}, nil, false, false, nil, true, 255)) + } + for _, path := range []string{"/d/a.txt", "/keep/b.txt"} { + require.NoError(t, f.CreateEntry(ctx, &Entry{ + FullPath: util.FullPath(path), + Attr: Attr{Mode: 0644, Mtime: time.Now()}, + HardLinkId: hardLinkId, + HardLinkCounter: 2, + Chunks: []*filer_pb.FileChunk{{FileId: "3,01637037d6"}}, + }, nil, false, false, nil, true, 255)) + } + + require.NoError(t, f.DeleteEntryMetaAndData(ctx, "/d", true, false, true, false, nil, 0)) + + _, err := store.KvGet(ctx, hardLinkId) + require.NoError(t, err, "surviving link must keep the inode record") + require.Equal(t, 0, f.pendingDeletionCount()) +} diff --git a/weed/filer/filer_lazy_remote.go b/weed/filer/filer_lazy_remote.go index 036ac5f5d..81d4e3996 100644 --- a/weed/filer/filer_lazy_remote.go +++ b/weed/filer/filer_lazy_remote.go @@ -202,13 +202,31 @@ func (f *Filer) retractLazyRemoteEntry(ctx context.Context, entry *Entry) { } } -func (f *Filer) maybeDeleteFromRemote(ctx context.Context, entry *Entry) (bool, error) { - if entry == nil || f.RemoteStorage == nil { - return false, nil - } - +// remoteStorageObject resolves the client and object location entry maps to +// on its mounted remote storage; a nil client means the path is not mounted. +func (f *Filer) remoteStorageObject(ctx context.Context, entry *Entry) (remote_storage.RemoteStorageClient, *remote_pb.RemoteStorageLocation, error) { mountDir, remoteLoc := f.RemoteStorage.FindMountDirectory(entry.FullPath) if remoteLoc == nil { + return nil, nil, nil + } + + remoteConf, found := f.RemoteStorage.GetRemoteStorageConf(remoteLoc.Name) + if !found { + return nil, nil, fmt.Errorf("resolve remote storage client for %s: not found", entry.FullPath) + } + client, clientErr := f.buildRemoteStorageClient(ctx, remoteConf) + if clientErr != nil { + return nil, nil, fmt.Errorf("resolve remote storage client for %s: %w", entry.FullPath, clientErr) + } + if client == nil { + return nil, nil, fmt.Errorf("resolve remote storage client for %s: initialization failed", entry.FullPath) + } + + return client, MapFullPathToRemoteStorageLocation(mountDir, remoteLoc, entry.FullPath), nil +} + +func (f *Filer) maybeDeleteFromRemote(ctx context.Context, entry *Entry) (bool, error) { + if entry == nil || f.RemoteStorage == nil { return false, nil } @@ -216,19 +234,10 @@ func (f *Filer) maybeDeleteFromRemote(ctx context.Context, entry *Entry) (bool, return false, nil } - remoteConf, found := f.RemoteStorage.GetRemoteStorageConf(remoteLoc.Name) - if !found { - return false, fmt.Errorf("resolve remote storage client for %s: not found", entry.FullPath) + client, objectLoc, err := f.remoteStorageObject(ctx, entry) + if err != nil || client == nil { + return false, err } - client, clientErr := f.buildRemoteStorageClient(ctx, remoteConf) - if clientErr != nil { - return false, fmt.Errorf("resolve remote storage client for %s: %w", entry.FullPath, clientErr) - } - if client == nil { - return false, fmt.Errorf("resolve remote storage client for %s: initialization failed", entry.FullPath) - } - - objectLoc := MapFullPathToRemoteStorageLocation(mountDir, remoteLoc, entry.FullPath) if entry.IsDirectory() { if err := client.RemoveDirectory(objectLoc); err != nil { @@ -252,6 +261,84 @@ func (f *Filer) maybeDeleteFromRemote(ctx context.Context, entry *Entry) (bool, return true, nil } +// deleteSweptEntryRemote frees the remote objects a swept entry still owns, +// mirroring the write-back daemon's delete guards (processDeleteEvent): the +// source a pending rename marker still reads goes first — nothing else +// releases it — then the object at the entry's own key, each only while it +// is still the object the entry's stamp describes. +func (f *Filer) deleteSweptEntryRemote(ctx context.Context, entry *Entry) error { + if entry == nil || f.RemoteStorage == nil || entry.Remote == nil { + return nil + } + client, objectLoc, err := f.remoteStorageObject(ctx, entry) + if err != nil || client == nil { + return err + } + if err := f.releasePendingRenameSource(ctx, client, entry); err != nil { + return err + } + return deleteRemoteFileIfUnchanged(ctx, client, objectLoc, entry.Remote) +} + +// releasePendingRenameSource deletes the object a pending-rename marker on +// entry still reads: the destination copy never landed, so the source key is +// the content's only location and dropping the entry would orphan it. The +// source survives while a remote-only entry occupies its path — that entry's +// content is the same object. The delete binds to entry's stamp, which is +// the stamp of the source the marker reads. Mirrors the write-back daemon's +// releasePendingSource. +func (f *Filer) releasePendingRenameSource(ctx context.Context, client remote_storage.RemoteStorageClient, entry *Entry) error { + src := f.RemoteSourcePath(entry, entry.FullPath) + if src == entry.FullPath { + return nil + } + srcEntry, err := f.Store.FindEntry(ctx, src) + if err != nil && !errors.Is(err, filer_pb.ErrNotFound) { + return fmt.Errorf("find rename source %s: %w", src, err) + } + if srcEntry != nil && len(srcEntry.Content) == 0 && srcEntry.IsInRemoteOnly() { + glog.V(0).InfofCtx(ctx, "keep remote source %s of pending rename of %s: a remote-only entry holds its path", src, entry.FullPath) + return nil + } + mountDir, remoteLoc := f.RemoteStorage.FindMountDirectory(src) + if remoteLoc == nil { + return nil + } + return deleteRemoteFileIfUnchanged(ctx, client, MapFullPathToRemoteStorageLocation(mountDir, remoteLoc, src), entry.Remote) +} + +// deleteRemoteFileIfUnchanged deletes the remote object only while it is +// still the one stamp describes. Backends offering a conditional delete bind +// the check and the delete into one preconditioned request; the rest fall +// back to a guarded stat-then-delete. +func deleteRemoteFileIfUnchanged(ctx context.Context, client remote_storage.RemoteStorageClient, objectLoc *remote_pb.RemoteStorageLocation, stamp *filer_pb.RemoteEntry) error { + if deleter, ok := client.(remote_storage.RemoteStorageConditionalDeleter); ok { + deleted, delErr := deleter.DeleteFileIfUnchanged(objectLoc, stamp) + if delErr != nil { + return fmt.Errorf("delete remote file %s: %w", remote_storage.FormatLocation(objectLoc), delErr) + } + if !deleted { + glog.V(0).InfofCtx(ctx, "keep %s: the remote object changed since the delete", remote_storage.FormatLocation(objectLoc)) + } + return nil + } + live, statErr := client.StatFile(objectLoc) + if statErr != nil { + if errors.Is(statErr, remote_storage.ErrRemoteObjectNotFound) { + return nil + } + return fmt.Errorf("stat remote file %s: %w", remote_storage.FormatLocation(objectLoc), statErr) + } + if !remote_storage.StillCurrent(stamp, live) { + glog.V(0).InfofCtx(ctx, "keep %s: the remote object changed since the delete", remote_storage.FormatLocation(objectLoc)) + return nil + } + if err := client.DeleteFile(objectLoc); err != nil && !errors.Is(err, remote_storage.ErrRemoteObjectNotFound) { + return fmt.Errorf("delete remote file %s: %w", remote_storage.FormatLocation(objectLoc), err) + } + return nil +} + // DeleteFromRemote deletes the object, or for a directory every object, the // entry maps to on its mounted remote storage. func (f *Filer) DeleteFromRemote(ctx context.Context, entry *Entry) error { diff --git a/weed/filer/filer_lazy_remote_test.go b/weed/filer/filer_lazy_remote_test.go index c56934c6f..f463bb618 100644 --- a/weed/filer/filer_lazy_remote_test.go +++ b/weed/filer/filer_lazy_remote_test.go @@ -259,8 +259,8 @@ func (c *stubRemoteClient) RemoveDirectory(loc *remote_pb.RemoteStorageLocation) func (c *stubRemoteClient) WriteFile(*remote_pb.RemoteStorageLocation, *filer_pb.Entry, io.Reader) (*filer_pb.RemoteEntry, error) { return nil, nil } -func (c *stubRemoteClient) UpdateFileMetadata(*remote_pb.RemoteStorageLocation, *filer_pb.Entry, *filer_pb.Entry) error { - return nil +func (c *stubRemoteClient) UpdateFileMetadata(*remote_pb.RemoteStorageLocation, *filer_pb.Entry, *filer_pb.Entry) (*filer_pb.RemoteEntry, error) { + return nil, nil } func (c *stubRemoteClient) DeleteFile(loc *remote_pb.RemoteStorageLocation) error { c.deleteCalls = append(c.deleteCalls, &remote_pb.RemoteStorageLocation{ @@ -281,6 +281,21 @@ func (c *stubRemoteClient) ListBuckets() ([]*remote_storage.Bucket, error) { ret func (c *stubRemoteClient) CreateBucket(string) error { return nil } func (c *stubRemoteClient) DeleteBucket(string) error { return nil } +// stubConditionalRemoteClient adds the optional conditional-delete interface: +// callers binding stat and delete into one preconditioned request never reach +// StatFile or DeleteFile. +type stubConditionalRemoteClient struct { + stubRemoteClient + conditionalResult bool + conditionalErr error + conditionalCalls []*remote_pb.RemoteStorageLocation +} + +func (c *stubConditionalRemoteClient) DeleteFileIfUnchanged(loc *remote_pb.RemoteStorageLocation, _ *filer_pb.RemoteEntry) (bool, error) { + c.conditionalCalls = append(c.conditionalCalls, loc) + return c.conditionalResult, c.conditionalErr +} + // --- stub RemoteStorageClientMaker --- type stubClientMaker struct { diff --git a/weed/filer/meta_aggregator_replay_test.go b/weed/filer/meta_aggregator_replay_test.go index e41c222c1..3e700a47e 100644 --- a/weed/filer/meta_aggregator_replay_test.go +++ b/weed/filer/meta_aggregator_replay_test.go @@ -12,6 +12,7 @@ import ( "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/stats" "github.com/seaweedfs/seaweedfs/weed/util" ) @@ -751,6 +752,46 @@ func TestReplayDirectoryDeleteSweepsLocalChildren(t *testing.T) { } } +// The sweep locks each child under locks Replay already holds, so it cannot +// wait on a child a local write holds — a rename could hold the child while +// waiting on one of the replayed paths. Contention fails the replay instead, +// and a retry after the local write sweeps normally. +func TestReplaySweepFailsOnChildLockContention(t *testing.T) { + store := newStubFilerStore() + f := newReplayTestFiler(store) + ctx := context.Background() + + if err := store.InsertEntry(ctx, &Entry{FullPath: "/d", Attr: Attr{Mode: os.ModeDir}}); err != nil { + t.Fatal(err) + } + if err := store.InsertEntry(ctx, &Entry{FullPath: "/d/a.txt"}); err != nil { + t.Fatal(err) + } + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "d", IsDirectory: true}, + }, + TsNs: time.Now().UnixNano(), + } + + held := f.EntryLockTable.AcquireLock("test local write", "/d/a.txt", util.ExclusiveLock) + if err := f.Replay(event); err == nil { + f.EntryLockTable.ReleaseLock("/d/a.txt", held) + t.Fatal("replay succeeded while a child lock was held by a local write") + } + f.EntryLockTable.ReleaseLock("/d/a.txt", held) + + if err := f.Replay(event); err != nil { + t.Fatalf("replay after the local write: %v", err) + } + for _, path := range []string{"/d", "/d/a.txt"} { + if _, err := store.FindEntry(ctx, util.FullPath(path)); !errors.Is(err, filer_pb.ErrNotFound) { + t.Fatalf("expected %s gone after the retried delete, got %v", path, err) + } + } +} + func TestReplayFileDeleteKeepsSiblingEntries(t *testing.T) { store := newStubFilerStore() f := newReplayTestFiler(store) @@ -878,3 +919,616 @@ func TestReplayKeptDirectoryDeleteFencesStaleCreates(t *testing.T) { t.Fatal("stale create resurrected a child under a directory the origin deleted") } } + +// A swept hard link drops the shared inode record with the entry. When the +// record is gone the sweep must free the chunks no sibling link can reach. +func TestReplayDirectoryDeleteFreesOrphanedHardLinkChunks(t *testing.T) { + store := newStubFilerStore() + f := newReplayTestFiler(store) + f.FilerConf = NewFilerConf() + f.FileIdDeletionQueue = util.NewUnboundedQueue() + ctx := context.Background() + + hardLinkId := NewHardLinkId() + inode := &Entry{ + FullPath: "/d/a.txt", + Attr: Attr{Mtime: time.Now()}, + HardLinkId: hardLinkId, + HardLinkCounter: 1, + Chunks: []*filer_pb.FileChunk{{FileId: "3,01637037d6"}}, + } + blob, err := inode.EncodeAttributesAndChunks() + if err != nil { + t.Fatal(err) + } + if err := store.KvPut(ctx, hardLinkId, blob); err != nil { + t.Fatal(err) + } + put := func(e *Entry) { + if err := store.InsertEntry(ctx, e); err != nil { + t.Fatal(err) + } + } + put(&Entry{FullPath: "/d", Attr: Attr{Mode: os.ModeDir}}) + put(&Entry{FullPath: "/d/a.txt", Attr: Attr{Mtime: time.Now()}, HardLinkId: hardLinkId, HardLinkCounter: 1}) + + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "d", IsDirectory: true}, + DeleteChunks: true, + }, + TsNs: time.Now().UnixNano(), + } + if err := f.Replay(event); err != nil { + t.Fatalf("replay: %v", err) + } + + if _, err := store.KvGet(ctx, hardLinkId); !errors.Is(err, ErrKvNotFound) { + t.Fatalf("hard link record should be gone, got %v", err) + } + if got := f.pendingDeletionCount(); got != 1 { + t.Fatalf("pending deletions = %d, want the orphaned shared chunks queued", got) + } +} + +// A sibling link outside the swept tree keeps the inode record, so the shared +// chunks stay owned and must not be deleted. +func TestReplayDirectoryDeleteKeepsSharedHardLinkChunks(t *testing.T) { + store := newStubFilerStore() + f := newReplayTestFiler(store) + f.FilerConf = NewFilerConf() + f.FileIdDeletionQueue = util.NewUnboundedQueue() + ctx := context.Background() + + hardLinkId := NewHardLinkId() + inode := &Entry{ + FullPath: "/d/a.txt", + Attr: Attr{Mtime: time.Now()}, + HardLinkId: hardLinkId, + HardLinkCounter: 2, + Chunks: []*filer_pb.FileChunk{{FileId: "3,01637037d6"}}, + } + blob, err := inode.EncodeAttributesAndChunks() + if err != nil { + t.Fatal(err) + } + if err := store.KvPut(ctx, hardLinkId, blob); err != nil { + t.Fatal(err) + } + put := func(e *Entry) { + if err := store.InsertEntry(ctx, e); err != nil { + t.Fatal(err) + } + } + put(&Entry{FullPath: "/d", Attr: Attr{Mode: os.ModeDir}}) + put(&Entry{FullPath: "/d/a.txt", Attr: Attr{Mtime: time.Now()}, HardLinkId: hardLinkId, HardLinkCounter: 2}) + put(&Entry{FullPath: "/keep", Attr: Attr{Mode: os.ModeDir}}) + put(&Entry{FullPath: "/keep/b.txt", Attr: Attr{Mtime: time.Now()}, HardLinkId: hardLinkId, HardLinkCounter: 2}) + + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "d", IsDirectory: true}, + DeleteChunks: true, + }, + TsNs: time.Now().UnixNano(), + } + if err := f.Replay(event); err != nil { + t.Fatalf("replay: %v", err) + } + + if _, err := store.KvGet(ctx, hardLinkId); err != nil { + t.Fatalf("surviving link must keep the inode record: %v", err) + } + if got := f.pendingDeletionCount(); got != 0 { + t.Fatalf("pending deletions = %d, want 0 while a sibling link owns the chunks", got) + } +} + +// A replayed directory delete must mirror the origin's remote cleanup: entries +// the lagging peer alone still holds lose their remote objects too, and record +// tombstones so lazy remote fetches cannot resurrect them. +func TestReplayDirectoryDeleteSweepsRemoteObjects(t *testing.T) { + const storageType = "stub_replay_remote" + stub := &stubRemoteClient{ + statResult: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 10}, + } + defer registerStubMaker(t, storageType, stub)() + + conf := &remote_pb.RemoteConf{Name: "replaystore", Type: storageType} + rs := NewFilerRemoteStorage() + rs.storageNameToConf[conf.Name] = conf + rs.mapDirectoryToRemoteStorage("/buckets/mybucket", &remote_pb.RemoteStorageLocation{ + Name: "replaystore", + Bucket: "mybucket", + Path: "/", + }) + + store := newStubFilerStore() + f := newTestFiler(t, store, rs) + ctx := context.Background() + + put := func(e *Entry) { + if err := store.InsertEntry(ctx, e); err != nil { + t.Fatal(err) + } + } + for _, dir := range []string{"/buckets", "/buckets/mybucket", "/buckets/mybucket/d"} { + put(&Entry{FullPath: util.FullPath(dir), Attr: Attr{Mode: os.ModeDir}}) + } + put(&Entry{ + FullPath: "/buckets/mybucket/d/f.txt", + Attr: Attr{Mtime: time.Now(), Mode: 0644}, + Remote: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 10}, + }) + + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/mybucket", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "d", IsDirectory: true}, + DeleteChunks: false, + }, + TsNs: time.Now().UnixNano(), + } + if err := f.Replay(event); err != nil { + t.Fatalf("replay: %v", err) + } + f.onMetadataChangeEvent(event) + + if _, err := store.FindEntry(ctx, "/buckets/mybucket/d/f.txt"); !errors.Is(err, filer_pb.ErrNotFound) { + t.Fatalf("swept child should be gone, got %v", err) + } + if len(stub.deleteCalls) != 1 || stub.deleteCalls[0].Path != "/d/f.txt" { + t.Fatalf("expected one remote delete of /d/f.txt, got %v", stub.deleteCalls) + } + if got := f.remoteTombstones.dirs["/buckets/mybucket/d"]; got != event.TsNs { + t.Fatalf("dir tombstone stamped %d, want the event ts %d", got, event.TsNs) + } +} + +// A metadata-only delete keeps remote objects: the event's marker means the +// remote tree stays authoritative, so the sweep must not touch it. +func TestReplayDirectoryDeleteKeepRemoteObjects(t *testing.T) { + const storageType = "stub_replay_remote_keep" + stub := &stubRemoteClient{} + defer registerStubMaker(t, storageType, stub)() + + conf := &remote_pb.RemoteConf{Name: "keepstore", Type: storageType} + rs := NewFilerRemoteStorage() + rs.storageNameToConf[conf.Name] = conf + rs.mapDirectoryToRemoteStorage("/buckets/mybucket", &remote_pb.RemoteStorageLocation{ + Name: "keepstore", + Bucket: "mybucket", + Path: "/", + }) + + store := newStubFilerStore() + f := newTestFiler(t, store, rs) + ctx := context.Background() + + put := func(e *Entry) { + if err := store.InsertEntry(ctx, e); err != nil { + t.Fatal(err) + } + } + for _, dir := range []string{"/buckets", "/buckets/mybucket", "/buckets/mybucket/d"} { + put(&Entry{FullPath: util.FullPath(dir), Attr: Attr{Mode: os.ModeDir}}) + } + put(&Entry{ + FullPath: "/buckets/mybucket/d/f.txt", + Attr: Attr{Mtime: time.Now(), Mode: 0644}, + Remote: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 10}, + }) + + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/mybucket", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{ + Name: "d", + IsDirectory: true, + Extended: map[string][]byte{ExtKeepRemoteObjectKey: []byte("true")}, + }, + }, + TsNs: time.Now().UnixNano(), + } + if err := f.Replay(event); err != nil { + t.Fatalf("replay: %v", err) + } + f.onMetadataChangeEvent(event) + + if len(stub.deleteCalls) != 0 { + t.Fatalf("metadata-only delete must not touch remote objects, got %v", stub.deleteCalls) + } + if len(f.remoteTombstones.files)+len(f.remoteTombstones.dirs) != 0 { + t.Fatalf("metadata-only delete must not record tombstones, got %v", f.remoteTombstones.dirs) + } +} + +// A delayed sweep must not erase an object another writer replaced since the +// entry's stamp was recorded — the same guard the write-back daemon applies. +func TestReplayDirectoryDeleteKeepsReplacedRemoteObject(t *testing.T) { + const storageType = "stub_replay_remote_replaced" + stub := &stubRemoteClient{ + statResult: &filer_pb.RemoteEntry{RemoteMtime: 1700000099, RemoteSize: 10}, + } + defer registerStubMaker(t, storageType, stub)() + + conf := &remote_pb.RemoteConf{Name: "replacestore", Type: storageType} + rs := NewFilerRemoteStorage() + rs.storageNameToConf[conf.Name] = conf + rs.mapDirectoryToRemoteStorage("/buckets/mybucket", &remote_pb.RemoteStorageLocation{ + Name: "replacestore", + Bucket: "mybucket", + Path: "/", + }) + + store := newStubFilerStore() + f := newTestFiler(t, store, rs) + ctx := context.Background() + + put := func(e *Entry) { + if err := store.InsertEntry(ctx, e); err != nil { + t.Fatal(err) + } + } + for _, dir := range []string{"/buckets", "/buckets/mybucket", "/buckets/mybucket/d"} { + put(&Entry{FullPath: util.FullPath(dir), Attr: Attr{Mode: os.ModeDir}}) + } + put(&Entry{ + FullPath: "/buckets/mybucket/d/f.txt", + Attr: Attr{Mtime: time.Now(), Mode: 0644}, + Remote: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 10}, + }) + + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/mybucket", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "d", IsDirectory: true}, + DeleteChunks: false, + }, + TsNs: time.Now().UnixNano(), + } + if err := f.Replay(event); err != nil { + t.Fatalf("replay: %v", err) + } + + if len(stub.deleteCalls) != 0 { + t.Fatalf("a replaced remote object must be kept, got %v", stub.deleteCalls) + } +} + +// A backend offering a conditional delete must serve the sweep: the check and +// the delete bind into one preconditioned request, so a replacement landing +// between them cannot be deleted. +func TestReplayDirectoryDeleteUsesConditionalRemoteDelete(t *testing.T) { + const storageType = "stub_replay_remote_cond" + stub := &stubConditionalRemoteClient{conditionalResult: true} + defer registerStubMaker(t, storageType, stub)() + + conf := &remote_pb.RemoteConf{Name: "condstore", Type: storageType} + rs := NewFilerRemoteStorage() + rs.storageNameToConf[conf.Name] = conf + rs.mapDirectoryToRemoteStorage("/buckets/mybucket", &remote_pb.RemoteStorageLocation{ + Name: "condstore", + Bucket: "mybucket", + Path: "/", + }) + + store := newStubFilerStore() + f := newTestFiler(t, store, rs) + ctx := context.Background() + + put := func(e *Entry) { + if err := store.InsertEntry(ctx, e); err != nil { + t.Fatal(err) + } + } + for _, dir := range []string{"/buckets", "/buckets/mybucket", "/buckets/mybucket/d"} { + put(&Entry{FullPath: util.FullPath(dir), Attr: Attr{Mode: os.ModeDir}}) + } + put(&Entry{ + FullPath: "/buckets/mybucket/d/f.txt", + Attr: Attr{Mtime: time.Now(), Mode: 0644}, + Remote: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 10}, + }) + + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/mybucket", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "d", IsDirectory: true}, + DeleteChunks: false, + }, + TsNs: time.Now().UnixNano(), + } + if err := f.Replay(event); err != nil { + t.Fatalf("replay: %v", err) + } + + if len(stub.conditionalCalls) != 1 || stub.conditionalCalls[0].Path != "/d/f.txt" { + t.Fatalf("expected one conditional delete of /d/f.txt, got %v", stub.conditionalCalls) + } + if len(stub.deleteCalls) != 0 { + t.Fatalf("an unconditional delete must not run beside the conditional one, got %v", stub.deleteCalls) + } +} + +// A conditional delete reporting the object changed keeps the remote object +// but still sweeps the local metadata: the entry the peer held is gone either +// way. +func TestReplayDirectoryDeleteKeepsChangedRemoteObject(t *testing.T) { + const storageType = "stub_replay_remote_cond_keep" + stub := &stubConditionalRemoteClient{conditionalResult: false} + defer registerStubMaker(t, storageType, stub)() + + conf := &remote_pb.RemoteConf{Name: "condkeepstore", Type: storageType} + rs := NewFilerRemoteStorage() + rs.storageNameToConf[conf.Name] = conf + rs.mapDirectoryToRemoteStorage("/buckets/mybucket", &remote_pb.RemoteStorageLocation{ + Name: "condkeepstore", + Bucket: "mybucket", + Path: "/", + }) + + store := newStubFilerStore() + f := newTestFiler(t, store, rs) + ctx := context.Background() + + put := func(e *Entry) { + if err := store.InsertEntry(ctx, e); err != nil { + t.Fatal(err) + } + } + for _, dir := range []string{"/buckets", "/buckets/mybucket", "/buckets/mybucket/d"} { + put(&Entry{FullPath: util.FullPath(dir), Attr: Attr{Mode: os.ModeDir}}) + } + put(&Entry{ + FullPath: "/buckets/mybucket/d/f.txt", + Attr: Attr{Mtime: time.Now(), Mode: 0644}, + Remote: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 10}, + }) + + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/mybucket", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "d", IsDirectory: true}, + DeleteChunks: false, + }, + TsNs: time.Now().UnixNano(), + } + if err := f.Replay(event); err != nil { + t.Fatalf("replay: %v", err) + } + if _, err := store.FindEntry(ctx, "/buckets/mybucket/d/f.txt"); !errors.Is(err, filer_pb.ErrNotFound) { + t.Fatalf("swept child should be gone, got %v", err) + } + if len(stub.deleteCalls) != 0 { + t.Fatalf("a changed remote object must be kept, got %v", stub.deleteCalls) + } +} + +// A failed remote delete aborts the sweep so the replication stream retries +// instead of orphaning the object no other event will ever name. +func TestReplayDirectoryDeleteFailsOnRemoteDeleteError(t *testing.T) { + const storageType = "stub_replay_remote_err" + stub := &stubRemoteClient{ + statResult: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 10}, + deleteErr: errors.New("remote unavailable"), + } + defer registerStubMaker(t, storageType, stub)() + + conf := &remote_pb.RemoteConf{Name: "errstore", Type: storageType} + rs := NewFilerRemoteStorage() + rs.storageNameToConf[conf.Name] = conf + rs.mapDirectoryToRemoteStorage("/buckets/mybucket", &remote_pb.RemoteStorageLocation{ + Name: "errstore", + Bucket: "mybucket", + Path: "/", + }) + + store := newStubFilerStore() + f := newTestFiler(t, store, rs) + ctx := context.Background() + + put := func(e *Entry) { + if err := store.InsertEntry(ctx, e); err != nil { + t.Fatal(err) + } + } + for _, dir := range []string{"/buckets", "/buckets/mybucket", "/buckets/mybucket/d"} { + put(&Entry{FullPath: util.FullPath(dir), Attr: Attr{Mode: os.ModeDir}}) + } + put(&Entry{ + FullPath: "/buckets/mybucket/d/f.txt", + Attr: Attr{Mtime: time.Now(), Mode: 0644}, + Remote: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 10}, + }) + + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/mybucket", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "d", IsDirectory: true}, + DeleteChunks: false, + }, + TsNs: time.Now().UnixNano(), + } + if err := f.Replay(event); err == nil { + t.Fatal("a failed remote delete must fail the replay for retry") + } + if _, err := store.FindEntry(ctx, "/buckets/mybucket/d/f.txt"); err != nil { + t.Fatalf("the child must survive for the retry, got %v", err) + } +} + +// A swept entry deleted while its rename copy was still pending reads its +// content from the source the marker names: removing the entry must release +// that object too, or it is orphaned — the destination key may hold nothing. +func TestReplayDirectoryDeleteReleasesPendingRenameSource(t *testing.T) { + const storageType = "stub_replay_remote_rename_src" + stub := &pathStatRemoteClient{objects: map[string]*filer_pb.RemoteEntry{ + "/d/src.txt": {RemoteMtime: 1700000000, RemoteSize: 10}, + }} + defer registerStubMaker(t, storageType, stub)() + + conf := &remote_pb.RemoteConf{Name: "srcstore", Type: storageType} + rs := NewFilerRemoteStorage() + rs.storageNameToConf[conf.Name] = conf + rs.mapDirectoryToRemoteStorage("/buckets/mybucket", &remote_pb.RemoteStorageLocation{ + Name: "srcstore", + Bucket: "mybucket", + Path: "/", + }) + + store := newStubFilerStore() + f := newTestFiler(t, store, rs) + ctx := context.Background() + + put := func(e *Entry) { + if err := store.InsertEntry(ctx, e); err != nil { + t.Fatal(err) + } + } + for _, dir := range []string{"/buckets", "/buckets/mybucket", "/buckets/mybucket/d"} { + put(&Entry{FullPath: util.FullPath(dir), Attr: Attr{Mode: os.ModeDir}}) + } + put(&Entry{ + FullPath: "/buckets/mybucket/d/dst.txt", + Attr: Attr{Mtime: time.Now(), Mode: 0644}, + Remote: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 10}, + Extended: map[string][]byte{ExtRemoteRenameSourceKey: []byte("/buckets/mybucket/d/src.txt")}, + }) + + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/mybucket", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "d", IsDirectory: true}, + DeleteChunks: false, + }, + TsNs: time.Now().UnixNano(), + } + if err := f.Replay(event); err != nil { + t.Fatalf("replay: %v", err) + } + if _, err := store.FindEntry(ctx, "/buckets/mybucket/d/dst.txt"); !errors.Is(err, filer_pb.ErrNotFound) { + t.Fatalf("swept child should be gone, got %v", err) + } + if len(stub.deleteCalls) != 1 || stub.deleteCalls[0].Path != "/d/src.txt" { + t.Fatalf("expected one remote delete of the pending rename's source /d/src.txt, got %v", stub.deleteCalls) + } +} + +// The source survives while a remote-only entry holds its path: that entry's +// content is the same object. The swept destination entry's own object still +// goes while its stamp describes it. +func TestReplayDirectoryDeleteKeepsPendingRenameSourceOwnedByRemoteOnly(t *testing.T) { + const storageType = "stub_replay_remote_rename_owned" + stub := &pathStatRemoteClient{objects: map[string]*filer_pb.RemoteEntry{ + "/d/dst.txt": {RemoteMtime: 1700000000, RemoteSize: 10}, + "/other/src.txt": {RemoteMtime: 1700000000, RemoteSize: 10}, + }} + defer registerStubMaker(t, storageType, stub)() + + conf := &remote_pb.RemoteConf{Name: "ownstore", Type: storageType} + rs := NewFilerRemoteStorage() + rs.storageNameToConf[conf.Name] = conf + rs.mapDirectoryToRemoteStorage("/buckets/mybucket", &remote_pb.RemoteStorageLocation{ + Name: "ownstore", + Bucket: "mybucket", + Path: "/", + }) + + store := newStubFilerStore() + f := newTestFiler(t, store, rs) + ctx := context.Background() + + put := func(e *Entry) { + if err := store.InsertEntry(ctx, e); err != nil { + t.Fatal(err) + } + } + for _, dir := range []string{"/buckets", "/buckets/mybucket", "/buckets/mybucket/d", "/buckets/mybucket/other"} { + put(&Entry{FullPath: util.FullPath(dir), Attr: Attr{Mode: os.ModeDir}}) + } + put(&Entry{ + FullPath: "/buckets/mybucket/d/dst.txt", + Attr: Attr{Mtime: time.Now(), Mode: 0644}, + Remote: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 10}, + Extended: map[string][]byte{ExtRemoteRenameSourceKey: []byte("/buckets/mybucket/other/src.txt")}, + }) + put(&Entry{ + FullPath: "/buckets/mybucket/other/src.txt", + Attr: Attr{Mtime: time.Now(), Mode: 0644}, + Remote: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 10}, + }) + + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/mybucket", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "d", IsDirectory: true}, + DeleteChunks: false, + }, + TsNs: time.Now().UnixNano(), + } + if err := f.Replay(event); err != nil { + t.Fatalf("replay: %v", err) + } + if len(stub.deleteCalls) != 1 || stub.deleteCalls[0].Path != "/d/dst.txt" { + t.Fatalf("expected one remote delete of /d/dst.txt, got %v", stub.deleteCalls) + } + if _, err := store.FindEntry(ctx, "/buckets/mybucket/other/src.txt"); err != nil { + t.Fatalf("the remote-only source holder must survive, got %v", err) + } +} + +// A remote-only entry that took the path since the listing reads the object +// at the key as its only copy — the sweep must keep both. +func TestReplaySweepKeepsRemoteOnlyOwnerOfPath(t *testing.T) { + const storageType = "stub_replay_remote_owner" + stub := &pathStatRemoteClient{objects: map[string]*filer_pb.RemoteEntry{ + "/d/f.txt": {RemoteMtime: 1700000099, RemoteSize: 20}, + }} + defer registerStubMaker(t, storageType, stub)() + + conf := &remote_pb.RemoteConf{Name: "ownerstore", Type: storageType} + rs := NewFilerRemoteStorage() + rs.storageNameToConf[conf.Name] = conf + rs.mapDirectoryToRemoteStorage("/buckets/mybucket", &remote_pb.RemoteStorageLocation{ + Name: "ownerstore", + Bucket: "mybucket", + Path: "/", + }) + + store := newStubFilerStore() + f := newTestFiler(t, store, rs) + ctx := context.Background() + + owner := &Entry{ + FullPath: "/buckets/mybucket/d/f.txt", + Attr: Attr{Mtime: time.Now(), Mode: 0644}, + Remote: &filer_pb.RemoteEntry{RemoteMtime: 1700000099, RemoteSize: 20}, + } + if err := store.InsertEntry(ctx, owner); err != nil { + t.Fatal(err) + } + + stale := &Entry{ + FullPath: "/buckets/mybucket/d/f.txt", + Attr: Attr{Mtime: time.Now(), Mode: 0644}, + Remote: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 10}, + } + sweep := &folderSweep{chunksByPath: make(map[util.FullPath][]*filer_pb.FileChunk)} + message := &filer_pb.EventNotification{OldEntry: &filer_pb.Entry{Name: "d", IsDirectory: true}} + kept, err := f.deleteSweptChild(ctx, stale, message, 0, sweep, false, true) + if err != nil { + t.Fatalf("deleteSweptChild: %v", err) + } + if !kept { + t.Fatal("a remote-only entry that took the path must be kept") + } + if _, err := store.FindEntry(ctx, "/buckets/mybucket/d/f.txt"); err != nil { + t.Fatalf("the remote-only owner must survive, got %v", err) + } + if len(stub.deleteCalls) != 0 { + t.Fatalf("the owned remote object must be kept, got %v", stub.deleteCalls) + } +} diff --git a/weed/filer/meta_replay.go b/weed/filer/meta_replay.go index cad4f41cf..ca517f865 100644 --- a/weed/filer/meta_replay.go +++ b/weed/filer/meta_replay.go @@ -2,6 +2,7 @@ package filer import ( "context" + "errors" "fmt" "slices" "strings" @@ -9,6 +10,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/protobuf/proto" ) // Replay applies the delete and the insert as separate, non-transactional @@ -85,7 +87,7 @@ func (f *Filer) Replay(resp *filer_pb.SubscribeMetadataResponse) error { keepDir := false if message.OldEntry.IsDirectory { var err error - keepDir, err = f.deleteReplayedFolderChildren(ctx, oldPath, message.DeleteChunks, resp.TsNs) + keepDir, err = f.deleteReplayedFolderChildren(ctx, oldPath, message, resp.TsNs) if err != nil { return err } @@ -125,9 +127,9 @@ func (f *Filer) Replay(resp *filer_pb.SubscribeMetadataResponse) error { // the chunks of removed files for deletion. Children this filer committed // after the event stay — a newer local write wins over a delayed delete — // along with the directories that hold them. -func (f *Filer) deleteReplayedFolderChildren(ctx context.Context, dir util.FullPath, deleteChunks bool, tsNs int64) (bool, error) { +func (f *Filer) deleteReplayedFolderChildren(ctx context.Context, dir util.FullPath, message *filer_pb.EventNotification, tsNs int64) (bool, error) { sweep := &folderSweep{chunksByPath: make(map[util.FullPath][]*filer_pb.FileChunk)} - kept, err := f.sweepReplayedFolder(ctx, dir, deleteChunks, tsNs, sweep) + kept, err := f.sweepReplayedFolder(ctx, dir, message, tsNs, sweep) // Flush what was collected even on error: the entries already deleted // cannot be listed again on retry, so their chunks must be queued now. sweep.flush(ctx, f) @@ -164,7 +166,9 @@ func (s *folderSweep) flush(ctx context.Context, f *Filer) { // newer than the event, working subtree-first so nested entries go too. // It reports whether a newer descendant keeps dir alive — the caller then // keeps the dir entry itself. -func (f *Filer) sweepReplayedFolder(ctx context.Context, dir util.FullPath, deleteChunks bool, tsNs int64, sweep *folderSweep) (kept bool, err error) { +func (f *Filer) sweepReplayedFolder(ctx context.Context, dir util.FullPath, message *filer_pb.EventNotification, tsNs int64, sweep *folderSweep) (kept bool, err error) { + keepRemote := IsMetadataOnlyDelete(message.OldEntry) + deleteRemote := !keepRemote && !message.GetIsFromOtherCluster() lastFileName := "" for { entries, _, lerr := f.ListDirectoryEntries(ctx, dir, lastFileName, false, PaginationSize, "", "", "") @@ -182,7 +186,7 @@ func (f *Filer) sweepReplayedFolder(ctx context.Context, dir util.FullPath, dele childKept = localTsNs > tsNs } if !childKept && sub.IsDirectory() { - childKept, err = f.sweepReplayedFolder(ctx, sub.FullPath, deleteChunks, tsNs, sweep) + childKept, err = f.sweepReplayedFolder(ctx, sub.FullPath, message, tsNs, sweep) if err != nil { return false, err } @@ -191,20 +195,12 @@ func (f *Filer) sweepReplayedFolder(ctx context.Context, dir util.FullPath, dele kept = true continue } - if derr := f.Store.DeleteEntry(ctx, sub.FullPath); derr != nil { - return false, fmt.Errorf("delete %s: %w", sub.FullPath, derr) + childKept, err = f.deleteSweptChild(ctx, sub, message, tsNs, sweep, keepRemote, deleteRemote) + if err != nil { + return false, err } - f.notePathChange(sub.FullPath, tsNs, true) - // hard-linked chunks are owned by the shared inode record; the - // origin's own events decide their deletion. A directory entry - // can still carry chunks of its own — an S3 child promotes an - // existing file to a directory without dropping them — so the - // entry's chunks are keyed to its own path whatever it is. - if deleteChunks && len(sub.HardLinkId) == 0 && len(sub.GetChunks()) > 0 { - sweep.add(sub.FullPath, sub.GetChunks()) - if sweep.count >= folderSweepChunkBatch { - sweep.flush(ctx, f) - } + if childKept { + kept = true } } if len(entries) < PaginationSize { @@ -214,6 +210,74 @@ func (f *Filer) sweepReplayedFolder(ctx context.Context, dir util.FullPath, dele return kept, nil } +// deleteSweptChild removes sub under its entry lock and reports whether the +// child was kept instead. The listing and the first commit-time check run +// without the lock, so a local write can land in between: the check runs +// again under the lock before anything is removed. A remote-only entry that +// has taken the path since — one the commit record may not cover, such as a +// lazy fetch materialized from the remote — reads the object at the key as +// its only copy: the write-back daemon's processDeleteEvent keeps such an +// object, and the sweep does too. +func (f *Filer) deleteSweptChild(ctx context.Context, sub *Entry, message *filer_pb.EventNotification, tsNs int64, sweep *folderSweep, keepRemote, deleteRemote bool) (bool, error) { + var childLock *util.ActiveLock + if lt := f.EntryLockTable; lt != nil { + // The sweep already holds locks on the event's paths; waiting for a + // descendant here can deadlock against a local rename that holds the + // child and waits for one of them. ErrLockContended is transient, so + // the replication retry picks the event up once the write settles. + var acquired bool + childLock, acquired = lt.TryAcquireLock("MetaReplay", sub.FullPath, util.ExclusiveLock) + if !acquired { + return false, fmt.Errorf("sweep %s: %w", sub.FullPath, util.ErrLockContended) + } + defer lt.ReleaseLock(sub.FullPath, childLock) + } + if tsNs > 0 { + localTsNs, tsErr := f.localCommitTsNs(ctx, sub.FullPath) + if tsErr != nil { + return false, tsErr + } + if localTsNs > tsNs { + return true, nil + } + } + current, findErr := f.Store.FindEntry(ctx, sub.FullPath) + if findErr != nil && !errors.Is(findErr, filer_pb.ErrNotFound) { + return false, fmt.Errorf("find %s: %w", sub.FullPath, findErr) + } + if current != nil && len(current.Content) == 0 && current.IsInRemoteOnly() && + !proto.Equal(current.ToProtoEntry(), sub.ToProtoEntry()) { + glog.V(0).InfofCtx(ctx, "keep %s: a remote-only entry has taken the path since the listing", sub.FullPath) + return true, nil + } + if !keepRemote && f.RemoteStorage != nil && f.remoteTombstones != nil { + if _, remoteLoc := f.RemoteStorage.FindMountDirectory(sub.FullPath); remoteLoc != nil { + f.remoteTombstones.addFromEvent(string(sub.FullPath), sub.IsDirectory(), tsNs) + } + } + if deleteRemote && !sub.IsDirectory() { + if remoteErr := f.deleteSweptEntryRemote(ctx, sub); remoteErr != nil { + return false, fmt.Errorf("remote delete %s: %w", sub.FullPath, remoteErr) + } + } + if derr := f.Store.DeleteEntry(ctx, sub.FullPath); derr != nil { + return false, fmt.Errorf("delete %s: %w", sub.FullPath, derr) + } + f.notePathChange(sub.FullPath, tsNs, true) + // chunks of a hard-linked child belong to the shared inode record; free + // them only once that record is gone. A directory entry can still carry + // chunks of its own — an S3 child promotes an existing file to a + // directory without dropping them — so the entry's chunks are keyed to + // its own path whatever it is. + if message.DeleteChunks && len(sub.GetChunks()) > 0 && (len(sub.HardLinkId) == 0 || f.hardLinkIsOrphaned(ctx, sub.HardLinkId)) { + sweep.add(sub.FullPath, sub.GetChunks()) + if sweep.count >= folderSweepChunkBatch { + sweep.flush(ctx, f) + } + } + return false, nil +} + // replayBlockedByLocal reports whether the local store already holds a // version of path committed after the event's timestamp. Lookup failures are // propagated: a replayed change must not be applied on the guess that the diff --git a/weed/filer/mongodb/mongodb_store_kv.go b/weed/filer/mongodb/mongodb_store_kv.go index 193bb247b..ce7ac8fed 100644 --- a/weed/filer/mongodb/mongodb_store_kv.go +++ b/weed/filer/mongodb/mongodb_store_kv.go @@ -46,7 +46,7 @@ func (store *MongodbStore) KvGet(ctx context.Context, key []byte) (value []byte, err = store.connect.Database(store.database).Collection(store.collectionName).FindOne(ctx, where).Decode(&data) if err != mongo.ErrNoDocuments && err != nil { glog.ErrorfCtx(ctx, "kv get: %v", err) - return nil, filer.ErrKvNotFound + return nil, err } if len(data.Meta) == 0 { diff --git a/weed/filer/reader_cache_memory_test.go b/weed/filer/reader_cache_memory_test.go index d363a5b4d..da73c7a2e 100644 --- a/weed/filer/reader_cache_memory_test.go +++ b/weed/filer/reader_cache_memory_test.go @@ -3,6 +3,7 @@ package filer import ( "context" "fmt" + "strconv" "sync" "sync/atomic" "testing" @@ -202,6 +203,12 @@ func TestReaderCacheFailedPrefetchReleasesBudget(t *testing.T) { } func TestReaderCacheUnboundedWithoutBudget(t *testing.T) { + if strconv.IntSize == 32 { + // Proving the cache has no implicit memory budget needs more than 64 + // in-flight 4 MiB buffers (>256 MiB) at once; the weed/filer test binary + // already runs near the 4 GiB address-space limit on 32-bit. + t.Skip("requires >256MiB of in-flight buffers; skipped on 32-bit") + } const readers = 96 // more than 256MiB / 4MiB = 64 buffers started := make(chan struct{}, readers) gate := make(chan struct{}) diff --git a/weed/remote_storage/azure/azure_storage_client.go b/weed/remote_storage/azure/azure_storage_client.go index e9d9d28e3..15ff2519d 100644 --- a/weed/remote_storage/azure/azure_storage_client.go +++ b/weed/remote_storage/azure/azure_storage_client.go @@ -2,6 +2,7 @@ package azure import ( "context" + "errors" "fmt" "io" "math" @@ -501,38 +502,54 @@ func toMetadata(attributes map[string][]byte) map[string]*string { return metadata } -func (az *azureRemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStorageLocation, oldEntry *filer_pb.Entry, newEntry *filer_pb.Entry) (err error) { +func (az *azureRemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStorageLocation, oldEntry *filer_pb.Entry, newEntry *filer_pb.Entry) (remoteEntry *filer_pb.RemoteEntry, err error) { if reflect.DeepEqual(oldEntry.Extended, newEntry.Extended) { - return nil + return nil, nil } metadata := toMetadata(newEntry.Extended) key := loc.Path[1:] blobClient := az.client.ServiceClient().NewContainerClient(loc.Bucket).NewBlobClient(key) + updated := false + var writeETag *azcore.ETag if !reflect.DeepEqual(toMetadata(oldEntry.Extended), metadata) { - if _, err = blobClient.SetMetadata(context.Background(), metadata, nil); err != nil { - return err + metaResp, metaErr := blobClient.SetMetadata(context.Background(), metadata, nil) + if metaErr != nil { + return nil, metaErr } + writeETag = metaResp.ETag + updated = true } if encoding := remote_storage.EntryContentEncoding(newEntry); encoding != remote_storage.EntryContentEncoding(oldEntry) { // SetHTTPHeaders replaces the whole header set, so carry the rest over props, getErr := blobClient.GetProperties(context.Background(), nil) if getErr != nil { - return fmt.Errorf("azure get properties %s%s: %w", loc.Bucket, loc.Path, getErr) + return nil, fmt.Errorf("azure get properties %s%s: %w", loc.Bucket, loc.Path, getErr) } httpHeaders := blob.ParseHTTPHeaders(props) httpHeaders.BlobContentEncoding = nil if encoding != "" { httpHeaders.BlobContentEncoding = &encoding } - if _, err = blobClient.SetHTTPHeaders(context.Background(), httpHeaders, nil); err != nil { - return fmt.Errorf("azure set http headers %s%s: %w", loc.Bucket, loc.Path, err) + hdrResp, hdrErr := blobClient.SetHTTPHeaders(context.Background(), httpHeaders, nil) + if hdrErr != nil { + return nil, fmt.Errorf("azure set http headers %s%s: %w", loc.Bucket, loc.Path, hdrErr) } + writeETag = hdrResp.ETag + updated = true } - - return + if !updated { + return nil, nil + } + remoteEntry, err = az.StatFile(loc) + if err == nil && writeETag != nil { + // the stat can describe a concurrent replacement; the write response + // names the object this update produced + remoteEntry.RemoteETag = string(*writeETag) + } + return remoteEntry, err } func (az *azureRemoteStorageClient) DeleteFile(loc *remote_pb.RemoteStorageLocation) (err error) { @@ -552,6 +569,45 @@ func (az *azureRemoteStorageClient) DeleteFile(loc *remote_pb.RemoteStorageLocat return } +// DeleteFileIfUnchanged deletes the blob only while it is still the one the +// stat describes, enforced with an If-Match ETag precondition. A replacement +// that lands between the stat and the delete fails the precondition instead +// of being deleted. +func (az *azureRemoteStorageClient) DeleteFileIfUnchanged(loc *remote_pb.RemoteStorageLocation, stamp *filer_pb.RemoteEntry) (bool, error) { + live, err := az.StatFile(loc) + if err != nil { + if errors.Is(err, remote_storage.ErrRemoteObjectNotFound) { + return true, nil + } + return false, err + } + if !remote_storage.StillCurrent(stamp, live) { + return false, nil + } + key := loc.Path[1:] + blobClient := az.client.ServiceClient().NewContainerClient(loc.Bucket).NewBlobClient(key) + options := &blob.DeleteOptions{ + DeleteSnapshots: to.Ptr(blob.DeleteSnapshotsOptionTypeInclude), + } + if live.RemoteETag != "" { + options.AccessConditions = &blob.AccessConditions{ + ModifiedAccessConditions: &blob.ModifiedAccessConditions{ + IfMatch: to.Ptr(azcore.ETag(live.RemoteETag)), + }, + } + } + if _, err := blobClient.Delete(context.Background(), options); err != nil { + if bloberror.HasCode(err, bloberror.BlobNotFound) { + return true, nil + } + if bloberror.HasCode(err, bloberror.ConditionNotMet) { + return false, nil + } + return false, fmt.Errorf("azure delete %s%s: %w", loc.Bucket, loc.Path, err) + } + return true, nil +} + func (az *azureRemoteStorageClient) ListBuckets() (buckets []*remote_storage.Bucket, err error) { pager := az.client.NewListContainersPager(nil) diff --git a/weed/remote_storage/azure/azure_storage_client_test.go b/weed/remote_storage/azure/azure_storage_client_test.go index 1de998791..5c58a7bd7 100644 --- a/weed/remote_storage/azure/azure_storage_client_test.go +++ b/weed/remote_storage/azure/azure_storage_client_test.go @@ -138,7 +138,7 @@ func TestAzureStorageClientBasic(t *testing.T) { "x-amz-meta-new-key": []byte("new-value"), }, } - err := azClient.UpdateFileMetadata(loc, oldEntry, newEntry) + _, err := azClient.UpdateFileMetadata(loc, oldEntry, newEntry) if err != nil { t.Fatalf("Failed to update metadata: %v", err) } diff --git a/weed/remote_storage/gcs/gcs_storage_client.go b/weed/remote_storage/gcs/gcs_storage_client.go index c37671d5c..c76881e98 100644 --- a/weed/remote_storage/gcs/gcs_storage_client.go +++ b/weed/remote_storage/gcs/gcs_storage_client.go @@ -21,6 +21,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/util" "golang.org/x/oauth2" "golang.org/x/oauth2/google" + "google.golang.org/api/googleapi" "google.golang.org/api/iterator" "google.golang.org/api/option" "google.golang.org/protobuf/proto" @@ -374,9 +375,9 @@ func toMetadata(attributes map[string][]byte) map[string]string { return metadata } -func (gcs *gcsRemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStorageLocation, oldEntry *filer_pb.Entry, newEntry *filer_pb.Entry) (err error) { +func (gcs *gcsRemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStorageLocation, oldEntry *filer_pb.Entry, newEntry *filer_pb.Entry) (remoteEntry *filer_pb.RemoteEntry, err error) { if reflect.DeepEqual(oldEntry.Extended, newEntry.Extended) { - return nil + return nil, nil } attrsToUpdate := storage.ObjectAttrsToUpdate{} if metadata := toMetadata(newEntry.Extended); len(metadata) > 0 { @@ -389,11 +390,14 @@ func (gcs *gcsRemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStora } if attrsToUpdate.Metadata == nil && attrsToUpdate.ContentEncoding == nil { - return nil + return nil, nil } key := loc.Path[1:] - _, err = gcs.client.Bucket(loc.Bucket).Object(key).Update(context.Background(), attrsToUpdate) - return + attr, err := gcs.client.Bucket(loc.Bucket).Object(key).Update(context.Background(), attrsToUpdate) + if err != nil { + return nil, err + } + return gcs.toRemoteEntry(attr), nil } func (gcs *gcsRemoteStorageClient) DeleteFile(loc *remote_pb.RemoteStorageLocation) (err error) { key := loc.Path[1:] @@ -406,6 +410,38 @@ func (gcs *gcsRemoteStorageClient) DeleteFile(loc *remote_pb.RemoteStorageLocati return } +// DeleteFileIfUnchanged deletes the object only while it is still the one the +// stat describes, enforced with a generation precondition. A replacement that +// lands between the stat and the delete gets a new generation and fails the +// precondition instead of being deleted. +func (gcs *gcsRemoteStorageClient) DeleteFileIfUnchanged(loc *remote_pb.RemoteStorageLocation, stamp *filer_pb.RemoteEntry) (bool, error) { + key := loc.Path[1:] + object := gcs.client.Bucket(loc.Bucket).Object(key) + ctx, cancel := context.WithTimeout(context.Background(), defaultGCSOpTimeout) + defer cancel() + attr, err := object.Attrs(ctx) + if err != nil { + if errors.Is(err, storage.ErrObjectNotExist) { + return true, nil + } + return false, fmt.Errorf("stat gcs %s%s: %w", loc.Bucket, loc.Path, err) + } + if !remote_storage.StillCurrent(stamp, gcs.toRemoteEntry(attr)) { + return false, nil + } + if err := object.If(storage.Conditions{GenerationMatch: attr.Generation}).Delete(ctx); err != nil { + if errors.Is(err, storage.ErrObjectNotExist) { + return true, nil + } + var gerr *googleapi.Error + if errors.As(err, &gerr) && gerr.Code == http.StatusPreconditionFailed { + return false, nil + } + return false, fmt.Errorf("gcs delete %s%s: %w", loc.Bucket, key, err) + } + return true, nil +} + func (gcs *gcsRemoteStorageClient) ListBuckets() (buckets []*remote_storage.Bucket, err error) { if gcs.projectID == "" { return nil, fmt.Errorf("gcs project id or GOOGLE_CLOUD_PROJECT env variable not set") diff --git a/weed/remote_storage/remote_storage.go b/weed/remote_storage/remote_storage.go index 0cefe362a..73f568f5a 100644 --- a/weed/remote_storage/remote_storage.go +++ b/weed/remote_storage/remote_storage.go @@ -26,6 +26,28 @@ func ParseLocationName(remote string) (locationName string) { return } +// Describes reports whether a stored remote entry's stamp still describes the +// live remote object: same size, and equal ETags when both are known. +func Describes(stamp, object *filer_pb.RemoteEntry) bool { + if stamp == nil || object == nil || stamp.RemoteSize != object.RemoteSize { + return false + } + return stamp.RemoteETag == "" || object.RemoteETag == "" || stamp.RemoteETag == object.RemoteETag +} + +// StillCurrent reports whether live can still be the object the stored stamp +// described: not newer than the stamp, matching it on size, and not having +// dropped an ETag the stamp carried. +func StillCurrent(stamp, live *filer_pb.RemoteEntry) bool { + if live != nil && live.RemoteMtime > stamp.GetRemoteMtime() { + return false + } + if !Describes(stamp, live) { + return false + } + return stamp.GetRemoteETag() == "" || live.RemoteETag != "" +} + func parseBucketLocation(remote string) (loc *remote_pb.RemoteStorageLocation) { loc = &remote_pb.RemoteStorageLocation{} remote = strings.TrimSuffix(remote, slash) @@ -91,7 +113,11 @@ type RemoteStorageClient interface { WriteDirectory(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry) (err error) RemoveDirectory(loc *remote_pb.RemoteStorageLocation) (err error) WriteFile(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry, reader io.Reader) (remoteEntry *filer_pb.RemoteEntry, err error) - UpdateFileMetadata(loc *remote_pb.RemoteStorageLocation, oldEntry *filer_pb.Entry, newEntry *filer_pb.Entry) (err error) + // UpdateFileMetadata returns the stamp of the object the update produced — + // a backend that rewrites the object to change its metadata can give it a + // new ETag or mtime — or nil when the object was not rewritten. Callers + // restamp their entries from the result and keep the previous stamp on nil. + UpdateFileMetadata(loc *remote_pb.RemoteStorageLocation, oldEntry *filer_pb.Entry, newEntry *filer_pb.Entry) (remoteEntry *filer_pb.RemoteEntry, err error) DeleteFile(loc *remote_pb.RemoteStorageLocation) (err error) ListBuckets() ([]*Bucket, error) CreateBucket(name string) (err error) @@ -118,6 +144,15 @@ type RemoteStorageObjectCopier interface { CopyFile(src *remote_pb.RemoteStorageLocation, dst *remote_pb.RemoteStorageLocation) (remoteEntry *filer_pb.RemoteEntry, err error) } +// RemoteStorageConditionalDeleter is an optional interface for remote storage +// clients whose backend can delete an object only while it is still the one a +// stat described, enforced with a backend version precondition (S3 and Azure: +// ETag, GCS: generation). deleted is false without an error when the object +// differs from stamp or changed between the stat and the delete. +type RemoteStorageConditionalDeleter interface { + DeleteFileIfUnchanged(loc *remote_pb.RemoteStorageLocation, stamp *filer_pb.RemoteEntry) (deleted bool, err error) +} + // CacheWaitTimeout is how long a read of an uncached remote-only object waits // for the local cache before serving another way: small files wait longer since // their cache completes quickly, large files fail fast for better TTFB. A mount diff --git a/weed/remote_storage/s3/s3_storage_client.go b/weed/remote_storage/s3/s3_storage_client.go index 4166df8a4..7664407eb 100644 --- a/weed/remote_storage/s3/s3_storage_client.go +++ b/weed/remote_storage/s3/s3_storage_client.go @@ -586,9 +586,9 @@ func (s *s3RemoteStorageClient) readFileRemoteEntry(loc *remote_pb.RemoteStorage // the largest object a single CopyObject call accepts const s3CopyObjectSizeLimit = 5 * 1024 * 1024 * 1024 -func (s *s3RemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStorageLocation, oldEntry *filer_pb.Entry, newEntry *filer_pb.Entry) (err error) { +func (s *s3RemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStorageLocation, oldEntry *filer_pb.Entry, newEntry *filer_pb.Entry) (remoteEntry *filer_pb.RemoteEntry, err error) { if reflect.DeepEqual(oldEntry.Extended, newEntry.Extended) { - return nil + return nil, nil } // Content-Encoding is S3 system metadata, changeable without a content @@ -605,7 +605,7 @@ func (s *s3RemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStorageL Key: aws.String(key), }) if headErr != nil { - return fmt.Errorf("stat %s/%s before metadata copy: %w", loc.Bucket, key, headErr) + return nil, fmt.Errorf("stat %s/%s before metadata copy: %w", loc.Bucket, key, headErr) } copyInput := &s3.CopyObjectInput{ Bucket: aws.String(loc.Bucket), @@ -636,8 +636,25 @@ func (s *s3RemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStorageL if s.conf.S3StorageClass != "" { copyInput.StorageClass = aws.String(s.conf.S3StorageClass) } - if _, err = s.conn.CopyObject(copyInput); err != nil { - return fmt.Errorf("update content encoding of %s/%s: %w", loc.Bucket, key, err) + copyOutput, copyErr := s.conn.CopyObject(copyInput) + if copyErr != nil { + return nil, fmt.Errorf("update content encoding of %s/%s: %w", loc.Bucket, key, copyErr) + } + // the copy rewrites the object and can mint a new ETag, so the + // stamp it produced is reported instead of a fresh stat — which + // could describe a replacement that landed in between + remoteEntry = &filer_pb.RemoteEntry{ + StorageName: s.conf.Name, + RemoteSize: aws.Int64Value(headOut.ContentLength), + } + if result := copyOutput.CopyObjectResult; result != nil { + remoteEntry.RemoteETag = aws.StringValue(result.ETag) + if result.LastModified != nil { + remoteEntry.RemoteMtime = result.LastModified.Unix() + } + } + if encoding != "" { + remoteEntry.RemoteContentEncoding = aws.String(encoding) } } } @@ -661,6 +678,9 @@ func (s *s3RemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStorageL Key: aws.String(loc.Path[1:]), }) } + if err != nil { + return nil, err + } return } @@ -833,6 +853,45 @@ func (s *s3RemoteStorageClient) DeleteFile(loc *remote_pb.RemoteStorageLocation) return } +// DeleteFileIfUnchanged deletes the object only while it is still the one the +// stat describes, enforced with the S3 conditional-delete If-Match header. A +// replacement that lands between the stat and the delete fails the +// precondition instead of being deleted. +func (s *s3RemoteStorageClient) DeleteFileIfUnchanged(loc *remote_pb.RemoteStorageLocation, stamp *filer_pb.RemoteEntry) (bool, error) { + live, err := s.StatFile(loc) + if err != nil { + if errors.Is(err, remote_storage.ErrRemoteObjectNotFound) { + return true, nil + } + return false, err + } + if !remote_storage.StillCurrent(stamp, live) { + return false, nil + } + if live.RemoteETag == "" { + return true, s.DeleteFile(loc) + } + req, _ := s.conn.DeleteObjectRequest(&s3.DeleteObjectInput{ + Bucket: aws.String(loc.Bucket), + Key: aws.String(loc.Path[1:]), + }) + req.Handlers.Build.PushBack(func(r *request.Request) { + r.HTTPRequest.Header.Set("If-Match", live.RemoteETag) + }) + if err := req.Send(); err != nil { + if reqErr, ok := err.(awserr.RequestFailure); ok { + switch reqErr.StatusCode() { + case http.StatusPreconditionFailed: + return false, nil + case http.StatusNotFound: + return true, nil + } + } + return false, fmt.Errorf("delete s3 %s%s: %w", loc.Bucket, loc.Path, err) + } + return true, nil +} + func (s *s3RemoteStorageClient) ListBuckets() (buckets []*remote_storage.Bucket, err error) { resp, err := s.conn.ListBuckets(&s3.ListBucketsInput{}) if err != nil { diff --git a/weed/remote_storage/s3/s3_storage_client_test.go b/weed/remote_storage/s3/s3_storage_client_test.go index 72e84a808..568e4a633 100644 --- a/weed/remote_storage/s3/s3_storage_client_test.go +++ b/weed/remote_storage/s3/s3_storage_client_test.go @@ -297,10 +297,12 @@ func TestS3UpdateFileMetadataSkipsTaggingWhenUnsupported(t *testing.T) { plain := &filer_pb.Entry{} tagged := &filer_pb.Entry{Extended: map[string][]byte{"k1": []byte("v1")}} - require.NoError(t, client.UpdateFileMetadata(loc, plain, tagged)) + _, err := client.UpdateFileMetadata(loc, plain, tagged) + require.NoError(t, err) require.Empty(t, rt.requests, "adding attributes must not send PutObjectTagging") - require.NoError(t, client.UpdateFileMetadata(loc, tagged, plain)) + _, err = client.UpdateFileMetadata(loc, tagged, plain) + require.NoError(t, err) require.Empty(t, rt.requests, "removing attributes must not send DeleteObjectTagging") } @@ -310,13 +312,15 @@ func TestS3UpdateFileMetadataSendsTaggingWhenSupported(t *testing.T) { plain := &filer_pb.Entry{} tagged := &filer_pb.Entry{Extended: map[string][]byte{"k1": []byte("v1")}} - require.NoError(t, client.UpdateFileMetadata(loc, plain, tagged)) + _, err := client.UpdateFileMetadata(loc, plain, tagged) + require.NoError(t, err) require.Len(t, rt.requests, 1) require.Equal(t, http.MethodPut, rt.requests[0].Method) require.Contains(t, rt.requests[0].URL.RawQuery, "tagging") rt.requests = nil - require.NoError(t, client.UpdateFileMetadata(loc, tagged, plain)) + _, err = client.UpdateFileMetadata(loc, tagged, plain) + require.NoError(t, err) require.Len(t, rt.requests, 1) require.Equal(t, http.MethodDelete, rt.requests[0].Method) require.Contains(t, rt.requests[0].URL.RawQuery, "tagging") diff --git a/weed/util/lock_table.go b/weed/util/lock_table.go index 65daae39c..3869ec52f 100644 --- a/weed/util/lock_table.go +++ b/weed/util/lock_table.go @@ -128,6 +128,39 @@ func (lt *LockTable[T]) AcquireLock(intention string, key T, lockType LockType) return lock } +// TryAcquireLock grants the lock only when nothing holds or waits for the +// key; it reports false on contention instead of queueing. Callers already +// holding other locks use it so acquisition stays deadlock-free. +func (lt *LockTable[T]) TryAcquireLock(intention string, key T, lockType LockType) (lock *ActiveLock, acquired bool) { + lt.mu.Lock() + entry, exists := lt.locks[key] + if !exists { + entry = &LockEntry{} + entry.cond = sync.NewCond(&entry.mu) + lt.locks[key] = entry + lt.locksInFlight[key] = 0 + } + lt.locksInFlight[key]++ + entry.mu.Lock() + blocked := len(entry.waiters) > 0 || entry.activeExclusiveLockOwnerCount > 0 || + (lockType == ExclusiveLock && entry.activeSharedLockOwnerCount > 0) + if blocked { + entry.mu.Unlock() + lt.locksInFlight[key]-- + lt.mu.Unlock() + return nil, false + } + lock = lt.NewActiveLock(intention, lockType) + if lockType == ExclusiveLock { + entry.activeExclusiveLockOwnerCount++ + } else { + entry.activeSharedLockOwnerCount++ + } + entry.mu.Unlock() + lt.mu.Unlock() + return lock, true +} + func (lt *LockTable[T]) ReleaseLock(key T, lock *ActiveLock) { lt.mu.Lock() defer lt.mu.Unlock() diff --git a/weed/util/retry.go b/weed/util/retry.go index fbed37a16..6f77be911 100644 --- a/weed/util/retry.go +++ b/weed/util/retry.go @@ -16,6 +16,12 @@ import ( var RetryWaitTime = 6 * time.Second +// ErrLockContended marks a lock acquisition abandoned instead of waited on — +// the caller already holds other locks and cannot queue without inverting +// their order. The holder's write settles the contested state, so callers +// treat it as transient and retry. +var ErrLockContended = errors.New("a local write holds the entry lock") + // transientErrorMessages are substrings of failures that a later attempt is // likely to get past: connection resets, timeouts, and the throttling or // overload replies S3 and gRPC hand back. Cloud SDKs bury the underlying net @@ -84,6 +90,9 @@ func IsTransientError(err error) bool { if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { return false } + if errors.Is(err, ErrLockContended) { + return true + } if errors.Is(err, io.ErrUnexpectedEOF) || errors.Is(err, syscall.ECONNRESET) || errors.Is(err, syscall.ECONNABORTED) || errors.Is(err, syscall.ECONNREFUSED) || errors.Is(err, syscall.EPIPE) ||