From 76ad3b963e12a91c1b8968df8cf886daa8bda3c8 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 1 Oct 2026 21:26:33 +0300 Subject: [PATCH 1/4] pilotctl: send-message reads its payload from stdin or a file --data took the body only as a command-line argument, which the OS caps: a 1 MB payload fails with "Argument list too long" and Linux stops a single argument at 128 KiB. Add "--data -" (stdin) and --data-file . Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 5 ++ cmd/pilotctl/main.go | 44 ++++++++++++++-- cmd/pilotctl/zz_send_message_payload_test.go | 53 ++++++++++++++++++++ 3 files changed, 98 insertions(+), 4 deletions(-) create mode 100644 cmd/pilotctl/zz_send_message_payload_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index fd723e61..2fd070f2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,11 @@ Detailed per-release notes are on the ## [Unreleased] ### Added +- **`pilotctl send-message` can take its payload from stdin or a file**: + `--data -` and `--data-file `. A payload passed as an argument is + capped by the OS — a 1 MB `--data` failed with "Argument list too long", + and on Linux a single argument stops at 128 KiB — which is why senders of + large bodies needed a separate stdin helper. - **A client can ask the daemon whether a datagram was actually sent.** The IPC `SendTo` command is fire-and-forget: when the daemon could not send a datagram (no route to the node, port policy, ephemeral ports exhausted) it diff --git a/cmd/pilotctl/main.go b/cmd/pilotctl/main.go index 418c9c80..e3415bff 100644 --- a/cmd/pilotctl/main.go +++ b/cmd/pilotctl/main.go @@ -895,7 +895,9 @@ var commandHelp = map[string]string{ Send a message to a remote agent and optionally wait for the reply. Flags: - --data message payload (required) + --data message payload; "-" reads it from stdin + --data-file read the payload from a file (for payloads too large + for a command-line argument) --type text|json|binary payload encoding (default: text) --count send N times (default: 1) --reuse-conn reuse the connection across --count sends (saves ~1 RTT) @@ -912,6 +914,8 @@ Examples: pilotctl send-message list-agents --data '/data {"search":"weather","limit":5}' pilotctl send-message my-peer --data "hello" --wait pilotctl send-message 0:0000.0000.400E --data "ping" --trace + pilotctl send-message my-peer --data-file report.json --type json + generate-report | pilotctl send-message my-peer --data - `, "ping": `Usage: pilotctl ping [flags] @@ -4744,6 +4748,38 @@ func streamSendFile(d *driver.Driver, target protocol.Addr, filePath, filename s return result, nil } +// messagePayload returns the body for send-message: --data, or the whole of +// stdin for "--data -", or the contents of --data-file. A payload passed as an +// argument is capped by the OS (ARG_MAX; 128 KiB per argument on Linux), which +// is why callers with large bodies needed a separate stdin helper. +func messagePayload(flags map[string]string, stdin io.Reader) (string, error) { + data, hasData := flags["data"] + file, hasFile := flags["data-file"] + switch { + case hasData && hasFile: + return "", fmt.Errorf("give --data or --data-file, not both") + case hasFile: + if file == "" || file == "true" { + return "", fmt.Errorf("--data-file needs a path") + } + b, err := os.ReadFile(file) // #nosec G304 -- the operator names the file to send + if err != nil { + return "", fmt.Errorf("read --data-file: %v", err) + } + data = string(b) + case data == "-": + b, err := io.ReadAll(stdin) + if err != nil { + return "", fmt.Errorf("read payload from stdin: %v", err) + } + data = string(b) + } + if data == "" { + return "", fmt.Errorf("--data is required") + } + return data, nil +} + func cmdSendMessage(args []string) { flags, pos := parseFlags(args) for name := range flags { @@ -4823,9 +4859,9 @@ func cmdSendMessage(args []string) { fatalCode("not_found", "%v", err) } - data := flagString(flags, "data", "") - if data == "" { - fatalCode("invalid_argument", "--data is required") + data, err := messagePayload(flags, os.Stdin) + if err != nil { + fatalCode("invalid_argument", "%v", err) } msgType := flagString(flags, "type", "text") diff --git a/cmd/pilotctl/zz_send_message_payload_test.go b/cmd/pilotctl/zz_send_message_payload_test.go new file mode 100644 index 00000000..82e17ff1 --- /dev/null +++ b/cmd/pilotctl/zz_send_message_payload_test.go @@ -0,0 +1,53 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package main + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +// send-message took its body only as a command-line argument, which the OS +// caps (a 1 MB --data fails with "Argument list too long"). It can now come +// from stdin or a file. +func TestMessagePayload(t *testing.T) { + big := strings.Repeat("x", 2<<20) + file := filepath.Join(t.TempDir(), "body") + if err := os.WriteFile(file, []byte(big), 0o600); err != nil { + t.Fatal(err) + } + + for _, tc := range []struct { + name string + flags map[string]string + stdin string + want string + wantErr string + }{ + {name: "argument", flags: map[string]string{"data": "hello"}, want: "hello"}, + {name: "stdin", flags: map[string]string{"data": "-"}, stdin: big, want: big}, + {name: "file", flags: map[string]string{"data-file": file}, want: big}, + {name: "neither", flags: map[string]string{}, wantErr: "--data is required"}, + {name: "empty stdin", flags: map[string]string{"data": "-"}, wantErr: "--data is required"}, + {name: "both", flags: map[string]string{"data": "a", "data-file": file}, wantErr: "not both"}, + {name: "file flag without a path", flags: map[string]string{"data-file": "true"}, wantErr: "needs a path"}, + {name: "missing file", flags: map[string]string{"data-file": filepath.Join(t.TempDir(), "nope")}, wantErr: "read --data-file"}, + } { + got, err := messagePayload(tc.flags, strings.NewReader(tc.stdin)) + if tc.wantErr != "" { + if err == nil || !strings.Contains(err.Error(), tc.wantErr) { + t.Errorf("%s: err = %v, want one containing %q", tc.name, err, tc.wantErr) + } + continue + } + if err != nil { + t.Errorf("%s: %v", tc.name, err) + continue + } + if got != tc.want { + t.Errorf("%s: got %d bytes, want %d", tc.name, len(got), len(tc.want)) + } + } +} From 539c5400c51bb4d3c956c3e42c8bb0257a789ef7 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 1 Oct 2026 21:53:58 +0300 Subject: [PATCH 2/4] Deliver messages over 256 KB; accept --data-file; fail an unacknowledged send Confirming --data-file against real nodes showed three problems: - send-message rejected --data-file: its flag allow-list was not updated. - The daemon refused any single stream write larger than MaxNagleBuf (256 KB) with ErrSendBufFull. An IPC send has no reply, so the refusal went nowhere: a 1 MB message was dropped while pilotctl printed status ok. SendData now feeds a large write through the buffer in whole-segment pieces, blocking on the window like any other write. The cap on the buffer is unchanged; the unit test that pinned the refusal is split so the cap is still asserted, now during a blocked oversized write. - send-message exited 0 when no ACK came back. It now fails. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 9 ++++ cmd/pilotctl/main.go | 13 ++++- pkg/daemon/daemon.go | 24 +++++++++ pkg/daemon/zz_nagle_buf_bug_test.go | 84 +++++++++++++++++++++-------- tests/zz_large_write_test.go | 71 ++++++++++++++++++++++++ 5 files changed, 179 insertions(+), 22 deletions(-) create mode 100644 tests/zz_large_write_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 2fd070f2..8f79b8e9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -361,6 +361,15 @@ Detailed per-release notes are on the with 2 of 5 failing before, 10.3–11.8s with none failing after; at 0.5% loss 9.4–10.5s before, 7.3–7.6s after; without loss 8.2–8.4s before, 7.6–7.7s after. +- **A message larger than 256 KB is delivered instead of silently dropped.** + The daemon refused any single stream write bigger than its send buffer, and + an IPC send has no reply, so the client never knew: `send-message` printed + `"status":"ok"` for a 1 MB message that never left the node. A large write is + now fed through the buffer in pieces, blocking on the window like any other + write. The buffer's size cap is unchanged. +- **`pilotctl send-message` fails when the receiver does not acknowledge the + message.** Every receiver answers a stored message with an ACK; with none, + the command used to exit 0. - **`-advertise-endpoint` survives a re-registration.** When the daemon re-registered (registry reconnect, transport watchdog recovery) it sent the tunnel socket's local address instead of the advertised endpoint, so diff --git a/cmd/pilotctl/main.go b/cmd/pilotctl/main.go index e3415bff..6f82500d 100644 --- a/cmd/pilotctl/main.go +++ b/cmd/pilotctl/main.go @@ -4784,7 +4784,7 @@ func cmdSendMessage(args []string) { flags, pos := parseFlags(args) for name := range flags { switch name { - case "data", "type", "count", "reuse-conn", "wait", "timeout", "no-resend", "trace", "no-auto-handshake": + case "data", "data-file", "type", "count", "reuse-conn", "wait", "timeout", "no-resend", "trace", "no-auto-handshake": default: fatalCode("invalid_argument", "send-message: unknown flag --%s", name) } @@ -5010,6 +5010,17 @@ func cmdSendMessage(args []string) { defer cl.Close() r := sendOne(cl, 0, false) ackAt := time.Now() + // Every receiver answers a stored message with an ACK frame. No ACK + // means the message was not stored, or was never sent — the daemon + // drops a write it cannot buffer without telling the client — so + // "ok" here was reporting messages that never arrived. + if _, failed := r["error"]; !failed { + if _, acked := r["ack"]; !acked { + fatalHint("connection_failed", + "the receiver did not confirm it stored the message; check `pilotctl peers` and the daemon log, then send again", + "%s did not acknowledge the message (%d bytes)", target, len(data)) + } + } // 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. diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index 0128113e..0fccf234 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -4299,6 +4299,30 @@ func (d *Daemon) SendData(conn *Connection, data []byte) error { return d.sendDataImmediate(conn, data) } + // A write larger than the send buffer can never fit in it, so it used + // to be refused outright with ErrSendBufFull — and since an IPC send has + // no reply, the client never knew: a message over MaxNagleBuf (256 KB) + // was dropped while the sender reported success. Feed a large write + // through the buffer in whole-segment pieces instead. Each piece is + // flushed before the next is taken (sendSegment blocks on the window), + // so the buffer never holds more than one piece of this write. + for len(data) > nagleWritePiece { + if err := d.sendDataPiece(conn, data[:nagleWritePiece]); err != nil { + return err + } + data = data[nagleWritePiece:] + } + return d.sendDataPiece(conn, data) +} + +// nagleWritePiece is how much of one large write SendData buffers at a time: +// a whole number of segments, so a piece leaves no tail behind for Nagle to +// hold, and well under MaxNagleBuf. +const nagleWritePiece = 16 * MaxSegmentSize + +// sendDataPiece buffers one write of at most nagleWritePiece bytes (or any +// smaller caller write) and flushes it under Nagle's algorithm. +func (d *Daemon) sendDataPiece(conn *Connection, data []byte) error { conn.NagleMu.Lock() // v1.9.1: cap NagleBuf at MaxNagleBuf. Without this, slow peers / // full cwnd / packet loss caused the buffer to grow without bound, diff --git a/pkg/daemon/zz_nagle_buf_bug_test.go b/pkg/daemon/zz_nagle_buf_bug_test.go index a3a95a44..1dfa2ad8 100644 --- a/pkg/daemon/zz_nagle_buf_bug_test.go +++ b/pkg/daemon/zz_nagle_buf_bug_test.go @@ -61,26 +61,9 @@ func TestSendDataNagleBufGrowsUnbounded(t *testing.T) { conn.PeerRecvWin = 1 << 20 // advertise 1 MB receive window so cwnd is the binding constraint conn.Mu.Unlock() - // FIXED (v1.9.1): SendData now rejects an oversized write up-front - // instead of appending it to NagleBuf. The 5 MiB write should - // return ErrSendBufFull immediately and NagleBuf should remain - // at 0 (not grow at all). MaxNagleBuf = 64 * MaxSegmentSize = 256 KB. - const payloadSize = 5 * 1024 * 1024 - payload := make([]byte, payloadSize) - for i := range payload { - payload[i] = byte(i) - } - err := d.SendData(conn, payload) - if !errors.Is(err, ErrSendBufFull) { - t.Fatalf("expected ErrSendBufFull on 5 MiB oversized write; got %v", err) - } - - conn.NagleMu.Lock() - bufLen := len(conn.NagleBuf) - conn.NagleMu.Unlock() - if bufLen != 0 { - t.Errorf("NagleBuf should be empty after rejected oversized write; got %d bytes", bufLen) - } + // The cap on NagleBuf is what fixed the OOM. (A single write larger + // than the cap used to be refused here too; it is now fed through the + // buffer in pieces — see TestSendDataOversizedWriteStaysWithinNagleCap.) // Cap is invariant: a sequence of small writes that, in aggregate, // would exceed the cap also rejects the offending one. Send up @@ -106,7 +89,7 @@ func TestSendDataNagleBufGrowsUnbounded(t *testing.T) { // And confirm NagleBuf still didn't grow past the cap. conn.NagleMu.Lock() - bufLen = len(conn.NagleBuf) + bufLen := len(conn.NagleBuf) conn.NagleMu.Unlock() if bufLen > MaxNagleBuf { t.Errorf("NagleBuf grew past cap %d; got %d", MaxNagleBuf, bufLen) @@ -120,3 +103,62 @@ func TestSendDataNagleBufGrowsUnbounded(t *testing.T) { d.ports.RemoveConnection(conn.ID) _ = time.Millisecond // keep `time` import live regardless of what's in cleanup } + +// A write larger than MaxNagleBuf used to be refused with ErrSendBufFull. +// Over IPC that refusal has nowhere to go (a send has no reply), so a client +// writing a 1 MB message lost it while reporting success. SendData now feeds +// such a write through the buffer in pieces and blocks on the window like any +// other write. The OOM guard is unchanged: against a peer that never ACKs, +// NagleBuf must stay within the cap for as long as the write is blocked. +func TestSendDataOversizedWriteStaysWithinNagleCap(t *testing.T) { + t.Parallel() + d := New(Config{}) + t.Cleanup(func() { d.tunnels.Close() }) + + const peerNode uint32 = 0xBA5EBA12 + peerConn := addPeerOnDaemon(t, d, peerNode) + t.Cleanup(func() { peerConn.Close() }) + d.setNodeID_testhelper(0x33330001) + + conn := d.ports.NewConnection(40001, protocol.Addr{Network: 0, Node: peerNode}, 80) + conn.Mu.Lock() + conn.LocalAddr = protocol.Addr{Network: 0, Node: 0x33330001} + conn.RemoteAddr = protocol.Addr{Network: 0, Node: peerNode} + conn.RemotePort = 80 + conn.State = StateEstablished + conn.PeerRecvWin = 1 << 20 + conn.Mu.Unlock() + + done := make(chan error, 1) + go func() { done <- d.SendData(conn, make([]byte, 5*1024*1024)) }() + + // The peer never ACKs, so the write blocks once the congestion window is + // full. Watch the buffer while it does. + maxBuf := 0 + for deadline := time.Now().Add(500 * time.Millisecond); time.Now().Before(deadline); time.Sleep(5 * time.Millisecond) { + conn.NagleMu.Lock() + if n := len(conn.NagleBuf); n > maxBuf { + maxBuf = n + } + conn.NagleMu.Unlock() + } + if maxBuf > MaxNagleBuf { + t.Errorf("NagleBuf reached %d bytes during an oversized write; cap is %d", maxBuf, MaxNagleBuf) + } + select { + case err := <-done: + t.Fatalf("a 5 MiB write to a peer that never ACKs returned (%v); it should block on the window", err) + default: + } + + // Closing the connection releases the blocked writer with an error. + d.ports.RemoveConnection(conn.ID) + select { + case err := <-done: + if err == nil { + t.Error("blocked write returned nil after the connection closed") + } + case <-time.After(5 * time.Second): + t.Fatal("blocked write did not return after the connection closed") + } +} diff --git a/tests/zz_large_write_test.go b/tests/zz_large_write_test.go new file mode 100644 index 00000000..0bdc6530 --- /dev/null +++ b/tests/zz_large_write_test.go @@ -0,0 +1,71 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package tests + +import ( + "bytes" + "crypto/sha256" + "io" + "testing" + "time" +) + +// One stream write larger than the daemon's send buffer (MaxNagleBuf, 256 KB) +// was refused with "send buffer full". An IPC send has no reply, so the +// client was told nothing: a 1 MB message was dropped while the sender +// reported success. The write is now fed through the buffer in pieces. +func TestSingleWriteLargerThanSendBufferIsDelivered(t *testing.T) { + requireRealNetwork(t) + t.Parallel() + env := NewTestEnv(t) + a := env.AddDaemon() + b := env.AddDaemon() + + const port = 5103 + payload := make([]byte, 1<<20) + for i := range payload { + payload[i] = byte(i * 131) + } + want := sha256.Sum256(payload) + + ln, err := a.Driver.Listen(port) + if err != nil { + t.Fatalf("listen: %v", err) + } + defer ln.Close() + got := make(chan []byte, 1) + go func() { + conn, err := ln.Accept() + if err != nil { + return + } + defer conn.Close() + buf := make([]byte, len(payload)) + if _, err := io.ReadFull(conn, buf); err != nil { + got <- nil + return + } + got <- buf + }() + + conn, err := b.Driver.DialAddr(a.Daemon.Addr(), port) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + if n, err := conn.Write(payload); err != nil || n != len(payload) { + t.Fatalf("write: n=%d err=%v", n, err) + } + + select { + case buf := <-got: + if buf == nil { + t.Fatal("receiver did not get the whole payload") + } + if sum := sha256.Sum256(buf); !bytes.Equal(sum[:], want[:]) { + t.Fatal("payload arrived corrupted") + } + case <-time.After(20 * time.Second): + t.Fatal("a 1 MB write was not delivered within 20s") + } +} From b3d0d1f964267d08f291685303ad8cf234804454 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Wed, 7 Oct 2026 16:26:29 +0300 Subject: [PATCH 3/4] send-message: keep large writes whole and report every undelivered message Review fixes for the large-payload change. Daemon: - A write over one Nagle piece went through the buffer in pieces with the send lock released between them, so two writers on one connection could interleave their bytes. A per-connection WriteMu now holds for the whole write. - The piece size is a whole number of 1152-byte segments, and whether a short tail may be held is decided once from the whole write, so a large write sends its last piece at once instead of holding it for an ACK. pilotctl: - The payload is read and checked before the daemon is contacted. A payload over the data-exchange frame limit is refused with a size error instead of being dropped by the receiver; an empty --data-file is named. - A send that failed outright exited 0 with an error field; it now fails. - With --count, a message that failed, was not acknowledged or was refused by the receiver made no difference to the exit status; any such message now fails the command with a count and the first reason. - The ACK read error is kept in the result and shown when a message is not acknowledged. - The usage text and docs/cli-reference.md mention --data-file. Co-Authored-By: Claude Opus 5.5 --- cmd/pilotctl/main.go | 95 +++++++++++--- cmd/pilotctl/zz_send_message_payload_test.go | 52 ++++++++ docs/cli-reference.md | 2 +- pkg/daemon/daemon.go | 44 ++++--- pkg/daemon/ports.go | 5 + pkg/daemon/zz_large_write_pieces_test.go | 124 +++++++++++++++++++ pkg/daemon/zz_nagle_buf_bug_test.go | 42 ++----- 7 files changed, 303 insertions(+), 61 deletions(-) create mode 100644 pkg/daemon/zz_large_write_pieces_test.go diff --git a/cmd/pilotctl/main.go b/cmd/pilotctl/main.go index 6f82500d..23bf0f06 100644 --- a/cmd/pilotctl/main.go +++ b/cmd/pilotctl/main.go @@ -1632,7 +1632,7 @@ Communication commands: pilotctl send --data [--timeout ] pilotctl recv [--count ] [--timeout ] pilotctl send-file - pilotctl send-message --data [--type text|json|binary] [--count ] [--reuse-conn] [--wait ] + pilotctl send-message --data | --data-file [--type text|json|binary] [--count ] [--reuse-conn] [--wait ] pilotctl dgram --data pilotctl subscribe [--count ] [--timeout ] pilotctl publish --data @@ -4775,11 +4775,62 @@ func messagePayload(flags map[string]string, stdin io.Reader) (string, error) { data = string(b) } if data == "" { + if hasFile { + return "", fmt.Errorf("--data-file %s is empty", file) + } return "", fmt.Errorf("--data is required") } + // One message is one data-exchange frame. A receiver drops a frame over + // its limit at the header, which the sender would only see as a missing + // acknowledgement, and past 4 GiB the length prefix would wrap. + if limit := maxMessageBytes(); len(data) > limit { + return "", fmt.Errorf("message is %d bytes; one message can carry at most %d (use send-file for larger payloads)", len(data), limit) + } return data, nil } +// maxMessageBytes is the most one send-message payload can be: the frame +// limit, less room for the frame and message-tag headers. +func maxMessageBytes() int { + return int(dataexchange.MaxFrameSize) - 4096 +} + +// failIfUndelivered ends a multi-message send with an error when any message +// failed to send, was not acknowledged, or was refused by the receiver: +// reporting "ok" for those hid messages that never arrived. +func failIfUndelivered(target string, results []map[string]interface{}) { + if failed, first := undelivered(results); failed > 0 { + fatalHint("connection_failed", first, "%d of %d messages to %s were not delivered", failed, len(results), target) + } +} + +// undelivered counts the send results that did not end in a stored message, +// and describes the first of them. +func undelivered(results []map[string]interface{}) (failed int, first string) { + for _, r := range results { + why := "" + ackText, acked := r["ack"].(string) + switch { + case r["error"] != nil: + why = fmt.Sprint(r["error"]) + case !acked: + why = "not acknowledged" + if e, ok := r["ack_error"].(string); ok { + why += " (" + e + ")" + } + case strings.HasPrefix(ackText, "ERR "): + why = "receiver rejected it: " + ackText + } + if why != "" { + failed++ + if first == "" { + first = fmt.Sprintf("message %v: %s", r["seq"], why) + } + } + } + return failed, first +} + func cmdSendMessage(args []string) { flags, pos := parseFlags(args) for name := range flags { @@ -4804,7 +4855,7 @@ func cmdSendMessage(args []string) { defer timer.Stop() } if len(pos) != 1 { - fatalCode("invalid_argument", "usage: pilotctl send-message --data [--type text|json|binary] [--trace] [--count ] [--reuse-conn] [--wait ] [--timeout ] [--no-resend]") + fatalCode("invalid_argument", "usage: pilotctl send-message --data | --data - | --data-file [--type text|json|binary] [--trace] [--count ] [--reuse-conn] [--wait ] [--timeout ] [--no-resend]") } sendCount := flagInt(flags, "count", 1) @@ -4848,6 +4899,13 @@ func cmdSendMessage(args []string) { } } + // Read and check the payload first, so a missing file or an oversized + // message is reported before the daemon or the network is touched. + data, err := messagePayload(flags, os.Stdin) + if err != nil { + fatalCode("invalid_argument", "%v", err) + } + d := connectDriver() tracef("connectDriver") // d may be replaced by a fresh connection when a dial is retried. @@ -4859,10 +4917,6 @@ func cmdSendMessage(args []string) { fatalCode("not_found", "%v", err) } - data, err := messagePayload(flags, os.Stdin) - if err != nil { - fatalCode("invalid_argument", "%v", err) - } msgType := flagString(flags, "type", "text") // First contact: the daemon holds no session with the peer yet, so this @@ -4939,6 +4993,9 @@ func cmdSendMessage(args []string) { if ack != nil { r["ack"] = string(ack.Payload) } + if ackErr != nil { + r["ack_error"] = ackErr.Error() + } if traceTime { r["total_ms"] = float64(time.Duration(ackRecvAtNs-sentAtNs).Microseconds()) / 1000.0 if ack != nil && ack.Type == dataexchange.TypeJSON { @@ -5013,13 +5070,22 @@ func cmdSendMessage(args []string) { // Every receiver answers a stored message with an ACK frame. No ACK // means the message was not stored, or was never sent — the daemon // drops a write it cannot buffer without telling the client — so - // "ok" here was reporting messages that never arrived. - if _, failed := r["error"]; !failed { - if _, acked := r["ack"]; !acked { - fatalHint("connection_failed", - "the receiver did not confirm it stored the message; check `pilotctl peers` and the daemon log, then send again", - "%s did not acknowledge the message (%d bytes)", target, len(data)) + // "ok" here was reporting messages that never arrived. A send that + // failed outright was reported as "ok" with an error field; it is a + // failure too. + if e, failed := r["error"].(string); failed { + fatalHint("connection_failed", + "the message was not sent; check `pilotctl peers` and the daemon log, then send again", + "sending to %s failed: %s", target, e) + } + if _, acked := r["ack"]; !acked { + why := "" + if e, ok := r["ack_error"].(string); ok { + why = ": " + e } + fatalHint("connection_failed", + "the receiver did not confirm it stored the message; check `pilotctl peers` and the daemon log, then send again", + "%s did not acknowledge the message (%d bytes)%s", target, len(data), why) } // The receiver answers "ERR ..." when it could not store the // message (disk full, inbox unwritable). That is a failed send, not @@ -5059,8 +5125,7 @@ func cmdSendMessage(args []string) { // more on a new stream if nothing arrived by mid-window. // --no-resend opts out for requests that are not safe to // repeat. - _, sendFailed := r["error"] - if firstContact && !sendFailed && !flagBool(flags, "no-resend") { + if firstContact && !flagBool(flags, "no-resend") { if after := resendDelay(waitDur); after > 0 { cfg.resendAfter = after cfg.resend = func() (time.Time, error) { @@ -5131,6 +5196,7 @@ func cmdSendMessage(args []string) { time.Sleep(50 * time.Millisecond) } } + failIfUndelivered(target.String(), results) outputOK(map[string]interface{}{ "target": target.String(), "to": target.String(), @@ -5151,6 +5217,7 @@ func cmdSendMessage(args []string) { time.Sleep(50 * time.Millisecond) } } + failIfUndelivered(target.String(), results) outputOK(map[string]interface{}{ "target": target.String(), "to": target.String(), diff --git a/cmd/pilotctl/zz_send_message_payload_test.go b/cmd/pilotctl/zz_send_message_payload_test.go index 82e17ff1..cf17e4e8 100644 --- a/cmd/pilotctl/zz_send_message_payload_test.go +++ b/cmd/pilotctl/zz_send_message_payload_test.go @@ -7,6 +7,8 @@ import ( "path/filepath" "strings" "testing" + + "github.com/pilot-protocol/dataexchange" ) // send-message took its body only as a command-line argument, which the OS @@ -18,6 +20,10 @@ func TestMessagePayload(t *testing.T) { if err := os.WriteFile(file, []byte(big), 0o600); err != nil { t.Fatal(err) } + empty := filepath.Join(t.TempDir(), "empty") + if err := os.WriteFile(empty, nil, 0o600); err != nil { + t.Fatal(err) + } for _, tc := range []struct { name string @@ -34,6 +40,7 @@ func TestMessagePayload(t *testing.T) { {name: "both", flags: map[string]string{"data": "a", "data-file": file}, wantErr: "not both"}, {name: "file flag without a path", flags: map[string]string{"data-file": "true"}, wantErr: "needs a path"}, {name: "missing file", flags: map[string]string{"data-file": filepath.Join(t.TempDir(), "nope")}, wantErr: "read --data-file"}, + {name: "empty file", flags: map[string]string{"data-file": empty}, wantErr: "is empty"}, } { got, err := messagePayload(tc.flags, strings.NewReader(tc.stdin)) if tc.wantErr != "" { @@ -51,3 +58,48 @@ func TestMessagePayload(t *testing.T) { } } } + +// One message is one data-exchange frame. A larger payload was sent anyway +// and dropped by the receiver at the frame header, which the sender only saw +// as a missing acknowledgement. +func TestMessagePayloadFrameLimit(t *testing.T) { + // A small frame limit keeps the test cheap. + defer func(old uint32) { dataexchange.MaxFrameSize = old }(dataexchange.MaxFrameSize) + dataexchange.MaxFrameSize = 64 << 10 + + limit := maxMessageBytes() + if _, err := messagePayload(map[string]string{"data": "-"}, strings.NewReader(strings.Repeat("x", limit+1))); err == nil || !strings.Contains(err.Error(), "at most") { + t.Errorf("a payload one byte over the limit: err = %v, want a size error", err) + } + if got, err := messagePayload(map[string]string{"data": "-"}, strings.NewReader(strings.Repeat("x", limit))); err != nil || len(got) != limit { + t.Errorf("a payload at the limit: %d bytes, err = %v", len(got), err) + } +} + +// A send of several messages reported "ok" even when some were never sent or +// never acknowledged, so a caller could not tell that messages were lost. +func TestUndeliveredMessages(t *testing.T) { + results := []map[string]interface{}{ + {"seq": 0, "ack": "ACK TEXT 5 bytes"}, + {"seq": 1, "error": "write: connection closed"}, + {"seq": 2, "ack_error": "EOF"}, + {"seq": 3, "ack": "ERR inbox not writable"}, + {"seq": 4, "ack": "ACK TEXT 5 bytes"}, + } + failed, first := undelivered(results) + if failed != 3 { + t.Errorf("failed = %d, want 3", failed) + } + if first != "message 1: write: connection closed" { + t.Errorf("first = %q", first) + } + if failed, _ := undelivered(results[4:]); failed != 0 { + t.Errorf("an acknowledged message counted as undelivered") + } + if _, first := undelivered(results[2:3]); !strings.Contains(first, "not acknowledged (EOF)") { + t.Errorf("first = %q, want the ACK read error", first) + } + if _, first := undelivered(results[3:4]); !strings.Contains(first, "ERR inbox not writable") { + t.Errorf("first = %q, want the receiver's refusal", first) + } +} diff --git a/docs/cli-reference.md b/docs/cli-reference.md index 28adad23..ad0cbdb6 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -43,7 +43,7 @@ Communication commands: pilotctl send --data [--timeout ] pilotctl recv [--count ] [--timeout ] pilotctl send-file - pilotctl send-message --data [--type text|json|binary] [--count ] [--reuse-conn] [--wait ] + pilotctl send-message --data | --data-file [--type text|json|binary] [--count ] [--reuse-conn] [--wait ] pilotctl dgram --data pilotctl subscribe [--count ] [--timeout ] pilotctl publish --data diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index 0fccf234..40f7baf6 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -4278,12 +4278,13 @@ const QuickACKBudget = 32 // segments unless NoDelay is set. Large writes (>= MSS) are sent immediately. // ErrSendBufFull is returned by SendData when the per-connection // NagleBuf would exceed MaxNagleBuf if the caller's write were -// appended. Callers must back off and retry — typically by waiting -// for a webhook or polling the connection's send-buffer state. +// appended. It bounds the daemon's memory per connection. // -// This error replaces the silent unbounded-growth behavior that -// could OOM the daemon when an application wrote faster than the -// network drained. Pinned by TestSendDataNagleBufGrowsUnbounded. +// SendData returns it only before any byte of the write has been buffered, +// so retrying the whole write is safe. With one write at a time per +// connection (WriteMu) and large writes fed through in pieces that drain +// before the next is taken, the buffer holds at most a held tail and one +// piece, so in practice it is not returned at all. var ErrSendBufFull = errors.New("send buffer full") func (d *Daemon) SendData(conn *Connection, data []byte) error { @@ -4294,35 +4295,48 @@ func (d *Daemon) SendData(conn *Connection, data []byte) error { return fmt.Errorf("connection not established") } + // One write at a time per connection: the pieces of a large write + // below must not interleave with another writer's bytes. + conn.WriteMu.Lock() + defer conn.WriteMu.Unlock() + // If Nagle is disabled (NoDelay), send everything immediately in segments if conn.NoDelay { return d.sendDataImmediate(conn, data) } + // A write of a segment or more takes its tail with it: it is the end of + // a message, not one of a run of small writes that would coalesce. + // Decided once, from the whole write, not from each piece. + tailNow := len(data) >= SendSegmentSize + // A write larger than the send buffer can never fit in it, so it used // to be refused outright with ErrSendBufFull — and since an IPC send has // no reply, the client never knew: a message over MaxNagleBuf (256 KB) // was dropped while the sender reported success. Feed a large write // through the buffer in whole-segment pieces instead. Each piece is // flushed before the next is taken (sendSegment blocks on the window), - // so the buffer never holds more than one piece of this write. + // and WriteMu keeps other writers out, so the buffer never holds more + // than a held tail and one piece of this write. for len(data) > nagleWritePiece { - if err := d.sendDataPiece(conn, data[:nagleWritePiece]); err != nil { + if err := d.sendDataPiece(conn, data[:nagleWritePiece], tailNow); err != nil { return err } data = data[nagleWritePiece:] } - return d.sendDataPiece(conn, data) + return d.sendDataPiece(conn, data, tailNow) } // nagleWritePiece is how much of one large write SendData buffers at a time: -// a whole number of segments, so a piece leaves no tail behind for Nagle to -// hold, and well under MaxNagleBuf. -const nagleWritePiece = 16 * MaxSegmentSize +// a whole number of SendSegmentSize segments, so only the last piece of a +// write can end in a short segment, and well under MaxNagleBuf. +const nagleWritePiece = 56 * SendSegmentSize // sendDataPiece buffers one write of at most nagleWritePiece bytes (or any -// smaller caller write) and flushes it under Nagle's algorithm. -func (d *Daemon) sendDataPiece(conn *Connection, data []byte) error { +// smaller caller write) and flushes it under Nagle's algorithm. tailNow sends +// a short remainder at once instead of letting Nagle hold it. Caller holds +// conn.WriteMu. +func (d *Daemon) sendDataPiece(conn *Connection, data []byte, tailNow bool) error { conn.NagleMu.Lock() // v1.9.1: cap NagleBuf at MaxNagleBuf. Without this, slow peers / // full cwnd / packet loss caused the buffer to grow without bound, @@ -4336,9 +4350,7 @@ func (d *Daemon) sendDataPiece(conn *Connection, data []byte) error { conn.NagleBuf = append(conn.NagleBuf, data...) conn.NagleMu.Unlock() - // A write of a segment or more takes its tail with it: it is the end of - // a message, not one of a run of small writes that would coalesce. - _, err := d.flushNagle(conn, len(data) >= SendSegmentSize, false) + _, err := d.flushNagle(conn, tailNow, false) return err } diff --git a/pkg/daemon/ports.go b/pkg/daemon/ports.go index d682d35a..03a23035 100644 --- a/pkg/daemon/ports.go +++ b/pkg/daemon/ports.go @@ -317,6 +317,11 @@ type Connection struct { // Nagle algorithm (write coalescing) NagleBuf []byte // pending small write data NagleMu sync.Mutex // protects NagleBuf and tailFlusher + // WriteMu is held by SendData for the whole of one write, so two + // writers on a connection cannot interleave: a write larger than + // nagleWritePiece goes through the buffer piece by piece, and without it + // another writer's bytes could land between pieces. Taken before SendMu. + WriteMu sync.Mutex // SendMu is held from taking bytes out of NagleBuf until they are handed // to sendSegment, so segments leave in the order the bytes were written // whichever goroutine sends them. Taken before NagleMu. diff --git a/pkg/daemon/zz_large_write_pieces_test.go b/pkg/daemon/zz_large_write_pieces_test.go new file mode 100644 index 00000000..9645f6ef --- /dev/null +++ b/pkg/daemon/zz_large_write_pieces_test.go @@ -0,0 +1,124 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package daemon + +import ( + "bytes" + "sort" + "sync" + "testing" + "time" +) + +// ackEverything stands in for a peer that acknowledges at once: it keeps +// emptying the connection's retransmission queue and waking the sender. +func ackEverything(t *testing.T, conn *Connection) { + t.Helper() + stop := make(chan struct{}) + t.Cleanup(func() { close(stop) }) + go func() { + for { + select { + case <-stop: + return + case <-time.After(time.Millisecond): + } + conn.RetxMu.Lock() + conn.Unacked = nil + conn.RetxMu.Unlock() + select { + case conn.WindowCh <- struct{}{}: + default: + } + select { + case conn.NagleCh <- struct{}{}: + default: + } + } + }() +} + +// A write larger than nagleWritePiece goes through the send buffer piece by +// piece. Two writers on one connection must still each come out whole and in +// one run: without a per-connection write lock, review saw a 70000-byte +// write leave as A×65536 B×100 A×4464, and writers retrying ErrSendBufFull +// put parts of their writes on the wire twice. +func TestConcurrentLargeWritesDoNotInterleave(t *testing.T) { + t.Parallel() + d, pc, conn := newSendDataFixture(t, 0xDADA0001) + conn.NoDelay = false + conn.CongWin = MaxCongWin + ackEverything(t, conn) + + const size = 3*nagleWritePiece + 777 + writes := map[byte][]byte{'A': bytes.Repeat([]byte{'A'}, size), 'B': bytes.Repeat([]byte{'B'}, size), 'C': bytes.Repeat([]byte{'C'}, size)} + var wg sync.WaitGroup + for _, w := range writes { + wg.Add(1) + go func(w []byte) { + defer wg.Done() + if err := d.SendData(conn, w); err != nil { + t.Errorf("SendData: %v", err) + } + }(w) + } + + type seg struct { + seq uint32 + data []byte + } + var segs []seg + total := 0 + for total < len(writes)*size { + pkt := readOneSegment(t, pc, 5*time.Second) + segs = append(segs, seg{pkt.Seq, pkt.Payload}) + total += len(pkt.Payload) + } + wg.Wait() + if total != len(writes)*size { + t.Fatalf("%d bytes on the wire for %d written", total, len(writes)*size) + } + sort.Slice(segs, func(i, j int) bool { return seqAfter(segs[j].seq, segs[i].seq) }) + var stream []byte + for _, s := range segs { + stream = append(stream, s.data...) + } + runs := 0 + for i := range stream { + if i == 0 || stream[i] != stream[i-1] { + runs++ + } + } + if runs != len(writes) { + t.Fatalf("the stream holds %d runs of bytes for %d writes: writes were interleaved", runs, len(writes)) + } +} + +// Whether a short remainder may be held is decided from the whole write, not +// from each piece: the last piece of a 65537-byte write is one byte, and it +// must leave with the write rather than wait for an ACK. +func TestLargeWriteSendsItsLastPieceAtOnce(t *testing.T) { + t.Parallel() + d, pc, conn := newSendDataFixture(t, 0xDADA0002) + conn.NoDelay = false + conn.CongWin = MaxCongWin + conn.PeerRecvWin = 1 << 20 + + const size = nagleWritePiece + 1 + if err := d.SendData(conn, make([]byte, size)); err != nil { + t.Fatalf("SendData: %v", err) + } + conn.NagleMu.Lock() + held := len(conn.NagleBuf) + conn.NagleMu.Unlock() + if held != 0 { + t.Fatalf("%d bytes of a %d-byte write were held back", held, size) + } + got := 0 + for got < size { + got += len(readOneSegment(t, pc, 2*time.Second).Payload) + } + if nagleWritePiece%SendSegmentSize != 0 { + t.Fatalf("nagleWritePiece (%d) is not a whole number of %d-byte segments", nagleWritePiece, SendSegmentSize) + } +} diff --git a/pkg/daemon/zz_nagle_buf_bug_test.go b/pkg/daemon/zz_nagle_buf_bug_test.go index 1dfa2ad8..d0f913bf 100644 --- a/pkg/daemon/zz_nagle_buf_bug_test.go +++ b/pkg/daemon/zz_nagle_buf_bug_test.go @@ -10,34 +10,14 @@ import ( "github.com/pilot-protocol/common/protocol" ) -// TestSendDataNagleBufGrowsUnbounded reproduces the NagleBuf-OOM bug. -// -// Symptom (theoretical, observed-once-and-cleared in pilot service-agent -// memory profiling): an application that calls SendData faster than the -// network can drain (slow peer, packet loss, full cwnd) accumulates the -// unsent bytes in conn.NagleBuf without bound. There is currently NO -// per-connection send-buffer cap. With many connections in this state, -// daemon RSS climbs linearly with offered-but-undeliverable load until -// the host OOM-kills the process. -// -// Reproducer: a real UDP peer is reachable but never ACKs anything, so -// the daemon's cwnd fills after IW10 (10 × 4 KB = 40 KB). The 11th -// MSS-sized chunk's sendSegment call blocks waiting for cwnd to open -// (which it never will). The single goroutine calling SendData is -// stuck in nagleFlush; meanwhile, the data slice the application -// passed (5 MiB here) sits in conn.NagleBuf, well over any reasonable -// memory budget. Multiple writers would amplify this further. -// -// What v1.9.1's NagleBuf cap fix will change: -// - introduce MaxNagleBuf = 64 * MaxSegmentSize (256 KB) -// - SendData returns ErrSendBufFull when len(NagleBuf) + len(data) > MaxNagleBuf -// - test assertion flips: NagleBuf plateaus at MaxNagleBuf; -// SendData returns ErrSendBufFull on the oversized write. -// -// This test pins CURRENT (buggy) behavior so the cap patch has a -// concrete regression target. After the fix, the bug-asserting block -// below is replaced with the post-fix block (already drafted in -// comments). +// TestSendDataNagleBufGrowsUnbounded guards the fix for a NagleBuf OOM: an +// application calling SendData faster than the network drains (slow peer, +// loss, a full congestion window) used to pile the unsent bytes up in +// conn.NagleBuf without bound, and with many such connections the daemon's +// memory grew until the host killed it. NagleBuf is now capped at +// MaxNagleBuf. This test checks the cap at its boundary: a small write that +// would take the buffer past it is refused with ErrSendBufFull and the +// buffer does not grow. (The name is kept from when it pinned the bug.) func TestSendDataNagleBufGrowsUnbounded(t *testing.T) { t.Parallel() d := New(Config{}) @@ -142,8 +122,10 @@ func TestSendDataOversizedWriteStaysWithinNagleCap(t *testing.T) { } conn.NagleMu.Unlock() } - if maxBuf > MaxNagleBuf { - t.Errorf("NagleBuf reached %d bytes during an oversized write; cap is %d", maxBuf, MaxNagleBuf) + // One piece of the write plus, at most, a short remainder held from + // before it: the write goes through the buffer piece by piece. + if limit := nagleWritePiece + SendSegmentSize; maxBuf > limit { + t.Errorf("NagleBuf reached %d bytes during an oversized write, want at most one piece and a tail (%d)", maxBuf, limit) } select { case err := <-done: From 66441ae7c54ccf43348b896fd033dde4dce69852 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Wed, 7 Oct 2026 16:55:08 +0300 Subject: [PATCH 4/4] send-message: a large write survives tunnel send errors and stops at a close Second review round. Daemon: - A tunnel send error part-way through a large write ended the write. The failed segment is tracked and retransmitted, but the pieces not yet buffered were dropped, and the connection's next write landed in their place in the stream. The write now keeps draining after such an error and stops only when the connection is gone. - A large write blocked on the window went on sending after a local close. The state is checked again after WriteMu and between pieces, so at most the piece already buffered follows the FIN. - The tail test now leaves a short segment in flight first, which is when deciding per piece would have held the last byte; it failed against that. pilotctl: - With --trace a receiver's refusal is in inner_ack; both the single and the multi-message checks now see it. - A failed --count run keeps every message's result in the error, so the caller can tell which messages to send again. A run where the receiver refused every failed message uses the same code as a single refusal. - The usage line, the command catalogue and the changelog mention --data -. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 3 + cmd/pilotctl/main.go | 60 ++++++++++---- cmd/pilotctl/zz_send_message_payload_test.go | 19 +++-- docs/cli-reference.md | 2 +- pkg/daemon/daemon.go | 46 +++++++++-- pkg/daemon/zz_large_write_pieces_test.go | 87 +++++++++++++++++++- 6 files changed, 185 insertions(+), 32 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8f79b8e9..21dd2f63 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,9 @@ Detailed per-release notes are on the capped by the OS — a 1 MB `--data` failed with "Argument list too long", and on Linux a single argument stops at 128 KiB — which is why senders of large bodies needed a separate stdin helper. + `--data -` now means stdin, so a message that is just `-` has to come from + `--data-file`. One message can carry up to the 64 MiB data-exchange frame + limit; a larger payload is refused before anything is sent. - **A client can ask the daemon whether a datagram was actually sent.** The IPC `SendTo` command is fire-and-forget: when the daemon could not send a datagram (no route to the node, port policy, ephemeral ports exhausted) it diff --git a/cmd/pilotctl/main.go b/cmd/pilotctl/main.go index 23bf0f06..a1e3f69c 100644 --- a/cmd/pilotctl/main.go +++ b/cmd/pilotctl/main.go @@ -185,6 +185,10 @@ func classifyDaemonError(err error) string { return "" } +// fatalResults, when set, goes into fatalHint's JSON envelope as "results", +// so a multi-message send that fails still says what became of each message. +var fatalResults []map[string]interface{} + // fatalHint is like fatalCode but adds an actionable hint telling the user what to do next. func fatalHint(code, hint, format string, args ...interface{}) { msg := fmt.Sprintf(format, args...) @@ -204,6 +208,9 @@ func fatalHint(code, hint, format string, args ...interface{}) { if ns := nextStepsEnvelope(exitNextSteps); ns != nil { env["next_steps"] = ns } + if fatalResults != nil { + env["results"] = fatalResults + } b, _ := json.Marshal(env) fmt.Fprintln(os.Stderr, string(b)) } else { @@ -1632,7 +1639,7 @@ Communication commands: pilotctl send --data [--timeout ] pilotctl recv [--count ] [--timeout ] pilotctl send-file - pilotctl send-message --data | --data-file [--type text|json|binary] [--count ] [--reuse-conn] [--wait ] + pilotctl send-message --data | --data - | --data-file [--type text|json|binary] [--count ] [--reuse-conn] [--wait ] pilotctl dgram --data pilotctl subscribe [--count ] [--timeout ] pilotctl publish --data @@ -2491,8 +2498,8 @@ func contextCatalog() map[string]interface{} { // Messaging "send-message": map[string]interface{}{ - "args": []string{"", "--data ", "[--type text|json|binary]", "[--count ]", "[--reuse-conn]", "[--wait ]", "[--timeout ]"}, - "description": "Send a typed message to a node via data exchange (port 1001). --count N sends N messages; --reuse-conn shares one connection across all N (env: PILOT_SENDMSG_REUSE_CONN=1). Default type: text", + "args": []string{"", "--data | --data - | --data-file ", "[--type text|json|binary]", "[--count ]", "[--reuse-conn]", "[--wait ]", "[--timeout ]"}, + "description": "Send a typed message to a node via data exchange (port 1001). --data - reads the payload from stdin, --data-file from a file (up to 64 MiB; a command-line argument is capped by the OS). --count N sends N messages; --reuse-conn shares one connection across all N (env: PILOT_SENDMSG_REUSE_CONN=1). Fails unless every message is acknowledged. Default type: text", "returns": "target, to, type, bytes, ack, reuse_conn", }, "send-file": map[string]interface{}{ @@ -4797,19 +4804,30 @@ func maxMessageBytes() int { // failIfUndelivered ends a multi-message send with an error when any message // failed to send, was not acknowledged, or was refused by the receiver: -// reporting "ok" for those hid messages that never arrived. +// reporting "ok" for those hid messages that never arrived. The error carries +// every message's result, so a caller can tell which ones to send again. func failIfUndelivered(target string, results []map[string]interface{}) { - if failed, first := undelivered(results); failed > 0 { - fatalHint("connection_failed", first, "%d of %d messages to %s were not delivered", failed, len(results), target) + failed, allRefused, first := undelivered(results) + if failed == 0 { + return } + code := "connection_failed" + if allRefused { + code = "internal" // as for a single message the receiver refuses + } + fatalResults = results + fatalHint(code, "the results list which messages were delivered; check `pilotctl peers` and the daemon log before sending the others again", + "%d of %d messages to %s were not delivered (first, %s)", failed, len(results), target, first) } // undelivered counts the send results that did not end in a stored message, -// and describes the first of them. -func undelivered(results []map[string]interface{}) (failed int, first string) { +// says whether all of those were refused by the receiver, and describes the +// first of them. +func undelivered(results []map[string]interface{}) (failed int, allRefused bool, first string) { + allRefused = true for _, r := range results { - why := "" - ackText, acked := r["ack"].(string) + why, refused := "", false + _, acked := r["ack"].(string) switch { case r["error"] != nil: why = fmt.Sprint(r["error"]) @@ -4818,17 +4836,29 @@ func undelivered(results []map[string]interface{}) (failed int, first string) { if e, ok := r["ack_error"].(string); ok { why += " (" + e + ")" } - case strings.HasPrefix(ackText, "ERR "): - why = "receiver rejected it: " + ackText + case refusal(r) != "": + why, refused = "receiver refused it: "+refusal(r), true } if why != "" { failed++ + allRefused = allRefused && refused if first == "" { first = fmt.Sprintf("message %v: %s", r["seq"], why) } } } - return failed, first + return failed, failed > 0 && allRefused, first +} + +// refusal returns the receiver's "ERR ..." answer in a send result: its ACK, +// or with --trace the inner ACK that the timing reply carries. +func refusal(r map[string]interface{}) string { + for _, k := range []string{"ack", "inner_ack"} { + if s, _ := r[k].(string); strings.HasPrefix(s, "ERR ") { + return s + } + } + return "" } func cmdSendMessage(args []string) { @@ -5090,8 +5120,8 @@ func cmdSendMessage(args []string) { // 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) + if rej := refusal(r); rej != "" { + fatalCode("internal", "receiver rejected message: %s", rej) } result := map[string]interface{}{ "target": target.String(), diff --git a/cmd/pilotctl/zz_send_message_payload_test.go b/cmd/pilotctl/zz_send_message_payload_test.go index cf17e4e8..3f9497ff 100644 --- a/cmd/pilotctl/zz_send_message_payload_test.go +++ b/cmd/pilotctl/zz_send_message_payload_test.go @@ -86,20 +86,25 @@ func TestUndeliveredMessages(t *testing.T) { {"seq": 3, "ack": "ERR inbox not writable"}, {"seq": 4, "ack": "ACK TEXT 5 bytes"}, } - failed, first := undelivered(results) - if failed != 3 { - t.Errorf("failed = %d, want 3", failed) + failed, allRefused, first := undelivered(results) + if failed != 3 || allRefused { + t.Errorf("failed = %d, allRefused = %v, want 3 and false", failed, allRefused) } if first != "message 1: write: connection closed" { t.Errorf("first = %q", first) } - if failed, _ := undelivered(results[4:]); failed != 0 { + if failed, _, _ := undelivered(results[4:]); failed != 0 { t.Errorf("an acknowledged message counted as undelivered") } - if _, first := undelivered(results[2:3]); !strings.Contains(first, "not acknowledged (EOF)") { + if _, _, first := undelivered(results[2:3]); !strings.Contains(first, "not acknowledged (EOF)") { t.Errorf("first = %q, want the ACK read error", first) } - if _, first := undelivered(results[3:4]); !strings.Contains(first, "ERR inbox not writable") { - t.Errorf("first = %q, want the receiver's refusal", first) + if _, allRefused, first := undelivered(results[3:4]); !allRefused || !strings.Contains(first, "ERR inbox not writable") { + t.Errorf("first = %q, allRefused = %v, want the receiver's refusal", first, allRefused) + } + // With --trace the ACK is timing JSON and a refusal is in inner_ack. + traced := []map[string]interface{}{{"seq": 0, "ack": `{"received_at_ns":1}`, "inner_ack": "ERR disk full"}} + if failed, allRefused, _ := undelivered(traced); failed != 1 || !allRefused { + t.Errorf("a refusal reported in inner_ack counted as delivered") } } diff --git a/docs/cli-reference.md b/docs/cli-reference.md index ad0cbdb6..6fd73958 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -43,7 +43,7 @@ Communication commands: pilotctl send --data [--timeout ] pilotctl recv [--count ] [--timeout ] pilotctl send-file - pilotctl send-message --data | --data-file [--type text|json|binary] [--count ] [--reuse-conn] [--wait ] + pilotctl send-message --data | --data - | --data-file [--type text|json|binary] [--count ] [--reuse-conn] [--wait ] pilotctl dgram --data pilotctl subscribe [--count ] [--timeout ] pilotctl publish --data diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index 40f7baf6..bbcf874f 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -4288,10 +4288,7 @@ const QuickACKBudget = 32 var ErrSendBufFull = errors.New("send buffer full") func (d *Daemon) SendData(conn *Connection, data []byte) error { - conn.Mu.Lock() - st := conn.State - conn.Mu.Unlock() - if st != StateEstablished { + if !connEstablished(conn) { return fmt.Errorf("connection not established") } @@ -4299,6 +4296,10 @@ func (d *Daemon) SendData(conn *Connection, data []byte) error { // below must not interleave with another writer's bytes. conn.WriteMu.Lock() defer conn.WriteMu.Unlock() + // The connection may have been closed while this writer queued. + if !connEstablished(conn) { + return fmt.Errorf("connection not established") + } // If Nagle is disabled (NoDelay), send everything immediately in segments if conn.NoDelay { @@ -4320,13 +4321,48 @@ func (d *Daemon) SendData(conn *Connection, data []byte) error { // than a held tail and one piece of this write. for len(data) > nagleWritePiece { if err := d.sendDataPiece(conn, data[:nagleWritePiece], tailNow); err != nil { - return err + if err := d.drainAfterSendError(conn, err, tailNow); err != nil { + return err + } } data = data[nagleWritePiece:] + // Closed while this write waited on the window: send no more of + // it, nothing may follow the FIN. + if !connEstablished(conn) { + return protocol.ErrConnClosed + } } return d.sendDataPiece(conn, data, tailNow) } +// connEstablished reports whether conn is still open for sending. +func connEstablished(conn *Connection) bool { + conn.Mu.Lock() + defer conn.Mu.Unlock() + return conn.State == StateEstablished +} + +// drainAfterSendError finishes flushing a piece of a large write after a +// send error, so the rest of the write can follow it. +// +// A tunnel send error does not lose the segment: it was given its sequence +// number and tracked before the send, and the retransmit loop resends it. +// Stopping the write there instead dropped the pieces not yet buffered — they +// never got sequence numbers — and the connection's next write landed in +// their place in the stream. Every failed attempt still commits a segment, so +// this ends (or blocks on the window like any write). It gives up only when +// the connection is gone. +func (d *Daemon) drainAfterSendError(conn *Connection, err error, tailNow bool) error { + for err != nil { + if errors.Is(err, protocol.ErrConnClosed) || errors.Is(err, ErrSendBufFull) || !connEstablished(conn) { + return err + } + slog.Debug("stream send failed; the segment will be retransmitted", "conn_id", conn.ID, "err", err) + _, err = d.flushNagle(conn, tailNow, false) + } + return nil +} + // nagleWritePiece is how much of one large write SendData buffers at a time: // a whole number of SendSegmentSize segments, so only the last piece of a // write can end in a short segment, and well under MaxNagleBuf. diff --git a/pkg/daemon/zz_large_write_pieces_test.go b/pkg/daemon/zz_large_write_pieces_test.go index 9645f6ef..aaef0d47 100644 --- a/pkg/daemon/zz_large_write_pieces_test.go +++ b/pkg/daemon/zz_large_write_pieces_test.go @@ -8,6 +8,8 @@ import ( "sync" "testing" "time" + + "github.com/pilot-protocol/common/protocol" ) // ackEverything stands in for a peer that acknowledges at once: it keeps @@ -95,8 +97,10 @@ func TestConcurrentLargeWritesDoNotInterleave(t *testing.T) { } // Whether a short remainder may be held is decided from the whole write, not -// from each piece: the last piece of a 65537-byte write is one byte, and it -// must leave with the write rather than wait for an ACK. +// from each piece: the last piece of a large write is short, and it must +// leave with the write rather than wait for an ACK. Nagle holds a short +// remainder only while an earlier short segment is unacknowledged, so one is +// left in flight first. func TestLargeWriteSendsItsLastPieceAtOnce(t *testing.T) { t.Parallel() d, pc, conn := newSendDataFixture(t, 0xDADA0002) @@ -104,6 +108,12 @@ func TestLargeWriteSendsItsLastPieceAtOnce(t *testing.T) { conn.CongWin = MaxCongWin conn.PeerRecvWin = 1 << 20 + // A short write goes out and is never acknowledged. + if err := d.SendData(conn, make([]byte, 10)); err != nil { + t.Fatalf("SendData: %v", err) + } + readOneSegment(t, pc, 2*time.Second) + const size = nagleWritePiece + 1 if err := d.SendData(conn, make([]byte, size)); err != nil { t.Fatalf("SendData: %v", err) @@ -118,7 +128,76 @@ func TestLargeWriteSendsItsLastPieceAtOnce(t *testing.T) { for got < size { got += len(readOneSegment(t, pc, 2*time.Second).Payload) } - if nagleWritePiece%SendSegmentSize != 0 { - t.Fatalf("nagleWritePiece (%d) is not a whole number of %d-byte segments", nagleWritePiece, SendSegmentSize) +} + +// nagleWritePiece is a whole number of segments, so only the last piece of a +// write can end in a short segment. +var _ [0]struct{} = [nagleWritePiece % SendSegmentSize]struct{}{} + +// A tunnel send error part-way through a large write used to end the write: +// the failed segment is tracked and retransmitted, but the pieces not yet +// buffered were dropped, and the connection's next write took their place in +// the stream. Every byte of the write must be committed to the stream — given +// a sequence number, or still in the send buffer — in order. +func TestLargeWriteSurvivesTunnelSendErrors(t *testing.T) { + t.Parallel() + d := New(Config{}) + t.Cleanup(func() { d.tunnels.Close() }) + const peer uint32 = 0xDADA0003 // no tunnel to it: every send fails + conn := d.ports.NewConnection(2003, protocol.Addr{Network: 0, Node: peer}, 80) + conn.State = StateEstablished + conn.LocalAddr = protocol.Addr{Network: 0, Node: 0x22222222} + conn.RemoteAddr = protocol.Addr{Network: 0, Node: peer} + conn.RemotePort = 80 + conn.SendSeq = 1000 + conn.CongWin = MaxCongWin + conn.PeerRecvWin = 1 << 20 + t.Cleanup(func() { d.ports.RemoveConnection(conn.ID) }) + ackEverything(t, conn) + + const size = 200000 + _ = d.SendData(conn, make([]byte, size)) // the last piece's error is reported + conn.Mu.Lock() + sent := int(conn.SendSeq - 1000) + conn.Mu.Unlock() + conn.NagleMu.Lock() + buffered := len(conn.NagleBuf) + conn.NagleMu.Unlock() + if sent+buffered != size { + t.Fatalf("%d bytes given sequence numbers and %d buffered, of a %d-byte write: the rest was dropped", sent, buffered, size) + } +} + +// A large write blocked on the window must not go on sending after the +// connection is closed: only the piece already in the buffer may follow. +func TestLargeWriteStopsWhenTheConnectionCloses(t *testing.T) { + t.Parallel() + d, pc, conn := newSendDataFixture(t, 0xDADA0004) + conn.NoDelay = false + conn.CongWin = InitialCongWin + + done := make(chan error, 1) + go func() { done <- d.SendData(conn, make([]byte, 4*nagleWritePiece)) }() + // The first piece fills the window and blocks. + readOneSegment(t, pc, 2*time.Second) + time.Sleep(50 * time.Millisecond) + conn.Mu.Lock() + conn.State = StateFinWait + conn.Mu.Unlock() + ackEverything(t, conn) + + select { + case err := <-done: + if err == nil { + t.Error("a write cut short by a close reported success") + } + case <-time.After(5 * time.Second): + t.Fatal("the write did not return after the close") + } + conn.Mu.Lock() + sent := int(conn.SendSeq - 1000) + conn.Mu.Unlock() + if sent > nagleWritePiece { + t.Fatalf("%d bytes sent after a close; at most the buffered piece (%d) may be", sent, nagleWritePiece) } }