From 6a594da7391a30b8549f042e5e0de805ae7f73f5 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 18 Oct 2026 12:23:30 -0700 Subject: [PATCH] filer: free orphaned hard-link chunks and remote objects in delete sweeps (#11809) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * filer: free shared chunks when a folder delete drops the last hard link doBatchDeleteFolderMetaAndData skipped chunk collection for every hard-linked child, relying on maybeDeleteHardLinks to retire the shared inode record. When the delete removes the last link the record is gone and nobody else will ever free those chunks. After the link callback runs, check whether the record survived; if it did not, queue the entry's chunks for deletion. * filer: free orphaned hard-link chunks in the replayed delete sweep The subtree sweep on a replayed directory delete skipped every hard-linked child, so a link that was the inode's last owner leaked its shared chunks. Check the inode record after the store delete; when it is gone, queue the child's chunks like any other swept entry. * filer: mirror remote-object cleanup in the replayed delete sweep Children a lagging peer alone still holds under a replayed directory delete were dropped with plain store deletes: their remote objects leaked and lazy remote fetches could resurrect them. The sweep now records event-stamped tombstones and deletes swept non-directory children through maybeDeleteFromRemoteIfUnchanged, which carries the same stat guard the write-back daemon's processDeleteEvent applies plus a modification-time check — a delayed replay cannot erase an object another writer replaced, even a same-size one. A failed remote delete fails the replay so the event is retried instead of orphaning the object. The directory's own tombstone comes from the event pipeline at the event timestamp and covers the subtree. Metadata-only deletes and cross-cluster events keep the origin's semantics. * filerstore: return real errors from KvGet on MongoDB, ArangoDB, Elastic These stores mapped every lookup failure to ErrKvNotFound, so a database error looked identical to a missing key: hard-link records could appear gone while links still owned their chunks, and path timestamp reads could treat committed state as absent. Return the underlying error and keep ErrKvNotFound for genuinely missing keys. * filer: bind remote-object deletes to the observed object version A guarded stat-then-delete still lets a replacement landing between the two requests go down with the deleted object. Add an optional conditional-delete operation to remote storage clients — S3 If-Match on the live ETag, GCS generation match, Azure IfMatch — and use it everywhere a delete is guarded by a stored stamp: the write-back daemon's delete, rename-source, and pending-rename paths, and the replayed-delete sweep. Backends without it keep the guarded stat-then-delete. * filer: restamp the remote entry after a metadata-only remote update Changing Content-Encoding rewrites the object in place on S3 and advances its LastModified, while the recorded stamp keeps the old mtime. A version-checked delete would then mistake the object for a replacement and keep it forever. Re-stat after UpdateFileMetadata so the saved stamp describes the live object. * filer: mirror the remote-delete ownership checks in the replayed sweep The sweep deleted swept children's remote objects on the listing's snapshot alone. Two gaps: a remote-only entry that took the path since — one a commit record may not cover, such as a lazy fetch materialized from the remote — reads the object as its only copy, and an entry removed while its rename copy was pending leaves the object at the source key orphaned. The per-child delete now runs under the child's entry lock with the commit-time check repeated, keeps a remote-only entry that holds the path, and releases the pending rename's source under the same ownership and stamp conditions the write-back daemon's releasePendingSource applies. * remote.sync: restamp only the object the metadata update touched The live stat after UpdateFileMetadata was saved unconditionally, so a replacement written between the update and the stat stamped the entry with an object it never described — and a later delete would then pass the freshness guard on the replacement's data. The restamp now requires the live object to still match the stored stamp on size and ETag; a different object keeps the old stamp, which the delete guard reads as replaced and keeps. * filer: skip unbudgeted reader-cache test on 32-bit TestReaderCacheUnboundedWithoutBudget proves the reader cache has no implicit memory budget by keeping 96 concurrent 4 MiB downloads in flight (>256 MiB). On linux/386 the weed/filer test binary already runs near the 4 GiB address-space limit, so the burst reliably crashes it with "out of memory" mid-suite. * filer: bail out of a swept child lock instead of waiting on it Replay already holds the event's path locks when the sweep reaches a descendant. A local rename can hold that descendant while waiting on one of the replayed paths, so blocking on the child lock deadlocks both. The sweep now tries the lock and reports ErrLockContended — transient, so the replication retry replays the event once the local write settles. * remote: report the stamp a metadata update produced A metadata write can rewrite the object it touches — S3 changes Content-Encoding through an in-place copy, which can mint a new ETag — so restamping from a fresh stat can reject the very object the update produced, or describe a replacement that landed in between. UpdateFileMetadata now returns the RemoteEntry of the object it wrote (S3 from the copy result, GCS from the update attributes, Azure from a post-update stat); nil means the object was untouched and callers keep the stored stamp. * remote: restamp Azure from the metadata write's own etag --- weed/command/filer_remote_gateway_buckets.go | 10 +- weed/command/filer_remote_sync_dir.go | 93 ++- weed/command/filer_remote_sync_dir_test.go | 71 +- weed/filer/arangodb/arangodb_store_kv.go | 2 +- weed/filer/elastic/v7/elastic_store_kv.go | 5 +- weed/filer/filer_delete_entry.go | 10 + weed/filer/filer_delete_entry_test.go | 56 ++ weed/filer/filer_lazy_remote.go | 121 +++- weed/filer/filer_lazy_remote_test.go | 19 +- weed/filer/meta_aggregator_replay_test.go | 654 ++++++++++++++++++ weed/filer/meta_replay.go | 100 ++- weed/filer/mongodb/mongodb_store_kv.go | 2 +- weed/filer/reader_cache_memory_test.go | 7 + .../azure/azure_storage_client.go | 74 +- .../azure/azure_storage_client_test.go | 2 +- weed/remote_storage/gcs/gcs_storage_client.go | 46 +- weed/remote_storage/remote_storage.go | 37 +- weed/remote_storage/s3/s3_storage_client.go | 69 +- .../s3/s3_storage_client_test.go | 12 +- weed/util/lock_table.go | 33 + weed/util/retry.go | 9 + 21 files changed, 1313 insertions(+), 119 deletions(-) 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) ||