Fix filer metadata-replay OOM under mount reconnect storms (#9901)

* fix(filer): propagate multi-filer metadata log read errors

A genuine (non not-found) read error in one filer's log stream was logged
and skipped, then the merged cursor advanced past the gap, silently
dropping that file's events. Abort the whole replay so the subscriber
re-reads from the unchanged position; chunk-not-found still skips.

* perf(mount): read persisted metadata log chunks directly from volume servers

Set LogFileReaderFn so the filer returns log file references and the mount
reads the chunk data itself, instead of the filer reading, decoding, and
streaming every persisted entry. Keeps a reconnect storm of many mounts
from concentrating hundreds of concurrent log replays in filer memory.

* perf(filer): pre-size chunk stream reader buffer to view size

The chunk size is known up front, so grow the buffer once instead of
letting bytes.Buffer double as the streamed pieces arrive (which
transiently overshoots to ~2x per reader).

* fix(filer): bound concurrent persisted-log replays

Each server-side replay holds an open chunk reader per source filer plus a
readahead buffer, so a reconnect storm of clients that predate the
metadata-chunks offload multiplies into many GB. Gate replays with a
semaphore; abort the acquire when the subscriber's stream is gone so
cancelled clients do not pile up parked goroutines.

weed/filer/filer_notify.go

@@ -200,7 +200,27 @@ func isChunkNotFoundError(err error) bool {
 		httpNotFoundPattern.MatchString(errMsg)
 }
 
-func (f *Filer) ReadPersistedLogBuffer(startPosition log_buffer.MessagePosition, stopTsNs int64, eachLogEntryFn log_buffer.EachLogEntryFuncType) (lastTsNs int64, isDone bool, err error) {
+// persistedLogReplayLimit caps concurrent legacy replays; each holds a chunk
+// reader per source filer, so a reconnect storm of pre-offload clients would
+// otherwise pin many GB. Metadata-chunks clients take sendLogFileRefs and never
+// reach this path.
+const persistedLogReplayLimit = 16
+
+var persistedLogReplaySem = make(chan struct{}, persistedLogReplayLimit)
+
+func (f *Filer) ReadPersistedLogBuffer(ctx context.Context, startPosition log_buffer.MessagePosition, stopTsNs int64, eachLogEntryFn log_buffer.EachLogEntryFuncType) (lastTsNs int64, isDone bool, err error) {
+
+	// Cap concurrent replays; bail if the stream is already gone so cancelled
+	// clients do not park on the semaphore.
+	if err := ctx.Err(); err != nil {
+		return 0, false, err
+	}
+	select {
+	case persistedLogReplaySem <- struct{}{}:
+		defer func() { <-persistedLogReplaySem }()
+	case <-ctx.Done():
+		return 0, false, ctx.Err()
+	}
 
 	visitor, visitErr := f.collectPersistedLogBuffer(startPosition, stopTsNs)
 	if visitErr != nil {

weed/filer/stream.go

@@ -484,6 +484,8 @@ func (c *ChunkStreamReader) fetchChunkToBuffer(chunkView *ChunkView) error {
 		return err
 	}
 	var buffer bytes.Buffer
+	// pre-size to the known chunk size; avoids bytes.Buffer's doubling regrowth
+	buffer.Grow(int(chunkView.ViewSize))
 	var shouldRetry bool
 	jwt := JwtForVolumeServer(chunkView.FileId)
 	for _, urlString := range urlStrings {

weed/mount/meta_cache/meta_cache_subscribe.go

@@ -2,8 +2,10 @@ package meta_cache
 
 import (
 	"context"
+	"io"
 	"strings"
 
+	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -66,6 +68,9 @@ func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.Fil
 		prefix = prefix + "/"
 	}
 
+	// Read persisted log chunks directly from volume servers, keeping the replay
+	// cost off the filer's heap (see LogFileReaderFn below).
+	lookupFn := filer.LookupFn(client)
 	metadataFollowOption := &pb.MetadataFollowOption{
 		ClientName:             "mount",
 		ClientId:               selfSignature,
@@ -77,6 +82,9 @@ func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.Fil
 		StartTsNs:              lastTsNs,
 		StopTsNs:               0,
 		EventErrorType:         pb.FatalOnError,
+		LogFileReaderFn: func(chunks []*filer_pb.FileChunk) (io.ReadCloser, error) {
+			return filer.NewChunkStreamReaderFromLookup(context.Background(), lookupFn, chunks), nil
+		},
 	}
 	util.RetryUntil("followMetaUpdates", func() error {
 		metadataFollowOption.ClientEpoch++

weed/pb/filer_pb_direct_read.go

@@ -144,6 +144,23 @@ func readMultiFilersMerged(
 	streams := make([]filerStream, len(filerOrder))
 	var wg sync.WaitGroup
 
+	// A genuine (non chunk-not-found) read error must fail the whole replay: the
+	// caller advances its cursor only on success, so a swallowed error leaves a
+	// permanent gap. stop aborts the other readers; fatalErr is read on exit.
+	stop := make(chan struct{})
+	var stopOnce sync.Once
+	closeStop := func() { stopOnce.Do(func() { close(stop) }) }
+	var fatalMu sync.Mutex
+	var fatalErr error
+	setFatal := func(e error) {
+		fatalMu.Lock()
+		if fatalErr == nil {
+			fatalErr = e
+		}
+		fatalMu.Unlock()
+		closeStop()
+	}
+
 	for i, filerId := range filerOrder {
 		entryCh := make(chan *filer_pb.LogEntry, 512)
 		streams[i] = filerStream{filerId: filerId, entryCh: entryCh}
@@ -152,10 +169,20 @@ func readMultiFilersMerged(
 		go func(refs []*filer_pb.LogFileChunkRef, ch chan *filer_pb.LogEntry) {
 			defer wg.Done()
 			defer close(ch)
-			readFilerFilesToChannel(refs, newReader, startTsNs, stopTsNs, ch)
+			readFilerFilesToChannel(refs, newReader, startTsNs, stopTsNs, ch, stop, setFatal)
 		}(perFiler[filerId], entryCh)
 	}
 
+	// Stop readers, drain channels so none block on a send, then wait for exit.
+	drainAndWait := func() {
+		closeStop()
+		for i := range streams {
+			for range streams[i].entryCh {
+			}
+		}
+		wg.Wait()
+	}
+
 	// Seed the min-heap with the first entry from each filer
 	pq := &logEntryHeap{}
 	heap.Init(pq)
@@ -167,15 +194,23 @@ func readMultiFilersMerged(
 
 	// Merge loop
 	for pq.Len() > 0 {
+		// stop is closed only by setFatal here, so a closed stop means a reader
+		// aborted; lock-free bail on the hot path.
+		select {
+		case <-stop:
+			drainAndWait()
+			fatalMu.Lock()
+			fe := fatalErr
+			fatalMu.Unlock()
+			return lastTsNs, fe
+		default:
+		}
+
 		item := heap.Pop(pq).(*logEntryHeapItem)
 
 		lastTsNs, err = processOneLogEntry(item.entry, filter, processEventFn)
 		if err != nil {
-			for i := range streams {
-				for range streams[i].entryCh {
-				}
-			}
-			wg.Wait()
+			drainAndWait()
 			return
 		}
 
@@ -184,7 +219,13 @@ func readMultiFilersMerged(
 		}
 	}
 
-	wg.Wait()
+	drainAndWait()
+	fatalMu.Lock()
+	fe := fatalErr
+	fatalMu.Unlock()
+	if fe != nil {
+		return lastTsNs, fe
+	}
 	return
 }
 
@@ -193,6 +234,8 @@ func readFilerFilesToChannel(
 	newReader LogFileReaderFn,
 	startTsNs, stopTsNs int64,
 	ch chan *filer_pb.LogEntry,
+	stop <-chan struct{},
+	setFatal func(error),
 ) {
 	type prefetchResult struct {
 		entries []*filer_pb.LogEntry
@@ -214,7 +257,12 @@ func readFilerFilesToChannel(
 	}
 
 	for i, ref := range refs {
-		result := <-pendingCh
+		var result prefetchResult
+		select {
+		case result = <-pendingCh:
+		case <-stop:
+			return
+		}
 
 		if i+1 < len(refs) {
 			pendingCh = startPrefetch(refs[i+1])
@@ -223,14 +271,19 @@ func readFilerFilesToChannel(
 		if result.err != nil {
 			if isChunkNotFound(result.err) {
 				glog.V(0).Infof("skip log file filer=%s ts=%d: %v", ref.FilerId, ref.FileTsNs, result.err)
-			} else {
-				glog.Errorf("read log file filer=%s ts=%d: %v", ref.FilerId, ref.FileTsNs, result.err)
+				continue
 			}
-			continue
+			glog.Errorf("read log file filer=%s ts=%d: %v", ref.FilerId, ref.FileTsNs, result.err)
+			setFatal(fmt.Errorf("read log file filer=%s ts=%d: %w", ref.FilerId, ref.FileTsNs, result.err))
+			return
 		}
 
 		for _, entry := range result.entries {
-			ch <- entry
+			select {
+			case ch <- entry:
+			case <-stop:
+				return
+			}
 		}
 	}
 }

weed/pb/filer_pb_direct_read_test.go

@@ -278,3 +278,54 @@ func TestDirectReadVsServerSideThroughput(t *testing.T) {
 		t.Logf("Speedup: %.1fx (parallel + prefetch + no gRPC vs server-side sequential)", directRate/serverRate)
 	}
 }
+
+// failingReaderFn returns err for the given file key, delegating otherwise.
+func failingReaderFn(base LogFileReaderFn, failKey string, err error) LogFileReaderFn {
+	return func(chunks []*filer_pb.FileChunk) (io.ReadCloser, error) {
+		if len(chunks) > 0 && chunks[0].FileId == failKey {
+			return nil, err
+		}
+		return base(chunks)
+	}
+}
+
+// A real (non not-found) read error must fail the whole replay, not silently
+// drop the file and advance the cursor.
+func TestReadLogFileRefsMultiFilerGenuineErrorAborts(t *testing.T) {
+	files := newTestLogFiles(3, 2, 10, 0)
+	failKey := files.refs[2].Chunks[0].FileId // filer01's first file
+	readerFn := failingReaderFn(files.readerFn(), failKey, fmt.Errorf("failed to locate %s", failKey))
+
+	var count int64
+	_, err := ReadLogFileRefs(files.refs, readerFn, 0, 0,
+		PathFilter{PathPrefix: "/"},
+		func(resp *filer_pb.SubscribeMetadataResponse) error {
+			atomic.AddInt64(&count, 1)
+			return nil
+		})
+	if err == nil {
+		t.Fatalf("expected error from genuine read failure, got nil (delivered=%d)", count)
+	}
+}
+
+// A chunk-not-found error skips only that file (volume gone), not the replay.
+func TestReadLogFileRefsMultiFilerNotFoundSkips(t *testing.T) {
+	files := newTestLogFiles(3, 2, 10, 0)
+	skipKey := files.refs[2].Chunks[0].FileId // filer01's first file
+	readerFn := failingReaderFn(files.readerFn(), skipKey, fmt.Errorf("volume not found: %s", skipKey))
+
+	var count int64
+	_, err := ReadLogFileRefs(files.refs, readerFn, 0, 0,
+		PathFilter{PathPrefix: "/"},
+		func(resp *filer_pb.SubscribeMetadataResponse) error {
+			atomic.AddInt64(&count, 1)
+			return nil
+		})
+	if err != nil {
+		t.Fatalf("chunk-not-found should be skipped, got error: %v", err)
+	}
+	expected := int64(files.totalEvents() - 10) // one skipped file's events
+	if count != expected {
+		t.Fatalf("expected %d events after skipping one file, got %d", expected, count)
+	}
+}

weed/server/filer_grpc_server_sub_meta.go

@@ -215,7 +215,7 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 		if req.ClientSupportsMetadataChunks {
 			processedTsNs, isDone, readPersistedLogErr = fs.sendLogFileRefs(ctx, stream, lastReadTime, req.UntilNs)
 		} else {
-			processedTsNs, isDone, readPersistedLogErr = fs.filer.ReadPersistedLogBuffer(lastReadTime, req.UntilNs, eachLogEntryFn)
+			processedTsNs, isDone, readPersistedLogErr = fs.filer.ReadPersistedLogBuffer(ctx, lastReadTime, req.UntilNs, eachLogEntryFn)
 		}
 		if readPersistedLogErr != nil {
 			return fmt.Errorf("reading from persisted logs: %w", readPersistedLogErr)
@@ -367,7 +367,7 @@ func (fs *FilerServer) SubscribeLocalMetadata(req *filer_pb.SubscribeMetadataReq
 			if req.ClientSupportsMetadataChunks {
 				processedTsNs, isDone, readPersistedLogErr = fs.sendLogFileRefs(ctx, stream, lastReadTime, req.UntilNs)
 			} else {
-				processedTsNs, isDone, readPersistedLogErr = fs.filer.ReadPersistedLogBuffer(lastReadTime, req.UntilNs, eachLogEntryFn)
+				processedTsNs, isDone, readPersistedLogErr = fs.filer.ReadPersistedLogBuffer(ctx, lastReadTime, req.UntilNs, eachLogEntryFn)
 			}
 			if readPersistedLogErr != nil {
 				glog.V(0).Infof("read on disk %v local subscribe %s from %+v: %v", clientName, req.PathPrefix, lastReadTime, readPersistedLogErr)