fix(filer): eliminate redundant disk reads causing memory/CPU regression (#9039)

* fix(filer): eliminate redundant disk reads causing memory/CPU regression (#9035)

Since 4.18, LocalMetaLogBuffer's ReadFromDiskFn was set to
readPersistedLogBufferPosition, causing LoopProcessLogData to call
ReadPersistedLogBuffer on every 250ms health-check tick when a
subscriber encounters ResumeFromDiskError.  Each call creates an
OrderedLogVisitor (ListDirectoryEntries on the filer store), spawns a
readahead goroutine with a 1024-element channel, finds no data, and
returns — 4 times per second even on an idle filer.

This is redundant because SubscribeLocalMetadata already manages disk
reads explicitly with its own shouldReadFromDisk / lastCheckedFlushTsNs
tracking in the outer loop.

Set ReadFromDiskFn back to nil for LocalMetaLogBuffer.  When
LoopProcessLogData encounters ResumeFromDiskError with nil
ReadFromDiskFn, the HasData() guard returns ResumeFromDiskError to the
caller (SubscribeLocalMetadata), which blocks efficiently on
listenersCond.Wait() instead of polling.

* fix(filer): add gap detection for slow consumers after disk-read stall

When a slow consumer falls behind and LoopProcessLogData returns
ResumeFromDiskError with no flush or read-position progress, there may
be a gap between persisted data and in-memory data (e.g. writes stopped
while consumer was still catching up). Without this, the consumer would
block on listenersCond.Wait() forever.

Skip forward to the earliest in-memory time to resume progress, matching
the gap-handling pattern already used in the shouldReadFromDisk path.

* fix(filer): clear stale ResumeFromDiskError after gap-skip to avoid stall

The gap-detection block added in the previous commit skips lastReadTime
forward to GetEarliestTime() and continues the outer loop.  On the next
iteration, shouldReadFromDisk becomes true (currentReadTsNs >
lastDiskReadTsNs), the disk read returns processedTsNs == 0, and the
existing gap handler at the top of the loop runs its own gap check.
That check uses readInMemoryLogErr == ResumeFromDiskError as the entry
condition — but readInMemoryLogErr is still the stale error from two
iterations ago.  GetEarliestTime() now equals lastReadTime.Time (we
already advanced to it), so earliestTime.After(lastReadTime.Time) is
false and the handler falls into listenersCond.Wait() — stuck.

Clear readInMemoryLogErr at the gap-skip point, matching the existing
pattern at the earlier gap handler that already clears it for the same
reason.

* fix(log_buffer): GetEarliestTime must include sealed prev buffers

GetEarliestTime previously returned only logBuffer.startTime (the active
buffer's first timestamp).  That is narrower than ReadFromBuffer's
tsMemory, which is the min across active + prev buffers.  Callers using
GetEarliestTime for gap detection after ResumeFromDiskError (the
SubscribeLocalMetadata outer loop's disk-read path, the new gap-skip in
the in-memory ResumeFromDiskError handler, and MQ HasData) saw a time
that was *newer* than the real earliest in-memory data.

Impact in SubscribeLocalMetadata's slow-consumer path:
  - tsMemory = earliest prev buffer time (T_prev)
  - GetEarliestTime() = active startTime (T_active, later than T_prev)
  - Consumer position = T1, with T_prev < T1 < T_active
  - ReadFromBuffer returns ResumeFromDiskError (T1 < tsMemory)
  - Gap detect: GetEarliestTime().After(T1) = T_active.After(T1) = true
  - Skip forward to T_active -- silently drops the prev-buffer data
  - And when T_active happens to equal the stuck position, gap detect
    evaluates false, and the subscriber stalls on listenersCond.Wait()

This reproduces the TestMetadataSubscribeSlowConsumerKeepsProgressing
failure in CI where the consumer stalled at 10220/20000 after writing
stopped -- the buffer still had data in prev[0..3], but gap detection
was comparing against the active buffer's startTime.

Fix: scan all sealed prev buffers under RLock, return the true minimum
startTime.  Matches the min-of-buffers logic in ReadFromBuffer.

* test(log_buffer): make DiskReadRetry test deterministic

The previous test added the message via AddToBuffer + ForceFlush and
relied on a race: the second disk read had to happen before the data
was delivered through the in-memory path.  Under the race detector or
on a slow CI runner, the reader is woken by AddToBuffer's notification,
finds the data in the active buffer or its prev slot, and returns after
exactly one disk read — failing the >= 2 disk reads assertion even
though the loop behaved correctly.

Reproduced on master with race detector (2/5 failures).

Rewrite the test to deliver the data exclusively through the disk-read
path: no AddToBuffer, no ForceFlush.  The test waits until the reader
has issued at least one no-op disk read, then atomically flips a
"dataReady" flag.  The reader's next iteration through readFromDiskFn
returns the entry.  This deterministically exercises the retry-loop
behavior the test was originally written to protect, and removes the
in-memory delivery race entirely.

weed/filer/filer.go

@@ -82,7 +82,13 @@ func NewFiler(masters pb.ServerDiscovery, grpcDialOption grpc.DialOption, filerH
 		f.UniqueFilerId = -f.UniqueFilerId
 	}
 
-	f.LocalMetaLogBuffer = log_buffer.NewLogBuffer("local", LogFlushInterval, f.logFlushFunc, f.readPersistedLogBufferPosition, notifyFn)
+	// ReadFromDiskFn is intentionally nil here.  SubscribeLocalMetadata already
+	// manages disk reads explicitly with shouldReadFromDisk / lastCheckedFlushTsNs
+	// tracking.  Setting ReadFromDiskFn would cause LoopProcessLogData to issue a
+	// redundant ReadPersistedLogBuffer call (ListDirectoryEntries + readahead
+	// goroutine) on every 250ms health-check tick when a subscriber encounters
+	// ResumeFromDiskError, adding significant CPU and GC pressure even when idle.
+	f.LocalMetaLogBuffer = log_buffer.NewLogBuffer("local", LogFlushInterval, f.logFlushFunc, nil, notifyFn)
 	f.metaLogCollection = collection
 	f.metaLogReplication = replication
 

weed/filer/filer_notify.go

@@ -262,16 +262,3 @@ func (f *Filer) ReadPersistedLogBuffer(startPosition log_buffer.MessagePosition,
 	return
 }
 
-func (f *Filer) readPersistedLogBufferPosition(startPosition log_buffer.MessagePosition, stopTsNs int64, eachLogEntryFn log_buffer.EachLogEntryFuncType) (lastReadPosition log_buffer.MessagePosition, isDone bool, err error) {
-	lastReadPosition = startPosition
-
-	lastTsNs, isDone, err := f.ReadPersistedLogBuffer(startPosition, stopTsNs, eachLogEntryFn)
-	if err != nil {
-		return lastReadPosition, isDone, err
-	}
-	if lastTsNs != 0 {
-		lastReadPosition = log_buffer.NewMessagePosition(lastTsNs, 1)
-	}
-
-	return lastReadPosition, isDone, nil
-}

weed/server/filer_grpc_server_sub_meta.go

@@ -415,6 +415,24 @@ func (fs *FilerServer) SubscribeLocalMetadata(req *filer_pb.SubscribeMetadataReq
 						time.Unix(0, lastDiskReadTsNs), time.Unix(0, currentReadTsNs))
 					continue
 				}
+				// No flush or read-position progress — there may be a gap
+				// between the last persisted data and the earliest in-memory
+				// data (e.g. a slow consumer that fell behind while writes
+				// already stopped). Skip forward to the earliest in-memory
+				// time so the consumer can resume instead of blocking forever.
+				earliestTime := fs.filer.LocalMetaLogBuffer.GetEarliestTime()
+				if !earliestTime.IsZero() && earliestTime.After(lastReadTime.Time) {
+					glog.V(3).Infof("gap detected: skipping from %v to earliest memory time %v for %v",
+						lastReadTime.Time, earliestTime, clientName)
+					lastReadTime = log_buffer.NewMessagePosition(earliestTime.UnixNano(), -2)
+					// Clear the stale ResumeFromDiskError so the next
+					// iteration's shouldReadFromDisk path (triggered by the
+					// advanced lastReadTime) doesn't re-enter the gap branch
+					// at line 360 with earliestTime == lastReadTime.Time and
+					// stall on listenersCond.Wait().
+					readInMemoryLogErr = nil
+					continue
+				}
 				// No progress possible, wait for new data to arrive (event-driven, not polling)
 				fs.listenersLock.Lock()
 				atomic.AddInt64(&fs.listenersWaits, 1)

weed/util/log_buffer/log_buffer.go

@@ -615,8 +615,27 @@ func (logBuffer *LogBuffer) invalidateAllDiskCacheChunks() {
 	}
 }
 
+// GetEarliestTime returns the oldest timestamp still resident in the buffer.
+// It must consider the sealed prev buffers in addition to the active buffer,
+// because ReadFromBuffer's tsMemory (and therefore ResumeFromDiskError) is
+// computed from the min across both.  Returning only the active startTime
+// would cause gap-detection callers to skip past data still living in prev
+// buffers, and can also silently equal the consumer's lastReadTime and
+// stall on listenersCond.Wait().
 func (logBuffer *LogBuffer) GetEarliestTime() time.Time {
-	return logBuffer.startTime
+	logBuffer.RLock()
+	defer logBuffer.RUnlock()
+
+	earliest := logBuffer.startTime
+	for _, prevBuf := range logBuffer.prevBuffers.buffers {
+		if prevBuf.startTime.IsZero() {
+			continue
+		}
+		if earliest.IsZero() || prevBuf.startTime.Before(earliest) {
+			earliest = prevBuf.startTime
+		}
+	}
+	return earliest
 }
 
 func (logBuffer *LogBuffer) HasData() bool {

weed/util/log_buffer/log_buffer_test.go

@@ -452,80 +452,60 @@ func TestReadFromBuffer_InitializedFromDisk(t *testing.T) {
 }
 
 // TestLoopProcessLogDataWithOffset_DiskReadRetry tests that when a subscriber
-// reads from disk before flush completes, it continues to retry disk reads
-// and eventually finds the data after flush completes.
-// This reproduces the Schema Registry timeout issue on first start.
+// reads from disk before data is available, it continues to retry disk reads
+// and eventually finds the data once it appears.  This reproduces the Schema
+// Registry timeout where the first read happened before the data landed on
+// disk and the loop never retried.
+//
+// The data is delivered exclusively through the disk-read path (no in-memory
+// AddToBuffer) so the test deterministically asserts that retries happen,
+// rather than racing the in-memory delivery path.
 func TestLoopProcessLogDataWithOffset_DiskReadRetry(t *testing.T) {
 	diskReadCallCount := 0
 	diskReadMu := sync.Mutex{}
-	dataFlushedToDisk := false
-	var flushedData []*filer_pb.LogEntry
+	dataReady := false
+	mockEntry := &filer_pb.LogEntry{
+		Key:    []byte("key-0"),
+		Data:   []byte("message-0"),
+		TsNs:   time.Now().UnixNano(),
+		Offset: 0,
+	}
 
-	// Create a readFromDiskFn that simulates the race condition
 	readFromDiskFn := func(startPosition MessagePosition, stopTsNs int64, eachLogEntryFn EachLogEntryFuncType) (MessagePosition, bool, error) {
 		diskReadMu.Lock()
 		diskReadCallCount++
 		callNum := diskReadCallCount
-		hasData := dataFlushedToDisk
+		ready := dataReady
 		diskReadMu.Unlock()
 
-		t.Logf("DISK READ #%d: startOffset=%d, dataFlushedToDisk=%v", callNum, startPosition.Offset, hasData)
+		t.Logf("DISK READ #%d: startOffset=%d, dataReady=%v", callNum, startPosition.Offset, ready)
 
-		if !hasData {
-			// Simulate: data not yet on disk (flush hasn't completed)
-			t.Logf("  → No data found (flush not completed yet)")
+		if !ready {
+			t.Logf("  → No data on disk yet")
 			return startPosition, false, nil
 		}
 
-		// Data is now on disk, process it
-		t.Logf("  → Found %d entries on disk", len(flushedData))
-		for _, entry := range flushedData {
-			if entry.Offset >= startPosition.Offset {
-				isDone, err := eachLogEntryFn(entry)
-				if err != nil || isDone {
-					return NewMessagePositionFromOffset(entry.Offset + 1), isDone, err
-				}
+		if mockEntry.Offset >= startPosition.Offset {
+			isDone, err := eachLogEntryFn(mockEntry)
+			if err != nil || isDone {
+				return NewMessagePositionFromOffset(mockEntry.Offset + 1), isDone, err
 			}
 		}
-		return NewMessagePositionFromOffset(int64(len(flushedData))), false, nil
+		return NewMessagePositionFromOffset(mockEntry.Offset + 1), false, nil
 	}
 
-	flushFn := func(logBuffer *LogBuffer, startTime, stopTime time.Time, buf []byte, minOffset, maxOffset int64) {
-		t.Logf("FLUSH: minOffset=%d maxOffset=%d size=%d bytes", minOffset, maxOffset, len(buf))
-		// Simulate writing to disk
-		diskReadMu.Lock()
-		dataFlushedToDisk = true
-		// Parse the buffer and add entries to flushedData
-		// For this test, we'll just create mock entries
-		flushedData = append(flushedData, &filer_pb.LogEntry{
-			Key:    []byte("key-0"),
-			Data:   []byte("message-0"),
-			TsNs:   time.Now().UnixNano(),
-			Offset: 0,
-		})
-		diskReadMu.Unlock()
-	}
-
-	logBuffer := NewLogBuffer("test", 1*time.Minute, flushFn, readFromDiskFn, nil)
+	logBuffer := NewLogBuffer("test", 1*time.Minute, nil, readFromDiskFn, nil)
 	defer logBuffer.ShutdownLogBuffer()
 
-	// Simulate the race condition:
-	// 1. Subscriber starts reading from offset 0
-	// 2. Data is not yet flushed
-	// 3. Loop calls readFromDiskFn → no data found
-	// 4. A bit later, data gets flushed
-	// 5. Loop should continue and call readFromDiskFn again
-
 	receivedMessages := 0
 	mu := sync.Mutex{}
-	maxIterations := 50 // Allow up to 50 iterations (500ms with 10ms sleep each)
+	maxIterations := 50
 	iterationCount := 0
 
 	waitForDataFn := func() bool {
 		mu.Lock()
 		defer mu.Unlock()
 		iterationCount++
-		// Stop after receiving message or max iterations
 		return receivedMessages == 0 && iterationCount < maxIterations
 	}
 
@@ -534,10 +514,9 @@ func TestLoopProcessLogDataWithOffset_DiskReadRetry(t *testing.T) {
 		receivedMessages++
 		mu.Unlock()
 		t.Logf("✉️  RECEIVED: offset=%d key=%s", offset, string(logEntry.Key))
-		return true, nil // Stop after first message
+		return true, nil
 	}
 
-	// Start the reader in a goroutine
 	var readerWg sync.WaitGroup
 	readerWg.Add(1)
 	go func() {
@@ -547,27 +526,29 @@ func TestLoopProcessLogDataWithOffset_DiskReadRetry(t *testing.T) {
 		t.Logf("📋 Reader finished: isDone=%v, err=%v", isDone, err)
 	}()
 
-	// Wait a bit to let the first disk read happen (returns no data)
-	time.Sleep(50 * time.Millisecond)
-
-	// Now add data and flush it
-	t.Logf("➕ Adding message to buffer...")
-	if err := logBuffer.AddToBuffer(&mq_pb.DataMessage{
-		Key:   []byte("key-0"),
-		Value: []byte("message-0"),
-		TsNs:  time.Now().UnixNano(),
-	}); err != nil {
-		t.Fatalf("Failed to add buffer: %v", err)
+	// Wait until the reader has issued at least one no-op disk read so we
+	// know it has entered the retry loop, then publish the data.
+	deadline := time.Now().Add(2 * time.Second)
+	for {
+		diskReadMu.Lock()
+		seen := diskReadCallCount
+		diskReadMu.Unlock()
+		if seen >= 1 {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("reader never called readFromDiskFn")
+		}
+		time.Sleep(5 * time.Millisecond)
 	}
 
-	// Force flush
-	t.Logf("Force flushing...")
-	logBuffer.ForceFlush()
+	t.Logf("➕ Marking data as ready on disk")
+	diskReadMu.Lock()
+	dataReady = true
+	diskReadMu.Unlock()
 
-	// Wait for reader to finish
 	readerWg.Wait()
 
-	// Check results
 	diskReadMu.Lock()
 	finalDiskReadCount := diskReadCallCount
 	diskReadMu.Unlock()
@@ -584,12 +565,12 @@ func TestLoopProcessLogDataWithOffset_DiskReadRetry(t *testing.T) {
 
 	if finalDiskReadCount < 2 {
 		t.Errorf("CRITICAL BUG REPRODUCED: Disk read was only called %d time(s)", finalDiskReadCount)
-		t.Errorf("Expected: Multiple disk reads as the loop continues after flush completes")
-		t.Errorf("This is why Schema Registry times out - it reads once before flush, never re-reads after flush")
+		t.Errorf("Expected: Multiple disk reads as the loop continues after data lands")
+		t.Errorf("This is why Schema Registry times out - it reads once before data is available, never re-reads")
 	}
 
 	if finalReceivedMessages == 0 {
-		t.Errorf("SCHEMA REGISTRY TIMEOUT REPRODUCED: No messages received even after flush")
+		t.Errorf("SCHEMA REGISTRY TIMEOUT REPRODUCED: No messages received even after data landed")
 		t.Errorf("The subscriber is stuck because disk reads are not retried")
 	} else {
 		t.Logf("✓ SUCCESS: Message received after %d disk read attempts", finalDiskReadCount)