diff --git a/CHANGELOG.md b/CHANGELOG.md index f4d54d24..baffc224 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,38 @@ 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` 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: 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 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 (`"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. - **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..b97bddaf 100644 --- a/cmd/pilotctl/firstcontact.go +++ b/cmd/pilotctl/firstcontact.go @@ -29,12 +29,14 @@ package main // first-contact reply does not show up, and say which step failed. import ( + "bytes" "encoding/json" "fmt" "os" "path/filepath" "sort" "strings" + "sync" "time" "github.com/pilot-protocol/common/driver" @@ -153,16 +155,86 @@ 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. +// - 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, 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. type inboxWatch struct { dir string from string cutoff time.Time seen map[string]bool + + // 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 + 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 + holdFrom time.Time // our request's acknowledgement; no hold starts earlier } +// 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. awaitReply +// polls early so the hold ends on time. The send-message help quotes this +// value. +const untaggedReplyGrace = 750 * time.Millisecond + +// 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 { @@ -184,11 +256,150 @@ func newInboxWatch(from string) *inboxWatch { for _, e := range entries { w.seen[e.Name()] = true } + // 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 } -// poll returns the oldest reply that arrived since the snapshot. +// 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. +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 the echoHistory constants. +func peerEchoedBefore(dir string, entries []os.DirEntry, peer string) bool { + if peer == "" { + return false + } + files := make([]os.DirEntry, 0, len(entries)) + for _, e := range entries { + if !e.IsDir() && strings.HasSuffix(e.Name(), ".json") { + files = append(files, e) + } + } + 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 _, e := range files { + info, err := e.Info() + 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 + } + 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 + } + 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 + } + } + 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) { + if id == "" { + return + } + w.mu.Lock() + defer w.mu.Unlock() + if w.ids == nil { + w.ids = map[string]bool{} + } + 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 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() + w.tagged = 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, subject to untaggedReply. func (w *inboxWatch) poll() (map[string]interface{}, error) { if w.dir == "" { dir, err := inboxDirPath() @@ -204,16 +415,9 @@ 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()] { continue @@ -235,9 +439,82 @@ 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, 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 { + untagged = msg + } + } } - return nil, 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, 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 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 + return nil + } + w.mu.Lock() + tagged := w.tagged + w.mu.Unlock() + 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 + } + } + // 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 { + 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 +} + +// 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. @@ -275,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{} @@ -308,7 +586,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) { @@ -317,10 +597,19 @@ 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 } - 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 f708c7e4..ae4f1f69 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,12 +911,28 @@ 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; 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 + --reply-to mark this message as the answer to a received message: + 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, 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 +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}' @@ -1035,6 +1052,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. @@ -2500,9 +2522,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 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{}{ "args": []string{"", ""}, @@ -2512,7 +2534,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 ]]"}, @@ -4867,7 +4889,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 +4909,18 @@ 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. + 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) @@ -4907,6 +4940,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 @@ -4987,40 +5023,103 @@ 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. + 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. - 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)) + // 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, redial func() (*dataexchange.Client, error)) map[string]interface{} { + messageID := dataexchange.NewMessageID() + if watch != nil { + watch.addID(messageID) + } + sentAtNs := time.Now().UnixNano() + frame := messageFrame(innerType, []byte(data), messageID, replyTo, traceTime, sentAtNs) + res, sendErr := cl.Send(frame) + retried := false + 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 + // 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 + 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() } - sentAtNs = sendStart.UnixNano() - } - if sendErr != nil { - return map[string]interface{}{"seq": seq, "error": sendErr.Error()} } - - ack, ackErr := cl.Recv() 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} + } + 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. + "message_id": messageID, + } + 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 } if ack != nil { r["ack"] = string(ack.Payload) @@ -5060,7 +5159,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) } @@ -5097,7 +5195,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 @@ -5157,7 +5255,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) { @@ -5171,7 +5269,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) } @@ -5222,7 +5320,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) @@ -5242,7 +5340,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 { @@ -5262,6 +5360,72 @@ 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 +} + +// 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)") + } + // 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 { + return "", fmt.Errorf("--reply-to %s: %v", raw, err) + } + 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) + } + 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 { + 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) @@ -7172,11 +7336,32 @@ 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)) + // 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) + } + 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 @@ -7318,6 +7503,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 { @@ -7352,6 +7546,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 new file mode 100644 index 00000000..8f4f2ee6 --- /dev/null +++ b/cmd/pilotctl/zz_reply_match_test.go @@ -0,0 +1,1084 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package main + +import ( + "bytes" + "encoding/binary" + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/pilot-protocol/dataexchange" +) + +func writeInboxRecord(t *testing.T, dir, name string, rec map[string]any) { + t.Helper() + 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 +// 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 ──────────────────── + +// dxDropConnection, returned by a dxReceiver's answer, closes the connection +// 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). +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) + 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 + 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(text)}) + 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 { + 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", + }); 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" + }) + + 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) + } + } +} + +// ── an untagged message ahead of our tagged reply ───────────────────────── + +// 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 >= atOnce { + 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() + + writeInboxRecord(t, dir, "TEXT-20260924-100005.000-000001.json", map[string]any{ + "from": replyPeer, "data": "answer to another client's untagged 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) + } +} + +// 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() + 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) + storeEchoHistory(t, dir, replyPeer) + 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, 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"}) + + 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 >= atOnce { + t.Fatalf("answer taken after %s; an untagged request has nothing to wait for", el) + } +} + +// 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) + 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 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) + } + + // 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 + 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": ackAt.Add(time.Millisecond).Format(time.RFC3339Nano), + }) + + start := time.Now() + 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-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) { + 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 { + 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" + }) + + 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(atOnce.Milliseconds()) { + 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) + inbox := filepath.Join(home, ".pilot", "inbox") + if err := os.MkdirAll(inbox, 0o700); err != nil { + t.Fatal(err) + } + const peer = "0:0000.0000.002A" + storeEchoHistory(t, inbox, peer) + written := make(chan (<-chan struct{}), 1) + newDXReceiver(sd, func(_ uint32, _ []byte, f *dataexchange.Frame) string { + 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, + }) + return "ACK TEXT 4 bytes" + }) + + data := sendMessageJSON(t, peer, "--data", "ping", "--wait", "5s") + 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"]) + } +} + +// ── --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) + } + // 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 +// 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{"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) + } + } +} + +// ── 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) + } + }) + + // 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. + 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) + } + } + }) +} + +// 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) + } +} 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) + } +}