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: } } }() 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) )