From 08d46e2db090f9fc4637c3ba57cc037dd7975ff0 Mon Sep 17 00:00:00 2001 From: Teodor Calin Date: Tue, 28 Jul 2026 19:14:40 +0300 Subject: [PATCH] publish: NewTimer+Stop instead of time.After in the fan-out hot path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit publishWith spawns one goroutine per subscriber per event, and each used time.After(publishWriteTimeout). time.After allocates a timer the runtime pins for the FULL 5 seconds, even when the write completes in microseconds. At the 100 events/s publish limit with S subscribers, steady state is up to 100*S*5 = 500*S live timers plus matching timer-heap entries — 50,000 pinned timers at S=100 — purely as GC and timer-heap pressure. Now time.NewTimer with defer Stop(), releasing the timer as soon as the write returns. Same correction already applied in the daemon's withRegistryDeadline, where the rationale is documented in-line for exactly this reason. Co-Authored-By: Claude Opus 5 --- service.go | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/service.go b/service.go index b16d1b1..c2bdd7e 100644 --- a/service.go +++ b/service.go @@ -408,6 +408,8 @@ func (b *broker) publishWith(evt *Event, sender *subscriber, write eventWriter) } done <- err }() + writeTimeout := time.NewTimer(publishWriteTimeout) + defer writeTimeout.Stop() select { case err := <-done: if err == nil { @@ -423,7 +425,14 @@ func (b *broker) publishWith(evt *Event, sender *subscriber, write eventWriter) dead = append(dead, s) deadMu.Unlock() } - case <-time.After(publishWriteTimeout): + // NewTimer + Stop, not time.After. This runs in a goroutine + // spawned per subscriber per event, and time.After pins its + // timer in the runtime heap for the full publishWriteTimeout + // even when the write completes immediately. At the 100 ev/s + // publish limit with S subscribers that is up to 500*S live + // timers in steady state — 50k at S=100 — purely as GC and + // timer-heap pressure. + case <-writeTimeout.C: if s.publishFailures.Add(1) >= maxConsecutivePublishFailures { slog.Debug("eventstream subscriber removed after write timeout", "remote", s.remote(),