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..1b7f4e7c --- /dev/null +++ b/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) + } + } +}