[filer] fix log buffer idle polling (#9012)

* fix log buffer idle polling

* log_buffer: document notificationHealthCheckInterval tradeoffs

Explain that notifyChan is the primary wakeup path and this interval only
bounds the fallback / state-recheck cadence, so future maintainers don't
tune it without understanding the implications for client-disconnect
detection latency.

* log_buffer: rename waitForNotification to awaitNotificationOrTimeout

The helper returns after either a notification or the health-check
timeout; the old name read like it blocked indefinitely. No behavior
change.

* log_buffer: wake blocked subscribers on shutdown

awaitNotificationOrTimeout previously only returned on notifyChan or the
health-check timeout, so ShutdownLogBuffer on an idle buffer (where
copyToFlush returns nil and loopFlush never fires the post-flush
notification) would leave subscribers parked for up to 250ms before they
noticed IsStopping.

Add an internal shutdownCh closed by ShutdownLogBuffer and select on it
from awaitNotificationOrTimeout, which is now a method on *LogBuffer.
Subscribers wake immediately, re-check IsStopping, and exit. No change
to LoopProcessLogData signatures or any caller (filer metadata
subscribers, MQ broker, local partition subscribe).

* log_buffer: regression tests for flush-notify wake-up

TestLoopFlush_NotifiesSubscribersAfterFlush directly verifies that
loopFlush calls notifySubscribers after processing a flush, so a reader
parked on notifyChan wakes promptly when a flush lands. Verified to fail
if that notification is removed.

TestLoopProcessLogDataWithOffset_WakesOnDataArrival is the end-to-end
counterpart: a real LoopProcessLogDataWithOffset reader parks on
notifyChan via the ResumeFromDiskError branch, then wakes and processes
the entry well under the 250ms fallback once data arrives.

* log_buffer: keep notification-timeout logs at V(4)

Revert the V(4)->V(5) demotion. Now that the shutdown wake-up path
exists and (with the follow-up fix) idle-polling CPU churn is bounded
by the 250ms health check, these timeout logs no longer flood at V=4
the way they did on the 10ms fallback, so the previous verbosity is
appropriate again.

* log_buffer: exit reader loops cleanly on shutdown

awaitNotificationOrTimeout returns true on both data notifications and
shutdown (shutdownCh closed). Without an explicit IsStopping() guard,
the ResumeFromDiskError, offset-based no-data, empty-buffer, and
timestamp-wait paths would either tight-spin against a closed shutdownCh
or, in the offset-based case, return ResumeFromDiskError to the caller
instead of exiting.

Add an IsStopping() check after each awaitNotificationOrTimeout call
that previously continued or returned ResumeFromDiskError, so subscribers
exit promptly with isDone=true and err=nil when ShutdownLogBuffer is
called.

* log_buffer: regression test for shutdown wake-up

Park a real LoopProcessLogDataWithOffset reader on notifyChan via the
ResumeFromDiskError branch, call ShutdownLogBuffer, and assert the
reader exits with isDone=true and err=nil well under the 250ms
fallback. Verified to fail (timeout) if the IsStopping() guards added
in the prior commit are removed.

* log_buffer: bump reader-park sleep to 50ms with rationale

Both wake-path tests use a sleep to give the goroutine time to reach
awaitNotificationOrTimeout before the test triggers the wake-up.
Bump from 20ms to 50ms and document the timing assumption to reduce
flakiness on slow CI. Both paths are race-free either way (a buffered
notification or a closed shutdownCh stays valid until consumed), so
this is purely about exercising the park-then-wake path rather than
the already-pending fast path.

weed/util/log_buffer/log_buffer.go

@@ -81,6 +81,7 @@ type LogBuffer struct {
 	subscribersMu sync.RWMutex
 	subscribers   map[string]chan struct{} // subscriberID -> notification channel
 	isStopping    *atomic.Bool
+	shutdownCh    chan struct{} // closed by ShutdownLogBuffer to wake blocked subscribers
 	isAllFlushed  bool
 	flushChan     chan *dataToFlush
 	// Offset range tracking for Kafka integration
@@ -104,6 +105,7 @@ func NewLogBuffer(name string, flushInterval time.Duration, flushFn LogFlushFunc
 		subscribers:    make(map[string]chan struct{}),
 		flushChan:      make(chan *dataToFlush, 256),
 		isStopping:     new(atomic.Bool),
+		shutdownCh:     make(chan struct{}),
 		offset:         0, // Will be initialized from existing data if available
 		diskChunkCache: &DiskChunkCache{
 			chunks:    make(map[int64]*CachedDiskChunk),
@@ -492,6 +494,10 @@ func (logBuffer *LogBuffer) ShutdownLogBuffer() {
 	if isAlreadyStopped {
 		return
 	}
+	// Wake any subscribers blocked in awaitNotificationOrTimeout so they can
+	// notice IsStopping() and exit promptly, even on an idle buffer where no
+	// flush notification would otherwise fire.
+	close(logBuffer.shutdownCh)
 	toFlush := logBuffer.copyToFlush()
 	logBuffer.flushChan <- toFlush
 	close(logBuffer.flushChan)
@@ -515,6 +521,12 @@ func (logBuffer *LogBuffer) loopFlush() {
 				logBuffer.lastFlushTsNs.Store(d.stopTime.UnixNano())
 			}
 
+			// Wake readers that may be waiting to retry disk reads after the flush lands.
+			if logBuffer.notifyFn != nil {
+				logBuffer.notifyFn()
+			}
+			logBuffer.notifySubscribers()
+
 			// Signal completion if there's a callback channel
 			if d.done != nil {
 				close(d.done)

weed/util/log_buffer/log_read.go

@@ -18,6 +18,15 @@ var (
 	ResumeFromDiskError = fmt.Errorf("resumeFromDisk")
 )
 
+// notificationHealthCheckInterval bounds how long an idle subscriber blocks
+// on the notification channel before re-checking state (client disconnect via
+// waitForDataFn, LogBuffer shutdown, timestamp advancement). notifyChan is the
+// primary wakeup path when new data arrives or a flush lands; this timeout is
+// the safety net for any missed notification and also caps the latency to
+// notice that the subscriber should exit. Balances idle CPU and log noise
+// against client-disconnect detection latency.
+const notificationHealthCheckInterval = 250 * time.Millisecond
+
 type MessagePosition struct {
 	Time          time.Time // timestamp of the message
 	Offset        int64     // Kafka offset for offset-based positioning, or batch index for timestamp-based
@@ -49,6 +58,26 @@ func (mp MessagePosition) GetOffset() int64 {
 	return mp.Offset // Offset is stored directly
 }
 
+// awaitNotificationOrTimeout blocks until one of:
+//   - a new-data / flush notification arrives on notifyChan (returns true)
+//   - the LogBuffer is shut down via ShutdownLogBuffer (returns true; callers
+//     re-check IsStopping() and exit)
+//   - notificationHealthCheckInterval elapses (returns false; caller
+//     re-checks client-disconnect and other state)
+func (logBuffer *LogBuffer) awaitNotificationOrTimeout(notifyChan <-chan struct{}) bool {
+	timer := time.NewTimer(notificationHealthCheckInterval)
+	defer timer.Stop()
+
+	select {
+	case <-notifyChan:
+		return true
+	case <-logBuffer.shutdownCh:
+		return true
+	case <-timer.C:
+		return false
+	}
+}
+
 func (logBuffer *LogBuffer) LoopProcessLogData(readerName string, startPosition MessagePosition, stopTsNs int64,
 	waitForDataFn func() bool, eachLogDataFn EachLogEntryFuncType) (lastReadPosition MessagePosition, isDone bool, err error) {
 
@@ -100,13 +129,19 @@ func (logBuffer *LogBuffer) LoopProcessLogData(readerName string, startPosition
 			}
 
 			// Wait for notification or timeout (instant wake-up when data arrives)
-			select {
-			case <-notifyChan:
-				// New data available, retry immediately
+			if logBuffer.awaitNotificationOrTimeout(notifyChan) {
 				glog.V(3).Infof("%s: Woke up from notification after ResumeFromDiskError", readerName)
-			case <-time.After(10 * time.Millisecond):
-				// Timeout, retry anyway (fallback for edge cases)
-				glog.V(4).Infof("%s: Notification timeout after ResumeFromDiskError, polling", readerName)
+			} else {
+				glog.V(4).Infof("%s: Notification timeout after ResumeFromDiskError, rechecking state", readerName)
+			}
+
+			// If the LogBuffer is shutting down, exit cleanly instead of looping
+			// on ResumeFromDiskError. awaitNotificationOrTimeout returns true on
+			// shutdown (shutdownCh closed), which would otherwise spin here
+			// because ReadFromBuffer keeps returning ResumeFromDiskError.
+			if logBuffer.IsStopping() {
+				isDone = true
+				return
 			}
 
 			// Continue to next iteration (don't return ResumeFromDiskError)
@@ -143,17 +178,18 @@ func (logBuffer *LogBuffer) LoopProcessLogData(readerName string, startPosition
 					return
 				}
 				// Wait for notification or timeout (instant wake-up when data arrives)
-				select {
-				case <-notifyChan:
-					// New data available, break and retry read
+				if logBuffer.awaitNotificationOrTimeout(notifyChan) {
 					glog.V(3).Infof("%s: Woke up from notification (LoopProcessLogData)", readerName)
+				} else if lastTsNs != logBuffer.LastTsNs.Load() {
 					break
-				case <-time.After(10 * time.Millisecond):
-					// Timeout, check if timestamp changed
-					if lastTsNs != logBuffer.LastTsNs.Load() {
-						break
-					}
-					glog.V(4).Infof("%s: Notification timeout (LoopProcessLogData), polling", readerName)
+				} else {
+					glog.V(4).Infof("%s: Notification timeout (LoopProcessLogData), rechecking state", readerName)
+				}
+				// Exit the wait loop on shutdown so we don't spin against a
+				// closed shutdownCh.
+				if logBuffer.IsStopping() {
+					isDone = true
+					return
 				}
 			}
 			if logBuffer.IsStopping() {
@@ -294,13 +330,15 @@ func (logBuffer *LogBuffer) LoopProcessLogDataWithOffset(readerName string, star
 			}
 
 			// Wait for notification or timeout (instant wake-up when data arrives)
-			select {
-			case <-notifyChan:
-				// New data available, retry immediately
+			if logBuffer.awaitNotificationOrTimeout(notifyChan) {
 				glog.V(3).Infof("%s: Woke up from notification after disk read", readerName)
-			case <-time.After(10 * time.Millisecond):
-				// Timeout, retry anyway (fallback for edge cases)
-				glog.V(4).Infof("%s: Notification timeout, polling", readerName)
+			} else {
+				glog.V(4).Infof("%s: Notification timeout, rechecking state", readerName)
+			}
+
+			// Exit cleanly on shutdown so we don't loop on ResumeFromDiskError.
+			if logBuffer.IsStopping() {
+				return lastReadPosition, true, nil
 			}
 
 			// Continue to next iteration (don't return ResumeFromDiskError)
@@ -338,13 +376,14 @@ func (logBuffer *LogBuffer) LoopProcessLogDataWithOffset(readerName string, star
 					return lastReadPosition, true, nil
 				}
 				// Wait for notification or timeout (instant wake-up when data arrives)
-				select {
-				case <-notifyChan:
-					// New data available, retry immediately
+				if logBuffer.awaitNotificationOrTimeout(notifyChan) {
 					glog.V(3).Infof("%s: Woke up from notification for offset-based read", readerName)
-				case <-time.After(10 * time.Millisecond):
-					// Timeout, retry anyway (fallback for edge cases)
-					glog.V(4).Infof("%s: Notification timeout for offset-based, polling", readerName)
+				} else {
+					glog.V(4).Infof("%s: Notification timeout for offset-based, rechecking state", readerName)
+				}
+				// On shutdown, exit cleanly instead of returning ResumeFromDiskError.
+				if logBuffer.IsStopping() {
+					return lastReadPosition, true, nil
 				}
 				return lastReadPosition, isDone, ResumeFromDiskError
 			}
@@ -357,17 +396,18 @@ func (logBuffer *LogBuffer) LoopProcessLogDataWithOffset(readerName string, star
 					return lastReadPosition, true, nil
 				}
 				// Wait for notification or timeout (instant wake-up when data arrives)
-				select {
-				case <-notifyChan:
-					// New data available, break and retry read
+				if logBuffer.awaitNotificationOrTimeout(notifyChan) {
 					glog.V(3).Infof("%s: Woke up from notification (main loop)", readerName)
+				} else if lastTsNs != logBuffer.LastTsNs.Load() {
 					break
-				case <-time.After(10 * time.Millisecond):
-					// Timeout, check if timestamp changed
-					if lastTsNs != logBuffer.LastTsNs.Load() {
-						break
-					}
-					glog.V(4).Infof("%s: Notification timeout (main loop), polling", readerName)
+				} else {
+					glog.V(4).Infof("%s: Notification timeout (main loop), rechecking state", readerName)
+				}
+				// Exit the wait loop on shutdown so we don't spin against a
+				// closed shutdownCh.
+				if logBuffer.IsStopping() {
+					glog.V(4).Infof("%s: LogBuffer is stopping", readerName)
+					return lastReadPosition, true, nil
 				}
 			}
 			if logBuffer.IsStopping() {
@@ -388,8 +428,15 @@ func (logBuffer *LogBuffer) LoopProcessLogDataWithOffset(readerName string, star
 				glog.V(4).Infof("%s: Client disconnected on empty buffer", readerName)
 				return lastReadPosition, true, nil
 			}
-			// Sleep to avoid busy-wait on empty buffer
-			time.Sleep(10 * time.Millisecond)
+			if logBuffer.awaitNotificationOrTimeout(notifyChan) {
+				glog.V(3).Infof("%s: Woke up from notification on empty buffer", readerName)
+			} else {
+				glog.V(4).Infof("%s: Empty buffer timeout, rechecking state", readerName)
+			}
+			// Exit cleanly on shutdown to avoid an idle spin on the empty buffer.
+			if logBuffer.IsStopping() {
+				return lastReadPosition, true, nil
+			}
 			continue
 		}
 

weed/util/log_buffer/log_read_test.go

@@ -61,7 +61,7 @@ func TestLoopProcessLogDataWithOffset_EmptyBuffer(t *testing.T) {
 	defer logBuffer.ShutdownLogBuffer()
 
 	callCount := 0
-	maxCalls := 10
+	maxCalls := 4
 	mu := sync.Mutex{}
 
 	waitForDataFn := func() bool {
@@ -87,14 +87,12 @@ func TestLoopProcessLogDataWithOffset_EmptyBuffer(t *testing.T) {
 		t.Errorf("Expected isDone=true when waitForDataFn returns false, got false")
 	}
 
-	// With 10ms sleep per iteration, 10 iterations should take ~100ms minimum
-	minExpectedTime := time.Duration(maxCalls-1) * 10 * time.Millisecond
+	minExpectedTime := time.Duration(maxCalls-1) * notificationHealthCheckInterval
 	if elapsed < minExpectedTime {
 		t.Errorf("Loop exited too quickly (%v), expected at least %v (suggests busy-waiting)", elapsed, minExpectedTime)
 	}
 
-	// But shouldn't take more than 2x expected (allows for some overhead)
-	maxExpectedTime := time.Duration(maxCalls) * 30 * time.Millisecond
+	maxExpectedTime := time.Duration(maxCalls+1) * notificationHealthCheckInterval
 	if elapsed > maxExpectedTime {
 		t.Errorf("Loop took too long: %v (expected < %v)", elapsed, maxExpectedTime)
 	}
@@ -122,7 +120,7 @@ func TestLoopProcessLogDataWithOffset_NoDataResumeFromDisk(t *testing.T) {
 	defer logBuffer.ShutdownLogBuffer()
 
 	callCount := 0
-	maxCalls := 5
+	maxCalls := 3
 	mu := sync.Mutex{}
 
 	waitForDataFn := func() bool {
@@ -148,15 +146,198 @@ func TestLoopProcessLogDataWithOffset_NoDataResumeFromDisk(t *testing.T) {
 		t.Errorf("Expected isDone=true when waitForDataFn returns false, got false")
 	}
 
-	// Should take at least (maxCalls-1) * 10ms due to sleep in ResumeFromDiskError path
-	minExpectedTime := time.Duration(maxCalls-1) * 10 * time.Millisecond
+	minExpectedTime := time.Duration(maxCalls-1) * notificationHealthCheckInterval
 	if elapsed < minExpectedTime {
-		t.Errorf("Loop exited too quickly (%v), expected at least %v (suggests missing sleep)", elapsed, minExpectedTime)
+		t.Errorf("Loop exited too quickly (%v), expected at least %v (suggests missing wait)", elapsed, minExpectedTime)
 	}
 
 	t.Logf("Loop exited cleanly in %v after %d iterations (proper sleep detected)", elapsed, callCount)
 }
 
+// TestLoopFlush_NotifiesSubscribersAfterFlush is a regression test for the
+// issue #9007 fix: loopFlush must call notifySubscribers() after processing a
+// flush so that readers parked on notifyChan wake up when a flush lands. The
+// classic bug scenario is a reader that got ResumeFromDiskError, did a disk
+// read that raced the flush and found nothing, and is now blocked on
+// notifyChan waiting for the data that just hit disk.
+//
+// We drain the AddToBuffer notification first, then ForceFlush, and assert a
+// new notification is delivered on notifyChan well before the fallback
+// timeout. If the loopFlush notification is removed, this test fails by
+// hitting the fallback.
+func TestLoopFlush_NotifiesSubscribersAfterFlush(t *testing.T) {
+	flushFn := func(logBuffer *LogBuffer, startTime, stopTime time.Time, buf []byte, minOffset, maxOffset int64) {
+	}
+	logBuffer := NewLogBuffer("test", 1*time.Minute, flushFn, nil, nil)
+	defer logBuffer.ShutdownLogBuffer()
+
+	notifyChan := logBuffer.RegisterSubscriber("flush-notify-test")
+	defer logBuffer.UnregisterSubscriber("flush-notify-test")
+
+	if err := logBuffer.AddToBuffer(&mq_pb.DataMessage{
+		Key:   []byte("k"),
+		Value: []byte("v"),
+		TsNs:  time.Now().UnixNano(),
+	}); err != nil {
+		t.Fatalf("AddToBuffer: %v", err)
+	}
+
+	// Consume the AddToBuffer notification so the channel starts empty.
+	select {
+	case <-notifyChan:
+	case <-time.After(100 * time.Millisecond):
+		t.Fatal("expected a notification from AddToBuffer")
+	}
+
+	// ForceFlush waits for loopFlush to process the flush. After it returns,
+	// loopFlush must have called notifySubscribers() again.
+	start := time.Now()
+	logBuffer.ForceFlush()
+
+	select {
+	case <-notifyChan:
+		elapsed := time.Since(start)
+		// The fallback timeout is notificationHealthCheckInterval; the flush
+		// notification should arrive well before that.
+		if elapsed >= notificationHealthCheckInterval {
+			t.Errorf("flush notification too slow: %v (>= fallback %v)", elapsed, notificationHealthCheckInterval)
+		}
+		t.Logf("flush notification delivered in %v", elapsed)
+	case <-time.After(notificationHealthCheckInterval):
+		t.Fatalf("loopFlush did not notify subscribers within %v", notificationHealthCheckInterval)
+	}
+}
+
+// TestLoopProcessLogDataWithOffset_WakesOnDataArrival drives a real
+// LoopProcessLogDataWithOffset reader from an empty buffer (readFromDiskFn
+// returns nothing, forcing the reader to park on notifyChan after the
+// ResumeFromDiskError branch), then adds data from another goroutine and
+// asserts the reader completes well before the fallback timeout would fire.
+// This protects the end-to-end wake-up path; the loopFlush-specific
+// notification is covered by TestLoopFlush_NotifiesSubscribersAfterFlush.
+func TestLoopProcessLogDataWithOffset_WakesOnDataArrival(t *testing.T) {
+	readFromDiskFn := func(startPosition MessagePosition, stopTsNs int64, eachLogEntryFn EachLogEntryFuncType) (MessagePosition, bool, error) {
+		// No data on disk; return unchanged so the reader parks on notifyChan.
+		return startPosition, false, nil
+	}
+	flushFn := func(logBuffer *LogBuffer, startTime, stopTime time.Time, buf []byte, minOffset, maxOffset int64) {
+	}
+	logBuffer := NewLogBuffer("test", 1*time.Minute, flushFn, readFromDiskFn, nil)
+	defer logBuffer.ShutdownLogBuffer()
+
+	received := make(chan struct{})
+	eachLogEntryFn := func(logEntry *filer_pb.LogEntry, offset int64) (bool, error) {
+		close(received)
+		return true, nil // isDone
+	}
+
+	waitForDataFn := func() bool { return true }
+
+	startPosition := NewMessagePositionFromOffset(0)
+
+	readerDone := make(chan struct{})
+	go func() {
+		_, _, _ = logBuffer.LoopProcessLogDataWithOffset(
+			"wake-test", startPosition, 0, waitForDataFn, eachLogEntryFn)
+		close(readerDone)
+	}()
+
+	// Give the reader time to reach awaitNotificationOrTimeout. Both wake
+	// paths under test (notifyChan via AddToBuffer and shutdownCh via
+	// ShutdownLogBuffer) are race-free even if the reader hasn't parked yet
+	// — the notification stays buffered / shutdownCh stays closed — but a
+	// generous head start makes it likelier we exercise the actual park-then-
+	// wake path rather than the already-pending fast path. 50ms is well below
+	// notificationHealthCheckInterval (250ms) and tolerates slow CI.
+	time.Sleep(50 * time.Millisecond)
+
+	start := time.Now()
+	if err := logBuffer.AddToBuffer(&mq_pb.DataMessage{
+		Key:   []byte("k"),
+		Value: []byte("v"),
+		TsNs:  time.Now().UnixNano(),
+	}); err != nil {
+		t.Fatalf("AddToBuffer: %v", err)
+	}
+
+	select {
+	case <-received:
+	case <-time.After(notificationHealthCheckInterval):
+		t.Fatalf("reader did not process the entry within %v (fallback timeout)", notificationHealthCheckInterval)
+	}
+	<-readerDone
+	elapsed := time.Since(start)
+
+	if elapsed >= notificationHealthCheckInterval {
+		t.Errorf("reader wake too slow: %v (>= fallback %v)", elapsed, notificationHealthCheckInterval)
+	}
+	t.Logf("reader processed the entry in %v after AddToBuffer", elapsed)
+}
+
+// TestLoopProcessLogDataWithOffset_WakesOnShutdown verifies that a reader
+// parked inside awaitNotificationOrTimeout via the ResumeFromDiskError branch
+// exits promptly when ShutdownLogBuffer is called, without waiting for the
+// 250ms health-check fallback. Regression guard for the IsStopping() shutdown
+// path: if awaitNotificationOrTimeout returns true via shutdownCh and the
+// caller does not check IsStopping(), the reader either spins against the
+// closed shutdownCh or returns ResumeFromDiskError instead of exiting.
+func TestLoopProcessLogDataWithOffset_WakesOnShutdown(t *testing.T) {
+	readFromDiskFn := func(startPosition MessagePosition, stopTsNs int64, eachLogEntryFn EachLogEntryFuncType) (MessagePosition, bool, error) {
+		// No data on disk; return unchanged so the reader parks on notifyChan.
+		return startPosition, false, nil
+	}
+	flushFn := func(logBuffer *LogBuffer, startTime, stopTime time.Time, buf []byte, minOffset, maxOffset int64) {
+	}
+	logBuffer := NewLogBuffer("test", 1*time.Minute, flushFn, readFromDiskFn, nil)
+	// Note: not deferring ShutdownLogBuffer; we trigger it explicitly below.
+
+	eachLogEntryFn := func(logEntry *filer_pb.LogEntry, offset int64) (bool, error) {
+		return false, nil
+	}
+	waitForDataFn := func() bool { return true }
+	startPosition := NewMessagePositionFromOffset(0)
+
+	type result struct {
+		isDone bool
+		err    error
+	}
+	resultCh := make(chan result, 1)
+	go func() {
+		_, isDone, err := logBuffer.LoopProcessLogDataWithOffset(
+			"shutdown-test", startPosition, 0, waitForDataFn, eachLogEntryFn)
+		resultCh <- result{isDone: isDone, err: err}
+	}()
+
+	// Give the reader time to reach awaitNotificationOrTimeout. Both wake
+	// paths under test (notifyChan via AddToBuffer and shutdownCh via
+	// ShutdownLogBuffer) are race-free even if the reader hasn't parked yet
+	// — the notification stays buffered / shutdownCh stays closed — but a
+	// generous head start makes it likelier we exercise the actual park-then-
+	// wake path rather than the already-pending fast path. 50ms is well below
+	// notificationHealthCheckInterval (250ms) and tolerates slow CI.
+	time.Sleep(50 * time.Millisecond)
+
+	start := time.Now()
+	logBuffer.ShutdownLogBuffer()
+
+	select {
+	case r := <-resultCh:
+		elapsed := time.Since(start)
+		if elapsed >= notificationHealthCheckInterval {
+			t.Errorf("reader did not wake on shutdown: %v (>= fallback %v)", elapsed, notificationHealthCheckInterval)
+		}
+		if !r.isDone {
+			t.Errorf("expected isDone=true on shutdown, got false")
+		}
+		if r.err != nil {
+			t.Errorf("expected err=nil on shutdown, got %v", r.err)
+		}
+		t.Logf("reader exited in %v after ShutdownLogBuffer", elapsed)
+	case <-time.After(2 * notificationHealthCheckInterval):
+		t.Fatalf("reader did not exit within %v after ShutdownLogBuffer", 2*notificationHealthCheckInterval)
+	}
+}
+
 // TestLoopProcessLogDataWithOffset_WithData tests normal operation with data
 func TestLoopProcessLogDataWithOffset_WithData(t *testing.T) {
 	flushFn := func(logBuffer *LogBuffer, startTime, stopTime time.Time, buf []byte, minOffset, maxOffset int64) {}