driver: SendToConfirmed reports whether the daemon sent a datagram

driver/driver.go

@@ -9,6 +9,8 @@ import (
 	"os"
 	"path/filepath"
 	"runtime"
+	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/pilot-protocol/common/protocol"
@@ -64,6 +66,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 +180,46 @@ 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 —
+//     in which case the datagram may or may not have left.
+//   - (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.
+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)
+
+	if _, err := d.ipc.sendAndWaitTimeout(msg, cmdSendToOK, defaultDialTimeout); err != nil {
+		if strings.Contains(err.Error(), 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, err
+	}
+	return true, nil
+}
+
 // 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

@@ -79,6 +79,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)
@@ -243,7 +251,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 —

driver/zz_sendto_confirmed_test.go

@@ -0,0 +1,116 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package driver
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+
+	"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)
+	}
+}