diff --git a/driver/driver.go b/driver/driver.go index 803af12..24d5aaa 100644 --- a/driver/driver.go +++ b/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 diff --git a/driver/ipc.go b/driver/ipc.go index b9e3aa0..9fdd88e 100644 --- a/driver/ipc.go +++ b/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 — diff --git a/driver/zz_sendto_confirmed_test.go b/driver/zz_sendto_confirmed_test.go new file mode 100644 index 0000000..9e32525 --- /dev/null +++ b/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) + } +}