log_buffer: stop closing a notification channel another reader still holds (#11177)

* fix(log_buffer): stop closing a notification channel another reader still holds - #10810

The report blames the polling loop for the busy spin, but that loop is
not what burns the core. LogBuffer keeps one notification channel per
subscriberID, and UnregisterSubscriber closes it. Two registrations that
share a subscriberID share that channel, which happens whenever a client
opens a second stream or an old stream has not yet noticed it was
replaced, so the first unregister closes a channel the other reader is
parked on. A closed channel makes every receive in
awaitNotificationOrTimeoutFor return instantly, and that reader then
spins at full speed for the rest of its life.

Subscriptions are now reference counted. Registering an existing
subscriberID hands back the same channel and raises the count; the
channel is only closed when the last holder unregisters.

* test: fail if the surviving reader stops instead of keeps reading

Review caught that the iteration count alone proves nothing: had
LoopProcessLogData returned when the duplicate reader unregistered, the
counter would sit at 0 and the assertion would pass without a reader
ever having been there to spin. Check the reader is still running before
trusting its low count.

---------

Co-authored-by: Junker der Provinz <jdp@braethoria.com>

weed/util/log_buffer/log_buffer.go

@@ -189,9 +189,9 @@ type LogBuffer struct {
 	notifyFn       func()
 	// Per-subscriber notification channels for instant wake-up
 	subscribersMu sync.RWMutex
-	subscribers   map[string]chan struct{} // subscriberID -> notification channel
+	subscribers   map[string]*subscription // subscriberID -> shared notification channel
 	// Notified only when a flush lands, for readers that cannot act on an append
-	flushSubscribers map[string]chan struct{}
+	flushSubscribers map[string]*subscription
 	isStopping       *atomic.Bool
 	shutdownCh       chan struct{}  // closed by ShutdownLogBuffer to wake blocked subscribers
 	loopsDone        sync.WaitGroup // loopFlush and loopInterval signal exit
@@ -223,8 +223,8 @@ func NewLogBuffer(name string, flushInterval time.Duration, flushFn LogFlushFunc
 		flushFn:          flushFn,
 		ReadFromDiskFn:   readFromDiskFn,
 		notifyFn:         notifyFn,
-		subscribers:      make(map[string]chan struct{}),
-		flushSubscribers: make(map[string]chan struct{}),
+		subscribers:      make(map[string]*subscription),
+		flushSubscribers: make(map[string]*subscription),
 		flushChan:        make(chan *dataToFlush, flushQueueDepth),
 		isStopping:       new(atomic.Bool),
 		shutdownCh:       make(chan struct{}),
@@ -242,32 +242,59 @@ func NewLogBuffer(name string, flushInterval time.Duration, flushFn LogFlushFunc
 	return lb
 }
 
+// subscription is one notification channel and the number of readers holding
+// it. Registrations that share a subscriberID share the channel - two streams
+// of the same client, or an old one that has not noticed its replacement yet -
+// so it may only be closed once the last of them unregisters. Closing it under
+// a reader parked in awaitNotificationOrTimeoutFor makes every receive there
+// succeed instantly, spinning that reader on a full core for the rest of its
+// life.
+type subscription struct {
+	notifyChan chan struct{}
+	refCount   int
+}
+
+func registerSubscription(subscriptions map[string]*subscription, subscriberID string) chan struct{} {
+	if existing, exists := subscriptions[subscriberID]; exists {
+		existing.refCount++
+		return existing.notifyChan
+	}
+
+	// Create buffered channel (size 1) so notifications never block
+	sub := &subscription{notifyChan: make(chan struct{}, 1), refCount: 1}
+	subscriptions[subscriberID] = sub
+	return sub.notifyChan
+}
+
+func unregisterSubscription(subscriptions map[string]*subscription, subscriberID string) {
+	sub, exists := subscriptions[subscriberID]
+	if !exists {
+		return
+	}
+	sub.refCount--
+	if sub.refCount > 0 {
+		return
+	}
+	close(sub.notifyChan)
+	delete(subscriptions, subscriberID)
+}
+
 // RegisterSubscriber registers a subscriber for instant notifications when data is written
 // Returns a channel that will receive notifications (<1ms latency)
 func (logBuffer *LogBuffer) RegisterSubscriber(subscriberID string) chan struct{} {
 	logBuffer.subscribersMu.Lock()
 	defer logBuffer.subscribersMu.Unlock()
 
-	// Check if already registered
-	if existingChan, exists := logBuffer.subscribers[subscriberID]; exists {
-		return existingChan
-	}
-
-	// Create buffered channel (size 1) so notifications never block
-	notifyChan := make(chan struct{}, 1)
-	logBuffer.subscribers[subscriberID] = notifyChan
-	return notifyChan
+	return registerSubscription(logBuffer.subscribers, subscriberID)
 }
 
 // UnregisterSubscriber removes a subscriber and closes its notification channel
+// once no other registration of the same subscriberID is left holding it.
 func (logBuffer *LogBuffer) UnregisterSubscriber(subscriberID string) {
 	logBuffer.subscribersMu.Lock()
 	defer logBuffer.subscribersMu.Unlock()
 
-	if ch, exists := logBuffer.subscribers[subscriberID]; exists {
-		close(ch)
-		delete(logBuffer.subscribers, subscriberID)
-	}
+	unregisterSubscription(logBuffer.subscribers, subscriberID)
 }
 
 // RegisterFlushSubscriber registers a subscriber woken only when a flush lands.
@@ -279,23 +306,16 @@ func (logBuffer *LogBuffer) RegisterFlushSubscriber(subscriberID string) chan st
 	logBuffer.subscribersMu.Lock()
 	defer logBuffer.subscribersMu.Unlock()
 
-	if existingChan, exists := logBuffer.flushSubscribers[subscriberID]; exists {
-		return existingChan
-	}
-	notifyChan := make(chan struct{}, 1)
-	logBuffer.flushSubscribers[subscriberID] = notifyChan
-	return notifyChan
+	return registerSubscription(logBuffer.flushSubscribers, subscriberID)
 }
 
 // UnregisterFlushSubscriber removes a flush subscriber and closes its channel
+// once no other registration of the same subscriberID is left holding it.
 func (logBuffer *LogBuffer) UnregisterFlushSubscriber(subscriberID string) {
 	logBuffer.subscribersMu.Lock()
 	defer logBuffer.subscribersMu.Unlock()
 
-	if ch, exists := logBuffer.flushSubscribers[subscriberID]; exists {
-		close(ch)
-		delete(logBuffer.flushSubscribers, subscriberID)
-	}
+	unregisterSubscription(logBuffer.flushSubscribers, subscriberID)
 }
 
 // IsOffsetInMemory checks if the given offset is available in the in-memory buffer
@@ -357,9 +377,9 @@ func (logBuffer *LogBuffer) notifySubscribers() {
 		return // No subscribers, skip notification
 	}
 
-	for _, notifyChan := range logBuffer.subscribers {
+	for _, sub := range logBuffer.subscribers {
 		select {
-		case notifyChan <- struct{}{}:
+		case sub.notifyChan <- struct{}{}:
 			// Notification sent successfully
 		default:
 			// Channel full - subscriber hasn't consumed previous notification yet
@@ -373,9 +393,9 @@ func (logBuffer *LogBuffer) notifyFlushSubscribers() {
 	logBuffer.subscribersMu.RLock()
 	defer logBuffer.subscribersMu.RUnlock()
 
-	for _, notifyChan := range logBuffer.flushSubscribers {
+	for _, sub := range logBuffer.flushSubscribers {
 		select {
-		case notifyChan <- struct{}{}:
+		case sub.notifyChan <- struct{}{}:
 		default:
 		}
 	}

weed/util/log_buffer/log_read_test.go

@@ -3,6 +3,7 @@ package log_buffer
 import (
 	"context"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -574,6 +575,71 @@ func TestLoopProcessLogData_SlowConsumerFallsBehind(t *testing.T) {
 	}
 }
 
+// TestLoopProcessLogData_DuplicateReaderLeaving is a regression test for issue
+// #10810: an idle metadata subscriber pinning a full core inside
+// awaitNotificationOrTimeoutFor. Two readers registered under the same name
+// share one notification channel, which happens whenever a client reconnects
+// before its previous stream noticed the disconnect. When the departing one
+// closed that channel, the surviving reader's select won on it instantly on
+// every pass, so the loop ran flat out - allocating a timer per iteration -
+// until the client went away.
+func TestLoopProcessLogData_DuplicateReaderLeaving(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)
+	defer logBuffer.ShutdownLogBuffer()
+
+	const readerName = "localMeta:s3@"
+	startPosition := NewMessagePosition(time.Now().UnixNano(), -2)
+	eachLogEntryFn := func(logEntry *filer_pb.LogEntry) (bool, error) { return false, nil }
+
+	// startReader runs a reader and reports how many times it went round the
+	// loop; the loop calls waitForDataFn exactly once per pass.
+	startReader := func(iterations *atomic.Int64, stop *atomic.Bool) chan struct{} {
+		done := make(chan struct{})
+		go func() {
+			defer close(done)
+			logBuffer.LoopProcessLogData(readerName, startPosition, 0, func() bool {
+				iterations.Add(1)
+				return !stop.Load()
+			}, eachLogEntryFn)
+		}()
+		for iterations.Load() == 0 {
+			time.Sleep(time.Millisecond)
+		}
+		return done
+	}
+
+	var leavingIterations, survivorIterations atomic.Int64
+	var stopLeaving, stopSurvivor atomic.Bool
+	leavingDone := startReader(&leavingIterations, &stopLeaving)
+	survivorDone := startReader(&survivorIterations, &stopSurvivor)
+
+	stopLeaving.Store(true)
+	<-leavingDone
+
+	const observeFor = 500 * time.Millisecond
+	before := survivorIterations.Load()
+	time.Sleep(observeFor)
+	spun := survivorIterations.Load() - before
+
+	// A low iteration count only means "not spinning" if the reader was still
+	// there to spin. Had LoopProcessLogData returned when its duplicate left,
+	// spun would be 0 and the count below would pass while proving nothing.
+	select {
+	case <-survivorDone:
+		t.Fatal("surviving reader returned while its duplicate unregistered; it must keep reading")
+	default:
+	}
+
+	stopSurvivor.Store(true)
+	<-survivorDone
+
+	maxIterations := int64(observeFor/notificationHealthCheckInterval) + 1
+	if spun > maxIterations {
+		t.Errorf("surviving reader looped %d times in %v, expected at most %d (suggests busy-waiting)", spun, observeFor, maxIterations)
+	}
+}
+
 // BenchmarkLoopProcessLogDataWithOffset_EmptyBuffer benchmarks the performance
 // of the loop with an empty buffer to ensure no busy-waiting
 func BenchmarkLoopProcessLogDataWithOffset_EmptyBuffer(b *testing.B) {