From c70593d487f86f8f6fc886196f5cd58ab2be9921 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 1 Oct 2026 17:43:55 +0300 Subject: [PATCH 1/2] Fix datagram port leak, 40ms stream stalls and several pilotctl failure paths Found by running a node in a 1 GB container under load and profiling it. Daemon - A datagram's source port was never released: only closing a connection cleared an ephemeral port, and a datagram has none. After ~16k datagrams every dial failed until restart. The port is released once the datagram is sent. - DelayedACKTimeout 40ms -> 5ms. A write held by Nagle waits for the data before it to be ACKed; a lone or odd segment's ACK waited 40ms. That was 40ms on the first exchange of every connection and one stall per 48KB file chunk (~1.5 MB/s on any link). The hold itself stays: removing it bursts a full window into the peer's socket buffer and collapses under loss. - A dial is woken when its handshake completes (conn.DialCh) instead of on the next 10ms poll; the poll remains as a backstop. - The tunnel socket requests 4 MB kernel buffers (best effort). - The info reply builds typed rows instead of a map per peer/connection; it was most of the daemon's CPU under send load. JSON is unchanged. pilotctl - send-message exits non-zero when the receiver answers "ERR ...". - appstore install removes its unpack directory, on success and failure. - appstore call waits up to 15s for a just-installed app's socket, and reports a suspended app as such. - received --clear also removes interrupted transfers in .partial. - A dial that fails for lack of local ports says so. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 42 ++++++ cmd/pilotctl/appstore.go | 49 +++++- cmd/pilotctl/appstore_catalogue.go | 8 +- cmd/pilotctl/fatal_trap.go | 8 + cmd/pilotctl/firstcontact.go | 4 + cmd/pilotctl/main.go | 33 +++- cmd/pilotctl/zz_appstore_socket_wait_test.go | 68 +++++++++ pkg/daemon/daemon.go | 94 ++++++++---- pkg/daemon/ipc.go | 110 +++++++++----- pkg/daemon/ports.go | 24 +++ pkg/daemon/udpio/socket.go | 11 ++ pkg/daemon/zz_ipc_info_rows_test.go | 45 ++++++ tests/zz_datagram_test.go | 35 +++++ tests/zz_stream_latency_test.go | 150 +++++++++++++++++++ 14 files changed, 611 insertions(+), 70 deletions(-) create mode 100644 cmd/pilotctl/zz_appstore_socket_wait_test.go create mode 100644 pkg/daemon/zz_ipc_info_rows_test.go create mode 100644 tests/zz_stream_latency_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index f20b3eb6..59fc163f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -228,6 +228,15 @@ Detailed per-release notes are on the - SIGHUP no longer reloads control state. It is still caught and ignored. ### Changed +- **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 + the request at `net.core.rmem_max` / `wmem_max`; raise those to get the full + size. +- **The daemon's `info` reply is cheaper to build.** `pilotctl send-message` + asks for it on every send, and with a few hundred connections in the table + it was three quarters of the daemon's CPU under send load. The JSON is + unchanged. - **Dependencies: skillinject v0.2.4, dataexchange v0.2.3, updater v0.2.5** (plus common v0.5.14, and the sigstore-go v1.3.0 / go-openapi versions updater v0.2.5 requires). @@ -251,6 +260,39 @@ Detailed per-release notes are on the `~/.pilot/update-state.json`. ### Fixed +- **Datagrams no longer use up the daemon's ports.** Every datagram sent + (`pilotctl dgram`, `SendTo`, broadcasts) took an ephemeral source port and + never gave it back, because only closing a connection released one. After + about 16,000 datagrams no stream could be opened — sends failed with what + looked like an unreachable peer — until the daemon restarted. The port is now + released as soon as the datagram is sent. A dial that does fail for lack of + ports says so instead of telling the operator to check the peer. +- **Large writes and first exchanges no longer stall for 40ms with both ends + idle.** A write held by Nagle waits for the data before it to be + acknowledged, and the receiver delayed a lone or odd segment's ACK by 40ms. + That cost 40ms on the first exchange of every connection, and one stall per + 48KB chunk of a file transfer — about 1.5 MB/s on any link. The delayed-ACK + timer is now 5ms: a 60 MB transfer between two containers went from 31s to + 3.5s. +- **A dial returns when its handshake completes**, not on the next 10ms poll. + A local dial took 11ms and now takes under 1ms; with the change above, + connecting and exchanging one message dropped from 55ms to 6ms. +- **`pilotctl send-message` fails when the receiver could not store the + message.** The receiver answers `ERR ...` when its disk is full or its inbox + is unwritable; the command printed that as the ack and exited 0. It now exits + non-zero, as `send-file` already did. +- **`pilotctl appstore install` cleans up after itself.** Each install unpacked + the bundle into a temporary directory and left it there — a copy of every app + ever installed, and on a small `/tmp` the reason later installs failed with + "no space left on device". The directory is removed when the install ends, + whether it succeeded or not. +- **`pilotctl appstore call` waits for an app that was just installed.** The + daemon starts an app a few seconds after `install` returns, so a call made + straight away failed with "socket not present". It now waits up to 15s for + the socket, and says so plainly when the app is suspended after crashing. +- **`pilotctl received --clear` removes interrupted transfers too.** Their + bytes stay in `received/.partial` for a resume and count against the disk, + but were never listed or cleared. The reply gains a `cleared_partial` count. - **Proxy credential hints no longer send an operator who already set `proxy_cmd` off to set it.** When the daemon re-reads its credentials with a proxy command and the proxy still rejects them (407, or Meta Muse's diff --git a/cmd/pilotctl/appstore.go b/cmd/pilotctl/appstore.go index beececeb..7e6a78fb 100644 --- a/cmd/pilotctl/appstore.go +++ b/cmd/pilotctl/appstore.go @@ -1234,6 +1234,14 @@ func cmdAppStoreInstall(args []string) { "the argument must be either a catalogue ID (`pilotctl appstore catalogue` to list) or a path to a bundle dir containing manifest.json", "%v", err) } + if source == installSourceCatalogue { + // The bundle was unpacked into a temporary directory for this + // install alone. It was never removed, so every install — failed or + // not — left a copy of the app behind in $TMPDIR. + removeUnpacked := func() { _ = os.RemoveAll(bundleDir) } + defer removeUnpacked() + fatalCleanups = append(fatalCleanups, removeUnpacked) + } if source == installSourceLocal && !allowLocal { fatalHint("invalid_argument", "local sideloads carry no catalogue signature; pass --local to confirm you trust this bundle's source. The supervisor will clamp the manifest to a small allow-list (fs.read/fs.write under $APP, audit.log). No net.dial, key.sign, ipc.call to other apps, or daemon hooks.", @@ -2663,13 +2671,14 @@ func cmdAppStoreCall(args []string) { } } - sockPath := filepath.Join(appStoreRoot(), appID, "app.sock") - if _, err := os.Stat(sockPath); err != nil { + appDir := filepath.Join(appStoreRoot(), appID) + sockPath := filepath.Join(appDir, "app.sock") + if err := waitForAppSocket(appDir, sockPath, appSocketWait); err != nil { // Not installed here at all: if the id is a rename tombstone, say so // instead of blaming the daemon. No silent retarget: the method // namespace changed too. (An installed app whose daemon is down skips // the catalogue fetch.) - if _, derr := os.Stat(filepath.Dir(sockPath)); errors.Is(derr, os.ErrNotExist) { // #nosec G703 -- only a stat of the dir whose app.sock was stat'ed just above; nothing is read or written + if _, derr := os.Stat(appDir); errors.Is(derr, os.ErrNotExist) { // #nosec G703 -- only a stat of the dir whose app.sock was stat'ed just above; nothing is read or written if c, lerr := loadCatalogue(); lerr == nil { if e := c.findEntry(appID); e != nil && e.RenamedTo != "" { fatalHint("invalid_argument", @@ -2678,6 +2687,11 @@ func cmdAppStoreCall(args []string) { } } } + if _, serr := os.Stat(filepath.Join(appDir, ".suspended")); serr == nil { + fatalHint("io_error", + "the app crashed repeatedly and was suspended: `pilotctl appstore audit "+appID+"` shows why, `pilotctl appstore restart "+appID+"` retries it", + "app %s is suspended", appID) + } fatalHint("io_error", "is the daemon running and has it supervised this app yet?", "socket %s not present: %v", sockPath, err) @@ -2787,3 +2801,32 @@ func cmdAppStoreCall(args []string) { _, _ = pretty.WriteTo(os.Stdout) fmt.Println() } + +// appSocketWait is how long `appstore call` waits for an installed app's +// socket to appear. The daemon starts an app on its next rescan after the +// install and the adapter then stages its own assets, so a call made right +// after `install` used to fail with "socket not present" for several seconds. +const appSocketWait = 15 * time.Second + +// waitForAppSocket returns once sockPath exists. It only waits for an app +// that is installed and not suspended; otherwise the socket is not coming and +// it reports that at once. +func waitForAppSocket(appDir, sockPath string, wait time.Duration) error { + deadline := time.Now().Add(wait) + for { + _, err := os.Stat(sockPath) + if err == nil { + return nil + } + if _, merr := os.Stat(filepath.Join(appDir, "manifest.json")); merr != nil { + return err // not installed + } + if _, serr := os.Stat(filepath.Join(appDir, ".suspended")); serr == nil { + return err + } + if time.Now().After(deadline) { + return err + } + time.Sleep(200 * time.Millisecond) + } +} diff --git a/cmd/pilotctl/appstore_catalogue.go b/cmd/pilotctl/appstore_catalogue.go index 131a83af..48788e8d 100644 --- a/cmd/pilotctl/appstore_catalogue.go +++ b/cmd/pilotctl/appstore_catalogue.go @@ -464,7 +464,8 @@ func resolveInstallTarget(target string) (string, installSource, error) { // fetchAndUnpackBundle downloads the catalogue entry's tarball, // verifies its sha256 against the catalogue value (defence against a // CDN substitute), and unpacks it into a tempdir whose path is -// returned for the install path to consume. +// returned for the install path to consume. The caller removes that +// directory when it is done; on error nothing is left behind. func fetchAndUnpackBundle(e catalogueEntry) (string, error) { bundleURL, bundleSHA, err := e.resolveBundle() if err != nil { @@ -518,15 +519,20 @@ func fetchAndUnpackBundle(e catalogueEntry) (string, error) { } f, err := os.Open(tmpTar.Name()) if err != nil { + _ = os.RemoveAll(unpackDir) return "", err } defer f.Close() gz, err := gzip.NewReader(f) if err != nil { + _ = os.RemoveAll(unpackDir) return "", fmt.Errorf("gzip: %w", err) } defer gz.Close() if err := untarUnder(gz, unpackDir); err != nil { + // A partial unpack is what fills a small $TMPDIR: remove it, or + // the next install starts with less room than this one had. + _ = os.RemoveAll(unpackDir) return "", fmt.Errorf("untar: %w", err) } return unpackDir, nil diff --git a/cmd/pilotctl/fatal_trap.go b/cmd/pilotctl/fatal_trap.go index ddf97aa3..f2ad84d8 100644 --- a/cmd/pilotctl/fatal_trap.go +++ b/cmd/pilotctl/fatal_trap.go @@ -27,9 +27,17 @@ func exitAfterFatal(code, msg string) { if fatalTrapDepth > 0 { panic(trappedFatal{Code: code, Message: msg}) } + for _, fn := range fatalCleanups { + fn() + } os.Exit(1) } +// fatalCleanups run before a fatal error exits the process. os.Exit skips +// deferred calls, so anything that must not outlive the command (a temporary +// directory, say) registers here as well as deferring its own removal. +var fatalCleanups []func() + // runTrappingFatal runs fn and returns the fatal error it ended with, or nil // when it returned normally. Any other panic propagates. func runTrappingFatal(fn func()) (failure *trappedFatal) { diff --git a/cmd/pilotctl/firstcontact.go b/cmd/pilotctl/firstcontact.go index fadd40ef..75f6a531 100644 --- a/cmd/pilotctl/firstcontact.go +++ b/cmd/pilotctl/firstcontact.go @@ -132,6 +132,10 @@ func dialFailureHint(target string, err error, attempts int, elapsed time.Durati s = err.Error() } switch { + case strings.Contains(s, "ephemeral ports exhausted"): + // A local condition: the peer was never contacted, so pointing at + // its reachability sends the operator the wrong way. + return "this node has no free local ports for a new connection; close idle connections (`pilotctl connections`) or restart the daemon" case strings.Contains(s, "key exchange"): return fmt.Sprintf("%s never completed the tunnel key exchange (%d dial attempt(s), %s): the service is overloaded or offline. "+ "Retry in a minute; `pilotctl peers` shows the tunnel state.", target, attempts, elapsed.Round(time.Second)) diff --git a/cmd/pilotctl/main.go b/cmd/pilotctl/main.go index 10494b74..418c9c80 100644 --- a/cmd/pilotctl/main.go +++ b/cmd/pilotctl/main.go @@ -4974,6 +4974,12 @@ func cmdSendMessage(args []string) { defer cl.Close() r := sendOne(cl, 0, false) ackAt := time.Now() + // The receiver answers "ERR ..." when it could not store the + // message (disk full, inbox unwritable). That is a failed send, not + // a delivered one — same rule send-file applies. + if ackText, _ := r["ack"].(string); strings.HasPrefix(ackText, "ERR ") { + fatalCode("internal", "receiver rejected message: %s", ackText) + } result := map[string]interface{}{ "target": target.String(), "to": target.String(), @@ -6821,10 +6827,35 @@ func cmdReceived(args []string) { count++ } } + // Interrupted transfers leave their bytes in .partial for a resume. + // They count against the disk like any received file but were never + // listed or cleared, so a failed large transfer stayed forever. + partials := 0 + partialDir := filepath.Join(dir, ".partial") + if entries, perr := os.ReadDir(partialDir); perr == nil { + for _, e := range entries { + info, ierr := e.Info() + if ierr != nil || e.IsDir() { + continue + } + if !sinceCutoff.IsZero() && info.ModTime().Before(sinceCutoff) { + continue + } + if !beforeCutoff.IsZero() && !info.ModTime().Before(beforeCutoff) { + continue + } + if os.Remove(filepath.Join(partialDir, e.Name())) == nil { + partials++ + } + } + } if jsonOutput { - outputOK(map[string]interface{}{"cleared": count, "remaining": len(all) - count}) + outputOK(map[string]interface{}{"cleared": count, "remaining": len(all) - count, "cleared_partial": partials}) } else { fmt.Printf("cleared %d received file(s), %d remaining\n", count, len(all)-count) + if partials > 0 { + fmt.Printf("cleared %d interrupted transfer(s)\n", partials) + } } return } diff --git a/cmd/pilotctl/zz_appstore_socket_wait_test.go b/cmd/pilotctl/zz_appstore_socket_wait_test.go new file mode 100644 index 00000000..3995d781 --- /dev/null +++ b/cmd/pilotctl/zz_appstore_socket_wait_test.go @@ -0,0 +1,68 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package main + +import ( + "os" + "path/filepath" + "testing" + "time" +) + +// `appstore call` right after `appstore install` used to fail: the daemon had +// not started the app yet. It now waits for the socket — but only for an app +// that is installed and not suspended, so a call that cannot succeed still +// fails at once. +func TestWaitForAppSocket(t *testing.T) { + t.Run("not installed fails at once", func(t *testing.T) { + dir := t.TempDir() + start := time.Now() + if err := waitForAppSocket(dir, filepath.Join(dir, "app.sock"), 5*time.Second); err == nil { + t.Fatal("want an error for an app with no manifest") + } + if time.Since(start) > time.Second { + t.Errorf("waited %v for an app that is not installed", time.Since(start)) + } + }) + + t.Run("suspended fails at once", func(t *testing.T) { + dir := t.TempDir() + for _, name := range []string{"manifest.json", ".suspended"} { + if err := os.WriteFile(filepath.Join(dir, name), nil, 0o600); err != nil { + t.Fatal(err) + } + } + start := time.Now() + if err := waitForAppSocket(dir, filepath.Join(dir, "app.sock"), 5*time.Second); err == nil { + t.Fatal("want an error for a suspended app") + } + if time.Since(start) > time.Second { + t.Errorf("waited %v for a suspended app", time.Since(start)) + } + }) + + t.Run("installed app is waited for", func(t *testing.T) { + dir := t.TempDir() + if err := os.WriteFile(filepath.Join(dir, "manifest.json"), nil, 0o600); err != nil { + t.Fatal(err) + } + sock := filepath.Join(dir, "app.sock") + go func() { + time.Sleep(400 * time.Millisecond) + _ = os.WriteFile(sock, nil, 0o600) + }() + if err := waitForAppSocket(dir, sock, 5*time.Second); err != nil { + t.Fatalf("socket appeared but wait failed: %v", err) + } + }) + + t.Run("gives up after the wait", func(t *testing.T) { + dir := t.TempDir() + if err := os.WriteFile(filepath.Join(dir, "manifest.json"), nil, 0o600); err != nil { + t.Fatal(err) + } + if err := waitForAppSocket(dir, filepath.Join(dir, "app.sock"), 300*time.Millisecond); err == nil { + t.Fatal("want an error when the socket never appears") + } + }) +} diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index ffd4ef79..5df46b67 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -279,7 +279,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 // poll interval for state changes during dial + DialCheckInterval = 10 * time.Millisecond // backstop poll for state changes during dial (conn.DialCh is the fast path) RetxCheckInterval = 100 * time.Millisecond // retransmission check ticker MaxRetxAttempts = 8 // abandon connection after this many retransmissions HeartbeatReregThresh = 3 // heartbeat failures before re-registration @@ -3447,6 +3447,7 @@ func (d *Daemon) handleStreamPacket(pkt *protocol.Packet) { sendSeq := conn.SendSeq recvAck := conn.RecvAck conn.Mu.Unlock() + conn.signalDial() conn.RecvMu.Lock() conn.ExpectedSeq = pkt.Seq + 1 // first data segment after SYN-ACK @@ -4012,6 +4013,40 @@ func (d *Daemon) dialConnectionLocked(ctx context.Context, dstAddr protocol.Addr check := time.NewTicker(DialCheckInterval) defer check.Stop() + // dialOutcome reports whether the dial is over, and how. It runs when + // the handshake signals conn.DialCh, and on a slow tick as a backstop. + dialOutcome := func() (*Connection, error, bool) { + conn.Mu.Lock() + st := conn.State + conn.Mu.Unlock() + if st == StateEstablished || st == StateFinWait || st == StateTimeWait { + // StateFinWait/StateTimeWait: the three-way handshake completed but + // the remote closed before we observed ESTABLISHED. The connection + // was successfully established — return it so the caller can handle + // the closed state normally. + if st == StateEstablished { + d.startRetxLoop(conn) + } + // A completed handshake proves the outbound path is alive — + // clears the rx-watchdog's partial-wedge signal. + d.lastDialOKNano.Store(time.Now().UnixNano()) + d.consecutiveDialTimeouts.Store(0) + return conn, nil, true + } + if st == StateClosed { + // Symmetric with the ctx.Done and retries>maxRetries arms: + // the connection is finished from this dial's perspective, + // so release the slot immediately. StaleConnections does + // catch StateClosed entries on its next tick, but until + // then the conn occupies an ephemeral port AND counts toward + // MaxTotalConnections — making the next dial fail under + // burst even though nothing real holds the slot. + d.ports.RemoveConnection(conn.ID) + return nil, protocol.ErrConnRefused, true + } + return nil, nil, false + } + for { select { case <-ctx.Done(): @@ -4029,34 +4064,13 @@ func (d *Daemon) dialConnectionLocked(ctx context.Context, dstAddr protocol.Addr d.tunnels.SetRelayPeer(dstAddr.Node, false) } return nil, ctx.Err() - case <-check.C: - conn.Mu.Lock() - st := conn.State - conn.Mu.Unlock() - if st == StateEstablished || st == StateFinWait || st == StateTimeWait { - // StateFinWait/StateTimeWait: the three-way handshake completed but - // the remote closed before we observed ESTABLISHED. The connection - // was successfully established — return it so the caller can handle - // the closed state normally. - if st == StateEstablished { - d.startRetxLoop(conn) - } - // A completed handshake proves the outbound path is alive — - // clears the rx-watchdog's partial-wedge signal. - d.lastDialOKNano.Store(time.Now().UnixNano()) - d.consecutiveDialTimeouts.Store(0) - return conn, nil + case <-conn.DialCh: + if c, err, done := dialOutcome(); done { + return c, err } - if st == StateClosed { - // Symmetric with the ctx.Done and retries>maxRetries arms: - // the connection is finished from this dial's perspective, - // so release the slot immediately. StaleConnections does - // catch StateClosed entries on its next tick, but until - // then the conn occupies an ephemeral port AND counts toward - // MaxTotalConnections — making the next dial fail under - // burst even though nothing real holds the slot. - d.ports.RemoveConnection(conn.ID) - return nil, protocol.ErrConnRefused + case <-check.C: + if c, err, done := dialOutcome(); done { + return c, err } case <-timer.C: if keyPending() { @@ -4160,8 +4174,20 @@ var ErrDialKeyExchange = fmt.Errorf("%w: key exchange with peer did not complete // NagleTimeout is the maximum time to buffer small writes before flushing. const NagleTimeout = 40 * time.Millisecond -// DelayedACKTimeout is the max time to delay an ACK (RFC 1122 suggests 500ms max, we use 40ms). -const DelayedACKTimeout = 40 * time.Millisecond +// DelayedACKTimeout is the max time to delay an ACK (RFC 1122 suggests 500ms max). +// +// It is also how long a write can stall with both ends idle. Nagle holds a +// short write — a frame's body after its header, the tail of a large write — +// until the data before it is ACKed, and the ACK of a lone or odd segment +// waits for this timer. At 40ms that was 40ms on the first exchange of every +// connection and one stall per 48KB file chunk (about 1.5 MB/s on any link). +// +// The timer is shortened rather than removed from the path. ACKing short +// segments at once, or sending tails without waiting, takes the pause between +// writes away entirely; several streams then burst into the peer's socket +// buffer faster than a stock kernel's can hold, and four concurrent 20MB +// transfers measured 34-40s instead of 1.4s. +const DelayedACKTimeout = 5 * time.Millisecond // DelayedACKThreshold is the number of segments to receive before sending an ACK immediately. const DelayedACKThreshold = 2 @@ -4766,6 +4792,10 @@ func (d *Daemon) SendDatagram(dstAddr protocol.Addr, dstPort uint16, data []byte if srcPort == 0 { return ErrEphemeralExhausted } + // A datagram creates no connection, so nothing else would ever return + // this port: left allocated, ~16k datagrams exhaust the ephemeral range + // and every later dial fails until the daemon restarts. + defer d.ports.ReleaseEphemeralPort(srcPort) if err := d.ensureTunnel(dstAddr.Node); err != nil { return err @@ -4806,6 +4836,10 @@ func (d *Daemon) BroadcastDatagram(netID uint16, dstPort uint16, data []byte, ad if srcPort == 0 { return ErrEphemeralExhausted } + // A datagram creates no connection, so nothing else would ever return + // this port: left allocated, ~16k datagrams exhaust the ephemeral range + // and every later dial fails until the daemon restarts. + defer d.ports.ReleaseEphemeralPort(srcPort) return d.broadcastDatagram(netID, srcPort, dstPort, data, adminToken) } diff --git a/pkg/daemon/ipc.go b/pkg/daemon/ipc.go index 9256af56..6b6bcdb2 100644 --- a/pkg/daemon/ipc.go +++ b/pkg/daemon/ipc.go @@ -1122,49 +1122,89 @@ func (s *IPCServer) handleBroadcast(conn *ipcConn, reqID uint64, payload []byte) // key request that every released daemon answers. var daemonFeatures = []string{"reply_window", "dial_awaits_key", "key_request"} +// ipcInfoPeer is one row of the info reply's peer_list. +type ipcInfoPeer struct { + Authenticated bool `json:"authenticated"` + Encrypted bool `json:"encrypted"` + Endpoint string `json:"endpoint"` + NodeID uint32 `json:"node_id"` + Relay bool `json:"relay"` +} + +// ipcInfoConn is one row of the info reply's conn_list. +type ipcInfoConn struct { + BytesRecv uint64 `json:"bytes_recv"` + BytesSent uint64 `json:"bytes_sent"` + CongWin int `json:"cong_win"` + DupACKs uint64 `json:"dup_acks"` + FastRetx uint64 `json:"fast_retx"` + ID uint32 `json:"id"` + InFlight int `json:"in_flight"` + InRecovery bool `json:"in_recovery"` + LocalPort uint16 `json:"local_port"` + OOOBuf int `json:"ooo_buf"` + PeerRecvWin int `json:"peer_recv_win"` + RecvWin int `json:"recv_win"` + RemoteAddr string `json:"remote_addr"` + RemotePort uint16 `json:"remote_port"` + Retransmits uint64 `json:"retransmits"` + RTTVARMs float64 `json:"rttvar_ms"` + SACKRecv uint64 `json:"sack_recv"` + SACKSent uint64 `json:"sack_sent"` + SegsRecv uint64 `json:"segs_recv"` + SegsSent uint64 `json:"segs_sent"` + SRTTMs float64 `json:"srtt_ms"` + SSThresh int `json:"ssthresh"` + State string `json:"state"` + Unacked int `json:"unacked"` +} + func (s *IPCServer) handleInfo(conn *ipcConn, reqID uint64) { info := s.daemon.Info() - // Build peer list for JSON - peers := make([]map[string]interface{}, len(info.PeerList)) + // Peer and connection rows are typed structs, not maps: `info` is called + // by every pilotctl send, and under load the connection table holds + // hundreds of rows, where building and key-sorting a map per row was + // most of the daemon's CPU. Fields are declared in key order so the + // JSON is byte-for-byte what the maps produced. + peers := make([]ipcInfoPeer, len(info.PeerList)) for i, p := range info.PeerList { - peers[i] = map[string]interface{}{ - "node_id": p.NodeID, - "endpoint": p.Endpoint, - "encrypted": p.Encrypted, - "authenticated": p.Authenticated, - "relay": p.Relay, + peers[i] = ipcInfoPeer{ + Authenticated: p.Authenticated, + Encrypted: p.Encrypted, + Endpoint: p.Endpoint, + NodeID: p.NodeID, + Relay: p.Relay, } } - // Build connection list for JSON - conns := make([]map[string]interface{}, len(info.ConnList)) + conns := make([]ipcInfoConn, len(info.ConnList)) for i, c := range info.ConnList { - conns[i] = map[string]interface{}{ - "id": c.ID, - "local_port": c.LocalPort, - "remote_addr": c.RemoteAddr, - "remote_port": c.RemotePort, - "state": c.State, - "cong_win": c.CongWin, - "ssthresh": c.SSThresh, - "in_flight": c.InFlight, - "srtt_ms": float64(c.SRTT.Milliseconds()), - "rttvar_ms": float64(c.RTTVAR.Milliseconds()), - "unacked": c.Unacked, - "ooo_buf": c.OOOBuf, - "peer_recv_win": c.PeerRecvWin, - "recv_win": c.RecvWin, - "in_recovery": c.InRecovery, - "bytes_sent": c.Stats.BytesSent, - "bytes_recv": c.Stats.BytesRecv, - "segs_sent": c.Stats.SegsSent, - "segs_recv": c.Stats.SegsRecv, - "retransmits": c.Stats.Retransmits, - "fast_retx": c.Stats.FastRetx, - "sack_recv": c.Stats.SACKRecv, - "sack_sent": c.Stats.SACKSent, - "dup_acks": c.Stats.DupACKs, + conns[i] = ipcInfoConn{ + BytesRecv: c.Stats.BytesRecv, + BytesSent: c.Stats.BytesSent, + CongWin: c.CongWin, + DupACKs: c.Stats.DupACKs, + FastRetx: c.Stats.FastRetx, + ID: c.ID, + InFlight: c.InFlight, + InRecovery: c.InRecovery, + LocalPort: c.LocalPort, + OOOBuf: c.OOOBuf, + PeerRecvWin: c.PeerRecvWin, + RecvWin: c.RecvWin, + RemoteAddr: c.RemoteAddr, + RemotePort: c.RemotePort, + Retransmits: c.Stats.Retransmits, + RTTVARMs: float64(c.RTTVAR.Milliseconds()), + SACKRecv: c.Stats.SACKRecv, + SACKSent: c.Stats.SACKSent, + SegsRecv: c.Stats.SegsRecv, + SegsSent: c.Stats.SegsSent, + SRTTMs: float64(c.SRTT.Milliseconds()), + SSThresh: c.SSThresh, + State: c.State, + Unacked: c.Unacked, } } diff --git a/pkg/daemon/ports.go b/pkg/daemon/ports.go index d62b046d..2d669d86 100644 --- a/pkg/daemon/ports.go +++ b/pkg/daemon/ports.go @@ -137,6 +137,15 @@ func (pm *PortManager) clearEphPortLocked(port uint16) { pm.ephBitmap[w] &^= 1 << b } +// ReleaseEphemeralPort returns a port taken with AllocEphemeralPort that no +// connection ended up owning (a datagram's source port). A port still held +// by a live connection stays allocated. +func (pm *PortManager) ReleaseEphemeralPort(port uint16) { + pm.mu.Lock() + defer pm.mu.Unlock() + pm.clearEphPortLocked(port) +} + type Listener struct { Port uint16 AcceptCh chan *Connection @@ -269,6 +278,7 @@ type Connection struct { NagleBuf []byte // pending small write data NagleMu sync.Mutex // protects NagleBuf NagleCh chan struct{} // signaled when Nagle should flush + DialCh chan struct{} // signaled when an outbound dial leaves SYN_SENT NoDelay bool // if true, disable Nagle (send immediately) // Receive window (reassembly) RecvMu sync.Mutex @@ -490,6 +500,7 @@ func (pm *PortManager) NewConnection(localPort uint16, remoteAddr protocol.Addr, SSThresh: MaxCongWin / 2, WindowCh: make(chan struct{}, 1), NagleCh: make(chan struct{}, 1), + DialCh: make(chan struct{}, 1), PeerRecvWin: -1, // sentinel: no window advertisement received yet // Initialize RetxStop here (instead of in startRetxLoop) so it is // safe to read concurrently from RemoveConnection / idleSweepLoop @@ -1339,6 +1350,19 @@ func (c *Connection) DeliverInOrder(seq uint32, data []byte) uint32 { // senders to drain (recvSenders WG), then closes the channel. This ordering // keeps `go test -race` clean — concurrent close-vs-send was previously // flagged even though the runtime made it safe via defer recover(). +// signalDial wakes a dialer waiting on this connection's handshake. The +// dialer re-reads the state itself, so a dropped signal only costs it one +// DialCheckInterval. +func (c *Connection) signalDial() { + if c.DialCh == nil { + return + } + select { + case c.DialCh <- struct{}{}: + default: + } +} + func (c *Connection) CloseRecvBuf() { c.CloseOnce.Do(func() { c.RecvMu.Lock() diff --git a/pkg/daemon/udpio/socket.go b/pkg/daemon/udpio/socket.go index 84347ffe..0eaf2dd4 100644 --- a/pkg/daemon/udpio/socket.go +++ b/pkg/daemon/udpio/socket.go @@ -100,9 +100,20 @@ func Listen(addr string) (*Socket, error) { if err != nil { return nil, fmt.Errorf("listen udp: %w", err) } + // Every tunnel shares this one socket. At the kernel default (~200KB on + // Linux) a few streams sending at once overflow it, and each dropped + // datagram is a whole segment for the stream layer to recover. Ask for + // more; the kernel silently caps the request at net.core.rmem_max / + // wmem_max, so this is best-effort and never an error. + _ = conn.SetReadBuffer(socketBufferBytes) + _ = conn.SetWriteBuffer(socketBufferBytes) return &Socket{conn: conn}, nil } +// socketBufferBytes is the kernel buffer size requested for the tunnel socket +// in each direction. +const socketBufferBytes = 4 << 20 + // WrapConn adopts an already-bound *net.UDPConn into a Socket. Used by // tests that build a UDP listener manually (so the test owns // configuration like SetReadBuffer) and want to drive the same Socket diff --git a/pkg/daemon/zz_ipc_info_rows_test.go b/pkg/daemon/zz_ipc_info_rows_test.go new file mode 100644 index 00000000..7c553f91 --- /dev/null +++ b/pkg/daemon/zz_ipc_info_rows_test.go @@ -0,0 +1,45 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package daemon + +import ( + "encoding/json" + "testing" +) + +// The info reply's rows moved from maps to structs for speed. Clients parse +// the JSON by key, so the keys and their order must not have changed. +func TestInfoRowsMarshalLikeTheMapsTheyReplaced(t *testing.T) { + conn := ipcInfoConn{ + BytesRecv: 1, BytesSent: 2, CongWin: 3, DupACKs: 4, FastRetx: 5, ID: 6, InFlight: 7, + InRecovery: true, LocalPort: 8, OOOBuf: 9, PeerRecvWin: 10, RecvWin: 11, + RemoteAddr: "0:0000.0000.0001", RemotePort: 12, Retransmits: 13, RTTVARMs: 14, + SACKRecv: 15, SACKSent: 16, SegsRecv: 17, SegsSent: 18, SRTTMs: 19, SSThresh: 20, + State: "ESTABLISHED", Unacked: 21, + } + connMap := map[string]interface{}{ + "id": uint32(6), "local_port": uint16(8), "remote_addr": "0:0000.0000.0001", "remote_port": uint16(12), + "state": "ESTABLISHED", "cong_win": 3, "ssthresh": 20, "in_flight": 7, "srtt_ms": float64(19), + "rttvar_ms": float64(14), "unacked": 21, "ooo_buf": 9, "peer_recv_win": 10, "recv_win": 11, + "in_recovery": true, "bytes_sent": uint64(2), "bytes_recv": uint64(1), "segs_sent": uint64(18), + "segs_recv": uint64(17), "retransmits": uint64(13), "fast_retx": uint64(5), "sack_recv": uint64(15), + "sack_sent": uint64(16), "dup_acks": uint64(4), + } + peer := ipcInfoPeer{Authenticated: true, Encrypted: true, Endpoint: "127.0.0.1:1", NodeID: 7, Relay: true} + peerMap := map[string]interface{}{ + "node_id": uint32(7), "endpoint": "127.0.0.1:1", "encrypted": true, "authenticated": true, "relay": true, + } + for name, pair := range map[string][2]interface{}{"conn": {conn, connMap}, "peer": {peer, peerMap}} { + got, err := json.Marshal(pair[0]) + if err != nil { + t.Fatal(err) + } + want, err := json.Marshal(pair[1]) + if err != nil { + t.Fatal(err) + } + if string(got) != string(want) { + t.Errorf("%s row changed:\n got %s\nwant %s", name, got, want) + } + } +} diff --git a/tests/zz_datagram_test.go b/tests/zz_datagram_test.go index ab0591d7..52a484e8 100644 --- a/tests/zz_datagram_test.go +++ b/tests/zz_datagram_test.go @@ -218,3 +218,38 @@ func TestMultipleDatagrams(t *testing.T) { } t.Logf("received all %d datagrams", len(msgs)) } + +// TestDatagramsDoNotExhaustEphemeralPorts sends more datagrams than there are +// ephemeral ports. Each one used to keep its source port forever, so after +// ~16k the daemon could not open any stream until it restarted. +func TestDatagramsDoNotExhaustEphemeralPorts(t *testing.T) { + requireRealNetwork(t) + t.Parallel() + env := NewTestEnv(t) + + a := env.AddDaemon() + b := env.AddDaemon() + + // Drain B so its receive queue does not back up. + go func() { + for { + if _, err := b.Driver.RecvFrom(); err != nil { + return + } + } + }() + + const sends = 17000 // the ephemeral range holds 16384 + for i := 0; i < sends; i++ { + if err := a.Driver.SendTo(b.Daemon.Addr(), 5000, []byte("d")); err != nil { + t.Fatalf("sendto %d: %v", i, err) + } + } + // SendTo is fire-and-forget over IPC; a stream dial is served after the + // queued sends and fails outright if the ports are gone. + conn, err := a.Driver.DialAddr(b.Daemon.Addr(), 7) + if err != nil { + t.Fatalf("dial after %d datagrams: %v", sends, err) + } + conn.Close() +} diff --git a/tests/zz_stream_latency_test.go b/tests/zz_stream_latency_test.go new file mode 100644 index 00000000..a4ec4794 --- /dev/null +++ b/tests/zz_stream_latency_test.go @@ -0,0 +1,150 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package tests + +import ( + "io" + "testing" + "time" +) + +// A request written as two small writes (a header, then a body) is how a +// data-exchange frame crosses a stream. Nagle holds the second write until the +// first is ACKed, and on a connection's first exchange that lone segment's ACK +// waits for the receiver's delayed-ACK timer: 40ms with both ends idle when +// the timer was 40ms, on top of a dial that polled for its handshake every +// 10ms. One message per connection is the common case (`pilotctl +// send-message`), so every message paid both. +// +// Not parallel, and judged on the fastest round: a timer in the path puts a +// floor under every round, while load only adds to some of them. +func TestFirstExchangeOnAConnectionIsNotDelayed(t *testing.T) { + requireRealNetwork(t) + env := NewTestEnv(t) + a := env.AddDaemon() + b := env.AddDaemon() + + const port = 5101 + ln, err := a.Driver.Listen(port) + if err != nil { + t.Fatalf("listen: %v", err) + } + defer ln.Close() + + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + go func() { + defer conn.Close() + req := make([]byte, 8+200) + if _, err := io.ReadFull(conn, req); err != nil { + return + } + _, _ = conn.Write(req[:28]) + }() + } + }() + + // Stay under the receiver's per-source limit of 10 new connections a + // second, which would otherwise be what this measures. + const rounds = 8 + header, body, reply := make([]byte, 8), make([]byte, 200), make([]byte, 28) + fastest := time.Hour + for i := 0; i < rounds; i++ { + start := time.Now() + conn, err := b.Driver.DialAddr(a.Daemon.Addr(), port) + if err != nil { + t.Fatalf("dial %d: %v", i, err) + } + if _, err := conn.Write(header); err != nil { + t.Fatalf("write header: %v", err) + } + if _, err := conn.Write(body); err != nil { + t.Fatalf("write body: %v", err) + } + if _, err := io.ReadFull(conn, reply); err != nil { + t.Fatalf("read reply: %v", err) + } + if d := time.Since(start); d < fastest { + fastest = d + } + conn.Close() + } + t.Logf("fastest of %d dial + write-write-read exchanges: %v", rounds, fastest) + // The 10ms dial poll plus the 40ms delayed ACK kept every round above + // 50ms. + if fastest > 25*time.Millisecond { + t.Errorf("fastest exchange took %v, want under 25ms", fastest) + } +} + +// A write larger than one segment ends in a short tail. Nagle holds that tail +// until everything before it is ACKed, and when the full segments are an odd +// number the last one's ACK waits for the receiver's delayed-ACK timer. At +// 40ms that was one stall per 48KB file chunk, capping any transfer near +// 1.5 MB/s. +func TestLargeWritesDoNotStallOnDelayedACK(t *testing.T) { + requireRealNetwork(t) + env := NewTestEnv(t) + a := env.AddDaemon() + b := env.AddDaemon() + + const port = 5102 + const chunk = 48*1024 + 20 // a data-exchange file chunk plus its framing + const chunks = 10 + ln, err := a.Driver.Listen(port) + if err != nil { + t.Fatalf("listen: %v", err) + } + defer ln.Close() + + received := make(chan error, 1) + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + _, err = io.CopyN(io.Discard, conn, chunk*chunks) + conn.Close() + received <- err + } + }() + + // Judged on the fastest of three runs: the timer puts a floor under + // every run (one 40ms stall per chunk is 400ms here), load only adds. + fastest := time.Hour + buf := make([]byte, chunk) + for run := 0; run < 3; run++ { + conn, err := b.Driver.DialAddr(a.Daemon.Addr(), port) + if err != nil { + t.Fatalf("dial: %v", err) + } + start := time.Now() + for i := 0; i < chunks; i++ { + if _, err := conn.Write(buf); err != nil { + t.Fatalf("write chunk %d: %v", i, err) + } + } + select { + case err := <-received: + if err != nil { + t.Fatalf("receive: %v", err) + } + case <-time.After(60 * time.Second): + t.Fatal("timeout receiving chunks") + } + if d := time.Since(start); d < fastest { + fastest = d + } + conn.Close() + } + rate := float64(chunk*chunks) / fastest.Seconds() / (1 << 20) + t.Logf("fastest run: %d chunks of %d bytes in %v (%.1f MB/s)", chunks, chunk, fastest, rate) + if fastest > 200*time.Millisecond { + t.Errorf("%d chunks took %v at best, want under 200ms: each large write is stalling on a delayed ACK", chunks, fastest) + } +} From 2b394936f8146480940e5ed921f8026dac29ef2a Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 1 Oct 2026 18:34:10 +0300 Subject: [PATCH 2/2] appstore call: say where a suspended app's error output is `appstore audit` lists the supervisor's events (exit codes), not the reason; the app's stderr goes to the daemon's log. Co-Authored-By: Claude Opus 5.5 --- cmd/pilotctl/appstore.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cmd/pilotctl/appstore.go b/cmd/pilotctl/appstore.go index 7e6a78fb..fe3b4c55 100644 --- a/cmd/pilotctl/appstore.go +++ b/cmd/pilotctl/appstore.go @@ -2689,7 +2689,7 @@ func cmdAppStoreCall(args []string) { } if _, serr := os.Stat(filepath.Join(appDir, ".suspended")); serr == nil { fatalHint("io_error", - "the app crashed repeatedly and was suspended: `pilotctl appstore audit "+appID+"` shows why, `pilotctl appstore restart "+appID+"` retries it", + "the app crashed repeatedly and was suspended: `pilotctl appstore audit "+appID+"` shows the exits, the app's own error output is in the daemon's log; `pilotctl appstore restart "+appID+"` retries it", "app %s is suspended", appID) } fatalHint("io_error",