From bb05393de7a02cfc3edfcbed18f17d62dcc87254 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Tue, 6 Oct 2026 02:33:27 +0300 Subject: [PATCH 1/2] daemon: info and trusted replies fit one IPC message One IPC message holds at most 1 MB. Past about 10,000 peers the info reply no longer fit, nor did the trusted list past about 8,000 records: the write failed, the daemon closed the client's connection, and every command that asked failed with "daemon disconnected" (pilotctl info, peers, connections, trust, and send-message, whose later requests used the closed connection). Seen on service agents with 10,909 peers and over 20,000 trust records. Send as many rows as fit, peers with an open connection and the newest trust records first, and mark the list as truncated; the counts keep the totals. pilotctl peers and trust say when a list was cut. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 12 +++ cmd/pilotctl/main.go | 26 ++++- cmd/pilotctl/zz_truncated_lists_test.go | 63 ++++++++++++ pkg/daemon/ipc.go | 74 ++++++++++++- pkg/daemon/zz_ipc_reply_budget_test.go | 131 ++++++++++++++++++++++++ 5 files changed, 298 insertions(+), 8 deletions(-) create mode 100644 cmd/pilotctl/zz_truncated_lists_test.go create mode 100644 pkg/daemon/zz_ipc_reply_budget_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 0ebeaec1..21cc3494 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -383,6 +383,18 @@ Detailed per-release notes are on the one whose hold-off ends soonest makes room. - **Bytes after a known beacon notify kind are ignored** instead of the whole notify being dropped, so a later beacon can extend the message. +- **`info` and the trusted-peers list work on nodes with very many peers.** + One IPC message holds at most 1 MB. Past about 10,000 peers the `info` + reply no longer fit, nor did the trusted list past about 8,000 records; + the daemon could not send it and closed the client's connection. Every + command that asked for it failed with `daemon disconnected`: `pilotctl + info`, `peers`, `connections` and `trust`, and `send-message`, whose + later requests went over the closed connection. Seen on service agents + with 10,909 peers and over 20,000 trust records. The daemon now sends as + many rows as fit, peers with an open connection and the newest trust + records first, and marks the list with `peer_list_truncated` or + `trusted_truncated`; the counts still give the totals. `pilotctl peers` + and `trust` say when a list was cut. - **Stream segments fit one packet.** A full stream segment was 4096 bytes, about 4.2 KB on the wire and three IP fragments on a 1500-byte path. NATs, firewalls and some virtual networks drop fragments, so on those paths diff --git a/cmd/pilotctl/main.go b/cmd/pilotctl/main.go index 153be469..121911ef 100644 --- a/cmd/pilotctl/main.go +++ b/cmd/pilotctl/main.go @@ -5871,6 +5871,9 @@ func cmdTrust(args []string) { }) total := len(trusted) + // A daemon with more trust records than one reply can carry sends + // the newest that fit. + cut, _ := result["trusted_truncated"].(bool) if jsonOutput { // Shape unchanged; "total" is the pre-limit count. The list is @@ -5880,7 +5883,11 @@ func cmdTrust(args []string) { if limitExplicit && limit > 0 && len(list) > limit { list = list[:limit] } - output(map[string]interface{}{"trusted": list, "total": total}) + out := map[string]interface{}{"trusted": list, "total": total} + if cut { + out["truncated"] = true + } + output(out) return } @@ -5906,6 +5913,9 @@ func cmdTrust(args []string) { if len(shown) < total { qualifier += fmt.Sprintf(" · showing %d newest", len(shown)) } + if cut { + qualifier += " · the daemon listed only its newest (one reply holds about 8,000)" + } fmt.Printf("%s%s\n\n", header, sDim(qualifier)) for _, t := range shown { @@ -6303,14 +6313,24 @@ func cmdPeers(args []string) { } direct := total - relayCount + // A daemon with more peers than one reply can carry lists only part. + cut, _ := info["peer_list_truncated"].(bool) if jsonOutput { - output(map[string]interface{}{ + out := map[string]interface{}{ "peers": filtered, "total": total, "encrypted": encCount, - }) + } + if cut { + out["truncated"] = true + out["daemon_peers"] = info["peers"] + } + output(out) return } + if cut { + fmt.Println(sDim(fmt.Sprintf("the daemon has %v peers and listed %d (one reply holds about 10,000; peers in use first)", info["peers"], len(peerList)))) + } if total == 0 { if search != "" { diff --git a/cmd/pilotctl/zz_truncated_lists_test.go b/cmd/pilotctl/zz_truncated_lists_test.go new file mode 100644 index 00000000..1716e27b --- /dev/null +++ b/cmd/pilotctl/zz_truncated_lists_test.go @@ -0,0 +1,63 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package main + +import ( + "encoding/json" + "strings" + "testing" +) + +// When the daemon could list only part of its peers, `peers` says so in +// both output modes instead of passing a partial list off as complete. +func TestCmdPeersReportsCutList(t *testing.T) { + const reply = `{"node_id": 1, "peers": 10909, "peer_list_truncated": true, + "peer_list": [{"node_id": 99, "encrypted": true, "authenticated": true, "relay": true}]}` + + d := newFakeDaemon(t) + d.useDaemon(t) + d.onJSON(tdCmdInfo, tdCmdInfoOK, reply) + out := captureStdout(t, func() { withText(func() { cmdPeers(nil) }) }) + if !strings.Contains(out, "the daemon has 10909 peers and listed 1") { + t.Fatalf("text output does not say the list was cut:\n%s", out) + } + + d2 := newFakeDaemon(t) + d2.useDaemon(t) + d2.onJSON(tdCmdInfo, tdCmdInfoOK, reply) + out = captureStdout(t, func() { withJSON(func() { cmdPeers(nil) }) }) + var env struct { + Data struct { + Truncated bool `json:"truncated"` + DaemonPeers float64 `json:"daemon_peers"` + } `json:"data"` + } + if err := json.Unmarshal([]byte(out), &env); err != nil { + t.Fatalf("json: %v\n%s", err, out) + } + if !env.Data.Truncated || env.Data.DaemonPeers != 10909 { + t.Fatalf("JSON output: truncated=%v daemon_peers=%v", env.Data.Truncated, env.Data.DaemonPeers) + } +} + +// The same for `trust`. +func TestCmdTrustReportsCutList(t *testing.T) { + const reply = `{"trusted_truncated": true, + "trusted": [{"node_id": 99, "public_key": "k", "approved_at": 1790000000, "mutual": true, "network": 0}]}` + + d := newFakeDaemon(t) + d.useDaemon(t) + d.onJSON(tdCmdHandshake, tdCmdHandshakeOK, reply) + out := captureStdout(t, func() { withText(func() { cmdTrust(nil) }) }) + if !strings.Contains(out, "the daemon listed only its newest") { + t.Fatalf("text output does not say the list was cut:\n%s", out) + } + + d2 := newFakeDaemon(t) + d2.useDaemon(t) + d2.onJSON(tdCmdHandshake, tdCmdHandshakeOK, reply) + out = captureStdout(t, func() { withJSON(func() { cmdTrust(nil) }) }) + if !strings.Contains(out, `"truncated":true`) { + t.Fatalf("JSON output does not say the list was cut:\n%s", out) + } +} diff --git a/pkg/daemon/ipc.go b/pkg/daemon/ipc.go index e9bb64ba..0a10a978 100644 --- a/pkg/daemon/ipc.go +++ b/pkg/daemon/ipc.go @@ -15,6 +15,7 @@ import ( "net" "os" "path/filepath" + "sort" "sync" "time" @@ -1172,6 +1173,42 @@ func (s *IPCServer) handleBroadcast(conn *ipcConn, reqID uint64, payload []byte) var daemonFeatures = []string{"reply_window", "dial_awaits_key", "key_request", "dgram_confirm"} // ipcInfoPeer is one row of the info reply's peer_list. +// ipcReplyBudget is the most JSON one IPC reply may carry. The framing +// refuses a message over ipcutil.MaxMessageSize, and a refused write closes +// the client's connection: the request failed with "daemon disconnected", +// and so did every later one on that connection. The margin covers the +// command byte. +const ipcReplyBudget = ipcutil.MaxMessageSize - 64 + +// marshalRowsWithinBudget marshals body with as many leading rows of the +// list under key as fit in budget bytes, and marks the list as cut with +// "_truncated": true. The counts elsewhere in body keep the totals. +func marshalRowsWithinBudget[T any](body map[string]interface{}, key string, rows []T, budget int) ([]byte, error) { + body[key] = []T{} + body[key+"_truncated"] = true + base, err := json.Marshal(body) + if err != nil { + return nil, err + } + size, keep := len(base), 0 + for ; keep < len(rows); keep++ { + row, err := json.Marshal(rows[keep]) + if err != nil { + return nil, err + } + add := len(row) + if keep > 0 { + add++ // separating comma + } + if size+add > budget { + break + } + size += add + } + body[key] = rows[:keep] + return json.Marshal(body) +} + type ipcInfoPeer struct { Authenticated bool `json:"authenticated"` Encrypted bool `json:"encrypted"` @@ -1257,7 +1294,7 @@ func (s *IPCServer) handleInfo(conn *ipcConn, reqID uint64) { } } - data, err := json.Marshal(map[string]interface{}{ + body := map[string]interface{}{ "node_id": info.NodeID, "address": info.Address, "endpoint": info.Endpoint, @@ -1291,7 +1328,26 @@ func (s *IPCServer) handleInfo(conn *ipcConn, reqID uint64) { "motd": info.MOTD, "features": daemonFeatures, "transport": info.Transport, - }) + } + data, err := json.Marshal(body) + if err == nil && len(data) > ipcReplyBudget { + // Past about 10,000 peers the reply no longer fits one IPC message. + // Peers in use come first so the rows a client looks for survive + // the cut; the rest in node order. If the connections alone are + // still too many, they are cut as well. + open := s.daemon.ports.ActiveNodeIDs() + sort.SliceStable(peers, func(i, j int) bool { + oi, oj := open[peers[i].NodeID], open[peers[j].NodeID] + if oi != oj { + return oi + } + return peers[i].NodeID < peers[j].NodeID + }) + data, err = marshalRowsWithinBudget(body, "peer_list", peers, ipcReplyBudget) + if err == nil && len(data) > ipcReplyBudget { + data, err = marshalRowsWithinBudget(body, "conn_list", conns, ipcReplyBudget) + } + } if err != nil { s.sendError(conn, reqID, fmt.Sprintf("info marshal: %v", err)) return @@ -2062,9 +2118,17 @@ func (s *IPCServer) handleHandshake(conn *ipcConn, reqID uint64, payload []byte) "network": t.Network, } } - data, _ := json.Marshal(map[string]interface{}{ - "trusted": list, - }) + body := map[string]interface{}{"trusted": list} + data, _ := json.Marshal(body) + if len(data) > ipcReplyBudget { + // About 8,000 trust records fill one IPC message; service + // agents hold over 20,000. Keep the newest, as pilotctl + // lists them. + sort.SliceStable(list, func(i, j int) bool { + return list[i]["approved_at"].(int64) > list[j]["approved_at"].(int64) + }) + data, _ = marshalRowsWithinBudget(body, "trusted", list, ipcReplyBudget) + } s.ipcWriteHandshakeOK(conn, reqID, data) case SubHandshakeRevoke: diff --git a/pkg/daemon/zz_ipc_reply_budget_test.go b/pkg/daemon/zz_ipc_reply_budget_test.go new file mode 100644 index 00000000..d4a9308b --- /dev/null +++ b/pkg/daemon/zz_ipc_reply_budget_test.go @@ -0,0 +1,131 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package daemon + +import ( + "encoding/base64" + "encoding/json" + "fmt" + "net" + "testing" + "time" + + "github.com/pilot-protocol/common/ipcutil" + "github.com/pilot-protocol/common/protocol" +) + +// A node with more peers than one IPC message can list still answers info: +// the scalar fields and connections are whole, the peer list is cut to fit +// and says so, and peers with an open connection are kept. Before, the +// daemon could not send the reply and closed the client's connection +// (seen on a service agent with 10,909 peers). +func TestInfoReplyFitsOneIPCMessage(t *testing.T) { + t.Parallel() + 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) + + const nPeers = 12000 + const inUse = 20000 + nPeers - 1 // last in node order: cut first unless kept for its connection + d.tunnels.mu.Lock() + for i := 0; i < nPeers; i++ { + d.tunnels.peers[uint32(20000+i)] = &net.UDPAddr{IP: net.IPv4(203, 0, byte(i/250), byte(i%250)), Port: 40000 + i%20000} + } + d.tunnels.mu.Unlock() + d.ports.NewConnection(49152, protocol.Addr{Node: inUse}, protocol.PortDataExchange).State = StateEstablished + + ic, conn := newIPCTestConn(t) + reply := runHandler(t, conn, func() { s.handleInfo(ic, 0) }) + if len(reply) > ipcutil.MaxMessageSize { + t.Fatalf("reply is %d bytes, over the IPC limit", len(reply)) + } + var info struct { + Peers int `json:"peers"` + Truncated bool `json:"peer_list_truncated"` + PeerList []struct { + NodeID uint32 `json:"node_id"` + } `json:"peer_list"` + ConnList []json.RawMessage `json:"conn_list"` + NodeID uint32 `json:"node_id"` + } + if err := json.Unmarshal(reply[1:], &info); err != nil { + t.Fatalf("info JSON: %v", err) + } + if info.Peers != nPeers || info.NodeID != 7 || len(info.ConnList) != 1 { + t.Fatalf("peers=%d node_id=%d conns=%d, want %d, 7, 1", info.Peers, info.NodeID, len(info.ConnList), nPeers) + } + if !info.Truncated || len(info.PeerList) >= nPeers || len(info.PeerList) < 9000 { + t.Fatalf("truncated=%v with %d of %d peer rows", info.Truncated, len(info.PeerList), nPeers) + } + if info.PeerList[0].NodeID != inUse { + t.Fatalf("first row is node %d; the peer with an open connection must come first", info.PeerList[0].NodeID) + } + t.Logf("%d of %d peer rows in %d bytes", len(info.PeerList), nPeers, len(reply)) +} + +// The same for the trusted-peers list: service agents hold over 20,000 +// trust records, and `pilotctl trust` failed on every one of them. +func TestTrustedReplyFitsOneIPCMessage(t *testing.T) { + t.Parallel() + d := New(Config{}) + fs := installFakeHandshake(d) + const n = 24000 + key := base64.StdEncoding.EncodeToString(make([]byte, 32)) + for i := 0; i < n; i++ { + fs.trustedRecs = append(fs.trustedRecs, HandshakeTrustRecord{ + NodeID: uint32(20000 + i), PublicKey: key, ApprovedAt: time.Unix(1790000000+int64(i), 0), Mutual: true, + }) + } + ic, conn := newIPCTestConn(t) + reply := runHandler(t, conn, func() { d.ipc.handleHandshake(ic, 0, []byte{SubHandshakeTrusted}) }) + var r struct { + Trusted []struct { + NodeID uint32 `json:"node_id"` + } `json:"trusted"` + Truncated bool `json:"trusted_truncated"` + } + if err := json.Unmarshal(reply[1:], &r); err != nil { + t.Fatalf("JSON: %v", err) + } + if !r.Truncated || len(r.Trusted) == 0 || len(r.Trusted) >= n { + t.Fatalf("truncated=%v with %d of %d records", r.Truncated, len(r.Trusted), n) + } + if newest := uint32(20000 + n - 1); r.Trusted[0].NodeID != newest { + t.Fatalf("first record is node %d, want the newest, %d", r.Trusted[0].NodeID, newest) + } + t.Logf("%d of %d trust records in %d bytes", len(r.Trusted), n, len(reply)) +} + +// The cut keeps as many rows as fit, and no more. +func TestMarshalRowsWithinBudgetKeepsAllThatFit(t *testing.T) { + t.Parallel() + rows := make([]string, 100) + for i := range rows { + rows[i] = fmt.Sprintf("row-%03d", i) + } + for _, budget := range []int{60, 100, 333, 1000} { + body := map[string]interface{}{"total": len(rows)} + data, err := marshalRowsWithinBudget(body, "list", rows, budget) + if err != nil { + t.Fatal(err) + } + var got struct { + Total int `json:"total"` + List []string `json:"list"` + Truncated bool `json:"list_truncated"` + } + if err := json.Unmarshal(data, &got); err != nil { + t.Fatalf("budget %d: %v", budget, err) + } + if len(data) > budget || !got.Truncated || got.Total != 100 { + t.Fatalf("budget %d: %d bytes, truncated=%v, total=%d", budget, len(data), got.Truncated, got.Total) + } + // One more row would not have fit. + body["list"] = rows[:len(got.List)+1] + if more, _ := json.Marshal(body); len(more) <= budget { + t.Fatalf("budget %d: kept %d rows, but %d fit", budget, len(got.List), len(got.List)+1) + } + } +} From 047d2a36aaaafefaff93726c2b02734b880d2576 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Tue, 6 Oct 2026 03:43:50 +0300 Subject: [PATCH 2/2] daemon tests: give the megabyte-reply tests time under -race runHandler reads with a 500 ms deadline; building a 12,000-row reply takes 0.6 to 1.1 s under -race, longer on CI runners. Co-Authored-By: Claude Opus 5.5 --- pkg/daemon/zz_ipc_reply_budget_test.go | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/pkg/daemon/zz_ipc_reply_budget_test.go b/pkg/daemon/zz_ipc_reply_budget_test.go index d4a9308b..1b7f4e7c 100644 --- a/pkg/daemon/zz_ipc_reply_budget_test.go +++ b/pkg/daemon/zz_ipc_reply_budget_test.go @@ -14,6 +14,14 @@ import ( "github.com/pilot-protocol/common/protocol" ) +// runLargeHandler is runHandler with time to build a megabyte reply on a +// slow CI runner under -race. +func runLargeHandler(t *testing.T, conn net.Conn, fn func()) []byte { + t.Helper() + go fn() + return readIPCReply(t, conn, 20*time.Second) +} + // A node with more peers than one IPC message can list still answers info: // the scalar fields and connections are whole, the peer list is cut to fit // and says so, and peers with an open connection are kept. Before, the @@ -37,7 +45,7 @@ func TestInfoReplyFitsOneIPCMessage(t *testing.T) { d.ports.NewConnection(49152, protocol.Addr{Node: inUse}, protocol.PortDataExchange).State = StateEstablished ic, conn := newIPCTestConn(t) - reply := runHandler(t, conn, func() { s.handleInfo(ic, 0) }) + reply := runLargeHandler(t, conn, func() { s.handleInfo(ic, 0) }) if len(reply) > ipcutil.MaxMessageSize { t.Fatalf("reply is %d bytes, over the IPC limit", len(reply)) } @@ -79,7 +87,7 @@ func TestTrustedReplyFitsOneIPCMessage(t *testing.T) { }) } ic, conn := newIPCTestConn(t) - reply := runHandler(t, conn, func() { d.ipc.handleHandshake(ic, 0, []byte{SubHandshakeTrusted}) }) + reply := runLargeHandler(t, conn, func() { d.ipc.handleHandshake(ic, 0, []byte{SubHandshakeTrusted}) }) var r struct { Trusted []struct { NodeID uint32 `json:"node_id"`