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

@@ -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

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
 }
 

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

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

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

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