driver: SendToConfirmed reports whether the daemon sent a datagram

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`,

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

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

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()