daemon: make connAdapter read deadlines real, stop leaking registry timers

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

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=

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
 	}
 }

pkg/daemon/services.go

@@ -30,20 +30,75 @@ 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 {
 		n := copy(p, a.buf)
 		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.

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