From fc07fbcdcffba26ac115d9783aa1a74f7613cfad Mon Sep 17 00:00:00 2001 From: Teodor Calin Date: Tue, 28 Jul 2026 16:56:34 +0300 Subject: [PATCH 1/2] streams: expose SetReadDeadline so idle teardown actually engages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Frame readers gate their slowloris/idle teardown on an optional interface: dl, canDeadline := conn.(readDeadliner) // dataexchange/service.go:227 if canDeadline && idle > 0 { _ = dl.SetReadDeadline(time.Now().Add(idle)) } streamAdapter did not implement SetReadDeadline, so canDeadline was false for every in-daemon plugin connection and dataexchange's DefaultIdleTimeout (2 minutes) was never applied. A peer that opened a stream, sent one frame and then stalled held its handler goroutine and the entire underlying Connection — a 512-slot RecvBuf, a 256-slot SendBuf and three goroutine stacks, roughly 43 KiB — until the process died. On 2026-07-28 that filled the daemon's documented connection bound (DefaultMaxTotalConnections = 65536 x ~43 KiB = ~2.8 GB per daemon) across 431 service agents on a 256 GB host, triggering system-wide kernel OOM. The agents were reaped, restart-looped, stopped heartbeating, and the registry expired every registration — the whole fleet became unresolvable. Idle daemons on other hosts were unaffected and had run 24 weeks at 17 MB, which is what identified this as per-request retention rather than a time-based leak. SetReadDeadline forwards to the underlying ConnReadWriter when that transport supports deadlines, and reports success otherwise so callers that treat a failure as fatal are not broken by transports that legitimately cannot do deadlines. Added a compile-time assertion that *streamAdapter satisfies the interface, so the capability cannot be silently dropped again. The failure mode here was invisible precisely because a failed type assertion has no diagnostic — it just quietly selects the no-timeout path. Note: this is only half the chain. The daemon's connAdapter.SetReadDeadline was itself a stub returning nil without storing anything; that is fixed in pilotprotocol separately. Both are required for the timeout to engage. Co-Authored-By: Claude Opus 5 --- streams.go | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/streams.go b/streams.go index 9ff1f46..b8fb9ba 100644 --- a/streams.go +++ b/streams.go @@ -5,6 +5,7 @@ package runtime import ( "context" "errors" + "time" "github.com/pilot-protocol/common/coreapi" "github.com/pilot-protocol/common/daemonapi" @@ -72,6 +73,27 @@ func (s *streamAdapter) Read(p []byte) (int, error) { return s.rw.Read(p) } func (s *streamAdapter) Write(p []byte) (int, error) { return s.rw.Write(p) } func (s *streamAdapter) Close() error { return s.rw.Close() } +// SetReadDeadline forwards to the underlying ConnReadWriter when it +// supports deadlines. +// +// This method existing at all is the point. Frame readers (notably +// dataexchange) gate their slowloris/idle teardown on a +// `conn.(interface{ SetReadDeadline(time.Time) error })` assertion. +// streamAdapter did not implement it, so that assertion failed for every +// in-daemon plugin connection and the configured idle timeout was never +// applied — a peer that opened a stream and stalled held its handler +// goroutine and the whole underlying connection until the process died. +// +// Reported as nil (no-op success) rather than an error when the transport +// cannot do deadlines, so callers that treat a failure as fatal are not +// broken by transports that legitimately lack the capability. +func (s *streamAdapter) SetReadDeadline(t time.Time) error { + if d, ok := s.rw.(interface{ SetReadDeadline(time.Time) error }); ok { + return d.SetReadDeadline(t) + } + return nil +} + func (s *streamAdapter) LocalAddr() coreapi.Addr { return s.conn.Info().LocalAddr } func (s *streamAdapter) LocalPort() uint16 { return s.conn.Info().LocalPort } func (s *streamAdapter) RemoteAddr() coreapi.Addr { return s.conn.Info().RemoteAddr } @@ -83,4 +105,12 @@ var ( _ coreapi.Streams = daemonStreams{} _ coreapi.Listener = (*daemonListener)(nil) _ coreapi.Stream = (*streamAdapter)(nil) + + // Frame readers (dataexchange) gate their idle/slowloris teardown on + // this exact optional interface. When streamAdapter stopped satisfying + // it, the assertion silently fell through to "no deadline" and stalled + // peers pinned a goroutine plus a full Connection forever — which is + // what OOM-killed the 431-agent service fleet on 2026-07-28. Assert it + // at compile time so the capability cannot be dropped again silently. + _ interface{ SetReadDeadline(time.Time) error } = (*streamAdapter)(nil) ) From bb16ae99931f06ace03a7d29778027e24ef02443 Mon Sep 17 00:00:00 2001 From: Teodor Calin Date: Tue, 28 Jul 2026 19:14:19 +0300 Subject: [PATCH 2/2] events: drop-on-full instead of a blocking forward that leaked goroutines MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit daemonEventBus.Subscribe wraps the daemon bus in a forwarder goroutine that did a BLOCKING send onto the channel it hands back: out <- coreapi.Event{...} The underlying bus deliberately uses non-blocking drop semantics so a slow subscriber can never stall its publishers. Forwarding with a blocking send silently converted that guarantee into a goroutine leak. If a plugin consumer stopped draining out — handler loop exited, panicked past its recover, or merely ran slower than the publisher for cap(src) events — this goroutine parked on the send permanently, retaining itself, the buffered contents of src, and every payload map they referenced. cancel() does not rescue it. cancel closes src, which does nothing for a goroutine already blocked on a send to out. Nothing else can ever unblock it, so the leak lasts the process lifetime. Every other subscriber in the tree consumes the bus channel directly rather than through an intermediate forwarder, which is why this only affects the plugin path. Dropping matches what the bus would have done once its buffer filled, so a slow consumer now loses events — the documented behaviour — instead of leaking a goroutine forever. Co-Authored-By: Claude Opus 5 --- events.go | 21 ++++++++++++++++++++- 1 file changed, 20 insertions(+), 1 deletion(-) diff --git a/events.go b/events.go index e8086a2..d0b41c7 100644 --- a/events.go +++ b/events.go @@ -32,11 +32,30 @@ func (b daemonEventBus) Subscribe(pattern string) (<-chan coreapi.Event, func()) go func() { defer close(out) for ev := range src { - out <- coreapi.Event{ + // Non-blocking, drop-on-full. The underlying bus deliberately + // drops rather than blocking its publishers; forwarding with a + // blocking send silently converted those semantics and turned + // this adapter into a goroutine leak. + // + // If a plugin consumer stopped draining `out` — its handler + // loop exited, panicked past a recover, or simply ran slower + // than the publisher for cap(src) events — this goroutine + // parked on the send forever, retaining itself, the buffered + // contents of src, and every payload map they referenced. + // cancel() does not rescue it: cancel closes src, which does + // nothing for a goroutine already blocked sending to out. + // + // Dropping matches what the bus would have done anyway once + // the buffer filled, so a slow consumer now loses events + // instead of leaking a goroutine for the process lifetime. + select { + case out <- coreapi.Event{ Topic: ev.Topic, NodeID: ev.NodeID, Time: ev.Time, Payload: ev.Payload, + }: + default: } } }()