daemon: info and trusted replies fit one IPC message

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

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 != "" {

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

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
+// "<key>_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:

pkg/daemon/zz_ipc_reply_budget_test.go

@@ -0,0 +1,139 @@
+// 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"
+)
+
+// 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
+// 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 := 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))
+	}
+	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 := runLargeHandler(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)
+		}
+	}
+}