daemon: much less CPU under send bursts

CHANGELOG.md

@@ -304,6 +304,20 @@ Detailed per-release notes are on the
   CLI 7.5 ms of CPU and 2.9 MB of allocations, more than the rest of a send.
   The second request happened on every send to an agent in the trusted list
   (`list-agents`, `pilot-mom`) and to any public peer not yet trusted.
+- **The daemon uses far less CPU under bursts of sends.** Measured on two
+  local daemons, the sending side's CPU per message fell from 4.5 ms to
+  1.6 ms at 32 concurrent sends and from 24 ms to 3.5 ms at 256. Two
+  changes:
+  - A dial no longer polls its connection every 10 ms. SYN-ACK, RST and
+    shutdown wake the dialer directly; the poll is a 250 ms backstop. With
+    hundreds of dials in flight the poll alone kept the Go scheduler busy.
+  - Concurrent `info` requests share one reply for 200 ms. Every
+    `pilotctl send-message` asks for `info`, and the reply lists every
+    connection, so a burst of N sends built N replies over a table that grew
+    with N: 73% of everything the daemon allocated during a 256-way burst.
+
+  Hosts with many cores can cut scheduler overhead further by setting
+  `GOMAXPROCS` (for example 4) in the daemon's environment.
 - **The tunnel socket asks the kernel for 4 MB buffers** in each direction
   instead of the default (about 200 KB on Linux). Every tunnel shares the one
   socket, and several streams sending at once overflowed it. The kernel caps

pkg/daemon/daemon.go

@@ -287,7 +287,7 @@ const (
 	DialMaxRetries       = 7                      // total attempts (direct + relay). 3 direct + 4 relay. With DialInitialRTO=250ms exponential-backoff capped at DialMaxRTO=8s, the relay phase is ~7.75s — covers cold-start handshake (key_exchange + flushPending + SYN/SYN-ACK round trip) for typical peers while keeping bad dials from blocking longer than the user's --timeout. The probe-and-adapt machinery (see srttHistory below) will let us shorten this for peers we've successfully dialed before.
 	DialInitialRTO       = 250 * time.Millisecond // initial SYN retransmission timeout. Lowered from 1s — modern relay RTT is <200ms; waiting a full second before assuming loss makes cold dials feel like a stall. Three direct retries with exponential backoff (250→500→1000) still cover up to 1.75s of jitter before flipping to relay; that's plenty for an unhealthy direct path while letting the common case (peer is reachable, single retry needed) feel snappy.
 	DialMaxRTO           = 8 * time.Second        // max backoff for SYN retransmission
-	DialCheckInterval    = 10 * time.Millisecond  // backstop poll for state changes during dial (conn.DialCh is the fast path)
+	DialCheckInterval    = 250 * time.Millisecond // backstop poll for state changes during dial; every transition that ends a dial (SYN-ACK, RST, shutdown) signals conn.DialCh
 	RetxCheckInterval    = 100 * time.Millisecond // retransmission check ticker
 	MaxRetxAttempts      = 8                      // abandon connection after this many retransmissions
 	HeartbeatReregThresh = 3                      // heartbeat failures before re-registration
@@ -1730,6 +1730,7 @@ func (d *Daemon) doStop() {
 		conn.Mu.Lock()
 		conn.State = StateClosed
 		conn.Mu.Unlock()
+		conn.signalDial()
 		conn.CloseRecvBuf()
 		d.ports.RemoveConnection(conn.ID)
 	}
@@ -3630,6 +3631,7 @@ func (d *Daemon) handleStreamPacket(pkt *protocol.Packet) {
 			conn.Mu.Lock()
 			conn.State = StateClosed
 			conn.Mu.Unlock()
+			conn.signalDial() // a refused dial ends now, not at the next backstop tick
 			conn.CloseRecvBuf()
 			d.ports.RemoveConnection(conn.ID)
 			d.publishEvent("conn.rst", map[string]interface{}{

pkg/daemon/ipc.go

@@ -542,8 +542,22 @@ type IPCServer struct {
 	clients    map[*ipcConn]bool
 	closeOnce  sync.Once
 	done       chan struct{}
+
+	// infoMu guards the last info reply. Every `pilotctl send-message`
+	// asks for info, and the reply serialises every peer and connection
+	// row, so a burst of N concurrent sends cost N full builds over a
+	// table that itself grows with N. Requests that arrive while a reply
+	// younger than infoReplyTTL exists share it; requests that arrive
+	// during a build wait for it instead of starting their own.
+	infoMu   sync.Mutex
+	infoAt   time.Time
+	infoData []byte
 }
 
+// infoReplyTTL is how long one info reply is reused. A var so tests that
+// assert on a value changed a moment earlier can turn reuse off.
+var infoReplyTTL = 200 * time.Millisecond
+
 func NewIPCServer(socketPath string, d *Daemon) *IPCServer {
 	return &IPCServer{
 		socketPath: socketPath,
@@ -1246,6 +1260,31 @@ type ipcInfoConn struct {
 }
 
 func (s *IPCServer) handleInfo(conn *ipcConn, reqID uint64) {
+	s.infoMu.Lock()
+	if infoReplyTTL > 0 && s.infoData != nil && time.Since(s.infoAt) < infoReplyTTL {
+		data := s.infoData
+		s.infoMu.Unlock()
+		if err := conn.writeReply(CmdInfoOK, reqID, data); err != nil {
+			slog.Debug("IPC info reply failed", "err", err)
+		}
+		return
+	}
+	data, err := s.buildInfoReply()
+	if err == nil {
+		s.infoData, s.infoAt = data, time.Now()
+	}
+	s.infoMu.Unlock()
+	if err != nil {
+		s.sendError(conn, reqID, fmt.Sprintf("info marshal: %v", err))
+		return
+	}
+	if err := conn.writeReply(CmdInfoOK, reqID, data); err != nil {
+		slog.Debug("IPC info reply failed", "err", err)
+	}
+}
+
+// buildInfoReply assembles the JSON body of an info reply.
+func (s *IPCServer) buildInfoReply() ([]byte, error) {
 	info := s.daemon.Info()
 
 	// Peer and connection rows are typed structs, not maps: `info` is called
@@ -1348,13 +1387,7 @@ func (s *IPCServer) handleInfo(conn *ipcConn, reqID uint64) {
 			data, err = marshalRowsWithinBudget(body, "conn_list", conns, ipcReplyBudget)
 		}
 	}
-	if err != nil {
-		s.sendError(conn, reqID, fmt.Sprintf("info marshal: %v", err))
-		return
-	}
-	if err := conn.writeReply(CmdInfoOK, reqID, data); err != nil {
-		slog.Debug("IPC info reply failed", "err", err)
-	}
+	return data, err
 }
 
 func (s *IPCServer) handleHealth(conn *ipcConn, reqID uint64) {

pkg/daemon/zz_burst_efficiency_test.go

@@ -0,0 +1,111 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package daemon
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/pilot-protocol/common/protocol"
+)
+
+// infoHostname runs one info request and returns the hostname in the reply.
+func infoHostname(t *testing.T, s *IPCServer) string {
+	t.Helper()
+	ic, client := newIPCTestConn(t)
+	reply := runHandler(t, client, func() { s.handleInfo(ic, 0) })
+	if reply[0] != CmdInfoOK {
+		t.Fatalf("opcode = 0x%02X, want CmdInfoOK", reply[0])
+	}
+	var info map[string]interface{}
+	if err := json.Unmarshal(reply[1:], &info); err != nil {
+		t.Fatalf("info json: %v", err)
+	}
+	h, _ := info["hostname"].(string)
+	return h
+}
+
+// A burst of info requests shares one reply for infoReplyTTL; a request
+// after that sees current state again. Not parallel: it reads the
+// package-level TTL.
+func TestInfoReplyIsSharedWithinTTLThenRebuilt(t *testing.T) {
+	client, cleanup := startFakeRegistry(t, func(req map[string]interface{}) map[string]interface{} {
+		return map[string]interface{}{"networks": []interface{}{float64(0)}}
+	})
+	defer cleanup()
+	d, s := newSimpleHandlerDaemon(t, client)
+
+	d.config.Hostname = "before"
+	if got := infoHostname(t, s); got != "before" {
+		t.Fatalf("first reply hostname = %q, want before", got)
+	}
+
+	d.config.Hostname = "after"
+	if got := infoHostname(t, s); got != "before" {
+		t.Fatalf("reply inside the TTL was rebuilt (hostname %q); concurrent senders would each rebuild it", got)
+	}
+
+	s.infoMu.Lock()
+	s.infoAt = time.Now().Add(-2 * infoReplyTTL)
+	s.infoMu.Unlock()
+	if got := infoHostname(t, s); got != "after" {
+		t.Fatalf("reply after the TTL still stale (hostname %q)", got)
+	}
+}
+
+// With reuse turned off every request is built fresh.
+func TestInfoReplyReuseCanBeTurnedOff(t *testing.T) {
+	client, cleanup := startFakeRegistry(t, func(req map[string]interface{}) map[string]interface{} {
+		return map[string]interface{}{"networks": []interface{}{float64(0)}}
+	})
+	defer cleanup()
+	d, s := newSimpleHandlerDaemon(t, client)
+
+	prev := infoReplyTTL
+	infoReplyTTL = 0
+	t.Cleanup(func() { infoReplyTTL = prev })
+
+	d.config.Hostname = "one"
+	_ = infoHostname(t, s)
+	d.config.Hostname = "two"
+	if got := infoHostname(t, s); got != "two" {
+		t.Fatalf("hostname = %q, want two with reuse off", got)
+	}
+}
+
+// A RST for a connection still in SYN_SENT must wake the dialer at once.
+// The dial loop's ticker is only a 250 ms backstop now; before, a refused
+// dial was noticed by a 10 ms poll that ran for every dialing connection.
+func TestResetWakesAWaitingDial(t *testing.T) {
+	t.Parallel()
+	d, _ := newPacketDaemon(t, nil)
+
+	remote := protocol.Addr{Network: 0, Node: 77}
+	conn := d.ports.NewConnection(41000, remote, protocol.PortDataExchange)
+	conn.Mu.Lock()
+	conn.State = StateSynSent
+	conn.Mu.Unlock()
+
+	d.handleStreamPacket(&protocol.Packet{
+		Version:  protocol.Version,
+		Protocol: protocol.ProtoStream,
+		Flags:    protocol.FlagRST,
+		Src:      remote,
+		Dst:      protocol.Addr{Network: 0, Node: 42},
+		SrcPort:  protocol.PortDataExchange,
+		DstPort:  41000,
+	})
+
+	select {
+	case <-conn.DialCh:
+	default:
+		t.Fatal("RST did not signal the dialer; it would wait for the backstop tick")
+	}
+	conn.Mu.Lock()
+	st := conn.State
+	conn.Mu.Unlock()
+	if st != StateClosed {
+		t.Fatalf("state = %v, want CLOSED", st)
+	}
+}