diff --git a/weed/filer/filer.go b/weed/filer/filer.go index 6a8590c26..c605396a6 100644 --- a/weed/filer/filer.go +++ b/weed/filer/filer.go @@ -82,7 +82,13 @@ func NewFiler(masters pb.ServerDiscovery, grpcDialOption grpc.DialOption, filerH f.UniqueFilerId = -f.UniqueFilerId } - f.LocalMetaLogBuffer = log_buffer.NewLogBuffer("local", LogFlushInterval, f.logFlushFunc, f.readPersistedLogBufferPosition, notifyFn) + // ReadFromDiskFn is intentionally nil here. SubscribeLocalMetadata already + // manages disk reads explicitly with shouldReadFromDisk / lastCheckedFlushTsNs + // tracking. Setting ReadFromDiskFn would cause LoopProcessLogData to issue a + // redundant ReadPersistedLogBuffer call (ListDirectoryEntries + readahead + // goroutine) on every 250ms health-check tick when a subscriber encounters + // ResumeFromDiskError, adding significant CPU and GC pressure even when idle. + f.LocalMetaLogBuffer = log_buffer.NewLogBuffer("local", LogFlushInterval, f.logFlushFunc, nil, notifyFn) f.metaLogCollection = collection f.metaLogReplication = replication diff --git a/weed/filer/filer_notify.go b/weed/filer/filer_notify.go index ba83aeb78..a0dc198d6 100644 --- a/weed/filer/filer_notify.go +++ b/weed/filer/filer_notify.go @@ -262,16 +262,3 @@ func (f *Filer) ReadPersistedLogBuffer(startPosition log_buffer.MessagePosition, return } -func (f *Filer) readPersistedLogBufferPosition(startPosition log_buffer.MessagePosition, stopTsNs int64, eachLogEntryFn log_buffer.EachLogEntryFuncType) (lastReadPosition log_buffer.MessagePosition, isDone bool, err error) { - lastReadPosition = startPosition - - lastTsNs, isDone, err := f.ReadPersistedLogBuffer(startPosition, stopTsNs, eachLogEntryFn) - if err != nil { - return lastReadPosition, isDone, err - } - if lastTsNs != 0 { - lastReadPosition = log_buffer.NewMessagePosition(lastTsNs, 1) - } - - return lastReadPosition, isDone, nil -} diff --git a/weed/server/filer_grpc_server_sub_meta.go b/weed/server/filer_grpc_server_sub_meta.go index c436dee3c..89c5f2324 100644 --- a/weed/server/filer_grpc_server_sub_meta.go +++ b/weed/server/filer_grpc_server_sub_meta.go @@ -415,6 +415,24 @@ func (fs *FilerServer) SubscribeLocalMetadata(req *filer_pb.SubscribeMetadataReq time.Unix(0, lastDiskReadTsNs), time.Unix(0, currentReadTsNs)) continue } + // No flush or read-position progress — there may be a gap + // between the last persisted data and the earliest in-memory + // data (e.g. a slow consumer that fell behind while writes + // already stopped). Skip forward to the earliest in-memory + // time so the consumer can resume instead of blocking forever. + earliestTime := fs.filer.LocalMetaLogBuffer.GetEarliestTime() + if !earliestTime.IsZero() && earliestTime.After(lastReadTime.Time) { + glog.V(3).Infof("gap detected: skipping from %v to earliest memory time %v for %v", + lastReadTime.Time, earliestTime, clientName) + lastReadTime = log_buffer.NewMessagePosition(earliestTime.UnixNano(), -2) + // Clear the stale ResumeFromDiskError so the next + // iteration's shouldReadFromDisk path (triggered by the + // advanced lastReadTime) doesn't re-enter the gap branch + // at line 360 with earliestTime == lastReadTime.Time and + // stall on listenersCond.Wait(). + readInMemoryLogErr = nil + continue + } // No progress possible, wait for new data to arrive (event-driven, not polling) fs.listenersLock.Lock() atomic.AddInt64(&fs.listenersWaits, 1) diff --git a/weed/util/log_buffer/log_buffer.go b/weed/util/log_buffer/log_buffer.go index e8ee7e5ed..13724fee3 100644 --- a/weed/util/log_buffer/log_buffer.go +++ b/weed/util/log_buffer/log_buffer.go @@ -615,8 +615,27 @@ func (logBuffer *LogBuffer) invalidateAllDiskCacheChunks() { } } +// GetEarliestTime returns the oldest timestamp still resident in the buffer. +// It must consider the sealed prev buffers in addition to the active buffer, +// because ReadFromBuffer's tsMemory (and therefore ResumeFromDiskError) is +// computed from the min across both. Returning only the active startTime +// would cause gap-detection callers to skip past data still living in prev +// buffers, and can also silently equal the consumer's lastReadTime and +// stall on listenersCond.Wait(). func (logBuffer *LogBuffer) GetEarliestTime() time.Time { - return logBuffer.startTime + logBuffer.RLock() + defer logBuffer.RUnlock() + + earliest := logBuffer.startTime + for _, prevBuf := range logBuffer.prevBuffers.buffers { + if prevBuf.startTime.IsZero() { + continue + } + if earliest.IsZero() || prevBuf.startTime.Before(earliest) { + earliest = prevBuf.startTime + } + } + return earliest } func (logBuffer *LogBuffer) HasData() bool { diff --git a/weed/util/log_buffer/log_buffer_test.go b/weed/util/log_buffer/log_buffer_test.go index 78a5d74f8..1d4243f54 100644 --- a/weed/util/log_buffer/log_buffer_test.go +++ b/weed/util/log_buffer/log_buffer_test.go @@ -452,80 +452,60 @@ func TestReadFromBuffer_InitializedFromDisk(t *testing.T) { } // TestLoopProcessLogDataWithOffset_DiskReadRetry tests that when a subscriber -// reads from disk before flush completes, it continues to retry disk reads -// and eventually finds the data after flush completes. -// This reproduces the Schema Registry timeout issue on first start. +// reads from disk before data is available, it continues to retry disk reads +// and eventually finds the data once it appears. This reproduces the Schema +// Registry timeout where the first read happened before the data landed on +// disk and the loop never retried. +// +// The data is delivered exclusively through the disk-read path (no in-memory +// AddToBuffer) so the test deterministically asserts that retries happen, +// rather than racing the in-memory delivery path. func TestLoopProcessLogDataWithOffset_DiskReadRetry(t *testing.T) { diskReadCallCount := 0 diskReadMu := sync.Mutex{} - dataFlushedToDisk := false - var flushedData []*filer_pb.LogEntry + dataReady := false + mockEntry := &filer_pb.LogEntry{ + Key: []byte("key-0"), + Data: []byte("message-0"), + TsNs: time.Now().UnixNano(), + Offset: 0, + } - // Create a readFromDiskFn that simulates the race condition readFromDiskFn := func(startPosition MessagePosition, stopTsNs int64, eachLogEntryFn EachLogEntryFuncType) (MessagePosition, bool, error) { diskReadMu.Lock() diskReadCallCount++ callNum := diskReadCallCount - hasData := dataFlushedToDisk + ready := dataReady diskReadMu.Unlock() - t.Logf("DISK READ #%d: startOffset=%d, dataFlushedToDisk=%v", callNum, startPosition.Offset, hasData) + t.Logf("DISK READ #%d: startOffset=%d, dataReady=%v", callNum, startPosition.Offset, ready) - if !hasData { - // Simulate: data not yet on disk (flush hasn't completed) - t.Logf(" → No data found (flush not completed yet)") + if !ready { + t.Logf(" → No data on disk yet") return startPosition, false, nil } - // Data is now on disk, process it - t.Logf(" → Found %d entries on disk", len(flushedData)) - for _, entry := range flushedData { - if entry.Offset >= startPosition.Offset { - isDone, err := eachLogEntryFn(entry) - if err != nil || isDone { - return NewMessagePositionFromOffset(entry.Offset + 1), isDone, err - } + if mockEntry.Offset >= startPosition.Offset { + isDone, err := eachLogEntryFn(mockEntry) + if err != nil || isDone { + return NewMessagePositionFromOffset(mockEntry.Offset + 1), isDone, err } } - return NewMessagePositionFromOffset(int64(len(flushedData))), false, nil + return NewMessagePositionFromOffset(mockEntry.Offset + 1), false, nil } - 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)) - // Simulate writing to disk - diskReadMu.Lock() - dataFlushedToDisk = true - // Parse the buffer and add entries to flushedData - // For this test, we'll just create mock entries - flushedData = append(flushedData, &filer_pb.LogEntry{ - Key: []byte("key-0"), - Data: []byte("message-0"), - TsNs: time.Now().UnixNano(), - Offset: 0, - }) - diskReadMu.Unlock() - } - - logBuffer := NewLogBuffer("test", 1*time.Minute, flushFn, readFromDiskFn, nil) + logBuffer := NewLogBuffer("test", 1*time.Minute, nil, readFromDiskFn, nil) defer logBuffer.ShutdownLogBuffer() - // Simulate the race condition: - // 1. Subscriber starts reading from offset 0 - // 2. Data is not yet flushed - // 3. Loop calls readFromDiskFn → no data found - // 4. A bit later, data gets flushed - // 5. Loop should continue and call readFromDiskFn again - receivedMessages := 0 mu := sync.Mutex{} - maxIterations := 50 // Allow up to 50 iterations (500ms with 10ms sleep each) + maxIterations := 50 iterationCount := 0 waitForDataFn := func() bool { mu.Lock() defer mu.Unlock() iterationCount++ - // Stop after receiving message or max iterations return receivedMessages == 0 && iterationCount < maxIterations } @@ -534,10 +514,9 @@ func TestLoopProcessLogDataWithOffset_DiskReadRetry(t *testing.T) { receivedMessages++ mu.Unlock() t.Logf("✉️ RECEIVED: offset=%d key=%s", offset, string(logEntry.Key)) - return true, nil // Stop after first message + return true, nil } - // Start the reader in a goroutine var readerWg sync.WaitGroup readerWg.Add(1) go func() { @@ -547,27 +526,29 @@ func TestLoopProcessLogDataWithOffset_DiskReadRetry(t *testing.T) { t.Logf("📋 Reader finished: isDone=%v, err=%v", isDone, err) }() - // Wait a bit to let the first disk read happen (returns no data) - time.Sleep(50 * time.Millisecond) - - // Now add data and flush it - t.Logf("➕ Adding message to buffer...") - if err := logBuffer.AddToBuffer(&mq_pb.DataMessage{ - Key: []byte("key-0"), - Value: []byte("message-0"), - TsNs: time.Now().UnixNano(), - }); err != nil { - t.Fatalf("Failed to add buffer: %v", err) + // Wait until the reader has issued at least one no-op disk read so we + // know it has entered the retry loop, then publish the data. + deadline := time.Now().Add(2 * time.Second) + for { + diskReadMu.Lock() + seen := diskReadCallCount + diskReadMu.Unlock() + if seen >= 1 { + break + } + if time.Now().After(deadline) { + t.Fatalf("reader never called readFromDiskFn") + } + time.Sleep(5 * time.Millisecond) } - // Force flush - t.Logf("Force flushing...") - logBuffer.ForceFlush() + t.Logf("➕ Marking data as ready on disk") + diskReadMu.Lock() + dataReady = true + diskReadMu.Unlock() - // Wait for reader to finish readerWg.Wait() - // Check results diskReadMu.Lock() finalDiskReadCount := diskReadCallCount diskReadMu.Unlock() @@ -584,12 +565,12 @@ func TestLoopProcessLogDataWithOffset_DiskReadRetry(t *testing.T) { if finalDiskReadCount < 2 { t.Errorf("CRITICAL BUG REPRODUCED: Disk read was only called %d time(s)", finalDiskReadCount) - t.Errorf("Expected: Multiple disk reads as the loop continues after flush completes") - t.Errorf("This is why Schema Registry times out - it reads once before flush, never re-reads after flush") + t.Errorf("Expected: Multiple disk reads as the loop continues after data lands") + t.Errorf("This is why Schema Registry times out - it reads once before data is available, never re-reads") } if finalReceivedMessages == 0 { - t.Errorf("SCHEMA REGISTRY TIMEOUT REPRODUCED: No messages received even after flush") + t.Errorf("SCHEMA REGISTRY TIMEOUT REPRODUCED: No messages received even after data landed") t.Errorf("The subscriber is stuck because disk reads are not retried") } else { t.Logf("✓ SUCCESS: Message received after %d disk read attempts", finalDiskReadCount)