openvibers/openvibe.link Commits

Keep the best server-clock sample for the whole device connection (#5)

OP OpenVibers committed 2 Oct 2026 · 4d2b8e14f6 · 19 files, +842 −178

local checks ✗ 0/2 · CI ✓ 9/9 · review SHIP by DeepSeek harness · 1 fix round

https://github.com/OpenVibers/OpenVibe.Node/pull/5

Changes

ddiff --git a/docs/HANDOFF.md b/docs/HANDOFF.md
iindex cdd6a0a..9241e40 100644
---- a/docs/HANDOFF.md
++++ b/docs/HANDOFF.md
@@@ -52,11 +52,12 @@ mode 0600 and its contents never appear in logs, errors or `status`.
## Safety (ADR-043 decision 6; not optional)
--- Every motion command carries a deadline relative to receipt (`deadline_ms`, default 300 ms, capped by the profile's `max_command_ms`); the
-- plugin stops at the deadline unless a newer command arrived. Held controls are re-sent by the operator every 150 ms.
++- Every motion command carries a deadline (`deadline_ms`: an absolute instant on the server's clock, 300 ms after Bot sent it,
++ capped by the profile's `max_command_ms`); the core nacks `expired` a command that arrives after it and gives the plugin
++ only the time left, and the plugin stops at that deadline unless a newer command arrived. Held controls are re-sent by the operator every 150 ms.
- Stop all actuators on: 2 missed heartbeats, link loss, core shutdown, plugin crash (the core restarts it with backoff
and it starts stopped), `openvibe-node stop`, an `estop`.
--- The e-stop is latched locally (persisted across restarts) and only a server `estop_clear` from the owner (or the local
++- The e-stop is latched locally (persisted across restarts) and only a server `estop` `latched:false` from the owner (or the local
`resume`) clears it. Owner limits (`max_speed`, `max_turn`) clamp every value in the core before it reaches a plugin.
- Never run or expose a kit's stock control server (the Adeept kit ships one on 0.0.0.0:8888 with the fixed login
`admin:123456` and an MJPEG stream on :5000 — the Node replaces it; `install.sh` disables it if it finds it enabled).
ddiff --git a/docs/install.md b/docs/install.md
iindex a032ec1..1eabe0a 100644
---- a/docs/install.md
++++ b/docs/install.md
@@@ -101,8 +101,8 @@ macOS: `/Library/Application Support/OpenVibe Node`. Windows: `%ProgramData%\Ope
e.g. `["ffmpeg", "-f", "v4l2", "-i", "/dev/video0", "-c:v", "libx264", "-preset", "ultrafast", "-tune",
"zerolatency", "-profile:v", "baseline", "-g", "30", "-f", "h264", "-"]`), `plugin` (with `video.plugin`), `off`.
`fps`, `width`, `height`, `ffmpeg` tune the rest. `whip_url`: where the camera publishes, with the publish key from
-- pairing. OpenVibe.Bot hands out the key but no endpoint, so until `video.whip_url` is set (an `https://` URL) video
-- stays off and the log says so.
++ pairing. It overrides the `whip_url` OpenVibe.Bot returns at pairing; with neither (Bot has no ingest configured)
++ video stays off and the log says so.
- `limits`: local caps on top of the owner's limits from openvibe.bot; the stricter value wins.
- `python`: interpreter for bundled plugins (default `<state>/venv/bin/python`, then `python3`).
ddiff --git a/docs/protocol.md b/docs/protocol.md
iindex df46d4c..a9539a4 100644
---- a/docs/protocol.md
++++ b/docs/protocol.md
@@@ -61,9 +61,12 @@ arrives (Bot answers any frame on a not-yet-authenticated socket with `error` `b
| code / answer | meaning | the Node |
|------------------------------|------------------------------------------------------|--------------------------------------------------------|
--| `4000` | a second connection with this credential replaced it | waits 60 s (+ jitter) before dialing, so two agents with one credential do not kick each other off every second |
--| `4002`, or HTTP `401`/`403` | the credential is wrong or revoked | retries after ≥ 10 s; gives up after 3 in a row, "pair again" |
--| `4003` | the owner revoked the device while it was connected | stops the link at once, "pair again" |
++| `4000` | a second connection with this credential replaced it | reconnects with the normal backoff |
++| `4002`, or HTTP `401`/`403` | the credential is wrong, rotated or revoked | retries no sooner than 10 s (jitter only adds), logs "pair again, or install the current credential" |
++| `4003` | the owner revoked the device (or rotated its credential) while it was connected | the same as `4002` |
++
++The link never gives up on its own: the owner may pair the device again. Bot's `error` frames (`bot.not_paired`,
++`bot.not_ready`, `bot.forbidden`, …) are logged and never end the connection.
### Frames
@@@ -86,13 +89,16 @@ carry their own `seq` (the echoed heartbeat number) in the same object as the en
|-----------------|--------------------------------------------------------------------------------------------------|
| `hello` | `session_id`, `device_id`, `robot_ids` [], `server_time` (RFC 3339) — on every authenticated connection |
| `config` | `heartbeat_ms`, `limits` {`max_speed` 0..1, `max_turn` 0..1, `max_command_ms`, `heartbeat_ms`}, `allowed_commands` [kinds], `estop_latched` — right after `hello` |
--| `command` | `id` (idempotency key), `kind`, `value`, `deadline_ms` (absolute, motion kinds only), `operator` {`subject`, `role`}, `robot_id`; Node only: `target` (plugin name) |
++| `command` | `id` (server-minted: the idempotency and ack key), `ref` (the operator's own id, logs only), `kind`, `value`, `deadline_ms` (absolute, motion kinds only), `operator` {`subject`, `role`}, `robot_id`; Node only: `target` (plugin name) |
| `estop` | `latched` (true: the e-stop latched; false: the owner cleared it), `by`, `at` |
| `heartbeat_ack` | `seq` (the heartbeat's), `server_time`; the Node measures RTT from its own send time |
| `error` | `code`, `detail` — a frame the server refused; logged, never fatal |
`config.estop_latched` is the robot's e-stop as the server holds it: an owner e-stop set while the device was offline
--latches the Node on reconnect, and a clear made while offline releases it.
++latches the Node on reconnect, and a clear made while offline releases it. Until a connection's `config` has been
++applied the Node runs nothing but `halt` (`nack not_ready`), so that latch is in force before any motion.
++`config.allowed_commands` absent allows every kind; present (Bot always sends it, `halt` included) it is the list.
++`estop` `latched:false` is the owner's clear: it clears the remote latch only, never the local kill switch.
### Device → server
@@@ -103,11 +109,12 @@ latches the Node on reconnect, and a clear made while offline releases it.
| `ack` | `id`; Node extra: `latency_ms` (receipt → plugin ack) |
| `nack` | `id`, `fault_code`; Node extra: `message` |
| `heartbeat` | `seq`, `rtt_ms` (the last measured round trip, once there is one) |
--| `estop_state` | `latched` (the remote e-stop), `by` (`device`), `at`; Node extras: `local_stop` (local kill switch), `reason` |
++| `estop_state` | `latched` (the device is stopped: remote e-stop or local kill switch), `by` (`device`), `at`; Node extras: `local_stop` (local kill switch), `reason` |
--`status` is sent when `hello` arrives and whenever a plugin changes state or faults; `estop_state` after every
--`config` and whenever the latch changes (never before `config`: Bot takes a device's `estop_state` as the robot's
--latch, so reporting "clear" before learning of an owner e-stop would clear it). Plugin events (`cliff`, `picked_up`,
++The first `status` and `estop_state` go out once the connection's `config` has been applied, never right after the
++upgrade or `hello`; then `status` whenever a plugin changes state or faults and `estop_state` after every `config`,
++`estop` and latch change. Bot reads `latched:true` as the device stopping the robot and latches it (only the owner
++clears that); `latched:false` is a report only. Plugin events (`cliff`, `picked_up`,
`low_battery`, `heartbeat_lost`, …) travel in `telemetry.events` and are sent within 100 ms; sensor telemetry is sent
at most every 500 ms (≤ 2 Hz, Bot's cap). There is no separate `event` message type.
@@@ -127,33 +134,36 @@ Sign conventions: `throttle`/`x` positive = forward; `steer` positive = turn rig
servo in their config to make the hardware match).
`deadline_ms` is an **absolute** instant in Unix milliseconds on the server's clock, present on `drive`, `actuator` and
--`ptz`. A device's clock may be minutes off, so the Node never compares it with its own clock: it takes
--`deadline_ms − ts` (both from the server's clock) as the time the command stays valid and starts that window at
--receipt. A deadline at or before `ts` → `nack bad_value`. No `deadline_ms` → 300 ms. Always capped at `max_command_ms`
--(default 1000). A held control is re-sent by the panel every 150 ms, each with a new `id`. Plugins still get a
--relative `deadline_ms` ([plugins.md](plugins.md)).
++`ptz`. A device's clock may be minutes off, so the Node never compares it with its own clock: it estimates the
++server's clock from the `ts` of every frame on the connection (the largest `ts` − receive time: the skew less the
++fastest transit, kept for the whole connection so a run of delayed frames cannot reopen an expired deadline) and
++compares `deadline_ms` with that at receipt. A command that arrives at or after its deadline →
++`nack expired` and never reaches a plugin; otherwise the plugin gets only the time that is left. No `deadline_ms` →
++300 ms. Always capped at `max_command_ms` (default 1000). A held control is re-sent by the panel every 150 ms, each
++with a new `id`. Plugins still get a relative `deadline_ms` and keep their own stop timer ([plugins.md](plugins.md)).
What the Node does with a command, in order:
1. A repeated `id` is answered with the first result and never executed again (the cache lives for 10 minutes and
across reconnects).
--2. Unknown kind → `nack unsupported`. Kind not in `config.allowed_commands` → `nack not_allowed` (`halt` is always allowed).
++2. Unknown kind → `nack unsupported`.
3. `halt` → every plugin stops, `ack`.
--4. `drive`, `ptz`, `actuator` while the local kill switch is latched → `nack local_stop`; while the remote e-stop is
-- latched → `nack estopped`.
--5. The value is clamped in the core, before any plugin sees it: `throttle`, `x`, `y` to ±`max_speed`; `steer`,
++4. No `config` yet on this connection → `nack not_ready`. Kind not in `config.allowed_commands` → `nack not_allowed`.
++5. `drive`, `ptz`, `actuator` while the local kill switch is latched → `nack local_stop`; while the remote e-stop is
++ latched → `nack estopped`; past its `deadline_ms` → `nack expired`.
++6. The value is clamped in the core, before any plugin sees it: `throttle`, `x`, `y` to ±`max_speed`; `steer`,
`rotation` to ±`max_turn`; `pan`, `tilt`, `zoom` and a numeric actuator `value` to ±1. The limits are the stricter of
the server's `config.limits` and the local config's `limits`. Non-numbers → `nack bad_value`.
--6. It goes to the plugin named by `target`, or the first plugin whose `describe` declares the kind (for actuators, the
++7. It goes to the plugin named by `target`, or the first plugin whose `describe` declares the kind (for actuators, the
first listing the name). None → `nack unsupported`.
--7. The plugin's `ack`/`nack` is relayed. No answer within max(deadline, 1 s) → the plugin is told to stop and the
++8. The plugin's `ack`/`nack` is relayed. No answer within max(deadline, 1 s) → the plugin is told to stop and the
server gets `nack plugin_timeout`.
Commands are never queued while offline and never replayed after a reconnect.
### Fault codes
--`bad_frame`, `bad_value`, `unsupported`, `not_allowed`, `estopped`, `local_stop`, `no_heartbeat`,
++`bad_frame`, `bad_value`, `unsupported`, `not_allowed`, `estopped`, `local_stop`, `expired`, `no_heartbeat`,
`not_ready`, `plugin_down`, `plugin_timeout`, `hardware`, `firmware_unsupported`, `not_connected`, `cliff`,
`picked_up`, `shutting_down`. Plugins may add their own; they are passed through.
ddiff --git a/internal/fakebot/fakebot.go b/internal/fakebot/fakebot.go
iindex 531e65e..18322e6 100644
---- a/internal/fakebot/fakebot.go
++++ b/internal/fakebot/fakebot.go
@@@ -1,8 +1,15 @@
// Package fakebot is a stand-in for OpenVibe.Bot used by tests. It speaks Bot's real device contract (docs/protocol.md
--// section 1 and POST /api/v1/pair, pinned in internal/protocol/testdata/bot): it redeems pairing codes, upgrades every
--// /device request and then authenticates the Authorization header (a bad credential closes with 4002, a revoked
--// device with 4003, a second connection replaces the first with 4000), sends hello and config, and answers
--// heartbeats. Tests drive it to send commands, e-stops and silence, and read back what the device sent.
++// section 1, server/realtime.js and POST /api/v1/pair in server/api/v1.js, pinned in internal/protocol/testdata/bot):
++// it redeems pairing codes (refusals are RFC 9457 problem bodies with Bot's codes), upgrades every /device request
++// and then authenticates the Authorization header (a bad credential closes with 4002, a revoked device with 4003, a
++// second connection replaces the first with 4000), sends hello and then config (always with allowed_commands and
++// the robot's estop_latched), sends commands with a server-minted id, an absolute deadline_ms, the operator and the
++// robot, and answers heartbeats. Like Bot, it keeps the robot's e-stop latch: an estop frame it sends sets or clears
++// it, a device's estop_state latched:true sets it, and latched:false is only a report. Tests drive it to send
++// commands, e-stops, errors and silence, and read back what the device sent.
++//
++// heartbeat_ack still echoes the heartbeat's seq (Bot d398445); Bot now echoes heartbeat.t as echo, and that change
++// is a follow-up on both sides.
package fakebot
import (
@@@ -43,10 +50,17 @@ type Server struct {
live map[string]*Conn // device id → its current connection
estop map[string]bool // robot id → the robot's e-stop latch, as Bot keeps it
pairRequests []protocol.PairRequest
++ pairRefusal *refusal
credInURL bool
conns chan *Conn
count atomic.Int64
++ dials atomic.Int64
++}
++
++type refusal struct {
++ status int
++ code, detail string
}
// RobotID is the robot AddCode and AddCredential attach devices to.
@@@ -116,6 +130,30 @@ func (s *Server) Revoke(cred string) {
}
}
++// RefusePair makes the next POST /api/v1/pair answer with this problem, as Bot's pairing.redeem refuses a code
++// (403 bot.pairing_code_invalid / _locked / _used / _expired, 404 bot.no_pairing_code, 422 bot.invalid_pairing_code,
++// 429 rate_limited).
++func (s *Server) RefusePair(status int, code, detail string) {
++ s.mu.Lock()
++ s.pairRefusal = &refusal{status, code, detail}
++ s.mu.Unlock()
++}
++
++// SetEstop sets or clears the robot's latch as the owner's REST call does: devices learn it from the next config
++// (connected devices get no frame here; send protocol.Estop for that).
++func (s *Server) SetEstop(latched bool) {
++ s.mu.Lock()
++ s.estop[RobotID] = latched
++ s.mu.Unlock()
++}
++
++// EstopLatched reports the robot's latch as Bot would hold it.
++func (s *Server) EstopLatched() bool {
++ s.mu.Lock()
++ defer s.mu.Unlock()
++ return s.estop[RobotID]
++}
++
// PairRequests returns what devices sent to /api/v1/pair.
func (s *Server) PairRequests() []protocol.PairRequest {
s.mu.Lock()
@@@ -130,6 +168,9 @@ func (s *Server) CredentialSeenInURL() bool {
return s.credInURL
}
++// Dials counts every /device request, authenticated or not.
++func (s *Server) Dials() int { return int(s.dials.Load()) }
++
// Connections counts accepted (authenticated) device connections.
func (s *Server) Connections() int { return int(s.count.Load()) }
@@@ -160,6 +201,12 @@ func (s *Server) pair(w http.ResponseWriter, r *http.Request) {
code := strings.ToUpper(strings.ReplaceAll(req.Code, "-", ""))
s.mu.Lock()
s.pairRequests = append(s.pairRequests, req)
++ if rf := s.pairRefusal; rf != nil {
++ s.pairRefusal = nil
++ s.mu.Unlock()
++ problem(w, rf.status, rf.code, rf.detail)
++ return
++ }
robot, ok := s.codes[code]
if ok && req.Robot != "" && req.Robot != robot {
ok = false
@@@ -174,8 +221,8 @@ func (s *Server) pair(w http.ResponseWriter, r *http.Request) {
}
body := s.PairBody
if body == nil {
-- cred, dev := token(), "dev_"+token()[:12]
-- body, _ = json.Marshal(protocol.PairResponse{DeviceID: dev, Credential: cred, PublishKey: "pk_" + token()[:16],
++ cred, dev, pk := token(), "dev_"+token()[:12], "pk_"+token()[:16]
++ body, _ = json.Marshal(protocol.PairResponse{DeviceID: dev, Credential: cred, PublishKey: pk, WHIPURL: s.HTTP.URL + "/whip/" + pk,
RobotID: robot, Profile: json.RawMessage(`{"id":"sim.rover","limits":{"max_command_ms":300,"heartbeat_ms":1000}}`)})
}
var pr protocol.PairResponse
@@@ -192,6 +239,7 @@ func (s *Server) pair(w http.ResponseWriter, r *http.Request) {
var upgrader = websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }}
func (s *Server) device(w http.ResponseWriter, r *http.Request) {
++ s.dials.Add(1)
s.mu.Lock()
for cred := range s.creds {
if strings.Contains(r.URL.RawQuery, cred) || strings.Contains(r.URL.Path, cred) {
@@@ -222,6 +270,9 @@ func (s *Server) device(w http.ResponseWriter, r *http.Request) {
greeting := s.Greeting
cfg := s.Config
cfg.EstopLatched = s.estop[c.RobotID]
++ if cfg.AllowedCommands == nil {
++ cfg.AllowedCommands = append([]string(nil), protocol.Kinds...) // Bot always sends the list, halt included
++ }
c.srv = s
s.mu.Unlock()
if c.DeviceID == "" {
@@@ -278,13 +329,22 @@ type Conn struct {
// Send writes one frame to the device.
func (c *Conn) Send(m protocol.Message) error { return c.sendAt(time.Now().UnixMilli(), m) }
--// Command sends a command the way Bot does: deadline_ms is the absolute instant ts + deadlineMS (none when 0), and
--// the operator is usr_test unless cmd names one.
++// Command sends a command the way Bot does: deadline_ms is the absolute instant ts + deadlineMS (none when 0), ref is
++// the operator's own id, the operator is usr_test unless cmd names one, and robot_id is the device's robot.
func (c *Conn) Command(cmd protocol.Command, deadlineMS int) error {
-- ts := time.Now().UnixMilli()
++ return c.CommandLate(cmd, deadlineMS, 0)
++}
++
++// CommandLate sends a command stamped late ago, as if it had sat in a queue on the way for that long: ts and the
++// deadline are both in the past by late.
++func (c *Conn) CommandLate(cmd protocol.Command, deadlineMS int, late time.Duration) error {
++ ts := time.Now().Add(-late).UnixMilli()
if deadlineMS > 0 {
cmd.DeadlineMS = ts + int64(deadlineMS)
}
++ if cmd.Ref == "" {
++ cmd.Ref = "op_" + cmd.ID
++ }
if cmd.Operator == nil {
cmd.Operator = &protocol.Operator{Subject: "usr_test", Role: "operator"}
}
@@@ -294,7 +354,18 @@ func (c *Conn) Command(cmd protocol.Command, deadlineMS int) error {
return c.sendAt(ts, cmd)
}
--// setEstop records the robot's latch: Bot keeps it from estop frames it sends and estop_state frames it receives.
++// SendLate writes one frame stamped late ago, as if it had sat in a queue on the way for that long.
++func (c *Conn) SendLate(m protocol.Message, late time.Duration) error {
++ return c.sendAt(time.Now().Add(-late).UnixMilli(), m)
++}
++
++// SendError sends Bot's error frame.
++func (c *Conn) SendError(code, detail string) error {
++ return c.Send(protocol.Error{Code: code, Detail: detail})
++}
++
++// setEstop records the robot's latch: Bot keeps it from the estop frames it sends and the latched:true estop_state
++// frames it receives.
func (c *Conn) setEstop(latched bool) {
if c.srv == nil {
return
@@@ -363,8 +434,8 @@ func (c *Conn) read() {
if err != nil {
continue
}
-- if es, ok := f.Msg.(protocol.EstopState); ok {
-- c.setEstop(es.Latched)
++ if es, ok := f.Msg.(protocol.EstopState); ok && es.Latched {
++ c.setEstop(true) // a report of latched:false never clears Bot's latch
}
if hb, ok := f.Msg.(protocol.Heartbeat); ok && c.autoAck.Load() && !c.muted.Load() {
_ = c.Send(protocol.HeartbeatAck{Seq: hb.Seq, ServerTime: time.Now().UTC().Format(time.RFC3339Nano)})
ddiff --git a/internal/link/botfixture_test.go b/internal/link/botfixture_test.go
iindex 41870d5..a0e8a6f 100644
---- a/internal/link/botfixture_test.go
++++ b/internal/link/botfixture_test.go
@@@ -49,7 +49,7 @@ func TestBotFixturesEndToEnd(t *testing.T) {
t.Fatal(err)
}
if creds.DeviceID != "dev_01J8Z4…" || creds.Credential.Reveal() != "Xb3…" || creds.PublishKey.Reveal() != "Vt9…" ||
-- creds.RobotID != "rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X" || len(creds.Profile) == 0 {
++ creds.RobotID != "rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X" || creds.WHIPURL != "https://ingest.openre.stream/whip/Vt9…" || len(creds.Profile) == 0 {
t.Fatalf("credentials %+v", creds)
}
if got := srv.PairRequests(); len(got) != 1 || !reflect.DeepEqual(got[0], req) {
@@@ -76,7 +76,7 @@ func TestBotFixturesEndToEnd(t *testing.T) {
t.Fatalf("hello %+v", h)
}
if cf := waitFrame(t, rec, protocol.TypeConfig).Msg.(protocol.Config); cf.HeartbeatMS != 1000 || cf.Limits.MaxCommandMS != 300 ||
-- len(cf.AllowedCommands) != 6 || cf.EstopLatched {
++ len(cf.AllowedCommands) != 3 || cf.EstopLatched {
t.Fatalf("config %+v", cf)
}
if !l.Connected() {
@@@ -109,6 +109,23 @@ func TestBotFixturesEndToEnd(t *testing.T) {
if !reflect.DeepEqual(r, want.Msg) {
t.Fatalf("ack %+v, want %+v", r, want.Msg)
}
++ // Bot's error frame is logged, never handed on and never fatal.
++ if err := c.SendRaw(botFixture(t, "error")); err != nil {
++ t.Fatal(err)
++ }
++ if err := c.SendRaw(botFixture(t, "estop")); err != nil {
++ t.Fatal(err)
++ }
++ if f := waitFrame(t, rec, protocol.TypeEstop); f.Msg.(protocol.Estop).Latched || !l.Connected() {
++ t.Fatalf("after error and estop: %+v, connected %v", f, l.Connected())
++ }
++ rec.mu.Lock()
++ for _, f := range rec.frames {
++ if f.Type == protocol.TypeError {
++ t.Errorf("error frame reached the handler: %+v", f)
++ }
++ }
++ rec.mu.Unlock()
// Heartbeats carry Bot's seq and are acked by it.
if f, err := c.Expect(protocol.TypeHeartbeat, 2*time.Second, nil); err != nil || f.Msg.(protocol.Heartbeat).Seq != f.Seq {
t.Fatalf("heartbeat %+v %v", f, err)
ddiff --git a/internal/link/link.go b/internal/link/link.go
iindex 95e1454..cfd2961 100644
---- a/internal/link/link.go
++++ b/internal/link/link.go
@@@ -5,9 +5,12 @@
// and the caller decides. The connection counts as up only once the server's hello arrives: OpenVibe.Bot
// authenticates after the upgrade and answers anything sent earlier with error bot.not_paired.
//
--// Close codes: 4000 (another connection with this credential replaced this one) waits ReplacedBackoff before trying
--// again, so two agents sharing a credential do not take turns kicking each other off; 4003 (revoked) ends Run;
--// 4002 (credential refused) ends Run after MaxRefused refusals in a row, so a server blip is survived.
++// Close codes: 4000 (another connection with this credential replaced this one) reconnects with the normal backoff;
++// 4002 (credential refused) and 4003 (revoked) keep retrying, never sooner than CredentialRetryMin, and log what to
++// do: pair again, or install the current credential. Bot's error frames are logged, never fatal.
++//
++// Every server frame's ts also feeds an estimate of the server's clock (ServerNow), which the Node uses to read the
++// absolute deadline_ms of a command.
package link
import (
@@@ -34,8 +37,8 @@ var ErrOffline = errors.New("link offline")
// Handler receives link events. Calls are made from the link's goroutines, one at a time per connection.
type Handler interface {
-- // Connected runs when the server's hello arrives, before the hello itself reaches Frame. The handler sends status
-- // and estop_state here.
++ // Connected runs when the server's hello arrives, before the hello itself reaches Frame. Bot sends config right
++ // after hello; the Node holds its first status and estop_state (and every motion command) until that config.
Connected()
// Frame is every decoded server frame except heartbeat_ack and error.
Frame(f protocol.Frame)
@@@ -55,10 +58,12 @@ type Options struct {
Deadman time.Duration // default 2 × heartbeat
BackoffMin time.Duration // default 500 ms
BackoffMax time.Duration // default 30 s
-- ReplacedBackoff time.Duration // wait after close 4000; default 60 s
-- MaxRefused int // consecutive 4002 / HTTP 401 before giving up; default 3
}
++// CredentialRetryMin is the least a refused or revoked credential (close 4002 / 4003, HTTP 401 / 403) waits before
++// the next try; jitter only ever adds to it.
++const CredentialRetryMin = 10 * time.Second
++
// Link is the reconnecting control connection.
type Link struct {
opt Options
@@@ -72,6 +77,12 @@ type Link struct {
heartbeat time.Duration
deadman time.Duration
++ // The server clock estimate of this connection (see ServerNow): skew is the largest ts − local receive time (ms)
++ // seen since clockBase, the local time of its first frame; hasSkew is false until a frame arrives.
++ skew int64
++ hasSkew bool
++ clockBase time.Time
++
connected atomic.Bool
lastRTT atomic.Int64
reconnects atomic.Int64
@@@ -92,12 +103,6 @@ func New(opt Options, h Handler) *Link {
if opt.BackoffMax <= 0 {
opt.BackoffMax = 30 * time.Second
}
-- if opt.ReplacedBackoff <= 0 {
-- opt.ReplacedBackoff = 60 * time.Second
-- }
-- if opt.MaxRefused <= 0 {
-- opt.MaxRefused = 3
-- }
if opt.Log == nil {
opt.Log = slog.Default()
}
@@@ -126,6 +131,40 @@ func (l *Link) timing() (time.Duration, time.Duration) {
return l.heartbeat, l.deadman
}
++// observe records one server frame's ts, read at local time at.
++func (l *Link) observe(ts int64, at time.Time) {
++ if ts <= 0 {
++ return
++ }
++ l.mu.Lock()
++ defer l.mu.Unlock()
++ if !l.hasSkew {
++ l.clockBase, l.skew, l.hasSkew = at, ts-at.UnixMilli(), true
++ return
++ }
++ l.skew = max(l.skew, ts-l.localMS(at))
++}
++
++// localMS is local time at in epoch ms, measured from clockBase on the monotonic clock so a step of the wall clock
++// cannot move the estimate.
++func (l *Link) localMS(at time.Time) int64 {
++ return l.clockBase.UnixMilli() + at.Sub(l.clockBase).Milliseconds()
++}
++
++// ServerNow estimates the server's clock (epoch ms) at local time at, from the frames of the current connection. The
++// offset is the largest ts − receive time seen on the connection: the clock skew less the fastest transit, so a frame
++// that sat in a queue on the way reads as late by however long it sat there. The best sample is kept for the whole
++// connection: a run of delayed frames can never make an expired deadline look open again. A server clock that steps
++// back only makes deadlines read as expired (fail closed) until the next connection. ok is false before any frame.
++func (l *Link) ServerNow(at time.Time) (ms int64, ok bool) {
++ l.mu.Lock()
++ defer l.mu.Unlock()
++ if !l.hasSkew {
++ return 0, false
++ }
++ return l.localMS(at) + l.skew, true
++}
++
// Connected reports whether a connection is up.
func (l *Link) Connected() bool { return l.connected.Load() }
@@@ -183,10 +222,10 @@ func (l *Link) Send(m protocol.Message) error {
}
}
--// Run connects and reconnects until ctx ends, the credential is revoked (close 4003), or the server has refused the
--// credential MaxRefused times in a row.
++// Run connects and reconnects until ctx ends. A refused or revoked credential does not end it: the owner may pair
++// the device again, and the Node retries every CredentialRetryMin or more, saying so in the log.
func (l *Link) Run(ctx context.Context) {
-- attempt, refused := 0, 0
++ attempt := 0
for ctx.Err() == nil {
start := time.Now()
err := l.session(ctx)
@@@ -195,37 +234,17 @@ func (l *Link) Run(ctx context.Context) {
}
msg := errMsg(err)
l.lastErr.Store(msg)
-- if errors.Is(err, errRevoked) {
-- l.log.Error("link stopped", "err", msg)
-- return
-- }
-- if errors.Is(err, errUnauthorized) {
-- if refused++; refused >= l.opt.MaxRefused {
-- l.log.Error("link stopped: the server refused the credential", "times", refused, "err", msg)
-- return
-- }
-- } else {
-- refused = 0
-- }
if time.Since(start) > 30*time.Second {
attempt = 0
}
-- max := l.opt.BackoffMin << attempt
-- if max > l.opt.BackoffMax || max <= 0 {
-- max = l.opt.BackoffMax
-- }
-- if errors.Is(err, errUnauthorized) && max < 10*time.Second {
-- max = 10 * time.Second
-- }
-- // Full jitter, with a floor so a flapping server is not hammered.
-- wait := l.opt.BackoffMin/2 + time.Duration(rand.Int63n(int64(max)))
-- if errors.Is(err, errReplaced) {
-- // Another agent holds this credential now. Fighting it would flap both; come back much later.
-- wait = l.opt.ReplacedBackoff + time.Duration(rand.Int63n(int64(l.opt.ReplacedBackoff)/4+1))
-- }
++ wait := l.backoff(err, attempt)
attempt++
l.reconnects.Add(1)
-- l.log.Warn("link down; reconnecting", "err", msg, "in", wait.Round(time.Millisecond))
++ if credentialRefused(err) {
++ l.log.Error("link down: the server refused the device credential", "err", msg, "retry_in", wait.Round(time.Millisecond))
++ } else {
++ l.log.Warn("link down; reconnecting", "err", msg, "in", wait.Round(time.Millisecond))
++ }
select {
case <-ctx.Done():
return
@@@ -234,12 +253,33 @@ func (l *Link) Run(ctx context.Context) {
}
}
++// backoff is the wait before reconnect attempt number attempt (0 first) after err: exponential with full jitter and
++// a floor so a flapping server is not hammered. A refused or revoked credential waits CredentialRetryMin plus the
++// jitter, never less.
++func (l *Link) backoff(err error, attempt int) time.Duration {
++ ceil := l.opt.BackoffMax
++ if attempt < 32 {
++ if c := l.opt.BackoffMin << attempt; c > 0 && c < ceil {
++ ceil = c
++ }
++ }
++ jitter := time.Duration(rand.Int63n(int64(ceil) + 1))
++ if credentialRefused(err) {
++ return CredentialRetryMin + jitter
++ }
++ return l.opt.BackoffMin/2 + jitter
++}
++
var (
-- errUnauthorized = errors.New("the server refused the device credential (revoked or rotated?); pair again with `openvibe-node pair <CODE>`")
-- errRevoked = errors.New("the owner revoked this device; pair again with `openvibe-node pair <CODE>`")
++ errUnauthorized = errors.New("the server refused the device credential (wrong, rotated or revoked): pair again with `openvibe-node pair <CODE>`, or install the device's current credential, then restart the Node")
++ errRevoked = errors.New("the owner revoked this device or rotated its credential: pair again with `openvibe-node pair <CODE>`, or install the device's current credential, then restart the Node")
errReplaced = errors.New("another connection with this device's credential replaced this one (is a second agent running with the same credential?)")
)
++func credentialRefused(err error) bool {
++ return errors.Is(err, errUnauthorized) || errors.Is(err, errRevoked)
++}
++
// closeErr maps the server's close codes to the errors Run acts on.
func closeErr(err error) error {
var ce *websocket.CloseError
@@@ -286,7 +326,7 @@ func (l *Link) session(ctx context.Context) error {
// Nothing is sent until the server's hello: until then the socket is not authenticated.
out := make(chan []byte, 256)
l.mu.Lock()
-- l.out, l.seq, l.hbSent = nil, 0, map[uint64]time.Time{}
++ l.out, l.seq, l.hbSent, l.hasSkew = nil, 0, map[uint64]time.Time{}, false
l.mu.Unlock()
sessCtx, cancel := context.WithCancel(ctx)
@@@ -373,7 +413,8 @@ func (l *Link) session(ctx context.Context) error {
}
return endErr
}
-- lastRx.Store(time.Now().UnixNano())
++ rx := time.Now()
++ lastRx.Store(rx.UnixNano())
f, err := protocol.Decode(data)
if err != nil {
if errors.Is(err, protocol.ErrUnknownType) {
@@@ -383,6 +424,7 @@ func (l *Link) session(ctx context.Context) error {
}
continue
}
++ l.observe(f.TS, rx)
switch m := f.Msg.(type) {
case protocol.HeartbeatAck:
l.mu.Lock()
@@@ -395,9 +437,12 @@ func (l *Link) session(ctx context.Context) error {
}
continue
case protocol.Error:
-- if m.Code == protocol.ErrNotPaired {
++ switch m.Code {
++ case protocol.ErrNotPaired:
l.log.Warn("server: this connection is not authenticated yet", "code", m.Code, "detail", m.Detail)
-- } else {
++ case protocol.ErrNotReady:
++ l.log.Warn("server: too many frames before authentication finished", "code", m.Code, "detail", m.Detail)
++ default:
l.log.Warn("server refused a frame", "code", m.Code, "detail", m.Detail)
}
continue
ddiff --git a/internal/link/link_test.go b/internal/link/link_test.go
iindex 611cc02..53ee35a 100644
---- a/internal/link/link_test.go
++++ b/internal/link/link_test.go
@@@ -11,6 +11,8 @@ import (
"testing"
"time"
++ "github.com/gorilla/websocket"
++
"github.com/OpenVibers/OpenVibe.Node/internal/credentials"
"github.com/OpenVibers/OpenVibe.Node/internal/fakebot"
"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
@@@ -188,52 +190,103 @@ func TestSendOffline(t *testing.T) {
}
}
--// A revoke closes the live socket with 4003: the link stops for good and never logs the credential.
--func TestRevokedStopsAndNeverLogsCredential(t *testing.T) {
++// A revoke closes the live socket with 4003: the link says to pair again, keeps trying no sooner than
++// CredentialRetryMin, and never logs the credential.
++func TestRevokedRetriesAfterTenSecondsAndNeverLogsCredential(t *testing.T) {
srv, l, rec, logs, cancel := setup(t, 100*time.Millisecond)
waitFrame(t, rec, protocol.TypeConfig)
srv.Revoke("cred-0123456789abcdef")
-- <-rec.gotDown
-- deadline := time.Now().Add(3 * time.Second)
-- for !strings.Contains(l.Stats().LastError, "pair again") {
-- if time.Now().After(deadline) {
-- t.Fatalf("last error %q", l.Stats().LastError)
-- }
-- time.Sleep(10 * time.Millisecond)
++ if err := <-rec.gotDown; err == nil || !strings.Contains(err.Error(), "revoked") {
++ t.Fatalf("disconnect error %v", err)
}
-- time.Sleep(200 * time.Millisecond) // well past BackoffMax: a running loop would have dialed again
-- if n := srv.Connections(); n != 1 {
-- t.Fatalf("reconnected after a revoke: %d connections", n)
++ waitLastError(t, l, "pair again")
++ if !strings.Contains(l.Stats().LastError, "current credential") {
++ t.Fatalf("last error %q does not say how to recover", l.Stats().LastError)
++ }
++ time.Sleep(time.Second) // ten times BackoffMax: an ordinary reconnect would have dialed again by now
++ if n := srv.Dials(); n != 1 {
++ t.Fatalf("dialed %d times within a second of a revoke", n)
}
cancel()
if strings.Contains(logs.String(), "cred-0123456789abcdef") {
t.Fatalf("credential in logs: %s", logs.String())
}
++ if !strings.Contains(logs.String(), "refused the device credential") {
++ t.Fatalf("no credential diagnosis in the log: %s", logs.String())
++ }
}
--// A credential the server does not know is closed with 4002 after the upgrade; the link gives up after MaxRefused.
--func TestRefusedCredentialGivesUp(t *testing.T) {
++// A credential the server does not know is closed with 4002 right after the upgrade: the link reports it and waits
++// at least CredentialRetryMin before the next try instead of giving up or hammering the server.
++func TestRefusedCredentialWaitsTenSeconds(t *testing.T) {
srv := fakebot.New()
defer srv.Close()
u, _ := DeviceURL(srv.URL())
rec := newRecorder()
-- l := New(Options{URL: u, Credential: credentials.NewSecret("cred-unknown"), BackoffMin: 10 * time.Millisecond, MaxRefused: 1,
-- Log: slog.New(slog.NewTextHandler(io.Discard, nil))}, rec)
++ l := New(Options{URL: u, Credential: credentials.NewSecret("cred-unknown"), BackoffMin: 10 * time.Millisecond,
++ BackoffMax: 50 * time.Millisecond, Log: slog.New(slog.NewTextHandler(io.Discard, nil))}, rec)
++ ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
-- go func() { l.Run(context.Background()); close(done) }()
++ go func() { l.Run(ctx); close(done) }()
++ defer func() { cancel(); <-done }()
++ waitLastError(t, l, "pair again")
++ time.Sleep(time.Second)
++ if n := srv.Dials(); n != 1 {
++ t.Fatalf("dialed %d times within a second of a 4002", n)
++ }
select {
case <-done:
-- case <-time.After(5 * time.Second):
-- t.Fatal("link kept dialing with a refused credential")
++ t.Fatal("the link gave up on a refused credential")
++ default:
++ }
++ rec.mu.Lock()
++ defer rec.mu.Unlock()
++ if rec.connected != 0 {
++ t.Fatalf("connected %d times with a refused credential", rec.connected)
++ }
++}
++
++// The wait after 4002 / 4003 is never under CredentialRetryMin, whatever the jitter draws; other errors keep the
++// short exponential backoff.
++func TestCredentialBackoffFloor(t *testing.T) {
++ l := New(Options{URL: "ws://127.0.0.1:1/device", BackoffMin: time.Millisecond, BackoffMax: 20 * time.Second}, newRecorder())
++ for _, err := range []error{closeErr(&websocket.CloseError{Code: protocol.CloseInvalid}), closeErr(&websocket.CloseError{Code: protocol.CloseRevoked}), errUnauthorized} {
++ if !credentialRefused(err) {
++ t.Fatalf("%v is not a credential refusal", err)
++ }
++ for attempt := 0; attempt < 40; attempt++ {
++ for i := 0; i < 50; i++ {
++ if w := l.backoff(err, attempt); w < CredentialRetryMin || w > CredentialRetryMin+20*time.Second {
++ t.Fatalf("attempt %d after %v: wait %s", attempt, err, w)
++ }
++ }
++ }
++ }
++ replaced := closeErr(&websocket.CloseError{Code: protocol.CloseReplaced})
++ if replaced == nil || credentialRefused(replaced) {
++ t.Fatalf("4000 maps to %v", replaced)
++ }
++ for i := 0; i < 200; i++ {
++ if w := l.backoff(replaced, 0); w > 2*time.Millisecond {
++ t.Fatalf("4000 waited %s on the first attempt", w)
++ }
}
-- if !strings.Contains(l.Stats().LastError, "pair again") || rec.connected != 0 {
-- t.Fatalf("last error %q, connected %d", l.Stats().LastError, rec.connected)
++}
++
++func waitLastError(t *testing.T, l *Link, want string) {
++ t.Helper()
++ deadline := time.Now().Add(3 * time.Second)
++ for !strings.Contains(l.Stats().LastError, want) {
++ if time.Now().After(deadline) {
++ t.Fatalf("last error %q, want %q", l.Stats().LastError, want)
++ }
++ time.Sleep(10 * time.Millisecond)
}
}
--// A second connection with the same credential replaces the first (4000); the replaced link waits ReplacedBackoff
--// instead of dialing straight back and kicking the newcomer off.
--func TestReplacedWaitsBeforeReconnecting(t *testing.T) {
++// A second connection with the same credential replaces the first (4000); the replaced link reconnects with its
++// normal backoff, as Bot expects of a device whose stale socket was replaced.
++func TestReplacedReconnects(t *testing.T) {
srv := fakebot.New()
defer srv.Close()
srv.AddCredential("cred-x", "dev_x")
@@@ -243,17 +296,47 @@ func TestReplacedWaitsBeforeReconnecting(t *testing.T) {
defer cancel()
rec := newRecorder()
first := New(Options{URL: u, Credential: credentials.NewSecret("cred-x"), BackoffMin: 10 * time.Millisecond,
-- ReplacedBackoff: time.Hour, Log: quiet}, rec)
++ BackoffMax: 50 * time.Millisecond, Log: quiet}, rec)
go first.Run(ctx)
waitFrame(t, rec, protocol.TypeConfig)
-- second := New(Options{URL: u, Credential: credentials.NewSecret("cred-x"), Log: quiet}, newRecorder())
++ second := New(Options{URL: u, Credential: credentials.NewSecret("cred-x"), BackoffMin: time.Hour, Log: quiet}, newRecorder())
go second.Run(ctx)
if err := <-rec.gotDown; err == nil || !strings.Contains(err.Error(), "replaced") {
t.Fatalf("err %v", err)
}
-- time.Sleep(300 * time.Millisecond)
-- if n := srv.Connections(); n != 2 || !second.Connected() {
-- t.Fatalf("connections %d, newcomer connected %v", n, second.Connected())
++ // The first link comes back within its normal backoff (and replaces the second in turn).
++ waitFrame(t, rec, protocol.TypeHello)
++ if n := srv.Connections(); n != 3 {
++ t.Fatalf("connections %d", n)
++ }
++}
++
++// ServerNow reads the server's clock from the frames' ts: the skew less the fastest transit.
++func TestServerNow(t *testing.T) {
++ l := New(Options{URL: "ws://127.0.0.1:1/device"}, newRecorder())
++ if _, ok := l.ServerNow(time.Now()); ok {
++ t.Fatal("an estimate before any frame")
++ }
++ l.observe(10_000, time.UnixMilli(9_900)) // server 100 ms ahead, fast frame
++ l.observe(10_000, time.UnixMilli(10_500)) // a frame that sat 600 ms in a queue
++ if ms, ok := l.ServerNow(time.UnixMilli(20_000)); !ok || ms != 20_100 {
++ t.Fatalf("server now %d %v", ms, ok)
++ }
++ // A long run of frames that each sat 500 ms on the way never displaces the timely one: an expired deadline must
++ // not read as open again.
++ for i := 0; i < 100; i++ {
++ l.observe(int64(30_000+i*10), time.UnixMilli(int64(30_000+i*10)+400))
++ }
++ if ms, _ := l.ServerNow(time.UnixMilli(40_000)); ms != 40_100 {
++ t.Fatalf("delayed frames moved the estimate: %d", ms)
++ }
++ // A new connection starts a new estimate.
++ l.mu.Lock()
++ l.hasSkew = false
++ l.mu.Unlock()
++ l.observe(50_000, time.UnixMilli(50_000))
++ if ms, _ := l.ServerNow(time.UnixMilli(60_000)); ms != 60_000 {
++ t.Fatalf("estimate across connections: %d", ms)
}
}
@@@ -323,6 +406,34 @@ func TestPair(t *testing.T) {
}
}
++// Each of Bot's pairing refusals (RFC 9457 problem bodies from pairing.redeem and the rate limiter) becomes a message
++// that says what to do next; anything else keeps Bot's status, code and detail.
++func TestPairRefusals(t *testing.T) {
++ srv := fakebot.New()
++ defer srv.Close()
++ req := protocol.PairRequest{Robot: fakebot.RobotID, Code: "ABCD1234", AgentVersion: "t", DeviceKind: "onboard"}
++ for _, tc := range []struct {
++ status int
++ code, detail string
++ want string
++ }{
++ {403, "bot.pairing_code_invalid", "that is not the pairing code", "not the pairing code for this robot"},
++ {403, "bot.pairing_code_locked", "too many wrong tries; the code is dead", "the code is dead"},
++ {403, "bot.pairing_code_used", "that pairing code has already been used", "already used"},
++ {403, "bot.pairing_code_expired", "that pairing code has expired", "codes last 10 minutes"},
++ {404, "bot.no_pairing_code", "this robot has no pairing code; ask the owner for a new one", "no live pairing code"},
++ {422, "bot.invalid_pairing_code", "the pairing code must be 8 characters (XXXX-XXXX)", "like ABCD-1234"},
++ {429, "rate_limited", "", "wait a minute"},
++ {422, "bot.invalid_input", "drivers must be a list", "HTTP 422 bot.invalid_input drivers must be a list"},
++ } {
++ srv.RefusePair(tc.status, tc.code, tc.detail)
++ _, err := Pair(context.Background(), nil, srv.URL(), req)
++ if err == nil || !strings.Contains(err.Error(), tc.want) {
++ t.Errorf("%s: %v, want %q", tc.code, err, tc.want)
++ }
++ }
++}
++
func TestNormalizeCode(t *testing.T) {
for in, want := range map[string]string{"abcd-1234": "ABCD1234", " WXYZ 7890 ": "WXYZ7890", "OOII-LLOO": "00111100"} {
got, err := NormalizeCode(in)
ddiff --git a/internal/link/pair.go b/internal/link/pair.go
iindex 2d19cf1..55c5bf8 100644
---- a/internal/link/pair.go
++++ b/internal/link/pair.go
@@@ -41,6 +41,7 @@ var pairMessages = map[string]string{
"bot.pairing_code_used": "the pairing code was already used; ask for a new one on openvibe.bot",
"bot.pairing_code_expired": "the pairing code expired (codes last 10 minutes); ask for a new one on openvibe.bot",
"bot.no_pairing_code": "this robot has no live pairing code; ask for a new one on openvibe.bot",
++ "rate_limited": "too many pairing attempts from here; wait a minute and try again",
}
// Pair redeems a one-time code with POST <server>/api/v1/pair.
@@@ -99,8 +100,8 @@ func Pair(ctx context.Context, client *http.Client, server string, req protocol.
}
return &credentials.Credentials{
DeviceID: pr.DeviceID, RobotID: pr.RobotID, Credential: credentials.NewSecret(pr.Credential),
-- PublishKey: credentials.NewSecret(pr.PublishKey), Server: strings.TrimSuffix(server, "/"), Profile: pr.Profile,
-- PairedAt: time.Now().UTC(),
++ PublishKey: credentials.NewSecret(pr.PublishKey), WHIPURL: pr.WHIPURL, Server: strings.TrimSuffix(server, "/"),
++ Profile: pr.Profile, PairedAt: time.Now().UTC(),
}, nil
}
ddiff --git a/internal/node/botcontract_test.go b/internal/node/botcontract_test.go
nnew file mode 100644
iindex 0000000..40a567b
---- /dev/null
++++ b/internal/node/botcontract_test.go
@@@ -0,0 +1,316 @@
++//go:build !windows
++
++package node
++
++import (
++ "bytes"
++ "encoding/json"
++ "os"
++ "path/filepath"
++ "testing"
++ "time"
++
++ "github.com/OpenVibers/OpenVibe.Node/internal/fakebot"
++ "github.com/OpenVibers/OpenVibe.Node/internal/localctl"
++ "github.com/OpenVibers/OpenVibe.Node/internal/plugins"
++ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++)
++
++// These tests drive the Node with the frames OpenVibe.Bot sends, recorded from its docs/protocol.md into
++// internal/protocol/testdata/bot (see its README for the Bot commit).
++
++func botFrame(t *testing.T, name string) []byte {
++ t.Helper()
++ b, err := os.ReadFile(filepath.Join("..", "protocol", "testdata", "bot", name+".json"))
++ if err != nil {
++ t.Fatal(err)
++ }
++ return bytes.TrimSpace(b)
++}
++
++// restamp moves a recorded frame to now: ts becomes now and deadline_ms keeps its distance from ts.
++func restamp(t *testing.T, frame []byte) []byte {
++ t.Helper()
++ var m map[string]any
++ if err := json.Unmarshal(frame, &m); err != nil {
++ t.Fatal(err)
++ }
++ now := time.Now().UnixMilli()
++ ts, _ := m["ts"].(float64)
++ if d, ok := m["deadline_ms"].(float64); ok {
++ m["deadline_ms"] = now + int64(d-ts)
++ }
++ m["ts"] = now
++ b, err := json.Marshal(m)
++ if err != nil {
++ t.Fatal(err)
++ }
++ return b
++}
++
++func (e *env) commandRecords(id string) []rec {
++ var out []rec
++ for _, r := range e.records() {
++ if r.What == "command" && r.ID == id {
++ out = append(out, r)
++ }
++ }
++ return out
++}
++
++func (e *env) expectEstopState(latched, local bool) {
++ e.t.Helper()
++ if _, err := e.conn.Expect(protocol.TypeEstopState, 5*time.Second, func(m protocol.Message) bool {
++ s := m.(protocol.EstopState)
++ return s.Latched == latched && s.LocalStop == local
++ }); err != nil {
++ e.t.Fatalf("estop_state latched %v local %v: %v", latched, local, err)
++ }
++}
++
++func (e *env) ctl(cmd string) {
++ e.t.Helper()
++ if r, err := localctl.Call(e.paths.SocketPath(), localctl.Request{Cmd: cmd}, 2*time.Second); err != nil || !r.OK {
++ e.t.Fatalf("%s: %+v %v", cmd, r, err)
++ }
++}
++
++// pair → hello + config → drive → ack, with Bot's own frames (only ts and deadline_ms moved to now).
++func TestBotFramesDriveAndAck(t *testing.T) {
++ e := startWith(t, func(s *fakebot.Server) {
++ s.PairBody = protocol.PairBodyFromPaired(botFrame(t, "paired"))
++ s.Greeting = [][]byte{restamp(t, botFrame(t, "hello")), restamp(t, botFrame(t, "config"))}
++ }, true)
++ if c := e.node.opt.Creds; c.DeviceID != "dev_01J8Z4…" || c.RobotID != "rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X" || c.WHIPURL == "" {
++ t.Fatalf("paired %+v", c)
++ }
++ // The server-minted id is the ack and dedup key: Bot never re-sends one, and if it came twice it would run once
++ // and be acked twice.
++ const id = "cmd_01J8Z4F…"
++ for i := 0; i < 2; i++ {
++ if err := e.conn.SendRaw(restamp(t, botFrame(t, "command_drive"))); err != nil {
++ t.Fatal(err)
++ }
++ r, err := e.conn.Reply(id, 5*time.Second)
++ if err != nil {
++ t.Fatal(err)
++ }
++ if !isAck(r) {
++ t.Fatalf("reply %+v", r)
++ }
++ }
++ got := e.commandRecords(id)
++ if len(got) != 1 {
++ t.Fatalf("ran %d times", len(got))
++ }
++ var v map[string]float64
++ _ = json.Unmarshal(got[0].Value, &v)
++ if v["throttle"] != 0.6 || v["steer"] != -0.25 || got[0].DeadlineMS < 150 || got[0].DeadlineMS > 300 {
++ t.Fatalf("plugin got %s for %d ms", got[0].Value, got[0].DeadlineMS)
++ }
++}
++
++// deadline_ms is an absolute instant: a drive that arrives after it is nacked and never reaches the plugin, and one
++// that arrives late gets only what is left of its window.
++func TestExpiredDriveNeverReachesPlugin(t *testing.T) {
++ e := start(t)
++ // Bot's example drive, replayed as recorded: its deadline passed in 2025.
++ if err := e.conn.SendRaw(botFrame(t, "command_drive")); err != nil {
++ t.Fatal(err)
++ }
++ r, err := e.conn.Reply("cmd_01J8Z4F…", 5*time.Second)
++ if err != nil {
++ t.Fatal(err)
++ }
++ if c := nackCode(r); c != protocol.FaultExpired {
++ t.Fatalf("recorded drive: %+v", r)
++ }
++ // A drive that sat 600 ms on the way: its 300 ms window closed before it arrived.
++ send := func(id string, late time.Duration) protocol.Message {
++ t.Helper()
++ if err := e.conn.CommandLate(protocol.Command{ID: id, Kind: protocol.KindDrive, Value: json.RawMessage(`{"throttle":0.3}`)}, 300, late); err != nil {
++ t.Fatal(err)
++ }
++ r, err := e.conn.Reply(id, 5*time.Second)
++ if err != nil {
++ t.Fatal(err)
++ }
++ return r
++ }
++ if r := send("late-600", 600*time.Millisecond); nackCode(r) != protocol.FaultExpired {
++ t.Fatalf("late drive: %+v", r)
++ }
++ // One that sat 100 ms runs for the remaining ~200 ms (less transit, give or take the clock estimate's rounding),
++ // not the full 300.
++ if r := send("late-100", 100*time.Millisecond); !isAck(r) {
++ t.Fatalf("partly late drive: %+v", r)
++ }
++ if got := e.commandRecords("late-100"); len(got) != 1 || got[0].DeadlineMS < 50 || got[0].DeadlineMS > 210 {
++ t.Fatalf("partly late drive reached the plugin as %+v", got)
++ }
++ if len(e.commandRecords("cmd_01J8Z4F…")) != 0 || len(e.commandRecords("late-600")) != 0 {
++ t.Fatalf("an expired drive reached the plugin: %+v", e.records())
++ }
++}
++
++// The clock estimate keeps the timely hello/config for the whole connection: after a long run of frames that each
++// sat 500 ms on the way, a drive whose 300 ms window closed on the way is still nacked, not handed 300 ms.
++func TestExpiredDriveAfterDelayedFrames(t *testing.T) {
++ e := start(t)
++ const late = 500 * time.Millisecond
++ for i := 0; i < 40; i++ {
++ if err := e.conn.SendLate(protocol.Error{Code: "test.delayed", Detail: "a frame that sat in a queue"}, late); err != nil {
++ t.Fatal(err)
++ }
++ }
++ const id = "late-after-delayed"
++ if err := e.conn.CommandLate(protocol.Command{ID: id, Kind: protocol.KindDrive, Value: json.RawMessage(`{"throttle":0.3}`)}, 300, late); err != nil {
++ t.Fatal(err)
++ }
++ r, err := e.conn.Reply(id, 5*time.Second)
++ if err != nil {
++ t.Fatal(err)
++ }
++ if c := nackCode(r); c != protocol.FaultExpired {
++ t.Fatalf("expired drive after delayed frames: %+v", r)
++ }
++ if got := e.commandRecords(id); len(got) != 0 {
++ t.Fatalf("an expired drive reached the plugin: %+v", got)
++ }
++}
++
++// Bot's config for the Adeept allows drive, actuator and halt: any other kind is nacked on the device and never
++// reaches the plugin.
++func TestDisallowedKindNacked(t *testing.T) {
++ e := startWith(t, func(s *fakebot.Server) {
++ f, err := protocol.Decode(botFrame(t, "config"))
++ if err != nil {
++ t.Fatal(err)
++ }
++ s.Config = f.Msg.(protocol.Config)
++ }, true)
++ for _, kind := range []string{protocol.KindSay, protocol.KindPTZ, protocol.KindDisplay} {
++ if c := nackCode(e.cmd(kind, `{"text":"hi"}`, 300)); c != protocol.FaultNotAllowed {
++ t.Fatalf("%s: %q", kind, c)
++ }
++ }
++ if r := e.cmd(protocol.KindDrive, `{"throttle":0.2}`, 300); !isAck(r) {
++ t.Fatalf("drive: %+v", r)
++ }
++ if r := e.cmd(protocol.KindHalt, `{}`, 0); !isAck(r) {
++ t.Fatalf("halt: %+v", r)
++ }
++ for _, r := range e.records() {
++ if r.What == "command" && r.Kind != protocol.KindDrive {
++ t.Fatalf("a disallowed kind reached the plugin: %+v", r)
++ }
++ }
++}
++
++// Bot sends config right after hello. Until it arrives the device reports nothing and runs nothing but halt, so an
++// owner latch set while it was offline is in force before any motion.
++func TestNothingBeforeConfig(t *testing.T) {
++ e := startWith(t, func(s *fakebot.Server) {
++ s.Greeting = [][]byte{restamp(t, botFrame(t, "hello"))}
++ s.SetEstop(true)
++ }, false)
++ if c := nackCode(e.cmdID("early", protocol.KindDrive, `{"throttle":0.3}`, 300)); c != protocol.FaultNotReady {
++ t.Fatalf("drive before config: %q", c)
++ }
++ if !isAck(e.cmdID("early-halt", protocol.KindHalt, `{}`, 0)) {
++ t.Fatal("halt refused before config")
++ }
++ deadline := time.Now().Add(20 * time.Second)
++ for e.node.mgr.Get("dryrun").Info().State != plugins.StateReady {
++ if time.Now().After(deadline) {
++ t.Fatal("plugin not ready")
++ }
++ time.Sleep(20 * time.Millisecond)
++ }
++ for _, f := range e.conn.Frames() {
++ if f.Type == protocol.TypeStatus || f.Type == protocol.TypeEstopState {
++ t.Fatalf("%s sent before config", f.Type)
++ }
++ }
++ cfg := e.srv.Config
++ cfg.EstopLatched = e.srv.EstopLatched()
++ if err := e.conn.Send(cfg); err != nil {
++ t.Fatal(err)
++ }
++ e.waitReady()
++ e.expectEstopState(true, false)
++ if c := nackCode(e.cmd(protocol.KindDrive, `{"throttle":0.3}`, 300)); c != protocol.FaultEstopped {
++ t.Fatalf("drive after a latched config: %q", c)
++ }
++ if len(e.commandRecords("early")) != 0 {
++ t.Fatal("the drive before config reached the plugin")
++ }
++}
++
++// The owner's latch, set or cleared live or while the device is offline, is what the device holds after each
++// reconnect.
++func TestOwnerLatchAcrossReconnects(t *testing.T) {
++ e := start(t)
++ drive := func() protocol.Message { return e.cmd(protocol.KindDrive, `{"throttle":0.3}`, 300) }
++ reconnect := func() {
++ e.conn.Close()
++ e.conn = e.nextConn()
++ }
++ e.conn.Send(protocol.Estop{Latched: true, By: "usr_owner", At: time.Now().UTC().Format(time.RFC3339Nano)})
++ e.expectEstopState(true, false)
++ if c := nackCode(drive()); c != protocol.FaultEstopped {
++ t.Fatalf("latched live: %q", c)
++ }
++ // Cleared while offline: the next config says so.
++ e.srv.SetEstop(false)
++ reconnect()
++ e.expectEstopState(false, false)
++ if r := drive(); !isAck(r) {
++ t.Fatalf("cleared offline: %+v", r)
++ }
++ // Latched while offline: the first drive after the reconnect is already refused.
++ e.srv.SetEstop(true)
++ reconnect()
++ if c := nackCode(drive()); c != protocol.FaultEstopped {
++ t.Fatalf("latched offline: %q", c)
++ }
++ // Cleared live, with Bot's recorded estop frame (latched:false).
++ if err := e.conn.SendRaw(botFrame(t, "estop")); err != nil {
++ t.Fatal(err)
++ }
++ e.expectEstopState(false, false)
++ if r := drive(); !isAck(r) {
++ t.Fatalf("cleared live: %+v", r)
++ }
++}
++
++// A remote clear never clears the local latch (`openvibe-node stop`): the device keeps refusing motion and keeps
++// reporting itself latched, across a reconnect, until the local resume (which clears both latches on the device).
++// Its latched:false after the resume is only a report; Bot's latch stays for the owner to clear.
++func TestLocalLatchSurvivesRemoteClear(t *testing.T) {
++ e := start(t)
++ e.ctl(localctl.CmdStop)
++ e.expectEstopState(true, true)
++ if !e.srv.EstopLatched() {
++ t.Fatal("Bot did not latch the robot from the device's report")
++ }
++ e.conn.Send(protocol.Estop{Latched: false, By: "usr_owner", At: time.Now().UTC().Format(time.RFC3339Nano)})
++ e.expectEstopState(true, true)
++ if !e.srv.EstopLatched() {
++ t.Fatal("the device did not report its local latch after the owner's clear")
++ }
++ if c := nackCode(e.cmd(protocol.KindDrive, `{"throttle":0.3}`, 300)); c != protocol.FaultLocalStop {
++ t.Fatalf("after the remote clear: %q", c)
++ }
++ e.conn.Close()
++ e.conn = e.nextConn()
++ e.expectEstopState(true, true)
++ if c := nackCode(e.cmd(protocol.KindDrive, `{"throttle":0.3}`, 300)); c != protocol.FaultLocalStop {
++ t.Fatalf("after a reconnect: %q", c)
++ }
++ e.ctl(localctl.CmdResume)
++ e.expectEstopState(false, false)
++ if !e.srv.EstopLatched() {
++ t.Fatal("the device's latched:false cleared Bot's latch")
++ }
++}
ddiff --git a/internal/node/node.go b/internal/node/node.go
iindex 96ae687..2fc0033 100644
---- a/internal/node/node.go
++++ b/internal/node/node.go
@@@ -72,6 +72,10 @@ type Node struct {
videoSrc string
lastTelem time.Time
started time.Time
++
++ // configured is set by the first config on a connection; until then only halt runs, so a latch the owner set
++ // while the link was down is applied before any motion.
++ configured bool
}
// New builds a Node. Run starts it.
@@@ -125,7 +129,7 @@ func New(opt Options) (*Node, error) {
}
n.link = link.New(link.Options{URL: u, Credential: opt.Creds.Credential, UserAgent: "openvibe-node/" + opt.Version,
Log: opt.Log, HeartbeatInterval: opt.HeartbeatInterval, BackoffMin: opt.LinkBackoffMin}, n)
-- // Bot hands out only the publish key; the WHIP endpoint is local config (or an older pairing's stored URL).
++ // The WHIP endpoint is the pairing's whip_url (built by Bot from the publish key) unless the config sets one.
whip := opt.Config.Video.WHIPURL
if whip == "" {
whip = opt.Creds.WHIPURL
@@@ -185,15 +189,20 @@ func (n *Node) Run(ctx context.Context) error {
// ---- link.Handler ----
--// Connected sends status. estop_state waits for config: Bot treats a device's estop_state as the robot's latch, so
--// reporting "clear" before config told us about an owner e-stop set while we were offline would clear it.
++// Connected sends nothing: Bot sends config right after hello, and the first status and estop_state go out once it
++// has been applied (applyConfig), so they report the latch the owner may have set while the link was down.
func (n *Node) Connected() {
-- n.sendStatus()
++ n.mu.Lock()
++ n.configured = false
++ n.mu.Unlock()
}
func (n *Node) Disconnected(err error) {
n.log.Warn("link lost: stopping every actuator", "err", err)
n.mgr.StopAll()
++ n.mu.Lock()
++ n.configured = false
++ n.mu.Unlock()
}
func (n *Node) Frame(f protocol.Frame) {
@@@ -205,6 +214,7 @@ func (n *Node) Frame(f protocol.Frame) {
case protocol.Command:
n.command(m, f.TS, time.Now())
case protocol.Estop:
++ // latched:false is the owner's clear. It clears only the remote latch: a local (CLI or file) stop stays.
if m.Latched {
n.remoteEstop("by "+m.By, m.By)
} else {
@@@ -236,18 +246,23 @@ func (n *Node) applyConfig(c protocol.Config) {
} else if !c.EstopLatched && st.Remote {
n.remoteClear("server")
}
-- n.sendEstopState()
l := safety.Merge(c.Limits, n.opt.Config.Limits.MaxSpeed, n.opt.Config.Limits.MaxTurn, n.opt.Config.Limits.MaxCommandMS)
n.mu.Lock()
n.limits = l
n.allowed = nil
-- if len(c.AllowedCommands) > 0 {
++ if c.AllowedCommands != nil {
n.allowed = map[string]bool{protocol.KindHalt: true}
for _, k := range c.AllowedCommands {
n.allowed[k] = true
}
}
++ first := !n.configured
++ n.configured = true
n.mu.Unlock()
++ if first {
++ n.sendStatus()
++ }
++ n.sendEstopState()
if n.link != nil {
hb := c.HeartbeatMS
if hb <= 0 {
@@@ -255,17 +270,22 @@ func (n *Node) applyConfig(c protocol.Config) {
}
n.link.SetTiming(hb)
}
-- n.log.Info("config", "max_speed", l.MaxSpeed, "max_turn", l.MaxTurn, "max_command_ms", l.MaxCommandMS)
++ n.log.Info("config", "max_speed", l.MaxSpeed, "max_turn", l.MaxTurn, "max_command_ms", l.MaxCommandMS,
++ "allowed", c.AllowedCommands, "estop_latched", c.EstopLatched)
}
// command validates and routes one command. The synchronous part (checks, clamping, queueing to the plugin) keeps
--// commands in arrival order; the reply is awaited on its own goroutine. sentMS is the frame's ts: the absolute
--// deadline_ms minus it is the time the server allowed, on the server's own clock, so clock skew does not matter.
++// commands in arrival order; the reply is awaited on its own goroutine. deadline_ms is an absolute instant on the
++// server's clock: it is compared with the link's estimate of that clock at receipt (serverNow), so a command that
++// arrives after its deadline is nacked and never reaches a plugin, and a plugin gets only the time that is left.
func (n *Node) command(c protocol.Command, sentMS int64, received time.Time) {
if c.ID == "" {
n.log.Warn("command without id ignored", "kind", c.Kind)
return
}
++ if c.Operator != nil {
++ n.log.Debug("command", "id", c.ID, "ref", c.Ref, "kind", c.Kind, "robot", c.RobotID, "operator", c.Operator.Subject, "role", c.Operator.Role)
++ }
entry, first := n.dedup.Begin(c.ID)
if !first {
go func() {
@@@ -291,7 +311,7 @@ func (n *Node) command(c protocol.Command, sentMS int64, received time.Time) {
return
}
n.mu.Lock()
-- limits, allowed := n.limits, n.allowed
++ limits, allowed, configured := n.limits, n.allowed, n.configured
n.mu.Unlock()
if c.Kind == protocol.KindHalt {
n.mgr.StopAll()
@@@ -300,8 +320,12 @@ func (n *Node) command(c protocol.Command, sentMS int64, received time.Time) {
n.send(r)
return
}
++ if !configured {
++ nack(protocol.FaultNotReady, "the server's config has not arrived on this connection yet")
++ return
++ }
if allowed != nil && !allowed[c.Kind] {
-- nack(protocol.FaultNotAllowed, "")
++ nack(protocol.FaultNotAllowed, c.Kind+" is not in this robot's allowed_commands")
return
}
if protocol.GuardedKinds[c.Kind] {
@@@ -315,6 +339,15 @@ func (n *Node) command(c protocol.Command, sentMS int64, received time.Time) {
return
}
}
++ deadline := limits.Deadline(0)
++ if c.DeadlineMS > 0 {
++ left := c.DeadlineMS - n.serverNow(sentMS, received)
++ if left <= 0 {
++ nack(protocol.FaultExpired, fmt.Sprintf("the deadline passed %d ms before the command arrived", -left))
++ return
++ }
++ deadline = limits.Deadline(int(min(left, int64(limits.MaxCommandMS)+1)))
++ }
value, err := limits.Clamp(c.Kind, c.Value)
if err != nil {
var fe *safety.FaultError
@@@ -338,19 +371,6 @@ func (n *Node) command(c protocol.Command, sentMS int64, received time.Time) {
nack(protocol.FaultUnsupported, "no driver on this device handles "+c.Kind)
return
}
-- deadline := limits.Deadline(0)
-- if c.DeadlineMS > 0 {
-- ref := sentMS
-- if ref <= 0 {
-- ref = received.UnixMilli()
-- }
-- left := c.DeadlineMS - ref
-- if left <= 0 {
-- nack(protocol.FaultBadValue, "the command's deadline had already passed")
-- return
-- }
-- deadline = limits.Deadline(int(min(left, int64(limits.MaxCommandMS)+1)))
-- }
wait := p.Begin(n.ctx, c.ID, c.Kind, value, deadline)
go func() {
r := wait()
@@@ -365,6 +385,20 @@ func (n *Node) command(c protocol.Command, sentMS int64, received time.Time) {
}()
}
++// serverNow is the server's clock when a frame stamped sentMS was received: the link's estimate, else the frame's
++// own ts.
++func (n *Node) serverNow(sentMS int64, received time.Time) int64 {
++ if n.link != nil {
++ if ms, ok := n.link.ServerNow(received); ok {
++ return ms
++ }
++ }
++ if sentMS > 0 {
++ return sentMS
++ }
++ return received.UnixMilli()
++}
++
func (n *Node) send(m protocol.Message) {
if n.link == nil {
return
@@@ -403,9 +437,15 @@ func (n *Node) pollLatch(ctx context.Context) {
}
}
++// sendEstopState reports the device's latch. Bot reads latched:true as "this device is stopped" and latches the robot
++// (so the local kill switch shows on the panel and the gate stops sending motion); latched:false is a report only
++// and never clears the owner's latch.
func (n *Node) sendEstopState() {
++ if !n.reporting() {
++ return
++ }
st := n.latch.State()
-- n.send(protocol.EstopState{Latched: st.Remote, By: "device", At: time.Now().UTC().Format(time.RFC3339Nano),
++ n.send(protocol.EstopState{Latched: st.Stopped(), By: "device", At: time.Now().UTC().Format(time.RFC3339Nano),
LocalStop: st.Local, Reason: st.RemoteReason})
}
@@@ -695,10 +735,20 @@ func (n *Node) Status() Status {
return s
}
++// reporting is whether status and estop_state may go out: only once this connection's config has been applied.
++func (n *Node) reporting() bool {
++ n.mu.Lock()
++ defer n.mu.Unlock()
++ return n.configured
++}
++
func (n *Node) sendStatus() {
++ if !n.reporting() {
++ return
++ }
st := n.latch.State()
s := protocol.Status{Firmware: "openvibe-node-" + n.opt.Version, Capabilities: map[string]any{}, AgentVersion: n.opt.Version,
-- DeviceKind: n.opt.Config.DeviceKind, OS: runtime.GOOS, Arch: runtime.GOARCH, EstopLatched: st.Remote, LocalStop: st.Local,
++ DeviceKind: n.opt.Config.DeviceKind, OS: runtime.GOOS, Arch: runtime.GOARCH, EstopLatched: st.Stopped(), LocalStop: st.Local,
Drivers: []protocol.DriverStatus{}, Faults: []protocol.Fault{}}
for _, i := range n.mgr.Infos() {
d := protocol.DriverStatus{Name: i.Name, State: i.State}
ddiff --git a/internal/node/node_test.go b/internal/node/node_test.go
iindex cb707fb..b99694c 100644
---- a/internal/node/node_test.go
++++ b/internal/node/node_test.go
@@@ -49,6 +49,13 @@ type env struct {
// start pairs a Node running the real Python dry-run plugin with the fake Bot server and waits until the plugin is
// ready and the link is up.
func start(t *testing.T) *env {
++ t.Helper()
++ return startWith(t, nil, true)
++}
++
++// startWith is start with prep run on the fake server before pairing; ready false returns once the device has
++// connected, without waiting for its first status.
++func startWith(t *testing.T, prep func(*fakebot.Server), ready bool) *env {
t.Helper()
if runtime.GOOS == "windows" {
t.Skip("end-to-end tests run on unix")
@@@ -67,8 +74,12 @@ func start(t *testing.T) *env {
srv := fakebot.New()
t.Cleanup(srv.Close)
-- srv.Config = protocol.Config{HeartbeatMS: 100, Limits: protocol.Limits{MaxSpeed: f64(0.5), MaxTurn: f64(0.4), MaxCommandMS: 1000}}
++ srv.Config = protocol.Config{HeartbeatMS: 100, Limits: protocol.Limits{MaxSpeed: f64(0.5), MaxTurn: f64(0.4), MaxCommandMS: 1000},
++ AllowedCommands: append([]string(nil), protocol.Kinds...)}
srv.AddCode("TEST2345")
++ if prep != nil {
++ prep(srv)
++ }
creds, err := link.Pair(context.Background(), nil, srv.URL(), protocol.PairRequest{Code: "TEST2345", AgentVersion: "test",
DeviceKind: "onboard", Drivers: []string{"dryrun"}})
if err != nil {
@@@ -97,7 +108,9 @@ func start(t *testing.T) *env {
e := &env{t: t, srv: srv, whip: whip, node: n, record: record, paths: paths, cancel: cancel, done: done}
t.Cleanup(e.stop)
e.conn = e.nextConn()
-- e.waitReady()
++ if ready {
++ e.waitReady()
++ }
return e
}
@@@ -263,8 +276,10 @@ func TestCommandsClampedAndIdempotent(t *testing.T) {
if v["throttle"] != 0.5 || v["steer"] != -0.4 {
t.Fatalf("not clamped to the owner limits: %s", drives[0].Value)
}
-- if drives[0].DeadlineMS != 300 {
-- t.Fatalf("deadline %d", drives[0].DeadlineMS)
++ // The plugin gets what is left of the 300 ms window (transit on loopback takes a few ms of it; the clock estimate
++ // rounds to whole milliseconds).
++ if d := drives[0].DeadlineMS; d < 150 || d > 305 {
++ t.Fatalf("deadline %d", d)
}
// Deadlines are capped at max_command_ms.
e.cmd("drive", `{"throttle":0.1}`, 60000)
@@@ -362,7 +377,7 @@ func TestLocalKillSwitch(t *testing.T) {
if c := nackCode(e.cmd("drive", `{"throttle":0.3}`, 300)); c != protocol.FaultLocalStop {
t.Fatalf("%q", c)
}
-- // The owner's estop_clear does not clear the local kill switch.
++ // The owner's clear (estop latched:false) does not clear the local kill switch.
e.conn.Send(protocol.Estop{Latched: false, By: "usr_owner"})
time.Sleep(100 * time.Millisecond)
if c := nackCode(e.cmd("drive", `{"throttle":0.3}`, 300)); c != protocol.FaultLocalStop {
ddiff --git a/internal/protocol/protocol.go b/internal/protocol/protocol.go
iindex d2f8118..7c953fb 100644
---- a/internal/protocol/protocol.go
++++ b/internal/protocol/protocol.go
@@@ -34,13 +34,16 @@ const (
// Close codes the server ends the device socket with.
const (
-- CloseReplaced = 4000 // a second connection with the same credential replaced this one
-- CloseInvalid = 4002 // the credential is wrong or revoked (sent right after the upgrade)
-- CloseRevoked = 4003 // the owner revoked the device while it was connected
++ CloseReplaced = 4000 // a second connection with the same credential replaced this one: reconnect normally
++ CloseInvalid = 4002 // the credential is wrong, rotated or revoked (sent right after the upgrade)
++ CloseRevoked = 4003 // the owner revoked the device (or rotated its credential) while it was connected
)
--// Error codes in error.code that the Node acts on.
--const ErrNotPaired = "bot.not_paired" // a frame arrived before the server authenticated the socket
++// Error codes in error.code that the Node tells apart.
++const (
++ ErrNotPaired = "bot.not_paired" // a frame arrived on a socket that has no credential
++ ErrNotReady = "bot.not_ready" // more than 64 frames arrived while the server was still authenticating
++)
// Device → server message types.
const (
@@@ -77,7 +80,8 @@ const (
FaultEstopped = "estopped" // the remote e-stop is latched
FaultLocalStop = "local_stop" // the local kill switch (`openvibe-node stop`) is latched
FaultNoHeartbeat = "no_heartbeat" // a plugin lost the core's heartbeat
-- FaultNotReady = "not_ready" // the driver is starting or restarting
++ FaultNotReady = "not_ready" // the driver is starting or restarting, or the server's config has not arrived
++ FaultExpired = "expired" // the command's deadline_ms had passed when it arrived
FaultPluginDown = "plugin_down" // the driver process is not running
FaultPluginTimeout = "plugin_timeout" // the driver did not answer in time
FaultHardware = "hardware" // the driver reported a hardware error
@@@ -131,9 +135,11 @@ type Limits struct {
}
type Config struct {
-- HeartbeatMS int `json:"heartbeat_ms"`
-- Limits Limits `json:"limits"`
-- AllowedCommands []string `json:"allowed_commands"` // command kinds operators may send; empty = all
++ HeartbeatMS int `json:"heartbeat_ms"`
++ Limits Limits `json:"limits"`
++ // AllowedCommands are the kinds the robot takes (Bot always includes halt). Absent (nil) allows every kind; an
++ // empty list allows only halt.
++ AllowedCommands []string `json:"allowed_commands"`
EstopLatched bool `json:"estop_latched"`
}
@@@ -144,7 +150,9 @@ type Operator struct {
}
type Command struct {
++ // ID is minted by the server: the dedup and ack key. Ref is the operator's own id, for logs only.
ID string `json:"id"`
++ Ref string `json:"ref,omitempty"`
Kind string `json:"kind"`
Value json.RawMessage `json:"value,omitempty"`
// DeadlineMS is an absolute instant in epoch milliseconds on the server's clock, present on motion kinds only.
@@@ -356,19 +364,21 @@ func derefMessage(m Message) Message {
// ---- pairing (HTTPS) ----
// PairBodyFromPaired turns Bot's `paired` frame (pairing over the socket) into the body its POST /api/v1/pair answers
--// with: the same device id, credential, publish key and profile, and robot_ids[0] as robot_id (server/api/v1.js).
++// with: the same device id, credential, publish key, WHIP URL and profile, and robot_ids[0] as robot_id
++// (server/api/v1.js).
func PairBodyFromPaired(frame []byte) []byte {
var p struct {
DeviceID string `json:"device_id"`
Credential string `json:"credential"`
PublishKey string `json:"publish_key"`
++ WHIPURL string `json:"whip_url"`
RobotIDs []string `json:"robot_ids"`
Profile json.RawMessage `json:"profile"`
}
if json.Unmarshal(frame, &p) != nil {
return nil
}
-- r := PairResponse{DeviceID: p.DeviceID, Credential: p.Credential, PublishKey: p.PublishKey, Profile: p.Profile}
++ r := PairResponse{DeviceID: p.DeviceID, Credential: p.Credential, PublishKey: p.PublishKey, WHIPURL: p.WHIPURL, Profile: p.Profile}
if len(p.RobotIDs) > 0 {
r.RobotID = p.RobotIDs[0]
}
@@@ -399,11 +409,13 @@ type ICEServer struct {
Credential string `json:"credential,omitempty"`
}
--// PairResponse is the 201 answer of POST /api/v1/pair. The credential and publish key appear only here.
++// PairResponse is the 201 answer of POST /api/v1/pair. The credential, publish key and WHIP URL (which embeds the
++// publish key; null when Bot has no ingest configured) appear only here.
type PairResponse struct {
DeviceID string `json:"device_id"`
Credential string `json:"credential"`
PublishKey string `json:"publish_key"`
++ WHIPURL string `json:"whip_url,omitempty"`
RobotID string `json:"robot_id"`
Profile json.RawMessage `json:"profile,omitempty"`
}
ddiff --git a/internal/protocol/protocol_test.go b/internal/protocol/protocol_test.go
iindex 4024103..f6ec4e1 100644
---- a/internal/protocol/protocol_test.go
++++ b/internal/protocol/protocol_test.go
@@@ -56,7 +56,7 @@ func TestCommandFields(t *testing.T) {
t.Fatal(err)
}
c := f.Msg.(Command)
-- if c.ID != "x" || c.DeadlineMS != 1738065600300 || c.Operator == nil || c.Operator.Subject != "usr_1" || c.Operator.Role != "operator" ||
++ if c.ID != "x" || c.Ref != "op-1" || c.DeadlineMS != 1738065600300 || c.Operator == nil || c.Operator.Subject != "usr_1" || c.Operator.Role != "operator" ||
c.RobotID != "rob_1" || string(c.Value) != `{"throttle":1}` {
t.Fatalf("%+v", c)
}
@@@ -87,7 +87,7 @@ func TestBotFixtures(t *testing.T) {
t.Errorf("hello %+v", h)
}
if c := decode("config").Msg.(Config); c.HeartbeatMS != 1000 || c.Limits.MaxCommandMS != 300 || c.Limits.HeartbeatMS != 1000 ||
-- c.Limits.MaxSpeed == nil || *c.Limits.MaxSpeed != 1 || len(c.AllowedCommands) != 6 || c.EstopLatched {
++ c.Limits.MaxSpeed == nil || *c.Limits.MaxSpeed != 1 || len(c.AllowedCommands) != 3 || c.AllowedCommands[2] != KindHalt || c.EstopLatched {
t.Errorf("config %+v", c)
}
f := decode("command_drive")
@@@ -95,6 +95,11 @@ func TestBotFixtures(t *testing.T) {
c.Operator.Role != "operator" || c.RobotID != "rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X" || string(c.Value) != `{ "throttle": 0.6, "steer": -0.25 }` {
t.Errorf("command %+v", c)
}
++ f = decode("command_actuator")
++ if c := f.Msg.(Command); c.ID != "cmd_01J8Z4E…" || c.Kind != KindActuator || c.DeadlineMS-f.TS != 300 ||
++ string(c.Value) != `{ "name": "pan", "value": -0.4 }` {
++ t.Errorf("actuator command %+v", c)
++ }
if f := decode("heartbeat_ack"); f.Seq != 41 || f.Msg.(HeartbeatAck).Seq != 41 {
t.Errorf("heartbeat_ack %+v", f)
}
@@@ -126,7 +131,7 @@ func TestBotFixtures(t *testing.T) {
}
// TestBotPairShapes checks the pairing bodies against Bot's pair frame: the request carries exactly its fields, and
--// the REST answer (v1.js: device_id, credential, publish_key, robot_id = robot_ids[0], profile) decodes.
++// the REST answer (v1.js: device_id, credential, publish_key, whip_url, robot_id = robot_ids[0], profile) decodes.
func TestBotPairShapes(t *testing.T) {
var frame map[string]any
if err := json.Unmarshal(fixture(t, "pair"), &frame); err != nil {
@@@ -152,7 +157,7 @@ func TestBotPairShapes(t *testing.T) {
t.Fatal(err)
}
if r.DeviceID != "dev_01J8Z4…" || r.Credential != "Xb3…" || r.PublishKey != "Vt9…" || r.RobotID != "rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X" ||
-- len(r.Profile) == 0 {
++ r.WHIPURL != "https://ingest.openre.stream/whip/Vt9…" || len(r.Profile) == 0 {
t.Fatalf("%+v", r)
}
}
ddiff --git a/internal/protocol/testdata/bot/README.md b/internal/protocol/testdata/bot/README.md
iindex 951e9f3..c532cfd 100644
---- a/internal/protocol/testdata/bot/README.md
++++ b/internal/protocol/testdata/bot/README.md
@@@ -1,20 +1,26 @@
# OpenVibe.Bot device-contract fixtures
--Pinned to **OpenVibe.Bot `d398445479edefa845f0eb32198537434fa11bbd`**.
++Pinned to **OpenVibe.Bot `b48da7c13dbe8e4ab2e3c2d7b3cf058c67e5fd80`**, except `heartbeat` and `heartbeat_ack`,
++which are still the `d398445479edefa845f0eb32198537434fa11bbd` shapes (`seq` both ways): Bot has since moved to
++`heartbeat.t` / `heartbeat_ack.echo`, and the Node's side of that change is a follow-up of its own.
Each `.json` file is one frame copied verbatim (whitespace and `…` included) from the JSON examples in Bot's
`docs/protocol.md` at that commit:
| file | Bot `docs/protocol.md` |
|----------------------------------------------------------------------------|-----------------------------------------------|
--| `pair`, `paired`, `hello`, `config`, `command_drive`, `ack`, `nack`, | section 1 (`/device`), "Examples" |
--| `heartbeat`, `heartbeat_ack`, `telemetry`, `status`, `estop_state`, `estop` | |
++| `pair`, `paired`, `hello`, `config`, `command_actuator`, `command_drive`, | section 1 (`/device`), "Examples" |
++| `ack`, `nack`, `heartbeat`, `heartbeat_ack`, `telemetry`, `status`, | |
++| `estop_state`, `estop` | |
| `error` | section 2 (`/control`): the same `error` frame shape Bot sends on `/device` |
The REST pairing answer has no example in the doc; tests derive it from `paired.json` the way Bot's
--`POST /api/v1/pair` handler (`server/api/v1.js`) builds it: `device_id, credential, publish_key, robot_id`
++`POST /api/v1/pair` handler (`server/api/v1.js`) builds it: `device_id, credential, publish_key, whip_url, robot_id`
(`robot_ids[0]`), `profile` (see `protocol.PairBodyFromPaired`).
--Used by `internal/protocol` (every frame decodes, with the fields the Node reads) and `internal/link`
--(`TestBotFixturesEndToEnd`: pair → hello + config → command → ack → reconnect through `internal/fakebot`). When Bot's
--contract changes, copy the new examples here, update the SHA above, and let the tests say what moved.
++Used by `internal/protocol` (every frame decodes, with the fields the Node reads), `internal/link`
++(`TestBotFixturesEndToEnd`: pair → hello + config → command → ack → reconnect through `internal/fakebot`) and
++`internal/node` (`TestBotFramesDriveAndAck`, `TestRecordedExpiredDriveNeverReachesPlugin`). The command examples carry a
++2025 deadline: replayed as recorded they are expired, and the node tests move `ts`/`deadline_ms` to now (keeping the
++300 ms window) to drive the robot. When Bot's contract changes, copy the new examples here, update the SHA above, and
++let the tests say what moved.
ddiff --git a/internal/protocol/testdata/bot/command_actuator.json b/internal/protocol/testdata/bot/command_actuator.json
nnew file mode 100644
iindex 0000000..f7c1d72
---- /dev/null
++++ b/internal/protocol/testdata/bot/command_actuator.json
@@@ -0,0 +1,3 @@
++{ "v": 1, "seq": 8, "ts": 1738065600000, "type": "command", "id": "cmd_01J8Z4E…", "kind": "actuator",
++ "value": { "name": "pan", "value": -0.4 }, "deadline_ms": 1738065600300,
++ "operator": { "subject": "usr_01J8…", "role": "operator" }, "robot_id": "rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X" }
ddiff --git a/internal/protocol/testdata/bot/config.json b/internal/protocol/testdata/bot/config.json
iindex d70d059..30d5f9c 100644
---- a/internal/protocol/testdata/bot/config.json
++++ b/internal/protocol/testdata/bot/config.json
@@@ -1,3 +1,3 @@
{ "v": 1, "seq": 3, "ts": 1738065600000, "type": "config",
"heartbeat_ms": 1000, "limits": { "max_speed": 1, "max_turn": 1, "max_command_ms": 300, "heartbeat_ms": 1000 },
-- "allowed_commands": ["drive", "actuator", "ptz", "say", "display", "halt"], "estop_latched": false }
++ "allowed_commands": ["drive", "actuator", "halt"], "estop_latched": false }
ddiff --git a/internal/protocol/testdata/bot/paired.json b/internal/protocol/testdata/bot/paired.json
iindex 165a7e2..8e0154a 100644
---- a/internal/protocol/testdata/bot/paired.json
++++ b/internal/protocol/testdata/bot/paired.json
@@@ -1,3 +1,4 @@
{ "v": 1, "seq": 1, "ts": 1738065600000, "type": "paired", "device_id": "dev_01J8Z4…",
-- "credential": "Xb3…", "publish_key": "Vt9…", "robot_ids": ["rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X"],
++ "credential": "Xb3…", "publish_key": "Vt9…", "whip_url": "https://ingest.openre.stream/whip/Vt9…",
++ "robot_ids": ["rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X"],
"profile_id": "adeept.adr036", "profile": { "id": "adeept.adr036", "limits": { "max_command_ms": 300, "heartbeat_ms": 1000 } } }
ddiff --git a/internal/safety/latch.go b/internal/safety/latch.go
iindex 62e1f5a..9b766c9 100644
---- a/internal/safety/latch.go
++++ b/internal/safety/latch.go
@@@ -15,7 +15,7 @@ import (
// LatchState is the persisted e-stop state. Two independent latches:
//
--// - Remote: set by a server `estop`; cleared by a server `estop_clear` (the owner) or a local `resume`.
++// - Remote: set by a server `estop`; cleared by a server `estop` with latched:false (the owner) or a local `resume`.
// - Local: set by `openvibe-node stop` on the machine itself; cleared only by `openvibe-node resume`.
//
// Anything that moves is refused while either is set.
@@@ -138,7 +138,7 @@ func (l *Latch) SetRemote(reason string) (LatchState, error) {
})
}
--// ClearRemote clears the server e-stop (owner's estop_clear). The local latch is untouched.
++// ClearRemote clears the server e-stop (the owner's estop latched:false). The local latch is untouched.
func (l *Latch) ClearRemote() (LatchState, error) {
return l.update(func(s *LatchState) { s.Remote, s.RemoteReason, s.RemoteAt = false, "", time.Time{} })
}
ddiff --git a/internal/video/whip.go b/internal/video/whip.go
iindex 72dc9cd..e110d53 100644
---- a/internal/video/whip.go
++++ b/internal/video/whip.go
@@@ -268,7 +268,7 @@ func (p *Publisher) session(ctx context.Context) error {
p.mu.Unlock()
p.requestKeyframe()
p.state.Store(StateLive)
-- p.log.Info("video live", "whip", redactURL(p.opt.WHIPURL))
++ p.log.Info("video live", "whip", scrub(redactURL(p.opt.WHIPURL), p.opt.PublishKey))
case webrtc.PeerConnectionStateDisconnected:
disconnectedAt = time.Now()
case webrtc.PeerConnectionStateFailed, webrtc.PeerConnectionStateClosed: