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= 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") + } +}