diff --git a/CHANGELOG.md b/CHANGELOG.md index fd723e61..21dd2f63 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,14 @@ 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. + `--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 @@ -356,6 +364,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 418c9c80..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 { @@ -895,7 +902,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 +921,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] @@ -1628,7 +1639,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 - | --data-file [--type text|json|binary] [--count ] [--reuse-conn] [--wait ] pilotctl dgram --data pilotctl subscribe [--count ] [--timeout ] pilotctl publish --data @@ -2487,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{}{ @@ -4744,11 +4755,117 @@ 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 == "" { + 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. 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{}) { + 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, +// 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, refused := "", false + _, 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 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, 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) { 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) } @@ -4768,7 +4885,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) @@ -4812,6 +4929,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. @@ -4823,10 +4947,6 @@ func cmdSendMessage(args []string) { fatalCode("not_found", "%v", err) } - data := flagString(flags, "data", "") - if data == "" { - fatalCode("invalid_argument", "--data is required") - } msgType := flagString(flags, "type", "text") // First contact: the daemon holds no session with the peer yet, so this @@ -4903,6 +5023,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 { @@ -4974,11 +5097,31 @@ 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. 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 // 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(), @@ -5012,8 +5155,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) { @@ -5084,6 +5226,7 @@ func cmdSendMessage(args []string) { time.Sleep(50 * time.Millisecond) } } + failIfUndelivered(target.String(), results) outputOK(map[string]interface{}{ "target": target.String(), "to": target.String(), @@ -5104,6 +5247,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 new file mode 100644 index 00000000..3f9497ff --- /dev/null +++ b/cmd/pilotctl/zz_send_message_payload_test.go @@ -0,0 +1,110 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package main + +import ( + "os" + "path/filepath" + "strings" + "testing" + + "github.com/pilot-protocol/dataexchange" +) + +// 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) + } + empty := filepath.Join(t.TempDir(), "empty") + if err := os.WriteFile(empty, nil, 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"}, + {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 != "" { + 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)) + } + } +} + +// 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, 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 { + 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 _, 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 28adad23..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 [--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 0128113e..bbcf874f 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -4278,19 +4278,26 @@ 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 { - conn.Mu.Lock() - st := conn.State - conn.Mu.Unlock() - if st != StateEstablished { + if !connEstablished(conn) { + 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() + // The connection may have been closed while this writer queued. + if !connEstablished(conn) { return fmt.Errorf("connection not established") } @@ -4299,6 +4306,73 @@ func (d *Daemon) SendData(conn *Connection, data []byte) error { 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), + // 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], tailNow); err != nil { + 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. +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. 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, @@ -4312,9 +4386,7 @@ func (d *Daemon) SendData(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..aaef0d47 --- /dev/null +++ b/pkg/daemon/zz_large_write_pieces_test.go @@ -0,0 +1,203 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package daemon + +import ( + "bytes" + "sort" + "sync" + "testing" + "time" + + "github.com/pilot-protocol/common/protocol" +) + +// 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 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) + conn.NoDelay = false + 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) + } + 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) + } +} + +// 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) + } +} diff --git a/pkg/daemon/zz_nagle_buf_bug_test.go b/pkg/daemon/zz_nagle_buf_bug_test.go index a3a95a44..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{}) @@ -61,26 +41,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 +69,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 +83,64 @@ 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() + } + // 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: + 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") + } +}