From f69e2a0c2d1c02b9183dcc0399a5bc3b5cf0ec60 Mon Sep 17 00:00:00 2001 From: chrislu Date: Fri, 17 Oct 2025 12:48:12 -0700 Subject: [PATCH] feat: Add unit tests to reproduce buffer flush offset gaps Phase 7: Unit Test Creation Created comprehensive unit tests in log_buffer_flush_gap_test.go: - TestFlushOffsetGap_ReproduceDataLoss: Tests for gaps between disk and memory - TestFlushOffsetGap_CheckPrevBuffers: Tests if data stuck in prevBuffers - TestFlushOffsetGap_ConcurrentWriteAndFlush: Tests race conditions - TestFlushOffsetGap_ForceFlushAdvancesBuffer: Tests offset advancement Initial Findings: - Tests run but don't reproduce exact production scenario - Reason: AddToBuffer doesn't auto-assign offsets (stays at 0) - In production: messages come with pre-assigned offsets from MQ broker - Need to use AddLogEntryToBuffer with explicit offsets instead Test Structure: - Flush callback captures minOffset, maxOffset, buffer contents - Parse flushed buffers to extract actual messages - Compare flushed offsets vs in-memory offsets - Detect gaps, overlaps, and missing data Next: Enhance tests to use explicit offset assignment to match production scenario --- .../log_buffer/log_buffer_flush_gap_test.go | 429 ++++++++++++++++++ 1 file changed, 429 insertions(+) create mode 100644 weed/util/log_buffer/log_buffer_flush_gap_test.go diff --git a/weed/util/log_buffer/log_buffer_flush_gap_test.go b/weed/util/log_buffer/log_buffer_flush_gap_test.go new file mode 100644 index 000000000..e00b3754f --- /dev/null +++ b/weed/util/log_buffer/log_buffer_flush_gap_test.go @@ -0,0 +1,429 @@ +package log_buffer + +import ( + "fmt" + "sync" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/mq_pb" + "google.golang.org/protobuf/proto" +) + +// TestFlushOffsetGap_ReproduceDataLoss reproduces the critical bug where messages +// are lost in the gap between flushed disk data and in-memory buffer. +// +// OBSERVED BEHAVIOR FROM LOGS: +// Request offset: 1764 +// Disk contains: 1000-1763 (764 messages) +// Memory buffer starts at: 1800 +// Gap: 1764-1799 (36 messages) ← MISSING! +// +// This test verifies: +// 1. All messages sent to buffer are accounted for +// 2. No gaps exist between disk and memory offsets +// 3. Flushed data and in-memory data have continuous offset ranges +func TestFlushOffsetGap_ReproduceDataLoss(t *testing.T) { + var flushedMessages []*filer_pb.LogEntry + var flushMu sync.Mutex + + 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)) + + // Parse and store flushed messages + flushMu.Lock() + defer flushMu.Unlock() + + // Parse buffer to extract messages + parsedCount := 0 + for pos := 0; pos+4 < len(buf); { + if pos+4 > len(buf) { + break + } + + size := uint32(buf[pos])<<24 | uint32(buf[pos+1])<<16 | uint32(buf[pos+2])<<8 | uint32(buf[pos+3]) + if pos+4+int(size) > len(buf) { + break + } + + entryData := buf[pos+4 : pos+4+int(size)] + logEntry := &filer_pb.LogEntry{} + if err := proto.Unmarshal(entryData, logEntry); err == nil { + flushedMessages = append(flushedMessages, logEntry) + parsedCount++ + } + + pos += 4 + int(size) + } + + t.Logf(" Parsed %d messages from flush buffer", parsedCount) + } + + logBuffer := NewLogBuffer("test", 100*time.Millisecond, flushFn, nil, nil) + defer logBuffer.ShutdownLogBuffer() + + // Send 100 messages + messageCount := 100 + t.Logf("Sending %d messages...", messageCount) + + for i := 0; i < messageCount; i++ { + logBuffer.AddToBuffer(&mq_pb.DataMessage{ + Key: []byte(fmt.Sprintf("key-%d", i)), + Value: []byte(fmt.Sprintf("message-%d", i)), + TsNs: time.Now().UnixNano(), + }) + } + + // Force flush multiple times to simulate real workload + t.Logf("Forcing flush...") + logBuffer.ForceFlush() + + // Add more messages after flush + for i := messageCount; i < messageCount+50; i++ { + logBuffer.AddToBuffer(&mq_pb.DataMessage{ + Key: []byte(fmt.Sprintf("key-%d", i)), + Value: []byte(fmt.Sprintf("message-%d", i)), + TsNs: time.Now().UnixNano(), + }) + } + + // Force another flush + logBuffer.ForceFlush() + time.Sleep(200 * time.Millisecond) // Wait for flush to complete + + // Now check the buffer state + logBuffer.RLock() + bufferStartOffset := logBuffer.bufferStartOffset + currentOffset := logBuffer.offset + pos := logBuffer.pos + logBuffer.RUnlock() + + flushMu.Lock() + flushedCount := len(flushedMessages) + var maxFlushedOffset int64 = -1 + var minFlushedOffset int64 = -1 + if flushedCount > 0 { + minFlushedOffset = flushedMessages[0].Offset + maxFlushedOffset = flushedMessages[flushedCount-1].Offset + } + flushMu.Unlock() + + t.Logf("\nBUFFER STATE AFTER FLUSH:") + t.Logf(" bufferStartOffset: %d", bufferStartOffset) + t.Logf(" currentOffset (HWM): %d", currentOffset) + t.Logf(" pos (bytes in buffer): %d", pos) + t.Logf(" Messages sent: %d (offsets 0-%d)", messageCount+50, messageCount+49) + t.Logf(" Messages flushed to disk: %d (offsets %d-%d)", flushedCount, minFlushedOffset, maxFlushedOffset) + + // CRITICAL CHECK: Is there a gap between flushed data and memory buffer? + if flushedCount > 0 && maxFlushedOffset >= 0 { + gap := bufferStartOffset - (maxFlushedOffset + 1) + + t.Logf("\nOFFSET CONTINUITY CHECK:") + t.Logf(" Last flushed offset: %d", maxFlushedOffset) + t.Logf(" Buffer starts at: %d", bufferStartOffset) + t.Logf(" Gap: %d offsets", gap) + + if gap > 0 { + t.Errorf("❌ CRITICAL BUG REPRODUCED: OFFSET GAP DETECTED!") + t.Errorf(" Disk has offsets %d-%d", minFlushedOffset, maxFlushedOffset) + t.Errorf(" Memory buffer starts at: %d", bufferStartOffset) + t.Errorf(" MISSING OFFSETS: %d-%d (%d messages)", maxFlushedOffset+1, bufferStartOffset-1, gap) + t.Errorf(" These messages are LOST - neither on disk nor in memory!") + } else if gap < 0 { + t.Errorf("❌ OFFSET OVERLAP: Memory buffer starts BEFORE last flushed offset!") + t.Errorf(" This indicates data corruption or race condition") + } else { + t.Logf("✅ PASS: No gap detected - offsets are continuous") + } + + // Check if we can read all expected offsets + t.Logf("\nREADABILITY CHECK:") + for testOffset := int64(0); testOffset < currentOffset; testOffset += 10 { + // Try to read from buffer + requestPosition := NewMessagePositionFromOffset(testOffset) + buf, _, err := logBuffer.ReadFromBuffer(requestPosition) + + isReadable := (buf != nil && len(buf.Bytes()) > 0) || err == ResumeFromDiskError + status := "✅" + if !isReadable && err == nil { + status = "❌ NOT READABLE" + } + + t.Logf(" Offset %d: %s (buf=%v, err=%v)", testOffset, status, buf != nil, err) + + // If offset is in the gap, it should fail to read + if flushedCount > 0 && testOffset > maxFlushedOffset && testOffset < bufferStartOffset { + if isReadable { + t.Errorf(" Unexpected: Offset %d in gap range should NOT be readable!", testOffset) + } else { + t.Logf(" Expected: Offset %d in gap is not readable (data lost)", testOffset) + } + } + } + } + + // Check that all sent messages are accounted for + expectedMessageCount := messageCount + 50 + messagesInMemory := int(currentOffset - bufferStartOffset) + totalAccountedFor := flushedCount + messagesInMemory + + t.Logf("\nMESSAGE ACCOUNTING:") + t.Logf(" Expected: %d messages", expectedMessageCount) + t.Logf(" Flushed to disk: %d", flushedCount) + t.Logf(" In memory buffer: %d (offset range %d-%d)", messagesInMemory, bufferStartOffset, currentOffset-1) + t.Logf(" Total accounted for: %d", totalAccountedFor) + t.Logf(" Missing: %d messages", expectedMessageCount-totalAccountedFor) + + if totalAccountedFor < expectedMessageCount { + t.Errorf("❌ DATA LOSS CONFIRMED: %d messages are missing!", expectedMessageCount-totalAccountedFor) + } else { + t.Logf("✅ All messages accounted for") + } +} + +// TestFlushOffsetGap_CheckPrevBuffers tests if messages might be stuck in prevBuffers +// instead of being properly flushed to disk. +func TestFlushOffsetGap_CheckPrevBuffers(t *testing.T) { + var flushCount int + var flushMu sync.Mutex + + flushFn := func(logBuffer *LogBuffer, startTime, stopTime time.Time, buf []byte, minOffset, maxOffset int64) { + flushMu.Lock() + flushCount++ + count := flushCount + flushMu.Unlock() + + t.Logf("FLUSH #%d: minOffset=%d maxOffset=%d size=%d bytes", count, minOffset, maxOffset, len(buf)) + } + + logBuffer := NewLogBuffer("test", 100*time.Millisecond, flushFn, nil, nil) + defer logBuffer.ShutdownLogBuffer() + + // Send messages in batches with flushes in between + for batch := 0; batch < 5; batch++ { + t.Logf("\nBatch %d:", batch) + + // Send 20 messages + for i := 0; i < 20; i++ { + offset := int64(batch*20 + i) + logBuffer.AddToBuffer(&mq_pb.DataMessage{ + Key: []byte(fmt.Sprintf("key-%d", offset)), + Value: []byte(fmt.Sprintf("message-%d", offset)), + TsNs: time.Now().UnixNano(), + }) + } + + // Check state before flush + logBuffer.RLock() + beforeFlushOffset := logBuffer.offset + beforeFlushStart := logBuffer.bufferStartOffset + logBuffer.RUnlock() + + // Force flush + logBuffer.ForceFlush() + time.Sleep(50 * time.Millisecond) + + // Check state after flush + logBuffer.RLock() + afterFlushOffset := logBuffer.offset + afterFlushStart := logBuffer.bufferStartOffset + prevBufferCount := len(logBuffer.prevBuffers.buffers) + + // Check prevBuffers state + t.Logf(" Before flush: offset=%d, bufferStartOffset=%d", beforeFlushOffset, beforeFlushStart) + t.Logf(" After flush: offset=%d, bufferStartOffset=%d, prevBuffers=%d", + afterFlushOffset, afterFlushStart, prevBufferCount) + + // Check each prevBuffer + for i, prevBuf := range logBuffer.prevBuffers.buffers { + if prevBuf.size > 0 { + t.Logf(" prevBuffer[%d]: offsets %d-%d, size=%d bytes (NOT FLUSHED!)", + i, prevBuf.startOffset, prevBuf.offset, prevBuf.size) + } + } + logBuffer.RUnlock() + + // CRITICAL: Check if bufferStartOffset advanced correctly + expectedNewStart := beforeFlushOffset + if afterFlushStart != expectedNewStart { + t.Errorf(" ❌ bufferStartOffset mismatch!") + t.Errorf(" Expected: %d (= offset before flush)", expectedNewStart) + t.Errorf(" Actual: %d", afterFlushStart) + t.Errorf(" Gap: %d offsets", expectedNewStart-afterFlushStart) + } + } +} + +// TestFlushOffsetGap_ConcurrentWriteAndFlush tests for race conditions +// between writing new messages and flushing old ones. +func TestFlushOffsetGap_ConcurrentWriteAndFlush(t *testing.T) { + var allFlushedOffsets []int64 + var flushMu sync.Mutex + + flushFn := func(logBuffer *LogBuffer, startTime, stopTime time.Time, buf []byte, minOffset, maxOffset int64) { + t.Logf("FLUSH: offsets %d-%d (%d bytes)", minOffset, maxOffset, len(buf)) + + flushMu.Lock() + // Record the offset range that was flushed + for offset := minOffset; offset <= maxOffset; offset++ { + allFlushedOffsets = append(allFlushedOffsets, offset) + } + flushMu.Unlock() + } + + logBuffer := NewLogBuffer("test", 50*time.Millisecond, flushFn, nil, nil) + defer logBuffer.ShutdownLogBuffer() + + // Concurrently write messages and force flushes + var wg sync.WaitGroup + + // Writer goroutine + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < 200; i++ { + logBuffer.AddToBuffer(&mq_pb.DataMessage{ + Key: []byte(fmt.Sprintf("key-%d", i)), + Value: []byte(fmt.Sprintf("message-%d", i)), + TsNs: time.Now().UnixNano(), + }) + if i%50 == 0 { + time.Sleep(10 * time.Millisecond) + } + } + }() + + // Flusher goroutine + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < 5; i++ { + time.Sleep(30 * time.Millisecond) + logBuffer.ForceFlush() + } + }() + + wg.Wait() + time.Sleep(200 * time.Millisecond) // Wait for final flush + + // Check final state + logBuffer.RLock() + finalOffset := logBuffer.offset + finalBufferStart := logBuffer.bufferStartOffset + logBuffer.RUnlock() + + flushMu.Lock() + flushedCount := len(allFlushedOffsets) + flushMu.Unlock() + + expectedCount := int(finalOffset) + inMemory := int(finalOffset - finalBufferStart) + totalAccountedFor := flushedCount + inMemory + + t.Logf("\nFINAL STATE:") + t.Logf(" Total messages sent: %d (offsets 0-%d)", expectedCount, expectedCount-1) + t.Logf(" Flushed to disk: %d", flushedCount) + t.Logf(" In memory: %d (offsets %d-%d)", inMemory, finalBufferStart, finalOffset-1) + t.Logf(" Total accounted: %d", totalAccountedFor) + t.Logf(" Missing: %d", expectedCount-totalAccountedFor) + + if totalAccountedFor < expectedCount { + t.Errorf("❌ DATA LOSS in concurrent scenario: %d messages missing!", expectedCount-totalAccountedFor) + } +} + +// TestFlushOffsetGap_ForceFlushAdvancesBuffer tests if ForceFlush +// properly advances bufferStartOffset after flushing. +func TestFlushOffsetGap_ForceFlushAdvancesBuffer(t *testing.T) { + flushedRanges := []struct{ min, max int64 }{} + var flushMu sync.Mutex + + flushFn := func(logBuffer *LogBuffer, startTime, stopTime time.Time, buf []byte, minOffset, maxOffset int64) { + flushMu.Lock() + flushedRanges = append(flushedRanges, struct{ min, max int64 }{minOffset, maxOffset}) + flushMu.Unlock() + t.Logf("FLUSH: offsets %d-%d", minOffset, maxOffset) + } + + logBuffer := NewLogBuffer("test", time.Hour, flushFn, nil, nil) // Long interval, manual flush only + defer logBuffer.ShutdownLogBuffer() + + // Send messages, flush, check state - repeat + for round := 0; round < 3; round++ { + t.Logf("\n=== ROUND %d ===", round) + + // Check state before adding messages + logBuffer.RLock() + beforeOffset := logBuffer.offset + beforeStart := logBuffer.bufferStartOffset + logBuffer.RUnlock() + + t.Logf("Before adding: offset=%d, bufferStartOffset=%d", beforeOffset, beforeStart) + + // Add 10 messages + for i := 0; i < 10; i++ { + logBuffer.AddToBuffer(&mq_pb.DataMessage{ + Key: []byte(fmt.Sprintf("round-%d-msg-%d", round, i)), + Value: []byte(fmt.Sprintf("data-%d-%d", round, i)), + TsNs: time.Now().UnixNano(), + }) + } + + // Check state after adding + logBuffer.RLock() + afterAddOffset := logBuffer.offset + afterAddStart := logBuffer.bufferStartOffset + logBuffer.RUnlock() + + t.Logf("After adding: offset=%d, bufferStartOffset=%d", afterAddOffset, afterAddStart) + + // Force flush + t.Logf("Forcing flush...") + logBuffer.ForceFlush() + time.Sleep(100 * time.Millisecond) + + // Check state after flush + logBuffer.RLock() + afterFlushOffset := logBuffer.offset + afterFlushStart := logBuffer.bufferStartOffset + logBuffer.RUnlock() + + t.Logf("After flush: offset=%d, bufferStartOffset=%d", afterFlushOffset, afterFlushStart) + + // CRITICAL CHECK: bufferStartOffset should advance to where offset was before flush + if afterFlushStart != afterAddOffset { + t.Errorf("❌ FLUSH BUG: bufferStartOffset did NOT advance correctly!") + t.Errorf(" Expected bufferStartOffset=%d (= offset after add)", afterAddOffset) + t.Errorf(" Actual bufferStartOffset=%d", afterFlushStart) + t.Errorf(" Gap: %d offsets WILL BE LOST", afterAddOffset-afterFlushStart) + } else { + t.Logf("✅ bufferStartOffset correctly advanced to %d", afterFlushStart) + } + } + + // Final verification: check all offset ranges are continuous + flushMu.Lock() + t.Logf("\n=== FLUSHED RANGES ===") + for i, r := range flushedRanges { + t.Logf("Flush #%d: offsets %d-%d", i, r.min, r.max) + + // Check continuity with previous flush + if i > 0 { + prevMax := flushedRanges[i-1].max + currentMin := r.min + gap := currentMin - (prevMax + 1) + + if gap > 0 { + t.Errorf("❌ GAP between flush #%d and #%d: %d offsets missing!", i-1, i, gap) + } else if gap < 0 { + t.Errorf("❌ OVERLAP between flush #%d and #%d: %d offsets duplicated!", i-1, i, -gap) + } else { + t.Logf(" ✅ Continuous with previous flush") + } + } + } + flushMu.Unlock() +} +