filer: share one log-buffer window snapshot across all subscriber reads (#10267)

* log_buffer: share one window snapshot across all subscriber reads

Every in-memory read handed each subscriber a private pooled copy of the
window it wanted, so N subscribers reading the same data cost N copies of
up to 8MB each -- and slow consumers (grpc send backpressure) held those
copies live for their whole iteration. With hundreds of mount subscribers
that multiplied into gigabytes of live heap on the filer.

Share the bytes instead of copying per reader:

- Sealed windows get a lazily created GC-owned snapshot, made once by the
  first reader and handed out zero-copy to the rest. The snapshot travels
  with its window when SealBuffer shifts slots, so recycling the sealed
  array never invalidates it.
- The current window keeps a shared snapshot of its append-only prefix
  buf[:pos], extended on demand; each byte is copied once per window
  (writer-rate-bound) instead of once per reader. At seal a fully
  extended prefix becomes the sealed window's snapshot.

ReadFromBuffer now reports whether the returned buffer is a pooled copy
(flush path) or a shared view that must not be released; the read loops
only recycle pooled buffers.

With 200 subscribers consuming at grpc pace over sealed and current
windows, peak live heap drops from 5.2GB to 178MB.

* log_buffer: clear released read buffer so a panic cannot double-free it

The read loops release the previous iteration's pooled buffer and then
call ReadFromBuffer. If that call panicked before reassigning bytesBuf,
the deferred cleanup would put the same buffer into the pool a second
time, letting two future readers share one backing array. Nil the
pointer at the release site so the defer sees nothing to free.

weed/util/log_buffer/log_buffer.go

@@ -94,6 +94,12 @@ type LogBuffer struct {
 	hasOffsets bool
 	// Disk chunk cache for historical data reads
 	diskChunkCache *DiskChunkCache
+	// curSnap is a GC-owned copy of the current window's append-only prefix
+	// buf[:len(curSnap)], shared by all readers so each byte is copied once per
+	// window instead of once per reader. Extended lazily under curSnapMu; reset
+	// at seal. Existing holders keep their prefix slices, which never mutate.
+	curSnapMu sync.Mutex
+	curSnap   []byte
 	sync.RWMutex
 }
 
@@ -607,6 +613,13 @@ 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)
+		// Hand a fully extended prefix snapshot to the sealed slot so sealed
+		// readers reuse it instead of re-copying the window; reset for the next
+		// window either way (holders keep their immutable prefix slices).
+		if len(logBuffer.curSnap) == logBuffer.pos {
+			logBuffer.prevBuffers.buffers[len(logBuffer.prevBuffers.buffers)-1].snapshot = logBuffer.curSnap[:logBuffer.pos:logBuffer.pos]
+		}
+		logBuffer.curSnap = nil
 		// 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{}
@@ -702,7 +715,11 @@ func (d *dataToFlush) releaseMemory() {
 	bufferPool.Put(d.data)
 }
 
-func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bufferCopy *bytes.Buffer, batchIndex int64, err error) {
+// ReadFromBuffer returns the in-memory log data at lastReadPosition. isPooled
+// reports whether the returned buffer is a private pooled copy the caller must
+// return via ReleaseMemory; when false the buffer wraps a snapshot shared with
+// other readers and must not be released (or written to).
+func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bufferCopy *bytes.Buffer, batchIndex int64, isPooled bool, err error) {
 	logBuffer.RLock()
 	defer logBuffer.RUnlock()
 
@@ -729,19 +746,19 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 				// Case 3: try disk read (historical data might exist)
 				if requestedOffset < logBuffer.offset {
 					// Data was in the buffer range but buffer is now empty = flushed to disk
-					return nil, -2, ResumeFromDiskError
+					return nil, -2, false, ResumeFromDiskError
 				}
 				// requestedOffset == logBuffer.offset: Current position
 				// CRITICAL: For subscribers starting from offset 0, try disk read first
 				// (historical data might exist from previous runs)
 				if requestedOffset == 0 && logBuffer.bufferStartOffset == 0 && logBuffer.offset == 0 {
 					// Initial state: try disk read before waiting for new data
-					return nil, -2, ResumeFromDiskError
+					return nil, -2, false, ResumeFromDiskError
 				}
 				// Otherwise, wait for new data to arrive
-				return nil, logBuffer.offset, nil
+				return nil, logBuffer.offset, false, nil
 			}
-			return copiedBytes(logBuffer.buf[:logBuffer.pos]), logBuffer.offset, nil
+			return logBuffer.currentSnapshotView(0, logBuffer.pos), logBuffer.offset, false, nil
 		}
 
 		// Check previous buffers for the requested offset
@@ -751,9 +768,9 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 				// (prevBuffers are created when buffer is flushed)
 				if buf.size == 0 {
 					// Empty prevBuffer covering this offset means data was flushed
-					return nil, -2, ResumeFromDiskError
+					return nil, -2, false, ResumeFromDiskError
 				}
-				return copiedBytes(buf.buf[:buf.size]), buf.offset, nil
+				return sharedBufferView(buf, 0), buf.offset, false, nil
 			}
 		}
 
@@ -761,16 +778,16 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 		if requestedOffset < logBuffer.bufferStartOffset {
 			// Data not in current buffers - must be on disk (flushed or never existed)
 			// Return ResumeFromDiskError to trigger disk read
-			return nil, -2, ResumeFromDiskError
+			return nil, -2, false, ResumeFromDiskError
 		}
 
 		if requestedOffset > logBuffer.offset {
 			// Future data, not available yet
-			return nil, logBuffer.offset, nil
+			return nil, logBuffer.offset, false, nil
 		}
 
 		// Offset not found - return nil
-		return nil, logBuffer.offset, nil
+		return nil, logBuffer.offset, false, nil
 	}
 
 	// TIMESTAMP-BASED READ (original logic)
@@ -799,7 +816,7 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 		// Buffer is empty - return ResumeFromDiskError so caller can read from disk
 		// This fixes issue #4977 where SubscribeMetadata stalls because
 		// MetaAggregator.MetaLogBuffer is empty in single-filer setups
-		return nil, -2, ResumeFromDiskError
+		return nil, -2, false, ResumeFromDiskError
 	} else if lastReadPosition.Time.Before(tsMemory) { // case 2.3
 		// For time-based reads, only check timestamp for disk reads
 		// Don't use offset comparisons as they're not meaningful for time-based subscriptions
@@ -813,7 +830,7 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 			// Treat first read with sentinel/zero offset as inclusive of earliest in-memory data
 		} else {
 			// Data not in memory buffers - read from disk
-			return nil, -2, ResumeFromDiskError
+			return nil, -2, false, ResumeFromDiskError
 		}
 	}
 
@@ -822,17 +839,17 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 	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
+			return nil, logBuffer.offset, false, nil
 		}
 	}
 	if lastReadPosition.Time.After(logBuffer.stopTime) && !logBuffer.stopTime.IsZero() {
-		return nil, logBuffer.offset, nil
+		return nil, logBuffer.offset, false, nil
 	}
 	// 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) {
-				return copiedBytes(buf.buf[:buf.size]), buf.offset, nil
+				return sharedBufferView(buf, 0), buf.offset, false, nil
 			}
 			if !buf.startTime.After(lastReadPosition.Time) && buf.stopTime.After(lastReadPosition.Time) {
 				searchTime := lastReadPosition.Time
@@ -843,19 +860,19 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 				if err != nil {
 					// Buffer corruption detected - return error wrapped with ErrBufferCorrupted
 					glog.Errorf("ReadFromBuffer: buffer corruption in prevBuffer: %v", err)
-					return nil, -1, fmt.Errorf("%w: %v", ErrBufferCorrupted, err)
+					return nil, -1, false, fmt.Errorf("%w: %v", ErrBufferCorrupted, err)
 				}
 				if pos < buf.size {
-					return copiedBytes(buf.buf[pos:buf.size]), buf.offset, nil
+					return sharedBufferView(buf, pos), buf.offset, false, nil
 				}
 			}
 		}
 		// If current buffer is not empty, return it
 		if logBuffer.pos > 0 {
-			return copiedBytes(logBuffer.buf[:logBuffer.pos]), logBuffer.offset, nil
+			return logBuffer.currentSnapshotView(0, logBuffer.pos), logBuffer.offset, false, nil
 		}
 		// Buffer is empty and no data in prevBuffers - wait for new data
-		return nil, logBuffer.offset, nil
+		return nil, logBuffer.offset, false, nil
 	}
 
 	lastTs := lastReadPosition.Time.UnixNano()
@@ -887,7 +904,7 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 		if err != nil {
 			// Buffer corruption detected in binary search
 			glog.Errorf("ReadFromBuffer: buffer corruption at idx[%d] pos %d: %v", mid, pos, err)
-			return nil, -1, fmt.Errorf("%w: %v", ErrBufferCorrupted, err)
+			return nil, -1, false, fmt.Errorf("%w: %v", ErrBufferCorrupted, err)
 		}
 		if t <= searchTs {
 			l = mid + 1
@@ -898,11 +915,11 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 				if err != nil {
 					// Buffer corruption detected in binary search (previous entry)
 					glog.Errorf("ReadFromBuffer: buffer corruption at idx[%d] pos %d: %v", mid-1, logBuffer.idx[mid-1], err)
-					return nil, -1, fmt.Errorf("%w: %v", ErrBufferCorrupted, err)
+					return nil, -1, false, fmt.Errorf("%w: %v", ErrBufferCorrupted, err)
 				}
 			}
 			if prevT <= searchTs {
-				return copiedBytes(logBuffer.buf[pos:logBuffer.pos]), logBuffer.offset, nil
+				return logBuffer.currentSnapshotView(pos, logBuffer.pos), logBuffer.offset, false, nil
 			}
 			h = mid
 		}
@@ -910,9 +927,13 @@ func (logBuffer *LogBuffer) ReadFromBuffer(lastReadPosition MessagePosition) (bu
 
 	// Binary search didn't find the timestamp - data may have been flushed to disk already
 	// Returning -2 signals to caller that data is not available in memory
-	return nil, -2, nil
+	return nil, -2, false, nil
 
 }
+
+// ReleaseMemory returns a pooled buffer for reuse. Only call it for buffers
+// ReadFromBuffer reported as pooled: recycling a shared sealed-window view
+// would let the next copiedBytes overwrite bytes other readers still hold.
 func (logBuffer *LogBuffer) ReleaseMemory(b *bytes.Buffer) {
 	bufferPool.Put(b)
 }
@@ -944,6 +965,37 @@ func copiedBytes(buf []byte) (copied *bytes.Buffer) {
 	return
 }
 
+// sharedBufferView wraps the sealed window's shared snapshot from pos onward.
+// The returned buffer aliases memory shared with other readers: it must be
+// treated as read-only and never passed to ReleaseMemory (the full-slice cap
+// makes an accidental append reallocate instead of scribbling on the snapshot).
+func sharedBufferView(mb *MemBuffer, pos int) *bytes.Buffer {
+	snap := mb.sharedSnapshot()
+	return bytes.NewBuffer(snap[pos:len(snap):len(snap)])
+}
+
+// currentSnapshotView returns buf[from:to] of the current window as a view of
+// the shared prefix snapshot, extending the snapshot to cover [0:to) first.
+// buf[:pos] is append-only until the window seals, so extension only ever
+// appends stable bytes; earlier holders' slices are unaffected. Callers must
+// hold the read lock (keeps buf and pos stable during extension).
+func (logBuffer *LogBuffer) currentSnapshotView(from, to int) *bytes.Buffer {
+	logBuffer.curSnapMu.Lock()
+	if cap(logBuffer.curSnap) < to {
+		// First use in this window (or the rare window whose array outgrew the
+		// previous one): size to the window array so extensions never reallocate.
+		grown := make([]byte, len(logBuffer.curSnap), len(logBuffer.buf))
+		copy(grown, logBuffer.curSnap)
+		logBuffer.curSnap = grown
+	}
+	if len(logBuffer.curSnap) < to {
+		logBuffer.curSnap = append(logBuffer.curSnap, logBuffer.buf[len(logBuffer.curSnap):to]...)
+	}
+	snap := logBuffer.curSnap[:to]
+	logBuffer.curSnapMu.Unlock()
+	return bytes.NewBuffer(snap[from:to:to])
+}
+
 func readTs(buf []byte, pos int) (size int, ts int64, err error) {
 	// Bounds check for size field (overflow-safe)
 	if pos < 0 || pos > len(buf)-4 {

weed/util/log_buffer/log_buffer_corruption_test.go

@@ -118,7 +118,7 @@ func TestReadFromBufferCorruption(t *testing.T) {
 
 	// Try to read - should detect corruption
 	startPos := MessagePosition{Time: lb.startTime}
-	buf, offset, err := lb.ReadFromBuffer(startPos)
+	buf, offset, _, err := lb.ReadFromBuffer(startPos)
 
 	// Should return corruption error
 	if err == nil {

weed/util/log_buffer/log_buffer_flush_gap_test.go

@@ -148,7 +148,7 @@ func TestFlushOffsetGap_ReproduceDataLoss(t *testing.T) {
 		for testOffset := int64(0); testOffset < currentOffset; testOffset += 10 {
 			// Try to read from buffer
 			requestPosition := NewMessagePositionFromOffset(testOffset)
-			buf, _, err := logBuffer.ReadFromBuffer(requestPosition)
+			buf, _, _, err := logBuffer.ReadFromBuffer(requestPosition)
 
 			isReadable := (buf != nil && len(buf.Bytes()) > 0) || err == ResumeFromDiskError
 			status := "OK"

weed/util/log_buffer/log_buffer_queryability_test.go

@@ -50,7 +50,7 @@ func TestBufferQueryability(t *testing.T) {
 
 	// Test immediate queryability - read from buffer starting from beginning
 	startPosition := NewMessagePosition(0, 0) // Start from beginning
-	bufferCopy, batchIndex, err := logBuffer.ReadFromBuffer(startPosition)
+	bufferCopy, batchIndex, _, err := logBuffer.ReadFromBuffer(startPosition)
 
 	if err != nil {
 		t.Fatalf("ReadFromBuffer failed: %v", err)
@@ -131,7 +131,7 @@ func TestMultipleEntriesQueryability(t *testing.T) {
 
 	// Read all entries
 	startPosition := NewMessagePosition(0, 0)
-	bufferCopy, batchIndex, err := logBuffer.ReadFromBuffer(startPosition)
+	bufferCopy, batchIndex, _, err := logBuffer.ReadFromBuffer(startPosition)
 
 	if err != nil {
 		t.Fatalf("ReadFromBuffer failed: %v", err)
@@ -207,7 +207,7 @@ func TestSchemaRegistryScenario(t *testing.T) {
 
 	// Simulate the SQL query scenario - read from offset 0
 	startPosition := NewMessagePosition(0, 0)
-	bufferCopy, _, err := logBuffer.ReadFromBuffer(startPosition)
+	bufferCopy, _, _, err := logBuffer.ReadFromBuffer(startPosition)
 
 	if err != nil {
 		t.Fatalf("Schema registry scenario failed: %v", err)
@@ -268,7 +268,7 @@ func TestTimeBasedFirstReadBeforeEarliest(t *testing.T) {
 
 	// Start read 1ns before earliest memory, with offset sentinel (-2)
 	startPos := NewMessagePosition(baseTs.Add(-time.Nanosecond).UnixNano(), -2)
-	buf, _, err := logBuffer.ReadFromBuffer(startPos)
+	buf, _, _, err := logBuffer.ReadFromBuffer(startPos)
 	if err != nil {
 		t.Fatalf("ReadFromBuffer returned err: %v", err)
 	}
@@ -293,7 +293,7 @@ func TestEarliestTimeExactRead(t *testing.T) {
 	}
 
 	startPos := NewMessagePosition(ts.UnixNano(), -2)
-	buf, _, err := logBuffer.ReadFromBuffer(startPos)
+	buf, _, _, err := logBuffer.ReadFromBuffer(startPos)
 	if err != nil {
 		t.Fatalf("ReadFromBuffer err: %v", err)
 	}

weed/util/log_buffer/log_buffer_test.go

@@ -101,7 +101,7 @@ func TestReadFromBufferTimestampBased_AfterFlushReturnsNewerData(t *testing.T) {
 		}
 	}
 
-	buf, _, err := lb.ReadFromBuffer(NewMessagePosition(sealed.stopTime.UnixNano(), sealed.offset))
+	buf, _, _, err := lb.ReadFromBuffer(NewMessagePosition(sealed.stopTime.UnixNano(), sealed.offset))
 	if err != nil {
 		t.Fatalf("ReadFromBuffer returned error: %v", err)
 	}
@@ -200,7 +200,7 @@ func TestReadFromBuffer_OldOffsetReturnsResumeFromDiskError(t *testing.T) {
 			requestPosition := NewMessagePositionFromOffset(tt.requestedOffset)
 
 			// Try to read from the buffer
-			buf, batchIdx, err := lb.ReadFromBuffer(requestPosition)
+			buf, batchIdx, _, err := lb.ReadFromBuffer(requestPosition)
 
 			// Verify the error matches expectations
 			if tt.expectError != nil {
@@ -257,7 +257,7 @@ func TestReadFromBuffer_OldOffsetWithNoPrevBuffers(t *testing.T) {
 
 	// Before the fix, this would return (nil, offset, nil) causing an infinite wait
 	// After the fix, this should return ResumeFromDiskError
-	buf, batchIdx, err := lb.ReadFromBuffer(requestPosition)
+	buf, batchIdx, _, err := lb.ReadFromBuffer(requestPosition)
 
 	t.Logf("DEBUG: ReadFromBuffer returned: buf=%v, batchIdx=%d, err=%v", buf != nil, batchIdx, err)
 	t.Logf("DEBUG: Buffer state: bufferStartOffset=%d, offset=%d, pos=%d",
@@ -295,7 +295,7 @@ func TestReadFromBuffer_EmptyBufferAtCurrentOffset(t *testing.T) {
 
 	// BUG: Without fix, this returns empty buffer instead of checking disk
 	// FIX: Should return ResumeFromDiskError because buffer is empty (pos=0) despite valid range
-	buf, batchIdx, err := lb.ReadFromBuffer(requestPosition)
+	buf, batchIdx, _, err := lb.ReadFromBuffer(requestPosition)
 
 	t.Logf("DEBUG: ReadFromBuffer returned: buf=%v, batchIdx=%d, err=%v", buf != nil, batchIdx, err)
 	t.Logf("DEBUG: Buffer state: bufferStartOffset=%d, offset=%d, pos=%d",
@@ -362,7 +362,7 @@ func TestReadFromBuffer_OffsetRanges(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			requestPosition := NewMessagePositionFromOffset(tc.requestedOffset)
-			_, _, err := lb.ReadFromBuffer(requestPosition)
+			_, _, _, err := lb.ReadFromBuffer(requestPosition)
 
 			if tc.expectedError != nil {
 				if err != tc.expectedError {
@@ -424,7 +424,7 @@ func TestReadFromBuffer_InitializedFromDisk(t *testing.T) {
 	// Schema Registry tries to read offset 0 (should be on disk)
 	requestPosition := NewMessagePositionFromOffset(0)
 
-	buf, batchIdx, err := lb.ReadFromBuffer(requestPosition)
+	buf, batchIdx, _, err := lb.ReadFromBuffer(requestPosition)
 
 	t.Logf("After writing new message:")
 	t.Logf("  bufferStartOffset=%d, offset=%d, pos=%d", lb.bufferStartOffset, lb.offset, lb.pos)

weed/util/log_buffer/log_read.go

@@ -101,6 +101,7 @@ func (logBuffer *LogBuffer) LoopProcessLogData(readerName string, startPosition
 
 	// loop through all messages
 	var bytesBuf *bytes.Buffer
+	var bytesBufPooled bool
 	var batchIndex int64
 	lastReadPosition = startPosition
 	var entryCounter int64
@@ -113,7 +114,7 @@ func (logBuffer *LogBuffer) LoopProcessLogData(readerName string, startPosition
 	// responsive 250ms cadence resumes for active readers.
 	caughtUpToDiskHead := false
 	defer func() {
-		if bytesBuf != nil {
+		if bytesBuf != nil && bytesBufPooled {
 			logBuffer.ReleaseMemory(bytesBuf)
 		}
 		// println("LoopProcessLogData", readerName, "sent messages total", entryCounter)
@@ -121,10 +122,11 @@ func (logBuffer *LogBuffer) LoopProcessLogData(readerName string, startPosition
 
 	for {
 
-		if bytesBuf != nil {
+		if bytesBuf != nil && bytesBufPooled {
 			logBuffer.ReleaseMemory(bytesBuf)
+			bytesBuf = nil // keep the deferred release from double-freeing if ReadFromBuffer panics
 		}
-		bytesBuf, batchIndex, err = logBuffer.ReadFromBuffer(lastReadPosition)
+		bytesBuf, batchIndex, bytesBufPooled, err = logBuffer.ReadFromBuffer(lastReadPosition)
 		if err == ResumeFromDiskError {
 			// Try to read from disk if readFromDiskFn is available
 			if logBuffer.ReadFromDiskFn != nil {
@@ -148,11 +150,11 @@ func (logBuffer *LogBuffer) LoopProcessLogData(readerName string, startPosition
 				// HasData() and ReadFromBuffer lock separately, so a racing write can make
 				// HasData() see data the empty-buffer read missed. Re-read; only bail if the
 				// position is genuinely behind the in-memory window (flushed to disk).
-				reBuf, _, reErr := logBuffer.ReadFromBuffer(lastReadPosition)
+				reBuf, _, rePooled, reErr := logBuffer.ReadFromBuffer(lastReadPosition)
 				if reErr == ResumeFromDiskError {
 					return lastReadPosition, isDone, ResumeFromDiskError
 				}
-				if reBuf != nil {
+				if reBuf != nil && rePooled {
 					logBuffer.ReleaseMemory(reBuf)
 				}
 				continue
@@ -315,13 +317,14 @@ func (logBuffer *LogBuffer) LoopProcessLogDataWithOffset(readerName string, star
 
 	// loop through all messages
 	var bytesBuf *bytes.Buffer
+	var bytesBufPooled bool
 	var offset int64
 	lastReadPosition = startPosition
 	var entryCounter int64
 	// See LoopProcessLogData for the caughtUpToDiskHead invariant.
 	caughtUpToDiskHead := false
 	defer func() {
-		if bytesBuf != nil {
+		if bytesBuf != nil && bytesBufPooled {
 			logBuffer.ReleaseMemory(bytesBuf)
 		}
 		// println("LoopProcessLogDataWithOffset", readerName, "sent messages total", entryCounter)
@@ -335,10 +338,11 @@ func (logBuffer *LogBuffer) LoopProcessLogDataWithOffset(readerName string, star
 			return
 		}
 
-		if bytesBuf != nil {
+		if bytesBuf != nil && bytesBufPooled {
 			logBuffer.ReleaseMemory(bytesBuf)
+			bytesBuf = nil // keep the deferred release from double-freeing if ReadFromBuffer panics
 		}
-		bytesBuf, offset, err = logBuffer.ReadFromBuffer(lastReadPosition)
+		bytesBuf, offset, bytesBufPooled, err = logBuffer.ReadFromBuffer(lastReadPosition)
 		glog.V(4).Infof("ReadFromBuffer for %s returned bytesBuf=%v, offset=%d, err=%v", readerName, bytesBuf != nil, offset, err)
 
 		// Check for buffer corruption error before other error handling
@@ -374,11 +378,11 @@ func (logBuffer *LogBuffer) LoopProcessLogDataWithOffset(readerName string, star
 				// HasData() and ReadFromBuffer lock separately, so a racing write can make
 				// HasData() see data the empty-buffer read missed. Re-read; only bail if the
 				// position is genuinely behind the in-memory window (flushed to disk).
-				reBuf, _, reErr := logBuffer.ReadFromBuffer(lastReadPosition)
+				reBuf, _, rePooled, reErr := logBuffer.ReadFromBuffer(lastReadPosition)
 				if reErr == ResumeFromDiskError {
 					return lastReadPosition, isDone, ResumeFromDiskError
 				}
-				if reBuf != nil {
+				if reBuf != nil && rePooled {
 					logBuffer.ReleaseMemory(reBuf)
 				}
 				continue

weed/util/log_buffer/sealed_buffer.go

@@ -2,6 +2,7 @@ package log_buffer
 
 import (
 	"fmt"
+	"sync"
 	"time"
 )
 
@@ -12,6 +13,25 @@ type MemBuffer struct {
 	stopTime    time.Time
 	startOffset int64 // First offset in this buffer
 	offset      int64 // Last offset in this buffer (endOffset)
+
+	// snapshot is a GC-owned copy of buf[:size] shared by all readers of this
+	// sealed window, so N subscribers reading the same window cost one copy
+	// instead of N pooled copies. Created lazily by the first reader; travels
+	// with the window when SealBuffer shifts slots. Immutable once created.
+	snapMu   sync.Mutex
+	snapshot []byte
+}
+
+// sharedSnapshot returns the shared read-only copy of this sealed window,
+// creating it on first use. Callers must hold the LogBuffer read lock, which
+// keeps buf stable (SealBuffer mutates slots only under the write lock).
+func (mb *MemBuffer) sharedSnapshot() []byte {
+	mb.snapMu.Lock()
+	defer mb.snapMu.Unlock()
+	if mb.snapshot == nil {
+		mb.snapshot = append([]byte(nil), mb.buf[:mb.size]...)
+	}
+	return mb.snapshot
 }
 
 type SealedBuffers struct {
@@ -41,6 +61,7 @@ func (sbs *SealedBuffers) SealBuffer(startTime, stopTime time.Time, buf []byte,
 		sbs.buffers[i].stopTime = sbs.buffers[i+1].stopTime
 		sbs.buffers[i].startOffset = sbs.buffers[i+1].startOffset
 		sbs.buffers[i].offset = sbs.buffers[i+1].offset
+		sbs.buffers[i].snapshot = sbs.buffers[i+1].snapshot // snapshot follows its window
 	}
 	sbs.buffers[size-1].buf = buf
 	sbs.buffers[size-1].size = pos
@@ -48,6 +69,7 @@ func (sbs *SealedBuffers) SealBuffer(startTime, stopTime time.Time, buf []byte,
 	sbs.buffers[size-1].stopTime = stopTime
 	sbs.buffers[size-1].startOffset = startOffset
 	sbs.buffers[size-1].offset = endOffset
+	sbs.buffers[size-1].snapshot = nil
 	return oldBuf
 }
 

weed/util/log_buffer/shared_snapshot_test.go

@@ -0,0 +1,268 @@
+package log_buffer
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"google.golang.org/protobuf/proto"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+)
+
+func addEntries(tb testing.TB, lb *LogBuffer, n int, payloadSize int, tag string) {
+	payload := make([]byte, payloadSize)
+	for i := 0; i < n; i++ {
+		key := []byte(fmt.Sprintf("/%s/%d", tag, i))
+		if err := lb.AddDataToBuffer(key, payload, time.Now().UnixNano()); err != nil {
+			tb.Fatal(err)
+		}
+	}
+}
+
+// Sealed-window reads must hand every reader the same shared snapshot rather
+// than a fresh copy each.
+func TestSealedReadsShareOneSnapshot(t *testing.T) {
+	lb := NewLogBuffer("test", time.Hour, func(*LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, nil, nil)
+	defer lb.ShutdownLogBuffer()
+
+	addEntries(t, lb, 10, 1024, "a")
+	lb.ForceFlush() // seal the window
+	addEntries(t, lb, 1, 1024, "b")
+
+	pos := NewMessagePosition(1, -2) // far in the past => sealed window hit
+	buf1, _, pooled1, err1 := lb.ReadFromBuffer(pos)
+	buf2, _, pooled2, err2 := lb.ReadFromBuffer(pos)
+	if err1 != nil || err2 != nil {
+		t.Fatalf("read errors: %v %v", err1, err2)
+	}
+	if buf1 == nil || buf2 == nil {
+		t.Fatal("expected sealed data in memory")
+	}
+	if pooled1 || pooled2 {
+		t.Fatalf("sealed reads should not be pooled copies: %v %v", pooled1, pooled2)
+	}
+	b1, b2 := buf1.Bytes(), buf2.Bytes()
+	if len(b1) == 0 || len(b1) != len(b2) {
+		t.Fatalf("unexpected lengths %d %d", len(b1), len(b2))
+	}
+	if &b1[0] != &b2[0] {
+		t.Fatal("sealed reads made independent copies; expected one shared snapshot")
+	}
+}
+
+// A shared snapshot must stay intact after its window rotates out and the
+// backing array is recycled and overwritten by new writes.
+func TestSharedSnapshotSurvivesRecycle(t *testing.T) {
+	lb := NewLogBuffer("test", time.Hour, func(*LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, nil, nil)
+	defer lb.ShutdownLogBuffer()
+
+	addEntries(t, lb, 5, 2048, "keep")
+	lb.ForceFlush()
+
+	pos := NewMessagePosition(1, -2)
+	buf, _, pooled, err := lb.ReadFromBuffer(pos)
+	if err != nil || buf == nil || pooled {
+		t.Fatalf("expected shared sealed read, got buf=%v pooled=%v err=%v", buf != nil, pooled, err)
+	}
+	before := append([]byte(nil), buf.Bytes()...)
+
+	// Rotate the sealed window all the way out so its array is recycled and
+	// overwritten with different content.
+	for i := 0; i < PreviousBufferCount+1; i++ {
+		addEntries(t, lb, 5, 2048, "overwrite")
+		lb.ForceFlush()
+	}
+	addEntries(t, lb, 5, 2048, "overwrite")
+
+	if string(before) != string(buf.Bytes()) {
+		t.Fatal("shared snapshot content changed after its window was recycled")
+	}
+}
+
+// Race test: concurrent lagging readers against a writer that keeps sealing
+// and recycling windows. Every delivered entry must unmarshal cleanly with the
+// expected payload — corruption here means a reader saw recycled bytes.
+func TestSharedSnapshotConcurrentIntegrity(t *testing.T) {
+	lb := NewLogBuffer("test", time.Hour, func(*LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, nil, nil)
+	defer lb.ShutdownLogBuffer()
+
+	const payloadByte = 0x5A
+	payload := make([]byte, 8*1024)
+	for i := range payload {
+		payload[i] = payloadByte
+	}
+
+	var stop sync.WaitGroup
+	done := make(chan struct{})
+	stop.Add(1)
+	go func() { // writer: fill and seal aggressively so arrays recycle under the readers
+		defer stop.Done()
+		for i := 0; ; i++ {
+			select {
+			case <-done:
+				return
+			default:
+			}
+			if err := lb.AddDataToBuffer([]byte(fmt.Sprintf("/k/%d", i)), payload, time.Now().UnixNano()); err != nil {
+				t.Error(err)
+				return
+			}
+			if i%200 == 0 {
+				lb.ForceFlush()
+			}
+		}
+	}()
+
+	var readers sync.WaitGroup
+	errCh := make(chan error, 16)
+	for r := 0; r < 8; r++ {
+		readers.Add(1)
+		go func(r int) {
+			defer readers.Done()
+			deadline := time.Now().Add(2 * time.Second)
+			for time.Now().Before(deadline) {
+				start := NewMessagePosition(time.Now().Add(-500*time.Millisecond).UnixNano(), -2)
+				lb.LoopProcessLogData(fmt.Sprintf("r%d", r), start, 0,
+					func() bool { return time.Now().Before(deadline) },
+					func(le *filer_pb.LogEntry) (bool, error) {
+						if len(le.Data) != len(payload) {
+							err := fmt.Errorf("payload length %d, want %d", len(le.Data), len(payload))
+							select {
+							case errCh <- err:
+							default:
+							}
+							return true, err
+						}
+						for _, b := range le.Data {
+							if b != payloadByte {
+								err := fmt.Errorf("corrupted payload byte %x", b)
+								select {
+								case errCh <- err:
+								default:
+								}
+								return true, err
+							}
+						}
+						return false, nil
+					})
+			}
+		}(r)
+	}
+	readers.Wait()
+	close(done)
+	stop.Wait()
+
+	select {
+	case err := <-errCh:
+		t.Fatal(err)
+	default:
+	}
+}
+
+// Current-window reads must share the prefix snapshot: two behind-readers get
+// views of the same backing array, and the view content must stay intact while
+// the writer keeps appending and eventually seals and recycles the window.
+func TestCurrentWindowPrefixSharing(t *testing.T) {
+	lb := NewLogBuffer("test", time.Hour, func(*LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, nil, nil)
+	defer lb.ShutdownLogBuffer()
+
+	addEntries(t, lb, 8, 1024, "cur")
+
+	pos := NewMessagePosition(1, -2) // behind => full current-window read
+	buf1, _, pooled1, err1 := lb.ReadFromBuffer(pos)
+	buf2, _, pooled2, err2 := lb.ReadFromBuffer(pos)
+	if err1 != nil || err2 != nil || buf1 == nil || buf2 == nil {
+		t.Fatalf("reads: %v %v %v %v", buf1 != nil, err1, buf2 != nil, err2)
+	}
+	if pooled1 || pooled2 {
+		t.Fatalf("current-window behind-reads should be shared, got pooled %v %v", pooled1, pooled2)
+	}
+	b1, b2 := buf1.Bytes(), buf2.Bytes()
+	if len(b1) == 0 || len(b1) != len(b2) || &b1[0] != &b2[0] {
+		t.Fatalf("expected one shared prefix snapshot, lens %d %d", len(b1), len(b2))
+	}
+	before := append([]byte(nil), b1...)
+
+	// Keep appending: the prefix must be extended, not reallocated or mutated.
+	addEntries(t, lb, 8, 1024, "more")
+	buf3, _, _, err3 := lb.ReadFromBuffer(pos)
+	if err3 != nil || buf3 == nil {
+		t.Fatalf("read3: %v %v", buf3 != nil, err3)
+	}
+	if len(buf3.Bytes()) <= len(before) {
+		t.Fatalf("extended read %d should exceed first read %d", len(buf3.Bytes()), len(before))
+	}
+	if string(buf3.Bytes()[:len(before)]) != string(before) {
+		t.Fatal("prefix changed when the snapshot was extended")
+	}
+
+	// Seal + rotate the window fully out; earlier views must stay intact.
+	for i := 0; i < PreviousBufferCount+1; i++ {
+		lb.ForceFlush()
+		addEntries(t, lb, 4, 1024, "rotate")
+	}
+	if string(b1) != string(before) {
+		t.Fatal("shared current-window view changed after seal and recycle")
+	}
+}
+
+// A sealed window whose prefix snapshot was fully extended must reuse it
+// rather than re-copying on the first sealed read.
+func TestSealHandsOffCompleteSnapshot(t *testing.T) {
+	lb := NewLogBuffer("test", time.Hour, func(*LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, nil, nil)
+	defer lb.ShutdownLogBuffer()
+
+	addEntries(t, lb, 6, 512, "h")
+	pos := NewMessagePosition(1, -2)
+	buf1, _, _, err := lb.ReadFromBuffer(pos) // extends prefix snapshot to full window
+	if err != nil || buf1 == nil {
+		t.Fatalf("read1: %v %v", buf1 != nil, err)
+	}
+	lb.ForceFlush()
+	addEntries(t, lb, 1, 512, "next")
+
+	buf2, _, pooled, err := lb.ReadFromBuffer(pos) // sealed read of the same window
+	if err != nil || buf2 == nil || pooled {
+		t.Fatalf("read2: buf=%v pooled=%v err=%v", buf2 != nil, pooled, err)
+	}
+	b1, b2 := buf1.Bytes(), buf2.Bytes()
+	if len(b1) == 0 || len(b1) > len(b2) || &b1[0] != &b2[0] {
+		t.Fatalf("sealed read should reuse the handed-off snapshot (lens %d %d)", len(b1), len(b2))
+	}
+}
+
+// Sanity: sealed-read content must byte-match what a proto round-trip expects.
+func TestSharedSnapshotContentMatches(t *testing.T) {
+	lb := NewLogBuffer("test", time.Hour, func(*LogBuffer, time.Time, time.Time, []byte, int64, int64) {}, nil, nil)
+	defer lb.ShutdownLogBuffer()
+
+	addEntries(t, lb, 3, 512, "x")
+	lb.ForceFlush()
+
+	buf, _, _, err := lb.ReadFromBuffer(NewMessagePosition(1, -2))
+	if err != nil || buf == nil {
+		t.Fatalf("read: buf=%v err=%v", buf != nil, err)
+	}
+	data := buf.Bytes()
+	count := 0
+	for pos := 0; pos+4 < len(data); {
+		size, _, err := readTs(data, pos)
+		if err != nil {
+			t.Fatalf("entry %d: %v", count, err)
+		}
+		var le filer_pb.LogEntry
+		if err := proto.Unmarshal(data[pos+4:pos+4+size], &le); err != nil {
+			t.Fatalf("entry %d unmarshal: %v", count, err)
+		}
+		if want := fmt.Sprintf("/x/%d", count); string(le.Key) != want {
+			t.Fatalf("entry %d key %q, want %q", count, le.Key, want)
+		}
+		pos += 4 + size
+		count++
+	}
+	if count != 3 {
+		t.Fatalf("read %d entries, want 3", count)
+	}
+}