diff --git a/weed/filer/filer.go b/weed/filer/filer.go index 204659372b..0a5d3540be 100644 --- a/weed/filer/filer.go +++ b/weed/filer/filer.go @@ -82,6 +82,11 @@ type Filer struct { persistedLogCache *persistedLogCache metaLogInflight metaLogInflight remoteTombstones *remoteDeletionTombstones + // EntryLockTable serializes mutations per entry path. The filer server + // holds it across local read-modify-write mutations and the meta + // aggregator across a replayed peer change's check-and-apply, so the two + // cannot interleave on the same path. + EntryLockTable *util.LockTable[util.FullPath] // remoteTombstonesDone, when non-nil, is closed once the startup tombstone // rebuild finishes; lazy remote reads wait on it so a pending delete // cannot resurrect in the gap. @@ -110,6 +115,12 @@ 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. @@ -133,6 +144,7 @@ func NewFiler(masters pb.ServerDiscovery, grpcDialOption grpc.DialOption, filerH DeletionRetryQueue: NewDeletionRetryQueue(), persistedLogCache: newPersistedLogCache(persistedLogCacheMaxBytes), remoteTombstones: newRemoteDeletionTombstones(), + EntryLockTable: util.NewLockTable[util.FullPath](), } f.flushCtx, f.flushCancel = context.WithCancel(context.Background()) if f.UniqueFilerId < 0 { @@ -913,6 +925,12 @@ func (f *Filer) Shutdown() { f.LocalMetaLogBuffer.ShutdownLogBuffer() // The final metadata-log flush still needs the store to append its entry. f.LocalMetaLogBuffer.WaitForShutdown() + // Stop peer subscriptions and let each persist its last applied offset; + // the replay on the next start otherwise resumes from a stale offset and + // applies those old events over newer local state. + if f.MetaAggregator != nil { + f.MetaAggregator.Shutdown() + } // Persist the deletion ledger one last time before the store closes, so a // clean shutdown leaves the recovery set exactly consistent with reality. f.snapshotDeletionLedger() diff --git a/weed/filer/filer_lazy_remote_test.go b/weed/filer/filer_lazy_remote_test.go index 04798c3f21..7a867fcc6d 100644 --- a/weed/filer/filer_lazy_remote_test.go +++ b/weed/filer/filer_lazy_remote_test.go @@ -35,6 +35,7 @@ type stubFilerStore struct { kv map[string][]byte insertErr error findErr error + kvPutErr error kvGetErr error kvDeleteErr error kvGetHook func(key []byte) @@ -49,6 +50,12 @@ func newStubFilerStore() *stubFilerStore { } } +func (s *stubFilerStore) setKvPutErr(err error) { + s.mu.Lock() + defer s.mu.Unlock() + s.kvPutErr = err +} + func (s *stubFilerStore) GetName() string { return "stub" } func (s *stubFilerStore) Initialize(util.Configuration, string) error { return nil } func (s *stubFilerStore) Shutdown() {} @@ -60,6 +67,9 @@ func (s *stubFilerStore) RollbackTransaction(context.Context) error { return nil func (s *stubFilerStore) KvPut(_ context.Context, key []byte, value []byte) error { s.mu.Lock() defer s.mu.Unlock() + if s.kvPutErr != nil { + return s.kvPutErr + } s.kv[string(key)] = append([]byte(nil), value...) return nil } diff --git a/weed/filer/filer_notify.go b/weed/filer/filer_notify.go index 6101715eb2..8e23cef44a 100644 --- a/weed/filer/filer_notify.go +++ b/weed/filer/filer_notify.go @@ -71,6 +71,10 @@ 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) } diff --git a/weed/filer/meta_aggregator.go b/weed/filer/meta_aggregator.go index 0f8e16d1b4..c5d6521c89 100644 --- a/weed/filer/meta_aggregator.go +++ b/weed/filer/meta_aggregator.go @@ -63,6 +63,11 @@ type MetaAggregator struct { flushAdvanced chan struct{} remotePeerArrived chan struct{} peerWatermarksLock sync.Mutex + peerWg sync.WaitGroup + // stopped is set under peerChansLock by Shutdown; OnPeerUpdate then + // rejects adds so no subscription can outlive the shutdown wait and run + // against a store Filer.Shutdown has already closed. + stopped bool } // MetaAggregator only aggregates data "on the fly". The logs are not re-persisted to disk. @@ -87,6 +92,13 @@ func (ma *MetaAggregator) OnPeerUpdate(update *master_pb.ClusterNodeUpdate, star ma.peerChansLock.Lock() defer ma.peerChansLock.Unlock() + // After Shutdown the master connection may still deliver an update; a + // subscription started now would have no stop signal and could run past + // the store close. + if ma.stopped { + return + } + address := pb.ServerAddress(update.Address) if update.IsAdd { // the peer is already followed, restarting would only lose the events @@ -102,7 +114,11 @@ func (ma *MetaAggregator) OnPeerUpdate(update *master_pb.ClusterNodeUpdate, star // Account for the peer before its stream signals; keep prior values // on reconnect. ma.initPeerWatermark(address) - go ma.loopSubscribeToOneFiler(ma.filer, ma.self, address, startFrom, stopChan) + ma.peerWg.Add(1) + go func() { + defer ma.peerWg.Done() + ma.loopSubscribeToOneFiler(ma.filer, ma.self, address, startFrom, stopChan) + }() } else { if prevChan, found := ma.peerChans[address]; found { close(prevChan) @@ -344,6 +360,42 @@ func (ma *MetaAggregator) HasPeer(address pb.ServerAddress) bool { return false } +// Shutdown stops every peer subscription and waits for each to return, +// letting the deferred offset save persist the last applied position before +// the store closes. Without it a graceful restart resumes from the last +// periodically-saved offset — up to a minute stale — and replays those +// events over newer local state. +func (ma *MetaAggregator) Shutdown() { + ma.peerChansLock.Lock() + ma.stopped = true + chans := ma.peerChans + ma.peerChans = make(map[pb.ServerAddress]chan struct{}) + for _, stopChan := range chans { + close(stopChan) + } + ma.peerChansLock.Unlock() + + // Wait without a cap: every blocking point in the subscribe loop is + // stop-aware, so a peer still running here is genuinely wedged and + // returning anyway would let Filer.Shutdown close the store under it and + // skip its deferred offset save. + done := make(chan struct{}) + go func() { + ma.peerWg.Wait() + close(done) + }() + ticker := time.NewTicker(5 * time.Second) + defer ticker.Stop() + for { + select { + case <-done: + return + case <-ticker.C: + glog.Warningf("meta aggregator shutdown: peer subscriptions still stopping") + } + } +} + func (ma *MetaAggregator) loopSubscribeToOneFiler(f *Filer, self pb.ServerAddress, peer pb.ServerAddress, startFrom time.Time, stopChan chan struct{}) { lastTsNs := startFrom.UnixNano() for { @@ -368,7 +420,12 @@ func (ma *MetaAggregator) loopSubscribeToOneFiler(f *Filer, self pb.ServerAddres if lastTsNs < nextLastTsNs { lastTsNs = nextLastTsNs } - time.Sleep(1733 * time.Millisecond) + select { + case <-stopChan: + glog.V(0).Infof("stop subscribing peer %s meta change", peer) + return + case <-time.After(1733 * time.Millisecond): + } } } @@ -384,11 +441,34 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress, */ - var maybeReplicateMetadataChange func(*filer_pb.SubscribeMetadataResponse) + var maybeReplicateMetadataChange func(*filer_pb.SubscribeMetadataResponse) error lastPersistTime := time.Now() lastTsNs := startFrom + // replayFloorTsNs is the earliest event this session gave up replaying. + // The persisted offset and the reconnect resume point must stay behind + // it so the event is redelivered on the next subscribe instead of being + // skipped forever. + var replayFloorTsNs int64 + persistableTsNs := func(tsNs int64) int64 { + if replayFloorTsNs > 0 && replayFloorTsNs-1 < tsNs { + return replayFloorTsNs - 1 + } + return tsNs + } - peerSignature, err := ma.readFilerStoreSignature(peer) + // Everything below derives from ctx, cancelled when the peer is removed + // or the aggregator shuts down, so no stage can block past its stop. + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { + select { + case <-stopChan: + cancel() + case <-ctx.Done(): + } + }() + + peerSignature, err := ma.readFilerStoreSignature(ctx, peer) if err != nil { return lastTsNs, fmt.Errorf("connecting to peer filer %s: %v", peer, err) } @@ -408,7 +488,7 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress, // different concept and can be set to arbitrary values. preTraverseTime := time.Now() glog.V(0).Infof("no previous offset for peer %s, starting full metadata sync", peer) - if traverseErr := ma.traversePeerMetadata(f, peer); traverseErr != nil { + if traverseErr := ma.traversePeerMetadata(ctx, f, peer); traverseErr != nil { return lastTsNs, fmt.Errorf("initial metadata sync from %s: %v", peer, traverseErr) } lastTsNs = preTraverseTime.Add(-time.Minute).UnixNano() @@ -420,11 +500,12 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress, return lastTsNs, fmt.Errorf("read offset for peer %s: %w", peer, err) } defer func(prevTsNs int64) { - if lastTsNs != prevTsNs && lastTsNs != lastPersistTime.UnixNano() { - if err := ma.updateOffset(f, peer, peerSignature, lastTsNs); err == nil { - glog.V(0).Infof("last sync time with %s at %v (%d)", peer, time.Unix(0, lastTsNs), lastTsNs) + saveTsNs := persistableTsNs(lastTsNs) + if saveTsNs != prevTsNs && lastTsNs != lastPersistTime.UnixNano() { + if err := ma.updateOffset(f, peer, peerSignature, saveTsNs); err == nil { + glog.V(0).Infof("last sync time with %s at %v (%d)", peer, time.Unix(0, saveTsNs), saveTsNs) } else { - glog.Errorf("failed to save last sync time with %s at %v (%d)", peer, time.Unix(0, lastTsNs), lastTsNs) + glog.Errorf("failed to save last sync time with %s at %v (%d)", peer, time.Unix(0, saveTsNs), saveTsNs) } } }(lastTsNs) @@ -432,11 +513,23 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress, glog.V(0).Infof("follow peer: %v, last %v (%d)", peer, time.Unix(0, lastTsNs), lastTsNs) var counter int64 var synced bool - maybeReplicateMetadataChange = func(event *filer_pb.SubscribeMetadataResponse) { - replicateMetadataChange(f.Store, peer, event) + maybeReplicateMetadataChange = func(event *filer_pb.SubscribeMetadataResponse) error { + if err := replicateMetadataChange(f, peer, event, stopChan); err != nil { + if errors.Is(err, errPeerSubscriptionStopped) { + return err + } + // A failed event becomes a barrier the persisted offset and + // the reconnect resume point cannot pass, while the stream + // keeps flowing: blocking on an event that can never replay + // would stall every later event from this peer. + if replayFloorTsNs == 0 || event.TsNs < replayFloorTsNs { + replayFloorTsNs = event.TsNs + } + return nil + } counter++ if lastPersistTime.Add(time.Minute).Before(time.Now()) { - if err := ma.updateOffset(f, peer, peerSignature, event.TsNs); err == nil { + if err := ma.updateOffset(f, peer, peerSignature, persistableTsNs(event.TsNs)); err == nil { if event.TsNs < time.Now().Add(-2*time.Minute).UnixNano() { glog.V(0).Infof("sync with %s progressed to: %v %0.2f/sec", peer, time.Unix(0, event.TsNs), float64(counter)/60.0) } else if !synced { @@ -449,6 +542,7 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress, glog.V(0).Infof("failed to update offset for %v: %v", peer, err) } } + return nil } } @@ -465,7 +559,9 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress, return err } if maybeReplicateMetadataChange != nil { - maybeReplicateMetadataChange(event) + if err := maybeReplicateMetadataChange(event); err != nil { + return err + } } return nil } @@ -476,17 +572,8 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress, glog.V(0).Infof("subscribing remote %s meta change: %v, clientId:%d", peer, time.Unix(0, lastTsNs), ma.filer.UniqueFilerId) err = pb.WithFilerClient(true, 0, peer, ma.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - // Stop the stream promptly on removal; the blocking Recv below would - // otherwise only notice when the stream errors on its own. - go func() { - select { - case <-stopChan: - cancel() - case <-ctx.Done(): - } - }() + // ctx is cancelled on stopChan by the watch above; the blocking Recv + // below stops promptly on removal or shutdown. atomic.AddInt32(&ma.filer.UniqueFilerEpoch, 1) // Construct a log file reader that reads chunks via the peer filer's LookupVolume. lookupFn := LookupFn(filerClient{client}) @@ -511,6 +598,14 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress, } processOne := func(event *filer_pb.SubscribeMetadataResponse) error { + // Bail promptly on stop; a cancelled stream may still deliver + // queued events while unwinding, and replaying them now only + // delays the offset save the shutdown is waiting on. + select { + case <-stopChan: + return errPeerSubscriptionStopped + default: + } if err := processEventFn(event); err != nil { glog.V(0).Infof("SubscribeLocalMetadata process %v: %v", event, err) return fmt.Errorf("process %v: %w", event, err) @@ -587,19 +682,37 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress, } } }) - return lastTsNs, err + return persistableTsNs(lastTsNs), err } // replicateMetadataChange retries transient Replay failures with bounded -// backoff. A failure that outlives the retry budget is counted and logged, -// then skipped: blocking on an event that can never replay would stall every -// later event from this peer, which is worse than one entry staying stale. -func replicateMetadataChange(store FilerStore, peer pb.ServerAddress, event *filer_pb.SubscribeMetadataResponse) { - err := util.Retry("replicate metadata change from "+string(peer), func() error { - return Replay(store, event) - }) - if err == nil { - return +// backoff and reports the outcome. errPeerSubscriptionStopped is returned +// when the subscription stops mid-retry so the caller aborts and the offset +// save keeps the last completed event. A failure that outlives the retry +// budget is counted and logged, then returned: the caller pins the peer's +// offset at the event so a restart replays it rather than skipping it. +func replicateMetadataChange(f *Filer, peer pb.ServerAddress, event *filer_pb.SubscribeMetadataResponse, stopChan <-chan struct{}) error { + waitTime := time.Second + var err error + for waitTime < util.RetryWaitTime { + select { + case <-stopChan: + return errPeerSubscriptionStopped + default: + } + if err = f.Replay(event); err == nil { + return nil + } + if !util.IsTransientError(err) { + break + } + glog.V(0).Infof("retry replicate metadata change from %s: %v", peer, err) + select { + case <-stopChan: + return errPeerSubscriptionStopped + case <-time.After(waitTime): + } + waitTime += waitTime / 2 } stats.FilerMetaAggregatorReplayFailures.WithLabelValues(string(peer)).Inc() name := event.GetEventNotification().GetNewEntry().GetName() @@ -607,15 +720,20 @@ func replicateMetadataChange(store FilerStore, peer pb.ServerAddress, event *fil name = event.GetEventNotification().GetOldEntry().GetName() } glog.Errorf("giving up replicating metadata change from %s for %s/%s (ts=%d): %v", peer, event.Directory, name, event.TsNs, err) + return err } +// errPeerSubscriptionStopped ends event processing once the subscription's +// stop channel closed. +var errPeerSubscriptionStopped = errors.New("peer subscription stopped") + // traversePeerMetadata does a full BFS traversal of a peer filer's metadata // and inserts all entries into the local store. This is used when a filer // connects to a peer for the first time and needs to bootstrap pre-existing data. -func (ma *MetaAggregator) traversePeerMetadata(f *Filer, peer pb.ServerAddress) error { +// The ctx cancels on the peer's stop channel so a shutdown does not have to +// out-wait a long traversal. +func (ma *MetaAggregator) traversePeerMetadata(ctx context.Context, f *Filer, peer pb.ServerAddress) error { return pb.WithFilerClient(true, 0, peer, ma.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() stream, err := client.TraverseBfsMetadata(ctx, &filer_pb.TraverseBfsMetadataRequest{ Directory: "/", ExcludedPrefixes: []string{SystemLogDir}, @@ -635,22 +753,17 @@ func (ma *MetaAggregator) traversePeerMetadata(f *Filer, peer pb.ServerAddress) if resp.Entry == nil { continue } - fullpath := util.Join(resp.Directory, resp.Entry.Name) entry := FromPbEntry(resp.Directory, resp.Entry) - if insertErr := f.Store.InsertEntry(context.Background(), entry); insertErr != nil { - // Entry may already exist (root dir, or partial previous bootstrap). - existing, findErr := f.Store.FindEntry(context.Background(), entry.FullPath) - if findErr != nil { - return fmt.Errorf("insert entry %s: %w", fullpath, insertErr) - } - // Only overwrite if the peer's entry is newer. - if entry.Attr.Mtime.After(existing.Attr.Mtime) { - if updateErr := f.Store.UpdateEntry(context.Background(), entry); updateErr != nil { - return fmt.Errorf("update entry %s: %w", fullpath, updateErr) - } - } else { - glog.V(1).Infof("skip older peer entry %s (peer mtime %v <= local mtime %v)", fullpath, entry.Attr.Mtime, existing.Attr.Mtime) - } + // 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. + peerTsNs := resp.EntryTsNs + if peerTsNs == 0 { + peerTsNs = entry.Attr.Mtime.UnixNano() + } + if applyErr := f.applyTraversedEntry(ctx, entry, peerTsNs); applyErr != nil { + return applyErr } count++ if count%10000 == 0 { @@ -662,9 +775,38 @@ func (ma *MetaAggregator) traversePeerMetadata(f *Filer, peer pb.ServerAddress) }) } -func (ma *MetaAggregator) readFilerStoreSignature(peer pb.ServerAddress) (sig int32, err error) { +// 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. +func (f *Filer) applyTraversedEntry(ctx context.Context, entry *Entry, peerTsNs int64) error { + unlock := f.lockReplayPaths(entry.FullPath, "") + defer unlock() + localTsNs, err := f.localCommitTsNs(ctx, entry.FullPath) + if err != nil { + return fmt.Errorf("check local state for %s: %w", entry.FullPath, err) + } + if localTsNs >= peerTsNs { + return nil + } + 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. + if _, findErr := f.Store.FindEntry(ctx, entry.FullPath); findErr != nil { + return fmt.Errorf("insert entry %s: %w", entry.FullPath, insertErr) + } + if updateErr := f.Store.UpdateEntry(ctx, entry); updateErr != nil { + return fmt.Errorf("update entry %s: %w", entry.FullPath, updateErr) + } + } + f.notePathChange(entry.FullPath, peerTsNs, false) + return nil +} + +func (ma *MetaAggregator) readFilerStoreSignature(ctx context.Context, peer pb.ServerAddress) (sig int32, err error) { err = pb.WithFilerClient(false, 0, peer, ma.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { - resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{}) + resp, err := client.GetFilerConfiguration(ctx, &filer_pb.GetFilerConfigurationRequest{}) if err != nil { return err } diff --git a/weed/filer/meta_aggregator_replay_test.go b/weed/filer/meta_aggregator_replay_test.go index 43990b2922..10522f7f14 100644 --- a/weed/filer/meta_aggregator_replay_test.go +++ b/weed/filer/meta_aggregator_replay_test.go @@ -15,6 +15,14 @@ import ( "github.com/seaweedfs/seaweedfs/weed/util" ) +// newReplayTestFiler builds the minimum Filer Replay needs over a stub store. +func newReplayTestFiler(store FilerStore) *Filer { + return &Filer{ + Store: NewFilerStoreWrapper(store), + EntryLockTable: util.NewLockTable[util.FullPath](), + } +} + // flakyReplayStore fails InsertEntry for its first failUntil calls, then // delegates to the embedded stubFilerStore. type flakyReplayStore struct { @@ -73,8 +81,11 @@ func TestReplicateMetadataChangeRetriesTransientFailure(t *testing.T) { err: errors.New("i/o timeout talking to store"), } peer := pb.ServerAddress("peer-transient:1") + f := newReplayTestFiler(store) - replicateMetadataChange(store, peer, quotaChangeEvent("my-bucket", wantQuota)) + if err := replicateMetadataChange(f, peer, quotaChangeEvent("my-bucket", wantQuota), nil); err != nil { + t.Fatalf("replicateMetadataChange: %v", err) + } inserted, err := store.FindEntry(context.Background(), util.NewFullPath("/buckets", "my-bucket")) if err != nil { @@ -97,10 +108,14 @@ func TestReplicateMetadataChangeGivesUpLoudlyOnPermanentFailure(t *testing.T) { err: errors.New("entry checksum mismatch"), } peer := pb.ServerAddress("peer-permanent:1") + f := newReplayTestFiler(store) before := replayFailureCount(t, peer) - replicateMetadataChange(store, peer, quotaChangeEvent("poison-bucket", 65536<<20)) + err := replicateMetadataChange(f, peer, quotaChangeEvent("poison-bucket", 65536<<20), nil) + if err == nil { + t.Fatal("expected the exhausted replay error so the caller can pin the peer offset at the failed event") + } if got := store.attemptCount(); got != 1 { t.Fatalf("attempts = %d, want exactly 1: a non-transient error must fail fast, not retry", got) @@ -109,3 +124,587 @@ func TestReplicateMetadataChangeGivesUpLoudlyOnPermanentFailure(t *testing.T) { t.Fatalf("FilerMetaAggregatorReplayFailures[%s] = %v, want %v", peer, got, before+1) } } + +func overwriteEvent(dir, name string, tsNs int64, content string) *filer_pb.SubscribeMetadataResponse { + return &filer_pb.SubscribeMetadataResponse{ + Directory: dir, + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: name}, + NewEntry: &filer_pb.Entry{Name: name, Content: []byte(content)}, + }, + TsNs: tsNs, + } +} + +func deleteEvent(dir, name string, tsNs int64) *filer_pb.SubscribeMetadataResponse { + return &filer_pb.SubscribeMetadataResponse{ + Directory: dir, + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: name}, + }, + TsNs: tsNs, + } +} + +func TestReplaySkipsEventOlderThanLocalEntry(t *testing.T) { + store := newStubFilerStore() + f := newReplayTestFiler(store) + now := time.Now() + path := util.NewFullPath("/dir", "file") + store.entries[string(path)] = &Entry{ + FullPath: path, + Attr: Attr{Mtime: now}, + Content: []byte("newer local write"), + } + + if err := f.Replay(overwriteEvent("/dir", "file", now.Add(-time.Minute).UnixNano(), "old replayed version")); err != nil { + t.Fatalf("Replay: %v", err) + } + entry, err := store.FindEntry(context.Background(), path) + if err != nil { + t.Fatalf("FindEntry: %v", err) + } + if string(entry.Content) != "newer local write" { + t.Fatalf("content = %q, want local write kept", entry.Content) + } +} + +func TestReplayAppliesEventNewerThanLocalEntry(t *testing.T) { + store := newStubFilerStore() + f := newReplayTestFiler(store) + now := time.Now() + path := util.NewFullPath("/dir", "file") + store.entries[string(path)] = &Entry{ + FullPath: path, + Attr: Attr{Mtime: now.Add(-time.Minute)}, + Content: []byte("old local write"), + } + + if err := f.Replay(overwriteEvent("/dir", "file", now.UnixNano(), "newer peer write")); err != nil { + t.Fatalf("Replay: %v", err) + } + entry, err := store.FindEntry(context.Background(), path) + if err != nil { + t.Fatalf("FindEntry: %v", err) + } + if string(entry.Content) != "newer peer write" { + t.Fatalf("content = %q, want peer write applied", entry.Content) + } +} + +func TestReplaySkipsDeleteOlderThanLocalEntry(t *testing.T) { + store := newStubFilerStore() + f := newReplayTestFiler(store) + now := time.Now() + path := util.NewFullPath("/dir", "file") + store.entries[string(path)] = &Entry{FullPath: path, Attr: Attr{Mtime: now}} + + if err := f.Replay(deleteEvent("/dir", "file", now.Add(-time.Minute).UnixNano())); err != nil { + t.Fatalf("Replay: %v", err) + } + if _, err := store.FindEntry(context.Background(), path); err != nil { + t.Fatalf("newer local entry must survive a stale delete: %v", err) + } +} + +func TestReplayAppliesNewerDelete(t *testing.T) { + store := newStubFilerStore() + f := newReplayTestFiler(store) + now := time.Now() + path := util.NewFullPath("/dir", "file") + store.entries[string(path)] = &Entry{FullPath: path, Attr: Attr{Mtime: now.Add(-time.Minute)}} + + if err := f.Replay(deleteEvent("/dir", "file", now.UnixNano())); err != nil { + t.Fatalf("Replay: %v", err) + } + if _, err := store.FindEntry(context.Background(), path); err == nil { + t.Fatal("newer delete must remove the local entry") + } +} + +// A move event touches two paths; the local check applies to each side +// independently, so a newer destination is kept while the source is removed. +func TestReplayMoveChecksEachPathSeparately(t *testing.T) { + store := newStubFilerStore() + f := newReplayTestFiler(store) + now := time.Now() + oldPath := util.NewFullPath("/a", "file") + newPath := util.NewFullPath("/b", "file") + store.entries[string(oldPath)] = &Entry{FullPath: oldPath, Attr: Attr{Mtime: now.Add(-time.Minute)}} + store.entries[string(newPath)] = &Entry{FullPath: newPath, Attr: Attr{Mtime: now}, Content: []byte("kept")} + + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/a", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "file"}, + NewEntry: &filer_pb.Entry{Name: "file"}, + NewParentPath: "/b", + }, + TsNs: now.Add(-time.Minute).UnixNano(), + } + if err := f.Replay(event); err != nil { + t.Fatalf("Replay: %v", err) + } + if _, err := store.FindEntry(context.Background(), oldPath); err == nil { + t.Fatal("old source entry should be deleted") + } + entry, err := store.FindEntry(context.Background(), newPath) + if err != nil { + t.Fatalf("FindEntry: %v", err) + } + if string(entry.Content) != "kept" { + t.Fatalf("content = %q, want newer destination kept", entry.Content) + } +} + +// 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 +// be overwritten when its commit record is older. +func TestReplayAppliesWhenCommitRecordIsOlderThanFutureMtime(t *testing.T) { + store := newStubFilerStore() + f := newReplayTestFiler(store) + now := time.Now() + path := util.NewFullPath("/b", "x") + store.entries[string(path)] = &Entry{ + 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. + f.notePathChange(path, now.Add(-time.Minute).UnixNano(), false) + + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/a", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "x"}, + NewEntry: &filer_pb.Entry{Name: "x", Content: []byte("renamed in")}, + NewParentPath: "/b", + }, + TsNs: now.UnixNano(), + } + if err := f.Replay(event); err != nil { + t.Fatalf("Replay: %v", err) + } + entry, err := store.FindEntry(context.Background(), path) + if err != nil { + t.Fatalf("FindEntry: %v", err) + } + if string(entry.Content) != "renamed in" { + t.Fatalf("content = %q, want the rename destination applied", entry.Content) + } +} + +// A store read failure is not a not-found: the replayed change must not be +// applied on the guess that no newer local entry exists. +func TestReplayPropagatesLookupError(t *testing.T) { + store := newStubFilerStore() + store.findErr = errors.New("store read failed") + f := newReplayTestFiler(store) + + err := f.Replay(overwriteEvent("/dir", "file", time.Now().UnixNano(), "v")) + if err == nil { + t.Fatal("expected Replay to propagate the FindEntry failure") + } +} + +// The check and the mutation share the per-path entry lock, so a local write +// holding that lock cannot be split by a replayed change. +func TestReplaySerializesWithEntryLock(t *testing.T) { + store := newStubFilerStore() + f := newReplayTestFiler(store) + now := time.Now() + path := util.NewFullPath("/dir", "file") + 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. + held := f.EntryLockTable.AcquireLock("test local write", path, util.ExclusiveLock) + f.notePathChange(path, now.UnixNano(), false) + + done := make(chan error, 1) + go func() { + done <- f.Replay(overwriteEvent("/dir", "file", now.Add(-time.Minute).UnixNano(), "stale")) + }() + select { + case <-done: + t.Fatal("replay finished while the path lock was still held") + case <-time.After(50 * time.Millisecond): + } + f.EntryLockTable.ReleaseLock(path, held) + if err := <-done; err != nil { + t.Fatalf("Replay: %v", err) + } + entry, err := store.FindEntry(context.Background(), path) + if err != nil { + t.Fatalf("FindEntry: %v", err) + } + if entry.Content != nil { + t.Fatalf("content = %q, want local entry kept", entry.Content) + } +} + +// A closed stop channel must abort the retry loop with +// errPeerSubscriptionStopped so the subscription tears down and the offset +// save keeps the last completed event instead of skipping this one. +func TestReplicateMetadataChangeAbortsOnStop(t *testing.T) { + store := &flakyReplayStore{ + stubFilerStore: newStubFilerStore(), + failUntil: 1 << 30, + err: errors.New("i/o timeout talking to store"), + } + peer := pb.ServerAddress("peer-stopped:1") + f := newReplayTestFiler(store) + + stopChan := make(chan struct{}) + close(stopChan) + err := replicateMetadataChange(f, peer, quotaChangeEvent("stuck-bucket", 1<<20), stopChan) + if !errors.Is(err, errPeerSubscriptionStopped) { + t.Fatalf("err = %v, want errPeerSubscriptionStopped", err) + } + if got := store.attemptCount(); got != 0 { + t.Fatalf("attempts = %d, want 0: a stopped subscription must not replay", got) + } +} + +// 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 +// tombstone. +func TestReplaySerializesWithAncestorLock(t *testing.T) { + store := newStubFilerStore() + f := newReplayTestFiler(store) + + 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", + EventNotification: &filer_pb.EventNotification{ + NewEntry: &filer_pb.Entry{Name: "file", Content: []byte("stale resurrect")}, + }, + TsNs: now.Add(-time.Minute).UnixNano(), + } + done := make(chan error, 1) + go func() { + done <- f.Replay(createEvent) + }() + select { + case <-done: + t.Fatal("replay finished while the ancestor lock was still held") + case <-time.After(50 * time.Millisecond): + } + f.EntryLockTable.ReleaseLock("/dir", held) + 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") + } +} + +// The bug this guards: a peer entry committed recently but stamped with an +// old client mtime must still apply over an absent-or-older local commit. +func TestBootstrapAppliesPeerCommitNewerThanLocal(t *testing.T) { + store := newStubFilerStore() + f := newReplayTestFiler(store) + now := time.Now() + path := util.NewFullPath("/dir", "file") + + entry := &Entry{ + FullPath: path, + Attr: Attr{Mtime: now.Add(-time.Hour)}, // old client-set mtime + Content: []byte("fresh peer write"), + } + if err := f.applyTraversedEntry(context.Background(), entry, now.UnixNano()); err != nil { + t.Fatalf("applyTraversedEntry: %v", err) + } + got, err := store.FindEntry(context.Background(), path) + if err != nil { + t.Fatalf("peer entry with newer commit must bootstrap: %v", err) + } + if string(got.Content) != "fresh peer write" { + t.Fatalf("content = %q, want peer entry applied", got.Content) + } +} + +// A bootstrap-applied entry records the sender's commit timestamp, so a +// replayed event older than it — but newer than the entry's mtime — is +// still skipped. +func TestBootstrapAppliedCommitBlocksStaleReplay(t *testing.T) { + store := newStubFilerStore() + f := newReplayTestFiler(store) + now := time.Now() + path := util.NewFullPath("/dir", "file") + + entry := &Entry{ + FullPath: path, + Attr: Attr{Mtime: now.Add(-time.Hour)}, + Content: []byte("fresh peer write"), + } + if err := f.applyTraversedEntry(context.Background(), entry, now.UnixNano()); err != nil { + t.Fatalf("applyTraversedEntry: %v", err) + } + if err := f.Replay(overwriteEvent("/dir", "file", now.Add(-time.Minute).UnixNano(), "stale replay")); err != nil { + t.Fatalf("Replay: %v", err) + } + got, _ := store.FindEntry(context.Background(), path) + if string(got.Content) != "fresh peer write" { + t.Fatalf("content = %q, stale replay must not overwrite the bootstrap entry", got.Content) + } +} + +// An existing local entry is replaced only when the peer commit is newer — +// not when its mtime happens to be. +func TestBootstrapUpdatesExistingByCommitTime(t *testing.T) { + store := newStubFilerStore() + f := newReplayTestFiler(store) + now := time.Now() + 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"), + } + f.notePathChange(path, now.Add(-time.Minute).UnixNano(), false) + + entry := &Entry{ + FullPath: path, + Attr: Attr{Mtime: now.Add(-time.Hour)}, + Content: []byte("newer peer commit"), + } + if err := f.applyTraversedEntry(context.Background(), entry, now.UnixNano()); err != nil { + t.Fatalf("applyTraversedEntry: %v", err) + } + got, _ := store.FindEntry(context.Background(), path) + if string(got.Content) != "newer peer commit" { + t.Fatalf("content = %q, newer peer commit must update despite the future local mtime", got.Content) + } +} + +// 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) { + 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) + } + f.notePathChange(path, 50, false) + if got, err := f.CommitTsNs(context.Background(), path, 7); err != nil || got != 50 { + t.Fatalf("recorded: 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) + } +} diff --git a/weed/filer/meta_aggregator_test.go b/weed/filer/meta_aggregator_test.go index 70adae3878..82e8abd1a1 100644 --- a/weed/filer/meta_aggregator_test.go +++ b/weed/filer/meta_aggregator_test.go @@ -34,3 +34,55 @@ func TestOnPeerUpdateRepeatedAdd(t *testing.T) { t.Fatal("expecting the subscription to be removed") } } + +// Shutdown must close every peer's stopChan and wait for the loops, which is +// what lets each loop's deferred offset save run before the store closes. +func TestMetaAggregatorShutdownStopsPeerLoops(t *testing.T) { + ma := NewMetaAggregator(nil, pb.ServerAddress("127.0.0.1:2"), nil) + + stopChan := make(chan struct{}) + peer := pb.ServerAddress("127.0.0.1:1") + ma.peerChans[peer] = stopChan + ma.peerWg.Add(1) + exited := make(chan struct{}) + go func() { + defer ma.peerWg.Done() + <-stopChan + close(exited) + }() + + done := make(chan struct{}) + go func() { + ma.Shutdown() + close(done) + }() + + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("Shutdown did not return") + } + select { + case <-exited: + default: + t.Fatal("peer loop was not stopped") + } +} + +// After Shutdown begins, a peer-add update arriving on the still-connected +// master watch must not start a subscription: it would have no stop signal +// and could run past the store close. +func TestOnPeerUpdateAfterShutdownRejected(t *testing.T) { + ma := NewMetaAggregator(nil, pb.ServerAddress("127.0.0.1:2"), nil) + ma.Shutdown() + + peer := pb.ServerAddress("127.0.0.1:1") + ma.OnPeerUpdate(&master_pb.ClusterNodeUpdate{Address: string(peer), IsAdd: true}, time.Now()) + + ma.peerChansLock.Lock() + _, found := ma.peerChans[peer] + ma.peerChansLock.Unlock() + if found { + t.Fatal("peer subscription started after shutdown") + } +} diff --git a/weed/filer/meta_path_ts.go b/weed/filer/meta_path_ts.go new file mode 100644 index 0000000000..844f744d3a --- /dev/null +++ b/weed/filer/meta_path_ts.go @@ -0,0 +1,324 @@ +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. +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)) + if err != nil { + return 0, err + } + if found { + tsNs = recTs + } + if pendTs, _ := f.pendingPathTsNs(path); pendTs > tsNs { + tsNs = pendTs + } + return tsNs, nil +} + +// 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) + } +} + +// 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. +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)) + 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 + } + } 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 + } + return tsNs, nil +} diff --git a/weed/filer/meta_replay.go b/weed/filer/meta_replay.go index 58542d305e..6895490b8b 100644 --- a/weed/filer/meta_replay.go +++ b/weed/filer/meta_replay.go @@ -2,6 +2,8 @@ package filer import ( "context" + "slices" + "strings" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -9,31 +11,166 @@ import ( ) // Replay applies the delete and the insert as separate, non-transactional -// store calls, so retrying it after a partial failure is not atomic. -func Replay(filerStore FilerStore, resp *filer_pb.SubscribeMetadataResponse) error { +// store calls, so retrying it after a partial failure is not atomic; both +// operations are idempotent and the timestamp check tolerates re-delivery. +// +// A replayed event can be older than local state: a peer resuming from a +// stale offset replays everything since the offset, and two writes through +// different filers race within the replication lag. Apply each path's change +// only when no newer local change is recorded, so a late event cannot revert +// or resurrect entries. +// +// Ordering uses the per-path commit timestamp recorded for every emitted +// mutation (meta_path_ts.go), which survives deletes as a tombstone and is +// not client-controllable the way entry mtime is — a mounted client can set +// mtime into the future, and a rename keeps the source file's mtime on the +// destination entry. Entries written before recording existed fall back to +// their mtime. +// +// The check and the mutation run under the same per-path entry lock the +// filer server's local writes hold, so they cannot be split by a concurrent +// local update or by another peer's replay. +func (f *Filer) Replay(resp *filer_pb.SubscribeMetadataResponse) error { message := resp.EventNotification - var oldPath util.FullPath - var newEntry *Entry + dir := resp.Directory + if message.NewParentPath != "" { + dir = message.NewParentPath + } + var oldPath, newPath util.FullPath if message.OldEntry != nil { oldPath = util.NewFullPath(resp.Directory, message.OldEntry.Name) - glog.V(4).Infof("deleting %v", oldPath) - if err := filerStore.DeleteEntry(context.Background(), oldPath); err != nil { + } + if message.NewEntry != nil { + newPath = util.NewFullPath(dir, message.NewEntry.Name) + } + ctx := context.Background() + + unlock := f.lockReplayPaths(oldPath, newPath) + defer unlock() + + samePath := oldPath != "" && oldPath == newPath + if samePath { + newer, err := f.replayBlockedByLocal(ctx, oldPath, resp.TsNs) + if err != nil || newer { return err } + 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 + } + + if message.OldEntry != nil { + newer, err := f.replayBlockedByLocal(ctx, oldPath, resp.TsNs) + if err != nil { + return err + } + if !newer { + glog.V(4).Infof("deleting %v", oldPath) + if err := f.Store.DeleteEntry(ctx, oldPath); err != nil { + return err + } + f.notePathChange(oldPath, resp.TsNs, true) + } } if message.NewEntry != nil { - dir := resp.Directory - if message.NewParentPath != "" { - dir = message.NewParentPath - } - key := util.NewFullPath(dir, message.NewEntry.Name) - glog.V(4).Infof("creating %v", key) - newEntry = FromPbEntry(dir, message.NewEntry) - if err := filerStore.InsertEntry(context.Background(), newEntry); err != nil { + newer, err := f.replayBlockedByLocal(ctx, newPath, resp.TsNs) + if err != nil { return err } + if !newer { + 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(newPath, resp.TsNs, false) + } } return nil } + +// replayBlockedByLocal reports whether the local store already holds a +// version of path committed after the event's timestamp. Lookup failures are +// propagated: a replayed change must not be applied on the guess that the +// path was simply absent, or a transient read error could let a stale event +// overwrite a newer entry past the point of no return. +func (f *Filer) replayBlockedByLocal(ctx context.Context, path util.FullPath, tsNs int64) (bool, error) { + if tsNs <= 0 { + return false, nil + } + localTsNs, err := f.localCommitTsNs(ctx, path) + if err != nil { + return false, err + } + return localTsNs > tsNs, nil +} + +// lockReplayPaths holds the shared entry locks for every path the event +// touches, in the same sorted order the rename handler takes them. +// lockReplayPaths takes the per-path entry locks for the event's paths and +// every ancestor of them, in lexicographic order — parents sort before +// children, the same ordering the local rename handler's two-path lock uses, +// so lock acquisition is deadlock-free across operations. Ancestors are +// included because the replay check consults ancestor delete records: a +// local recursive delete or directory rename holds only the ancestor's lock, +// and without contending on it a replay could reinsert a stale child after +// its check. Ancestors are locked shared — they only need to exclude an +// exclusive holder, not each other, or every replay under a busy parent +// would serialize — while a path the event actually changes stays exclusive +// even when it is also the other path's ancestor. The root is never locked: +// it cannot be deleted, and a shared root lock would still be excluded by +// every top-level delete. +func (f *Filer) lockReplayPaths(oldPath, newPath util.FullPath) func() { + lt := f.EntryLockTable + if lt == nil { + return func() {} + } + var paths []util.FullPath + addAncestors := func(p util.FullPath) { + for i := strings.LastIndexByte(string(p), '/'); i > 0; i = strings.LastIndexByte(string(p), '/') { + p = p[:i] + paths = append(paths, p) + } + } + addAncestors(oldPath) + addAncestors(newPath) + if oldPath != "" { + paths = append(paths, oldPath) + } + if newPath != "" { + paths = append(paths, newPath) + } + if len(paths) == 0 { + return func() {} + } + slices.Sort(paths) + paths = slices.Compact(paths) + locks := make([]*util.ActiveLock, len(paths)) + for i, p := range paths { + lockType := util.SharedLock + if p == oldPath || p == newPath { + lockType = util.ExclusiveLock + } + locks[i] = lt.AcquireLock("MetaReplay", p, lockType) + } + return func() { + for i := len(paths) - 1; i >= 0; i-- { + lt.ReleaseLock(paths[i], locks[i]) + } + } +} diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index 27f31b0fec..a97fbfdf95 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -714,6 +714,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/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index 1e2dee3938..fe0d0767a8 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -4777,9 +4777,13 @@ func (x *TraverseBfsMetadataRequest) GetExcludedPrefixes() []string { } type TraverseBfsMetadataResponse struct { - state protoimpl.MessageState `protogen:"open.v1"` - Directory string `protobuf:"bytes,1,opt,name=directory,proto3" json:"directory,omitempty"` - Entry *Entry `protobuf:"bytes,2,opt,name=entry,proto3" json:"entry,omitempty"` + state protoimpl.MessageState `protogen:"open.v1"` + Directory string `protobuf:"bytes,1,opt,name=directory,proto3" json:"directory,omitempty"` + Entry *Entry `protobuf:"bytes,2,opt,name=entry,proto3" json:"entry,omitempty"` + // 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. + EntryTsNs int64 `protobuf:"varint,3,opt,name=entry_ts_ns,json=entryTsNs,proto3" json:"entry_ts_ns,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -4828,6 +4832,13 @@ func (x *TraverseBfsMetadataResponse) GetEntry() *Entry { return nil } +func (x *TraverseBfsMetadataResponse) GetEntryTsNs() int64 { + if x != nil { + return x.EntryTsNs + } + return 0 +} + type LogEntry struct { state protoimpl.MessageState `protogen:"open.v1"` TsNs int64 `protobuf:"varint,1,opt,name=ts_ns,json=tsNs,proto3" json:"ts_ns,omitempty"` @@ -7688,10 +7699,11 @@ const file_filer_proto_rawDesc = "" + "\bfiler_id\x18\x03 \x01(\tR\afilerId\"g\n" + "\x1aTraverseBfsMetadataRequest\x12\x1c\n" + "\tdirectory\x18\x01 \x01(\tR\tdirectory\x12+\n" + - "\x11excluded_prefixes\x18\x02 \x03(\tR\x10excludedPrefixes\"b\n" + + "\x11excluded_prefixes\x18\x02 \x03(\tR\x10excludedPrefixes\"\x82\x01\n" + "\x1bTraverseBfsMetadataResponse\x12\x1c\n" + "\tdirectory\x18\x01 \x01(\tR\tdirectory\x12%\n" + - "\x05entry\x18\x02 \x01(\v2\x0f.filer_pb.EntryR\x05entry\"\x8b\x01\n" + + "\x05entry\x18\x02 \x01(\v2\x0f.filer_pb.EntryR\x05entry\x12\x1e\n" + + "\ventry_ts_ns\x18\x03 \x01(\x03R\tentryTsNs\"\x8b\x01\n" + "\bLogEntry\x12\x13\n" + "\x05ts_ns\x18\x01 \x01(\x03R\x04tsNs\x12,\n" + "\x12partition_key_hash\x18\x02 \x01(\x05R\x10partitionKeyHash\x12\x12\n" + diff --git a/weed/pb/filer_pb/filer_vtproto.pb.go b/weed/pb/filer_pb/filer_vtproto.pb.go index 8f256d85cd..46d4b6e0c6 100644 --- a/weed/pb/filer_pb/filer_vtproto.pb.go +++ b/weed/pb/filer_pb/filer_vtproto.pb.go @@ -4379,6 +4379,11 @@ func (m *TraverseBfsMetadataResponse) MarshalToSizedBufferVT(dAtA []byte) (int, i -= len(m.unknownFields) copy(dAtA[i:], m.unknownFields) } + if m.EntryTsNs != 0 { + i = protohelpers.EncodeVarint(dAtA, i, uint64(m.EntryTsNs)) + i-- + dAtA[i] = 0x18 + } if m.Entry != nil { size, err := m.Entry.MarshalToSizedBufferVT(dAtA[:i]) if err != nil { @@ -8267,6 +8272,9 @@ func (m *TraverseBfsMetadataResponse) SizeVT() (n int) { l = m.Entry.SizeVT() n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) } + if m.EntryTsNs != 0 { + n += 1 + protohelpers.SizeOfVarint(uint64(m.EntryTsNs)) + } n += len(m.unknownFields) return n } @@ -20865,6 +20873,25 @@ func (m *TraverseBfsMetadataResponse) UnmarshalVT(dAtA []byte) error { return err } iNdEx = postIndex + case 3: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field EntryTsNs", wireType) + } + m.EntryTsNs = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.EntryTsNs |= int64(b&0x7F) << shift + if b < 0x80 { + break + } + } default: iNdEx = preIndex skippy, err := protohelpers.Skip(dAtA[iNdEx:]) diff --git a/weed/server/filer_grpc_server_traverse_meta.go b/weed/server/filer_grpc_server_traverse_meta.go index 2ebf33d05d..c59d52d79d 100644 --- a/weed/server/filer_grpc_server_traverse_meta.go +++ b/weed/server/filer_grpc_server_traverse_meta.go @@ -42,9 +42,14 @@ func (fs *FilerServer) TraverseBfsMetadata(req *filer_pb.TraverseBfsMetadataRequ // the filer on large trees (e.g. during a peer's first-time bootstrap). sendEntry := func(entry *filer.Entry) error { parent, _ := entry.FullPath.DirAndName() + entryTsNs, err := fs.filer.CommitTsNs(ctx, entry.FullPath, entry.Mtime.UnixNano()) + if err != nil { + return fmt.Errorf("commit timestamp of %s: %w", entry.FullPath, err) + } if err := stream.Send(&filer_pb.TraverseBfsMetadataResponse{ Directory: parent, Entry: entry.ToProtoEntry(), + EntryTsNs: entryTsNs, }); err != nil { return fmt.Errorf("send traverse bfs metadata response: %w", err) } diff --git a/weed/server/filer_grpc_server_traverse_meta_test.go b/weed/server/filer_grpc_server_traverse_meta_test.go index cb20951111..f6205bb235 100644 --- a/weed/server/filer_grpc_server_traverse_meta_test.go +++ b/weed/server/filer_grpc_server_traverse_meta_test.go @@ -4,6 +4,7 @@ import ( "context" "testing" + "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/util" "github.com/stretchr/testify/assert" @@ -37,8 +38,9 @@ func TestPtrie(t *testing.T) { type captureTraverseStream struct { grpc.ServerStream - ctx context.Context - visited []string + ctx context.Context + visited []string + entryTsNs map[string]int64 } func (s *captureTraverseStream) Context() context.Context { return s.ctx } @@ -48,7 +50,12 @@ func (s *captureTraverseStream) Send(resp *filer_pb.TraverseBfsMetadataResponse) if dir != "/" { dir += "/" } - s.visited = append(s.visited, dir+resp.Entry.Name) + path := dir + resp.Entry.Name + s.visited = append(s.visited, path) + if s.entryTsNs == nil { + s.entryTsNs = make(map[string]int64) + } + s.entryTsNs[path] = resp.EntryTsNs return nil } @@ -89,6 +96,16 @@ func TestTraverseBfsMetadata(t *testing.T) { } seen[path] = true } + + // Every response carries a commit timestamp for ordering; with no + // recorded commit it falls back to the entry's mtime. + for path, tsNs := range stream.entryTsNs { + if path == "/" { + assert.Equal(t, filer.Root.Attr.Mtime.UnixNano(), tsNs) + continue + } + assert.Equalf(t, store.entries[path].Attr.Mtime.UnixNano(), tsNs, "entry_ts_ns for %s", path) + } } var _ filer_pb.SeaweedFiler_TraverseBfsMetadataServer = (*captureTraverseStream)(nil) diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index e4871f914f..58748422c4 100644 --- a/weed/server/filer_server.go +++ b/weed/server/filer_server.go @@ -268,6 +268,9 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) maxFilenameLength := v.GetUint32("filer.options.max_file_name_length") glog.V(0).Infof("max_file_name_length %d", maxFilenameLength) fs.filer = filer.NewFiler(*option.Masters, fs.grpcDialOption, option.Host, option.FilerGroup, option.Collection, option.DefaultReplication, option.DataCenter, maxFilenameLength, nil) + // Peer metadata replay takes the same per-path locks, so a replayed change + // cannot split a local mutation's read-modify-write. + fs.entryLockTable = fs.filer.EntryLockTable fs.filer.Cipher = option.Cipher fs.filer.DefaultDiskType = option.DiskType fs.filer.BuildGuardedRemoteClient = BuildGuardedRemoteStorageClient diff --git a/weed/server/filer_server_handlers_write.go b/weed/server/filer_server_handlers_write.go index 799f764bb9..b1a96e07c4 100644 --- a/weed/server/filer_server_handlers_write.go +++ b/weed/server/filer_server_handlers_write.go @@ -255,7 +255,15 @@ func (fs *FilerServer) DeleteHandler(w http.ResponseWriter, r *http.Request) { return } - err = fs.filer.DeleteEntryMetaAndData(context.Background(), util.FullPath(objectPath), isRecursive, ignoreRecursiveError, !skipChunkDeletion, false, nil, 0) + // The delete and the commit record it emits must serialize with peer + // replay of this path, same as the gRPC DeleteEntry handler. + fullPath := util.FullPath(objectPath) + if fs.entryLockTable != nil { + pathLock := fs.entryLockTable.AcquireLock("DeleteHandler", fullPath, util.ExclusiveLock) + defer fs.entryLockTable.ReleaseLock(fullPath, pathLock) + } + + err = fs.filer.DeleteEntryMetaAndData(context.Background(), fullPath, isRecursive, ignoreRecursiveError, !skipChunkDeletion, false, nil, 0) if err != nil && err != filer_pb.ErrNotFound { glog.V(1).Infoln("deleting", objectPath, ":", err.Error()) writeJsonError(w, r, http.StatusInternalServerError, err)