db: reap ephemeral GC watcher goroutine on cancel and reschedule

A stopped timer never fires, so the per-node watcher leaked until close.
This commit is contained in:
Kristoffer Dalby authored and Kristoffer Dalby committed 2026-06-08 10:04:49 +02:00
1 parent 06d6816dc9
commit 4c165ae5e7
2 files changed
+58 -10

No files matched your search

@@ -89,6 +89,36 @@ func TestEphemeralGarbageCollectorGoRoutineLeak(t *testing.T) {
t.Logf("Final number of goroutines: %d", runtime.NumGoroutine()) t.Logf("Final number of goroutines: %d", runtime.NumGoroutine())
} }
// TestEphemeralGarbageCollectorCancelReapsGoroutine verifies that Cancel (and
// reschedule) reaps the per-node watcher goroutine while the collector is still
// running, rather than leaking it until Close. The production churn path is an
// ephemeral node disconnecting (Schedule) then reconnecting (Cancel) before the
// long expiry timer fires; a stopped timer never fires, so a watcher parked
// only on <-timer.C would otherwise leak on every cycle.
func TestEphemeralGarbageCollectorCancelReapsGoroutine(t *testing.T) {
gc := NewEphemeralGarbageCollector(func(types.NodeID) {})
go gc.Start()
defer gc.Close()
baseline := runtime.NumGoroutine()
const (
iterations = 1000
nodeID = types.NodeID(42)
)
for range iterations {
gc.Schedule(nodeID, time.Hour) // disconnect: long timer, will not fire
gc.Cancel(nodeID) // reconnect: must reap the watcher
}
assert.EventuallyWithT(t, func(c *assert.CollectT) {
assert.LessOrEqual(c, runtime.NumGoroutine(), baseline+10,
"per-node goroutines leaked on Cancel/reschedule")
}, 2*time.Second, 20*time.Millisecond, "watcher goroutines should be reaped")
}
// TestEphemeralGarbageCollectorReschedule is a test for the rescheduling of nodes in [EphemeralGarbageCollector]. // TestEphemeralGarbageCollectorReschedule is a test for the rescheduling of nodes in [EphemeralGarbageCollector].
// It creates a new [EphemeralGarbageCollector], schedules a node for deletion with a longer expiry, // It creates a new [EphemeralGarbageCollector], schedules a node for deletion with a longer expiry,
// and then reschedules it with a shorter expiry, and verifies that the node is deleted only once. // and then reschedules it with a shorter expiry, and verifies that the node is deleted only once.
+28 -10
View File
@@ -393,17 +393,25 @@ type EphemeralGarbageCollector struct {
mu sync.Mutex mu sync.Mutex
deleteFunc func(types.NodeID) deleteFunc func(types.NodeID)
toBeDeleted map[types.NodeID]*time.Timer toBeDeleted map[types.NodeID]ephemeralTimer
deleteCh chan types.NodeID deleteCh chan types.NodeID
cancelCh chan struct{} cancelCh chan struct{}
} }
// ephemeralTimer pairs a node's pending-deletion timer with a done channel
// used to reap its watcher goroutine on Cancel or reschedule. Without it a
// stopped timer never fires and the goroutine leaks until Close.
type ephemeralTimer struct {
timer *time.Timer
done chan struct{}
}
// NewEphemeralGarbageCollector creates a new [EphemeralGarbageCollector], it takes // NewEphemeralGarbageCollector creates a new [EphemeralGarbageCollector], it takes
// a deleteFunc that will be called when a node is scheduled for deletion. // a deleteFunc that will be called when a node is scheduled for deletion.
func NewEphemeralGarbageCollector(deleteFunc func(types.NodeID)) *EphemeralGarbageCollector { func NewEphemeralGarbageCollector(deleteFunc func(types.NodeID)) *EphemeralGarbageCollector {
return &EphemeralGarbageCollector{ return &EphemeralGarbageCollector{
toBeDeleted: make(map[types.NodeID]*time.Timer), toBeDeleted: make(map[types.NodeID]ephemeralTimer),
deleteCh: make(chan types.NodeID, 10), deleteCh: make(chan types.NodeID, 10),
cancelCh: make(chan struct{}), cancelCh: make(chan struct{}),
deleteFunc: deleteFunc, deleteFunc: deleteFunc,
@@ -416,8 +424,8 @@ func (e *EphemeralGarbageCollector) Close() {
defer e.mu.Unlock() defer e.mu.Unlock()
// Stop all timers // Stop all timers
for _, timer := range e.toBeDeleted { for _, t := range e.toBeDeleted {
timer.Stop() t.timer.Stop()
} }
// Close the cancel channel to signal all goroutines to exit // Close the cancel channel to signal all goroutines to exit
@@ -440,13 +448,16 @@ func (e *EphemeralGarbageCollector) Schedule(nodeID types.NodeID, expiry time.Du
// Continue with scheduling // Continue with scheduling
} }
// If a timer already exists for this node, stop it first // If a timer already exists for this node, stop it and reap its
if oldTimer, exists := e.toBeDeleted[nodeID]; exists { // watcher goroutine before scheduling a fresh one.
oldTimer.Stop() if old, exists := e.toBeDeleted[nodeID]; exists {
old.timer.Stop()
close(old.done)
} }
timer := time.NewTimer(expiry) timer := time.NewTimer(expiry)
e.toBeDeleted[nodeID] = timer done := make(chan struct{})
e.toBeDeleted[nodeID] = ephemeralTimer{timer: timer, done: done}
// Start a goroutine to handle the timer completion // Start a goroutine to handle the timer completion
go func() { go func() {
select { select {
@@ -461,7 +472,13 @@ func (e *EphemeralGarbageCollector) Schedule(nodeID types.NodeID, expiry time.Du
case <-e.cancelCh: case <-e.cancelCh:
// GC is shutting down, don't send to deleteCh // GC is shutting down, don't send to deleteCh
return return
case <-done:
// Cancelled or rescheduled before the send landed.
return
} }
case <-done:
// Cancelled or rescheduled before the timer fired.
return
case <-e.cancelCh: case <-e.cancelCh:
// If the GC is closed, exit the goroutine // If the GC is closed, exit the goroutine
return return
@@ -474,8 +491,9 @@ func (e *EphemeralGarbageCollector) Cancel(nodeID types.NodeID) {
e.mu.Lock() e.mu.Lock()
defer e.mu.Unlock() defer e.mu.Unlock()
if timer, ok := e.toBeDeleted[nodeID]; ok { if t, ok := e.toBeDeleted[nodeID]; ok {
timer.Stop() t.timer.Stop()
close(t.done)
delete(e.toBeDeleted, nodeID) delete(e.toBeDeleted, nodeID)
} }
} }