fix reading

Gap detection and skipping to earliest memory time
Time-based reads that include events at boundary times for first reads (offset ≤ 0)
Aggregated subscriber wake-up via ListenersWaits signaling

4 files changed, 123 insertions, 36 deletions

diff --git a/weed/util/log_buffer/log_buffer.go b/weed/util/log_buffer/log_buffer.go
index 5aeb285ca..5f4dc8099 100644
--- a/weed/util/log_buffer/log_buffer.go
+++ b/weed/util/log_buffer/log_buffer.go
@@ -235,7 +235,7 @@ func (logBuffer *LogBuffer) InitializeOffsetFromExistingData(getHighestOffsetFn
 		// Set the next offset to be one after the highest existing offset
 		nextOffset := highestOffset + 1
 		logBuffer.offset = nextOffset
-		// CRITICAL FIX: bufferStartOffset should match offset after initialization
+		// bufferStartOffset should match offset after initialization
 		// This ensures that reads for old offsets (0...highestOffset) will trigger disk reads
 		// New data written after this will start at nextOffset
 		logBuffer.bufferStartOffset = nextOffset
@@ -299,7 +299,7 @@ func (logBuffer *LogBuffer) AddLogEntryToBuffer(logEntry *filer_pb.LogEntry) {
 	}
 
 	// Track offset ranges for Kafka integration
-	// CRITICAL FIX: Use >= 0 to include offset 0 (first message in a topic)
+	// Use >= 0 to include offset 0 (first message in a topic)
 	if logEntry.Offset >= 0 {
 		if !logBuffer.hasOffsets {
 			logBuffer.minOffset = logEntry.Offset
@@ -385,7 +385,7 @@ func (logBuffer *LogBuffer) AddDataToBuffer(partitionKey, data []byte, processin
 		logBuffer.LastTsNs.Store(processingTsNs)
 	}
 
-	// CRITICAL FIX: Set the offset in the LogEntry before marshaling
+	// Set the offset in the LogEntry before marshaling
 	// This ensures the flushed data contains the correct offset information
 	// Note: This also enables AddToBuffer to work correctly with Kafka-style offset-based reads
 	logEntry.Offset = logBuffer.offset
@@ -414,7 +414,7 @@ func (logBuffer *LogBuffer) AddDataToBuffer(partitionKey, data []byte, processin
 	}
 
 	// Track offset ranges for Kafka integration
-	// CRITICAL FIX: Track the current offset being written
+	// Track the current offset being written
 	if !logBuffer.hasOffsets {
 		logBuffer.minOffset = logBuffer.offset
 		logBuffer.maxOffset = logBuffer.offset
@@ -518,7 +518,7 @@ func (logBuffer *LogBuffer) loopFlush() {
 			logBuffer.lastFlushDataTime = d.stopTime
 
 			// CRITICAL: Track what's been flushed to disk for both offset-based and time-based reads
-			// CRITICAL FIX: Use >= 0 to include offset 0 (first message in a topic)
+			// Use >= 0 to include offset 0 (first message in a topic)
 			if d.maxOffset >= 0 {
 				logBuffer.lastFlushedOffset.Store(d.maxOffset)
 			}
@@ -586,8 +586,10 @@ func (logBuffer *LogBuffer) copyToFlushInternal(withCallback bool) *dataToFlush
 		// CRITICAL: logBuffer.offset is the "next offset to assign", so last offset in buffer is offset-1
 		lastOffsetInBuffer := logBuffer.offset - 1
 		logBuffer.buf = logBuffer.prevBuffers.SealBuffer(logBuffer.startTime, logBuffer.stopTime, logBuffer.buf, logBuffer.pos, logBuffer.bufferStartOffset, lastOffsetInBuffer)
-		logBuffer.startTime = time.Unix(0, 0)
-		logBuffer.stopTime = time.Unix(0, 0)
+		// Use zero time (time.Time{}) not epoch time (time.Unix(0,0))
+		// Epoch time (1970) breaks time-based reads after flush
+		logBuffer.startTime = time.Time{}
+		logBuffer.stopTime = time.Time{}
 		logBuffer.pos = 0
 		logBuffer.idx = logBuffer.idx[:0]
 		// DON'T increment offset - it's already pointing to the next offset!
@@ -598,7 +600,7 @@ func (logBuffer *LogBuffer) copyToFlushInternal(withCallback bool) *dataToFlush
 		logBuffer.minOffset = 0
 		logBuffer.maxOffset = 0
 
-		// CRITICAL FIX: Invalidate disk cache chunks after flush
+		// Invalidate disk cache chunks after flush
 		// The cache may contain stale data from before this flush
 		// Invalidating ensures consumers will re-read fresh data from disk after flush
 		logBuffer.invalidateAllDiskCacheChunks()
@@ -639,8 +641,10 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 	defer logBuffer.RUnlock()
 
 	isOffsetBased := lastReadPosition.IsOffsetBased
+	glog.V(2).Infof("[ReadFromBuffer] %s: isOffsetBased=%v, position=%+v, bufferStartOffset=%d, offset=%d, pos=%d",
+		logBuffer.name, isOffsetBased, lastReadPosition, logBuffer.bufferStartOffset, logBuffer.offset, logBuffer.pos)
 
-	// CRITICAL FIX: For offset-based subscriptions, use offset comparisons, not time comparisons!
+	// For offset-based subscriptions, use offset comparisons, not time comparisons!
 	if isOffsetBased {
 		requestedOffset := lastReadPosition.Offset
 
@@ -648,7 +652,7 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 		if requestedOffset >= logBuffer.bufferStartOffset && requestedOffset <= logBuffer.offset {
 			// If current buffer is empty (pos=0), check if data is on disk or not yet written
 			if logBuffer.pos == 0 {
-				// CRITICAL FIX: If buffer is empty but offset range covers the request,
+				// If buffer is empty but offset range covers the request,
 				// it means data was in memory and has been flushed/moved out.
 				// The bufferStartOffset advancing to cover this offset proves data existed.
 				//
@@ -719,15 +723,22 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 	if !logBuffer.startTime.IsZero() {
 		tsMemory = logBuffer.startTime
 	}
-	for _, prevBuf := range logBuffer.prevBuffers.buffers {
-		if !prevBuf.startTime.IsZero() && prevBuf.startTime.Before(tsMemory) {
-			tsMemory = prevBuf.startTime
+	glog.V(2).Infof("[ReadFromBuffer] %s: checking prevBuffers, count=%d, currentStartTime=%v",
+		logBuffer.name, len(logBuffer.prevBuffers.buffers), logBuffer.startTime)
+	for i, prevBuf := range logBuffer.prevBuffers.buffers {
+		glog.V(2).Infof("[ReadFromBuffer] %s: prevBuf[%d]: startTime=%v stopTime=%v size=%d startOffset=%d endOffset=%d",
+			logBuffer.name, i, prevBuf.startTime, prevBuf.stopTime, prevBuf.size, prevBuf.startOffset, prevBuf.offset)
+		if !prevBuf.startTime.IsZero() {
+			// If tsMemory is zero, assign directly; otherwise compare
+			if tsMemory.IsZero() || prevBuf.startTime.Before(tsMemory) {
+				tsMemory = prevBuf.startTime
+			}
 		}
 	}
 	if tsMemory.IsZero() { // case 2.2
 		return nil, -2, nil
 	} else if lastReadPosition.Time.Before(tsMemory) { // case 2.3
-		// CRITICAL FIX: For time-based reads, only check timestamp for disk reads
+		// For time-based reads, only check timestamp for disk reads
 		// Don't use offset comparisons as they're not meaningful for time-based subscriptions
 
 		// Special case: If requested time is zero (Unix epoch), treat as "start from beginning"
@@ -735,45 +746,67 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 		if lastReadPosition.Time.IsZero() || lastReadPosition.Time.Unix() == 0 {
 			// Start from the beginning of memory
 			// Fall through to case 2.1 to read from earliest buffer
-		} else if lastReadPosition.Offset == 0 && lastReadPosition.Time.Before(tsMemory) {
-			// CRITICAL FIX: If this is the first read (offset=0) and time is slightly before memory,
-			// it's likely a race between starting to read and first message being written
-			// Fall through to case 2.1 to read from earliest buffer instead of triggering disk read
-			glog.V(2).Infof("first read at time %v before earliest memory %v, reading from memory",
-				lastReadPosition.Time, tsMemory)
+		} else if lastReadPosition.Offset <= 0 && lastReadPosition.Time.Before(tsMemory) {
+			// Treat first read with sentinel/zero offset as inclusive of earliest in-memory data
+			glog.V(4).Infof("first read (offset=%d) at time %v before earliest memory %v, reading from memory",
+				lastReadPosition.Offset, lastReadPosition.Time, tsMemory)
 		} else {
 			// Data not in memory buffers - read from disk
-			glog.V(0).Infof("resume from disk: requested time %v < earliest memory time %v",
-				lastReadPosition.Time, tsMemory)
+			glog.V(0).Infof("[ReadFromBuffer] %s resume from disk: requested time %v < earliest memory time %v",
+				logBuffer.name, lastReadPosition.Time, tsMemory)
 			return nil, -2, ResumeFromDiskError
 		}
 	}
 
+	glog.V(2).Infof("[ReadFromBuffer] %s: time-based read continuing, tsMemory=%v, lastReadPos=%v",
+		logBuffer.name, tsMemory, lastReadPosition.Time)
+
 	// the following is case 2.1
 
-	if lastReadPosition.Time.Equal(logBuffer.stopTime) {
-		return nil, logBuffer.offset, nil
+	if lastReadPosition.Time.Equal(logBuffer.stopTime) && !logBuffer.stopTime.IsZero() {
+		// For first-read sentinel/zero offset, allow inclusive read at the boundary
+		if lastReadPosition.Offset > 0 {
+			return nil, logBuffer.offset, nil
+		}
 	}
-	if lastReadPosition.Time.After(logBuffer.stopTime) {
+	if lastReadPosition.Time.After(logBuffer.stopTime) && !logBuffer.stopTime.IsZero() {
 		// glog.Fatalf("unexpected last read time %v, older than latest %v", lastReadPosition, m.stopTime)
 		return nil, logBuffer.offset, nil
 	}
-	if lastReadPosition.Time.Before(logBuffer.startTime) {
+	// Also check prevBuffers when current buffer is empty (startTime is zero)
+	if lastReadPosition.Time.Before(logBuffer.startTime) || logBuffer.startTime.IsZero() {
 		for _, buf := range logBuffer.prevBuffers.buffers {
 			if buf.startTime.After(lastReadPosition.Time) {
 				// glog.V(4).Infof("%s return the %d sealed buffer %v", m.name, i, buf.startTime)
 				return copiedBytes(buf.buf[:buf.size]), buf.offset, nil
 			}
 			if !buf.startTime.After(lastReadPosition.Time) && buf.stopTime.After(lastReadPosition.Time) {
-				pos := buf.locateByTs(lastReadPosition.Time)
+				searchTime := lastReadPosition.Time
+				if lastReadPosition.Offset <= 0 {
+					searchTime = searchTime.Add(-time.Nanosecond)
+				}
+				pos := buf.locateByTs(searchTime)
+				glog.V(2).Infof("[ReadFromBuffer] %s: found data in prevBuffer at pos %d, bufSize=%d", logBuffer.name, pos, buf.size)
 				return copiedBytes(buf.buf[pos:buf.size]), buf.offset, nil
 			}
 		}
-		// glog.V(4).Infof("%s return the current buf %v", m.name, lastReadPosition)
-		return copiedBytes(logBuffer.buf[:logBuffer.pos]), logBuffer.offset, nil
+		// If current buffer is not empty, return it
+		if logBuffer.pos > 0 {
+			// glog.V(4).Infof("%s return the current buf %v", m.name, lastReadPosition)
+			return copiedBytes(logBuffer.buf[:logBuffer.pos]), logBuffer.offset, nil
+		}
+		// Buffer is empty and no data in prevBuffers - wait for new data
+		return nil, logBuffer.offset, nil
 	}
 
 	lastTs := lastReadPosition.Time.UnixNano()
+	// Inclusive boundary for first-read sentinel/zero offset
+	searchTs := lastTs
+	if lastReadPosition.Offset <= 0 {
+		if searchTs > math.MinInt64+1 { // prevent underflow
+			searchTs = searchTs - 1
+		}
+	}
 	l, h := 0, len(logBuffer.idx)-1
 
 	/*
@@ -792,14 +825,14 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 		mid := (l + h) / 2
 		pos := logBuffer.idx[mid]
 		_, t := readTs(logBuffer.buf, pos)
-		if t <= lastTs {
+		if t <= searchTs {
 			l = mid + 1
-		} else if lastTs < t {
+		} else if searchTs < t {
 			var prevT int64
 			if mid > 0 {
 				_, prevT = readTs(logBuffer.buf, logBuffer.idx[mid-1])
 			}
-			if prevT <= lastTs {
+			if prevT <= searchTs {
 				return copiedBytes(logBuffer.buf[pos:logBuffer.pos]), logBuffer.offset, nil
 			}
 			h = mid
diff --git a/weed/util/log_buffer/log_buffer_queryability_test.go b/weed/util/log_buffer/log_buffer_queryability_test.go
index 6e372d2b3..16dd0f9b0 100644
--- a/weed/util/log_buffer/log_buffer_queryability_test.go
+++ b/weed/util/log_buffer/log_buffer_queryability_test.go
@@ -236,3 +236,58 @@ func TestSchemaRegistryScenario(t *testing.T) {
 
 	t.Logf("Schema registry scenario test passed - schema value preserved: %d bytes", len(retrievedEntry.Data))
 }
+
+// TestTimeBasedFirstReadBeforeEarliest ensures starting slightly before earliest memory
+// does not force a disk resume and returns in-memory data (regression test)
+func TestTimeBasedFirstReadBeforeEarliest(t *testing.T) {
+	flushed := false
+	logBuffer := NewLogBuffer("local", 10*time.Minute,
+		func(logBuffer *LogBuffer, startTime, stopTime time.Time, buf []byte, minOffset, maxOffset int64) {
+			// keep in memory; we just want earliest time populated
+			_ = buf
+		},
+		func(startPosition MessagePosition, stopTsNs int64, eachLogEntryFn EachLogEntryFuncType) (MessagePosition, bool, error) {
+			// disk should not be consulted in this regression path
+			return startPosition, false, nil
+		},
+		func() {})
+
+	// Seed one entry so earliestTime is set
+	baseTs := time.Now().Add(-time.Second)
+	entry := &filer_pb.LogEntry{TsNs: baseTs.UnixNano(), Data: []byte("x"), Key: []byte("k"), Offset: 0}
+	logBuffer.AddLogEntryToBuffer(entry)
+	_ = flushed
+
+	// Start read 1ns before earliest memory, with offset sentinel (-2)
+	startPos := NewMessagePosition(baseTs.Add(-time.Nanosecond).UnixNano(), -2)
+	buf, _, err := logBuffer.ReadFromBuffer(startPos)
+	if err != nil {
+		t.Fatalf("ReadFromBuffer returned err: %v", err)
+	}
+	if buf == nil {
+		t.Fatalf("Expected in-memory data, got nil buffer")
+	}
+}
+
+// TestEarliestTimeExactRead ensures starting exactly at earliest time returns first entry (no skip)
+func TestEarliestTimeExactRead(t *testing.T) {
+	logBuffer := NewLogBuffer("local", 10*time.Minute,
+		func(logBuffer *LogBuffer, startTime, stopTime time.Time, buf []byte, minOffset, maxOffset int64) {},
+		func(startPosition MessagePosition, stopTsNs int64, eachLogEntryFn EachLogEntryFuncType) (MessagePosition, bool, error) {
+			return startPosition, false, nil
+		},
+		func() {})
+
+	ts := time.Now()
+	entry := &filer_pb.LogEntry{TsNs: ts.UnixNano(), Data: []byte("a"), Key: []byte("k"), Offset: 0}
+	logBuffer.AddLogEntryToBuffer(entry)
+
+	startPos := NewMessagePosition(ts.UnixNano(), -2)
+	buf, _, err := logBuffer.ReadFromBuffer(startPos)
+	if err != nil {
+		t.Fatalf("ReadFromBuffer err: %v", err)
+	}
+	if buf == nil || buf.Len() == 0 {
+		t.Fatalf("Expected data at earliest time, got nil/empty")
+	}
+}
diff --git a/weed/util/log_buffer/log_read.go b/weed/util/log_buffer/log_read.go
index 3b7b99ada..950604022 100644
--- a/weed/util/log_buffer/log_read.go
+++ b/weed/util/log_buffer/log_read.go
@@ -270,7 +270,7 @@ func (logBuffer *LogBuffer) LoopProcessLogDataWithOffset(readerName string, star
 				return
 			}
 
-			// CRITICAL FIX: If we're reading offset-based and there's no data in LogBuffer,
+			// If we're reading offset-based and there's no data in LogBuffer,
 			// return ResumeFromDiskError to let Subscribe try reading from disk again.
 			// This prevents infinite blocking when all data is on disk (e.g., after restart).
 			if startPosition.IsOffsetBased {
@@ -355,7 +355,6 @@ func (logBuffer *LogBuffer) LoopProcessLogDataWithOffset(readerName string, star
 				continue
 			}
 
-
 			glog.V(4).Infof("Unmarshaled log entry %d: TsNs=%d, Offset=%d, Key=%s", batchSize+1, logEntry.TsNs, logEntry.Offset, string(logEntry.Key))
 
 			// Handle offset-based filtering for offset-based start positions
@@ -377,7 +376,7 @@ func (logBuffer *LogBuffer) LoopProcessLogDataWithOffset(readerName string, star
 				// println("stopTsNs", stopTsNs, "logEntry.TsNs", logEntry.TsNs)
 				return
 			}
-			// CRITICAL FIX: Use logEntry.Offset + 1 to move PAST the current entry
+			// Use logEntry.Offset + 1 to move PAST the current entry
 			// This prevents infinite loops where we keep requesting the same offset
 			lastReadPosition = NewMessagePosition(logEntry.TsNs, logEntry.Offset+1)
 
diff --git a/weed/util/log_buffer/log_read_stateless.go b/weed/util/log_buffer/log_read_stateless.go
index b48413bc8..abc7d9ac0 100644
--- a/weed/util/log_buffer/log_read_stateless.go
+++ b/weed/util/log_buffer/log_read_stateless.go
@@ -121,7 +121,7 @@ func (logBuffer *LogBuffer) ReadMessagesAtOffset(startOffset int64, maxMessages
 		logBuffer.RUnlock()
 
 		// Data not in memory - attempt disk read if configured
-		// CRITICAL FIX: Don't return error here - data may be on disk!
+		// Don't return error here - data may be on disk!
 		// Fall through to disk read logic below
 		glog.V(2).Infof("[StatelessRead] Data at offset %d not in memory (buffer: %d-%d), attempting disk read",
 			startOffset, bufferStartOffset, currentBufferEnd)