From 11d8bd16d8b76c0a6ca3d597286017dc87bd0049 Mon Sep 17 00:00:00 2001 From: Teodor Calin Date: Tue, 28 Jul 2026 16:57:14 +0300 Subject: [PATCH 1/2] daemon: make connAdapter read deadlines real, stop leaking registry timers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit connAdapter's deadline setters were stubs: func (a *connAdapter) SetReadDeadline(t time.Time) error { return nil } They stored nothing and reported success. Anything that set a read deadline got no deadline and no error — a silent lie, which is worse than not implementing the method, because callers cannot detect the failure. Read consequently blocked on <-a.conn.RecvBuf forever. A peer that opened a stream and then went silent pinned its handler goroutine plus the whole Connection: a 512-slot RecvBuf, a 256-slot SendBuf and three goroutine stacks, roughly 43 KiB each. On 2026-07-28 that filled the daemon's own documented bound — DefaultMaxTotalConnections (65536) x ~43 KiB = ~2.8 GB — across 431 service agents on a 256 GB host, causing system-wide kernel OOM. The agents were reaped, restart-looped, stopped heartbeating, and the registry expired every registration, so the whole fleet went unresolvable. The daemon was not leaking past a bound; it was filling its bound and sitting there, which is why growth was ~100x and then plateaued. Idle daemons elsewhere were fine at 17 MB after 24 weeks, identifying this as per-request retention. Changes: - SetReadDeadline now stores an absolute deadline (atomic; it may be set from a goroutine other than the one parked in Read), and Read selects on RecvBuf against a timer, returning a net.Error with Timeout() == true. - SetDeadline delegates to SetReadDeadline instead of silently doing nothing. SetWriteDeadline stays a no-op but is now documented as such — Write already bounds itself via connAdapterWriteDeadline. - withRegistryDeadline no longer leaks a timer per call. time.After pins its timer for the full 8s duration, and this path runs repeatedly when the registry conn is half-open. Now an explicitly stopped timer. The result channel was already buffered, so the worker goroutine can always deposit and exit rather than parking on the send. Regression coverage in zz_connadapter_deadline_test.go: the deadline fires and reports Timeout(); a cleared deadline restores block-forever semantics; an armed deadline does not break normal reads; an already-expired deadline returns immediately. The first of these hangs for the full timeout and then fails against the old stub. NOTE: this is half the chain. Frame readers reach connAdapter through runtime's streamAdapter, which did not expose SetReadDeadline at all, so the type assertion that gates the idle teardown failed regardless of what the daemon implemented. Fixed in pilot-protocol/runtime#31. web4 pins runtime v0.3.1, so that must be released and bumped here before the timeout engages in production. Co-Authored-By: Claude Opus 5 --- pkg/daemon/daemon.go | 12 +- pkg/daemon/services.go | 67 ++++++++++- pkg/daemon/zz_connadapter_deadline_test.go | 124 +++++++++++++++++++++ 3 files changed, 199 insertions(+), 4 deletions(-) create mode 100644 pkg/daemon/zz_connadapter_deadline_test.go diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index 8231e53a..bbc1b87b 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -78,15 +78,25 @@ func withRegistryDeadline(timeout time.Duration, fn func() (map[string]interface resp map[string]interface{} err error } + // Buffered so the worker can always deposit its result and exit even + // after we have given up — otherwise it would park on the send + // forever, retaining its stack and the decoded response map. resultCh := make(chan registryCallResult, 1) go func() { resp, err := fn() resultCh <- registryCallResult{resp: resp, err: err} }() + + // time.After leaks its timer until the duration elapses. On a wedged + // registry conn this path runs repeatedly, so use an explicitly + // stopped timer instead of pinning one 8s timer per call. + timer := time.NewTimer(timeout) + defer timer.Stop() + select { case r := <-resultCh: return r.resp, r.err - case <-time.After(timeout): + case <-timer.C: return nil, errRegistryCallTimedOut } } diff --git a/pkg/daemon/services.go b/pkg/daemon/services.go index 1d40b0cf..c6942d52 100644 --- a/pkg/daemon/services.go +++ b/pkg/daemon/services.go @@ -30,12 +30,50 @@ type connAdapter struct { // to 0 on the first successful WriteEvent. Atomic because publish() // runs concurrently across publishers without holding a write lock. publishFailures atomic.Uint32 + + // readDeadline is the absolute time at which a blocked Read gives up, + // as UnixNano; 0 means "no deadline" (block forever, the historical + // behaviour). Stored atomically because SetReadDeadline may be called + // from a different goroutine than the one parked in Read. + // + // Without this, Read blocks on <-RecvBuf with no way out. That made + // dataexchange's slowloris guard dead code in-daemon: it gates its + // idle teardown on a `conn.(readDeadliner)` type assertion, which no + // in-daemon stream satisfied, so DefaultIdleTimeout was never applied + // and a peer that opened a connection and then went silent pinned the + // handler goroutine plus the whole Connection (~43 KiB of buffers and + // stacks) forever. See TestConnAdapterReadDeadline. + readDeadline atomic.Int64 } func newConnAdapter(d *Daemon, conn *Connection) *connAdapter { return &connAdapter{conn: conn, daemon: d} } +// SetReadDeadline implements the optional deadline surface that +// dataexchange (and any other frame reader) probes for. A zero time +// clears the deadline. Satisfying this interface is what re-arms the +// idle timeout for in-daemon plugin connections. +func (a *connAdapter) SetReadDeadline(t time.Time) error { + if t.IsZero() { + a.readDeadline.Store(0) + return nil + } + a.readDeadline.Store(t.UnixNano()) + return nil +} + +// errReadDeadlineExceeded is returned by Read when the deadline set via +// SetReadDeadline elapses. It reports Timeout() == true so callers that +// distinguish timeouts from hard failures (net.Error) behave correctly. +type errReadDeadlineExceeded struct{} + +func (errReadDeadlineExceeded) Error() string { return "pilot: read deadline exceeded" } +func (errReadDeadlineExceeded) Timeout() bool { return true } +func (errReadDeadlineExceeded) Temporary() bool { return true } + +var _ net.Error = errReadDeadlineExceeded{} + func (a *connAdapter) Read(p []byte) (int, error) { // Drain leftover buffer first if len(a.buf) > 0 { @@ -43,7 +81,24 @@ func (a *connAdapter) Read(p []byte) (int, error) { a.buf = a.buf[n:] return n, nil } - data, ok := <-a.conn.RecvBuf + + var data []byte + var ok bool + if dl := a.readDeadline.Load(); dl != 0 { + d := time.Until(time.Unix(0, dl)) + if d <= 0 { + return 0, errReadDeadlineExceeded{} + } + timer := time.NewTimer(d) + select { + case data, ok = <-a.conn.RecvBuf: + timer.Stop() + case <-timer.C: + return 0, errReadDeadlineExceeded{} + } + } else { + data, ok = <-a.conn.RecvBuf + } if !ok { return 0, io.EOF } @@ -122,8 +177,14 @@ func (p pilotAddr) String() string { return fmt.Sprintf("%s:%d", p.addr.String(), p.port) } -func (a *connAdapter) SetDeadline(t time.Time) error { return nil } -func (a *connAdapter) SetReadDeadline(t time.Time) error { return nil } +// SetDeadline sets the read deadline. The write side has its own bound +// (connAdapterWriteDeadline), so only the read half is honoured here. +func (a *connAdapter) SetDeadline(t time.Time) error { return a.SetReadDeadline(t) } + +// SetWriteDeadline is not implemented: Write already fails after +// connAdapterWriteDeadline against a persistently-full send buffer. +// It returns nil (rather than an error) to preserve net.Conn semantics +// for callers that set both deadlines unconditionally. func (a *connAdapter) SetWriteDeadline(t time.Time) error { return nil } // startBuiltinServices starts all enabled built-in port services. diff --git a/pkg/daemon/zz_connadapter_deadline_test.go b/pkg/daemon/zz_connadapter_deadline_test.go new file mode 100644 index 00000000..3f434314 --- /dev/null +++ b/pkg/daemon/zz_connadapter_deadline_test.go @@ -0,0 +1,124 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package daemon + +import ( + "errors" + "net" + "testing" + "time" +) + +// newTestConnAdapter builds a connAdapter over a bare Connection with a +// RecvBuf nobody ever writes to — the exact shape of a peer that opens a +// stream, sends nothing, and stalls. +func newTestConnAdapter() *connAdapter { + c := &Connection{ + RecvBuf: make(chan []byte, RecvBufSize), + } + return &connAdapter{conn: c} +} + +// TestConnAdapterReadDeadlineFires is the regression guard for the leak +// that OOM-killed the 431-agent service fleet on 2026-07-28. +// +// connAdapter.SetReadDeadline used to be a stub returning nil without +// storing anything, so Read blocked on <-RecvBuf forever. A stalled peer +// pinned its handler goroutine plus the whole Connection (512-slot +// RecvBuf + 256-slot SendBuf + goroutine stacks, ~43 KiB) until the +// process died. Against that stub this test hangs until the deadline +// arrives and then fails on the nil error. +func TestConnAdapterReadDeadlineFires(t *testing.T) { + a := newTestConnAdapter() + + if err := a.SetReadDeadline(time.Now().Add(100 * time.Millisecond)); err != nil { + t.Fatalf("SetReadDeadline: %v", err) + } + + start := time.Now() + done := make(chan error, 1) + go func() { + buf := make([]byte, 64) + _, err := a.Read(buf) + done <- err + }() + + select { + case err := <-done: + if err == nil { + t.Fatal("Read returned nil error — deadline did not fire (stub regression)") + } + var ne net.Error + if !errors.As(err, &ne) || !ne.Timeout() { + t.Fatalf("expected a net.Error with Timeout()==true, got %T: %v", err, err) + } + if elapsed := time.Since(start); elapsed > 2*time.Second { + t.Fatalf("Read blocked %v — far past the 100ms deadline", elapsed) + } + case <-time.After(3 * time.Second): + t.Fatal("Read never returned — deadline is not enforced; this is the fleet-killing leak") + } +} + +// TestConnAdapterZeroDeadlineBlocks pins the documented default: with no +// deadline set, Read keeps its historical block-forever behaviour, and a +// zero time clears a previously-set deadline. +func TestConnAdapterZeroDeadlineBlocks(t *testing.T) { + a := newTestConnAdapter() + + if err := a.SetReadDeadline(time.Now().Add(50 * time.Millisecond)); err != nil { + t.Fatalf("SetReadDeadline: %v", err) + } + if err := a.SetReadDeadline(time.Time{}); err != nil { + t.Fatalf("clearing SetReadDeadline: %v", err) + } + + done := make(chan error, 1) + go func() { + buf := make([]byte, 64) + _, err := a.Read(buf) + done <- err + }() + + select { + case err := <-done: + t.Fatalf("Read returned %v — a cleared deadline must not time out", err) + case <-time.After(300 * time.Millisecond): + // Correct: still blocked with no deadline armed. + } +} + +// TestConnAdapterReadDeliversBeforeDeadline guards the other direction — +// arming a deadline must not break normal reads. +func TestConnAdapterReadDeliversBeforeDeadline(t *testing.T) { + a := newTestConnAdapter() + a.conn.RecvBuf <- []byte("hello") + + if err := a.SetReadDeadline(time.Now().Add(2 * time.Second)); err != nil { + t.Fatalf("SetReadDeadline: %v", err) + } + + buf := make([]byte, 64) + n, err := a.Read(buf) + if err != nil { + t.Fatalf("Read: %v", err) + } + if got := string(buf[:n]); got != "hello" { + t.Fatalf("got %q, want %q", got, "hello") + } +} + +// TestConnAdapterAlreadyExpiredDeadline covers the boundary where the +// deadline is already in the past when Read is entered. +func TestConnAdapterAlreadyExpiredDeadline(t *testing.T) { + a := newTestConnAdapter() + + if err := a.SetReadDeadline(time.Now().Add(-time.Second)); err != nil { + t.Fatalf("SetReadDeadline: %v", err) + } + + buf := make([]byte, 64) + if _, err := a.Read(buf); err == nil { + t.Fatal("Read with an already-expired deadline must not block or succeed") + } +} From f63f0767d29144d058d6d5db00b889e56bcf7a3d Mon Sep 17 00:00:00 2001 From: Teodor Calin Date: Wed, 29 Jul 2026 14:49:03 +0300 Subject: [PATCH 2/2] =?UTF-8?q?deps:=20bump=20runtime=20to=20v0.3.2=20?= =?UTF-8?q?=E2=80=94=20closes=20the=20idle-timeout=20chain?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Without this the connAdapter deadline work in this PR is a no-op in production. Frame readers reach connAdapter through runtime's streamAdapter, and streamAdapter only gained SetReadDeadline in runtime v0.3.2 (#31). Pinned at v0.3.1 the type assertion that gates dataexchange's idle teardown still fails, so the timeout never arms no matter what the daemon implements. Both halves are now in place: - runtime v0.3.2: streamAdapter exposes SetReadDeadline (+ a compile-time assertion so the capability cannot be dropped silently again) - this repo: connAdapter.SetReadDeadline stores a real deadline instead of being a stub that returned nil Verified with GOWORK=off so the resolution matches CI rather than the local go.work overlay, which shadows the module with the checkout and would have hidden a missing go.sum entry: build clean, vet clean, pkg/daemon suite green. Co-Authored-By: Claude Opus 5 --- go.mod | 2 +- go.sum | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/go.mod b/go.mod index 269eb891..647f9069 100644 --- a/go.mod +++ b/go.mod @@ -13,7 +13,7 @@ require ( github.com/pilot-protocol/nameserver v0.2.1 github.com/pilot-protocol/policy v0.2.3 github.com/pilot-protocol/rendezvous v0.2.7 - github.com/pilot-protocol/runtime v0.3.1 + github.com/pilot-protocol/runtime v0.3.2 github.com/pilot-protocol/skillinject v0.2.4-0.20260717204101-902f745138da github.com/pilot-protocol/trustedagents v0.2.5 github.com/pilot-protocol/updater v0.2.4 diff --git a/go.sum b/go.sum index 735d5760..42518fb4 100644 --- a/go.sum +++ b/go.sum @@ -255,6 +255,8 @@ github.com/pilot-protocol/rendezvous v0.2.7 h1:676cSBYGpkkg5FK6AioGLarixPrTwlky5 github.com/pilot-protocol/rendezvous v0.2.7/go.mod h1:sv0TuqAosOCe3f3EIuZ9lmrNaHJTqyy5k/UGFqHskB4= github.com/pilot-protocol/runtime v0.3.1 h1:+W9ww0dZY/FgOBtCmIOV3w5L5Z4Upt/RIsrYElXZ1zs= github.com/pilot-protocol/runtime v0.3.1/go.mod h1:GfFEIji0w7H9SSNR9Wl2q72pd2OYN3PHY9Qhcbvyrqk= +github.com/pilot-protocol/runtime v0.3.2 h1:21lgUfYNvpls0Vd3V2v9lfs13G7mT8pcT3D2QzcMueE= +github.com/pilot-protocol/runtime v0.3.2/go.mod h1:CXEmjKF/HozhIxn9QZxO13Lxdnkok3XKEkYS/jFjQKs= github.com/pilot-protocol/skillinject v0.2.4-0.20260717204101-902f745138da h1:supBO6UzykRn5Ta5pCLXjTvAPqTKOgfr2ZpZQugBNuI= github.com/pilot-protocol/skillinject v0.2.4-0.20260717204101-902f745138da/go.mod h1:+jEW+uCFkA6TnmZc41Ev5oczXCbOHD0NqVWG8RzjwXQ= github.com/pilot-protocol/trustedagents v0.2.5 h1:zdeezxalidXanOkEcdsageyAp6jVbfhAnPumREbEZt0=