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