From dfd9ecab05539f2f789ad99ca955aeaa6036157b Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 26 Oct 2026 08:06:15 -0700 Subject: [PATCH] mount: resume metadata-log replay where a dropped stream stopped (#12079) * pb: checkpoint metadata-log replay progress per settled entry A chunk-mode subscriber's resume cursor only advanced after the whole LogFileRefs batch drained, so a stream dropped mid-replay resubscribed from the original stale position and re-read the same log files while live traffic kept appending - under a flapping filer the replay window never shrank, and each reconnect re-listed and re-fetched the same backlog. ReadLogFileRefs now reports every entry the merge has settled - delivered or filter-skipped - and FollowMetadata checkpoints StartTsNs on it, so a reconnect resumes where the previous stream stopped and the backlog drains across restarts. The peer aggregator wires the same progress into its resubscribe/offset position. * pb: emit a replay checkpoint only after the merge passes its timestamp Checkpointing an entry's timestamp the moment it settles could strand a same-timestamp entry still queued on another filer's stream: resuming is exclusive, so a reconnect would skip the unsettled half of the tie. The merge now emits a timestamp only once a later entry proves every tied entry was delivered (and once at clean EOF). * pb+filer: never regress the replay cursor below the settled checkpoint The drain's final assignment took the last settled entry's own timestamp, which can lag the log position progressFn just checkpointed - a skewed or legacy entry rewinds option.StartTsNs (and the peer's lastTsNs), so the next drain or a resubscribe re-reads the settled prefix. Guard both assignments so the cursor only moves forward. * pb: cover the resubscribe path of replay checkpoints A drain that aborts mid-replay leaves StartTsNs at the last checkpoint the merge passed, and the next subscribe resumes there instead of re-reading the whole ref set. The test drives the real makeSubscribeMetadataFunc path: refs, a mid-batch apply failure, then a resubscribe whose SinceNs must carry the checkpoint. * pb: emit the tail replay checkpoint only for a settled timestamp Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/filer/meta_aggregator.go | 6 +- weed/pb/filer_pb_direct_read.go | 31 +++++- weed/pb/filer_pb_direct_read_test.go | 134 +++++++++++++++++++++-- weed/pb/filer_pb_tail.go | 15 ++- weed/pb/filer_pb_tail_test.go | 117 ++++++++++++++++++++ weed/server/filer_subscribe_loop_test.go | 4 +- 6 files changed, 290 insertions(+), 17 deletions(-) diff --git a/weed/filer/meta_aggregator.go b/weed/filer/meta_aggregator.go index c4f17d8734..e095108676 100644 --- a/weed/filer/meta_aggregator.go +++ b/weed/filer/meta_aggregator.go @@ -640,11 +640,15 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress, lastTsNs, 0, pb.PathFilter{PathPrefix: "/"}, func(event *filer_pb.SubscribeMetadataResponse) error { return processOne(event) + }, func(tsNs int64) { + if tsNs > lastTsNs { + lastTsNs = tsNs + } }) if readErr != nil { return fmt.Errorf("%w from %s: %w", pb.ErrLogFileRead, peer, readErr) } - if lastTs > 0 { + if lastTs > lastTsNs { lastTsNs = lastTs } pendingRefs = nil diff --git a/weed/pb/filer_pb_direct_read.go b/weed/pb/filer_pb_direct_read.go index f536b93a14..b221c26142 100644 --- a/weed/pb/filer_pb_direct_read.go +++ b/weed/pb/filer_pb_direct_read.go @@ -49,12 +49,19 @@ type PathFilter struct { // // Filers are read in parallel (one goroutine per filer), each streaming // entries through a bounded channel. +// +// progressFn, when non-nil, reports checkpoints the caller may resume from: +// a timestamp is emitted once the merge has moved past it, so every entry +// carrying it is already settled and the exclusive startTsNs filter cannot +// skip a straggler sharing it. A stream dropped partway through then resumes +// at the last checkpoint instead of re-reading the whole ref set. func ReadLogFileRefs( refs []*filer_pb.LogFileChunkRef, newReader LogFileReaderFn, startTsNs, stopTsNs int64, filter PathFilter, processEventFn ProcessMetadataFunc, + progressFn func(tsNs int64), ) (lastTsNs int64, err error) { if len(refs) == 0 { @@ -78,7 +85,7 @@ func ReadLogFileRefs( return } - return readFilersMerged(filerOrder, perFiler, newReader, startTsNs, stopTsNs, filter, processEventFn) + return readFilersMerged(filerOrder, perFiler, newReader, startTsNs, stopTsNs, filter, processEventFn, progressFn) } // readFilersMerged reads files from each filer in parallel (one producer @@ -92,6 +99,7 @@ func readFilersMerged( startTsNs, stopTsNs int64, filter PathFilter, processEventFn ProcessMetadataFunc, + progressFn func(tsNs int64), ) (lastTsNs int64, err error) { type filerStream struct { @@ -139,6 +147,13 @@ func readFilersMerged( }(perFiler[filerId], entryCh) } + progress := func(tsNs int64) { + if progressFn != nil { + progressFn(tsNs) + } + } + var pendingProgress int64 + // Seed the min-heap with the first entry from each filer pq := &logEntryHeap{} heap.Init(pq) @@ -169,6 +184,14 @@ func readFilersMerged( if err != nil { return } + // Checkpoint a timestamp only once the merge moves past it: per-filer + // streams are ts-ordered, so an entry above pendingProgress proves no + // stream can still deliver a tied entry the exclusive resume filter + // would then skip. + if pendingProgress > 0 && item.entry.TsNs > pendingProgress { + progress(pendingProgress) + } + pendingProgress = item.entry.TsNs select { case entry, ok := <-streams[item.filerIdx].entryCh: @@ -181,10 +204,14 @@ func readFilersMerged( } // All channels closed: every producer has finished (setFatal, if any, - // happened before its close). + // happened before its close). The tail timestamp is settled now - no + // stream can still deliver a tie. if fe := fatal(); fe != nil { return lastTsNs, fe } + if pendingProgress > 0 { + progress(pendingProgress) + } return } diff --git a/weed/pb/filer_pb_direct_read_test.go b/weed/pb/filer_pb_direct_read_test.go index 04c73fd49e..c7f283c41a 100644 --- a/weed/pb/filer_pb_direct_read_test.go +++ b/weed/pb/filer_pb_direct_read_test.go @@ -158,7 +158,7 @@ func assertOrderedReplay(t *testing.T, files *testLogFiles) { func(resp *filer_pb.SubscribeMetadataResponse) error { timestamps = append(timestamps, resp.TsNs) return nil - }) + }, nil) if err != nil { t.Fatalf("ReadLogFileRefs: %v", err) } @@ -191,7 +191,7 @@ func TestReadLogFileRefsPathFilter(t *testing.T) { func(resp *filer_pb.SubscribeMetadataResponse) error { allCount++ return nil - }) + }, nil) if err != nil { t.Fatalf("ReadLogFileRefs (all): %v", err) } @@ -201,7 +201,7 @@ func TestReadLogFileRefsPathFilter(t *testing.T) { func(resp *filer_pb.SubscribeMetadataResponse) error { filteredCount++ return nil - }) + }, nil) if err != nil { t.Fatalf("ReadLogFileRefs (filtered): %v", err) } @@ -267,7 +267,7 @@ func TestDirectReadVsServerSideThroughput(t *testing.T) { func(resp *filer_pb.SubscribeMetadataResponse) error { atomic.AddInt64(&processed, 1) return nil - }) + }, nil) if err != nil { t.Fatalf("ReadLogFileRefs: %v", err) } @@ -305,7 +305,7 @@ func TestReadLogFileRefsMultiFilerGenuineErrorAborts(t *testing.T) { func(resp *filer_pb.SubscribeMetadataResponse) error { atomic.AddInt64(&count, 1) return nil - }) + }, nil) if err == nil { t.Fatalf("expected error from genuine read failure, got nil (delivered=%d)", count) } @@ -323,7 +323,7 @@ func TestReadLogFileRefsMultiFilerNotFoundSkips(t *testing.T) { func(resp *filer_pb.SubscribeMetadataResponse) error { atomic.AddInt64(&count, 1) return nil - }) + }, nil) if err != nil { t.Fatalf("chunk-not-found should be skipped, got error: %v", err) } @@ -354,7 +354,7 @@ func TestReadLogFileRefsSingleFilerProcessErrorStops(t *testing.T) { return wantErr } return nil - }) + }, nil) if !errors.Is(err, wantErr) { t.Fatalf("expected processing error to propagate, got: %v", err) } @@ -378,7 +378,7 @@ func TestReadLogFileRefsCorruptSizePrefix(t *testing.T) { } _, err := ReadLogFileRefs(refs, readerFn, 0, 0, PathFilter{PathPrefix: "/"}, - func(*filer_pb.SubscribeMetadataResponse) error { return nil }) + func(*filer_pb.SubscribeMetadataResponse) error { return nil }, nil) if err == nil || !strings.Contains(err.Error(), "exceeds") { t.Fatalf("expected size-cap error, got: %v", err) } @@ -414,7 +414,7 @@ func TestReadLogFileRefsAbortDoesNotJoinWedgedReader(t *testing.T) { done := make(chan error, 1) go func() { _, err := ReadLogFileRefs(files.refs, readerFn, 0, 0, PathFilter{PathPrefix: "/"}, - func(*filer_pb.SubscribeMetadataResponse) error { return nil }) + func(*filer_pb.SubscribeMetadataResponse) error { return nil }, nil) done <- err }() @@ -441,7 +441,7 @@ func TestReadLogFileRefsSingleFilerNotFoundSkips(t *testing.T) { func(resp *filer_pb.SubscribeMetadataResponse) error { count++ return nil - }) + }, nil) if err != nil { t.Fatalf("chunk-not-found should be skipped, got error: %v", err) } @@ -450,6 +450,120 @@ func TestReadLogFileRefsSingleFilerNotFoundSkips(t *testing.T) { } } +// TestReadLogFileRefsProgressCheckpoints verifies the progress callback fires +// for every settled entry — including entries the path filter skips — in +// merge order, so a dropped replay can resume from the last checkpoint. +func TestReadLogFileRefsProgressCheckpoints(t *testing.T) { + files := newTestLogFiles(2, 2, 50, 0) + + var checkpoints []int64 + _, err := ReadLogFileRefs(files.refs, files.readerFn(), 0, 0, + PathFilter{PathPrefix: "/data/filer00/"}, + func(resp *filer_pb.SubscribeMetadataResponse) error { return nil }, + func(tsNs int64) { checkpoints = append(checkpoints, tsNs) }) + if err != nil { + t.Fatalf("ReadLogFileRefs: %v", err) + } + if got, want := len(checkpoints), files.totalEvents(); got != want { + t.Fatalf("expected %d checkpoints, got %d", want, got) + } + for i := 1; i < len(checkpoints); i++ { + if checkpoints[i] < checkpoints[i-1] { + t.Fatalf("checkpoints out of order at %d: %d > %d", i, checkpoints[i-1], checkpoints[i]) + } + } +} + +// A timestamp shared by entries from different filers must not be +// checkpointed until the merge has passed it: resuming is exclusive (<= +// startTsNs skipped), so checkpointing a tie mid-settle would lose the other +// half on reconnect. +func TestReadLogFileRefsProgressWaitsOutTimestampTies(t *testing.T) { + tieTs := time.Now().Add(-time.Hour).UnixNano() + fileData := map[string][]byte{ + "fa:1": buildLogFileData([]*filer_pb.SubscribeMetadataResponse{ + makeSubEvent("/data/fa", "a1", tieTs), + makeSubEvent("/data/fa", "a2", tieTs+10), + }), + "fb:1": buildLogFileData([]*filer_pb.SubscribeMetadataResponse{ + makeSubEvent("/data/fb", "b1", tieTs), + makeSubEvent("/data/fb", "b2", tieTs+20), + }), + } + refs := []*filer_pb.LogFileChunkRef{ + {Chunks: []*filer_pb.FileChunk{{FileId: "fa:1"}}, FileTsNs: 1, FilerId: "fa"}, + {Chunks: []*filer_pb.FileChunk{{FileId: "fb:1"}}, FileTsNs: 1, FilerId: "fb"}, + } + readerFn := func(chunks []*filer_pb.FileChunk) (io.ReadCloser, error) { + return io.NopCloser(bytes.NewReader(fileData[chunks[0].FileId])), nil + } + + var checkpoints []int64 + count := 0 + wantErr := fmt.Errorf("boom") + _, err := ReadLogFileRefs(refs, readerFn, 0, 0, + PathFilter{PathPrefix: "/"}, + func(resp *filer_pb.SubscribeMetadataResponse) error { + count++ + if count == 2 { + return wantErr + } + return nil + }, + func(tsNs int64) { checkpoints = append(checkpoints, tsNs) }) + if !errors.Is(err, wantErr) { + t.Fatalf("expected processing error, got: %v", err) + } + if len(checkpoints) != 0 { + t.Fatalf("tie was checkpointed before both entries settled: %v", checkpoints) + } + + checkpoints = nil + _, err = ReadLogFileRefs(refs, readerFn, 0, 0, + PathFilter{PathPrefix: "/"}, + func(resp *filer_pb.SubscribeMetadataResponse) error { return nil }, + func(tsNs int64) { checkpoints = append(checkpoints, tsNs) }) + if err != nil { + t.Fatalf("ReadLogFileRefs: %v", err) + } + if want := []int64{tieTs, tieTs + 10, tieTs + 20}; fmt.Sprint(checkpoints) != fmt.Sprint(want) { + t.Fatalf("checkpoints = %v, want %v", checkpoints, want) + } +} + +// A failed event must not be checkpointed: the callback fires only once the +// merge has moved past a timestamp, so an abort replays at most the last +// timestamp's worth of settled entries. +func TestReadLogFileRefsProgressStopsAtFailedEvent(t *testing.T) { + files := newTestLogFiles(1, 2, 10, 0) + + var delivered []int64 + var lastCheckpoint int64 + count := 0 + wantErr := fmt.Errorf("boom") + _, err := ReadLogFileRefs(files.refs, files.readerFn(), 0, 0, + PathFilter{PathPrefix: "/"}, + func(resp *filer_pb.SubscribeMetadataResponse) error { + count++ + if count == 5 { + return wantErr + } + delivered = append(delivered, resp.TsNs) + return nil + }, + func(tsNs int64) { lastCheckpoint = tsNs }) + if !errors.Is(err, wantErr) { + t.Fatalf("expected processing error, got: %v", err) + } + if lastCheckpoint == 0 { + t.Fatal("expected checkpoints for the settled prefix") + } + if lastCheckpoint != delivered[len(delivered)-2] { + t.Fatalf("last checkpoint %d, want the next-to-last settled event %d", + lastCheckpoint, delivered[len(delivered)-2]) + } +} + // TestProcessOneLogEntrySkipsInvalidUTF8 guards that an event whose string field // is not valid UTF-8 is rejected by proto.Unmarshal and never reaches path // filtering or the consumer. UnmarshalVT would accept it, so the decode here diff --git a/weed/pb/filer_pb_tail.go b/weed/pb/filer_pb_tail.go index edbd68c01a..db7adf42d0 100644 --- a/weed/pb/filer_pb_tail.go +++ b/weed/pb/filer_pb_tail.go @@ -200,11 +200,20 @@ func makeSubscribeMetadataFunc(option *MetadataFollowOption, processEventFn Proc return nil } readFromNs := option.StartTsNs + var progressFn func(tsNs int64) if option.GetResumeTsNs != nil { // The resume cursor lives in the processed watermark, so a // resubscribed ref replay must not be filtered by positions the // previous stream had only seen. readFromNs = sinceNs + } else { + // Checkpoint each settled entry so a dropped stream resumes the + // replay where it stopped instead of re-reading the whole set. + progressFn = func(tsNs int64) { + if tsNs > option.StartTsNs { + option.StartTsNs = tsNs + } + } } lastTs, readErr := ReadLogFileRefs(pendingRefs, option.LogFileReaderFn, readFromNs, option.StopTsNs, @@ -213,11 +222,13 @@ func makeSubscribeMetadataFunc(option *MetadataFollowOption, processEventFn Proc AdditionalPathPrefixes: option.AdditionalPathPrefixes, DirectoriesToWatch: option.DirectoriesToWatch, }, - processEventFn) + processEventFn, progressFn) if readErr != nil { return fmt.Errorf("%w: %w", ErrLogFileRead, readErr) } - if lastTs > 0 && option.GetResumeTsNs == nil { + // lastTs is the last settled entry's own timestamp; never rewind + // below the progress checkpoint it may lag. + if option.GetResumeTsNs == nil && lastTs > option.StartTsNs { option.StartTsNs = lastTs } pendingRefs = nil diff --git a/weed/pb/filer_pb_tail_test.go b/weed/pb/filer_pb_tail_test.go index 63909f4d4a..1fb62bce84 100644 --- a/weed/pb/filer_pb_tail_test.go +++ b/weed/pb/filer_pb_tail_test.go @@ -1,8 +1,10 @@ package pb import ( + "bytes" "context" "errors" + "fmt" "io" "sync" "sync/atomic" @@ -10,7 +12,9 @@ import ( "time" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" "google.golang.org/grpc" + "google.golang.org/protobuf/proto" ) // fakeSubscribeStream plays back a scripted sequence of responses, one per Recv. @@ -439,6 +443,54 @@ func TestFilerSyncReconnectReadsWatermarkEachSubscribe(t *testing.T) { } } +// A drained ref batch must leave StartTsNs at the settled checkpoint even when +// the last entry's own timestamp lags its log position, so the next drain does +// not re-read the settled prefix. +func TestChunkDrainCursorKeepsProgressCheckpoint(t *testing.T) { + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/watched", + TsNs: 100, + EventNotification: &filer_pb.EventNotification{ + NewEntry: &filer_pb.Entry{Name: "file"}, + }, + } + eventData, err := proto.Marshal(event) + if err != nil { + t.Fatalf("marshal event: %v", err) + } + entryData, err := (&filer_pb.LogEntry{TsNs: 300, Data: eventData}).MarshalVT() + if err != nil { + t.Fatalf("marshal log entry: %v", err) + } + var buf bytes.Buffer + var sizeBuf [4]byte + util.Uint32toBytes(sizeBuf[:], uint32(len(entryData))) + buf.Write(sizeBuf[:]) + buf.Write(entryData) + + option := &MetadataFollowOption{ + ClientName: "mount", + StartTsNs: 0, + PathPrefix: "/", + LogFileReaderFn: func(chunks []*filer_pb.FileChunk) (io.ReadCloser, error) { + return io.NopCloser(bytes.NewReader(buf.Bytes())), nil + }, + } + stream := &fakeSubscribeStream{responses: []*filer_pb.SubscribeMetadataResponse{ + {LogFileRefs: []*filer_pb.LogFileChunkRef{ + {FilerId: "f0", Chunks: []*filer_pb.FileChunk{{FileId: "1,01"}}}, + }}, + }} + if err := makeSubscribeMetadataFunc(option, func(*filer_pb.SubscribeMetadataResponse) error { + return nil + })(&fakeFilerClient{stream: stream}); err != nil { + t.Fatalf("follow: %v", err) + } + if option.StartTsNs != 300 { + t.Fatalf("expected StartTsNs at the settled checkpoint 300, got %d", option.StartTsNs) + } +} + // blockingFilerClient returns a stream whose Recv parks until the subscribe // context ends, standing in for a quiet source stream during a sink outage. type blockingFilerClient struct { @@ -487,3 +539,68 @@ func TestFilerSyncResubscribeSignalEndsStream(t *testing.T) { t.Fatal("follower did not end after the resubscribe signal") } } + +// A dropped stream mid-replay must keep the checkpoints the drain already +// settled: the resubscribe's SinceNs resumes at the last settled entry +// instead of re-reading the whole ref set. +func TestChunkReplayCheckpointSurvivesReconnect(t *testing.T) { + var events []*filer_pb.SubscribeMetadataResponse + for i, ts := range []int64{101, 102, 103, 104} { + events = append(events, makeSubEvent("/watched", fmt.Sprintf("f%d", i), ts)) + } + logData := buildLogFileData(events) + refsResp := &filer_pb.SubscribeMetadataResponse{ + LogFileRefs: []*filer_pb.LogFileChunkRef{{ + FilerId: "f0", + Chunks: []*filer_pb.FileChunk{{FileId: "log-1"}}, + }}, + } + markerResp := &filer_pb.SubscribeMetadataResponse{ + EventNotification: &filer_pb.EventNotification{}, + TsNs: 200, + } + option := &MetadataFollowOption{ + ClientName: "mount", + StartTsNs: 100, + PathPrefix: "/watched/", + LogFileReaderFn: func(chunks []*filer_pb.FileChunk) (io.ReadCloser, error) { + return io.NopCloser(bytes.NewReader(logData)), nil + }, + } + + var sinceNs []int64 + client := &recordingFilerClient{ + onSubscribe: func(req *filer_pb.SubscribeMetadataRequest) { + sinceNs = append(sinceNs, req.SinceNs) + }, + stream: &fakeSubscribeStream{responses: []*filer_pb.SubscribeMetadataResponse{refsResp, markerResp}}, + } + + count := 0 + wantErr := fmt.Errorf("apply boom") + err := makeSubscribeMetadataFunc(option, func(resp *filer_pb.SubscribeMetadataResponse) error { + count++ + if count == 3 { + return wantErr + } + return nil + })(client) + if !errors.Is(err, ErrLogFileRead) { + t.Fatalf("expected ErrLogFileRead, got %v", err) + } + // Entries 101 and 102 settled; 102's checkpoint emits when 103 pops, but + // 103 failed first, so the last checkpoint is 101. + if option.StartTsNs != 101 { + t.Fatalf("StartTsNs = %d, want 101", option.StartTsNs) + } + + client.stream = &fakeSubscribeStream{} + if err := makeSubscribeMetadataFunc(option, func(*filer_pb.SubscribeMetadataResponse) error { + return nil + })(client); err != nil { + t.Fatalf("resubscribe: %v", err) + } + if len(sinceNs) != 2 || sinceNs[0] != 100 || sinceNs[1] != 101 { + t.Fatalf("expected subscribes at 100 then 101, got %v", sinceNs) + } +} diff --git a/weed/server/filer_subscribe_loop_test.go b/weed/server/filer_subscribe_loop_test.go index f44b7ead20..a72e7e8fc6 100644 --- a/weed/server/filer_subscribe_loop_test.go +++ b/weed/server/filer_subscribe_loop_test.go @@ -705,7 +705,7 @@ func TestSubscribeLoop_ChunkModeMarkerMatchesClientReplay(t *testing.T) { func(resp *filer_pb.SubscribeMetadataResponse) error { applied = append(applied, resp.TsNs) return nil - }) + }, nil) if err != nil { t.Fatalf("client replay: %v", err) } @@ -778,7 +778,7 @@ func TestSubscribeLoop_ChunkModeDeadVolumeAgreesWithClient(t *testing.T) { func(resp *filer_pb.SubscribeMetadataResponse) error { applied = append(applied, resp.TsNs) return nil - }) + }, nil) if err != nil { t.Fatalf("client replay: %v", err) }