diff --git a/CHANGELOG.md b/CHANGELOG.md index 91eea53..5bdf76f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- `driver.SendToConfirmed`: sends a datagram like `SendTo` and reports + whether the daemon sent it, or why not (no route, port policy, ephemeral + ports exhausted). Against a daemon without the `dgram_confirm` feature it + falls back to the fire-and-forget send and reports the outcome as unknown. + `ErrConfirmTimeout` (outcome unknown) and `ErrConfirmQueueTimeout` + (nothing sent) tell the two timeouts apart. + ### Removed - `decision` and `actionhook`. They existed for the hosted control plane, which was retired on 2026-10-01. Their last importers (`dataexchange`, diff --git a/driver/driver.go b/driver/driver.go index 803af12..0fa8236 100644 --- a/driver/driver.go +++ b/driver/driver.go @@ -5,10 +5,13 @@ package driver import ( "encoding/binary" "encoding/json" + "errors" "fmt" "os" "path/filepath" "runtime" + "strings" + "sync/atomic" "time" "github.com/pilot-protocol/common/protocol" @@ -64,6 +67,11 @@ func (d *Driver) jsonRPC(msg []byte, expectCmd byte, label string) (map[string]i type Driver struct { ipc *ipcClient socketPath string + + // noDgramConfirm is set once the daemon has answered cmdSendToConfirm + // with "unknown command", so later SendToConfirmed calls go straight + // to the legacy send instead of probing again. + noDgramConfirm atomic.Bool } // Connect creates a new driver connected to the local daemon. @@ -173,6 +181,74 @@ func (d *Driver) SendTo(dst protocol.Addr, port uint16, data []byte) error { return d.ipc.send(msg) } +// SendToConfirmed sends an unreliable unicast datagram like SendTo, and +// waits for the daemon to say whether it sent it. +// +// - (true, nil): the daemon handed the datagram to its tunnel. Datagrams +// are unreliable, so this still says nothing about delivery to the peer. +// - (false, err): the daemon could not send it (no route to the node, port +// policy, ephemeral ports exhausted, ...), or did not answer in time +// (ErrConfirmTimeout: the datagram may or may not have left), or this +// call never got its turn (ErrConfirmQueueTimeout: nothing was sent). +// - (false, nil): the daemon predates confirmed sends (no "dgram_confirm" +// feature). The datagram was written with the legacy fire-and-forget +// command, exactly as SendTo does, and its outcome is unknown. +// +// Unlike SendTo this is a request/reply exchange, and a Driver runs one of +// those at a time: a confirmed send waits behind an in-flight Dial on the +// same Driver. Use SendTo where throughput matters more than the outcome. +// +// When the daemon does not answer in time, the next confirmed send on this +// Driver waits until that late answer has arrived: with no request IDs on the +// IPC, that is what keeps it from being taken as the next one's answer. Other +// requests are not held up. +func (d *Driver) SendToConfirmed(dst protocol.Addr, port uint16, data []byte) (confirmed bool, err error) { + if dst.IsBroadcast() { + return false, fmt.Errorf("broadcast address requires admin token: use Driver.Broadcast") + } + if d.noDgramConfirm.Load() { + return false, d.SendTo(dst, port, data) + } + msg := make([]byte, 1+protocol.AddrSize+2+len(data)) + msg[0] = cmdSendToConfirm + dst.MarshalTo(msg, 1) + binary.BigEndian.PutUint16(msg[1+protocol.AddrSize:], port) + copy(msg[1+protocol.AddrSize+2:], data) + + resp, err := d.ipc.sendConfirmAndWait(msg, sendToConfirmTimeout, sendToConfirmTimeout) + if err != nil { + return false, err + } + if resp.cmd == cmdSendToOK { + return true, nil + } + text := "" + if len(resp.payload) >= 2 { + text = string(resp.payload[2:]) + } + if strings.HasPrefix(text, fmt.Sprintf("unknown command: 0x%02X", cmdSendToConfirm)) { + // Older daemon: it rejected the command without sending anything, + // so fall back to the send it does understand. + d.noDgramConfirm.Store(true) + return false, d.SendTo(dst, port, data) + } + return false, fmt.Errorf("daemon: %s", text) +} + +// sendToConfirmTimeout bounds how long SendToConfirmed waits for its turn on +// the Driver, and then for the daemon's answer. A variable so tests can +// shorten it. +var sendToConfirmTimeout = defaultDialTimeout + +// ErrConfirmTimeout is returned by SendToConfirmed when the daemon did not +// answer in time. The datagram may or may not have been sent. +var ErrConfirmTimeout = errors.New("datagram send not confirmed: the daemon did not answer in time; it may or may not have been sent") + +// ErrConfirmQueueTimeout is returned by SendToConfirmed when it waited too +// long for its turn on the Driver: behind other requests, or for an earlier +// confirmed send's late answer. Nothing was sent. +var ErrConfirmQueueTimeout = errors.New("datagram not sent: timed out waiting for this Driver's earlier requests to be answered") + // Broadcast fans an unreliable datagram out to every member of a network. // The admin token must match the daemon's configured Config.AdminToken; an // empty token or mismatched token is rejected. Permitted on every network diff --git a/driver/ipc.go b/driver/ipc.go index b9e3aa0..1f6c08b 100644 --- a/driver/ipc.go +++ b/driver/ipc.go @@ -6,6 +6,7 @@ import ( "encoding/binary" "fmt" "net" + "strings" "sync" "time" @@ -79,6 +80,14 @@ const ( cmdSignEnvelopeOK byte = 0x34 cmdVerifyEnvelope byte = 0x35 cmdVerifyEnvelopeOK byte = 0x36 + // 0x37/0x38 are the daemon's CmdUnbind/CmdUnbindOK. + // + // cmdSendToConfirm is cmdSendTo with a reply: cmdSendToOK once the + // daemon has handed the datagram to its tunnel, cmdError when it could + // not send it. Daemons that support it list "dgram_confirm" in the info + // reply's features; older daemons reply cmdError "unknown command". + cmdSendToConfirm byte = 0x39 + cmdSendToOK byte = 0x3A ) // Network sub-commands (must match daemon SubNetwork* constants) @@ -146,6 +155,13 @@ type ipcClient struct { // while many goroutines are queued behind a slow sendAndWait. waitSem chan struct{} // capacity 1 + // confirmSem (capacity 1) is held by a confirmed send from before its + // request is written until its answer has arrived or the connection is + // gone, with no time limit: a confirmed send that timed out keeps the + // next one waiting until its late answer has gone by (see + // sendConfirmAndWait). Taken before waitSem. + confirmSem chan struct{} + // waiterMu guards the active-waiter slot. The IPC wire protocol has no // request IDs and the daemon dispatches requests concurrently, so a // reply that arrives AFTER its request timed out (or was abandoned) @@ -165,6 +181,11 @@ type ipcClient struct { // mitigation that keeps a late reply from being mis-delivered. waiterMu sync.Mutex activeWaiter *ipcWaiter // current in-flight waiter, nil when idle + // confirmWaiter is the confirmed send (cmdSendToConfirm) whose answer is + // outstanding, nil when none is. Its answer is recognised by its content + // (isConfirmAnswer) and only ever delivered here, so it cannot be taken + // as some other request's reply, nor another request's for it. + confirmWaiter *ipcWaiter recvMu sync.Mutex recvChs map[uint32]chan []byte // conn_id → data channel @@ -189,6 +210,7 @@ func newIPCClient(socketPath string) (*ipcClient, error) { c := &ipcClient{ conn: conn, waitSem: make(chan struct{}, 1), + confirmSem: make(chan struct{}, 1), recvChs: make(map[uint32]chan []byte), pendRecv: make(map[uint32][][]byte), pendAccept: make(map[uint16][][]byte), @@ -243,7 +265,7 @@ func (c *ipcClient) readLoop() { cmdDeregisterOK, cmdSetTagsOK, cmdSetWebhookOK, cmdNetworkOK, cmdHealthOK, cmdManagedOK, cmdRotateKeyOK, cmdBroadcastOK, cmdPreferDirectOK, cmdSubmitBadgeOK, cmdEnrollRecoveryOK, - cmdSignEnvelopeOK, cmdVerifyEnvelopeOK: + cmdSignEnvelopeOK, cmdVerifyEnvelopeOK, cmdSendToOK: // Known response cmds: deliver to the active sendAndWait waiter. // If there is no active waiter (the request timed out / was // abandoned, or this is a duplicate), the reply is dropped — @@ -407,6 +429,16 @@ func (c *ipcClient) sendAndWait(data []byte, expectCmd byte) ([]byte, error) { // never blocks. When a frame is delivered the slot is cleared so a second // (duplicate) frame for the same request is dropped rather than re-delivered. func (c *ipcClient) deliverReply(resp *pendingResponse) { + if isConfirmAnswer(resp) { + c.waiterMu.Lock() + w := c.confirmWaiter + c.confirmWaiter = nil + c.waiterMu.Unlock() + if w != nil { + w.ch <- resp + } + return + } c.waiterMu.Lock() w := c.activeWaiter if w == nil || (resp.cmd != w.expect && resp.cmd != cmdError) { @@ -418,6 +450,96 @@ func (c *ipcClient) deliverReply(resp *pendingResponse) { w.ch <- resp } +// isConfirmAnswer reports whether resp is the daemon's answer to a confirmed +// send: cmdSendToOK, or an error from its handler — "sendto: ..." from a +// daemon that supports confirmed sends, "unknown command: 0x39" from one that +// predates them. Fire-and-forget cmdSendTo is never answered, so nothing else +// produces these. +func isConfirmAnswer(resp *pendingResponse) bool { + if resp.cmd == cmdSendToOK { + return true + } + if resp.cmd != cmdError || len(resp.payload) < 2 { + return false + } + msg := string(resp.payload[2:]) + return strings.HasPrefix(msg, "sendto: ") || + strings.HasPrefix(msg, fmt.Sprintf("unknown command: 0x%02X", cmdSendToConfirm)) +} + +// sendConfirmAndWait writes a cmdSendToConfirm request and waits for its +// answer. The IPC has no request IDs, so a confirmed send that times out is +// still answered later. Its answer is recognised by content, so it can never +// be taken as another kind of request's reply; to keep it from being taken as +// the next confirmed send's, confirmSem stays held until it has arrived (or +// the connection is gone). The caller gets its timeout at once and the +// request/reply lane (waitSem) is released then, so other requests are not +// held up; only the next confirmed send waits, and it gets +// ErrConfirmQueueTimeout, nothing sent, if the answer never comes. +// +// queueTimeout bounds the wait for a turn, replyTimeout the wait for the +// answer once the request is written — time spent queued behind another +// request does not count against it. +func (c *ipcClient) sendConfirmAndWait(data []byte, queueTimeout, replyTimeout time.Duration) (*pendingResponse, error) { + qt := time.NewTimer(queueTimeout) + defer qt.Stop() + select { + case c.confirmSem <- struct{}{}: + case <-c.doneCh: + return nil, fmt.Errorf("daemon disconnected") + case <-qt.C: + return nil, ErrConfirmQueueTimeout + } + select { + case c.waitSem <- struct{}{}: + case <-c.doneCh: + <-c.confirmSem + return nil, fmt.Errorf("daemon disconnected") + case <-qt.C: + <-c.confirmSem + return nil, ErrConfirmQueueTimeout + } + w := &ipcWaiter{expect: cmdSendToOK, ch: make(chan *pendingResponse, 1)} + c.waiterMu.Lock() + c.confirmWaiter = w + c.waiterMu.Unlock() + release := func() { + c.waiterMu.Lock() + if c.confirmWaiter == w { + c.confirmWaiter = nil + } + c.waiterMu.Unlock() + <-c.confirmSem + } + if err := c.writeFrame(data[0], data[1:]); err != nil { + <-c.waitSem + release() + return nil, err + } + rt := time.NewTimer(replyTimeout) + defer rt.Stop() + select { + case resp := <-w.ch: + <-c.waitSem + <-c.confirmSem + return resp, nil + case <-c.doneCh: + <-c.waitSem + release() + return nil, fmt.Errorf("daemon disconnected") + case <-rt.C: + } + <-c.waitSem + go func() { + select { + case <-w.ch: + case <-c.doneCh: + } + release() + }() + return nil, ErrConfirmTimeout +} + // registerWaiter installs a fresh active waiter for a request expecting // expect (or cmdError) and returns it. Any previous waiter is replaced; // since waitSem serialises callers there is normally at most one, but diff --git a/driver/zz_driver_test.go b/driver/zz_driver_test.go index 85ae821..f18d7a6 100644 --- a/driver/zz_driver_test.go +++ b/driver/zz_driver_test.go @@ -42,7 +42,8 @@ type fakeDaemon struct { conn net.Conn connSet chan struct{} // closed once conn is stored in acceptLoop mu sync.Mutex - received [][]byte // all frames received + received [][]byte // all frames received + wmu sync.Mutex // serialises frame writes (replies and push) handlers map[byte]func(frame []byte) [][]byte } @@ -90,12 +91,25 @@ func (d *fakeDaemon) acceptLoop() { } } d.mu.Unlock() + d.wmu.Lock() for _, r := range resp { _ = ipcutil.Write(conn, r) } + d.wmu.Unlock() } } +// push writes frame to the driver unprompted, like a late reply. +func (d *fakeDaemon) push(frame []byte) { + <-d.connSet + d.mu.Lock() + conn := d.conn + d.mu.Unlock() + d.wmu.Lock() + defer d.wmu.Unlock() + _ = ipcutil.Write(conn, frame) +} + func (d *fakeDaemon) onCmd(cmd byte, respond func(frame []byte) [][]byte) { d.mu.Lock() defer d.mu.Unlock() diff --git a/driver/zz_sendto_confirmed_test.go b/driver/zz_sendto_confirmed_test.go new file mode 100644 index 0000000..56535c1 --- /dev/null +++ b/driver/zz_sendto_confirmed_test.go @@ -0,0 +1,434 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package driver + +import ( + "errors" + "fmt" + "strings" + "testing" + "time" + + "github.com/pilot-protocol/common/protocol" +) + +func ipcErrorFrame(msg string) []byte { + return append([]byte{cmdError, 0, 1}, msg...) +} + +func framesOf(d *fakeDaemon, cmd byte) int { + n := 0 + for _, f := range d.allFrames() { + if f[0] == cmd { + n++ + } + } + return n +} + +// TestSendToConfirmedReportsDaemonOutcome: against a daemon that supports +// cmdSendToConfirm, the daemon's OK and its error both reach the caller. +func TestSendToConfirmedReportsDaemonOutcome(t *testing.T) { + t.Parallel() + d := newFakeDaemon(t) + defer d.close() + + fail := false + d.onCmd(cmdSendToConfirm, func(frame []byte) [][]byte { + if fail { + return [][]byte{ipcErrorFrame("sendto: ephemeral ports exhausted")} + } + return [][]byte{{cmdSendToOK}} + }) + + drv, err := Connect(d.path) + if err != nil { + t.Fatalf("Connect: %v", err) + } + defer drv.Close() + dst := protocol.Addr{Network: 0, Node: 7} + + confirmed, err := drv.SendToConfirmed(dst, 5000, []byte("hi")) + if err != nil || !confirmed { + t.Fatalf("SendToConfirmed = (%v, %v), want (true, nil)", confirmed, err) + } + want := make([]byte, 1+protocol.AddrSize+2) + want[0] = cmdSendToConfirm + dst.MarshalTo(want, 1) + want[1+protocol.AddrSize], want[2+protocol.AddrSize] = 0x13, 0x88 + if got := d.lastFrame(); string(got) != string(want)+"hi" { + t.Fatalf("frame = %x, want %x", got, string(want)+"hi") + } + + d.mu.Lock() + fail = true + d.mu.Unlock() + confirmed, err = drv.SendToConfirmed(dst, 5000, []byte("hi")) + if confirmed || err == nil || !strings.Contains(err.Error(), "ephemeral ports exhausted") { + t.Fatalf("SendToConfirmed = (%v, %v), want the daemon's error", confirmed, err) + } + if n := framesOf(d, cmdSendTo); n != 0 { + t.Fatalf("%d legacy cmdSendTo frames sent to a daemon that supports confirmed sends", n) + } + + if _, err := drv.SendToConfirmed(protocol.BroadcastAddr(1), 5000, []byte("hi")); err == nil { + t.Fatal("broadcast destination accepted") + } +} + +// TestSendToConfirmedFallsBackOnOlderDaemon: a daemon without the command +// answers "unknown command". The driver then sends the datagram with the +// legacy fire-and-forget command, reports it as unconfirmed, and does not +// probe again. +func TestSendToConfirmedFallsBackOnOlderDaemon(t *testing.T) { + t.Parallel() + d := newFakeDaemon(t) + defer d.close() + + d.onCmd(cmdSendToConfirm, func(frame []byte) [][]byte { + return [][]byte{ipcErrorFrame(fmt.Sprintf("unknown command: 0x%02X", frame[0]))} + }) + d.onCmd(cmdInfo, func(frame []byte) [][]byte { + return [][]byte{append([]byte{cmdInfoOK}, "{}"...)} + }) + + drv, err := Connect(d.path) + if err != nil { + t.Fatalf("Connect: %v", err) + } + defer drv.Close() + dst := protocol.Addr{Network: 0, Node: 7} + + for i := 0; i < 3; i++ { + confirmed, err := drv.SendToConfirmed(dst, 5000, []byte("hi")) + if err != nil || confirmed { + t.Fatalf("call %d: SendToConfirmed = (%v, %v), want (false, nil)", i, confirmed, err) + } + } + // A request/reply round trip puts the fire-and-forget frames behind us. + if _, err := drv.Info(); err != nil { + t.Fatalf("Info: %v", err) + } + if n := framesOf(d, cmdSendToConfirm); n != 1 { + t.Errorf("daemon was probed %d times, want 1", n) + } + if n := framesOf(d, cmdSendTo); n != 3 { + t.Errorf("%d legacy datagrams sent, want 3", n) + } +} + +// shortConfirmTimeout makes SendToConfirmed time out after timeout, for the +// duration of the test. +func shortConfirmTimeout(t *testing.T, timeout time.Duration) { + t.Helper() + ot := sendToConfirmTimeout + sendToConfirmTimeout = timeout + t.Cleanup(func() { sendToConfirmTimeout = ot }) +} + +// A confirmed send that times out is answered by the daemon later; that +// answer must not become the next confirmed send's. In review, a second send +// that the daemon rejected came back confirmed=true because it received the +// first send's late OK. +func TestSendToConfirmedLateOKIsNotTheNextSendsAnswer(t *testing.T) { + shortConfirmTimeout(t, 200*time.Millisecond) + d := newFakeDaemon(t) + defer d.close() + calls := 0 + d.onCmd(cmdSendToConfirm, func(frame []byte) [][]byte { + calls++ + if calls == 1 { + time.Sleep(300 * time.Millisecond) // answered in order, like the daemon + return [][]byte{{cmdSendToOK}} + } + return [][]byte{ipcErrorFrame("sendto: port 5000 not allowed by network 0 policy")} + }) + drv, err := Connect(d.path) + if err != nil { + t.Fatalf("Connect: %v", err) + } + defer drv.Close() + dst := protocol.Addr{Network: 0, Node: 7} + + if ok, err := drv.SendToConfirmed(dst, 5000, []byte("first")); ok || !errors.Is(err, ErrConfirmTimeout) { + t.Fatalf("first send = (%v, %v), want ErrConfirmTimeout", ok, err) + } + ok, err := drv.SendToConfirmed(dst, 5000, []byte("second")) + if ok || err == nil || !strings.Contains(err.Error(), "not allowed") { + t.Fatalf("second send = (%v, %v), want its own rejection", ok, err) + } +} + +// A late answer that is an error, arriving while nothing is waiting, used to +// leave the Driver convinced an answer was still owed, after which every +// confirmed send timed out. The next send must get its own answer. +func TestSendToConfirmedLateErrorWhileIdleDoesNotWedgeTheDriver(t *testing.T) { + shortConfirmTimeout(t, 100*time.Millisecond) + d := newFakeDaemon(t) + defer d.close() + calls := 0 + d.onCmd(cmdSendToConfirm, func(frame []byte) [][]byte { + calls++ + if calls == 1 { + time.Sleep(250 * time.Millisecond) + return [][]byte{ipcErrorFrame("sendto: resolve node 7: not found")} + } + return [][]byte{{cmdSendToOK}} + }) + drv, err := Connect(d.path) + if err != nil { + t.Fatalf("Connect: %v", err) + } + defer drv.Close() + dst := protocol.Addr{Network: 0, Node: 7} + + if ok, _ := drv.SendToConfirmed(dst, 5000, []byte("first")); ok { + t.Fatal("first send should time out") + } + time.Sleep(300 * time.Millisecond) // the late error arrives with nothing waiting + for i := 0; i < 3; i++ { + if ok, err := drv.SendToConfirmed(dst, 5000, []byte("next")); err != nil || !ok { + t.Fatalf("send %d after the late error = (%v, %v), want (true, nil)", i+2, ok, err) + } + } +} + +// A late error from a different request (here an Info that timed out) is +// never taken as a confirmed send's answer. +func TestSendToConfirmedNeverTakesAnotherRequestsAnswer(t *testing.T) { + shortConfirmTimeout(t, 2*time.Second) + d := newFakeDaemon(t) + defer d.close() + d.onCmd(cmdInfo, func(frame []byte) [][]byte { + time.Sleep(300 * time.Millisecond) + return [][]byte{ipcErrorFrame("info: registry unreachable")} + }) + d.onCmd(cmdSendToConfirm, func(frame []byte) [][]byte { + return [][]byte{ipcErrorFrame("sendto: port 5000 not allowed by network 0 policy")} + }) + drv, err := Connect(d.path) + if err != nil { + t.Fatalf("Connect: %v", err) + } + defer drv.Close() + + // The Info request is abandoned before its error arrives. + if _, err := drv.ipc.sendAndWaitTimeout([]byte{cmdInfo}, cmdInfoOK, 100*time.Millisecond); err == nil { + t.Fatal("info should time out") + } + ok, err := drv.SendToConfirmed(protocol.Addr{Network: 0, Node: 7}, 5000, []byte("x")) + if ok || err == nil || !strings.Contains(err.Error(), "not allowed") { + t.Fatalf("confirmed send = (%v, %v), want its own rejection, not the Info's late error", ok, err) + } +} + +// Against a daemon that predates confirmed sends, a first probe that times +// out must not stop the fallback: the next call still learns the daemon is +// old and sends the datagram the legacy way. +func TestSendToConfirmedFallsBackAfterALegacyProbeTimedOut(t *testing.T) { + shortConfirmTimeout(t, 200*time.Millisecond) + d := newFakeDaemon(t) + defer d.close() + calls := 0 + unknown := fmt.Sprintf("unknown command: 0x%02X", cmdSendToConfirm) + d.onCmd(cmdSendToConfirm, func(frame []byte) [][]byte { + calls++ + if calls == 1 { + time.Sleep(250 * time.Millisecond) + } + return [][]byte{ipcErrorFrame(unknown)} + }) + drv, err := Connect(d.path) + if err != nil { + t.Fatalf("Connect: %v", err) + } + defer drv.Close() + dst := protocol.Addr{Network: 0, Node: 7} + + if ok, _ := drv.SendToConfirmed(dst, 5000, []byte("first")); ok { + t.Fatal("first probe should time out") + } + // The second call waits for the first's late answer (50ms on) and then + // gets its own. + ok, err := drv.SendToConfirmed(dst, 5000, []byte("second")) + if ok || err != nil { + t.Fatalf("second send = (%v, %v), want (false, nil) via the legacy fallback", ok, err) + } + deadline := time.Now().Add(2 * time.Second) + for framesOf(d, cmdSendTo) == 0 && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + if framesOf(d, cmdSendTo) == 0 { + t.Fatal("the fallback sent no legacy datagram") + } +} + +// Time spent queued behind another request does not count against the wait +// for the daemon's answer. +func TestSendToConfirmedReplyWaitStartsWhenTheRequestIsWritten(t *testing.T) { + shortConfirmTimeout(t, 300*time.Millisecond) + d := newFakeDaemon(t) + defer d.close() + d.onCmd(cmdInfo, func(frame []byte) [][]byte { + time.Sleep(250 * time.Millisecond) + return [][]byte{append([]byte{cmdInfoOK}, []byte("{}")...)} + }) + d.onCmd(cmdSendToConfirm, func(frame []byte) [][]byte { + time.Sleep(150 * time.Millisecond) + return [][]byte{{cmdSendToOK}} + }) + drv, err := Connect(d.path) + if err != nil { + t.Fatalf("Connect: %v", err) + } + defer drv.Close() + + infoDone := make(chan struct{}) + go func() { + defer close(infoDone) + _, _ = drv.ipc.sendAndWaitTimeout([]byte{cmdInfo}, cmdInfoOK, 5*time.Second) + }() + time.Sleep(20 * time.Millisecond) // the Info holds the lane for ~250ms + ok, err := drv.SendToConfirmed(protocol.Addr{Network: 0, Node: 7}, 5000, []byte("x")) + <-infoDone + if err != nil || !ok { + t.Fatalf("confirmed send queued ~230ms behind an Info and answered 150ms after its write = (%v, %v), want (true, nil)", ok, err) + } +} + +// A confirmed send's late answer never reaches another kind of request, and +// waiting for it does not hold other requests up: an Info sent while the +// answer is still owed gets its own reply, even when the late answer is an +// error frame arriving just before it. +func TestSendToConfirmedLateAnswerDoesNotReachOrHoldOtherRequests(t *testing.T) { + shortConfirmTimeout(t, 100*time.Millisecond) + d := newFakeDaemon(t) + defer d.close() + answered := false + d.onCmd(cmdSendToConfirm, func(frame []byte) [][]byte { + if !answered { + return nil // answered late, below + } + return [][]byte{{cmdSendToOK}} + }) + infoOK := append([]byte{cmdInfoOK}, `{"node_id":7}`...) + d.onCmd(cmdInfo, func(frame []byte) [][]byte { + return [][]byte{infoOK} + }) + drv, err := Connect(d.path) + if err != nil { + t.Fatalf("Connect: %v", err) + } + defer drv.Close() + dst := protocol.Addr{Network: 0, Node: 7} + + if _, err := drv.SendToConfirmed(dst, 5000, []byte("x")); !errors.Is(err, ErrConfirmTimeout) { + t.Fatalf("confirmed send = %v, want ErrConfirmTimeout", err) + } + // The answer is still owed. An Info is not held up by it... + start := time.Now() + if _, err := drv.Info(); err != nil { + t.Fatalf("Info while a confirmed send's answer is owed: %v", err) + } + if waited := time.Since(start); waited > time.Second { + t.Errorf("Info waited %s behind an unanswered confirmed send", waited) + } + // ...and the late error, arriving just before an Info's reply, is not + // taken as that reply. + d.onCmd(cmdInfo, func(frame []byte) [][]byte { + answered = true + return [][]byte{ipcErrorFrame("sendto: resolve node 7: not found"), infoOK} + }) + info, err := drv.Info() + if err != nil { + t.Fatalf("Info took the confirmed send's late error as its reply: %v", err) + } + if info["node_id"] != float64(7) { + t.Errorf("Info = %v", info) + } + // The late answer has gone by, so the next confirmed send gets its own. + deadline := time.Now().Add(2 * time.Second) + for { + ok, err := drv.SendToConfirmed(dst, 5000, []byte("y")) + if errors.Is(err, ErrConfirmQueueTimeout) && time.Now().Before(deadline) { + continue + } + if err != nil || !ok { + t.Fatalf("send after the late answer = (%v, %v), want (true, nil)", ok, err) + } + break + } +} + +// The next confirmed send waits for a late answer however late it is: while +// it is owed, the next send gets ErrConfirmQueueTimeout and nothing is +// written; once it arrives, the next send gets its own answer. In review, a +// time limit on that wait let a very late OK become the next send's answer. +func TestSendToConfirmedWaitsOutAVeryLateAnswer(t *testing.T) { + shortConfirmTimeout(t, 100*time.Millisecond) + d := newFakeDaemon(t) + defer d.close() + calls := 0 + d.onCmd(cmdSendToConfirm, func(frame []byte) [][]byte { + calls++ + if calls == 1 { + return nil // answered late, below + } + return [][]byte{ipcErrorFrame("sendto: port 5000 not allowed by network 0 policy")} + }) + drv, err := Connect(d.path) + if err != nil { + t.Fatalf("Connect: %v", err) + } + defer drv.Close() + dst := protocol.Addr{Network: 0, Node: 7} + + if _, err := drv.SendToConfirmed(dst, 5000, []byte("first")); !errors.Is(err, ErrConfirmTimeout) { + t.Fatalf("first send = %v, want ErrConfirmTimeout", err) + } + for i := 0; i < 3; i++ { + if ok, err := drv.SendToConfirmed(dst, 5000, []byte("waits")); ok || !errors.Is(err, ErrConfirmQueueTimeout) { + t.Fatalf("send while the first answer is owed = (%v, %v), want ErrConfirmQueueTimeout", ok, err) + } + } + if n := framesOf(d, cmdSendToConfirm); n != 1 { + t.Fatalf("%d confirmed sends written while an answer was owed, want only the first", n) + } + d.push([]byte{cmdSendToOK}) // the first send's very late OK + deadline := time.Now().Add(2 * time.Second) + for { + ok, err := drv.SendToConfirmed(dst, 5000, []byte("second")) + if errors.Is(err, ErrConfirmQueueTimeout) && time.Now().Before(deadline) { + continue // the late OK is still on its way + } + if ok || err == nil || !strings.Contains(err.Error(), "not allowed") { + t.Fatalf("send after the late OK = (%v, %v), want its own rejection", ok, err) + } + break + } +} + +// Closing the Driver while a late answer is owed releases the waiter: later +// calls fail at once instead of queueing. +func TestSendToConfirmedCloseWhileAnAnswerIsOwed(t *testing.T) { + shortConfirmTimeout(t, 100*time.Millisecond) + d := newFakeDaemon(t) + defer d.close() + d.onCmd(cmdSendToConfirm, func(frame []byte) [][]byte { return nil }) + drv, err := Connect(d.path) + if err != nil { + t.Fatalf("Connect: %v", err) + } + dst := protocol.Addr{Network: 0, Node: 7} + if _, err := drv.SendToConfirmed(dst, 5000, []byte("x")); !errors.Is(err, ErrConfirmTimeout) { + t.Fatalf("send = %v, want ErrConfirmTimeout", err) + } + drv.Close() + start := time.Now() + if _, err := drv.SendToConfirmed(dst, 5000, []byte("x")); err == nil || errors.Is(err, ErrConfirmQueueTimeout) { + t.Fatalf("send after Close = %v, want a disconnect error", err) + } + if waited := time.Since(start); waited > 50*time.Millisecond { + t.Errorf("send after Close took %s", waited) + } +}