log_buffer: end bounded reads that find the buffer empty (#10750)

A bounded LoopProcessLogData (stopTsNs set) on a buffer that never took a
write since process start fell into the ResumeFromDiskError branch, which
never checks stopTsNs when ReadFromDiskFn is nil and HasData() is false.
The read parked on the notification loop forever while the subscription's
idle heartbeats kept the stream looking alive, so a bounded
SubscribeMetadata pass on a freshly restarted idle filer never completed.

Terminate like the caught-up path does, returning a nil error: leaking
the pending ResumeFromDiskError would latch the filer's outer loop into
its gap machinery, which parks the bounded subscriber all over again.

weed/server/filer_subscribe_loop_test.go

@@ -524,6 +524,31 @@ func TestSubscribeLoop_BoundedSubscriptionTerminates(t *testing.T) {
 	}
 }
 
+// A bounded subscription against a filer whose meta-log buffer has never taken
+// a write - fresh restart, zero metadata traffic since - must still terminate.
+// The empty ring makes every memory read return ResumeFromDiskError, and with
+// no gap to resolve the loop used to park inside LoopProcessLogData forever,
+// its idle heartbeats keeping the stream looking healthy to the client.
+func TestSubscribeLoop_BoundedEmptyFilerTerminates(t *testing.T) {
+	h := newSubscribeHarness(t)
+
+	r := h.subscribe(h.base, func(req *filer_pb.SubscribeMetadataRequest) {
+		req.UntilNs = time.Now().UnixNano()
+	})
+
+	select {
+	case err := <-r.done:
+		if err != nil {
+			t.Fatalf("bounded subscription failed: %v", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("bounded subscription on an idle filer did not terminate")
+	}
+	if got := eventTimestamps(r.stream.snapshot()); len(got) > 0 {
+		t.Fatalf("delivered %v from an empty filer", got)
+	}
+}
+
 // Vacuumed logs: the flush watermark proves a gap empty even though the files
 // are gone - the resolver must skip to the retained ring and deliver its
 // earliest window intact, including a single-entry window whose start and stop

weed/util/log_buffer/log_read.go

@@ -160,6 +160,16 @@ func (logBuffer *LogBuffer) LoopProcessLogData(readerName string, startPosition
 				continue
 			}
 
+			// Nothing readable anywhere for this cursor: a bounded subscription is
+			// done, same as the caught-up path below. Without this a bounded
+			// subscriber on a buffer that never took a write (empty since process
+			// start, ReadFromDiskFn nil) parks here forever while heartbeats keep
+			// the stream looking alive. The nil error matters: err still holds
+			// ResumeFromDiskError, which callers treat as retry-the-disk-pass.
+			if stopTsNs != 0 {
+				return lastReadPosition, true, nil
+			}
+
 			// CRITICAL: Check if client is still connected
 			if !waitForDataFn() {
 				// Client disconnected - exit cleanly

weed/util/log_buffer/log_read_test.go

@@ -488,6 +488,51 @@ func TestLoopProcessLogDataWithOffset_StopTime(t *testing.T) {
 	t.Logf("Loop correctly exited for past stopTsNs in %v (waitForDataFn called %d times)", elapsed, callCount)
 }
 
+// TestLoopProcessLogData_BoundedEmptyBufferTerminates: a bounded subscription
+// (stopTsNs != 0) against a buffer that has never received a write - so
+// ReadFromBuffer returns ResumeFromDiskError - with no ReadFromDiskFn must
+// terminate like the caught-up path does, not park on the notification loop
+// forever. This is the filer SubscribeMetadata configuration: both meta log
+// buffers are built with ReadFromDiskFn == nil, and a freshly restarted filer
+// that has serviced zero metadata writes wedges every UntilNs-bounded
+// subscriber while heartbeats keep the stream looking alive.
+func TestLoopProcessLogData_BoundedEmptyBufferTerminates(t *testing.T) {
+	logBuffer := NewLogBuffer("test", time.Minute, nil, nil, nil)
+	defer logBuffer.ShutdownLogBuffer()
+
+	// Client stays connected, like the filer's heartbeat-sending closure.
+	waitForDataFn := func() bool { return true }
+	eachLogEntryFn := func(logEntry *filer_pb.LogEntry) (bool, error) {
+		t.Error("no entries should be delivered from an empty buffer")
+		return false, nil
+	}
+
+	startPosition := NewMessagePosition(time.Now().Add(-time.Hour).UnixNano(), EvictionGatedOffset)
+	stopTsNs := time.Now().UnixNano()
+
+	done := make(chan struct{})
+	var isDone bool
+	var err error
+	go func() {
+		_, isDone, err = logBuffer.LoopProcessLogData("bounded-empty", startPosition, stopTsNs, waitForDataFn, eachLogEntryFn)
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		if !isDone {
+			t.Errorf("expected isDone=true for a bounded read of an empty buffer, got false")
+		}
+		// A leaked ResumeFromDiskError would send the filer's outer loop into
+		// its gap machinery, which parks the bounded subscriber all over again.
+		if err != nil {
+			t.Errorf("expected err=nil for a bounded read of an empty buffer, got %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("bounded LoopProcessLogData wedged on an empty buffer instead of terminating")
+	}
+}
+
 func TestLoopProcessLogData_SlowConsumerFallsBehind(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)