From 9620e7ebfe907316c5ada0ad30fa74a3e355d3d2 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 25 Oct 2026 11:37:06 -0700 Subject: [PATCH] filer: auto-enable per-path commit records when a peer has its own store (#12062) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * filer: enable per-path commit records only when a peer has its own store Since seaweedfs#11698 every metadata change writes a per-path commit record into the filer store (an MpTs key, plus an MpDl key on deletes), and the records stay after the path is deleted. They order the events a filer replays from peer filers that keep their own store, since replay only runs when peerSignature != f.Signature. Filers that all share one store (one Redis, MySQL, TiKV, ...) never replay each other's events, so the records are never read, yet each change still costs one or two extra store writes and the key count grows with every path ever changed — unbounded growth for in-memory stores. This turns the records off by default and latches them on the first time a peer reports a different store signature — no option needed. The latch is persisted in the store (MpRep) so it survives restarts and converges filers sharing a store even before they discover the peer themselves. While off, a filer keeps an MpOff marker so enabling later bumps the records' generation (MpGen, folded into each record key) past the unrecorded window: a record older than a change made while off cannot let an older replayed peer event win. Ordering for unrecorded changes falls back to the entry's mtime, the same as entries written before the records existed. Supersedes the manual option approach in seaweedfs#12060. * filer: keep delete tombstones while records off, harden enable path Review on the first commit: - A delete committed while the records were off left no entry and no usable tombstone: after enabling, an older replayed create could resurrect the path. Deletes and rename sources are now always recorded under the generation-0 MpDl key — they are never fenced by a generation bump, and they cannot go stale because every delete is recorded. Creates/updates still write nothing while off. - bumpPathCommitGen could regress MpGen under concurrent bumpers on a shared store. It now loops: each pass writes only a generation higher than any it has seen, so concurrent bumps converge upward. - doSubscribeToOneFiler proceeded into replay when enablePathCommitRecords failed, running the whole session without ordering. It now aborts the subscribe so the reconnect retries with the records on. * filer: bind pending commit-record repairs to their generation A failed MpTs write kept in metaTsPending could land after a shared-store generation bump, writing a pre-bump timestamp as a current-generation record and overriding newer unrecorded state. Each pending change record now carries the generation it failed under: repairs write it only while that generation is current, and ordering reads ignore it otherwise. Pending delete records need no binding — MpDl keys are never fenced. * filer: never move the local records' generation backward The converge loop in bumpPathCommitGen seeded from zero and both it and syncPathCommitGen stored unconditionally: a stale writer could leave the stored generation below what this filer already adopted, and the next sync or bump then walked the local value back — orphaning records it wrote under the higher generation. The bump now seeds from the local generation and every store only moves it forward. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer: name a records generation after the off marker it consumes Two filers consuming the same off marker each counted one past the stored generation, so the second could store N+2 after the first stored N+1 and write records under a key the other could not read until the next check. The bump now takes the marker's own timestamp — filers racing on one marker land on one value — a marker older than the stored generation is cleaned as a leftover, and a stored value that already passed the bump is adopted instead of counted past. A newly enabling filer whose marker was already consumed names its own off window with the current time, so its unrecorded period is still fenced. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer: tell a fresh off marker from one the generation already consumed A marker whose timestamp reads at or below the stored generation was dropped as leftover, but a new off period can also read that old — the off filer's clock lagging, or a host clock moving back — and dropping it kept records from before that period live, so stale replay could win. The stored generation now names the marker it was bumped for: a marker the store still shows is consumed unless it is that one, however old it reads, and only the consumed marker itself is cleared without a bump. The generation trails in the value, so a whole-slice read still yields it on stores written before the pair was stored. Enabling fences the filer's own off window with the current time instead of a marker a sharer may already have consumed. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer: cover racing off-marker consumers settling on one generation * filer: adopt the stored generation, not the consumed marker it names A filer syncing with no off marker present read the value's first field — the consumed marker's timestamp — so a restart after a lagging-clock off period under-counted the generation and kept records the bump dropped. * filer: re-check the off marker before consuming it A sync whose marker read ran behind a later off period's consume cycle still bumped for its stale read: the stored generation already covered a different marker, so the bump overwrote the consumed name, counted the generation past records a newer consumption meant to keep, and cleared whatever marker was live by then. Each pass now re-reads the marker and only bumps or clears while it is still the one the pass saw. * filer: keep a failed generation read from resetting the records' generation The final generation read ignored errors and then stored whatever gen held — possibly zero — so a transient store failure could hide records the current generation wrote. Unexpected read errors now return before the local generation changes. Also cover a sync whose marker read runs behind another consumer's delete: it must adopt the stored generation, not bump for the stale read. Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer: fail a bump that cannot verify the stored generation A generation read that failed mid-bump was treated as absent: the write could land below a generation a racing bumper had stored, or re-store a marker that bumper had already consumed. Read failures now abort the pass like every other store error on the path. Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer: cover generation reads that fail instead of returning a value Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer: carry the commit timestamp on the entry itself The per-path MpTs record only exists because a commit timestamp had nowhere else to live. Stamping commit_ts_ns on the entry before its store write - and reusing that stamp as the emitted event's TsNs - makes the entry its own commit record: replay and bootstrap traversal order on it with zero extra writes for creates and updates. Deletes still need a durable key since a removed entry cannot carry a timestamp, so MpDl tombstones stay (one write per delete, growth bounded by delete history). Entries predating the field fall back to mtime, matching what peers and old records already assume. This replaces the MpTs/MpOff/MpGen/MpRep machinery entirely: shared- store filers write nothing extra, and dedicated-store replication keeps last-writer-wins ordering plus tombstone coverage for deletes and ancestor drops. * filer: keep a directory update replay from tombstoning its subtree A same-path update wrote a delete record before the delete-and-reinsert, so a directory metadata change read as a subtree delete: a later peer event for a child older than the update was then skipped even though nobody deleted the children, leaving files missing on this filer. Only file updates take the tombstone now — the re-inserted directory entry's own commit stamp orders the path itself. * filer: order unstamped entries by their recorded commit before mtime Entries written before stamping existed have no CommitTsNs, so the ordering check fell back to the client-settable mtime and dropped the MpTs record those versions wrote: an entry committed at event time 300 with mtime 100 lost to a delayed peer event at 200 after upgrade. Read the legacy record for unstamped entries in localCommitTsNs, CommitTsNs, and the traverse stamp (which also stopped falling back to mtime, regressing entry_ts_ns to 0). * filer: release queued rename event stamps when the rename fails Each moved entry carried an in-flight stamp its queued event would adopt at emit time. When the move succeeded but a later child move or the commit failed, the handler rolled back without emitting the queued events or releasing their stamps, pinning the delivered and flushed watermarks below the aborted rename forever. Release every queued stamp on both failure exits in both rename handlers. * filer: apply a directory update replay with an in-place upsert Deleting the stored entry before inserting its replacement left a window where the directory had no entry and no ordering record — a tombstone cannot stand in for it without reading as a subtree delete that suppresses the children's own records. A stale peer event accepted in that gap could replay an older update or delete and sweep the directory's children. Upsert the replacement instead: the entry keeps its recorded commit order across the whole update, and a directory has no chunks or hard links for a delete to release. * filer: update an existing directory entry on replay instead of upserting InsertEntry is not an upsert on SQL stores configured without upsert support — PostgreSQL with enableUpsert=false issues a plain INSERT and its duplicate-key error does not match the MySQL-only fallback, so a replay updating a present directory failed on every retry. Pick UpdateEntry when the entry exists and InsertEntry when it is absent; UpdateEntry touches no row that is not there. * filer: detect a shared-store delete racing a replayed directory update UpdateEntry reports success even when the row was deleted between the find and the update — the UPDATE changes zero rows and the replayed change is dropped while the subscription offset advances. Re-read the entry after the update and retry the write, re-checking the recorded commit order each pass so a newer peer change wins instead of being resurrected over. * filer: declare commit_ts_ns in field-number order Field 6 sits between extended and hard_link_id; the declaration now reads in the same order. Wire format unchanged. Also syncs the Java client's checked-in copy of filer.proto, which the pb Makefile copies on gen. * filer: option to skip delete tombstones entirely filer.options.deleteTombstones=false stops writing MpDl records on deletes - a single filer or shared-store cluster never replays peer events, so the tombstones are dead weight. The tradeoff is explicit: if a dedicated-store peer does replicate, a stale event can resurrect a deleted path. * filer: drop delete tombstones entirely Deletes no longer write MpDl records, and the pending-repair machinery they needed is gone with them: per-mutation metadata overhead is now zero for every store topology. Ordering information lives only on the entry itself via commit_ts_ns. The accepted tradeoff: a dedicated-store peer replaying an event older than a delete can resurrect the path, and a kept directory no longer fences stale child creates. Recorded tombstones and MpTs records left by older builds are still read, so upgrades keep the protections they already have. * filer: stop reading legacy per-path records too Drops the MpTs/MpDl read path along with the writes: meta_path_ts.go keeps only the entry-field readers (CommitTsNs, EntryCommitTsNs, localCommitTsNs). Entries written before stamping existed fall back to mtime, and tombstones or commit records older builds left behind no longer fence replayed events. * filer: trim commit timestamp comments --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- other/java/client/src/main/proto/filer.proto | 19 ++ weed/filer/entry.go | 7 + weed/filer/filer.go | 35 +- weed/filer/filer_lazy_remote_listing.go | 4 + weed/filer/filer_notify.go | 72 +++- weed/filer/meta_aggregator.go | 15 +- weed/filer/meta_aggregator_replay_test.go | 298 ++++------------- weed/filer/meta_path_ts.go | 314 ++---------------- weed/filer/meta_path_ts_records_test.go | 91 +++++ weed/filer/meta_replay.go | 72 +++- weed/pb/filer.proto | 4 + weed/pb/filer_pb/filer.pb.go | 39 ++- weed/pb/filer_pb/filer_vtproto.pb.go | 27 ++ weed/server/filer_grpc_server.go | 8 + weed/server/filer_grpc_server_remote.go | 2 + weed/server/filer_grpc_server_rename.go | 26 +- weed/server/filer_grpc_server_rename_test.go | 27 ++ .../server/filer_grpc_server_traverse_meta.go | 2 +- 18 files changed, 489 insertions(+), 573 deletions(-) create mode 100644 weed/filer/meta_path_ts_records_test.go diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto index 93e7e3ed32..42f772a57a 100644 --- a/other/java/client/src/main/proto/filer.proto +++ b/other/java/client/src/main/proto/filer.proto @@ -166,6 +166,10 @@ message Entry { repeated FileChunk chunks = 3; FuseAttributes attributes = 4; map extended = 5; + // filer-local commit timestamp of the version stored here: the meta log + // event's ts_ns for the mutation that produced this entry, or 0 for + // entries written before the field existed + int64 commit_ts_ns = 6; bytes hard_link_id = 7; int32 hard_link_counter = 8; // only exists in hard link meta data bytes content = 9; // if not empty, the file content @@ -281,6 +285,7 @@ message WriteCondition { IF_EXTENDED_NOT_EQUAL = 7; // fail if present and extended[ext_key] == ext_value IF_EXTENDED_TIME_ELAPSED = 8; // fail if present and extended[ext_key] (unix seconds) is in the future IF_CHUNKS_EQUAL = 9; // fail unless the stored chunk fid multiset equals fids (absent entry = no chunks) + IF_ENTRY_EQUAL = 10; // fail unless the stored entry equals expected_entry } // Clause is one primitive comparison. IF_ETAG_MATCH holds when the current // entry's ETag equals any value in etags; IF_ETAG_NOT_MATCH holds when it @@ -309,6 +314,7 @@ message WriteCondition { string gate_key = 7; // IF_EXTENDED_TIME_ELAPSED: only enforce when extended[gate_key] == gate_value string gate_value = 8; // gate value (e.g. retention mode COMPLIANCE for governance bypass) repeated string fids = 9; // chunk fid strings for IF_CHUNKS_EQUAL + Entry expected_entry = 10; // whole-entry comparison for IF_ENTRY_EQUAL } repeated Clause clauses = 1; // all must hold (logical AND) } @@ -323,6 +329,7 @@ enum FilerError { EXISTING_IS_FILE = 4; // cannot overwrite file with directory ENTRY_ALREADY_EXISTS = 5; // O_EXCL and entry already exists PRECONDITION_FAILED = 6; // WriteCondition not satisfied + READ_ONLY = 7; // path matches a read-only storage rule (e.g. bucket over quota) } // ObjectMutation is one entry-level change applied by ObjectTransaction. All @@ -351,6 +358,7 @@ message ObjectMutation { bytes content = 11; // PATCH_EXTENDED: new Entry.content when set_content bool touch_mtime = 12; // PATCH_EXTENDED: set the entry's Mtime to now (e.g. a metadata-replace copy) bool remove_empty_parent = 13; // DELETE: also remove the parent directory when the delete leaves it empty (best-effort) + bool keep_remote_object = 14; // DELETE: leave the object on the mounted remote storage } // Recompute re-derives a pointer entry (directory/name on the mutation) from the @@ -388,6 +396,8 @@ message ObjectTransactionRequest { string condition_key = 6; // if set, evaluate the condition against this entry instead of lock_key (still locking lock_key) string route_key = 7; // ring key identifying the owner filer; a non-owner forwards the whole transaction to it bool is_moved = 8; // set on a forwarded transaction so the receiver applies it locally instead of forwarding again + string request_id = 9; // idempotency token: the owner records the outcome and replays it on a retried request instead of re-applying + string condition_alt_key = 10; // if the condition's entry is absent, evaluate it against this entry instead (e.g. null slot, else latest version) } message ObjectTransactionResponse { @@ -477,6 +487,7 @@ message UpdateEntryRequest { // the write, under the filer's per-path lock. The caller must route the // key's writes to this entry's owner filer for the check to be authoritative. WriteCondition condition = 6; + bool is_moved = 7; // set on a forwarded request so the owner applies it locally } message UpdateEntryResponse { SubscribeMetadataResponse metadata_event = 1; @@ -514,6 +525,7 @@ message DeleteEntryRequest { bool is_from_other_cluster = 7; repeated int32 signatures = 8; int64 if_not_modified_after = 9; + bool keep_remote_object = 10; } message DeleteEntryResponse { @@ -573,6 +585,9 @@ message AssignVolumeResponse { // fsync is the storage rule's fsync decision for the assigned path, so the // client can carry it onto the volume server upload request. bool fsync = 11; + // error_code classifies error for callers that must branch on it; error + // keeps the human-readable text for clients that predate the field. + FilerError error_code = 12; } message LookupVolumeRequest { @@ -709,6 +724,10 @@ message TraverseBfsMetadataRequest { message TraverseBfsMetadataResponse { string directory = 1; Entry entry = 2; + // The sender's per-path commit timestamp (see meta_path_ts.go): the newest + // event time this filer committed for the path, 0 when unrecorded so + // subscribers fall back to the entry's mtime. + int64 entry_ts_ns = 3; } message LogEntry { diff --git a/weed/filer/entry.go b/weed/filer/entry.go index 796674e062..0813b75f1c 100644 --- a/weed/filer/entry.go +++ b/weed/filer/entry.go @@ -48,6 +48,9 @@ type Entry struct { Remote *filer_pb.RemoteEntry Quota int64 WORMEnforcedAtTsNs int64 + // CommitTsNs is the meta log event timestamp of the mutation that + // produced this version. Zero for pre-field versions; they order by mtime. + CommitTsNs int64 } func (entry *Entry) Size() uint64 { @@ -76,6 +79,8 @@ func (entry *Entry) ShallowClone() *Entry { newEntry.Content = entry.Content newEntry.Remote = entry.Remote newEntry.Quota = entry.Quota + newEntry.WORMEnforcedAtTsNs = entry.WORMEnforcedAtTsNs + newEntry.CommitTsNs = entry.CommitTsNs return newEntry } @@ -109,6 +114,7 @@ func (entry *Entry) ToExistingProtoEntry(message *filer_pb.Entry) { message.RemoteEntry = entry.Remote message.Quota = entry.Quota message.WormEnforcedAtTsNs = entry.WORMEnforcedAtTsNs + message.CommitTsNs = entry.CommitTsNs } func FromPbEntryToExistingEntry(message *filer_pb.Entry, fsEntry *Entry) { @@ -122,6 +128,7 @@ func FromPbEntryToExistingEntry(message *filer_pb.Entry, fsEntry *Entry) { fsEntry.Quota = message.Quota fsEntry.FileSize = FileSize(message) fsEntry.WORMEnforcedAtTsNs = message.WormEnforcedAtTsNs + fsEntry.CommitTsNs = message.CommitTsNs } func (entry *Entry) ToProtoFullEntry() *filer_pb.FullEntry { diff --git a/weed/filer/filer.go b/weed/filer/filer.go index 59e9bfc714..0330ec2395 100644 --- a/weed/filer/filer.go +++ b/weed/filer/filer.go @@ -115,12 +115,6 @@ type Filer struct { // overwriting the unread ledger with a partial set. deletionLedgerBlocked atomic.Bool deletionLedgerFlush chan struct{} - // metaTsPending holds per-path commit records whose KV write failed, - // keyed by path; ordering reads consult it until the retry lands them - // (see meta_path_ts.go). metaTsRepairing marks paths with a live - // repair goroutine so failures neither strand a record nor pile on. - metaTsPending sync.Map - metaTsRepairing sync.Map // flushCtx is the context metadata-log flushes append under. It stays // live for the process lifetime and is cancelled once during Shutdown, // bounding every retry - in-flight and queued alike - to one deadline. @@ -483,8 +477,20 @@ func (f *Filer) CreateEntry(ctx context.Context, entry *Entry, existing *Entry, } */ + emitted := false + stamped := false + // On abort, release the stamp this call made — one the entry already + // carried is a concurrent mutation's and stays for its own event. + defer func() { + if !emitted && stamped { + f.releaseEntryCommitTsNs(entry.CommitTsNs) + } + }() + if oldEntry == nil { f.ensureEntryInode(entry) + f.stampEntryCommitTsNs(entry) + stamped = true glog.V(4).InfofCtx(ctx, "InsertEntry %s: new entry: %v", entry.FullPath, entry.Name()) if err := f.Store.InsertEntry(ctx, entry); err != nil { @@ -517,6 +523,8 @@ func (f *Filer) CreateEntry(ctx context.Context, entry *Entry, existing *Entry, return fmt.Errorf("%s: %w", entry.FullPath, filer_pb.ErrEntryAlreadyExists) } glog.V(4).InfofCtx(ctx, "UpdateEntry %s: old entry: %v", entry.FullPath, oldEntry.Name()) + f.stampEntryCommitTsNs(entry) + stamped = true if err := f.UpdateEntry(ctx, oldEntry, entry, isFromOtherCluster); err != nil { if errors.Is(err, filer_pb.ErrExistingIsDirectory) || errors.Is(err, filer_pb.ErrExistingIsFile) { glog.V(2).InfofCtx(ctx, "update entry %s: %v", entry.FullPath, err) @@ -528,6 +536,7 @@ func (f *Filer) CreateEntry(ctx context.Context, entry *Entry, existing *Entry, } f.NotifyUpdateEvent(ctx, oldEntry, entry, true, isFromOtherCluster, signatures) + emitted = true f.deleteChunksIfNotNew(ctx, oldEntry, entry) @@ -592,16 +601,18 @@ func (f *Filer) ensureParentDirectoryEntry(ctx context.Context, entry *Entry, di } glog.V(2).InfofCtx(ctx, "create directory: %s %v", dirPath, dirEntry.Mode) + f.stampEntryCommitTsNs(dirEntry) mkdirErr := f.Store.InsertEntry(ctx, dirEntry) if mkdirErr != nil { + f.releaseEntryCommitTsNs(dirEntry.CommitTsNs) if fEntry, err := f.FindEntry(ctx, util.FullPath(dirPath)); err == filer_pb.ErrNotFound || fEntry == nil { glog.V(3).InfofCtx(ctx, "mkdir %s: %v", dirPath, mkdirErr) return fmt.Errorf("mkdir %s: %v", dirPath, mkdirErr) } + } else if !strings.HasPrefix("/"+util.Join(dirParts[:]...), SystemLogDir) { + f.NotifyUpdateEvent(ctx, nil, dirEntry, false, isFromOtherCluster, nil) } else { - if !strings.HasPrefix("/"+util.Join(dirParts[:]...), SystemLogDir) { - f.NotifyUpdateEvent(ctx, nil, dirEntry, false, isFromOtherCluster, nil) - } + f.releaseEntryCommitTsNs(dirEntry.CommitTsNs) } } else if !dirEntry.IsDirectory() { @@ -621,7 +632,9 @@ func (f *Filer) ensureParentDirectoryEntry(ctx context.Context, entry *Entry, di dirEntry.Extended = make(map[string][]byte) } dirEntry.Extended[s3_constants.SeaweedFSPrefixObject] = []byte("true") + f.stampEntryCommitTsNs(dirEntry) if updateErr := f.Store.UpdateEntry(ctx, dirEntry); updateErr != nil { + f.releaseEntryCommitTsNs(dirEntry.CommitTsNs) return fmt.Errorf("promote %s to directory: %v", dirPath, updateErr) } f.NotifyUpdateEvent(ctx, nil, dirEntry, false, isFromOtherCluster, nil) @@ -673,7 +686,9 @@ func (f *Filer) EnsureDirectoryEntry(ctx context.Context, dirPath util.FullPath, narrowed := existing.ShallowClone() narrowed.Mode = existing.Mode&^restorableModeBits | kept glog.V(1).InfofCtx(ctx, "restore directory %s: narrowing %v to %v", dirPath, existing.Mode, narrowed.Mode) + f.stampEntryCommitTsNs(narrowed) if err := f.UpdateEntry(ctx, existing, narrowed, false); err != nil { + f.releaseEntryCommitTsNs(narrowed.CommitTsNs) return err } f.NotifyUpdateEvent(ctx, existing, narrowed, false, false, nil) @@ -698,7 +713,9 @@ func (f *Filer) EnsureDirectoryEntry(ctx context.Context, dirPath util.FullPath, UserName: attrs.UserName, GroupNames: attrs.GroupNames, }} f.ensureEntryInode(dirEntry) + f.stampEntryCommitTsNs(dirEntry) if err := f.Store.InsertEntry(ctx, dirEntry); err != nil { + f.releaseEntryCommitTsNs(dirEntry.CommitTsNs) return fmt.Errorf("restore directory %s: %v", dirPath, err) } f.NotifyUpdateEvent(ctx, nil, dirEntry, false, false, nil) diff --git a/weed/filer/filer_lazy_remote_listing.go b/weed/filer/filer_lazy_remote_listing.go index 52b43981fe..7b5fa7620b 100644 --- a/weed/filer/filer_lazy_remote_listing.go +++ b/weed/filer/filer_lazy_remote_listing.go @@ -136,9 +136,11 @@ func (f *Filer) maybeLazyListFromRemote(ctx context.Context, p util.FullPath) { existingEntry.Attr.FileSize = uint64(remoteEntry.RemoteSize) existingEntry.Extended = MergeRemoteContentEncoding(remoteEntry, existingEntry.Extended) } + f.stampEntryCommitTsNs(existingEntry) if saveErr := f.Store.UpdateEntry(persistCtx, existingEntry); saveErr != nil { glog.Warningf("maybeLazyListFromRemote: update %s: %v", childPath, saveErr) } + f.releaseEntryCommitTsNs(existingEntry.CommitTsNs) } else { // New entry not yet in local store var entry *Entry @@ -228,7 +230,9 @@ func (f *Filer) updateDirectoryListingSyncedAt(ctx context.Context, p util.FullP dirEntry.Extended = make(map[string][]byte) } dirEntry.Extended[xattrRemoteListingSyncedAt] = []byte(fmt.Sprintf("%d", syncTime.Unix())) + f.stampEntryCommitTsNs(dirEntry) if saveErr := f.Store.UpdateEntry(ctx, dirEntry); saveErr != nil { glog.Warningf("maybeLazyListFromRemote: update synced_at for %s: %v", p, saveErr) } + f.releaseEntryCommitTsNs(dirEntry.CommitTsNs) } diff --git a/weed/filer/filer_notify.go b/weed/filer/filer_notify.go index 8e23cef44a..6161a9aa0b 100644 --- a/weed/filer/filer_notify.go +++ b/weed/filer/filer_notify.go @@ -29,6 +29,8 @@ func (f *Filer) NotifyUpdateEvent(ctx context.Context, oldEntry, newEntry *Entry func (f *Filer) notifyUpdateEvent(ctx context.Context, oldEntry, newEntry *Entry, deleteChunks, isFromOtherCluster bool, signatures []int32) *filer_pb.SubscribeMetadataResponse { if metadataEventsSuppressed(ctx) { + // The entry's stamp stays in flight for the covering event the + // caller emits itself (the rename handler does). return nil } @@ -44,6 +46,10 @@ func (f *Filer) notifyUpdateEvent(ctx context.Context, oldEntry, newEntry *Entry // println("fullpath:", fullpath) if strings.HasPrefix(fullpath, SystemLogDir) { + // No event covers this write: release its stamp. + if newEntry != nil { + f.metaLogInflight.release(newEntry.CommitTsNs) + } return nil } foundSelf := false @@ -56,7 +62,7 @@ func (f *Filer) notifyUpdateEvent(ctx context.Context, oldEntry, newEntry *Entry signatures = append(signatures, f.Signature) } - event := f.newMetadataEvent(oldEntry, newEntry, deleteChunks, isFromOtherCluster, signatures) + event := f.newMetadataEvent(oldEntry, newEntry, deleteChunks, isFromOtherCluster, signatures, f.stampOf(newEntry)) // Clear the stamp after the buffer append below - deliberately also on // append failure (see the metaLogInflight comment). defer f.metaLogInflight.done(event.TsNs) @@ -71,10 +77,6 @@ func (f *Filer) notifyUpdateEvent(ctx context.Context, oldEntry, newEntry *Entry } f.logMetaEvent(ctx, event) - // Record the commit timestamp so a peer replay ordered by the entry lock - // this mutation may still be under sees the local change, including as a - // tombstone after the entry itself is gone. - f.noteCommittedEvent(event) if sink := metadataEventSinkFromContext(ctx); sink != nil { sink.Record(event) } @@ -134,6 +136,28 @@ func (t *metaLogInflight) done(tsNs int64) { } } +// inFlight reports whether tsNs is a pre-commit stamp still awaiting its +// buffer append — how an event adopts the stamp an entry carries. +func (t *metaLogInflight) inFlight(tsNs int64) bool { + t.Lock() + defer t.Unlock() + return t.stamped[tsNs] > 0 +} + +// release drops a stamp whose event will never be appended. +func (t *metaLogInflight) release(tsNs int64) { + if tsNs == 0 { + return + } + t.Lock() + defer t.Unlock() + if t.stamped[tsNs] <= 1 { + delete(t.stamped, tsNs) + } else { + t.stamped[tsNs]-- + } +} + // minTsNs returns the oldest in-flight stamp, or 0 when nothing is in flight. func (t *metaLogInflight) minTsNs() int64 { t.Lock() @@ -197,7 +221,41 @@ func (f *Filer) DoneMetaLogInflightForTest(tsNs int64) { f.metaLogInflight.done(tsNs) } -func (f *Filer) newMetadataEvent(oldEntry, newEntry *Entry, deleteChunks, isFromOtherCluster bool, signatures []int32) *filer_pb.SubscribeMetadataResponse { +// stampEntryCommitTsNs assigns the mutation's event timestamp to the entry +// pre-write so the emitted event and stored version carry the same commit +// time. The stamp stays in flight until the event appends; a write no event +// covers must release it via releaseEntryCommitTsNs. +func (f *Filer) stampEntryCommitTsNs(entry *Entry) int64 { + tsNs := f.metaLogInflight.stamp() + entry.CommitTsNs = tsNs + return tsNs +} + +// releaseEntryCommitTsNs drops a stamp no event will adopt. +func (f *Filer) releaseEntryCommitTsNs(tsNs int64) { + f.metaLogInflight.release(tsNs) +} + +// StampEntryCommitTsNs/ReleaseEntryCommitTsNs expose the lifecycle to the +// filer server's mutation paths. +func (f *Filer) StampEntryCommitTsNs(entry *Entry) int64 { + return f.stampEntryCommitTsNs(entry) +} + +func (f *Filer) ReleaseEntryCommitTsNs(tsNs int64) { + f.metaLogInflight.release(tsNs) +} + +// stampOf returns the event's timestamp: the entry's in-flight stamp, or a +// fresh one for events with no newly stamped entry (deletes). +func (f *Filer) stampOf(newEntry *Entry) int64 { + if newEntry != nil && f.metaLogInflight.inFlight(newEntry.CommitTsNs) { + return newEntry.CommitTsNs + } + return f.metaLogInflight.stamp() +} + +func (f *Filer) newMetadataEvent(oldEntry, newEntry *Entry, deleteChunks, isFromOtherCluster bool, signatures []int32, tsNs int64) *filer_pb.SubscribeMetadataResponse { if oldEntry == nil && newEntry == nil { return nil } @@ -224,7 +282,7 @@ func (f *Filer) newMetadataEvent(oldEntry, newEntry *Entry, deleteChunks, isFrom Signatures: signatures, }, // In flight until appended to the local log buffer (see metaLogInflight). - TsNs: f.metaLogInflight.stamp(), + TsNs: tsNs, } } diff --git a/weed/filer/meta_aggregator.go b/weed/filer/meta_aggregator.go index 310874a2e2..c4f17d8734 100644 --- a/weed/filer/meta_aggregator.go +++ b/weed/filer/meta_aggregator.go @@ -754,10 +754,8 @@ func (ma *MetaAggregator) traversePeerMetadata(ctx context.Context, f *Filer, pe continue } entry := FromPbEntry(resp.Directory, resp.Entry) - // Order on the sender's per-path commit timestamp, not the - // entry's mtime — mtime is client-settable and does not track - // when the peer committed the change. Peers that predate the - // field send 0 and fall back to mtime. + // Order on the sender's commit timestamp, not client-settable + // mtime. Peers predating the field send 0: fall back to mtime. peerTsNs := resp.EntryTsNs if peerTsNs == 0 { peerTsNs = entry.Attr.Mtime.UnixNano() @@ -776,10 +774,9 @@ func (ma *MetaAggregator) traversePeerMetadata(ctx context.Context, f *Filer, pe } // applyTraversedEntry merges one bootstrap entry under the path's replay -// locks, keeping local state when its commit is newer than the sender's — -// most importantly a delete, whose tombstone must not let the bootstrap -// resurrect the path. An applied entry records the sender's commit time so a -// later peer replay orders against it rather than the entry's mtime. +// locks, keeping local state when its commit is newer than the sender's. +// An applied entry keeps the sender's commit time on its stamp so a later +// peer replay orders against it rather than the entry's mtime. func (f *Filer) applyTraversedEntry(ctx context.Context, entry *Entry, peerTsNs int64) error { unlock := f.lockReplayPaths(entry.FullPath, "") defer unlock() @@ -805,6 +802,7 @@ func (f *Filer) applyTraversedEntry(ctx context.Context, entry *Entry, peerTsNs return nil } } + entry.CommitTsNs = peerTsNs if insertErr := f.Store.InsertEntry(ctx, entry); insertErr != nil { // Entry already exists (root dir, or partial previous bootstrap); // the lock kept the check above current, so the peer commit is newer. @@ -815,7 +813,6 @@ func (f *Filer) applyTraversedEntry(ctx context.Context, entry *Entry, peerTsNs return fmt.Errorf("update entry %s: %w", entry.FullPath, updateErr) } } - f.notePathChange(entry.FullPath, peerTsNs, false) return nil } diff --git a/weed/filer/meta_aggregator_replay_test.go b/weed/filer/meta_aggregator_replay_test.go index efbc9968f2..3d2966e321 100644 --- a/weed/filer/meta_aggregator_replay_test.go +++ b/weed/filer/meta_aggregator_replay_test.go @@ -18,6 +18,7 @@ import ( ) // newReplayTestFiler builds the minimum Filer Replay needs over a stub store. + func newReplayTestFiler(store FilerStore) *Filer { return &Filer{ Store: NewFilerStoreWrapper(store), @@ -259,58 +260,6 @@ func TestReplayMoveChecksEachPathSeparately(t *testing.T) { } } -// A local delete leaves a recorded tombstone, so a replayed create older than -// the delete must not resurrect the path even though no entry remains to -// compare against. -func TestReplaySkipsCreateOlderThanLocalDelete(t *testing.T) { - store := newStubFilerStore() - f := newReplayTestFiler(store) - now := time.Now() - path := util.NewFullPath("/dir", "file") - - // The delete removed the entry; its commit record survives. - f.notePathChange(path, now.UnixNano(), true) - - createEvent := &filer_pb.SubscribeMetadataResponse{ - Directory: "/dir", - EventNotification: &filer_pb.EventNotification{ - NewEntry: &filer_pb.Entry{Name: "file", Content: []byte("stale resurrect")}, - }, - TsNs: now.Add(-time.Minute).UnixNano(), - } - if err := f.Replay(createEvent); err != nil { - t.Fatalf("Replay: %v", err) - } - if _, err := store.FindEntry(context.Background(), path); err == nil { - t.Fatal("stale replayed create resurrected a locally deleted entry") - } -} - -// A deleted directory covers its children: a bulk drop emits no per-child -// events, so the tombstone on the ancestor is what a replayed child create -// must check. -func TestReplaySkipsCreateUnderDeletedAncestor(t *testing.T) { - store := newStubFilerStore() - f := newReplayTestFiler(store) - now := time.Now() - - f.notePathChange("/dir", now.UnixNano(), true) - - createEvent := &filer_pb.SubscribeMetadataResponse{ - Directory: "/dir", - EventNotification: &filer_pb.EventNotification{ - NewEntry: &filer_pb.Entry{Name: "child", Content: []byte("stale")}, - }, - TsNs: now.Add(-time.Minute).UnixNano(), - } - if err := f.Replay(createEvent); err != nil { - t.Fatalf("Replay: %v", err) - } - if _, err := store.FindEntry(context.Background(), "/dir/child"); err == nil { - t.Fatal("stale replayed create resurrected a child of a deleted directory") - } -} - // A recorded commit timestamp orders replay, not the entry's mtime: a rename // keeps the source's mtime on the destination, and a mounted client can set // mtime arbitrarily, so an entry whose mtime postdates the event must still @@ -324,9 +273,9 @@ func TestReplayAppliesWhenCommitRecordIsOlderThanFutureMtime(t *testing.T) { FullPath: path, Attr: Attr{Mtime: now.Add(time.Hour)}, // future mtime, e.g. client-set or kept from a rename source Content: []byte("old destination"), + // The last committed change predates the event. + CommitTsNs: now.Add(-time.Minute).UnixNano(), } - // The last committed change predates the event. - f.notePathChange(path, now.Add(-time.Minute).UnixNano(), false) event := &filer_pb.SubscribeMetadataResponse{ Directory: "/a", @@ -372,9 +321,9 @@ func TestReplaySerializesWithEntryLock(t *testing.T) { store.entries[string(path)] = &Entry{FullPath: path, Attr: Attr{Mtime: now}} // A local mutation holding the path lock; a replay started behind it - // sees the recorded commit and skips. + // sees the entry's commit stamp and skips. held := f.EntryLockTable.AcquireLock("test local write", path, util.ExclusiveLock) - f.notePathChange(path, now.UnixNano(), false) + store.entries[string(path)].CommitTsNs = now.UnixNano() done := make(chan error, 1) go func() { @@ -421,18 +370,6 @@ func TestReplicateMetadataChangeAbortsOnStop(t *testing.T) { } } -// Commit-record keys must stay small: SQL KV stores keep the key suffix in a -// VARCHAR(766) column, so a raw path over ~578 bytes silently drops the -// record (the insert fails and the fallback update touches no row). -func TestMetaPathKeyFixedSizeForLongPaths(t *testing.T) { - path := util.NewFullPath("/", string(make([]rune, 600))) - for _, p := range []util.FullPath{"/a", path} { - if got := len(metaPathKey(metaPathTsPrefix, p)); got != len(metaPathTsPrefix)+16 { - t.Fatalf("metaPathKey(%d-byte path) = %d bytes, want %d", len(p), got, len(metaPathTsPrefix)+16) - } - } -} - // Exact-path locking alone lets a replayed child create slip between a local // ancestor delete's tombstone check and the sweep: replay locks every // ancestor too, so it waits for the delete's lock and then sees the @@ -444,7 +381,6 @@ func TestReplaySerializesWithAncestorLock(t *testing.T) { now := time.Now() child := util.NewFullPath("/dir", "file") held := f.EntryLockTable.AcquireLock("test local delete", "/dir", util.ExclusiveLock) - f.notePathChange("/dir", now.UnixNano(), true) createEvent := &filer_pb.SubscribeMetadataResponse{ Directory: "/dir", @@ -466,32 +402,9 @@ func TestReplaySerializesWithAncestorLock(t *testing.T) { if err := <-done; err != nil { t.Fatalf("Replay: %v", err) } - if _, err := store.FindEntry(context.Background(), child); err == nil { - t.Fatal("stale replayed create resurrected a child of a locally deleted directory") - } -} - -// Bootstrap ordering uses the sender's commit timestamp: a local commit -// newer than the traversed entry keeps its state even when the peer entry's -// mtime is far in the future. -func TestBootstrapSkipsPeerEntryBehindLocalDelete(t *testing.T) { - store := newStubFilerStore() - f := newReplayTestFiler(store) - now := time.Now() - path := util.NewFullPath("/dir", "file") - - f.notePathChange(path, now.UnixNano(), true) - - entry := &Entry{ - FullPath: path, - Attr: Attr{Mtime: now.Add(time.Hour)}, // client-settable, must not win - Content: []byte("stale peer entry"), - } - if err := f.applyTraversedEntry(context.Background(), entry, now.Add(-time.Minute).UnixNano()); err != nil { - t.Fatalf("applyTraversedEntry: %v", err) - } - if _, err := store.FindEntry(context.Background(), path); err == nil { - t.Fatal("bootstrap resurrected a path newer local delete covered") + // No tombstone protects the deleted ancestor: the stale create lands. + if _, err := store.FindEntry(context.Background(), child); err != nil { + t.Fatalf("stale create applies once the lock releases: %v", err) } } @@ -555,11 +468,11 @@ func TestBootstrapUpdatesExistingByCommitTime(t *testing.T) { path := util.NewFullPath("/dir", "file") store.entries[string(path)] = &Entry{ - FullPath: path, - Attr: Attr{Mtime: now.Add(time.Hour)}, // future mtime, e.g. client-set - Content: []byte("old local write"), + FullPath: path, + Attr: Attr{Mtime: now.Add(time.Hour)}, // future mtime, e.g. client-set + Content: []byte("old local write"), + CommitTsNs: now.Add(-time.Minute).UnixNano(), } - f.notePathChange(path, now.Add(-time.Minute).UnixNano(), false) entry := &Entry{ FullPath: path, @@ -612,139 +525,23 @@ func TestBootstrapCopiesMissingEntryWithoutCommitTime(t *testing.T) { } } -// A delete whose commit record failed to persist still blocks a stale -// replayed create: the pending record stands in until the retry lands it. -func TestPendingDeleteRecordBlocksStaleReplay(t *testing.T) { - store := newStubFilerStore() - store.setKvPutErr(errors.New("kv store unreachable")) - f := newReplayTestFiler(store) - f.flushCtx, f.flushCancel = context.WithCancel(context.Background()) - defer f.flushCancel() - now := time.Now() - path := util.NewFullPath("/dir", "file") - - // The delete already committed; only its ordering record failed. - f.notePathChange(path, now.UnixNano(), true) - - createEvent := &filer_pb.SubscribeMetadataResponse{ - Directory: "/dir", - EventNotification: &filer_pb.EventNotification{ - NewEntry: &filer_pb.Entry{Name: "file", Content: []byte("stale resurrect")}, - }, - TsNs: now.Add(-time.Minute).UnixNano(), - } - if err := f.Replay(createEvent); err != nil { - t.Fatalf("Replay: %v", err) - } - if _, err := store.FindEntry(context.Background(), path); err == nil { - t.Fatal("stale create resurrected a path whose delete record is still pending") - } -} - -// The failed record write retries until it lands, then leaves the pending set. -func TestPendingCommitRecordRepairs(t *testing.T) { - store := newStubFilerStore() - f := newReplayTestFiler(store) - f.flushCtx, f.flushCancel = context.WithCancel(context.Background()) - defer f.flushCancel() - path := util.NewFullPath("/dir", "file") - - store.setKvPutErr(errors.New("kv store unreachable")) - f.notePathChange(path, 100, true) - if _, loaded := f.metaTsPending.Load(string(path)); !loaded { - t.Fatal("failed record write must stay pending") - } - store.setKvPutErr(nil) - deadline := time.Now().Add(5 * time.Second) - for time.Now().Before(deadline) { - if _, loaded := f.metaTsPending.Load(string(path)); !loaded { - ts, found, err := f.kvReadTsNs(context.Background(), metaPathKey(metaPathDelPrefix, path)) - if err != nil || !found || ts != 100 { - t.Fatalf("delete record after repair: ts=%v found=%v err=%v", ts, found, err) - } - return - } - time.Sleep(10 * time.Millisecond) - } - t.Fatal("pending record was not repaired") -} - -// CommitTsNs reports the recorded commit time over the entry mtime, and a -// still-pending record over both. -func TestCommitTsNsPrefersRecordAndPending(t *testing.T) { +// CommitTsNs reports the commit timestamp stamped on the entry, or the +// fallback when the entry predates stamping or is gone. +func TestCommitTsNsFromEntry(t *testing.T) { store := newStubFilerStore() f := newReplayTestFiler(store) path := util.NewFullPath("/dir", "file") if got, err := f.CommitTsNs(context.Background(), path, 7); err != nil || got != 7 { - t.Fatalf("no record: CommitTsNs = %d, %v; want fallback 7", got, err) + t.Fatalf("no entry: CommitTsNs = %d, %v; want fallback 7", got, err) } - f.notePathChange(path, 50, false) + store.entries[string(path)] = &Entry{FullPath: path, CommitTsNs: 50} if got, err := f.CommitTsNs(context.Background(), path, 7); err != nil || got != 50 { - t.Fatalf("recorded: CommitTsNs = %d, %v; want 50", got, err) + t.Fatalf("stamped: CommitTsNs = %d, %v; want 50", got, err) } - pending := &metaTsPendingRec{} - pending.changeTsNs.Store(80) - f.metaTsPending.Store(string(path), pending) - if got, err := f.CommitTsNs(context.Background(), path, 7); err != nil || got != 80 { - t.Fatalf("pending: CommitTsNs = %d, %v; want 80", got, err) - } -} - -// A successful write discharges what the pending record owed before it, so a -// repair pass can never put an older timestamp back over it. -func TestPendingCommitRecordDischargedByNewerWrite(t *testing.T) { - store := newStubFilerStore() - f := newReplayTestFiler(store) - f.flushCtx, f.flushCancel = context.WithCancel(context.Background()) - defer f.flushCancel() - path := util.NewFullPath("/dir", "file") - - store.setKvPutErr(errors.New("kv store unreachable")) - f.notePathChange(path, 100, false) - store.setKvPutErr(nil) - f.notePathChange(path, 200, false) - - deadline := time.Now().Add(5 * time.Second) - for time.Now().Before(deadline) { - if _, loaded := f.metaTsPending.Load(string(path)); !loaded { - break - } - time.Sleep(10 * time.Millisecond) - } - ts, found, err := f.kvReadTsNs(context.Background(), metaPathKey(metaPathTsPrefix, path)) - if err != nil || !found || ts != 200 { - t.Fatalf("MpTs after repair = %v, %v, %v; want 200", ts, found, err) - } -} - -// A failed delete record keeps its own time even when a later failed change -// merges into the pending record: the repair must not stamp the delete at -// the newer change's time. -func TestPendingDeleteRecordKeepsItsTime(t *testing.T) { - store := newStubFilerStore() - store.setKvPutErr(errors.New("kv store unreachable")) - f := newReplayTestFiler(store) - f.flushCtx, f.flushCancel = context.WithCancel(context.Background()) - defer f.flushCancel() - path := util.NewFullPath("/dir", "file") - - f.notePathChange(path, 100, true) - f.notePathChange(path, 200, false) - store.setKvPutErr(nil) - - deadline := time.Now().Add(5 * time.Second) - for time.Now().Before(deadline) { - if _, loaded := f.metaTsPending.Load(string(path)); !loaded { - break - } - time.Sleep(10 * time.Millisecond) - } - if ts, found, err := f.kvReadTsNs(context.Background(), metaPathKey(metaPathTsPrefix, path)); err != nil || !found || ts != 200 { - t.Fatalf("MpTs = %v, %v, %v; want 200", ts, found, err) - } - if ts, found, err := f.kvReadTsNs(context.Background(), metaPathKey(metaPathDelPrefix, path)); err != nil || !found || ts != 100 { - t.Fatalf("MpDl = %v, %v, %v; want 100", ts, found, err) + store.entries[string(path)].CommitTsNs = 0 + if got, err := f.CommitTsNs(context.Background(), path, 7); err != nil || got != 7 { + t.Fatalf("unstamped: CommitTsNs = %d, %v; want fallback 7", got, err) } } @@ -906,9 +703,10 @@ func TestReplayDirectoryDeleteKeepsNewerChildren(t *testing.T) { } } -// A directory kept for newer descendants still records the delete: a stale -// create replayed later for another child must be fenced by it. -func TestReplayKeptDirectoryDeleteFencesStaleCreates(t *testing.T) { +// A directory kept for newer descendants records no delete — tombstones are +// gone — so a stale create replayed later for another child resurrects +// under it. This documents the accepted tradeoff. +func TestReplayKeptDirectoryDeleteAllowsStaleCreates(t *testing.T) { store := newStubFilerStore() f := newReplayTestFiler(store) ctx := context.Background() @@ -952,8 +750,8 @@ func TestReplayKeptDirectoryDeleteFencesStaleCreates(t *testing.T) { if err := f.Replay(stale); err != nil { t.Fatalf("replay stale create: %v", err) } - if _, err := store.FindEntry(ctx, "/d/late.txt"); !errors.Is(err, filer_pb.ErrNotFound) { - t.Fatal("stale create resurrected a child under a directory the origin deleted") + if _, err := store.FindEntry(ctx, "/d/late.txt"); err != nil { + t.Fatalf("no tombstone exists, so the stale create lands: %v", err) } } @@ -1569,3 +1367,45 @@ func TestReplaySweepKeepsRemoteOnlyOwnerOfPath(t *testing.T) { t.Fatalf("the owned remote object must be kept, got %v", stub.deleteCalls) } } + +// A directory metadata update is not a subtree delete: replaying one must +// leave no tombstone, or a child event older than the update is suppressed +// even though nobody deleted the children. +func TestReplayDirUpdateDoesNotSuppressChildren(t *testing.T) { + store := newStubFilerStore() + f := newReplayTestFiler(store) + now := time.Now() + + store.entries["/dir"] = &Entry{ + FullPath: "/dir", + Attr: Attr{Mtime: now.Add(-time.Hour), Mode: os.ModeDir | 0755}, + } + update := &filer_pb.SubscribeMetadataResponse{ + Directory: "/", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "dir", IsDirectory: true}, + NewEntry: &filer_pb.Entry{Name: "dir", IsDirectory: true}, + }, + TsNs: now.UnixNano(), + } + if err := f.Replay(update); err != nil { + t.Fatalf("Replay dir update: %v", err) + } + if _, err := store.FindEntry(context.Background(), "/dir"); err != nil { + t.Fatalf("the updated directory must be back in place: %v", err) + } + + create := &filer_pb.SubscribeMetadataResponse{ + Directory: "/dir", + EventNotification: &filer_pb.EventNotification{ + NewEntry: &filer_pb.Entry{Name: "child", Content: []byte("data")}, + }, + TsNs: now.Add(-time.Minute).UnixNano(), + } + if err := f.Replay(create); err != nil { + t.Fatalf("Replay child create: %v", err) + } + if _, err := store.FindEntry(context.Background(), "/dir/child"); err != nil { + t.Fatalf("a child create older than the dir update must apply: %v", err) + } +} diff --git a/weed/filer/meta_path_ts.go b/weed/filer/meta_path_ts.go index 844f744d3a..69d7109268 100644 --- a/weed/filer/meta_path_ts.go +++ b/weed/filer/meta_path_ts.go @@ -2,323 +2,55 @@ package filer import ( "context" - "crypto/sha256" "errors" - "strings" - "sync" - "sync/atomic" - "time" - "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/util" ) -const ( - // metaPathTsPrefix keys per-path commit timestamps in the filer store KV: - // the newest event timestamp this filer committed or applied for the - // path. Written for every emitted mutation event and kept after the entry - // is deleted, so a replayed peer event can still be ordered against a - // change that left no entry behind. - metaPathTsPrefix = "MpTs" - // metaPathDelPrefix keys delete-only timestamps. A replayed event checks - // the path's ancestors too, so a directory delete also covers children - // swept without per-child events (a dropped bucket's whole-store delete, - // or a walk aborted partway). - metaPathDelPrefix = "MpDl" -) - -// metaPathKey hashes the path into a fixed-size key. SQL-backed KV stores -// split the key at byte 8 and store the remainder base64-encoded in a -// VARCHAR(766) name column: a raw path over ~578 bytes overflows it, the -// KvPut insert fails, and the fallback update touches no row yet returns no -// error — the commit record would never land and stale replay could restore -// the path. -func metaPathKey(prefix string, path util.FullPath) []byte { - sum := sha256.Sum256([]byte(path)) - key := make([]byte, len(prefix)+16) - copy(key, prefix) - copy(key[len(prefix):], sum[:16]) - return key -} - -// kvReadTsNs returns the timestamp recorded for key, or found=false when the -// key is absent or malformed. Any other store read failure is an error: a -// caller deciding whether a replayed change is stale must not guess. -func (f *Filer) kvReadTsNs(ctx context.Context, key []byte) (tsNs int64, found bool, err error) { - value, err := f.Store.KvGet(ctx, key) - if err != nil { - if errors.Is(err, ErrKvNotFound) { - return 0, false, nil - } - return 0, false, err - } - if len(value) < 8 { - return 0, false, nil - } - return int64(util.BytesToUint64(value)), true, nil -} - -// notePathChange persists tsNs as path's newest committed change. Local event -// timestamps are stamped monotonically (metaLogInflight.stamp), so a plain -// put cannot regress an earlier local record; a higher peer record it may -// overwrite was already newer under last-writer-wins ordering anyway. A -// delete additionally writes a delete record the ancestor lookup consults. -// -// The write uses a fresh context: callers notify after CommitTransaction, -// and SQL stores keep the already-committed *sql.Tx in the request context — -// a record written through it would fail and the path would lose ordering. -// -// A failed write is kept in metaTsPending and retried in the background: the -// mutation the record orders has already committed, so dropping it would let -// a stale replayed change overwrite or resurrect the path while the store is -// unreachable. Ordering reads consult the pending record until the retry -// lands it. -func (f *Filer) notePathChange(path util.FullPath, tsNs int64, deleted bool) { - if f.Store == nil { - return - } - ctx := context.Background() - key := string(path) - rec := f.pendingRec(key) - defer rec.mu.Unlock() - - value := make([]byte, 8) - util.Uint64toBytes(value, uint64(tsNs)) - if err := f.Store.KvPut(ctx, metaPathKey(metaPathTsPrefix, path), value); err != nil { - glog.WarningfCtx(ctx, "record meta change time %s: %v", path, err) - if tsNs > rec.changeTsNs.Load() { - rec.changeTsNs.Store(tsNs) - } - } else if rec.changeTsNs.Load() <= tsNs { - // This write covers every change record owed so far. - rec.changeTsNs.Store(0) - } - if deleted { - if err := f.Store.KvPut(ctx, metaPathKey(metaPathDelPrefix, path), value); err != nil { - glog.WarningfCtx(ctx, "record meta delete time %s: %v", path, err) - if tsNs > rec.deleteTsNs.Load() { - rec.deleteTsNs.Store(tsNs) - } - } else if rec.deleteTsNs.Load() <= tsNs { - rec.deleteTsNs.Store(0) - } - } - if rec.changeTsNs.Load() == 0 && rec.deleteTsNs.Load() == 0 { - f.metaTsPending.CompareAndDelete(key, rec) - return - } - if _, running := f.metaTsRepairing.LoadOrStore(key, struct{}{}); !running { - go f.repairMetaPathTs(path) - } -} - -// metaTsPendingRec is a commit record still on its way to the store, holding -// only work not yet written: changeTsNs is the newest committed change the -// MpTs record lacks and deleteTsNs the newest delete the MpDl record lacks — -// the two keep their own times so a later update cannot masquerade as the -// delete. Store writes and the bookkeeping they discharge share mu, so a -// repair pass can never write an older record over a newer one that already -// landed; readers take the atomic fields directly. -type metaTsPendingRec struct { - mu sync.Mutex - changeTsNs atomic.Int64 - deleteTsNs atomic.Int64 -} - -// pendingRec returns the live pending record for key, locked for mutation, -// recreating it when a repair already dropped the previous one. -func (f *Filer) pendingRec(key string) *metaTsPendingRec { - for { - cur, _ := f.metaTsPending.LoadOrStore(key, &metaTsPendingRec{}) - rec := cur.(*metaTsPendingRec) - rec.mu.Lock() - if live, ok := f.metaTsPending.Load(key); ok && live == cur { - return rec - } - rec.mu.Unlock() - } -} - -// repairMetaPathTs retries a pending commit record until it lands. Each pass -// writes only what the record still owes, so a merge by a later failure is -// picked up without another goroutine and a landed write is never rewritten. -// The retry is unbounded — the mutation is already committed and only the -// record preserves its ordering. -func (f *Filer) repairMetaPathTs(path util.FullPath) { - key := string(path) - ctx := f.flushCtx - if ctx == nil { - ctx = context.Background() - } - defer func() { - f.metaTsRepairing.Delete(key) - // A failure landing between the last pass and the flag release - // still needs a repair pass — unless the retry context is done. - if ctx.Err() != nil { - return - } - if _, ok := f.metaTsPending.Load(key); ok { - if _, running := f.metaTsRepairing.LoadOrStore(key, struct{}{}); !running { - go f.repairMetaPathTs(path) - } - } - }() - for { - if _, ok := f.metaTsPending.Load(key); !ok { - return - } - rec := f.pendingRec(key) - var tsErr, delErr error - if tsNs := rec.changeTsNs.Load(); tsNs > 0 { - value := make([]byte, 8) - util.Uint64toBytes(value, uint64(tsNs)) - if tsErr = f.Store.KvPut(ctx, metaPathKey(metaPathTsPrefix, path), value); tsErr == nil { - rec.changeTsNs.Store(0) - } - } - if tsNs := rec.deleteTsNs.Load(); tsNs > 0 { - value := make([]byte, 8) - util.Uint64toBytes(value, uint64(tsNs)) - if delErr = f.Store.KvPut(ctx, metaPathKey(metaPathDelPrefix, path), value); delErr == nil { - rec.deleteTsNs.Store(0) - } - } - done := rec.changeTsNs.Load() == 0 && rec.deleteTsNs.Load() == 0 - if done { - f.metaTsPending.CompareAndDelete(key, rec) - } - rec.mu.Unlock() - if !done { - glog.V(0).InfofCtx(ctx, "retry record meta change time %s: %v %v", path, tsErr, delErr) - select { - case <-ctx.Done(): - return - case <-time.After(time.Second): - } - } - } -} - -// pendingPathTsNs returns the newest pending commit timestamp for path and -// the newest pending delete timestamp. -func (f *Filer) pendingPathTsNs(path util.FullPath) (changeTsNs, deleteTsNs int64) { - cur, ok := f.metaTsPending.Load(string(path)) - if !ok { - return 0, 0 - } - rec := cur.(*metaTsPendingRec) - return rec.changeTsNs.Load(), rec.deleteTsNs.Load() -} - -// CommitTsNs returns the newest commit timestamp known for path — the -// recorded event time, or a still-pending one — and fallbackTsNs when none -// exists. Peers stamp traversal entries with it so a bootstrap subscriber -// orders each entry on the sender's commit clock rather than the entry's -// client-settable mtime. A store read failure is an error, not a fallback: -// the mtime it would mint could outdate real commits. +// CommitTsNs resolves the commit timestamp at path for traversal: the +// entry's stamp, else fallbackTsNs for pre-stamp or missing entries. func (f *Filer) CommitTsNs(ctx context.Context, path util.FullPath, fallbackTsNs int64) (int64, error) { if f.Store == nil { return fallbackTsNs, nil } - tsNs := fallbackTsNs - recTs, found, err := f.kvReadTsNs(ctx, metaPathKey(metaPathTsPrefix, path)) + entry, err := f.Store.FindEntry(ctx, path) if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + return fallbackTsNs, nil + } return 0, err } - if found { - tsNs = recTs + if entry == nil { + return fallbackTsNs, nil } - if pendTs, _ := f.pendingPathTsNs(path); pendTs > tsNs { - tsNs = pendTs - } - return tsNs, nil + return f.EntryCommitTsNs(ctx, entry, fallbackTsNs) } -// noteCommittedEvent records the event's commit timestamp for every path it -// touches: the source of a delete or rename, and the target of a create, -// update, or rename. It runs after the local mutation commits while the -// caller still holds the path's entry lock (for filer-server mutations), so -// a peer replay ordered by the same lock sees either the whole local change -// or none of it. -func (f *Filer) noteCommittedEvent(event *filer_pb.SubscribeMetadataResponse) { - message := event.EventNotification - if message == nil { - return - } - var source, target util.FullPath - if message.OldEntry != nil { - source = util.FullPath(filer_pb.MetadataEventSourceFullPath(event)) - } - if message.NewEntry != nil { - target = util.FullPath(filer_pb.MetadataEventTargetFullPath(event)) - } - // A delete or a rename removes the source path; an update reports the - // same path on both sides and counts once. - if source != "" && source != target { - f.notePathChange(source, event.TsNs, true) - } - if target != "" { - f.notePathChange(target, event.TsNs, false) +// EntryCommitTsNs returns the stamp on an already-loaded entry, else +// fallbackTsNs. +func (f *Filer) EntryCommitTsNs(ctx context.Context, entry *Entry, fallbackTsNs int64) (int64, error) { + if entry.CommitTsNs != 0 { + return entry.CommitTsNs, nil } + return fallbackTsNs, nil } -// deletedAfterTsNs returns the newest recorded delete timestamp among path -// and its ancestors. -func (f *Filer) deletedAfterTsNs(ctx context.Context, path util.FullPath) (int64, error) { - var tsNs int64 - for p := string(path); p != "" && p != "/"; { - ts, found, err := f.kvReadTsNs(ctx, metaPathKey(metaPathDelPrefix, util.FullPath(p))) - if err != nil { - return 0, err - } - if found && ts > tsNs { - tsNs = ts - } - if _, pendDel := f.pendingPathTsNs(util.FullPath(p)); pendDel > tsNs { - tsNs = pendDel - } - i := strings.LastIndexByte(p, '/') - if i <= 0 { - break - } - p = p[:i] - } - return tsNs, nil -} - -// localCommitTsNs returns the newest timestamp this filer is known to have -// committed for path: the recorded event time when one exists — it is not -// client-controllable the way an entry's mtime is — else the entry's mtime -// for entries written before recording existed, plus any recorded delete of -// the path or an ancestor. +// localCommitTsNs returns this filer's newest known commit time for path: +// the entry's stamp — not client-controllable like mtime — else mtime. func (f *Filer) localCommitTsNs(ctx context.Context, path util.FullPath) (int64, error) { if f.Store == nil { return 0, nil } - var tsNs int64 - recTs, found, err := f.kvReadTsNs(ctx, metaPathKey(metaPathTsPrefix, path)) + entry, err := f.Store.FindEntry(ctx, path) if err != nil { - return 0, err - } - if found { - tsNs = recTs - } else if entry, findErr := f.Store.FindEntry(ctx, path); findErr != nil { - if !errors.Is(findErr, filer_pb.ErrNotFound) { - return 0, findErr + if errors.Is(err, filer_pb.ErrNotFound) { + return 0, nil } - } else if entry != nil { - tsNs = entry.Mtime.UnixNano() - } - if pendTs, _ := f.pendingPathTsNs(path); pendTs > tsNs { - tsNs = pendTs - } - delTs, err := f.deletedAfterTsNs(ctx, path) - if err != nil { return 0, err } - if delTs > tsNs { - tsNs = delTs + if entry == nil { + return 0, nil } - return tsNs, nil + return f.EntryCommitTsNs(ctx, entry, entry.Mtime.UnixNano()) } diff --git a/weed/filer/meta_path_ts_records_test.go b/weed/filer/meta_path_ts_records_test.go new file mode 100644 index 0000000000..72376857cf --- /dev/null +++ b/weed/filer/meta_path_ts_records_test.go @@ -0,0 +1,91 @@ +package filer + +import ( + "context" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/seaweedfs/seaweedfs/weed/util/log_buffer" +) + +// Committed entries carry their own commit timestamp; deletes write +// nothing — no per-path record and no tombstone. +func TestPathCommitStampedOnEntry(t *testing.T) { + store := newStubFilerStore() + f := newReplayTestFiler(store) + ctx := context.Background() + entry := &Entry{FullPath: "/dir/file", Content: []byte("data")} + f.StampEntryCommitTsNs(entry) + if err := store.InsertEntry(ctx, entry); err != nil { + t.Fatalf("InsertEntry: %v", err) + } + f.releaseEntryCommitTsNs(entry.CommitTsNs) + if got, err := f.CommitTsNs(ctx, "/dir/file", 0); err != nil || got != entry.CommitTsNs { + t.Fatalf("CommitTsNs = %d,%v; want %d", got, err, entry.CommitTsNs) + } + if len(store.kv) != 0 { + t.Fatalf("kv keys written for a change event: %v", store.kv) + } + + if err := store.DeleteEntry(ctx, entry.FullPath); err != nil { + t.Fatalf("DeleteEntry: %v", err) + } + if len(store.kv) != 0 { + t.Fatalf("kv keys written for a change or delete: %v", store.kv) + } +} + +// The emitted event's TsNs is the commit timestamp stamped on the entry +// before its store write — identical values so peers and the local store +// agree on last-writer-wins. +func TestEventTsNsEqualsEntryCommitTs(t *testing.T) { + f := &Filer{ + Signature: 7, + LocalMetaLogBuffer: log_buffer.NewLogBuffer( + "commit-ts-test", time.Minute, + func(*log_buffer.LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, + nil, nil, + ), + } + defer f.LocalMetaLogBuffer.ShutdownLogBuffer() + + ctx, sink := WithMetadataEventSink(context.Background()) + newEntry := &Entry{FullPath: util.FullPath("/dir/file")} + f.StampEntryCommitTsNs(newEntry) + f.NotifyUpdateEvent(ctx, nil, newEntry, true, false, nil) + + event := sink.Last() + if event == nil { + t.Fatal("expected metadata event") + } + if event.TsNs != newEntry.CommitTsNs { + t.Fatalf("event.TsNs = %d, entry.CommitTsNs = %d; must be equal", event.TsNs, newEntry.CommitTsNs) + } + if got := f.metaLogInflight.minTsNs(); got != 0 { + t.Fatalf("stamp still in flight after emit: %d", got) + } +} + +// A notify that never saw a pre-stamp still emits at a fresh stamp — the +// stamp/release pairing must hold either way. +func TestEventWithoutPreStampStillEmits(t *testing.T) { + f := &Filer{ + Signature: 7, + LocalMetaLogBuffer: log_buffer.NewLogBuffer( + "commit-ts-fresh", time.Minute, + func(*log_buffer.LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, + nil, nil, + ), + } + defer f.LocalMetaLogBuffer.ShutdownLogBuffer() + + ctx, sink := WithMetadataEventSink(context.Background()) + f.NotifyUpdateEvent(ctx, nil, &Entry{FullPath: util.FullPath("/dir/file")}, true, false, nil) + if sink.Last() == nil || sink.Last().TsNs == 0 { + t.Fatal("expected stamped metadata event") + } + if got := f.metaLogInflight.minTsNs(); got != 0 { + t.Fatalf("stamp leaked: %d", got) + } +} diff --git a/weed/filer/meta_replay.go b/weed/filer/meta_replay.go index ca517f8659..0906542363 100644 --- a/weed/filer/meta_replay.go +++ b/weed/filer/meta_replay.go @@ -57,21 +57,20 @@ func (f *Filer) Replay(resp *filer_pb.SubscribeMetadataResponse) error { if err != nil || newer { return err } + newEntry := FromPbEntry(dir, message.NewEntry) + newEntry.CommitTsNs = resp.TsNs + if message.OldEntry.IsDirectory { + glog.V(4).Infof("updating %v", newPath) + return f.updateExistingEntry(ctx, oldPath, newEntry) + } glog.V(4).Infof("deleting %v", oldPath) if err := f.Store.DeleteEntry(ctx, oldPath); err != nil { return err } - // Record the commit before the insert too: if the insert fails and - // the event is retried later, the path sits deleted in between — - // without the record a stale peer create could resurrect it in the - // gap. The entry still exists afterward, so no delete record. - f.notePathChange(oldPath, resp.TsNs, false) glog.V(4).Infof("creating %v", newPath) - newEntry := FromPbEntry(dir, message.NewEntry) if err := f.Store.InsertEntry(ctx, newEntry); err != nil { return err } - f.notePathChange(oldPath, resp.TsNs, false) return nil } @@ -98,10 +97,6 @@ func (f *Filer) Replay(resp *filer_pb.SubscribeMetadataResponse) error { return err } } - // The origin deleted the directory even when a newer descendant - // keeps its entry here — recording the delete fences stale child - // creates that would otherwise resurrect under it. - f.notePathChange(oldPath, resp.TsNs, true) } } @@ -113,16 +108,68 @@ func (f *Filer) Replay(resp *filer_pb.SubscribeMetadataResponse) error { if !newer { glog.V(4).Infof("creating %v", newPath) newEntry := FromPbEntry(dir, message.NewEntry) + newEntry.CommitTsNs = resp.TsNs if err := f.Store.InsertEntry(ctx, newEntry); err != nil { return err } - f.notePathChange(newPath, resp.TsNs, false) } } return nil } +// updateExistingEntry applies a same-path update without deleting the row in +// between: the path never sits absent, so a failed retry cannot leave it +// unordered — a tombstone would instead read as a subtree delete that +// suppresses a directory's children. UpdateEntry goes on a present row and +// InsertEntry on an absent one (InsertEntry is not an upsert on stores +// configured without upsert support, and UpdateEntry touches no row that is +// not there). +// +// A peer filer sharing the store can still delete the row between the find +// and the update — the UPDATE then changes zero rows yet reports success, +// silently dropping the replay while its subscription offset advances. The +// write is verified by re-reading the entry, and each pass re-checks the +// recorded commit order so a newer peer change wins instead of being +// resurrected over. +func (f *Filer) updateExistingEntry(ctx context.Context, path util.FullPath, newEntry *Entry) error { + var lastErr error + for attempt := 0; attempt < 4; attempt++ { + if attempt > 0 { + newer, err := f.replayBlockedByLocal(ctx, path, newEntry.CommitTsNs) + if err != nil { + return err + } + if newer { + return nil + } + } + if _, findErr := f.Store.FindEntry(ctx, path); findErr == nil { + if err := f.Store.UpdateEntry(ctx, newEntry); err != nil { + return err + } + if _, err := f.Store.FindEntry(ctx, path); err == nil { + return nil + } else if !errors.Is(err, filer_pb.ErrNotFound) { + return err + } else { + lastErr = err + } + } else if errors.Is(findErr, filer_pb.ErrNotFound) { + // An insert colliding with a racing peer create retries as an + // update on the next pass. + if err := f.Store.InsertEntry(ctx, newEntry); err != nil { + lastErr = err + } else { + return nil + } + } else { + return findErr + } + } + return fmt.Errorf("update %s: %w", path, lastErr) +} + // deleteReplayedFolderChildren removes local entries under dir, and queues // the chunks of removed files for deletion. Children this filer committed // after the event stay — a newer local write wins over a delayed delete — @@ -263,7 +310,6 @@ func (f *Filer) deleteSweptChild(ctx context.Context, sub *Entry, message *filer 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 diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index 1919108fcd..42f772a57a 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -166,6 +166,10 @@ message Entry { repeated FileChunk chunks = 3; FuseAttributes attributes = 4; map extended = 5; + // filer-local commit timestamp of the version stored here: the meta log + // event's ts_ns for the mutation that produced this entry, or 0 for + // entries written before the field existed + int64 commit_ts_ns = 6; bytes hard_link_id = 7; int32 hard_link_counter = 8; // only exists in hard link meta data bytes content = 9; // if not empty, the file content diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index f00ed17771..8257c18b8b 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -672,18 +672,22 @@ func (x *RemoteEntry) GetRemoteContentEncoding() string { } type Entry struct { - state protoimpl.MessageState `protogen:"open.v1"` - Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"` - IsDirectory bool `protobuf:"varint,2,opt,name=is_directory,json=isDirectory,proto3" json:"is_directory,omitempty"` - Chunks []*FileChunk `protobuf:"bytes,3,rep,name=chunks,proto3" json:"chunks,omitempty"` - Attributes *FuseAttributes `protobuf:"bytes,4,opt,name=attributes,proto3" json:"attributes,omitempty"` - Extended map[string][]byte `protobuf:"bytes,5,rep,name=extended,proto3" json:"extended,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` - HardLinkId []byte `protobuf:"bytes,7,opt,name=hard_link_id,json=hardLinkId,proto3" json:"hard_link_id,omitempty"` - HardLinkCounter int32 `protobuf:"varint,8,opt,name=hard_link_counter,json=hardLinkCounter,proto3" json:"hard_link_counter,omitempty"` // only exists in hard link meta data - Content []byte `protobuf:"bytes,9,opt,name=content,proto3" json:"content,omitempty"` // if not empty, the file content - RemoteEntry *RemoteEntry `protobuf:"bytes,10,opt,name=remote_entry,json=remoteEntry,proto3" json:"remote_entry,omitempty"` - Quota int64 `protobuf:"varint,11,opt,name=quota,proto3" json:"quota,omitempty"` // for bucket only. Positive/Negative means enabled/disabled. - WormEnforcedAtTsNs int64 `protobuf:"varint,12,opt,name=worm_enforced_at_ts_ns,json=wormEnforcedAtTsNs,proto3" json:"worm_enforced_at_ts_ns,omitempty"` + state protoimpl.MessageState `protogen:"open.v1"` + Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"` + IsDirectory bool `protobuf:"varint,2,opt,name=is_directory,json=isDirectory,proto3" json:"is_directory,omitempty"` + Chunks []*FileChunk `protobuf:"bytes,3,rep,name=chunks,proto3" json:"chunks,omitempty"` + Attributes *FuseAttributes `protobuf:"bytes,4,opt,name=attributes,proto3" json:"attributes,omitempty"` + Extended map[string][]byte `protobuf:"bytes,5,rep,name=extended,proto3" json:"extended,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + // filer-local commit timestamp of the version stored here: the meta log + // event's ts_ns for the mutation that produced this entry, or 0 for + // entries written before the field existed + CommitTsNs int64 `protobuf:"varint,6,opt,name=commit_ts_ns,json=commitTsNs,proto3" json:"commit_ts_ns,omitempty"` + HardLinkId []byte `protobuf:"bytes,7,opt,name=hard_link_id,json=hardLinkId,proto3" json:"hard_link_id,omitempty"` + HardLinkCounter int32 `protobuf:"varint,8,opt,name=hard_link_counter,json=hardLinkCounter,proto3" json:"hard_link_counter,omitempty"` // only exists in hard link meta data + Content []byte `protobuf:"bytes,9,opt,name=content,proto3" json:"content,omitempty"` // if not empty, the file content + RemoteEntry *RemoteEntry `protobuf:"bytes,10,opt,name=remote_entry,json=remoteEntry,proto3" json:"remote_entry,omitempty"` + Quota int64 `protobuf:"varint,11,opt,name=quota,proto3" json:"quota,omitempty"` // for bucket only. Positive/Negative means enabled/disabled. + WormEnforcedAtTsNs int64 `protobuf:"varint,12,opt,name=worm_enforced_at_ts_ns,json=wormEnforcedAtTsNs,proto3" json:"worm_enforced_at_ts_ns,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -753,6 +757,13 @@ func (x *Entry) GetExtended() map[string][]byte { return nil } +func (x *Entry) GetCommitTsNs() int64 { + if x != nil { + return x.CommitTsNs + } + return 0 +} + func (x *Entry) GetHardLinkId() []byte { if x != nil { return x.HardLinkId @@ -7300,7 +7311,7 @@ const file_filer_proto_rawDesc = "" + "\vremote_size\x18\x05 \x01(\x03R\n" + "remoteSize\x12;\n" + "\x17remote_content_encoding\x18\x06 \x01(\tH\x00R\x15remoteContentEncoding\x88\x01\x01B\x1a\n" + - "\x18_remote_content_encoding\"\x89\x04\n" + + "\x18_remote_content_encoding\"\xab\x04\n" + "\x05Entry\x12\x12\n" + "\x04name\x18\x01 \x01(\tR\x04name\x12!\n" + "\fis_directory\x18\x02 \x01(\bR\visDirectory\x12+\n" + @@ -7309,6 +7320,8 @@ const file_filer_proto_rawDesc = "" + "attributes\x18\x04 \x01(\v2\x18.filer_pb.FuseAttributesR\n" + "attributes\x129\n" + "\bextended\x18\x05 \x03(\v2\x1d.filer_pb.Entry.ExtendedEntryR\bextended\x12 \n" + + "\fcommit_ts_ns\x18\x06 \x01(\x03R\n" + + "commitTsNs\x12 \n" + "\fhard_link_id\x18\a \x01(\fR\n" + "hardLinkId\x12*\n" + "\x11hard_link_counter\x18\b \x01(\x05R\x0fhardLinkCounter\x12\x18\n" + diff --git a/weed/pb/filer_pb/filer_vtproto.pb.go b/weed/pb/filer_pb/filer_vtproto.pb.go index e86380465e..ea2d3747aa 100644 --- a/weed/pb/filer_pb/filer_vtproto.pb.go +++ b/weed/pb/filer_pb/filer_vtproto.pb.go @@ -389,6 +389,11 @@ func (m *Entry) MarshalToSizedBufferVT(dAtA []byte) (int, error) { i-- dAtA[i] = 0x3a } + if m.CommitTsNs != 0 { + i = protohelpers.EncodeVarint(dAtA, i, uint64(m.CommitTsNs)) + i-- + dAtA[i] = 0x30 + } if len(m.Extended) > 0 { for k := range m.Extended { v := m.Extended[k] @@ -6702,6 +6707,9 @@ func (m *Entry) SizeVT() (n int) { n += mapEntrySize + 1 + protohelpers.SizeOfVarint(uint64(mapEntrySize)) } } + if m.CommitTsNs != 0 { + n += 1 + protohelpers.SizeOfVarint(uint64(m.CommitTsNs)) + } l = len(m.HardLinkId) if l > 0 { n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) @@ -10230,6 +10238,25 @@ func (m *Entry) UnmarshalVT(dAtA []byte) error { } m.Extended[mapkey] = mapvalue iNdEx = postIndex + case 6: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field CommitTsNs", wireType) + } + m.CommitTsNs = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.CommitTsNs |= int64(b&0x7F) << shift + if b < 0x80 { + break + } + } case 7: if wireType != 2 { return fmt.Errorf("proto: wrong wireType = %d for field HardLinkId", wireType) diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index c338e051c5..8989e10730 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -668,7 +668,9 @@ func (fs *FilerServer) applyObjectMutation(ctx context.Context, m *filer_pb.Obje if m.TouchMtime { newEntry.Attr.Mtime = time.Now() } + fs.filer.StampEntryCommitTsNs(newEntry) if err := fs.filer.UpdateEntry(ctx, oldEntry, newEntry, fromOtherCluster); err != nil { + fs.filer.ReleaseEntryCommitTsNs(newEntry.CommitTsNs) return err } // Emit the metadata event so the update replicates and subscribers see it, @@ -773,7 +775,9 @@ func (fs *FilerServer) applyRecomputeLatest(ctx context.Context, m *filer_pb.Obj } } + fs.filer.StampEntryCommitTsNs(pointer) if err := fs.filer.UpdateEntry(ctx, oldPointer, pointer, fromOtherCluster); err != nil { + fs.filer.ReleaseEntryCommitTsNs(pointer.CommitTsNs) return err } // Replicate the recomputed pointer to peer filers and subscribers. Without @@ -803,7 +807,9 @@ func (fs *FilerServer) applyRecomputeLatest(ctx context.Context, m *filer_pb.Obj priorEntry.Extended = make(map[string][]byte) } priorEntry.Extended[rc.DemoteKey] = rc.DemoteValue + fs.filer.StampEntryCommitTsNs(priorEntry) if err := fs.filer.UpdateEntry(ctx, oldPrior, priorEntry, fromOtherCluster); err != nil { + fs.filer.ReleaseEntryCommitTsNs(priorEntry.CommitTsNs) return err } fs.filer.NotifyUpdateEvent(ctx, oldPrior, priorEntry, false, fromOtherCluster, signatures) @@ -899,6 +905,7 @@ func (fs *FilerServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntr ctx, eventSink := filer.WithMetadataEventSink(ctx) resp := &filer_pb.UpdateEntryResponse{LogTsNs: logTsNs, LogSignature: fs.filer.Signature} + fs.filer.StampEntryCommitTsNs(newEntry) if err = fs.filer.UpdateEntry(ctx, entry, newEntry, req.IsFromOtherCluster); err == nil { fs.filer.DeleteChunksNotRecursive(garbage) @@ -906,6 +913,7 @@ func (fs *FilerServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntr resp.MetadataEvent = eventSink.Last() } else { + fs.filer.ReleaseEntryCommitTsNs(newEntry.CommitTsNs) glog.V(3).InfofCtx(ctx, "UpdateEntry %s: %v", filepath.Join(req.Directory, req.Entry.Name), err) } diff --git a/weed/server/filer_grpc_server_remote.go b/weed/server/filer_grpc_server_remote.go index 892789fb7f..18be095c59 100644 --- a/weed/server/filer_grpc_server_remote.go +++ b/weed/server/filer_grpc_server_remote.go @@ -318,7 +318,9 @@ func (fs *FilerServer) doCacheRemoteObjectToLocalCluster(ctx context.Context, re // this skips meta data log events + fs.filer.StampEntryCommitTsNs(newEntry) if err := fs.filer.Store.UpdateEntry(context.Background(), newEntry); err != nil { + fs.filer.ReleaseEntryCommitTsNs(newEntry.CommitTsNs) fs.filer.DeleteUncommittedChunks(ctx, chunks) return nil, err } diff --git a/weed/server/filer_grpc_server_rename.go b/weed/server/filer_grpc_server_rename.go index a1a6c682cc..7c40b9cb64 100644 --- a/weed/server/filer_grpc_server_rename.go +++ b/weed/server/filer_grpc_server_rename.go @@ -78,10 +78,12 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom var pendingRemoteCopies []*pendingRemoteCopy moveErr := fs.moveEntry(ctx, nil, oldParent, oldEntry, newParent, req.NewName, req.Signatures, false, &metadataEvents, &pendingChunkDeletes, &pendingRemoteCopies) if moveErr != nil { + releaseMetadataEventStamps(fs.filer, metadataEvents) fs.filer.RollbackTransaction(ctx) return nil, fmt.Errorf("%s/%s move error: %v", req.OldDirectory, req.OldName, moveErr) } else { if commitError := fs.filer.CommitTransaction(ctx); commitError != nil { + releaseMetadataEventStamps(fs.filer, metadataEvents) fs.filer.RollbackTransaction(ctx) return nil, fmt.Errorf("%s/%s move commit error: %v", req.OldDirectory, req.OldName, commitError) } @@ -149,10 +151,12 @@ func (fs *FilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest, var pendingRemoteCopies []*pendingRemoteCopy moveErr := fs.moveEntry(ctx, stream, oldParent, oldEntry, newParent, req.NewName, req.Signatures, false, &metadataEvents, &pendingChunkDeletes, &pendingRemoteCopies) if moveErr != nil { + releaseMetadataEventStamps(fs.filer, metadataEvents) fs.filer.RollbackTransaction(ctx) return fmt.Errorf("%s/%s move error: %v", req.OldDirectory, req.OldName, moveErr) } else { if commitError := fs.filer.CommitTransaction(ctx); commitError != nil { + releaseMetadataEventStamps(fs.filer, metadataEvents) fs.filer.RollbackTransaction(ctx) return fmt.Errorf("%s/%s move commit error: %v", req.OldDirectory, req.OldName, commitError) } @@ -178,6 +182,17 @@ type metadataEvent struct { deleteOldRemote bool } +// releaseMetadataEventStamps drops the commit stamps queued events would +// never adopt when the rename fails — left in flight they pin the delivered +// and flushed watermarks below the aborted rename forever. +func releaseMetadataEventStamps(f *filer.Filer, events []metadataEvent) { + for _, event := range events { + if event.newEntry != nil { + f.ReleaseEntryCommitTsNs(event.newEntry.CommitTsNs) + } + } +} + // pendingRemoteCopy is a remote-only rename over a remote-only target whose // copy runs after the commit: the store transaction cannot roll a remote // write back, so the new entry keeps the old key's stamp until the copy lands. @@ -320,7 +335,7 @@ func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, stream filer_pb return nil } -func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, moveFolderSubEntries func() error, signatures []int32, skipTargetLookup bool, metadataEvents *[]metadataEvent, pendingChunkDeletes *[]*filer_pb.FileChunk, pendingRemoteCopies *[]*pendingRemoteCopy) error { +func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, moveFolderSubEntries func() error, signatures []int32, skipTargetLookup bool, metadataEvents *[]metadataEvent, pendingChunkDeletes *[]*filer_pb.FileChunk, pendingRemoteCopies *[]*pendingRemoteCopy) (retErr error) { oldPath, newPath := oldParent.Child(entry.Name()), newParent.Child(newName) @@ -415,6 +430,13 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee Remote: remoteEntry, Quota: entry.Quota, } + defer func() { + // On failure no covering event adopts the insert's stamp. + if retErr != nil { + fs.filer.ReleaseEntryCommitTsNs(newEntry.CommitTsNs) + } + }() + if deferredCopy { // a pending entry renamed again still reads its original source newEntry.Extended = maps.Clone(newEntry.Extended) @@ -433,7 +455,9 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee if newEntry.FullPath.IsLongerFileName(fs.filer.MaxFilenameLength) { return filer_pb.ErrEntryNameTooLong } + fs.filer.StampEntryCommitTsNs(newEntry) if createErr := fs.filer.Store.InsertEntryKnownAbsent(filer.WithSuppressedMetadataEvents(ctx), newEntry); createErr != nil { + fs.filer.ReleaseEntryCommitTsNs(newEntry.CommitTsNs) return fmt.Errorf("insert entry %s: %v", newEntry.FullPath, createErr) } } else { diff --git a/weed/server/filer_grpc_server_rename_test.go b/weed/server/filer_grpc_server_rename_test.go index 1a36bc45d3..10d6941406 100644 --- a/weed/server/filer_grpc_server_rename_test.go +++ b/weed/server/filer_grpc_server_rename_test.go @@ -532,3 +532,30 @@ func TestAtomicRenameEntrySkipsDescendantTargetLookups(t *testing.T) { t.Fatalf("event count = %d, want 3", got) } } + +// The stamps queued rename events carry are released when the rename fails: +// left in flight they would pin the delivered watermark below the aborted +// rename forever. +func TestAtomicRenameReleasesQueuedStampsOnCommitFailure(t *testing.T) { + store := newRenameTestStore() + store.entries["/dir"] = newDirectoryEntry("/dir", 100) + store.entries["/dir/a.txt"] = newFileEntry("/dir/a.txt", 101) + store.entries["/dir/b.txt"] = newFileEntry("/dir/b.txt", 102) + store.commitErr = errors.New("commit failed") + + f := newRenameTestFiler(t, store) + server := &FilerServer{filer: f, entryLockTable: util.NewLockTable[util.FullPath]()} + _, err := server.AtomicRenameEntry(context.Background(), &filer_pb.AtomicRenameEntryRequest{ + OldDirectory: "/", + OldName: "dir", + NewDirectory: "/", + NewName: "dstdir", + }) + if err == nil { + t.Fatal("expected commit failure") + } + now := time.Now().UnixNano() + if got := f.LocalDeliveredThroughTsNs(now); got != now { + t.Fatalf("LocalDeliveredThroughTsNs = %d, want %d: a leaked stamp still caps the claim", got, now) + } +} diff --git a/weed/server/filer_grpc_server_traverse_meta.go b/weed/server/filer_grpc_server_traverse_meta.go index 7ad1014f81..8f413a82d0 100644 --- a/weed/server/filer_grpc_server_traverse_meta.go +++ b/weed/server/filer_grpc_server_traverse_meta.go @@ -61,7 +61,7 @@ func (fs *FilerServer) TraverseBfsMetadata(req *filer_pb.TraverseBfsMetadataRequ return nil } sendEntry := func(entry *filer.Entry) error { - entryTsNs, err := fs.filer.CommitTsNs(ctx, entry.FullPath, entry.Mtime.UnixNano()) + entryTsNs, err := fs.filer.EntryCommitTsNs(ctx, entry, entry.Mtime.UnixNano()) if err != nil { return fmt.Errorf("commit timestamp of %s: %w", entry.FullPath, err) }