From b2f0b29f95d4d5bfff3e35bc6248203dc36d7bbd Mon Sep 17 00:00:00 2001
From: Milos Pesic <milos.pesic@livekit.io>
Date: Mon, 13 Jul 2026 14:18:50 +0200
Subject: [PATCH] Allow updating the pulse sink reaper at runtime

Follow-up to #1289, addressing the review comment about updating the reaper config after service start.
The monitor previously snapshotted pulse_sink_reap_grace_sec once in NewMonitor and only started the reaper goroutine if reaping was enabled at boot, so the grace could not be tuned and reaping could not be enabled or disabled without a restart.

The reaper goroutine now always runs and re-reads the grace (an atomic.Duration on the monitor) every tick. Monitor.SetPulseSinkReapGraceSec lets embedders apply an updated value on a live service: positive values take effect on the
next tick, non-positive disables reaping. While disabled, the loop skips pactl entirely and clears its orphan tracking, so re-enabling starts a fresh grace period instead of instantly reaping sinks tracked before the flip. Config semantics are unchanged (0 = default, negative = disable).
---
 pkg/stats/monitor.go      |  8 +++-----
 pkg/stats/pulse_reaper.go | 29 ++++++++++++++++++++++++++---
 2 files changed, 29 insertions(+), 8 deletions(-)

diff --git a/pkg/stats/monitor.go b/pkg/stats/monitor.go
index 620aefaa..04413199 100644
--- a/pkg/stats/monitor.go
+++ b/pkg/stats/monitor.go
@@ -58,7 +58,7 @@ type Monitor struct {
 	nodeID             string
 	clusterID          string
 	cpuCostConfig      *config.CPUCostConfig
-	pulseSinkReapGrace time.Duration
+	pulseSinkReapGrace atomic.Duration
 
 	promCPULoad           prometheus.Gauge
 	promCgroupMemory      prometheus.Gauge
@@ -122,10 +122,8 @@ func NewMonitor(conf *config.ServiceConfig, svc Service) (*Monitor, error) {
 
 	m.initPrometheus()
 
-	if conf.PulseSinkReapGraceSec > 0 {
-		m.pulseSinkReapGrace = time.Duration(conf.PulseSinkReapGraceSec) * time.Second
-		go m.runPulseSinkReaper()
-	}
+	m.SetPulseSinkReapGraceSec(conf.PulseSinkReapGraceSec)
+	go m.runPulseSinkReaper()
 
 	procStats, err := hwstats.NewProcMonitor(m.updateEgressStats)
 	if err != nil {
diff --git a/pkg/stats/pulse_reaper.go b/pkg/stats/pulse_reaper.go
index ac8feaaa..82732051 100644
--- a/pkg/stats/pulse_reaper.go
+++ b/pkg/stats/pulse_reaper.go
@@ -30,17 +30,40 @@ const (
 	pulseReapCmdTimeout = 10 * time.Second
 )
 
+// SetPulseSinkReapGraceSec applies an updated reap grace, taking effect on the
+// reaper's next tick. A non-positive grace disables reaping.
+func (m *Monitor) SetPulseSinkReapGraceSec(graceSec int) {
+	m.pulseSinkReapGrace.Store(time.Duration(graceSec) * time.Second)
+}
+
+// a grace shorter than the default interval would otherwise be quantized back up to it
+func pulseReapTick(grace time.Duration) time.Duration {
+	if grace > 0 {
+		return min(pulseReapInterval, grace)
+	}
+	return pulseReapInterval
+}
+
 // runPulseSinkReaper unloads null-sinks left on the shared pulse daemon by handlers that
 // exited without cleanup, before they accumulate and hit the pulse client limit.
+// The grace is re-read every tick, so reaping can be reconfigured or disabled at runtime.
 func (m *Monitor) runPulseSinkReaper() {
 	firstOrphaned := make(map[string]time.Time)
 	listFailing := false
 
-	// a grace shorter than the default interval would otherwise be quantized back up to it
-	ticker := time.NewTicker(min(pulseReapInterval, m.pulseSinkReapGrace))
+	ticker := time.NewTicker(pulseReapTick(m.pulseSinkReapGrace.Load()))
 	defer ticker.Stop()
 
 	for range ticker.C {
+		grace := m.pulseSinkReapGrace.Load()
+		ticker.Reset(pulseReapTick(grace))
+
+		if grace <= 0 {
+			// re-enabling starts a fresh grace period instead of reaping instantly
+			clear(firstOrphaned)
+			continue
+		}
+
 		info, err := listPulse()
 		if err != nil {
 			if !listFailing {
@@ -51,7 +74,7 @@ func (m *Monitor) runPulseSinkReaper() {
 		}
 		listFailing = false
 
-		toReap, egressSinks := planPulseSinkReaps(info.Sinks, m.activeEgressIDs(), firstOrphaned, m.pulseSinkReapGrace, time.Now())
+		toReap, egressSinks := planPulseSinkReaps(info.Sinks, m.activeEgressIDs(), firstOrphaned, grace, time.Now())
 		m.promPulseSinks.Set(float64(egressSinks))
 
 		for _, sink := range toReap {