filer: stream offloaded metadata-log entries to fix concurrent-write OOM (#10203)

* filer: stream offloaded metadata-log entries instead of buffering whole files

The client metadata-chunks read path (ReadLogFileRefs, used by the meta
aggregator to consume peer filers and by mounts) decoded every entry of a
log file into a slice before handing it to the consumer, and prefetched the
next whole file the same way. Peak memory scaled with log-file size: under
heavy concurrent writes the per-event chunk lists grow and minute-files reach
hundreds of MB to GBs, so a filer aggregating a few peers held many GBs of
decoded entries at once (heap dominated by readLogFileEntries ->
consumeBytesNoZero) and OOMed.

Stream entries through a bounded channel: a producer decodes one entry at a
time and the next file's read overlaps processing via the channel buffer, so
peak memory is bounded by the channel depth rather than O(file size). In a
synthetic replay peak live heap dropped from ~1.3x the file size to a flat
few MB regardless of file size.

* filer: tighten offload replay tests

Share one ordered-replay assertion between the merge-order and single-filer
tests, assert the callback's own error is what propagates, and drop atomic
counters from callbacks that run on a single goroutine.

* filer: abort offloaded log replay promptly instead of joining wedged readers

Collapse the single-filer fast path into the merged reader: it was a second
copy of the producer/stop lifecycle with its own subtler synchronization, and
a one-stream merge does the same job.

On abort (processing error or a fatal read error from one filer), the
consumer used to drain channels and join every producer. A producer blocked
in an uncancellable chunk read cannot observe stop until that read returns,
so an abort could stall the caller's retry loop behind a dead volume-server
connection. Closing stop is now the only cleanup: producers check it at every
send and file boundary and exit on their own, and the merge loop's blocking
receives also escape on stop. Producers also check stop before opening each
file, so an aborted replay no longer keeps reading remaining files whose
entries never reach the channel.

A mid-file chunk-not-found still skips to the next file, but the log line now
reports how many entries were delivered first instead of pretending the whole
file was skipped; the redundant error log before setFatal is gone since the
error propagates to callers that already log it.

* filer: cap offloaded log entry allocation against corrupt size prefix

A garbage 4-byte size prefix (torn chunk or stream desync) drove
make([]byte, size) up to 4GiB per entry. Reject sizes above the same 1GiB
bound the filer-side log readers enforce.

weed/pb/filer_pb_direct_read.go

@@ -2,6 +2,7 @@ package pb
 
 import (
 	"container/heap"
+	"errors"
 	"fmt"
 	"io"
 	"strings"
@@ -17,6 +18,17 @@ import (
 // LogFileReaderFn creates an io.ReadCloser for a set of file chunks.
 type LogFileReaderFn func(chunks []*filer_pb.FileChunk) (io.ReadCloser, error)
 
+// logEntryChannelSize bounds decoded entries in flight per filer stream.
+const logEntryChannelSize = 512
+
+// maxLogEntrySize guards the per-entry allocation against a corrupt size
+// prefix, mirroring the filer package's unexported constant.
+const maxLogEntrySize = 1 << 30
+
+// errReaderStopped signals that the entry consumer asked to stop (the merge
+// loop aborted or the caller hit a processing error). It is not a read failure.
+var errReaderStopped = errors.New("log entry reader stopped")
+
 // PathFilter holds subscription path filtering parameters, matching the
 // server-side eachEventNotificationFn filtering logic.
 type PathFilter struct {
@@ -30,8 +42,8 @@ type PathFilter struct {
 // (same algorithm as the server's OrderedLogVisitor), applies path filtering,
 // and invokes processEventFn for each matching event.
 //
-// Filers are read in parallel (one goroutine per filer). Within each filer,
-// the next file is prefetched while the current file's entries are consumed.
+// Filers are read in parallel (one goroutine per filer), each streaming
+// entries through a bounded channel.
 func ReadLogFileRefs(
 	refs []*filer_pb.LogFileChunkRef,
 	newReader LogFileReaderFn,
@@ -61,73 +73,14 @@ func ReadLogFileRefs(
 		return
 	}
 
-	// Single filer fast path: no merge heap needed.
-	if len(filerOrder) == 1 {
-		return readFilerFilesWithPrefetch(perFiler[filerOrder[0]], newReader, startTsNs, stopTsNs, filter, processEventFn)
-	}
-
-	// Multiple filers: read each in parallel with prefetching, merge via min-heap.
-	return readMultiFilersMerged(filerOrder, perFiler, newReader, startTsNs, stopTsNs, filter, processEventFn)
+	return readFilersMerged(filerOrder, perFiler, newReader, startTsNs, stopTsNs, filter, processEventFn)
 }
 
-// readFilerFilesWithPrefetch reads files for a single filer, prefetching the
-// next file while processing entries from the current one.
-func readFilerFilesWithPrefetch(
-	refs []*filer_pb.LogFileChunkRef,
-	newReader LogFileReaderFn,
-	startTsNs, stopTsNs int64,
-	filter PathFilter,
-	processEventFn ProcessMetadataFunc,
-) (lastTsNs int64, err error) {
-
-	type prefetchResult struct {
-		entries []*filer_pb.LogEntry
-		err     error
-	}
-
-	startPrefetch := func(ref *filer_pb.LogFileChunkRef) chan prefetchResult {
-		ch := make(chan prefetchResult, 1)
-		go func() {
-			entries, readErr := readLogFileEntries(newReader, ref.Chunks, startTsNs, stopTsNs)
-			ch <- prefetchResult{entries, readErr}
-		}()
-		return ch
-	}
-
-	var pendingCh chan prefetchResult
-	if len(refs) > 0 {
-		pendingCh = startPrefetch(refs[0])
-	}
-
-	for i, ref := range refs {
-		result := <-pendingCh
-
-		// Start prefetching next file while we process current
-		if i+1 < len(refs) {
-			pendingCh = startPrefetch(refs[i+1])
-		}
-
-		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)
-				continue
-			}
-			return lastTsNs, fmt.Errorf("read log file filer=%s ts=%d: %w", ref.FilerId, ref.FileTsNs, result.err)
-		}
-
-		for _, logEntry := range result.entries {
-			lastTsNs, err = processOneLogEntry(logEntry, filter, processEventFn)
-			if err != nil {
-				return
-			}
-		}
-	}
-	return
-}
-
-// readMultiFilersMerged reads files from multiple filers in parallel (one goroutine
-// per filer with prefetching), then merges entries in timestamp order via min-heap.
-func readMultiFilersMerged(
+// readFilersMerged reads files from each filer in parallel (one producer
+// goroutine per filer streaming decoded entries through a bounded channel),
+// then merges entries in timestamp order via min-heap. A single filer is the
+// degenerate one-stream merge.
+func readFilersMerged(
 	filerOrder []string,
 	perFiler map[string][]*filer_pb.LogFileChunkRef,
 	newReader LogFileReaderFn,
@@ -142,14 +95,19 @@ 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.
+	//
+	// Closing stop is the only cleanup: producers observe it at every send and
+	// every file boundary and exit on their own. An abort must not wait for
+	// them — a producer can be wedged in an uncancellable chunk read, and
+	// joining it would stall the caller's retry loop behind a dead connection.
 	stop := make(chan struct{})
 	var stopOnce sync.Once
 	closeStop := func() { stopOnce.Do(func() { close(stop) }) }
+	defer closeStop()
 	var fatalMu sync.Mutex
 	var fatalErr error
 	setFatal := func(e error) {
@@ -160,35 +118,33 @@ func readMultiFilersMerged(
 		fatalMu.Unlock()
 		closeStop()
 	}
+	fatal := func() error {
+		fatalMu.Lock()
+		defer fatalMu.Unlock()
+		return fatalErr
+	}
 
 	for i, filerId := range filerOrder {
-		entryCh := make(chan *filer_pb.LogEntry, 512)
+		entryCh := make(chan *filer_pb.LogEntry, logEntryChannelSize)
 		streams[i] = filerStream{filerId: filerId, entryCh: entryCh}
 
-		wg.Add(1)
 		go func(refs []*filer_pb.LogFileChunkRef, ch chan *filer_pb.LogEntry) {
-			defer wg.Done()
 			defer close(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)
 	for i := range streams {
-		if entry, ok := <-streams[i].entryCh; ok {
-			heap.Push(pq, &logEntryHeapItem{entry: entry, filerIdx: i})
+		select {
+		case entry, ok := <-streams[i].entryCh:
+			if ok {
+				heap.Push(pq, &logEntryHeapItem{entry: entry, filerIdx: i})
+			}
+		case <-stop:
+			return lastTsNs, fatal()
 		}
 	}
 
@@ -198,11 +154,7 @@ func readMultiFilersMerged(
 		// aborted; lock-free bail on the hot path.
 		select {
 		case <-stop:
-			drainAndWait()
-			fatalMu.Lock()
-			fe := fatalErr
-			fatalMu.Unlock()
-			return lastTsNs, fe
+			return lastTsNs, fatal()
 		default:
 		}
 
@@ -210,20 +162,22 @@ func readMultiFilersMerged(
 
 		lastTsNs, err = processOneLogEntry(item.entry, filter, processEventFn)
 		if err != nil {
-			drainAndWait()
 			return
 		}
 
-		if entry, ok := <-streams[item.filerIdx].entryCh; ok {
-			heap.Push(pq, &logEntryHeapItem{entry: entry, filerIdx: item.filerIdx})
+		select {
+		case entry, ok := <-streams[item.filerIdx].entryCh:
+			if ok {
+				heap.Push(pq, &logEntryHeapItem{entry: entry, filerIdx: item.filerIdx})
+			}
+		case <-stop:
+			return lastTsNs, fatal()
 		}
 	}
 
-	drainAndWait()
-	fatalMu.Lock()
-	fe := fatalErr
-	fatalMu.Unlock()
-	if fe != nil {
+	// All channels closed: every producer has finished (setFatal, if any,
+	// happened before its close).
+	if fe := fatal(); fe != nil {
 		return lastTsNs, fe
 	}
 	return
@@ -237,53 +191,45 @@ func readFilerFilesToChannel(
 	stop <-chan struct{},
 	setFatal func(error),
 ) {
-	type prefetchResult struct {
-		entries []*filer_pb.LogEntry
-		err     error
+	var sent int
+	sendEntry := func(entry *filer_pb.LogEntry) error {
+		// Prefer stop over a send the buffer could still absorb, so an aborted
+		// producer quits at the next entry instead of filling the channel.
+		select {
+		case <-stop:
+			return errReaderStopped
+		default:
+		}
+		select {
+		case ch <- entry:
+			sent++
+			return nil
+		case <-stop:
+			return errReaderStopped
+		}
 	}
 
-	startPrefetch := func(ref *filer_pb.LogFileChunkRef) chan prefetchResult {
-		resultCh := make(chan prefetchResult, 1)
-		go func() {
-			entries, err := readLogFileEntries(newReader, ref.Chunks, startTsNs, stopTsNs)
-			resultCh <- prefetchResult{entries, err}
-		}()
-		return resultCh
-	}
-
-	var pendingCh chan prefetchResult
-	if len(refs) > 0 {
-		pendingCh = startPrefetch(refs[0])
-	}
-
-	for i, ref := range refs {
-		var result prefetchResult
+	for _, ref := range refs {
 		select {
-		case result = <-pendingCh:
 		case <-stop:
 			return
+		default:
 		}
 
-		if i+1 < len(refs) {
-			pendingCh = startPrefetch(refs[i+1])
-		}
-
-		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)
-				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))
+		sentBefore := sent
+		streamErr := streamLogFileEntries(newReader, ref.Chunks, startTsNs, stopTsNs, sendEntry)
+		if streamErr == errReaderStopped {
 			return
 		}
-
-		for _, entry := range result.entries {
-			select {
-			case ch <- entry:
-			case <-stop:
-				return
+		if streamErr != nil {
+			if isChunkNotFound(streamErr) {
+				// A mid-file not-found still delivered the readable prefix; say so
+				// rather than pretending the whole file was skipped.
+				glog.V(0).Infof("skip log file filer=%s ts=%d after %d entries: %v", ref.FilerId, ref.FileTsNs, sent-sentBefore, streamErr)
+				continue
 			}
+			setFatal(fmt.Errorf("read log file filer=%s ts=%d: %w", ref.FilerId, ref.FileTsNs, streamErr))
+			return
 		}
 	}
 }
@@ -352,45 +298,50 @@ func (h *logEntryHeap) Pop() any {
 
 // --- log file parsing (uses io.ReadFull for correct partial-read handling) ---
 
-func readLogFileEntries(newReader LogFileReaderFn, chunks []*filer_pb.FileChunk, startTsNs, stopTsNs int64) ([]*filer_pb.LogEntry, error) {
+// streamLogFileEntries reads a log file's chunks and invokes eachFn for every
+// entry as it is decoded, without buffering the whole file: a multi-GB log
+// file costs one entry plus the chunk reader's buffer at a time, not O(file
+// size). An eachFn error stops the read and is propagated verbatim.
+func streamLogFileEntries(newReader LogFileReaderFn, chunks []*filer_pb.FileChunk, startTsNs, stopTsNs int64, eachFn func(*filer_pb.LogEntry) error) error {
 	reader, err := newReader(chunks)
 	if err != nil {
-		return nil, fmt.Errorf("create reader: %w", err)
+		return fmt.Errorf("create reader: %w", err)
 	}
 	defer reader.Close()
 
 	sizeBuf := make([]byte, 4)
-	var entries []*filer_pb.LogEntry
 
 	for {
-		_, err := io.ReadFull(reader, sizeBuf)
-		if err != nil {
+		if _, err := io.ReadFull(reader, sizeBuf); err != nil {
 			if err == io.EOF || err == io.ErrUnexpectedEOF {
-				break
+				return nil
 			}
-			return entries, err
+			return err
 		}
 
 		size := util.BytesToUint32(sizeBuf)
+		if size > maxLogEntrySize {
+			return fmt.Errorf("entry size %d exceeds %d", size, maxLogEntrySize)
+		}
 		entryData := make([]byte, size)
-		_, err = io.ReadFull(reader, entryData)
-		if err != nil {
-			return entries, err
+		if _, err := io.ReadFull(reader, entryData); err != nil {
+			return err
 		}
 
 		logEntry := &filer_pb.LogEntry{}
-		if err = proto.Unmarshal(entryData, logEntry); err != nil {
-			return entries, err
+		if err := proto.Unmarshal(entryData, logEntry); err != nil {
+			return err
 		}
 
 		if logEntry.TsNs <= startTsNs {
 			continue
 		}
 		if stopTsNs != 0 && logEntry.TsNs > stopTsNs {
-			break
+			return nil
 		}
 
-		entries = append(entries, logEntry)
+		if err := eachFn(logEntry); err != nil {
+			return err
+		}
 	}
-	return entries, nil
 }

weed/pb/filer_pb_direct_read_test.go

@@ -2,8 +2,10 @@ package pb
 
 import (
 	"bytes"
+	"errors"
 	"fmt"
 	"io"
+	"strings"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -145,10 +147,10 @@ func (t *testLogFiles) totalEvents() int {
 	return total
 }
 
-// TestReadLogFileRefsMergeOrder verifies that entries from multiple filers are
-// delivered in correct timestamp order.
-func TestReadLogFileRefsMergeOrder(t *testing.T) {
-	files := newTestLogFiles(3, 2, 50, 0)
+// assertOrderedReplay reads all refs and checks every event arrives exactly
+// once, in timestamp order.
+func assertOrderedReplay(t *testing.T, files *testLogFiles) {
+	t.Helper()
 
 	var timestamps []int64
 	_, err := ReadLogFileRefs(files.refs, files.readerFn(), 0, 0,
@@ -161,20 +163,21 @@ func TestReadLogFileRefsMergeOrder(t *testing.T) {
 		t.Fatalf("ReadLogFileRefs: %v", err)
 	}
 
-	expected := files.totalEvents()
-	if len(timestamps) != expected {
-		t.Fatalf("expected %d events, got %d", expected, len(timestamps))
+	if got, want := len(timestamps), files.totalEvents(); got != want {
+		t.Fatalf("expected %d events, got %d", want, got)
 	}
-
 	for i := 1; i < len(timestamps); i++ {
 		if timestamps[i] < timestamps[i-1] {
-			t.Errorf("out of order at index %d: ts[%d]=%d > ts[%d]=%d",
+			t.Fatalf("out of order at index %d: ts[%d]=%d > ts[%d]=%d",
 				i, i-1, timestamps[i-1], i, timestamps[i])
-			break
 		}
 	}
+}
 
-	t.Logf("Verified %d events from 3 filers in correct timestamp order", len(timestamps))
+// TestReadLogFileRefsMergeOrder verifies that entries from multiple filers are
+// delivered in correct timestamp order.
+func TestReadLogFileRefsMergeOrder(t *testing.T) {
+	assertOrderedReplay(t, newTestLogFiles(3, 2, 50, 0))
 }
 
 // TestReadLogFileRefsPathFilter verifies path filtering including system log exclusion.
@@ -218,7 +221,7 @@ func TestReadLogFileRefsPathFilter(t *testing.T) {
 
 // TestDirectReadVsServerSideThroughput compares:
 //   - Server-side: sequential file read → gRPC send per event
-//   - Client direct-read: parallel filers + prefetching + no gRPC
+//   - Client direct-read: parallel filers + streaming + no gRPC
 func TestDirectReadVsServerSideThroughput(t *testing.T) {
 	const (
 		numFilers     = 3
@@ -255,7 +258,7 @@ func TestDirectReadVsServerSideThroughput(t *testing.T) {
 	})
 
 	var directRate float64
-	t.Run("client_direct_read_parallel_prefetch", func(t *testing.T) {
+	t.Run("client_direct_read_parallel_streaming", func(t *testing.T) {
 		var processed int64
 		start := time.Now()
 
@@ -270,12 +273,12 @@ func TestDirectReadVsServerSideThroughput(t *testing.T) {
 		}
 		elapsed := time.Since(start)
 		directRate = float64(processed) / elapsed.Seconds()
-		t.Logf("direct-read: %d events  %v  %6.0f events/sec  (%d filers parallel + prefetch, no gRPC)",
+		t.Logf("direct-read: %d events  %v  %6.0f events/sec  (%d filers parallel + streaming, no gRPC)",
 			processed, elapsed.Round(time.Millisecond), directRate, numFilers)
 	})
 
 	if serverRate > 0 {
-		t.Logf("Speedup: %.1fx (parallel + prefetch + no gRPC vs server-side sequential)", directRate/serverRate)
+		t.Logf("Speedup: %.1fx (parallel + streaming + no gRPC vs server-side sequential)", directRate/serverRate)
 	}
 }
 
@@ -329,3 +332,120 @@ func TestReadLogFileRefsMultiFilerNotFoundSkips(t *testing.T) {
 		t.Fatalf("expected %d events after skipping one file, got %d", expected, count)
 	}
 }
+
+// TestReadLogFileRefsSingleFilerOrder covers the single-filer path: every
+// entry across all files, in order.
+func TestReadLogFileRefsSingleFilerOrder(t *testing.T) {
+	assertOrderedReplay(t, newTestLogFiles(1, 4, 50, 0))
+}
+
+// TestReadLogFileRefsSingleFilerProcessErrorStops verifies that the callback's
+// own error propagates and aborts the stream promptly, mid-file.
+func TestReadLogFileRefsSingleFilerProcessErrorStops(t *testing.T) {
+	files := newTestLogFiles(1, 3, 100, 0)
+
+	var count int
+	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
+			}
+			return nil
+		})
+	if !errors.Is(err, wantErr) {
+		t.Fatalf("expected processing error to propagate, got: %v", err)
+	}
+	// Should stop near the failing event, not process the whole 300-event set.
+	if count > 20 {
+		t.Fatalf("expected prompt stop after error, processed %d events", count)
+	}
+}
+
+// A corrupt size prefix must fail the replay instead of allocating gigabytes.
+func TestReadLogFileRefsCorruptSizePrefix(t *testing.T) {
+	data := make([]byte, 4)
+	util.Uint32toBytes(data, 0xFFFFFFF0)
+	refs := []*filer_pb.LogFileChunkRef{{
+		Chunks:   []*filer_pb.FileChunk{{FileId: "corrupt"}},
+		FileTsNs: 1,
+		FilerId:  "filer00",
+	}}
+	readerFn := func(chunks []*filer_pb.FileChunk) (io.ReadCloser, error) {
+		return io.NopCloser(bytes.NewReader(data)), nil
+	}
+
+	_, err := ReadLogFileRefs(refs, readerFn, 0, 0, PathFilter{PathPrefix: "/"},
+		func(*filer_pb.SubscribeMetadataResponse) error { return nil })
+	if err == nil || !strings.Contains(err.Error(), "exceeds") {
+		t.Fatalf("expected size-cap error, got: %v", err)
+	}
+}
+
+// blockingReader blocks in Read until released.
+type blockingReader struct{ release chan struct{} }
+
+func (r *blockingReader) Read(p []byte) (int, error) { <-r.release; return 0, io.EOF }
+func (r *blockingReader) Close() error               { return nil }
+
+// An abort (fatal error on one filer) must not wait for another filer's
+// in-flight chunk read: the replay returns promptly and the wedged producer
+// exits on its own once its read completes.
+func TestReadLogFileRefsAbortDoesNotJoinWedgedReader(t *testing.T) {
+	files := newTestLogFiles(2, 1, 10, 0)
+	release := make(chan struct{})
+	t.Cleanup(func() { close(release) })
+
+	wedgedKey := files.refs[0].Chunks[0].FileId // filer00 wedges mid-read
+	failKey := files.refs[1].Chunks[0].FileId   // filer01 fails for real
+	base := files.readerFn()
+	readerFn := func(chunks []*filer_pb.FileChunk) (io.ReadCloser, error) {
+		switch chunks[0].FileId {
+		case wedgedKey:
+			return &blockingReader{release: release}, nil
+		case failKey:
+			return nil, fmt.Errorf("failed to locate %s", failKey)
+		}
+		return base(chunks)
+	}
+
+	done := make(chan error, 1)
+	go func() {
+		_, err := ReadLogFileRefs(files.refs, readerFn, 0, 0, PathFilter{PathPrefix: "/"},
+			func(*filer_pb.SubscribeMetadataResponse) error { return nil })
+		done <- err
+	}()
+
+	select {
+	case err := <-done:
+		if err == nil {
+			t.Fatalf("expected the fatal read error to propagate")
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatalf("ReadLogFileRefs did not return while a peer reader was wedged")
+	}
+}
+
+// TestReadLogFileRefsSingleFilerNotFoundSkips confirms a chunk-not-found on the
+// single-filer path skips just that file, not the whole replay.
+func TestReadLogFileRefsSingleFilerNotFoundSkips(t *testing.T) {
+	files := newTestLogFiles(1, 3, 10, 0)
+	skipKey := files.refs[1].Chunks[0].FileId // second file
+	readerFn := failingReaderFn(files.readerFn(), skipKey, fmt.Errorf("volume not found: %s", skipKey))
+
+	var count int
+	_, err := ReadLogFileRefs(files.refs, readerFn, 0, 0,
+		PathFilter{PathPrefix: "/"},
+		func(resp *filer_pb.SubscribeMetadataResponse) error {
+			count++
+			return nil
+		})
+	if err != nil {
+		t.Fatalf("chunk-not-found should be skipped, got error: %v", err)
+	}
+	if want := files.totalEvents() - 10; count != want {
+		t.Fatalf("expected %d events after skipping one file, got %d", want, count)
+	}
+}