From a27d431c6eb96deac987e814def0b1cce84de82f Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Thu, 1 Oct 2026 18:07:08 +0300 Subject: [PATCH 1/6] pilotctl: send-message --wait matches the reply by message ID --wait took the oldest new inbox message from the peer. Two concurrent requests to one peer, or anything else the peer sent in the window, could hand a caller another request's answer. send-message now sends every message through dataexchange's tagged path (Client.Send) with a new message ID, and the inbox watch uses it: - a message from the peer whose reply_to is one of our IDs is the reply; - a message whose reply_to names another request is never taken; - a message without reply_to is matched by sender and arrival time, as before, so peers that do not echo the ID (every responder today) keep working unchanged. The first-contact re-send gets an ID of its own, because the receiver drops a repeat of the same ID and bytes as a duplicate; the watch accepts a reply to either. --trace is covered: the TRACE wrapper is built here and carries the ID outside it. New flag --reply-to sends a message as the answer to a received one. The JSON result gains message_id, tagged and (with --reply-to) reply_to. Receivers that predate message IDs are reached through the library's fallback: the frame is re-sent in the old format on the same connection, and the result says "tagged": false. An "ERR ..." answer is still reported as the ack, and a send whose ack could not be read still succeeds, as before. No dependency change: dataexchange v0.2.3, already pinned, has the IDs. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 16 ++ cmd/pilotctl/firstcontact.go | 80 +++++- cmd/pilotctl/main.go | 99 ++++++-- cmd/pilotctl/zz_reply_match_test.go | 375 ++++++++++++++++++++++++++++ tests/zz_reply_correlation_test.go | 87 +++++++ 5 files changed, 626 insertions(+), 31 deletions(-) create mode 100644 cmd/pilotctl/zz_reply_match_test.go create mode 100644 tests/zz_reply_correlation_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index f4d54d24..a4e8bbce 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,22 @@ Detailed per-release notes are on the existing `SendTo` command is unchanged and still never replies, so current clients and SDKs are unaffected; `pilotctl dgram` switches to the confirmed send once the driver release that carries it is picked up. +- **`pilotctl send-message --wait` matches the reply by message ID.** The + wait took the oldest new message from the peer, so two concurrent requests + to one peer, or anything else the peer sent in the window, could hand a + caller another request's answer. Every message is now sent with a new + message ID (`message_id` in the `--json` result, and in the receiver's inbox + record), and a message from the peer whose `reply_to` is that ID is the + reply. A message whose `reply_to` names another request is never taken. A + peer that does not echo the ID is matched by sender and arrival time as + before, so nothing changes for it — that is every service responder today. + New flag `--reply-to ` sends a message as the answer to a received one. + A first-contact re-send uses an ID of its own (the receiver would drop a + repeat of the same ID as a duplicate) and a reply to either is accepted. + Receivers that predate message IDs still get the message, in the old format + (`"tagged": false` in the result); that costs one extra exchange on the same + connection. No dependency change: dataexchange v0.2.3 already carries the + IDs. - **The daemon caps its own log file.** launchd never rotates the daemon's `StandardOutPath`/`StandardErrorPath` (`~/.pilot/daemon.log`), which grew without bound — 22 MB on one laptop. When stderr is a regular file the diff --git a/cmd/pilotctl/firstcontact.go b/cmd/pilotctl/firstcontact.go index 75f6a531..e7512bc7 100644 --- a/cmd/pilotctl/firstcontact.go +++ b/cmd/pilotctl/firstcontact.go @@ -35,6 +35,7 @@ import ( "path/filepath" "sort" "strings" + "sync" "time" "github.com/pilot-protocol/common/driver" @@ -153,14 +154,28 @@ func dialFailureHint(target string, err error, attempts int, elapsed time.Durati // inboxWatch finds the reply to one request in the inbox. It snapshots // the files present before the request is sent, so a reply to an // earlier request (or the late duplicate some services send) that is -// already sitting in the inbox is never taken for this one, and it -// returns the oldest new file from the peer rather than whichever one -// the directory listing happens to put first. +// already sitting in the inbox is never taken for this one. +// +// Which new message is the reply: +// +// - Every request is sent with a message ID (addID). A message from the +// peer whose reply_to is one of our IDs is the reply, whatever else +// arrived before it. +// - A message whose reply_to names some other request is never taken: it +// answers another send-message --wait to the same peer. +// - A message without reply_to comes from a peer that does not echo the +// ID (every responder older than this, and old daemons, which drop the +// ID). The oldest such message from the peer is taken, as before IDs +// existed. Two concurrent requests to such a peer, or anything else it +// sends in the window, still cannot be told apart. type inboxWatch struct { dir string from string cutoff time.Time seen map[string]bool + + mu sync.Mutex + ids map[string]bool // message IDs of our request (and of its re-send) } func inboxDirPath() (string, error) { @@ -188,7 +203,49 @@ func newInboxWatch(from string) *inboxWatch { return w } -// poll returns the oldest reply that arrived since the snapshot. +// addID registers the message ID of a request this watch waits on. The +// re-sent request has an ID of its own; a reply to either one is ours. +func (w *inboxWatch) addID(id string) { + if id == "" { + return + } + w.mu.Lock() + defer w.mu.Unlock() + if w.ids == nil { + w.ids = map[string]bool{} + } + w.ids[id] = true +} + +// replyKind says how a message's reply_to relates to this watch's requests. +type replyKind int + +const ( + replyUntagged replyKind = iota // no reply_to: only sender and time to go by + replyOurs // reply_to is one of our message IDs + replyOther // reply_to names somebody else's request +) + +func (w *inboxWatch) classify(msg map[string]interface{}) replyKind { + replyTo, _ := msg["reply_to"].(string) + if replyTo == "" { + return replyUntagged + } + w.mu.Lock() + defer w.mu.Unlock() + if len(w.ids) == 0 { + // A caller that sent without an ID has nothing to compare with. + return replyUntagged + } + if w.ids[replyTo] { + return replyOurs + } + return replyOther +} + +// poll returns the reply that arrived since the snapshot: the message that +// names one of our requests in reply_to, else the oldest one from the peer +// that names none. func (w *inboxWatch) poll() (map[string]interface{}, error) { if w.dir == "" { dir, err := inboxDirPath() @@ -214,6 +271,7 @@ func (w *inboxWatch) poll() (map[string]interface{}, error) { } return ni[di:] < nj[dj:] }) + var untagged map[string]interface{} for _, e := range entries { if e.IsDir() || w.seen[e.Name()] { continue @@ -235,9 +293,19 @@ func (w *inboxWatch) poll() (map[string]interface{}, error) { continue } } - return msg, nil + switch w.classify(msg) { + case replyOurs: + return msg, nil + case replyOther: + // Another request's answer. It stays out of w.seen: an ID + // added later (the re-send) is checked against it again. + case replyUntagged: + if untagged == nil { + untagged = msg + } + } } - return nil, nil + return untagged, nil } // replyWait configures awaitReply. diff --git a/cmd/pilotctl/main.go b/cmd/pilotctl/main.go index f708c7e4..e4819f46 100644 --- a/cmd/pilotctl/main.go +++ b/cmd/pilotctl/main.go @@ -916,6 +916,13 @@ Flags: --timeout total wall time, including connect, handshake and reply --trace print per-step timing breakdown to stderr --no-auto-handshake skip automatic trust handshake with known agents + --reply-to mark this message as the answer to a received message: + is that message's message_id (pilotctl inbox), + and the sender's --wait matches the reply on it + +Every message is sent with a new message ID (message_id in the --json result). +--wait takes the message from the peer whose reply_to is that ID. A peer that +does not echo the ID is matched by sender and arrival time, as before. Examples: pilotctl send-message list-agents --data '/data {"search":"weather","limit":5}' @@ -4867,7 +4874,7 @@ func cmdSendMessage(args []string) { flags, pos := parseFlags(args) for name := range flags { switch name { - case "data", "data-file", "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", "reply-to": default: fatalCode("invalid_argument", "send-message: unknown flag --%s", name) } @@ -4887,7 +4894,14 @@ func cmdSendMessage(args []string) { defer timer.Stop() } if len(pos) != 1 { - fatalCode("invalid_argument", "usage: pilotctl send-message --data | --data - | --data-file [--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] [--reply-to ]") + } + // --reply-to marks this message as the answer to a message received + // earlier: the value is that message's message_id (see `pilotctl inbox`), + // and the sender's --wait matches on it. + replyTo := flagString(flags, "reply-to", "") + if _, given := flags["reply-to"]; given && !dataexchange.ValidMessageID(replyTo) { + fatalCode("invalid_argument", "--reply-to must be a message ID: 1-%d characters of A-Z a-z 0-9 . _ : -", dataexchange.MaxMessageIDLen) } sendCount := flagInt(flags, "count", 1) @@ -4988,39 +5002,53 @@ func cmdSendMessage(args []string) { return c } + // Set below when --wait is given. Declared here so sendOne can register + // each request's message ID with it before the request leaves. + var watch *inboxWatch + // sendOne sends one message on cl and returns timing/ack metadata. // reused=true is recorded when the connection was dialled on a prior call. + // + // Every send carries a new message ID, so the receiver can tell a + // re-delivery from a new message and a reply can name the request it + // answers (reply_to). Client.Send delivers the message without the ID to + // a receiver too old to know it: such a receiver stores nothing for the + // tagged frame, says so, and gets the plain frame on the same connection. sendOne := func(cl *dataexchange.Client, seq int, reused bool) map[string]interface{} { - var sentAtNs int64 - var sendErr error - if traceTime { - sentAtNs, sendErr = cl.SendTrace(innerType, []byte(data)) - } else { - sendStart := time.Now() - switch msgType { - case "text": - sendErr = cl.SendText(data) - case "json": - sendErr = cl.SendJSON([]byte(data)) - case "binary": - sendErr = cl.SendBinary([]byte(data)) - } - sentAtNs = sendStart.UnixNano() + messageID := dataexchange.NewMessageID() + if watch != nil { + watch.addID(messageID) } - if sendErr != nil { - return map[string]interface{}{"seq": seq, "error": sendErr.Error()} - } - - ack, ackErr := cl.Recv() + sentAtNs := time.Now().UnixNano() + res, sendErr := cl.Send(messageFrame(innerType, []byte(data), messageID, replyTo, traceTime, sentAtNs)) ackRecvAtNs := time.Now().UnixNano() - if ackErr != nil { - slog.Debug("send-message ACK read failed", "err", ackErr) + var ack *dataexchange.Frame + var ackErr error + switch { + case res != nil: + // Delivered, or answered "ERR ..." (ErrRejected): either way the + // receiver's answer is the ack. + ack = res.Ack + case isAckReadError(sendErr): + // The message was written; only the ack did not come back. + slog.Debug("send-message ACK read failed", "err", sendErr) + ackErr = sendErr + default: + return map[string]interface{}{"seq": seq, "error": sendErr.Error(), "message_id": messageID} } r := map[string]interface{}{ "seq": seq, "bytes": len(data), "reused": reused, + // message_id is the ID this message was sent with. tagged says + // whether the receiver got it: false for a receiver too old to + // know message IDs, which can then not be asked to echo it. + "message_id": messageID, + "tagged": res != nil && res.Tagged, + } + if replyTo != "" { + r["reply_to"] = replyTo } if ack != nil { r["ack"] = string(ack.Payload) @@ -5060,7 +5088,6 @@ func cmdSendMessage(args []string) { agentHint := target.String() // Snapshot the inbox before the send so --wait only takes a reply that // arrives after it (never one already sitting there). - var watch *inboxWatch if waitDur > 0 { watch = newInboxWatch(agentHint) } @@ -5262,6 +5289,28 @@ func cmdSendMessage(args []string) { maybePromptPilotReview() } +// messageFrame builds the data-exchange frame for one send-message. With +// trace it is the TypeTrace wrapper (inner type, send time, payload) the +// receiver answers with a timing ACK; the message ID and reply-to ride +// outside it either way. +func messageFrame(innerType uint32, data []byte, messageID, replyTo string, trace bool, sentAtNs int64) *dataexchange.Frame { + f := &dataexchange.Frame{Type: innerType, Payload: data, MessageID: messageID, ReplyTo: replyTo} + if trace { + buf := make([]byte, 12+len(data)) + binary.BigEndian.PutUint32(buf[0:4], innerType) + binary.BigEndian.PutUint64(buf[4:12], uint64(sentAtNs)) // #nosec G115 -- a wall-clock nanosecond count is positive + copy(buf[12:], data) + f.Type, f.Payload = dataexchange.TypeTrace, buf + } + return f +} + +// isAckReadError reports whether a Client.Send error means the frame was +// written and only the receiver's acknowledgement could not be read. +func isAckReadError(err error) bool { + return err != nil && strings.HasPrefix(err.Error(), "dataexchange: read ack:") +} + // maybePromptPilotReview occasionally prints a Pilot review nudge to stderr // after a send-message completes. Gated by: // - reviews consent (default on; no-op when off) diff --git a/cmd/pilotctl/zz_reply_match_test.go b/cmd/pilotctl/zz_reply_match_test.go new file mode 100644 index 00000000..ca17bc06 --- /dev/null +++ b/cmd/pilotctl/zz_reply_match_test.go @@ -0,0 +1,375 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package main + +import ( + "bytes" + "encoding/binary" + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "testing" + + "github.com/pilot-protocol/dataexchange" +) + +func writeInboxRecord(t *testing.T, dir, name string, rec map[string]any) { + t.Helper() + body, _ := json.Marshal(rec) + if err := os.WriteFile(filepath.Join(dir, name), body, 0o600); err != nil { + t.Fatal(err) + } +} + +const replyPeer = "0:0000.0002.BBE4" + +// Two `send-message --wait` to the same peer are in flight (A and B). The +// peer's answer to B lands first and names B's request in reply_to. A used +// to take it as its own reply. +func TestInboxWatchLeavesAnotherRequestsReply(t *testing.T) { + dir := tempInbox(t) + watchA := newInboxWatch(replyPeer) + watchA.addID("request-a") + watchB := newInboxWatch(replyPeer) + watchB.addID("request-b") + + writeInboxRecord(t, dir, "TEXT-20260924-100005.000-000001.json", map[string]any{ + "from": replyPeer, "data": "answer to B", "reply_to": "request-b", + }) + gotB, err := watchB.poll() + if err != nil || gotB == nil || gotB["data"] != "answer to B" { + t.Fatalf("B's watch = %v, %v; want B's answer", gotB, err) + } + if gotA, err := watchA.poll(); err != nil || gotA != nil { + t.Fatalf("A's watch = %v, %v: the answer to another request was handed to A", gotA, err) + } + + writeInboxRecord(t, dir, "TEXT-20260924-100006.000-000002.json", map[string]any{ + "from": replyPeer, "data": "answer to A", "reply_to": "request-a", + }) + gotA, err := watchA.poll() + if err != nil || gotA == nil || gotA["data"] != "answer to A" { + t.Fatalf("A's watch = %v, %v; want A's answer", gotA, err) + } +} + +// The reply that names our request wins over an older message from the peer +// that names none (here a request of the peer's own). +func TestInboxWatchPrefersTheReplyNamingOurRequest(t *testing.T) { + dir := tempInbox(t) + watch := newInboxWatch(replyPeer) + watch.addID("our-request") + + writeInboxRecord(t, dir, "TEXT-20260924-100005.000-000001.json", map[string]any{ + "from": replyPeer, "data": "/data {\"unrelated\":\"request from the peer\"}", "message_id": "peers-own-request", + }) + writeInboxRecord(t, dir, "TEXT-20260924-100006.000-000002.json", map[string]any{ + "from": replyPeer, "data": "the actual answer", "reply_to": "our-request", + }) + got, err := watch.poll() + if err != nil || got == nil || got["data"] != "the actual answer" { + t.Fatalf("poll = %v, %v; want the reply, not the peer's unrelated message", got, err) + } +} + +// A peer that does not echo the ID (every responder so far) is matched as +// before: the oldest new message from it, whether or not it carries a +// message_id of its own. +func TestInboxWatchFallsBackToSenderAndTime(t *testing.T) { + dir := tempInbox(t) + watch := newInboxWatch(replyPeer) + watch.addID("our-request") + + writeInboxRecord(t, dir, "TEXT-20260924-100007.000-000003.json", map[string]any{"from": replyPeer, "data": "second"}) + writeInboxRecord(t, dir, "TEXT-20260924-100006.000-000002.json", map[string]any{"from": replyPeer, "data": "first", "message_id": "its-own-id"}) + writeInboxRecord(t, dir, "TEXT-20260924-100005.000-000001.json", map[string]any{"from": "0:0000.0009.0001", "data": "other sender"}) + writeInboxRecord(t, dir, "TEXT-20260924-100004.000-000000.json", map[string]any{"from": replyPeer, "data": "for someone else", "reply_to": "not-ours"}) + got, err := watch.poll() + if err != nil || got == nil || got["data"] != "first" { + t.Fatalf("poll = %v, %v; want the oldest message from the peer that names no request", got, err) + } +} + +// The re-sent request has an ID of its own; a reply to either is the reply, +// including one that arrived before the re-send's ID was registered. +func TestInboxWatchTakesReplyToEitherSend(t *testing.T) { + dir := tempInbox(t) + watch := newInboxWatch(replyPeer) + watch.addID("first-send") + + writeInboxRecord(t, dir, "TEXT-20260924-100005.000-000001.json", map[string]any{ + "from": replyPeer, "data": "answer to the re-send", "reply_to": "second-send", + }) + if got, err := watch.poll(); err != nil || got != nil { + t.Fatalf("poll before the re-send's ID is known = %v, %v; want nothing", got, err) + } + watch.addID("second-send") + got, err := watch.poll() + if err != nil || got == nil || got["data"] != "answer to the re-send" { + t.Fatalf("poll = %v, %v; want the answer to the re-sent request", got, err) + } +} + +// A watch that was given no ID has nothing to compare reply_to with and +// keeps the sender-and-time rule for every message. +func TestInboxWatchWithoutIDsIgnoresReplyTo(t *testing.T) { + dir := tempInbox(t) + watch := newInboxWatch(replyPeer) + writeInboxRecord(t, dir, "TEXT-20260924-100005.000-000001.json", map[string]any{ + "from": replyPeer, "data": "tagged", "reply_to": "whatever", + }) + got, err := watch.poll() + if err != nil || got == nil || got["data"] != "tagged" { + t.Fatalf("poll = %v, %v", got, err) + } +} + +// ── send-message against receivers of each generation ──────────────────── + +// dxReceiver turns the stream daemon's echo into a data-exchange receiver: +// answer is called with each complete wire frame (its type and payload) the +// client wrote and returns the acknowledgement text. +type dxReceiver struct { + mu sync.Mutex + buf map[uint32][]byte + types []uint32 + frames []*dataexchange.Frame // decoded with the current library +} + +func newDXReceiver(sd *streamDaemon, answer func(ftype uint32, payload []byte, decoded *dataexchange.Frame) string) *dxReceiver { + r := &dxReceiver{buf: map[uint32][]byte{}} + sd.on(tdCmdSend, func(frame []byte) [][]byte { + sd.sendCount.Add(1) + if len(frame) < 5 { + return nil + } + connID := binary.BigEndian.Uint32(frame[1:5]) + r.mu.Lock() + defer r.mu.Unlock() + r.buf[connID] = append(r.buf[connID], frame[5:]...) + var out [][]byte + for { + b := r.buf[connID] + if len(b) < 8 { + break + } + n := int(binary.BigEndian.Uint32(b[4:8])) + if len(b) < 8+n { + break + } + ftype, payload := binary.BigEndian.Uint32(b[0:4]), append([]byte(nil), b[8:8+n]...) + decoded, _ := dataexchange.ReadFrame(bytes.NewReader(b[:8+n])) + r.buf[connID] = b[8+n:] + r.types = append(r.types, ftype) + r.frames = append(r.frames, decoded) + var ack bytes.Buffer + _ = dataexchange.WriteFrame(&ack, &dataexchange.Frame{Type: dataexchange.TypeText, Payload: []byte(answer(ftype, payload, decoded))}) + push := make([]byte, 5+ack.Len()) + push[0] = tdCmdRecv + binary.BigEndian.PutUint32(push[1:5], connID) + copy(push[5:], ack.Bytes()) + out = append(out, push) + } + return out + }) + return r +} + +func (r *dxReceiver) seen() ([]uint32, []*dataexchange.Frame) { + r.mu.Lock() + defer r.mu.Unlock() + return append([]uint32(nil), r.types...), append([]*dataexchange.Frame(nil), r.frames...) +} + +func sendMessageJSON(t *testing.T, args ...string) map[string]interface{} { + t.Helper() + out := captureStdout(t, func() { withJSON(func() { cmdSendMessage(args) }) }) + var env map[string]interface{} + if err := json.Unmarshal([]byte(out), &env); err != nil { + t.Fatalf("json: %v\n%s", err, out) + } + data, ok := env["data"].(map[string]interface{}) + if !ok { + t.Fatalf("no data object: %s", out) + } + return data +} + +// A current receiver gets the message with its ID, and --reply-to is carried +// as reply_to. The JSON result reports the ID that was sent. +func TestSendMessageTagsEveryMessage(t *testing.T) { + sd := newStreamDaemon(t) + sd.useDaemonNoRegistry(t) + rcv := newDXReceiver(sd, func(_ uint32, _ []byte, f *dataexchange.Frame) string { + return fmt.Sprintf("ACK %s %d bytes", dataexchange.TypeName(f.Type), len(f.Payload)) + }) + + data := sendMessageJSON(t, "0:0000.0000.002A", "--data", "hello", "--reply-to", "req-42") + types, frames := rcv.seen() + if len(types) != 1 || types[0] != dataexchange.TypeTagged { + t.Fatalf("receiver saw frame types %v, want one TAGGED frame", types) + } + f := frames[0] + if f == nil || f.Type != dataexchange.TypeText || string(f.Payload) != "hello" { + t.Fatalf("receiver decoded %+v", f) + } + if !dataexchange.ValidMessageID(f.MessageID) || f.ReplyTo != "req-42" { + t.Fatalf("frame message_id=%q reply_to=%q", f.MessageID, f.ReplyTo) + } + if data["message_id"] != f.MessageID || data["tagged"] != true || data["reply_to"] != "req-42" || data["ack"] != "ACK TEXT 5 bytes" { + t.Fatalf("result = %v; frame message_id %q", data, f.MessageID) + } +} + +// Each of several sends has an ID of its own: the receiver suppresses a +// second delivery of the same ID and bytes as a duplicate. +func TestSendMessageCountUsesAnIDPerSend(t *testing.T) { + sd := newStreamDaemon(t) + sd.useDaemonNoRegistry(t) + newDXReceiver(sd, func(uint32, []byte, *dataexchange.Frame) string { return "ACK TEXT 1 bytes" }) + multi := sendMessageJSON(t, "0:0000.0000.002A", "--data", "x", "--count", "3") + ids := map[interface{}]bool{} + for _, r := range multi["results"].([]interface{}) { + ids[r.(map[string]interface{})["message_id"]] = true + } + if len(ids) != 3 || ids[nil] { + t.Fatalf("--count 3 used message IDs %v, want three different ones", ids) + } +} + +// --trace keeps its TRACE wrapper and carries the ID outside it. +func TestSendMessageTraceIsTagged(t *testing.T) { + sd := newStreamDaemon(t) + sd.useDaemonNoRegistry(t) + rcv := newDXReceiver(sd, func(uint32, []byte, *dataexchange.Frame) string { return "ACK TEXT 5 bytes" }) + + data := sendMessageJSON(t, "0:0000.0000.002A", "--data", "hello", "--trace") + types, frames := rcv.seen() + if len(types) != 1 || types[0] != dataexchange.TypeTagged || frames[0] == nil || frames[0].Type != dataexchange.TypeTrace { + t.Fatalf("receiver saw types %v frame %+v, want one TAGGED frame wrapping TRACE", types, frames) + } + tf, err := dataexchange.ReadTracePayload(frames[0]) + if err != nil || tf.InnerType != dataexchange.TypeText || string(tf.Payload) != "hello" || tf.SentAtNs <= 0 { + t.Fatalf("trace payload = %+v, %v", tf, err) + } + if data["message_id"] != frames[0].MessageID || data["tagged"] != true { + t.Fatalf("result = %v; frame message_id %q", data, frames[0].MessageID) + } +} + +// A receiver that predates message IDs stores nothing for the tagged frame +// and says so in one of two ways, depending on its age. The message then +// reaches it in the old format, once, and the result says the ID did not get +// through. +func TestSendMessageReachesReceiversThatPredateMessageIDs(t *testing.T) { + for name, unknownAck := range map[string]string{ + "dataexchange up to v0.2.1 (ACK for a frame it dropped)": "ACK UNKNOWN(10) %d bytes", + "dataexchange v0.2.2 (rejects unknown types)": "ERR UNKNOWN(10) save failed: unsupported frame type 10", + } { + t.Run(name, func(t *testing.T) { + sd := newStreamDaemon(t) + sd.useDaemonNoRegistry(t) + var stored []string + rcv := newDXReceiver(sd, func(ftype uint32, payload []byte, _ *dataexchange.Frame) string { + if ftype != dataexchange.TypeText { + if bytes.Contains([]byte(unknownAck), []byte("%d")) { + return fmt.Sprintf(unknownAck, len(payload)) + } + return unknownAck + } + stored = append(stored, string(payload)) + return fmt.Sprintf("ACK TEXT %d bytes", len(payload)) + }) + + data := sendMessageJSON(t, "0:0000.0000.002A", "--data", "hello") + types, _ := rcv.seen() + if len(types) != 2 || types[0] != dataexchange.TypeTagged || types[1] != dataexchange.TypeText { + t.Fatalf("old receiver saw frame types %v, want [TAGGED TEXT]", types) + } + if len(stored) != 1 || stored[0] != "hello" { + t.Fatalf("old receiver stored %q, want exactly [hello]", stored) + } + if data["ack"] != "ACK TEXT 5 bytes" || data["tagged"] != false { + t.Fatalf("result = %v, want the plain frame's ack and tagged=false", data) + } + if id, _ := data["message_id"].(string); !dataexchange.ValidMessageID(id) { + t.Fatalf("result has no message_id: %v", data) + } + }) + } +} + +// A receiver that could not store the message answers "ERR ...". The message +// is sent once (the fallback is only for receivers that do not know the +// frame), and the answer reaches the caller as it did when the message was +// sent untagged: as the ack, or, once send-message treats an "ERR " ack as a +// failed send, as that error. +func TestSendMessageReportsReceiverRejection(t *testing.T) { + sd := newStreamDaemon(t) + sd.useDaemonNoRegistry(t) + const rejection = "ERR TEXT save failed: no space left on device" + rcv := newDXReceiver(sd, func(uint32, []byte, *dataexchange.Frame) string { return rejection }) + + var stdout string + var failure *trappedFatal + withJSON(func() { + stdout, _, failure = runTrapped(t, func() { cmdSendMessage([]string{"0:0000.0000.002A", "--data", "hello"}) }) + }) + if types, _ := rcv.seen(); len(types) != 1 { + t.Fatalf("a rejected message was sent %d times, want once", len(types)) + } + if failure != nil { + if !strings.Contains(failure.Message, rejection) { + t.Fatalf("send failed with %+v, want the receiver's rejection", failure) + } + return + } + if !strings.Contains(stdout, `"ack":"`+rejection+`"`) { + t.Fatalf("result = %s, want the rejection as the ack", stdout) + } +} + +// End to end: the peer answers two requests at once and echoes each +// request's ID. The answer to the other request arrives first; --wait +// returns the one that names this request. +func TestSendMessageWaitTakesTheReplyToItsOwnRequest(t *testing.T) { + sd := newStreamDaemon(t) + home := sd.useDaemonNoRegistry(t) + inbox := filepath.Join(home, ".pilot", "inbox") + if err := os.MkdirAll(inbox, 0o700); err != nil { + t.Fatal(err) + } + const peer = "0:0000.0000.002A" + newDXReceiver(sd, func(_ uint32, _ []byte, f *dataexchange.Frame) string { + writeInboxRecord(t, inbox, "TEXT-20260924-100005.000-000001.json", map[string]any{ + "from": peer, "data": "answer to somebody else's request", "reply_to": "somebody-else", + }) + writeInboxRecord(t, inbox, "TEXT-20260924-100006.000-000002.json", map[string]any{ + "from": peer, "data": "pong", "reply_to": f.MessageID, + }) + return "ACK TEXT 4 bytes" + }) + + data := sendMessageJSON(t, peer, "--data", "ping", "--wait", "5s") + reply, ok := data["reply"].(map[string]interface{}) + if !ok { + t.Fatalf("no reply in %v", data) + } + if reply["data"] != "pong" || reply["reply_to"] != data["message_id"] { + t.Fatalf("reply = %v for message_id %v; want the answer to this request", reply, data["message_id"]) + } +} + +func TestSendMessageRejectsInvalidReplyTo(t *testing.T) { + sd := newStreamDaemon(t) + sd.useDaemonNoRegistry(t) + for _, bad := range []string{"has space", "semi;colon", ""} { + _, _, f := runTrapped(t, func() { cmdSendMessage([]string{"0:0000.0000.002A", "--data", "x", "--reply-to=" + bad}) }) + if f == nil || f.Code != "invalid_argument" { + t.Errorf("--reply-to %q: %+v, want invalid_argument", bad, f) + } + } +} diff --git a/tests/zz_reply_correlation_test.go b/tests/zz_reply_correlation_test.go new file mode 100644 index 00000000..3006aa5f --- /dev/null +++ b/tests/zz_reply_correlation_test.go @@ -0,0 +1,87 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +//go:build nightly + +package tests + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" + "time" + + "github.com/pilot-protocol/dataexchange" +) + +// inboxRecordWith returns the inbox record whose key has the given value, +// waiting for the daemon to write it. +func inboxRecordWith(t *testing.T, dir, key, value string) map[string]any { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for { + entries, _ := os.ReadDir(dir) + for _, e := range entries { + raw, err := os.ReadFile(filepath.Join(dir, e.Name())) + if err != nil { + continue + } + var rec map[string]any + if json.Unmarshal(raw, &rec) == nil && rec[key] == value { + return rec + } + } + if time.Now().After(deadline) { + t.Fatalf("no inbox record with %s=%q in %s (%d files)", key, value, dir, len(entries)) + } + time.Sleep(50 * time.Millisecond) + } +} + +// What `pilotctl send-message --wait` relies on, between two real daemons: a +// request sent with a message ID is stored with that message_id, and an +// answer sent with reply_to is stored with that reply_to, so the requester +// can pick its own answer out of the inbox. +func TestDataExchangeRecordsMessageIDAndReplyTo(t *testing.T) { + requireRealNetwork(t) + // Both daemons run in this process and write to $HOME/.pilot/inbox. + home := t.TempDir() + t.Setenv("HOME", home) + inbox := filepath.Join(home, ".pilot", "inbox") + + env := NewTestEnv(t) + a := env.AddDaemon() + b := env.AddDaemon() + + request, err := dataexchange.Dial(b.Driver, a.Daemon.Addr()) + if err != nil { + t.Fatalf("dial a: %v", err) + } + defer request.Close() + id := dataexchange.NewMessageID() + res, err := request.Send(&dataexchange.Frame{Type: dataexchange.TypeText, Payload: []byte("ping"), MessageID: id}) + if err != nil || !res.Tagged { + t.Fatalf("send request: %+v, %v; want it delivered with its ID", res, err) + } + got := inboxRecordWith(t, inbox, "message_id", id) + if got["data"] != "ping" || got["from"] != b.Daemon.Addr().String() { + t.Fatalf("request record = %v", got) + } + if _, has := got["reply_to"]; has { + t.Fatalf("request record carries a reply_to: %v", got) + } + + answer, err := dataexchange.Dial(a.Driver, b.Daemon.Addr()) + if err != nil { + t.Fatalf("dial b: %v", err) + } + defer answer.Close() + res, err = answer.Send(&dataexchange.Frame{Type: dataexchange.TypeText, Payload: []byte("pong"), ReplyTo: id}) + if err != nil || !res.Tagged { + t.Fatalf("send answer: %+v, %v", res, err) + } + got = inboxRecordWith(t, inbox, "reply_to", id) + if got["data"] != "pong" || got["from"] != a.Daemon.Addr().String() { + t.Fatalf("answer record = %v", got) + } +} From d52f377c0d39cc9d2b45b0807634ce848b090f1b Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Wed, 7 Oct 2026 16:46:41 +0300 Subject: [PATCH 2/6] pilotctl: hold an untagged reply for a tagged one; safer --reply-to; inbox shows IDs send-message --wait took the first untagged message from the peer at once, so a reply naming our request that arrived a poll later lost to another client's untagged answer. When our request reached the receiver with its ID, an untagged message is now held for 0.75 s (bounded by the wait) in case our tagged reply follows, and once the peer has been seen naming another request in reply_to, only a reply naming ours is taken. --reply-to refuses a bare flag (parsed as "true") and maps an inbox file id to the message_id stored in that record, refusing it when the record has none. pilotctl inbox shows message_id and reply_to in the listing, --json, and inbox read; the help says which field to use. A lost ack on the tagged first attempt is retried once on a new connection with the same frame and ID: a current receiver keeps one copy, one through v1.13.9 gets the untagged copy. --no-resend opts out. tagged is reported only when the receiver's ack was read. CHANGELOG no longer claims the concurrency bug is fixed for responders that do not echo reply_to. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 38 +-- cmd/pilotctl/firstcontact.go | 85 ++++++- cmd/pilotctl/main.go | 179 +++++++++++-- cmd/pilotctl/zz_reply_match_test.go | 378 +++++++++++++++++++++++++++- 4 files changed, 633 insertions(+), 47 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a4e8bbce..6995774f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,22 +29,32 @@ Detailed per-release notes are on the existing `SendTo` command is unchanged and still never replies, so current clients and SDKs are unaffected; `pilotctl dgram` switches to the confirmed send once the driver release that carries it is picked up. -- **`pilotctl send-message --wait` matches the reply by message ID.** The - wait took the oldest new message from the peer, so two concurrent requests - to one peer, or anything else the peer sent in the window, could hand a - caller another request's answer. Every message is now sent with a new - message ID (`message_id` in the `--json` result, and in the receiver's inbox - record), and a message from the peer whose `reply_to` is that ID is the - reply. A message whose `reply_to` names another request is never taken. A - peer that does not echo the ID is matched by sender and arrival time as - before, so nothing changes for it — that is every service responder today. - New flag `--reply-to ` sends a message as the answer to a received one. +- **`pilotctl send-message --wait` can match the reply by message ID.** The + wait takes the oldest new message from the peer, so two concurrent + requests to one peer, or anything else the peer sends in the window, can + hand a caller another request's answer. Every message is now sent with a + new message ID (`message_id` in the `--json` result, and in the receiver's + inbox record). A reply whose `reply_to` is that ID is matched exactly, and + a message whose `reply_to` names another request is never taken. Exact + matching needs the responder to echo the request's `message_id` as + `reply_to`, which no service responder does yet: an untagged reply still + works and is matched by sender and arrival time as before, so for such + responders concurrent waits can still be crossed. An untagged message is + held back up to 0.75 s in case the reply naming the request follows, which + delays replies from those responders by that much; once the peer has been + seen naming another request in `reply_to`, only a reply naming ours is + taken. New flag `--reply-to ` sends a message as the answer to + a received one (an inbox file id is looked up; a bare flag is refused). + `pilotctl inbox` now shows each message's `message_id` and `reply_to`, in + the listing and with `--json` (new fields; `id` is still the file name). A first-contact re-send uses an ID of its own (the receiver would drop a repeat of the same ID as a duplicate) and a reply to either is accepted. - Receivers that predate message IDs still get the message, in the old format - (`"tagged": false` in the result); that costs one extra exchange on the same - connection. No dependency change: dataexchange v0.2.3 already carries the - IDs. + When the ack is lost, the message is sent once more on a new connection + with the same ID, which a current receiver keeps only once (`"retried": + true`; `--no-resend` opts out). Receivers that predate message IDs still + get the message, in the old format (`"tagged": false` in the result); that + costs one extra exchange on the same connection. No dependency change: the + dataexchange release already required carries the IDs. - **The daemon caps its own log file.** launchd never rotates the daemon's `StandardOutPath`/`StandardErrorPath` (`~/.pilot/daemon.log`), which grew without bound — 22 MB on one laptop. When stderr is a regular file the diff --git a/cmd/pilotctl/firstcontact.go b/cmd/pilotctl/firstcontact.go index e7512bc7..8eb40226 100644 --- a/cmd/pilotctl/firstcontact.go +++ b/cmd/pilotctl/firstcontact.go @@ -168,16 +168,38 @@ func dialFailureHint(target string, err error, attempts int, elapsed time.Durati // ID). The oldest such message from the peer is taken, as before IDs // existed. Two concurrent requests to such a peer, or anything else it // sends in the window, still cannot be told apart. +// - When our request reached the receiver with its ID (markTagged), an +// untagged message can also be the answer to someone else's untagged +// request to a peer that does echo IDs, with our own tagged reply still +// on its way. So it is held for untaggedReplyGrace first, and taken only +// if no reply naming our request arrives meanwhile (or when the wait +// ends). Once the peer has been seen naming another request in +// reply_to during this wait, it evidently echoes IDs, and only a reply +// naming ours is taken. type inboxWatch struct { dir string from string cutoff time.Time seen map[string]bool - mu sync.Mutex - ids map[string]bool // message IDs of our request (and of its re-send) + mu sync.Mutex + ids map[string]bool // message IDs of our request (and of its re-send) + tagged bool // a request of ours reached the receiver with its ID + + // Used only by the goroutine that polls. + peerEchoes bool // the peer named another request in reply_to + held map[string]interface{} // untagged message held back (see untaggedReply) + heldSince time.Time } +// untaggedReplyGrace is how long an untagged message from the peer is held +// back, after our request reached the receiver with its ID, in case the +// reply that names our request follows. A peer that does not echo IDs +// (every service responder today) has its reply delayed by this much; a +// peer that echoes them is not delayed at all. The send-message help +// quotes this value. +const untaggedReplyGrace = 750 * time.Millisecond + func inboxDirPath() (string, error) { home, err := os.UserHomeDir() if err != nil { @@ -217,6 +239,16 @@ func (w *inboxWatch) addID(id string) { w.ids[id] = true } +// markTagged records that a request reached the receiver with its message +// ID (SendResult.Tagged), so the peer can name it in reply_to. Until then +// an untagged message is taken at once: a receiver that predates IDs +// dropped ours, and its peer can never name our request. +func (w *inboxWatch) markTagged() { + w.mu.Lock() + defer w.mu.Unlock() + w.tagged = true +} + // replyKind says how a message's reply_to relates to this watch's requests. type replyKind int @@ -245,7 +277,7 @@ func (w *inboxWatch) classify(msg map[string]interface{}) replyKind { // poll returns the reply that arrived since the snapshot: the message that // names one of our requests in reply_to, else the oldest one from the peer -// that names none. +// that names none, subject to untaggedReply. func (w *inboxWatch) poll() (map[string]interface{}, error) { if w.dir == "" { dir, err := inboxDirPath() @@ -299,13 +331,51 @@ func (w *inboxWatch) poll() (map[string]interface{}, error) { case replyOther: // Another request's answer. It stays out of w.seen: an ID // added later (the re-send) is checked against it again. + w.peerEchoes = true case replyUntagged: if untagged == nil { untagged = msg } } } - return untagged, nil + return w.untaggedReply(untagged, time.Now()), nil +} + +// untaggedReply decides whether msg, the oldest untagged message from the +// peer (nil if there is none), is taken as the reply now. It is when our +// request went out without its ID. Otherwise it is held back for +// untaggedReplyGrace from the poll that first saw it, so a reply naming +// our request that arrives meanwhile wins; and it is never taken from a +// peer seen echoing IDs. +func (w *inboxWatch) untaggedReply(msg map[string]interface{}, now time.Time) map[string]interface{} { + if msg == nil { + w.held = nil + return nil + } + w.mu.Lock() + tagged := w.tagged + w.mu.Unlock() + if !tagged { + return msg + } + if w.held == nil { + w.heldSince = now + } + w.held = msg + if w.peerEchoes || now.Sub(w.heldSince) < untaggedReplyGrace { + return nil + } + return msg +} + +// heldReply returns the untagged message untaggedReply is holding back, if +// it may still be taken. awaitReply takes it when the wait runs out before +// its grace period does, rather than report no reply at all. +func (w *inboxWatch) heldReply() map[string]interface{} { + if w.peerEchoes { + return nil + } + return w.held } // replyWait configures awaitReply. @@ -376,7 +446,9 @@ func awaitReply(w *inboxWatch, ackAt time.Time, cfg replyWait) (replyOutcome, er default: } } - if !out.resent && cfg.resend != nil && cfg.resendAfter > 0 && now.Sub(ackAt) >= cfg.resendAfter && now.Before(deadline) { + // No re-send while an untagged message is held back: something did + // arrive, and it is most likely the reply. + if !out.resent && cfg.resend != nil && cfg.resendAfter > 0 && now.Sub(ackAt) >= cfg.resendAfter && now.Before(deadline) && w.heldReply() == nil { out.resent = true resendCh = make(chan resendResult, 1) go func(ch chan<- resendResult) { @@ -385,6 +457,9 @@ func awaitReply(w *inboxWatch, ackAt time.Time, cfg replyWait) (replyOutcome, er }(resendCh) } if now.After(deadline) && resendCh == nil { + // The grace period for a held untagged message is bounded by + // the wait: when the wait ends first, it is the reply. + out.reply = w.heldReply() out.waited = time.Since(ackAt) return out, nil } diff --git a/cmd/pilotctl/main.go b/cmd/pilotctl/main.go index e4819f46..9358b846 100644 --- a/cmd/pilotctl/main.go +++ b/cmd/pilotctl/main.go @@ -16,6 +16,7 @@ import ( "os" "os/exec" "path/filepath" + "regexp" "runtime" "sort" "strconv" @@ -910,19 +911,24 @@ Flags: --reuse-conn reuse the connection across --count sends (saves ~1 RTT) --wait [] wait for a reply in the inbox (default timeout: 30s, counted from the receiver's acknowledgement) - --no-resend on first contact, never send the request a second - time (by default a request with no reply by mid-wait - is re-sent once on a new stream) + --no-resend never send the request a second time (by default it + is re-sent once on a new stream when, on first + contact, no reply came by mid-wait, or when its + acknowledgement was lost) --timeout total wall time, including connect, handshake and reply --trace print per-step timing breakdown to stderr --no-auto-handshake skip automatic trust handshake with known agents --reply-to mark this message as the answer to a received message: - is that message's message_id (pilotctl inbox), - and the sender's --wait matches the reply on it + is that message's message_id, shown by + pilotctl inbox (an inbox id such as + TEXT-20260924-100005.000-000001 is looked up), and + the sender's --wait matches the reply on it Every message is sent with a new message ID (message_id in the --json result). ---wait takes the message from the peer whose reply_to is that ID. A peer that -does not echo the ID is matched by sender and arrival time, as before. +--wait takes the message from the peer whose reply_to is that ID, and never one +whose reply_to names another message. A peer that does not echo the ID is +matched by sender and arrival time, as before; its reply is held back up to +0.75s in case a reply naming the ID follows. Examples: pilotctl send-message list-agents --data '/data {"search":"weather","limit":5}' @@ -1042,6 +1048,11 @@ Agent patterns: pilotctl --json inbox --latest # newest reply, full body pilotctl --json inbox --from list-agents --limit 3 # last 3 from one peer pilotctl inbox --clear --before 24h # keep today, purge older + pilotctl send-message --data "..." --reply-to + # answer one message + +A message sent with an ID shows it as message_id (the id field is the inbox +file name, not the ID), and an answer shows the ID it names as reply_to. Tip: send-message --wait already returns the matching reply inline; the inbox is for replies that arrive later or that you want to re-read. @@ -2507,9 +2518,9 @@ func contextCatalog() map[string]interface{} { // Messaging "send-message": map[string]interface{}{ - "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", + "args": []string{"", "--data | --data - | --data-file ", "[--type text|json|binary]", "[--count ]", "[--reuse-conn]", "[--wait ]", "[--timeout ]", "[--reply-to ]"}, + "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. Every message carries a new message_id; --wait returns the reply whose reply_to names it (or, from a peer that does not echo IDs, the next message from the peer); --reply-to answers a received message", + "returns": "target, to, type, bytes, ack, reuse_conn, message_id, tagged, reply_to, reply", }, "send-file": map[string]interface{}{ "args": []string{"", ""}, @@ -2519,7 +2530,7 @@ func contextCatalog() map[string]interface{} { "inbox": map[string]interface{}{ "args": []string{"[read ]", "[--latest]", "[--limit ]", "[--from ]", "[--since ]", "[--full]", "[--clear [--before ]]"}, "description": "List received messages newest-first (~/.pilot/inbox/). Default limit 10 with previews; --latest for the newest full body; read for one message", - "returns": "messages [{id, from, received_at, type, bytes, preview|data}], total, shown, dir", + "returns": "messages [{id, from, received_at, type, bytes, message_id?, reply_to?, preview|data}], total, shown, dir", }, "received": map[string]interface{}{ "args": []string{"[--limit ]", "[--since ]", "[--clear [--before ]]"}, @@ -4899,9 +4910,13 @@ func cmdSendMessage(args []string) { // --reply-to marks this message as the answer to a message received // earlier: the value is that message's message_id (see `pilotctl inbox`), // and the sender's --wait matches on it. - replyTo := flagString(flags, "reply-to", "") - if _, given := flags["reply-to"]; given && !dataexchange.ValidMessageID(replyTo) { - fatalCode("invalid_argument", "--reply-to must be a message ID: 1-%d characters of A-Z a-z 0-9 . _ : -", dataexchange.MaxMessageIDLen) + var replyTo string + if raw, given := flags["reply-to"]; given { + id, err := resolveReplyTo(raw) + if err != nil { + fatalCode("invalid_argument", "%v", err) + } + replyTo = id } sendCount := flagInt(flags, "count", 1) @@ -4921,6 +4936,9 @@ func cmdSendMessage(args []string) { } } } + // --no-resend: the request is never sent a second time, neither on + // first contact nor after a lost ack, for requests not safe to repeat. + noResend := flagBool(flags, "no-resend") // --reuse-conn (or PILOT_SENDMSG_REUSE_CONN=1): when --count > 1, dial once // and reuse the same data-exchange connection for all N sends. Default false @@ -5001,6 +5019,9 @@ func cmdSendMessage(args []string) { } return c } + // redial dials like dialOnce but returns the error: when sendOne's + // retry after a lost ack cannot connect, the first attempt stands. + redial := func() (*dataexchange.Client, error) { return dataexchange.Dial(d, target) } // Set below when --wait is given. Declared here so sendOne can register // each request's message ID with it before the request leaves. @@ -5008,19 +5029,43 @@ func cmdSendMessage(args []string) { // sendOne sends one message on cl and returns timing/ack metadata. // reused=true is recorded when the connection was dialled on a prior call. + // redial opens a new connection to the same receiver, for the one retry + // after a lost ack (below). // // Every send carries a new message ID, so the receiver can tell a // re-delivery from a new message and a reply can name the request it // answers (reply_to). Client.Send delivers the message without the ID to // a receiver too old to know it: such a receiver stores nothing for the // tagged frame, says so, and gets the plain frame on the same connection. - sendOne := func(cl *dataexchange.Client, seq int, reused bool) map[string]interface{} { + sendOne := func(cl *dataexchange.Client, seq int, reused bool, redial func() (*dataexchange.Client, error)) map[string]interface{} { messageID := dataexchange.NewMessageID() if watch != nil { watch.addID(messageID) } sentAtNs := time.Now().UnixNano() - res, sendErr := cl.Send(messageFrame(innerType, []byte(data), messageID, replyTo, traceTime, sentAtNs)) + frame := messageFrame(innerType, []byte(data), messageID, replyTo, traceTime, sentAtNs) + res, sendErr := cl.Send(frame) + retried := false + if res == nil && isAckReadError(sendErr) && !noResend { + // The frame was written but no ack came back, so whether it was + // stored is unknown. A receiver through v1.13.9 never stores the + // tagged form, and the untagged copy only follows its answer to + // that, so with the first ack lost the message is lost too. + // Send the same frame once more on a new connection: a receiver + // that knows message IDs recognises the repeat by its ID and + // keeps one copy, and an older one answers the tagged frame + // again and gets the untagged copy. The one case that stores + // the message twice is an older receiver that lost the ack of + // the untagged copy; --no-resend opts out of the retry. + slog.Debug("send-message ACK read failed; sending again on a new connection", "err", sendErr) + if c2, err := redial(); err == nil { + retried = true + if res2, err2 := c2.Send(frame); res2 != nil { + res, sendErr = res2, err2 + } + c2.Close() + } + } ackRecvAtNs := time.Now().UnixNano() var ack *dataexchange.Frame var ackErr error @@ -5036,16 +5081,26 @@ func cmdSendMessage(args []string) { default: return map[string]interface{}{"seq": seq, "error": sendErr.Error(), "message_id": messageID} } + if res != nil && res.Tagged && watch != nil { + watch.markTagged() + } r := map[string]interface{}{ "seq": seq, "bytes": len(data), "reused": reused, - // message_id is the ID this message was sent with. tagged says - // whether the receiver got it: false for a receiver too old to - // know message IDs, which can then not be asked to echo it. + // message_id is the ID this message was sent with. "message_id": messageID, - "tagged": res != nil && res.Tagged, + } + if res != nil { + // tagged says whether the receiver got the ID: false for a + // receiver too old to know message IDs, which can then not be + // asked to echo it. Without an ack it is not known, and left out. + r["tagged"] = res.Tagged + } + if retried { + // The first ack was lost and the message was sent once more. + r["retried"] = true } if replyTo != "" { r["reply_to"] = replyTo @@ -5124,7 +5179,7 @@ func cmdSendMessage(args []string) { } tracef("dataexchange.Dial") defer cl.Close() - r := sendOne(cl, 0, false) + r := sendOne(cl, 0, false, redial) 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 @@ -5184,7 +5239,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. - if firstContact && !flagBool(flags, "no-resend") { + if firstContact && !noResend { if after := resendDelay(waitDur); after > 0 { cfg.resendAfter = after cfg.resend = func() (time.Time, error) { @@ -5198,7 +5253,7 @@ func cmdSendMessage(args []string) { return time.Time{}, err } defer c.Close() - rr := sendOne(c, 1, false) + rr := sendOne(c, 1, false, func() (*dataexchange.Client, error) { return dataexchange.Dial(rd, target) }) if e, failed := rr["error"].(string); failed { return time.Time{}, errors.New(e) } @@ -5249,7 +5304,7 @@ func cmdSendMessage(args []string) { defer cl.Close() var results []map[string]interface{} for i := 0; i < sendCount; i++ { - result := sendOne(cl, i, i > 0) + result := sendOne(cl, i, i > 0, redial) results = append(results, result) if i < sendCount-1 { time.Sleep(50 * time.Millisecond) @@ -5269,7 +5324,7 @@ func cmdSendMessage(args []string) { var results []map[string]interface{} for i := 0; i < sendCount; i++ { cl := dialOnce() - result := sendOne(cl, i, false) + result := sendOne(cl, i, false, redial) results = append(results, result) cl.Close() if i < sendCount-1 { @@ -5305,6 +5360,46 @@ func messageFrame(innerType uint32, data []byte, messageID, replyTo string, trac return f } +// inboxFileID matches the id `pilotctl inbox` lists for a message: its file +// name without .json, {TYPE}-{date}-{time.ms}-{seq}. It is not the message's +// message_id, though it passes as one. +var inboxFileID = regexp.MustCompile(`^[A-Z_]+-[0-9]{8}-[0-9]{6}\.[0-9]{3}-[0-9]+$`) + +// resolveReplyTo turns a --reply-to value into the message ID to send as +// reply_to. The sender's --wait only takes a reply whose reply_to is its +// own message ID, and never one that names another, so a wrong value costs +// the sender a timeout: catch the two easy mistakes here. A bare +// --reply-to parses as "true". An inbox file id is what the inbox listing +// shows first; it is mapped to the message_id stored in that record. +func resolveReplyTo(raw string) (string, error) { + if raw == "true" { + return "", errors.New("--reply-to needs a value: the message_id of the message you are answering (pilotctl inbox shows it)") + } + if inboxFileID.MatchString(raw) { + dir, err := inboxDirPath() + if err != nil { + return "", fmt.Errorf("--reply-to %s: %v", raw, err) + } + body, err := os.ReadFile(filepath.Join(dir, raw+".json")) + if err != nil { + return "", fmt.Errorf("--reply-to %s is an inbox file id, not a message_id, and no such message is in the inbox; pass the message_id that pilotctl inbox shows", raw) + } + var rec map[string]interface{} + if err := json.Unmarshal(body, &rec); err != nil { + return "", fmt.Errorf("--reply-to %s: parse inbox record: %v", raw, err) + } + id, _ := rec["message_id"].(string) + if !dataexchange.ValidMessageID(id) { + return "", fmt.Errorf("--reply-to %s: that message carries no message_id (its sender predates message IDs), so a reply cannot name it; send without --reply-to", raw) + } + return id, nil + } + if !dataexchange.ValidMessageID(raw) { + return "", fmt.Errorf("--reply-to must be a message ID: 1-%d characters of A-Z a-z 0-9 . _ : -", dataexchange.MaxMessageIDLen) + } + return raw, nil +} + // isAckReadError reports whether a Client.Send error means the frame was // written and only the receiver's acknowledgement could not be read. func isAckReadError(err error) bool { @@ -7221,11 +7316,31 @@ func cmdInboxRead(dir, id string) { ts, _ := m["received_at"].(string) msgType, _ := m["type"].(string) bytes, _ := m["bytes"].(float64) - fmt.Printf("ID: %s\nFrom: %s\nWhen: %s\nType: %s\nBytes: %d\n\n", m["id"], from, ts, msgType, int(bytes)) + fmt.Printf("ID: %s\nFrom: %s\nWhen: %s\nType: %s\nBytes: %d\n", m["id"], from, ts, msgType, int(bytes)) + if id, _ := m["message_id"].(string); id != "" { + fmt.Printf("Message ID: %s\n", id) + } + if id, _ := m["reply_to"].(string); id != "" { + fmt.Printf("Reply to: %s\n", id) + } + fmt.Println() body, _ := m["data"].(string) fmt.Println(body) } +// inboxCorrelation is the listing's line for a record's message_id and +// reply_to, or "" when it has neither (a sender that predates message IDs). +func inboxCorrelation(msg map[string]interface{}) string { + var parts []string + if id, _ := msg["message_id"].(string); id != "" { + parts = append(parts, "message_id "+id) + } + if id, _ := msg["reply_to"].(string); id != "" { + parts = append(parts, "reply_to "+id) + } + return strings.Join(parts, " · ") +} + // Messages are saved to ~/.pilot/inbox/ by the daemon's built-in service. // // Agent-first design: newest-first, bounded by default (--limit 10), full @@ -7367,6 +7482,15 @@ func cmdInbox(args []string) { "type": im.msg["type"], "bytes": im.msg["bytes"], } + // The ID a reply names (send-message --reply-to) is message_id, + // not id. message_id and reply_to are listed when the record + // has them, so an agent can answer a message or see what an + // answer answers without reading each record. + for _, key := range []string{"message_id", "reply_to"} { + if v, _ := im.msg[key].(string); v != "" { + m[key] = v + } + } if body, _ := im.msg["data"].(string); full { m["data"] = body } else { @@ -7401,6 +7525,9 @@ func cmdInbox(args []string) { } fmt.Printf(" %s\n", sAccent(im.id)) fmt.Printf(" %s\n", sDim(fmt.Sprintf("%s · %s · %s · %s", from, msgType, age, formatBytes(uint64(bytes))))) + if ids := inboxCorrelation(im.msg); ids != "" { + fmt.Printf(" %s\n", sDim(ids)) + } body, _ := im.msg["data"].(string) if full { fmt.Printf(" %s\n\n", body) diff --git a/cmd/pilotctl/zz_reply_match_test.go b/cmd/pilotctl/zz_reply_match_test.go index ca17bc06..90c7bc94 100644 --- a/cmd/pilotctl/zz_reply_match_test.go +++ b/cmd/pilotctl/zz_reply_match_test.go @@ -12,6 +12,7 @@ import ( "strings" "sync" "testing" + "time" "github.com/pilot-protocol/dataexchange" ) @@ -129,9 +130,13 @@ func TestInboxWatchWithoutIDsIgnoresReplyTo(t *testing.T) { // ── send-message against receivers of each generation ──────────────────── +// dxDropConnection, returned by a dxReceiver's answer, closes the connection +// instead of acknowledging the frame: the sender's ack read fails. +const dxDropConnection = "\x00drop" + // dxReceiver turns the stream daemon's echo into a data-exchange receiver: // answer is called with each complete wire frame (its type and payload) the -// client wrote and returns the acknowledgement text. +// client wrote and returns the acknowledgement text (or dxDropConnection). type dxReceiver struct { mu sync.Mutex buf map[uint32][]byte @@ -165,8 +170,16 @@ func newDXReceiver(sd *streamDaemon, answer func(ftype uint32, payload []byte, d r.buf[connID] = b[8+n:] r.types = append(r.types, ftype) r.frames = append(r.frames, decoded) + text := answer(ftype, payload, decoded) + if text == dxDropConnection { + closed := make([]byte, 5) + closed[0] = tdCmdCloseOK + binary.BigEndian.PutUint32(closed[1:5], connID) + out = append(out, closed) + break + } var ack bytes.Buffer - _ = dataexchange.WriteFrame(&ack, &dataexchange.Frame{Type: dataexchange.TypeText, Payload: []byte(answer(ftype, payload, decoded))}) + _ = dataexchange.WriteFrame(&ack, &dataexchange.Frame{Type: dataexchange.TypeText, Payload: []byte(text)}) push := make([]byte, 5+ack.Len()) push[0] = tdCmdRecv binary.BigEndian.PutUint32(push[1:5], connID) @@ -373,3 +386,364 @@ func TestSendMessageRejectsInvalidReplyTo(t *testing.T) { } } } + +// ── an untagged message ahead of our tagged reply ───────────────────────── + +// Our request reached the receiver with its ID, and the peer echoes IDs. +// First comes an untagged message from the peer (its answer to an untagged +// request another client on this host sent it), then, 400 ms later, the +// reply that names our request. The untagged one used to be taken at once. +func TestAwaitReplyHoldsAnUntaggedMessageForALaterTaggedReply(t *testing.T) { + dir := tempInbox(t) + watch := newInboxWatch(replyPeer) + watch.addID("our-request") + watch.markTagged() + + writeInboxRecord(t, dir, "TEXT-20260924-100005.000-000001.json", map[string]any{ + "from": replyPeer, "data": "answer to another client's untagged request", + }) + written := make(chan struct{}) + defer func() { <-written }() + go func() { + defer close(written) + time.Sleep(400 * time.Millisecond) + writeInboxRecord(t, dir, "TEXT-20260924-100006.000-000002.json", map[string]any{ + "from": replyPeer, "data": "our answer", "reply_to": "our-request", + }) + }() + out, err := awaitReply(watch, time.Now(), replyWait{wait: 5 * time.Second}) + if err != nil || out.reply == nil || out.reply["data"] != "our answer" { + t.Fatalf("awaitReply = %+v, %v; want the reply naming our request", out, err) + } +} + +// A peer that does not echo IDs (every responder today) still has its reply +// taken, once the grace period is over. +func TestAwaitReplyTakesAnUntaggedReplyAfterTheGracePeriod(t *testing.T) { + dir := tempInbox(t) + watch := newInboxWatch(replyPeer) + watch.addID("our-request") + watch.markTagged() + writeInboxRecord(t, dir, "TEXT-20260924-100005.000-000001.json", map[string]any{"from": replyPeer, "data": "the answer"}) + + start := time.Now() + out, err := awaitReply(watch, start, replyWait{wait: 5 * time.Second}) + if err != nil || out.reply == nil || out.reply["data"] != "the answer" { + t.Fatalf("awaitReply = %+v, %v; want the untagged answer", out, err) + } + if el := time.Since(start); el < untaggedReplyGrace || el > untaggedReplyGrace+time.Second { + t.Fatalf("answer taken after %s, want about %s", el, untaggedReplyGrace) + } +} + +// The grace period never outlasts the wait: when the wait ends first, the +// held message is the reply rather than a timeout. +func TestAwaitReplyTakesTheHeldMessageWhenTheWaitEnds(t *testing.T) { + dir := tempInbox(t) + watch := newInboxWatch(replyPeer) + watch.addID("our-request") + watch.markTagged() + writeInboxRecord(t, dir, "TEXT-20260924-100005.000-000001.json", map[string]any{"from": replyPeer, "data": "the answer"}) + + out, err := awaitReply(watch, time.Now(), replyWait{wait: 300 * time.Millisecond}) + if err != nil || out.reply == nil || out.reply["data"] != "the answer" { + t.Fatalf("awaitReply = %+v, %v; want the held answer when the wait ends", out, err) + } +} + +// A request delivered without its ID (a receiver that predates IDs) cannot +// be named by any reply: an untagged message is taken at once, as before. +func TestAwaitReplyTakesAnUntaggedReplyAtOnceForAnUntaggedRequest(t *testing.T) { + dir := tempInbox(t) + watch := newInboxWatch(replyPeer) + watch.addID("our-request") + writeInboxRecord(t, dir, "TEXT-20260924-100005.000-000001.json", map[string]any{"from": replyPeer, "data": "the answer"}) + + start := time.Now() + out, err := awaitReply(watch, start, replyWait{wait: 5 * time.Second}) + if err != nil || out.reply == nil || out.reply["data"] != "the answer" { + t.Fatalf("awaitReply = %+v, %v", out, err) + } + if el := time.Since(start); el >= untaggedReplyGrace { + t.Fatalf("answer taken after %s; an untagged request has nothing to wait for", el) + } +} + +// A peer seen naming another request in reply_to echoes IDs, so an untagged +// message from it is not the answer to our tagged request, not even when the +// wait ends. +func TestAwaitReplyWantsATaggedReplyFromAPeerThatEchoesIDs(t *testing.T) { + dir := tempInbox(t) + watch := newInboxWatch(replyPeer) + watch.addID("our-request") + watch.markTagged() + writeInboxRecord(t, dir, "TEXT-20260924-100005.000-000001.json", map[string]any{ + "from": replyPeer, "data": "answer to another request", "reply_to": "another-request", + }) + writeInboxRecord(t, dir, "TEXT-20260924-100006.000-000002.json", map[string]any{ + "from": replyPeer, "data": "answer to an untagged request", + }) + + out, err := awaitReply(watch, time.Now(), replyWait{wait: 1200 * time.Millisecond}) + if err != nil || out.reply != nil { + t.Fatalf("awaitReply = %+v, %v; want no reply", out, err) + } +} + +// The same through send-message: the receiver knows message IDs, so the +// request is tagged; the peer's untagged message lands first and the reply +// naming the request 400 ms later. +func TestSendMessageWaitHoldsAnUntaggedMessageForItsTaggedReply(t *testing.T) { + sd := newStreamDaemon(t) + home := sd.useDaemonNoRegistry(t) + inbox := filepath.Join(home, ".pilot", "inbox") + if err := os.MkdirAll(inbox, 0o700); err != nil { + t.Fatal(err) + } + const peer = "0:0000.0000.002A" + written := make(chan struct{}) + newDXReceiver(sd, func(_ uint32, _ []byte, f *dataexchange.Frame) string { + writeInboxRecord(t, inbox, "TEXT-20260924-100005.000-000001.json", map[string]any{ + "from": peer, "data": "answer to another client's untagged request", + }) + go func() { + defer close(written) + time.Sleep(400 * time.Millisecond) + writeInboxRecord(t, inbox, "TEXT-20260924-100006.000-000002.json", map[string]any{ + "from": peer, "data": "pong", "reply_to": f.MessageID, + }) + }() + return "ACK TEXT 4 bytes" + }) + + data := sendMessageJSON(t, peer, "--data", "ping", "--wait", "5s") + <-written + reply, _ := data["reply"].(map[string]interface{}) + if reply == nil || reply["data"] != "pong" || reply["reply_to"] != data["message_id"] { + t.Fatalf("reply = %v for message_id %v; want the answer to this request", reply, data["message_id"]) + } +} + +// ── --reply-to values ───────────────────────────────────────────────────── + +// withoutDaemon points pilotctl at a socket nobody listens on: a --reply-to +// value that got past validation fails on connecting instead. +func withoutDaemon(t *testing.T) { + t.Helper() + t.Setenv("PILOT_SOCKET", filepath.Join(t.TempDir(), "no-daemon.sock")) +} + +// A bare --reply-to parses as "true", a valid message ID, which no request +// was ever sent with. +func TestSendMessageRejectsABareReplyTo(t *testing.T) { + withTempHomeFull(t) + withoutDaemon(t) + for _, args := range [][]string{ + {"0:0000.0000.002A", "--data", "x", "--reply-to"}, + {"0:0000.0000.002A", "--reply-to", "--data", "x"}, + } { + _, _, f := runTrapped(t, func() { cmdSendMessage(args) }) + if f == nil || f.Code != "invalid_argument" || !strings.Contains(f.Message, "needs a value") { + t.Errorf("%v: %+v, want invalid_argument asking for a value", args, f) + } + } +} + +// The id `pilotctl inbox` lists is the record's file name. Given as +// --reply-to, it is replaced by the message_id stored in that record. +func TestSendMessageReplyToLooksUpAnInboxID(t *testing.T) { + sd := newStreamDaemon(t) + home := sd.useDaemonNoRegistry(t) + inbox := filepath.Join(home, ".pilot", "inbox") + if err := os.MkdirAll(inbox, 0o700); err != nil { + t.Fatal(err) + } + const requestID = "5d1e9c0a7b3f4e21a8c6d0b9e2f71a34" + writeInboxRecord(t, inbox, "TEXT-20260924-100005.000-000001.json", map[string]any{ + "from": "0:0000.0000.002A", "data": "ping", "message_id": requestID, + }) + rcv := newDXReceiver(sd, func(uint32, []byte, *dataexchange.Frame) string { return "ACK TEXT 4 bytes" }) + + data := sendMessageJSON(t, "0:0000.0000.002A", "--data", "pong", "--reply-to", "TEXT-20260924-100005.000-000001") + _, frames := rcv.seen() + if len(frames) != 1 || frames[0] == nil || frames[0].ReplyTo != requestID { + t.Fatalf("receiver got %+v, want reply_to %s", frames, requestID) + } + if data["reply_to"] != requestID { + t.Fatalf("result reply_to = %v, want %s", data["reply_to"], requestID) + } +} + +// An inbox id whose record has no message_id (or no record at all) has no +// ID to name: refused before anything is sent. +func TestSendMessageReplyToRefusesAnInboxIDWithoutMessageID(t *testing.T) { + inbox := tempInbox(t) + withoutDaemon(t) + writeInboxRecord(t, inbox, "TEXT-20260924-100005.000-000001.json", map[string]any{"from": "0:0000.0000.002A", "data": "ping"}) + for id, want := range map[string]string{ + "TEXT-20260924-100005.000-000001": "carries no message_id", + "TEXT-20260924-100009.000-000009": "not a message_id", + } { + _, _, f := runTrapped(t, func() { cmdSendMessage([]string{"0:0000.0000.002A", "--data", "x", "--reply-to", id}) }) + if f == nil || f.Code != "invalid_argument" || !strings.Contains(f.Message, want) { + t.Errorf("--reply-to %s: %+v, want invalid_argument %q", id, f, want) + } + } +} + +// ── the inbox shows message_id and reply_to ─────────────────────────────── + +func TestInboxShowsMessageIDAndReplyTo(t *testing.T) { + dir := tempInbox(t) + for name, rec := range map[string]map[string]any{ + "TEXT-20260924-100004.000-000000.json": {"from": replyPeer, "type": "TEXT", "data": "untagged", "bytes": 8, "received_at": "2026-09-24T10:00:04Z"}, + "TEXT-20260924-100005.000-000001.json": {"from": replyPeer, "type": "TEXT", "data": "ping", "bytes": 4, "received_at": "2026-09-24T10:00:05Z", "message_id": "req-1"}, + "TEXT-20260924-100006.000-000002.json": {"from": replyPeer, "type": "TEXT", "data": "pong", "bytes": 4, "received_at": "2026-09-24T10:00:06Z", "message_id": "ans-1", "reply_to": "req-1"}, + } { + writeInboxRecord(t, dir, name, rec) + } + + var env struct { + Data struct { + Messages []map[string]interface{} `json:"messages"` + } `json:"data"` + } + out := captureStdout(t, func() { withJSON(func() { cmdInbox(nil) }) }) + if err := json.Unmarshal([]byte(out), &env); err != nil || len(env.Data.Messages) != 3 { + t.Fatalf("inbox --json: %v\n%s", err, out) + } + byID := map[string]map[string]interface{}{} + for _, m := range env.Data.Messages { + byID[m["id"].(string)] = m + } + if m := byID["TEXT-20260924-100005.000-000001"]; m["message_id"] != "req-1" || m["from"] != replyPeer || m["preview"] != "ping" { + t.Errorf("request entry = %v", m) + } + if m := byID["TEXT-20260924-100006.000-000002"]; m["message_id"] != "ans-1" || m["reply_to"] != "req-1" { + t.Errorf("answer entry = %v", m) + } + m := byID["TEXT-20260924-100004.000-000000"] + if _, has := m["message_id"]; has { + t.Errorf("untagged entry = %v, want no message_id", m) + } + if _, has := m["reply_to"]; has { + t.Errorf("untagged entry = %v, want no reply_to", m) + } + + text := captureStdout(t, func() { withText(func() { cmdInbox(nil) }) }) + for _, want := range []string{"message_id req-1", "message_id ans-1 · reply_to req-1"} { + if !strings.Contains(text, want) { + t.Errorf("inbox listing lacks %q:\n%s", want, text) + } + } + read := captureStdout(t, func() { withText(func() { cmdInbox([]string{"read", "TEXT-20260924-100006.000-000002"}) }) }) + for _, want := range []string{"Message ID: ans-1", "Reply to: req-1"} { + if !strings.Contains(read, want) { + t.Errorf("inbox read lacks %q:\n%s", want, read) + } + } +} + +// ── a lost ack ──────────────────────────────────────────────────────────── + +// The first connection drops before the receiver's ack comes back. +func TestSendMessageSendsAgainWhenTheAckIsLost(t *testing.T) { + const peer = "0:0000.0000.002A" + + // A receiver through v1.13.9 stored nothing for the tagged frame. Sent + // again on a new connection, the message reaches it untagged, once. + t.Run("receiver that predates message IDs", func(t *testing.T) { + sd := newStreamDaemon(t) + sd.useDaemonNoRegistry(t) + calls := 0 + var stored []string + rcv := newDXReceiver(sd, func(ftype uint32, payload []byte, _ *dataexchange.Frame) string { + calls++ + if calls == 1 { + return dxDropConnection + } + if ftype != dataexchange.TypeText { + return fmt.Sprintf("ACK UNKNOWN(10) %d bytes", len(payload)) + } + stored = append(stored, string(payload)) + return fmt.Sprintf("ACK TEXT %d bytes", len(payload)) + }) + + data := sendMessageJSON(t, peer, "--data", "hello") + types, _ := rcv.seen() + if len(types) != 3 || types[0] != dataexchange.TypeTagged || types[1] != dataexchange.TypeTagged || types[2] != dataexchange.TypeText { + t.Fatalf("receiver saw frame types %v, want [TAGGED TAGGED TEXT]", types) + } + if len(stored) != 1 || stored[0] != "hello" { + t.Fatalf("receiver stored %q, want exactly [hello]", stored) + } + if data["retried"] != true || data["tagged"] != false || data["ack"] != "ACK TEXT 5 bytes" { + t.Fatalf("result = %v", data) + } + }) + + // A receiver that knows message IDs may have stored the first copy. The + // second carries the same ID, which it recognises as a repeat. + t.Run("receiver that knows message IDs", func(t *testing.T) { + sd := newStreamDaemon(t) + sd.useDaemonNoRegistry(t) + calls := 0 + rcv := newDXReceiver(sd, func(uint32, []byte, *dataexchange.Frame) string { + calls++ + if calls == 1 { + return dxDropConnection + } + return "ACK TEXT 5 bytes (duplicate)" + }) + + data := sendMessageJSON(t, peer, "--data", "hello") + _, frames := rcv.seen() + if len(frames) != 2 || frames[0] == nil || frames[1] == nil { + t.Fatalf("receiver saw %+v, want two frames", frames) + } + if frames[0].MessageID != frames[1].MessageID || frames[0].MessageID != data["message_id"] { + t.Fatalf("message IDs %q then %q (result %v), want the same one twice", frames[0].MessageID, frames[1].MessageID, data["message_id"]) + } + if data["retried"] != true || data["tagged"] != true { + t.Fatalf("result = %v", data) + } + }) + + // --no-resend: sent once, and with no ack the send fails. Whether the + // receiver got the ID is not known without its ack, so the result does + // not claim either. + t.Run("--no-resend", func(t *testing.T) { + sd := newStreamDaemon(t) + sd.useDaemonNoRegistry(t) + rcv := newDXReceiver(sd, func(uint32, []byte, *dataexchange.Frame) string { return dxDropConnection }) + t.Cleanup(func() { fatalResults = nil }) + + var stderr string + var f *trappedFatal + withJSON(func() { + _, stderr, f = runTrapped(t, func() { cmdSendMessage([]string{peer, "--data", "hello", "--no-resend", "--count", "2"}) }) + }) + if f == nil || f.Code != "connection_failed" || !strings.Contains(f.Message, "not acknowledged") { + t.Fatalf("send = %+v, want connection_failed for the unacknowledged messages", f) + } + if types, _ := rcv.seen(); len(types) != 2 { + t.Fatalf("receiver saw frame types %v, want one frame per message", types) + } + var env struct { + Results []map[string]interface{} `json:"results"` + } + if err := json.Unmarshal([]byte(stderr[strings.Index(stderr, "{"):]), &env); err != nil || len(env.Results) != 2 { + t.Fatalf("error envelope: %v\n%s", err, stderr) + } + for _, r := range env.Results { + for _, key := range []string{"tagged", "retried", "ack"} { + if v, has := r[key]; has { + t.Errorf("result has %s=%v: %v", key, v, r) + } + } + if _, has := r["ack_error"]; !has { + t.Errorf("result has no ack_error: %v", r) + } + } + }) +} From 19ff8c976c461381e12e6896bb4cddc8b166a8eb Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Wed, 7 Oct 2026 16:53:41 +0300 Subject: [PATCH 3/6] pilotctl: hold an untagged reply only from a peer known to echo IDs No service responder echoes reply_to yet, so holding every untagged reply for 0.75 s whenever the receiver's daemon knows message IDs would slow nearly every send-message --wait once daemons upgrade, for nothing. An untagged reply is now taken at once, as before, unless the peer is known to echo IDs: one of its newest messages already in the inbox carries a reply_to. newInboxWatch checks this once, from the snapshot it already reads, looking at no more than the newest 1000 records and 200 from the peer. For such a peer the 0.75 s hold stays, and a peer seen naming another request during the wait still gets only a reply naming ours taken. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 18 ++-- cmd/pilotctl/firstcontact.go | 125 +++++++++++++++++++------- cmd/pilotctl/main.go | 9 +- cmd/pilotctl/zz_reply_match_test.go | 131 ++++++++++++++++++++++++---- 4 files changed, 225 insertions(+), 58 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6995774f..baca543c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -37,14 +37,16 @@ Detailed per-release notes are on the inbox record). A reply whose `reply_to` is that ID is matched exactly, and a message whose `reply_to` names another request is never taken. Exact matching needs the responder to echo the request's `message_id` as - `reply_to`, which no service responder does yet: an untagged reply still - works and is matched by sender and arrival time as before, so for such - responders concurrent waits can still be crossed. An untagged message is - held back up to 0.75 s in case the reply naming the request follows, which - delays replies from those responders by that much; once the peer has been - seen naming another request in `reply_to`, only a reply naming ours is - taken. New flag `--reply-to ` sends a message as the answer to - a received one (an inbox file id is looked up; a bare flag is refused). + `reply_to`, which no service responder does yet. An untagged reply still + works: it is matched by sender and arrival time and taken as soon as it + arrives, as before, so for such responders concurrent waits can still be + crossed. From a peer known to echo IDs (one of its newest messages in the + inbox carries a `reply_to`) an untagged message is held back up to 0.75 s + in case the reply naming the request follows, and once the peer has been + seen naming another request in `reply_to` during the wait, only a reply + naming ours is taken. New flag `--reply-to ` sends a message + as the answer to a received one (an inbox file id is looked up; a bare + flag is refused). `pilotctl inbox` now shows each message's `message_id` and `reply_to`, in the listing and with `--json` (new fields; `id` is still the file name). A first-contact re-send uses an ID of its own (the receiver would drop a diff --git a/cmd/pilotctl/firstcontact.go b/cmd/pilotctl/firstcontact.go index 8eb40226..830ebb81 100644 --- a/cmd/pilotctl/firstcontact.go +++ b/cmd/pilotctl/firstcontact.go @@ -168,20 +168,28 @@ func dialFailureHint(target string, err error, attempts int, elapsed time.Durati // ID). The oldest such message from the peer is taken, as before IDs // existed. Two concurrent requests to such a peer, or anything else it // sends in the window, still cannot be told apart. -// - When our request reached the receiver with its ID (markTagged), an -// untagged message can also be the answer to someone else's untagged -// request to a peer that does echo IDs, with our own tagged reply still -// on its way. So it is held for untaggedReplyGrace first, and taken only -// if no reply naming our request arrives meanwhile (or when the wait -// ends). Once the peer has been seen naming another request in -// reply_to during this wait, it evidently echoes IDs, and only a reply -// naming ours is taken. +// - From a peer known to echo IDs, an untagged message can instead be the +// answer to someone else's untagged request, with our own tagged reply +// still on its way. A peer is known to echo when one of its earlier +// messages in the inbox carries a reply_to (peerEchoedBefore). If our +// request reached the receiver with its ID (markTagged), an untagged +// message from such a peer is held for untaggedReplyGrace first, and +// taken only if no reply naming our request arrives meanwhile (or when +// the wait ends). Holding replies from peers with no such history would +// only delay them: no service responder echoes IDs yet. +// - Once the peer has been seen naming another request in reply_to during +// this wait, it evidently echoes IDs, and only a reply naming ours is +// taken. type inboxWatch struct { dir string from string cutoff time.Time seen map[string]bool + // echoedBefore is set by newInboxWatch and only read afterwards: an + // earlier message from the peer in the inbox carries a reply_to. + echoedBefore bool + mu sync.Mutex ids map[string]bool // message IDs of our request (and of its re-send) tagged bool // a request of ours reached the receiver with its ID @@ -192,14 +200,23 @@ type inboxWatch struct { heldSince time.Time } -// untaggedReplyGrace is how long an untagged message from the peer is held -// back, after our request reached the receiver with its ID, in case the -// reply that names our request follows. A peer that does not echo IDs -// (every service responder today) has its reply delayed by this much; a -// peer that echoes them is not delayed at all. The send-message help -// quotes this value. +// untaggedReplyGrace is how long an untagged message from a peer known to +// echo IDs is held back, after our request reached the receiver with its +// ID, in case the reply that names our request follows. Replies from such +// a peer normally name our request and are not delayed at all. The +// send-message help quotes this value. const untaggedReplyGrace = 750 * time.Millisecond +// The look for a peer's echo history reads at most echoHistoryFiles of the +// newest inbox records, and stops after echoHistoryPeerRecords of them from +// the peer: the sender is only known once a record is read, and a big inbox +// must not slow the send. A peer whose last reply_to lies further back is +// treated as not echoing, which is how every peer was treated before. +const ( + echoHistoryPeerRecords = 200 + echoHistoryFiles = 1000 +) + func inboxDirPath() (string, error) { home, err := os.UserHomeDir() if err != nil { @@ -221,10 +238,62 @@ func newInboxWatch(from string) *inboxWatch { for _, e := range entries { w.seen[e.Name()] = true } + w.echoedBefore = peerEchoedBefore(dir, entries, from) } return w } +// inboxNameKey orders inbox files by arrival. Their names are +// {TYPE}-{timestamp}-{seq}.json, so the part after the type sorts by time +// whatever the message types. +func inboxNameKey(name string) string { + if i := strings.Index(name, "-"); i >= 0 { + return name[i:] + } + return name +} + +// peerEchoedBefore reports whether one of the newest records in entries +// (the inbox at dir) from peer carries a reply_to: the peer has answered a +// message by naming its ID, so it is known to echo IDs. The scan is bounded +// by echoHistoryFiles and echoHistoryPeerRecords. +func peerEchoedBefore(dir string, entries []os.DirEntry, peer string) bool { + if peer == "" { + return false + } + names := make([]string, 0, len(entries)) + for _, e := range entries { + if !e.IsDir() && strings.HasSuffix(e.Name(), ".json") { + names = append(names, e.Name()) + } + } + sort.Slice(names, func(i, j int) bool { return inboxNameKey(names[i]) > inboxNameKey(names[j]) }) + if len(names) > echoHistoryFiles { + names = names[:echoHistoryFiles] + } + fromPeer := 0 + for _, name := range names { + body, err := os.ReadFile(filepath.Join(dir, name)) + if err != nil { + continue + } + var rec struct { + From string `json:"from"` + ReplyTo string `json:"reply_to"` + } + if json.Unmarshal(body, &rec) != nil || rec.From != peer { + continue + } + if rec.ReplyTo != "" { + return true + } + if fromPeer++; fromPeer == echoHistoryPeerRecords { + break + } + } + return false +} + // addID registers the message ID of a request this watch waits on. The // re-sent request has an ID of its own; a reply to either one is ours. func (w *inboxWatch) addID(id string) { @@ -241,8 +310,9 @@ func (w *inboxWatch) addID(id string) { // markTagged records that a request reached the receiver with its message // ID (SendResult.Tagged), so the peer can name it in reply_to. Until then -// an untagged message is taken at once: a receiver that predates IDs -// dropped ours, and its peer can never name our request. +// an untagged message is taken at once even from a peer known to echo: a +// receiver that predates IDs dropped ours, and the peer can never name our +// request. func (w *inboxWatch) markTagged() { w.mu.Lock() defer w.mu.Unlock() @@ -293,16 +363,8 @@ func (w *inboxWatch) poll() (map[string]interface{}, error) { } return nil, fmt.Errorf("read inbox: %w", err) } - // Inbox files are {TYPE}-{timestamp}-{seq}.json: order by the part - // after the type so the oldest new message wins. - sort.Slice(entries, func(i, j int) bool { - ni, nj := entries[i].Name(), entries[j].Name() - di, dj := strings.Index(ni, "-"), strings.Index(nj, "-") - if di < 0 || dj < 0 { - return ni < nj - } - return ni[di:] < nj[dj:] - }) + // Oldest first, so the oldest new message wins. + sort.Slice(entries, func(i, j int) bool { return inboxNameKey(entries[i].Name()) < inboxNameKey(entries[j].Name()) }) var untagged map[string]interface{} for _, e := range entries { if e.IsDir() || w.seen[e.Name()] { @@ -343,10 +405,11 @@ func (w *inboxWatch) poll() (map[string]interface{}, error) { // untaggedReply decides whether msg, the oldest untagged message from the // peer (nil if there is none), is taken as the reply now. It is when our -// request went out without its ID. Otherwise it is held back for -// untaggedReplyGrace from the poll that first saw it, so a reply naming -// our request that arrives meanwhile wins; and it is never taken from a -// peer seen echoing IDs. +// request went out without its ID, or when the peer is not known to echo +// IDs (as before IDs existed). From a peer with an echo history it is held +// back for untaggedReplyGrace from the poll that first saw it, so a reply +// naming our request that arrives meanwhile wins; and it is never taken +// from a peer seen naming another request during this wait. func (w *inboxWatch) untaggedReply(msg map[string]interface{}, now time.Time) map[string]interface{} { if msg == nil { w.held = nil @@ -355,7 +418,7 @@ func (w *inboxWatch) untaggedReply(msg map[string]interface{}, now time.Time) ma w.mu.Lock() tagged := w.tagged w.mu.Unlock() - if !tagged { + if !tagged || !w.echoedBefore && !w.peerEchoes { return msg } if w.held == nil { diff --git a/cmd/pilotctl/main.go b/cmd/pilotctl/main.go index 9358b846..62463f16 100644 --- a/cmd/pilotctl/main.go +++ b/cmd/pilotctl/main.go @@ -926,9 +926,10 @@ Flags: Every message is sent with a new message ID (message_id in the --json result). --wait takes the message from the peer whose reply_to is that ID, and never one -whose reply_to names another message. A peer that does not echo the ID is -matched by sender and arrival time, as before; its reply is held back up to -0.75s in case a reply naming the ID follows. +whose reply_to names another message. A reply without reply_to is matched by +sender and arrival time, as before. From a peer known to echo IDs (one of its +earlier messages in the inbox carries a reply_to) such a reply is held back up +to 0.75s in case the reply naming the ID follows. Examples: pilotctl send-message list-agents --data '/data {"search":"weather","limit":5}' @@ -2519,7 +2520,7 @@ func contextCatalog() map[string]interface{} { // Messaging "send-message": map[string]interface{}{ "args": []string{"", "--data | --data - | --data-file ", "[--type text|json|binary]", "[--count ]", "[--reuse-conn]", "[--wait ]", "[--timeout ]", "[--reply-to ]"}, - "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. Every message carries a new message_id; --wait returns the reply whose reply_to names it (or, from a peer that does not echo IDs, the next message from the peer); --reply-to answers a received message", + "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. Every message carries a new message_id; --wait returns the reply whose reply_to names it, or else the next message from the peer without reply_to (held up to 0.75s if the peer is known to echo IDs); --reply-to answers a received message", "returns": "target, to, type, bytes, ack, reuse_conn, message_id, tagged, reply_to, reply", }, "send-file": map[string]interface{}{ diff --git a/cmd/pilotctl/zz_reply_match_test.go b/cmd/pilotctl/zz_reply_match_test.go index 90c7bc94..1646e9e2 100644 --- a/cmd/pilotctl/zz_reply_match_test.go +++ b/cmd/pilotctl/zz_reply_match_test.go @@ -389,12 +389,47 @@ func TestSendMessageRejectsInvalidReplyTo(t *testing.T) { // ── an untagged message ahead of our tagged reply ───────────────────────── -// Our request reached the receiver with its ID, and the peer echoes IDs. -// First comes an untagged message from the peer (its answer to an untagged -// request another client on this host sent it), then, 400 ms later, the -// reply that names our request. The untagged one used to be taken at once. +// storeEchoHistory puts an earlier answer from peer into the inbox, one +// that named a request in reply_to: the peer is known to echo IDs. It has +// to be there before the watch snapshots the inbox. +func storeEchoHistory(t *testing.T, dir, peer string) { + t.Helper() + writeInboxRecord(t, dir, "TEXT-20260924-090000.000-000000.json", map[string]any{ + "from": peer, "data": "an earlier answer", "reply_to": "an-earlier-request", + }) +} + +// A peer with no echo history (every service responder today) has its +// untagged reply taken at once, even though our request was tagged. The +// inbox holds an earlier message from the peer without reply_to, and one +// from another peer with reply_to, which says nothing about this peer. +func TestAwaitReplyTakesAnUntaggedReplyAtOnceFromAPeerWithNoEchoHistory(t *testing.T) { + dir := tempInbox(t) + writeInboxRecord(t, dir, "TEXT-20260924-090000.000-000000.json", map[string]any{"from": replyPeer, "data": "an earlier answer"}) + storeEchoHistory(t, dir, "0:0000.0009.0001") + watch := newInboxWatch(replyPeer) + watch.addID("our-request") + watch.markTagged() + writeInboxRecord(t, dir, "TEXT-20260924-100005.000-000001.json", map[string]any{"from": replyPeer, "data": "the answer"}) + + start := time.Now() + out, err := awaitReply(watch, start, replyWait{wait: 5 * time.Second}) + if err != nil || out.reply == nil || out.reply["data"] != "the answer" { + t.Fatalf("awaitReply = %+v, %v; want the untagged answer", out, err) + } + if el := time.Since(start); el >= untaggedReplyGrace/3 { + t.Fatalf("answer taken after %s; a peer with no echo history is not held back", el) + } +} + +// Our request reached the receiver with its ID, and the peer is known to +// echo IDs. First comes an untagged message from the peer (its answer to an +// untagged request another client on this host sent it), then, 400 ms +// later, the reply that names our request. The untagged one used to be +// taken at once. func TestAwaitReplyHoldsAnUntaggedMessageForALaterTaggedReply(t *testing.T) { dir := tempInbox(t) + storeEchoHistory(t, dir, replyPeer) watch := newInboxWatch(replyPeer) watch.addID("our-request") watch.markTagged() @@ -417,10 +452,11 @@ func TestAwaitReplyHoldsAnUntaggedMessageForALaterTaggedReply(t *testing.T) { } } -// A peer that does not echo IDs (every responder today) still has its reply -// taken, once the grace period is over. -func TestAwaitReplyTakesAnUntaggedReplyAfterTheGracePeriod(t *testing.T) { +// From a peer known to echo IDs, an untagged reply is held for the grace +// period, then taken when no reply naming our request came. +func TestAwaitReplyHoldsAnUntaggedReplyFromAPeerThatEchoedBefore(t *testing.T) { dir := tempInbox(t) + storeEchoHistory(t, dir, replyPeer) watch := newInboxWatch(replyPeer) watch.addID("our-request") watch.markTagged() @@ -440,6 +476,7 @@ func TestAwaitReplyTakesAnUntaggedReplyAfterTheGracePeriod(t *testing.T) { // held message is the reply rather than a timeout. func TestAwaitReplyTakesTheHeldMessageWhenTheWaitEnds(t *testing.T) { dir := tempInbox(t) + storeEchoHistory(t, dir, replyPeer) watch := newInboxWatch(replyPeer) watch.addID("our-request") watch.markTagged() @@ -452,9 +489,11 @@ func TestAwaitReplyTakesTheHeldMessageWhenTheWaitEnds(t *testing.T) { } // A request delivered without its ID (a receiver that predates IDs) cannot -// be named by any reply: an untagged message is taken at once, as before. +// be named by any reply: an untagged message is taken at once, even from a +// peer known to echo IDs. func TestAwaitReplyTakesAnUntaggedReplyAtOnceForAnUntaggedRequest(t *testing.T) { dir := tempInbox(t) + storeEchoHistory(t, dir, replyPeer) watch := newInboxWatch(replyPeer) watch.addID("our-request") writeInboxRecord(t, dir, "TEXT-20260924-100005.000-000001.json", map[string]any{"from": replyPeer, "data": "the answer"}) @@ -464,14 +503,14 @@ func TestAwaitReplyTakesAnUntaggedReplyAtOnceForAnUntaggedRequest(t *testing.T) if err != nil || out.reply == nil || out.reply["data"] != "the answer" { t.Fatalf("awaitReply = %+v, %v", out, err) } - if el := time.Since(start); el >= untaggedReplyGrace { + if el := time.Since(start); el >= untaggedReplyGrace/3 { t.Fatalf("answer taken after %s; an untagged request has nothing to wait for", el) } } -// A peer seen naming another request in reply_to echoes IDs, so an untagged -// message from it is not the answer to our tagged request, not even when the -// wait ends. +// A peer seen naming another request in reply_to during the wait echoes +// IDs, even with no history in the inbox, so an untagged message from it is +// not the answer to our tagged request, not even when the wait ends. func TestAwaitReplyWantsATaggedReplyFromAPeerThatEchoesIDs(t *testing.T) { dir := tempInbox(t) watch := newInboxWatch(replyPeer) @@ -490,9 +529,70 @@ func TestAwaitReplyWantsATaggedReplyFromAPeerThatEchoesIDs(t *testing.T) { } } -// The same through send-message: the receiver knows message IDs, so the -// request is tagged; the peer's untagged message lands first and the reply -// naming the request 400 ms later. +// The look for echo history stops after the newest echoHistoryPeerRecords +// records from the peer and the newest echoHistoryFiles records in all, so +// a big inbox does not slow the send. +func TestPeerEchoedBeforeIsBounded(t *testing.T) { + echoed := func(dir string) bool { + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatal(err) + } + return peerEchoedBefore(dir, entries, replyPeer) + } + fill := func(dir, from string, n int) { + for i := 0; i < n; i++ { + writeInboxRecord(t, dir, fmt.Sprintf("TEXT-20260924-100000.000-%06d.json", i+1), map[string]any{"from": from, "data": "x"}) + } + } + + dir := t.TempDir() + storeEchoHistory(t, dir, replyPeer) + fill(dir, replyPeer, echoHistoryPeerRecords-1) + if !echoed(dir) { + t.Fatalf("echo record behind %d newer records from the peer not found", echoHistoryPeerRecords-1) + } + fill(dir, replyPeer, echoHistoryPeerRecords) + if echoed(dir) { + t.Fatalf("echo record behind %d newer records from the peer was read", echoHistoryPeerRecords) + } + + dir = t.TempDir() + storeEchoHistory(t, dir, replyPeer) + fill(dir, "0:0000.0009.0001", echoHistoryFiles) + if echoed(dir) { + t.Fatalf("echo record behind %d newer files was read", echoHistoryFiles) + } +} + +// Through send-message, to a peer with no echo history: the request is +// tagged, and the peer's untagged reply is returned without a hold. +func TestSendMessageWaitTakesAnUntaggedReplyAtOnceWithoutEchoHistory(t *testing.T) { + sd := newStreamDaemon(t) + home := sd.useDaemonNoRegistry(t) + inbox := filepath.Join(home, ".pilot", "inbox") + if err := os.MkdirAll(inbox, 0o700); err != nil { + t.Fatal(err) + } + const peer = "0:0000.0000.002A" + newDXReceiver(sd, func(uint32, []byte, *dataexchange.Frame) string { + writeInboxRecord(t, inbox, "TEXT-20260924-100005.000-000001.json", map[string]any{"from": peer, "data": "pong"}) + return "ACK TEXT 4 bytes" + }) + + data := sendMessageJSON(t, peer, "--data", "ping", "--wait", "5s") + reply, _ := data["reply"].(map[string]interface{}) + if reply == nil || reply["data"] != "pong" || data["tagged"] != true { + t.Fatalf("result = %v; want the untagged pong for a tagged request", data) + } + if ms, _ := data["reply_after_ms"].(float64); ms >= float64(untaggedReplyGrace.Milliseconds()/3) { + t.Fatalf("reply taken after %v ms; a peer with no echo history is not held back", ms) + } +} + +// The same through send-message, to a peer known to echo IDs: the request +// is tagged; the peer's untagged message lands first and the reply naming +// the request 400 ms later. func TestSendMessageWaitHoldsAnUntaggedMessageForItsTaggedReply(t *testing.T) { sd := newStreamDaemon(t) home := sd.useDaemonNoRegistry(t) @@ -501,6 +601,7 @@ func TestSendMessageWaitHoldsAnUntaggedMessageForItsTaggedReply(t *testing.T) { t.Fatal(err) } const peer = "0:0000.0000.002A" + storeEchoHistory(t, inbox, peer) written := make(chan struct{}) newDXReceiver(sd, func(_ uint32, _ []byte, f *dataexchange.Frame) string { writeInboxRecord(t, inbox, "TEXT-20260924-100005.000-000001.json", map[string]any{ From bd8a986ebca0e6843ff472596fa11f2772c794d2 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Wed, 7 Oct 2026 17:29:49 +0300 Subject: [PATCH 4/6] pilotctl: review fixes for send-message reply matching after #501 - A --trace message is not sent again after a lost ack: receivers do not suppress repeated trace frames (dedupeEligible), so even a current receiver would store it twice. - The lost-ack retry still runs inside sendOne, before #501 judges the send: a retry that is acknowledged is delivered, and a send whose retry is not acknowledged fails with the retry's ack error. - The echo-history scan runs in a goroutine started with the inbox snapshot and is waited for only when an untagged reply has to be judged, so it no longer delays the request. It checks for the sender and reply_to bytes before parsing, skips records over 1 MiB and stops after 16 MiB as well as after 1000 files or 200 records from the peer. - The hold runs until 0.75 s after the message arrived (its received_at), and awaitReply polls when it ends, so it is no longer up to a poll interval longer than stated. - --reply-to accepts an inbox file name with .json. - The help and CHANGELOG say a receiver through v1.13.9 can store a retried message twice when the ack of its untagged copy is lost; from v1.13.10 the repeat is suppressed. - inbox read lines up its labels; a misleading poll comment is fixed. - Tests no longer call t.Fatal from goroutines or receiver callbacks, wait with a timeout, and allow 400 ms for a reply taken at once. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 20 ++- cmd/pilotctl/firstcontact.go | 152 ++++++++++++---- cmd/pilotctl/main.go | 44 +++-- cmd/pilotctl/zz_reply_match_test.go | 266 ++++++++++++++++++++++++---- 4 files changed, 394 insertions(+), 88 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index baca543c..baffc224 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -41,19 +41,23 @@ Detailed per-release notes are on the works: it is matched by sender and arrival time and taken as soon as it arrives, as before, so for such responders concurrent waits can still be crossed. From a peer known to echo IDs (one of its newest messages in the - inbox carries a `reply_to`) an untagged message is held back up to 0.75 s - in case the reply naming the request follows, and once the peer has been - seen naming another request in `reply_to` during the wait, only a reply - naming ours is taken. New flag `--reply-to ` sends a message - as the answer to a received one (an inbox file id is looked up; a bare - flag is refused). + inbox carries a `reply_to`) an untagged message is held back until 0.75 s + after it arrived, in case the reply naming the request follows; the inbox + is searched for that history beside the send, not before it. Once the + peer has been seen naming another request in `reply_to` during the wait, + only a reply naming ours is taken. New flag `--reply-to ` + sends a message as the answer to a received one (an inbox file id is + looked up; a bare flag is refused). `pilotctl inbox` now shows each message's `message_id` and `reply_to`, in the listing and with `--json` (new fields; `id` is still the file name). A first-contact re-send uses an ID of its own (the receiver would drop a repeat of the same ID as a duplicate) and a reply to either is accepted. When the ack is lost, the message is sent once more on a new connection - with the same ID, which a current receiver keeps only once (`"retried": - true`; `--no-resend` opts out). Receivers that predate message IDs still + with the same ID (`"retried": true`; `--no-resend` opts out). A receiver + from v1.13.10 on recognises the repeat and keeps one copy; one through + v1.13.9 can store the message twice when the ack of its untagged copy is + the one lost. A `--trace` message is never sent again: receivers do not + suppress repeated trace frames. Receivers that predate message IDs still get the message, in the old format (`"tagged": false` in the result); that costs one extra exchange on the same connection. No dependency change: the dataexchange release already required carries the IDs. diff --git a/cmd/pilotctl/firstcontact.go b/cmd/pilotctl/firstcontact.go index 830ebb81..f813590e 100644 --- a/cmd/pilotctl/firstcontact.go +++ b/cmd/pilotctl/firstcontact.go @@ -29,6 +29,7 @@ package main // first-contact reply does not show up, and say which step failed. import ( + "bytes" "encoding/json" "fmt" "os" @@ -171,12 +172,14 @@ func dialFailureHint(target string, err error, attempts int, elapsed time.Durati // - From a peer known to echo IDs, an untagged message can instead be the // answer to someone else's untagged request, with our own tagged reply // still on its way. A peer is known to echo when one of its earlier -// messages in the inbox carries a reply_to (peerEchoedBefore). If our -// request reached the receiver with its ID (markTagged), an untagged -// message from such a peer is held for untaggedReplyGrace first, and -// taken only if no reply naming our request arrives meanwhile (or when -// the wait ends). Holding replies from peers with no such history would -// only delay them: no service responder echoes IDs yet. +// messages in the inbox carries a reply_to (peerEchoedBefore, which runs +// beside the send and is waited for only when an untagged message +// arrives). If our request reached the receiver with its ID +// (markTagged), an untagged message from such a peer is held until +// untaggedReplyGrace after it arrived, and taken only if no reply naming +// our request arrives meanwhile (or when the wait ends). Holding replies +// from peers with no such history would only delay them: no service +// responder echoes IDs yet. // - Once the peer has been seen naming another request in reply_to during // this wait, it evidently echoes IDs, and only a reply naming ours is // taken. @@ -186,8 +189,9 @@ type inboxWatch struct { cutoff time.Time seen map[string]bool - // echoedBefore is set by newInboxWatch and only read afterwards: an - // earlier message from the peer in the inbox carries a reply_to. + // The echo-history scan newInboxWatch starts: echoedBefore is its + // result, valid once echoDone is closed (nil when no scan was started). + echoDone chan struct{} echoedBefore bool mu sync.Mutex @@ -203,20 +207,33 @@ type inboxWatch struct { // untaggedReplyGrace is how long an untagged message from a peer known to // echo IDs is held back, after our request reached the receiver with its // ID, in case the reply that names our request follows. Replies from such -// a peer normally name our request and are not delayed at all. The -// send-message help quotes this value. +// a peer normally name our request and are not delayed at all. awaitReply +// polls early so the hold ends on time. The send-message help quotes this +// value. const untaggedReplyGrace = 750 * time.Millisecond -// The look for a peer's echo history reads at most echoHistoryFiles of the -// newest inbox records, and stops after echoHistoryPeerRecords of them from -// the peer: the sender is only known once a record is read, and a big inbox -// must not slow the send. A peer whose last reply_to lies further back is -// treated as not echoing, which is how every peer was treated before. +// replyPollInterval is how often awaitReply looks at the inbox. +const replyPollInterval = 250 * time.Millisecond + +// The look for a peer's echo history reads only the newest inbox records: +// at most echoHistoryFiles files and echoHistoryBytes bytes, and no further +// than echoHistoryPeerRecords records from the peer. A record over +// echoHistoryRecordBytes is a large message rather than an answer worth +// reading, and is skipped. A record's sender is only known once it is read, +// and the inbox holds up to 256 MiB. A peer whose last reply_to lies further +// back is treated as not echoing, which is how every peer was treated +// before. const ( echoHistoryPeerRecords = 200 echoHistoryFiles = 1000 + echoHistoryBytes = 16 << 20 + echoHistoryRecordBytes = 1 << 20 ) +// peerEchoScan is peerEchoedBefore; tests replace it to see that the send +// does not wait for it. +var peerEchoScan = peerEchoedBefore + func inboxDirPath() (string, error) { home, err := os.UserHomeDir() if err != nil { @@ -238,11 +255,30 @@ func newInboxWatch(from string) *inboxWatch { for _, e := range entries { w.seen[e.Name()] = true } - w.echoedBefore = peerEchoedBefore(dir, entries, from) + // The scan reads files, tens of milliseconds on a full inbox, and + // its answer matters only once an untagged message arrives: run it + // beside the send instead of before it. + w.echoDone = make(chan struct{}) + scan := peerEchoScan + go func() { + defer close(w.echoDone) + w.echoedBefore = scan(dir, entries, from) + }() } return w } +// knownToEcho reports whether the peer is known to echo IDs from its +// messages already in the inbox, waiting for newInboxWatch's scan if it is +// still running. +func (w *inboxWatch) knownToEcho() bool { + if w.echoDone == nil { + return false + } + <-w.echoDone + return w.echoedBefore +} + // inboxNameKey orders inbox files by arrival. Their names are // {TYPE}-{timestamp}-{seq}.json, so the part after the type sorts by time // whatever the message types. @@ -256,36 +292,51 @@ func inboxNameKey(name string) string { // peerEchoedBefore reports whether one of the newest records in entries // (the inbox at dir) from peer carries a reply_to: the peer has answered a // message by naming its ID, so it is known to echo IDs. The scan is bounded -// by echoHistoryFiles and echoHistoryPeerRecords. +// by the echoHistory constants. func peerEchoedBefore(dir string, entries []os.DirEntry, peer string) bool { if peer == "" { return false } - names := make([]string, 0, len(entries)) + files := make([]os.DirEntry, 0, len(entries)) for _, e := range entries { if !e.IsDir() && strings.HasSuffix(e.Name(), ".json") { - names = append(names, e.Name()) + files = append(files, e) } } - sort.Slice(names, func(i, j int) bool { return inboxNameKey(names[i]) > inboxNameKey(names[j]) }) - if len(names) > echoHistoryFiles { - names = names[:echoHistoryFiles] + sort.Slice(files, func(i, j int) bool { return inboxNameKey(files[i].Name()) > inboxNameKey(files[j].Name()) }) + if len(files) > echoHistoryFiles { + files = files[:echoHistoryFiles] } + // The daemon writes records with encoding/json, without spaces, so these + // byte patterns find the sender and a reply_to without parsing the + // record; only a record that has both is parsed to be sure. + fromPattern := []byte(`"from":"` + peer + `"`) + replyToPattern := []byte(`"reply_to":"`) + var read int64 fromPeer := 0 - for _, name := range names { - body, err := os.ReadFile(filepath.Join(dir, name)) - if err != nil { + for _, e := range files { + info, err := e.Info() + if err != nil || info.Size() > echoHistoryRecordBytes { continue } - var rec struct { - From string `json:"from"` - ReplyTo string `json:"reply_to"` + if read += info.Size(); read > echoHistoryBytes { + break } - if json.Unmarshal(body, &rec) != nil || rec.From != peer { + body, err := os.ReadFile(filepath.Join(dir, e.Name())) + if err != nil { continue } - if rec.ReplyTo != "" { - return true + if !bytes.Contains(body, fromPattern) { + continue + } + if bytes.Contains(body, replyToPattern) { + var rec struct { + From string `json:"from"` + ReplyTo string `json:"reply_to"` + } + if json.Unmarshal(body, &rec) == nil && rec.From == peer && rec.ReplyTo != "" { + return true + } } if fromPeer++; fromPeer == echoHistoryPeerRecords { break @@ -391,8 +442,9 @@ func (w *inboxWatch) poll() (map[string]interface{}, error) { case replyOurs: return msg, nil case replyOther: - // Another request's answer. It stays out of w.seen: an ID - // added later (the re-send) is checked against it again. + // Another request's answer, not taken. Every poll reads it + // again, so an ID added later (the re-send's) is still + // checked against it. w.peerEchoes = true case replyUntagged: if untagged == nil { @@ -407,9 +459,10 @@ func (w *inboxWatch) poll() (map[string]interface{}, error) { // peer (nil if there is none), is taken as the reply now. It is when our // request went out without its ID, or when the peer is not known to echo // IDs (as before IDs existed). From a peer with an echo history it is held -// back for untaggedReplyGrace from the poll that first saw it, so a reply -// naming our request that arrives meanwhile wins; and it is never taken -// from a peer seen naming another request during this wait. +// back until untaggedReplyGrace after it arrived (its received_at, or the +// poll that first saw it), so a reply naming our request that arrives +// meanwhile wins; and it is never taken from a peer seen naming another +// request during this wait. func (w *inboxWatch) untaggedReply(msg map[string]interface{}, now time.Time) map[string]interface{} { if msg == nil { w.held = nil @@ -418,11 +471,16 @@ func (w *inboxWatch) untaggedReply(msg map[string]interface{}, now time.Time) ma w.mu.Lock() tagged := w.tagged w.mu.Unlock() - if !tagged || !w.echoedBefore && !w.peerEchoes { + if !tagged || !w.peerEchoes && !w.knownToEcho() { return msg } if w.held == nil { w.heldSince = now + if ts, _ := msg["received_at"].(string); ts != "" { + if at, err := time.Parse(time.RFC3339Nano, ts); err == nil && at.Before(now) { + w.heldSince = at + } + } } w.held = msg if w.peerEchoes || now.Sub(w.heldSince) < untaggedReplyGrace { @@ -441,6 +499,18 @@ func (w *inboxWatch) heldReply() map[string]interface{} { return w.held } +// holdLeft is how much of the grace period a held message has left at now, +// or 0 when nothing is held that may still be taken. +func (w *inboxWatch) holdLeft(now time.Time) time.Duration { + if w.heldReply() == nil { + return 0 + } + if left := w.heldSince.Add(untaggedReplyGrace).Sub(now); left > 0 { + return left + } + return 0 +} + // replyWait configures awaitReply. type replyWait struct { // wait is the reply budget, counted from the moment the request was @@ -526,7 +596,13 @@ func awaitReply(w *inboxWatch, ackAt time.Time, cfg replyWait) (replyOutcome, er out.waited = time.Since(ackAt) return out, nil } - time.Sleep(250 * time.Millisecond) + // Poll early when a held message's grace period ends sooner, so it + // is taken on time rather than up to a poll interval late. + sleep := replyPollInterval + if left := w.holdLeft(time.Now()); left > 0 && left < sleep { + sleep = left + } + time.Sleep(sleep) } } diff --git a/cmd/pilotctl/main.go b/cmd/pilotctl/main.go index 62463f16..5e4c2df5 100644 --- a/cmd/pilotctl/main.go +++ b/cmd/pilotctl/main.go @@ -914,7 +914,10 @@ Flags: --no-resend never send the request a second time (by default it is re-sent once on a new stream when, on first contact, no reply came by mid-wait, or when its - acknowledgement was lost) + acknowledgement was lost; a receiver from v1.13.10 + on keeps one copy of a repeat, but one through + v1.13.9 can store it twice if the ack of its + untagged copy is lost) --timeout total wall time, including connect, handshake and reply --trace print per-step timing breakdown to stderr --no-auto-handshake skip automatic trust handshake with known agents @@ -928,8 +931,8 @@ Every message is sent with a new message ID (message_id in the --json result). --wait takes the message from the peer whose reply_to is that ID, and never one whose reply_to names another message. A reply without reply_to is matched by sender and arrival time, as before. From a peer known to echo IDs (one of its -earlier messages in the inbox carries a reply_to) such a reply is held back up -to 0.75s in case the reply naming the ID follows. +earlier messages in the inbox carries a reply_to) such a reply is held back +until 0.75s after it arrived, in case the reply naming the ID follows. Examples: pilotctl send-message list-agents --data '/data {"search":"weather","limit":5}' @@ -2520,7 +2523,7 @@ func contextCatalog() map[string]interface{} { // Messaging "send-message": map[string]interface{}{ "args": []string{"", "--data | --data - | --data-file ", "[--type text|json|binary]", "[--count ]", "[--reuse-conn]", "[--wait ]", "[--timeout ]", "[--reply-to ]"}, - "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. Every message carries a new message_id; --wait returns the reply whose reply_to names it, or else the next message from the peer without reply_to (held up to 0.75s if the peer is known to echo IDs); --reply-to answers a received message", + "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. Every message carries a new message_id; --wait returns the reply whose reply_to names it, or else the next message from the peer without reply_to (held until 0.75s after it arrived if the peer is known to echo IDs); --reply-to answers a received message", "returns": "target, to, type, bytes, ack, reuse_conn, message_id, tagged, reply_to, reply", }, "send-file": map[string]interface{}{ @@ -5047,22 +5050,34 @@ func cmdSendMessage(args []string) { frame := messageFrame(innerType, []byte(data), messageID, replyTo, traceTime, sentAtNs) res, sendErr := cl.Send(frame) retried := false - if res == nil && isAckReadError(sendErr) && !noResend { + if res == nil && isAckReadError(sendErr) && !noResend && !traceTime { // The frame was written but no ack came back, so whether it was // stored is unknown. A receiver through v1.13.9 never stores the // tagged form, and the untagged copy only follows its answer to // that, so with the first ack lost the message is lost too. // Send the same frame once more on a new connection: a receiver - // that knows message IDs recognises the repeat by its ID and - // keeps one copy, and an older one answers the tagged frame - // again and gets the untagged copy. The one case that stores - // the message twice is an older receiver that lost the ack of - // the untagged copy; --no-resend opts out of the retry. + // from v1.13.10 on recognises the repeat by its ID and keeps one + // copy, and an older one answers the tagged frame again and gets + // the untagged copy. The one case that stores the message twice + // is a receiver through v1.13.9 that stored the untagged copy + // and lost its ack; --no-resend opts out of the retry. A --trace + // message is never sent again: receivers do not suppress + // repeated trace frames, so every receiver would store it twice. + // + // This runs before the send is judged: a message whose retry is + // acknowledged is delivered, and the ack error reported for one + // that is not is the retry's. slog.Debug("send-message ACK read failed; sending again on a new connection", "err", sendErr) if c2, err := redial(); err == nil { retried = true - if res2, err2 := c2.Send(frame); res2 != nil { + res2, err2 := c2.Send(frame) + switch { + case res2 != nil, isAckReadError(err2): res, sendErr = res2, err2 + default: + // The retry was not written: the first attempt, written + // but unacknowledged, is what happened to the message. + sendErr = fmt.Errorf("%w; sending again failed: %v", sendErr, err2) } c2.Close() } @@ -5376,6 +5391,10 @@ func resolveReplyTo(raw string) (string, error) { if raw == "true" { return "", errors.New("--reply-to needs a value: the message_id of the message you are answering (pilotctl inbox shows it)") } + // The file name itself, as ls shows it, is the same mistake. + if name := strings.TrimSuffix(raw, ".json"); name != raw && inboxFileID.MatchString(name) { + raw = name + } if inboxFileID.MatchString(raw) { dir, err := inboxDirPath() if err != nil { @@ -7317,7 +7336,8 @@ func cmdInboxRead(dir, id string) { ts, _ := m["received_at"].(string) msgType, _ := m["type"].(string) bytes, _ := m["bytes"].(float64) - fmt.Printf("ID: %s\nFrom: %s\nWhen: %s\nType: %s\nBytes: %d\n", m["id"], from, ts, msgType, int(bytes)) + // Labels padded to the longest, "Message ID:", so the values line up. + fmt.Printf("ID: %s\nFrom: %s\nWhen: %s\nType: %s\nBytes: %d\n", m["id"], from, ts, msgType, int(bytes)) if id, _ := m["message_id"].(string); id != "" { fmt.Printf("Message ID: %s\n", id) } diff --git a/cmd/pilotctl/zz_reply_match_test.go b/cmd/pilotctl/zz_reply_match_test.go index 1646e9e2..8bc85a3b 100644 --- a/cmd/pilotctl/zz_reply_match_test.go +++ b/cmd/pilotctl/zz_reply_match_test.go @@ -11,6 +11,7 @@ import ( "path/filepath" "strings" "sync" + "sync/atomic" "testing" "time" @@ -19,12 +20,50 @@ import ( func writeInboxRecord(t *testing.T, dir, name string, rec map[string]any) { t.Helper() - body, _ := json.Marshal(rec) - if err := os.WriteFile(filepath.Join(dir, name), body, 0o600); err != nil { + if err := putInboxRecord(dir, name, rec); err != nil { t.Fatal(err) } } +// putInboxRecord writes an inbox record as the daemon does. It returns the +// error rather than failing the test, for goroutines and receiver callbacks, +// which must not call t.Fatal. +func putInboxRecord(dir, name string, rec map[string]any) error { + body, err := json.Marshal(rec) + if err != nil { + return err + } + return os.WriteFile(filepath.Join(dir, name), body, 0o600) +} + +// putInboxRecordLater writes an inbox record after delay, from a goroutine. +// The returned channel is closed once it is written (or failed to be). +func putInboxRecordLater(t *testing.T, delay time.Duration, dir, name string, rec map[string]any) <-chan struct{} { + written := make(chan struct{}) + go func() { + defer close(written) + time.Sleep(delay) + if err := putInboxRecord(dir, name, rec); err != nil { + t.Error(err) + } + }() + return written +} + +// awaitClosed fails the test if ch is not closed within a few seconds. +func awaitClosed(t *testing.T, ch <-chan struct{}, what string) { + t.Helper() + select { + case <-ch: + case <-time.After(5 * time.Second): + t.Fatalf("timed out waiting for %s", what) + } +} + +// atOnce bounds how long an untagged reply that is not held may take: well +// under the hold, with room for a loaded machine. +const atOnce = 400 * time.Millisecond + const replyPeer = "0:0000.0002.BBE4" // Two `send-message --wait` to the same peer are in flight (A and B). The @@ -134,6 +173,11 @@ func TestInboxWatchWithoutIDsIgnoresReplyTo(t *testing.T) { // instead of acknowledging the frame: the sender's ack read fails. const dxDropConnection = "\x00drop" +// dxCutAck, returned by a dxReceiver's answer, sends the first bytes of an +// ack and then closes the connection: the sender's ack read fails with +// "unexpected EOF" rather than "EOF". +const dxCutAck = "\x00cut" + // dxReceiver turns the stream daemon's echo into a data-exchange receiver: // answer is called with each complete wire frame (its type and payload) the // client wrote and returns the acknowledgement text (or dxDropConnection). @@ -171,6 +215,13 @@ func newDXReceiver(sd *streamDaemon, answer func(ftype uint32, payload []byte, d r.types = append(r.types, ftype) r.frames = append(r.frames, decoded) text := answer(ftype, payload, decoded) + if text == dxCutAck { + part := make([]byte, 5+3) + part[0] = tdCmdRecv + binary.BigEndian.PutUint32(part[1:5], connID) + out = append(out, part) + text = dxDropConnection + } if text == dxDropConnection { closed := make([]byte, 5) closed[0] = tdCmdCloseOK @@ -357,12 +408,16 @@ func TestSendMessageWaitTakesTheReplyToItsOwnRequest(t *testing.T) { } const peer = "0:0000.0000.002A" newDXReceiver(sd, func(_ uint32, _ []byte, f *dataexchange.Frame) string { - writeInboxRecord(t, inbox, "TEXT-20260924-100005.000-000001.json", map[string]any{ + if err := putInboxRecord(inbox, "TEXT-20260924-100005.000-000001.json", map[string]any{ "from": peer, "data": "answer to somebody else's request", "reply_to": "somebody-else", - }) - writeInboxRecord(t, inbox, "TEXT-20260924-100006.000-000002.json", map[string]any{ + }); err != nil { + t.Error(err) + } + if err := putInboxRecord(inbox, "TEXT-20260924-100006.000-000002.json", map[string]any{ "from": peer, "data": "pong", "reply_to": f.MessageID, - }) + }); err != nil { + t.Error(err) + } return "ACK TEXT 4 bytes" }) @@ -417,7 +472,7 @@ func TestAwaitReplyTakesAnUntaggedReplyAtOnceFromAPeerWithNoEchoHistory(t *testi if err != nil || out.reply == nil || out.reply["data"] != "the answer" { t.Fatalf("awaitReply = %+v, %v; want the untagged answer", out, err) } - if el := time.Since(start); el >= untaggedReplyGrace/3 { + if el := time.Since(start); el >= atOnce { t.Fatalf("answer taken after %s; a peer with no echo history is not held back", el) } } @@ -437,15 +492,10 @@ func TestAwaitReplyHoldsAnUntaggedMessageForALaterTaggedReply(t *testing.T) { writeInboxRecord(t, dir, "TEXT-20260924-100005.000-000001.json", map[string]any{ "from": replyPeer, "data": "answer to another client's untagged request", }) - written := make(chan struct{}) - defer func() { <-written }() - go func() { - defer close(written) - time.Sleep(400 * time.Millisecond) - writeInboxRecord(t, dir, "TEXT-20260924-100006.000-000002.json", map[string]any{ - "from": replyPeer, "data": "our answer", "reply_to": "our-request", - }) - }() + written := putInboxRecordLater(t, 400*time.Millisecond, dir, "TEXT-20260924-100006.000-000002.json", map[string]any{ + "from": replyPeer, "data": "our answer", "reply_to": "our-request", + }) + defer awaitClosed(t, written, "the tagged reply to be written") out, err := awaitReply(watch, time.Now(), replyWait{wait: 5 * time.Second}) if err != nil || out.reply == nil || out.reply["data"] != "our answer" { t.Fatalf("awaitReply = %+v, %v; want the reply naming our request", out, err) @@ -503,7 +553,7 @@ func TestAwaitReplyTakesAnUntaggedReplyAtOnceForAnUntaggedRequest(t *testing.T) if err != nil || out.reply == nil || out.reply["data"] != "the answer" { t.Fatalf("awaitReply = %+v, %v", out, err) } - if el := time.Since(start); el >= untaggedReplyGrace/3 { + if el := time.Since(start); el >= atOnce { t.Fatalf("answer taken after %s; an untagged request has nothing to wait for", el) } } @@ -563,6 +613,96 @@ func TestPeerEchoedBeforeIsBounded(t *testing.T) { if echoed(dir) { t.Fatalf("echo record behind %d newer files was read", echoHistoryFiles) } + + // By bytes: past echoHistoryBytes nothing more is read. A record over + // echoHistoryRecordBytes (a large message) is skipped, not read. + big := func(dir, name string, size int) { + writeInboxRecord(t, dir, name, map[string]any{"from": "0:0000.0009.0001", "data": strings.Repeat("x", size)}) + } + dir = t.TempDir() + storeEchoHistory(t, dir, replyPeer) + half := echoHistoryRecordBytes / 2 + for i := 0; i <= echoHistoryBytes/half; i++ { + big(dir, fmt.Sprintf("TEXT-20260924-100000.000-%06d.json", i+1), half) + } + if echoed(dir) { + t.Fatalf("echo record behind %d MiB of newer records was read", echoHistoryBytes>>20) + } + dir = t.TempDir() + storeEchoHistory(t, dir, replyPeer) + big(dir, "TEXT-20260924-100000.000-000001.json", echoHistoryBytes) + if !echoed(dir) { + t.Fatal("a record as large as the byte budget stopped the scan instead of being skipped") + } +} + +// The echo-history scan runs beside the send: the request leaves while it +// is still running, and the scan is waited for only when the untagged reply +// has to be judged. +func TestSendMessageDoesNotWaitForTheEchoHistoryScan(t *testing.T) { + release := make(chan struct{}) + var releaseOnce sync.Once + var scanDone atomic.Bool + defer func(old func(string, []os.DirEntry, string) bool) { peerEchoScan = old }(peerEchoScan) + peerEchoScan = func(string, []os.DirEntry, string) bool { + select { + case <-release: + case <-time.After(3 * time.Second): + } + scanDone.Store(true) + return false + } + + sd := newStreamDaemon(t) + home := sd.useDaemonNoRegistry(t) + inbox := filepath.Join(home, ".pilot", "inbox") + if err := os.MkdirAll(inbox, 0o700); err != nil { + t.Fatal(err) + } + const peer = "0:0000.0000.002A" + var sentDuringScan atomic.Bool + newDXReceiver(sd, func(uint32, []byte, *dataexchange.Frame) string { + sentDuringScan.Store(!scanDone.Load()) + if err := putInboxRecord(inbox, "TEXT-20260924-100005.000-000001.json", map[string]any{"from": peer, "data": "pong"}); err != nil { + t.Error(err) + } + releaseOnce.Do(func() { close(release) }) + return "ACK TEXT 4 bytes" + }) + + data := sendMessageJSON(t, peer, "--data", "ping", "--wait", "5s") + if !sentDuringScan.Load() { + t.Fatal("the request left only after the echo-history scan finished") + } + if reply, _ := data["reply"].(map[string]interface{}); reply == nil || reply["data"] != "pong" { + t.Fatalf("result = %v; want the pong", data) + } +} + +// The hold runs from the message's arrival (its received_at), not from the +// poll that first sees it, and awaitReply polls again when it ends rather +// than at the next 250 ms tick: a message that arrived 50 ms short of the +// grace period before the first poll is taken about 50 ms later. +func TestAwaitReplyHoldEndsTheGracePeriodAfterArrival(t *testing.T) { + dir := tempInbox(t) + storeEchoHistory(t, dir, replyPeer) + watch := newInboxWatch(replyPeer) + watch.addID("our-request") + watch.markTagged() + const left = 50 * time.Millisecond + arrived := time.Now().Add(-(untaggedReplyGrace - left)) + writeInboxRecord(t, dir, "TEXT-20260924-100005.000-000001.json", map[string]any{ + "from": replyPeer, "data": "the answer", "received_at": arrived.Format(time.RFC3339Nano), + }) + + start := time.Now() + out, err := awaitReply(watch, start, replyWait{wait: 5 * time.Second}) + if err != nil || out.reply == nil || out.reply["data"] != "the answer" { + t.Fatalf("awaitReply = %+v, %v; want the untagged answer", out, err) + } + if el := time.Since(start); el < left-10*time.Millisecond || el >= replyPollInterval-50*time.Millisecond { + t.Fatalf("answer taken %s after the first poll, want about %s", el, left) + } } // Through send-message, to a peer with no echo history: the request is @@ -576,7 +716,9 @@ func TestSendMessageWaitTakesAnUntaggedReplyAtOnceWithoutEchoHistory(t *testing. } const peer = "0:0000.0000.002A" newDXReceiver(sd, func(uint32, []byte, *dataexchange.Frame) string { - writeInboxRecord(t, inbox, "TEXT-20260924-100005.000-000001.json", map[string]any{"from": peer, "data": "pong"}) + if err := putInboxRecord(inbox, "TEXT-20260924-100005.000-000001.json", map[string]any{"from": peer, "data": "pong"}); err != nil { + t.Error(err) + } return "ACK TEXT 4 bytes" }) @@ -585,7 +727,7 @@ func TestSendMessageWaitTakesAnUntaggedReplyAtOnceWithoutEchoHistory(t *testing. if reply == nil || reply["data"] != "pong" || data["tagged"] != true { t.Fatalf("result = %v; want the untagged pong for a tagged request", data) } - if ms, _ := data["reply_after_ms"].(float64); ms >= float64(untaggedReplyGrace.Milliseconds()/3) { + if ms, _ := data["reply_after_ms"].(float64); ms >= float64(atOnce.Milliseconds()) { t.Fatalf("reply taken after %v ms; a peer with no echo history is not held back", ms) } } @@ -602,23 +744,26 @@ func TestSendMessageWaitHoldsAnUntaggedMessageForItsTaggedReply(t *testing.T) { } const peer = "0:0000.0000.002A" storeEchoHistory(t, inbox, peer) - written := make(chan struct{}) + written := make(chan (<-chan struct{}), 1) newDXReceiver(sd, func(_ uint32, _ []byte, f *dataexchange.Frame) string { - writeInboxRecord(t, inbox, "TEXT-20260924-100005.000-000001.json", map[string]any{ + if err := putInboxRecord(inbox, "TEXT-20260924-100005.000-000001.json", map[string]any{ "from": peer, "data": "answer to another client's untagged request", + }); err != nil { + t.Error(err) + } + written <- putInboxRecordLater(t, 400*time.Millisecond, inbox, "TEXT-20260924-100006.000-000002.json", map[string]any{ + "from": peer, "data": "pong", "reply_to": f.MessageID, }) - go func() { - defer close(written) - time.Sleep(400 * time.Millisecond) - writeInboxRecord(t, inbox, "TEXT-20260924-100006.000-000002.json", map[string]any{ - "from": peer, "data": "pong", "reply_to": f.MessageID, - }) - }() return "ACK TEXT 4 bytes" }) data := sendMessageJSON(t, peer, "--data", "ping", "--wait", "5s") - <-written + select { + case w := <-written: + awaitClosed(t, w, "the tagged reply to be written") + case <-time.After(5 * time.Second): + t.Fatal("the receiver never got the request") + } reply, _ := data["reply"].(map[string]interface{}) if reply == nil || reply["data"] != "pong" || reply["reply_to"] != data["message_id"] { t.Fatalf("reply = %v for message_id %v; want the answer to this request", reply, data["message_id"]) @@ -673,6 +818,10 @@ func TestSendMessageReplyToLooksUpAnInboxID(t *testing.T) { if data["reply_to"] != requestID { t.Fatalf("result reply_to = %v, want %s", data["reply_to"], requestID) } + // The file name as ls shows it, with .json, is looked up the same way. + if got, err := resolveReplyTo("TEXT-20260924-100005.000-000001.json"); err != nil || got != requestID { + t.Fatalf("--reply-to with .json = %q, %v; want %s", got, err, requestID) + } } // An inbox id whose record has no message_id (or no record at all) has no @@ -738,7 +887,7 @@ func TestInboxShowsMessageIDAndReplyTo(t *testing.T) { } } read := captureStdout(t, func() { withText(func() { cmdInbox([]string{"read", "TEXT-20260924-100006.000-000002"}) }) }) - for _, want := range []string{"Message ID: ans-1", "Reply to: req-1"} { + for _, want := range []string{"ID: TEXT-20260924-100006.000-000002\n", "Bytes: 4\n", "Message ID: ans-1\n", "Reply to: req-1\n"} { if !strings.Contains(read, want) { t.Errorf("inbox read lacks %q:\n%s", want, read) } @@ -810,6 +959,51 @@ func TestSendMessageSendsAgainWhenTheAckIsLost(t *testing.T) { } }) + // The retry is lost too: the message was sent twice before the send was + // judged, and with no ack at all it fails, with the retry's ack error. + t.Run("retry unacknowledged too", func(t *testing.T) { + sd := newStreamDaemon(t) + sd.useDaemonNoRegistry(t) + calls := 0 + rcv := newDXReceiver(sd, func(uint32, []byte, *dataexchange.Frame) string { + if calls++; calls == 1 { + return dxDropConnection + } + return dxCutAck + }) + + var f *trappedFatal + withJSON(func() { + _, _, f = runTrapped(t, func() { cmdSendMessage([]string{peer, "--data", "hello"}) }) + }) + if f == nil || f.Code != "connection_failed" || !strings.Contains(f.Message, "did not acknowledge") || !strings.HasSuffix(f.Message, ": dataexchange: read ack: unexpected EOF") || strings.Contains(f.Message, "sending again failed") { + t.Fatalf("send = %+v, want connection_failed: not acknowledged, with only the retry's ack error", f) + } + if types, _ := rcv.seen(); len(types) != 2 { + t.Fatalf("receiver saw frame types %v, want the message and its retry", types) + } + }) + + // --trace: receivers do not suppress a repeated trace frame, so even a + // current receiver would store the message twice. It is sent once, and + // with no ack the send fails. + t.Run("--trace", func(t *testing.T) { + sd := newStreamDaemon(t) + sd.useDaemonNoRegistry(t) + rcv := newDXReceiver(sd, func(uint32, []byte, *dataexchange.Frame) string { return dxDropConnection }) + + var f *trappedFatal + withJSON(func() { + _, _, f = runTrapped(t, func() { cmdSendMessage([]string{peer, "--data", "hello", "--trace"}) }) + }) + if f == nil || f.Code != "connection_failed" || !strings.Contains(f.Message, "did not acknowledge") { + t.Fatalf("send = %+v, want connection_failed: not acknowledged", f) + } + if types, _ := rcv.seen(); len(types) != 1 { + t.Fatalf("receiver saw frame types %v, want the traced message once", types) + } + }) + // --no-resend: sent once, and with no ack the send fails. Whether the // receiver got the ID is not known without its ack, so the result does // not claim either. @@ -848,3 +1042,15 @@ func TestSendMessageSendsAgainWhenTheAckIsLost(t *testing.T) { } }) } + +// A message whose first ack was lost and whose retry was acknowledged was +// delivered: a --count run with it succeeds. +func TestUndeliveredCountsARetriedAndAcknowledgedMessageAsDelivered(t *testing.T) { + results := []map[string]interface{}{ + {"seq": 0, "ack": "ACK TEXT 5 bytes", "retried": true, "tagged": true}, + {"seq": 1, "ack": "ACK TEXT 5 bytes"}, + } + if failed, _, first := undelivered(results); failed != 0 { + t.Fatalf("undelivered = %d (%s), want 0", failed, first) + } +} From 59040758db8ffa1f3032aea506c56d4a487ab252 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Wed, 7 Oct 2026 17:43:13 +0300 Subject: [PATCH 5/6] pilotctl: the untagged-reply hold starts no earlier than the ack Review fix. The hold counted from the message's received_at even when that was before our request was acknowledged, so after a slow first-contact dial, a large payload or a lost-ack retry, an untagged message from a peer known to echo IDs was taken at the first poll, ahead of the tagged reply. A message that arrived before the ack is the least likely to be the reply; its hold now starts at the ack. The echo-history scan also skips anything that is not a regular file, so a FIFO in the inbox cannot block it. Co-Authored-By: Claude Opus 5.5 --- cmd/pilotctl/firstcontact.go | 11 +++++++-- cmd/pilotctl/zz_reply_match_test.go | 36 +++++++++++++++++++++++++---- 2 files changed, 41 insertions(+), 6 deletions(-) diff --git a/cmd/pilotctl/firstcontact.go b/cmd/pilotctl/firstcontact.go index f813590e..2af5a946 100644 --- a/cmd/pilotctl/firstcontact.go +++ b/cmd/pilotctl/firstcontact.go @@ -202,6 +202,7 @@ type inboxWatch struct { peerEchoes bool // the peer named another request in reply_to held map[string]interface{} // untagged message held back (see untaggedReply) heldSince time.Time + holdFrom time.Time // our request's acknowledgement; no hold starts earlier } // untaggedReplyGrace is how long an untagged message from a peer known to @@ -316,8 +317,8 @@ func peerEchoedBefore(dir string, entries []os.DirEntry, peer string) bool { fromPeer := 0 for _, e := range files { info, err := e.Info() - if err != nil || info.Size() > echoHistoryRecordBytes { - continue + if err != nil || !info.Mode().IsRegular() || info.Size() > echoHistoryRecordBytes { + continue // a FIFO or device would block the read } if read += info.Size(); read > echoHistoryBytes { break @@ -481,6 +482,11 @@ func (w *inboxWatch) untaggedReply(msg map[string]interface{}, now time.Time) ma w.heldSince = at } } + // A message that arrived before our request was acknowledged is the + // least likely to be its reply: its hold starts at the ack. + if w.heldSince.Before(w.holdFrom) { + w.heldSince = w.holdFrom + } } w.held = msg if w.peerEchoes || now.Sub(w.heldSince) < untaggedReplyGrace { @@ -546,6 +552,7 @@ func awaitReply(w *inboxWatch, ackAt time.Time, cfg replyWait) (replyOutcome, er err error } var out replyOutcome + w.holdFrom = ackAt deadline := ackAt.Add(cfg.wait) var resendCh chan resendResult // non-nil while the re-send is in flight lastObserve := time.Time{} diff --git a/cmd/pilotctl/zz_reply_match_test.go b/cmd/pilotctl/zz_reply_match_test.go index 8bc85a3b..8f4f2ee6 100644 --- a/cmd/pilotctl/zz_reply_match_test.go +++ b/cmd/pilotctl/zz_reply_match_test.go @@ -690,21 +690,49 @@ func TestAwaitReplyHoldEndsTheGracePeriodAfterArrival(t *testing.T) { watch.addID("our-request") watch.markTagged() const left = 50 * time.Millisecond - arrived := time.Now().Add(-(untaggedReplyGrace - left)) + ackAt := time.Now().Add(-(untaggedReplyGrace - left)) writeInboxRecord(t, dir, "TEXT-20260924-100005.000-000001.json", map[string]any{ - "from": replyPeer, "data": "the answer", "received_at": arrived.Format(time.RFC3339Nano), + "from": replyPeer, "data": "the answer", "received_at": ackAt.Add(time.Millisecond).Format(time.RFC3339Nano), }) start := time.Now() - out, err := awaitReply(watch, start, replyWait{wait: 5 * time.Second}) + out, err := awaitReply(watch, ackAt, replyWait{wait: 5 * time.Second}) if err != nil || out.reply == nil || out.reply["data"] != "the answer" { t.Fatalf("awaitReply = %+v, %v; want the untagged answer", out, err) } - if el := time.Since(start); el < left-10*time.Millisecond || el >= replyPollInterval-50*time.Millisecond { + if el := time.Since(start); el < left-20*time.Millisecond || el >= replyPollInterval-50*time.Millisecond { t.Fatalf("answer taken %s after the first poll, want about %s", el, left) } } +// An untagged message that arrived before our request was acknowledged (a +// slow first-contact dial, a large payload, a lost-ack retry) is held for +// the whole grace period from the acknowledgement, not from its arrival. In +// review it was taken at the first poll, ahead of our tagged reply. +func TestAwaitReplyHoldStartsNoEarlierThanTheAck(t *testing.T) { + dir := tempInbox(t) + storeEchoHistory(t, dir, replyPeer) + watch := newInboxWatch(replyPeer) + watch.addID("our-request") + watch.markTagged() + ackAt := time.Now() + writeInboxRecord(t, dir, "TEXT-20260924-100005.000-000001.json", map[string]any{ + "from": replyPeer, "data": "something else", "received_at": ackAt.Add(-untaggedReplyGrace).Format(time.RFC3339Nano), + }) + go func() { + time.Sleep(200 * time.Millisecond) + if err := putInboxRecord(dir, "TEXT-20260924-100006.000-000002.json", map[string]any{ + "from": replyPeer, "data": "the answer", "reply_to": "our-request", + }); err != nil { + t.Error(err) + } + }() + out, err := awaitReply(watch, ackAt, replyWait{wait: 5 * time.Second}) + if err != nil || out.reply == nil || out.reply["data"] != "the answer" { + t.Fatalf("awaitReply = %+v, %v; want the tagged answer, not the earlier untagged message", out, err) + } +} + // Through send-message, to a peer with no echo history: the request is // tagged, and the peer's untagged reply is returned without a hold. func TestSendMessageWaitTakesAnUntaggedReplyAtOnceWithoutEchoHistory(t *testing.T) { From bc6506a6d4cbd3e14d338f7d28e5e3dea4ef1df4 Mon Sep 17 00:00:00 2001 From: Teo Calin Date: Wed, 7 Oct 2026 17:53:20 +0300 Subject: [PATCH 6/6] pilotctl: justify the inbox reads gosec flags; close the retry connection explicitly Co-Authored-By: Claude Opus 5.5 --- cmd/pilotctl/firstcontact.go | 2 +- cmd/pilotctl/main.go | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/cmd/pilotctl/firstcontact.go b/cmd/pilotctl/firstcontact.go index 2af5a946..b97bddaf 100644 --- a/cmd/pilotctl/firstcontact.go +++ b/cmd/pilotctl/firstcontact.go @@ -323,7 +323,7 @@ func peerEchoedBefore(dir string, entries []os.DirEntry, peer string) bool { if read += info.Size(); read > echoHistoryBytes { break } - body, err := os.ReadFile(filepath.Join(dir, e.Name())) + body, err := os.ReadFile(filepath.Join(dir, e.Name())) // #nosec G304 -- a regular file listed in this node's own inbox directory if err != nil { continue } diff --git a/cmd/pilotctl/main.go b/cmd/pilotctl/main.go index 5e4c2df5..ae4f1f69 100644 --- a/cmd/pilotctl/main.go +++ b/cmd/pilotctl/main.go @@ -5079,7 +5079,7 @@ func cmdSendMessage(args []string) { // but unacknowledged, is what happened to the message. sendErr = fmt.Errorf("%w; sending again failed: %v", sendErr, err2) } - c2.Close() + _ = c2.Close() } } ackRecvAtNs := time.Now().UnixNano() @@ -5400,7 +5400,7 @@ func resolveReplyTo(raw string) (string, error) { if err != nil { return "", fmt.Errorf("--reply-to %s: %v", raw, err) } - body, err := os.ReadFile(filepath.Join(dir, raw+".json")) + body, err := os.ReadFile(filepath.Join(dir, raw+".json")) // #nosec G304 -- raw matched inboxFileID (no path separators); a file in this node's own inbox if err != nil { return "", fmt.Errorf("--reply-to %s is an inbox file id, not a message_id, and no such message is in the inbox; pass the message_id that pilotctl inbox shows", raw) }