openvibers/openvibe.link Commits
Send heartbeat send time and measure RTT from Bot's ack echo (#6)
Changes
@@@ -79,9 +79,10 @@ Every frame is one JSON object in one text message with the envelope fields at t| `ts` | int | sender's clock, Unix milliseconds || `type` | string | message type |--Unknown `type`s are ignored (a newer server may send more). Frames are at most 1 MiB. `heartbeat` and `heartbeat_ack`--carry their own `seq` (the echoed heartbeat number) in the same object as the envelope's; the Node sets a heartbeat's--`seq` equal to its envelope `seq`, so either reading gives the same number.++Unknown `type`s are ignored (a newer server may send more). Frames are at most 1 MiB. A `heartbeat` carries the++device's send time in `t` (Unix ms) and the last measured `rtt_ms`; the server echoes `t` back in `heartbeat_ack` as++`echo` (and `t`), so the Node measures the round trip against its own clock. A `heartbeat_ack`'s envelope `seq` is the++server's own counter, never the heartbeat's.### Server → device@@@ -91,7 +92,7 @@ carry their own `seq` (the echoed heartbeat number) in the same object as the en| `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` (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 |++| `heartbeat_ack` | `echo` (the heartbeat's `t`, or null), `t` (the same send time when the server has it), `server_time` (RFC 3339); the Node measures RTT from the echoed 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@@@ -108,7 +109,7 @@ applied the Node runs nothing but `halt` (`nack not_ready`), so that latch is in| `telemetry` | `battery` (0..1), `voltage` (V), `rssi`, `sensors` {…}; Node extra: `events` [{`name`, `driver`, `ts`, `fields`}] || `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) |++| `heartbeat` | `t` (send time, Unix ms), `rtt_ms` (the last measured round trip, once there is one; 0 is a real measurement) || `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` |The first `status` and `estop_state` go out once the connection's `config` has been applied, never right after the@@@ -137,8 +137,8 @@ type VideoConfig struct {FPS int `json:"fps,omitempty"`Width int `json:"width,omitempty"`Height int `json:"height,omitempty"`-- // WHIPURL is where the camera publishes (with the pairing's publish key). OpenVibe.Bot does not hand one out,-- // so video stays off until it is set here.++ // WHIPURL is where the camera publishes (with the pairing's publish key). The pair response carries it (Bot++ // mints one per device); when it is empty, video stays off until it is set here.WHIPURL string `json:"whip_url,omitempty"`}@@@ -8,8 +8,8 @@// 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.++// heartbeat_ack carries Bot's own envelope seq and echoes the heartbeat's t as both t and echo (echo null when the++// heartbeat had no t), with server_time as Bot's clock; the device measures RTT from the echoed send time.package fakebotimport (@@@ -438,7 +438,13 @@ func (c *Conn) read() {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)})++ // Bot answers with its own envelope seq (sendAt sets it): echo and t are the heartbeat's send time.++ ack := protocol.HeartbeatAck{ServerTime: time.Now().UTC().Format(time.RFC3339Nano)}++ if hb.T != 0 {++ t := hb.T++ ack.T, ack.Echo = t, &t++ }++ _ = c.Send(ack)}c.fmu.Lock()c.frames = append(c.frames, f)@@@ -64,6 +64,13 @@ type Options struct {// the next try; jitter only ever adds to it.const CredentialRetryMin = 10 * time.Second++// hbSend is one heartbeat the link sent: its monotonic send time and the t it carried, so an ack can be matched++// back either by the echoed t (Bot) or by the body seq (older servers).++type hbSend struct {++ at time.Time++ t int64++}++// Link is the reconnecting control connection.type Link struct {opt Options@@@ -73,7 +80,7 @@ type Link struct {mu sync.Mutexout chan []byteseq uint64-- hbSent map[uint64]time.Time // heartbeat seq → send time, for the RTT++ hbSent map[uint64]hbSend // heartbeat seq → send time/t, bounded to the last fewheartbeat time.Durationdeadman time.Duration@@@ -196,8 +203,10 @@ func (l *Link) Send(m protocol.Message) error {l.seq++now := time.Now()if hb, ok := m.(protocol.Heartbeat); ok {-- // The body seq equals the envelope seq, so heartbeat_ack's echo is the same whichever the server reads.++ // The body seq still mirrors the envelope's (older servers echo it); t is the send time Bot echoes back as++ // heartbeat_ack.t/echo, and rtt_ms is the last measured round trip, once there is one on this connection.hb.Seq = l.seq++ hb.T = now.UnixMilli()if rtt := l.lastRTT.Load(); rtt > 0 {ms := time.Duration(rtt).Milliseconds()hb.RTTMS = &ms@@@ -208,7 +217,7 @@ func (l *Link) Send(m protocol.Message) error {delete(l.hbSent, s)}}-- l.hbSent[l.seq] = now++ l.hbSent[l.seq] = hbSend{at: now, t: hb.T}}b, err := protocol.Encode(l.seq, now.UnixMilli(), m)if err != nil {@@@ -326,8 +335,10 @@ 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, l.hasSkew = nil, 0, map[uint64]time.Time{}, false++ l.out, l.seq, l.hbSent, l.hasSkew = nil, 0, map[uint64]hbSend{}, falsel.mu.Unlock()++ // A fresh connection has no measured round trip: the first heartbeat must not claim one.++ l.lastRTT.Store(0)sessCtx, cancel := context.WithCancel(ctx)var endErr error@@@ -427,11 +438,32 @@ func (l *Link) session(ctx context.Context) error {l.observe(f.TS, rx)switch m := f.Msg.(type) {case protocol.HeartbeatAck:++ // Bot's envelope seq is its own counter, so the ack is matched to a stored send time by the echoed t++ // (echo, or older Bot's t); a server that only echoes the heartbeat's seq is matched by that. The++ // stored time is monotonic, so a wall-clock adjustment between send and ack cannot skew the round trip.++ echo := m.T++ if echo == 0 && m.Echo != nil {++ echo = *m.Echo++ }l.mu.Lock()-- sent, ok := l.hbSent[m.Seq]-- delete(l.hbSent, m.Seq)++ var sent time.Time++ if echo > 0 {++ for s, hb := range l.hbSent {++ if hb.t == echo {++ sent = hb.at++ delete(l.hbSent, s)++ break++ }++ }++ }++ if sent.IsZero() && m.Seq != 0 {++ if hb, ok := l.hbSent[m.Seq]; ok {++ sent = hb.at++ delete(l.hbSent, m.Seq)++ }++ }l.mu.Unlock()-- if ok {++ if !sent.IsZero() {// A coarse clock (Windows) can read 0 on a fast loopback; 0 means "not measured yet", so floor at 1 ns.l.lastRTT.Store(int64(max(time.Since(sent), 1)))}@@@ -143,6 +143,119 @@ func TestConnectHelloConfigHeartbeat(t *testing.T) {}}++// Bot's heartbeat_ack echoes the device's send time (echo/t) over its own envelope seq, so the Node measures the++// round trip from that; a fresh connection starts with no measurement.++func TestHeartbeatRTTReportedAfterAck(t *testing.T) {++ srv, l, rec, _, _ := setup(t, 60*time.Millisecond)++ c, err := srv.NextConn(3 * time.Second)++ if err != nil {++ t.Fatal(err)++ }++ waitFrame(t, rec, protocol.TypeHello)++ waitFrame(t, rec, protocol.TypeConfig)++++ first, err := c.Expect(protocol.TypeHeartbeat, 2*time.Second, nil)++ if err != nil {++ t.Fatal(err)++ }++ if hb := first.Msg.(protocol.Heartbeat); hb.T == 0 {++ t.Fatalf("heartbeat carried no t: %+v", hb)++ } else if hb.RTTMS != nil {++ t.Fatalf("first heartbeat on a connection carried rtt_ms %d", *hb.RTTMS)++ }++++ // fakebot auto-acks, so a later heartbeat carries the measurement: present, and never negative.++ for {++ f, err := c.Expect(protocol.TypeHeartbeat, 3*time.Second, nil)++ if err != nil {++ t.Fatal(err)++ }++ hb := f.Msg.(protocol.Heartbeat)++ if hb.RTTMS == nil {++ continue++ }++ if *hb.RTTMS < 0 {++ t.Fatalf("rtt_ms %d", *hb.RTTMS)++ }++ break++ }++ if l.Stats().RTT <= 0 {++ t.Fatal("no RTT in stats")++ }++++ // A reconnect starts fresh: the first heartbeat on the new connection again carries no rtt_ms.++ c.Close()++ <-rec.gotDown++ c2, err := srv.NextConn(3 * time.Second)++ if err != nil {++ t.Fatal(err)++ }++ waitFrame(t, rec, protocol.TypeHello)++ waitFrame(t, rec, protocol.TypeConfig)++ first2, err := c2.Expect(protocol.TypeHeartbeat, 2*time.Second, nil)++ if err != nil {++ t.Fatal(err)++ }++ if hb := first2.Msg.(protocol.Heartbeat); hb.RTTMS != nil {++ t.Fatalf("first heartbeat on a new connection carried rtt_ms %d", *hb.RTTMS)++ }++}++++// An ack that carries only echo (Bot's newer field, no t) still sets the RTT.++func TestHeartbeatAckEchoOnlySetsRTT(t *testing.T) {++ srv, l, rec, _, _ := setup(t, 200*time.Millisecond)++ c, err := srv.NextConn(3 * time.Second)++ if err != nil {++ t.Fatal(err)++ }++ waitFrame(t, rec, protocol.TypeConfig)++ c.Mute() // stop fakebot's automatic t+echo answers; this test sends the ack itself++ f, err := c.Expect(protocol.TypeHeartbeat, 2*time.Second, nil)++ if err != nil {++ t.Fatal(err)++ }++ echo := f.Msg.(protocol.Heartbeat).T++ if echo == 0 {++ t.Fatal("heartbeat carried no t to echo")++ }++ if err := c.Send(protocol.HeartbeatAck{Echo: &echo}); err != nil {++ t.Fatal(err)++ }++ deadline := time.Now().Add(2 * time.Second)++ for l.Stats().RTT <= 0 {++ if time.Now().After(deadline) {++ t.Fatal("an echo-only heartbeat_ack did not set the RTT")++ }++ time.Sleep(5 * time.Millisecond)++ }++}++++// An ack that carries only seq (older Bot, no t/echo) still sets the RTT from the bounded send-time lookup.++func TestHeartbeatAckSeqOnlySetsRTT(t *testing.T) {++ srv, l, rec, _, _ := setup(t, 200*time.Millisecond)++ c, err := srv.NextConn(3 * time.Second)++ if err != nil {++ t.Fatal(err)++ }++ waitFrame(t, rec, protocol.TypeConfig)++ c.Mute() // stop fakebot's automatic t+echo answers; this test sends the ack itself++ f, err := c.Expect(protocol.TypeHeartbeat, 2*time.Second, nil)++ if err != nil {++ t.Fatal(err)++ }++ hb := f.Msg.(protocol.Heartbeat)++ if err := c.Send(protocol.HeartbeatAck{Seq: hb.Seq}); err != nil {++ t.Fatal(err)++ }++ deadline := time.Now().Add(2 * time.Second)++ for l.Stats().RTT <= 0 {++ if time.Now().After(deadline) {++ t.Fatal("a seq-only heartbeat_ack did not set the RTT")++ }++ time.Sleep(5 * time.Millisecond)++ }++}++func TestDeadmanOnSilentServer(t *testing.T) {srv, _, rec, _, _ := setup(t, 50*time.Millisecond)c, _ := srv.NextConn(3 * time.Second)@@@ -169,10 +169,13 @@ type Estop struct {At string `json:"at,omitempty"` // RFC 3339}--// HeartbeatAck echoes the heartbeat's seq. The server writes it over the envelope's seq, so the decoded frame's--// Seq is the echoed value too.++// HeartbeatAck answers a heartbeat. Bot's envelope seq is its own counter and never the device's, so the body++// carries the device's send time back as t (and, on newer Bot, echo: the same value, or null when the heartbeat++// had no t) and Bot's clock in server_time. Seq is present only from older servers that echoed the heartbeat's seq.type HeartbeatAck struct {-- Seq uint64 `json:"seq"`++ Seq uint64 `json:"seq,omitempty"`++ T int64 `json:"t,omitempty"`++ Echo *int64 `json:"echo,omitempty"`ServerTime string `json:"server_time,omitempty"`}@@@ -238,9 +241,12 @@ type Nack struct {Message string `json:"message,omitempty"`}--// Heartbeat carries the same seq as its envelope (the link sets it), so the server's echo is unambiguous.++// Heartbeat is sent every config.heartbeat_ms. Seq mirrors the envelope seq for older servers that echo it; T is the++// device's send time (unix ms), which Bot echoes back as heartbeat_ack.t/echo and the Node measures RTT from. RTTMS is++// the last measured round trip, absent until one has been measured (a measured 0 ms is still sent, hence the pointer).type Heartbeat struct {Seq uint64 `json:"seq"`++ T int64 `json:"t,omitempty"`RTTMS *int64 `json:"rtt_ms,omitempty"`}@@@ -32,7 +32,7 @@ func TestEncodeDecodeRoundTrip(t *testing.T) {if err := json.Unmarshal(b, &raw); err != nil {t.Fatalf("%s: %v: %s", m.MessageType(), err, b)}-- // heartbeat and heartbeat_ack carry their own seq, which wins over the envelope's (as on Bot's wire).++ // heartbeat carries its own seq (mirroring the envelope's); heartbeat_ack does too only from older servers._, ownSeq := m.(Heartbeat)if _, ok := m.(HeartbeatAck); ok {ownSeq = true@@@ -103,6 +103,11 @@ func TestBotFixtures(t *testing.T) {if f := decode("heartbeat_ack"); f.Seq != 41 || f.Msg.(HeartbeatAck).Seq != 41 {t.Errorf("heartbeat_ack %+v", f)}++ // Bot's newer ack: its own envelope seq, the device's send time echoed as echo and t.++ if f := decode("heartbeat_ack_echo"); f.Seq != 4 || f.Msg.(HeartbeatAck).T != 1738065600500 ||++ f.Msg.(HeartbeatAck).Echo == nil || *f.Msg.(HeartbeatAck).Echo != 1738065600500 || f.Msg.(HeartbeatAck).ServerTime == "" {++ t.Errorf("heartbeat_ack echo %+v", f)++ }if e := decode("estop").Msg.(Estop); e.Latched || e.By != "usr_01J8…" {t.Errorf("estop %+v", e)}@@@ -1,8 +1,8 @@# OpenVibe.Bot device-contract fixtures--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.++Pinned to **OpenVibe.Bot `b48da7c13dbe8e4ab2e3c2d7b3cf058c67e5fd80`**, except the heartbeat exchange: the++`heartbeat_ack_echo` fixture is Bot `ace1671`, where `heartbeat_ack` echoes the heartbeat's `t` as both `echo` and++`t` over Bot's own envelope `seq`. The older `heartbeat_ack` fixture (the `seq` shape) stays as a decode case.Each `.json` file is one frame copied verbatim (whitespace and `…` included) from the JSON examples in Bot's`docs/protocol.md` at that commit:@@@ -13,6 +13,7 @@ Each `.json` file is one frame copied verbatim (whitespace and `…` included) f| `ack`, `nack`, `heartbeat`, `heartbeat_ack`, `telemetry`, `status`, | || `estop_state`, `estop` | || `error` | section 2 (`/control`): the same `error` frame shape Bot sends on `/device` |++| `heartbeat_ack_echo` | section 1 (`/device`), "Examples" (Bot `ace1671`) |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, whip_url, robot_id`@@@ -20,7 +21,7 @@ The REST pairing answer has no example in the doc; tests derive it from `paired.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++`internal/node` (`TestBotFramesDriveAndAck`, `TestExpiredDriveNeverReachesPlugin`). The command examples carry a2025 deadline: replayed as recorded they are expired, and the node tests move `ts`/`deadline_ms` to now (keeping the300 ms window) to drive the robot. When Bot's contract changes, copy the new examples here, update the SHA above, andlet the tests say what moved.nnew file mode 100644@@@ -0,0 +1 @@++{ "v": 1, "seq": 4, "ts": 1738065600501, "type": "heartbeat_ack", "echo": 1738065600500, "t": 1738065600500, "server_time": "2026-09-29T19:20:00.501Z" }
@@ -79,9 +79,10 @@ Every frame is one JSON object in one text message with the envelope fields at t
@@ -79,9 +79,10 @@ Every frame is one JSON object in one text message with the envelope fields at t
| `ts` | int | sender's clock, Unix milliseconds |
| `ts` | int | sender's clock, Unix milliseconds |
| `type` | string | message type |
| `type` | string | message type |
-Unknown `type`s are ignored (a newer server may send more). Frames are at most 1 MiB. `heartbeat` and `heartbeat_ack`
-carry their own `seq` (the echoed heartbeat number) in the same object as the envelope's; the Node sets a heartbeat's
-`seq` equal to its envelope `seq`, so either reading gives the same number.
+Unknown `type`s are ignored (a newer server may send more). Frames are at most 1 MiB. A `heartbeat` carries the
+device's send time in `t` (Unix ms) and the last measured `rtt_ms`; the server echoes `t` back in `heartbeat_ack` as
+`echo` (and `t`), so the Node measures the round trip against its own clock. A `heartbeat_ack`'s envelope `seq` is the
+server's own counter, never the heartbeat's.
### Server → device
### Server → device
@@ -91,7 +92,7 @@ carry their own `seq` (the echoed heartbeat number) in the same object as the en
@@ -91,7 +92,7 @@ carry their own `seq` (the echoed heartbeat number) in the same object as the en
| `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` |
| `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` (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) |
| `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` |
| `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 |
+| `heartbeat_ack` | `echo` (the heartbeat's `t`, or null), `t` (the same send time when the server has it), `server_time` (RFC 3339); the Node measures RTT from the echoed send time |
| `error` | `code`, `detail` — a frame the server refused; logged, never fatal |
| `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
`config.estop_latched` is the robot's e-stop as the server holds it: an owner e-stop set while the device was offline
@@ -108,7 +109,7 @@ applied the Node runs nothing but `halt` (`nack not_ready`), so that latch is in
@@ -108,7 +109,7 @@ applied the Node runs nothing but `halt` (`nack not_ready`), so that latch is in
| `telemetry` | `battery` (0..1), `voltage` (V), `rssi`, `sensors` {…}; Node extra: `events` [{`name`, `driver`, `ts`, `fields`}] |
| `telemetry` | `battery` (0..1), `voltage` (V), `rssi`, `sensors` {…}; Node extra: `events` [{`name`, `driver`, `ts`, `fields`}] |
| `ack` | `id`; Node extra: `latency_ms` (receipt → plugin ack) |
| `ack` | `id`; Node extra: `latency_ms` (receipt → plugin ack) |
| `nack` | `id`, `fault_code`; Node extra: `message` |
| `nack` | `id`, `fault_code`; Node extra: `message` |
-| `heartbeat` | `seq`, `rtt_ms` (the last measured round trip, once there is one) |
+| `heartbeat` | `t` (send time, Unix ms), `rtt_ms` (the last measured round trip, once there is one; 0 is a real measurement) |
| `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` |
| `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` |
The first `status` and `estop_state` go out once the connection's `config` has been applied, never right after the
The first `status` and `estop_state` go out once the connection's `config` has been applied, never right after the
@@ -137,8 +137,8 @@ type VideoConfig struct {
@@ -137,8 +137,8 @@ type VideoConfig struct {
FPS int `json:"fps,omitempty"`
FPS int `json:"fps,omitempty"`
Width int `json:"width,omitempty"`
Width int `json:"width,omitempty"`
Height int `json:"height,omitempty"`
Height int `json:"height,omitempty"`
- // WHIPURL is where the camera publishes (with the pairing's publish key). OpenVibe.Bot does not hand one out,
- // so video stays off until it is set here.
+ // WHIPURL is where the camera publishes (with the pairing's publish key). The pair response carries it (Bot
+ // mints one per device); when it is empty, video stays off until it is set here.
WHIPURL string `json:"whip_url,omitempty"`
WHIPURL string `json:"whip_url,omitempty"`
}
}
@@ -8,8 +8,8 @@
@@ -8,8 +8,8 @@
// it, a device's estop_state latched:true sets it, and latched:false is only a report. Tests drive it to send
// 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.
// 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.
+// heartbeat_ack carries Bot's own envelope seq and echoes the heartbeat's t as both t and echo (echo null when the
+// heartbeat had no t), with server_time as Bot's clock; the device measures RTT from the echoed send time.
package fakebot
package fakebot
import (
import (
@@ -438,7 +438,13 @@ func (c *Conn) read() {
@@ -438,7 +438,13 @@ func (c *Conn) read() {
c.setEstop(true) // a report of latched:false never clears Bot's latch
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() {
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)})
+ // Bot answers with its own envelope seq (sendAt sets it): echo and t are the heartbeat's send time.
+ ack := protocol.HeartbeatAck{ServerTime: time.Now().UTC().Format(time.RFC3339Nano)}
+ if hb.T != 0 {
+ t := hb.T
+ ack.T, ack.Echo = t, &t
+ }
+ _ = c.Send(ack)
}
}
c.fmu.Lock()
c.fmu.Lock()
c.frames = append(c.frames, f)
c.frames = append(c.frames, f)
@@ -64,6 +64,13 @@ type Options struct {
@@ -64,6 +64,13 @@ type Options struct {
// the next try; jitter only ever adds to it.
// the next try; jitter only ever adds to it.
const CredentialRetryMin = 10 * time.Second
const CredentialRetryMin = 10 * time.Second
+// hbSend is one heartbeat the link sent: its monotonic send time and the t it carried, so an ack can be matched
+// back either by the echoed t (Bot) or by the body seq (older servers).
+type hbSend struct {
+ at time.Time
+ t int64
+}
+
// Link is the reconnecting control connection.
// Link is the reconnecting control connection.
type Link struct {
type Link struct {
opt Options
opt Options
@@ -73,7 +80,7 @@ type Link struct {
@@ -73,7 +80,7 @@ type Link struct {
mu sync.Mutex
mu sync.Mutex
out chan []byte
out chan []byte
seq uint64
seq uint64
- hbSent map[uint64]time.Time // heartbeat seq → send time, for the RTT
+ hbSent map[uint64]hbSend // heartbeat seq → send time/t, bounded to the last few
heartbeat time.Duration
heartbeat time.Duration
deadman time.Duration
deadman time.Duration
@@ -196,8 +203,10 @@ func (l *Link) Send(m protocol.Message) error {
@@ -196,8 +203,10 @@ func (l *Link) Send(m protocol.Message) error {
l.seq++
l.seq++
now := time.Now()
now := time.Now()
if hb, ok := m.(protocol.Heartbeat); ok {
if hb, ok := m.(protocol.Heartbeat); ok {
- // The body seq equals the envelope seq, so heartbeat_ack's echo is the same whichever the server reads.
+ // The body seq still mirrors the envelope's (older servers echo it); t is the send time Bot echoes back as
+ // heartbeat_ack.t/echo, and rtt_ms is the last measured round trip, once there is one on this connection.
hb.Seq = l.seq
hb.Seq = l.seq
+ hb.T = now.UnixMilli()
if rtt := l.lastRTT.Load(); rtt > 0 {
if rtt := l.lastRTT.Load(); rtt > 0 {
ms := time.Duration(rtt).Milliseconds()
ms := time.Duration(rtt).Milliseconds()
hb.RTTMS = &ms
hb.RTTMS = &ms
@@ -208,7 +217,7 @@ func (l *Link) Send(m protocol.Message) error {
@@ -208,7 +217,7 @@ func (l *Link) Send(m protocol.Message) error {
delete(l.hbSent, s)
delete(l.hbSent, s)
}
}
}
}
- l.hbSent[l.seq] = now
+ l.hbSent[l.seq] = hbSend{at: now, t: hb.T}
}
}
b, err := protocol.Encode(l.seq, now.UnixMilli(), m)
b, err := protocol.Encode(l.seq, now.UnixMilli(), m)
if err != nil {
if err != nil {
@@ -326,8 +335,10 @@ func (l *Link) session(ctx context.Context) error {
@@ -326,8 +335,10 @@ func (l *Link) session(ctx context.Context) error {
// Nothing is sent until the server's hello: until then the socket is not authenticated.
// Nothing is sent until the server's hello: until then the socket is not authenticated.
out := make(chan []byte, 256)
out := make(chan []byte, 256)
l.mu.Lock()
l.mu.Lock()
- l.out, l.seq, l.hbSent, l.hasSkew = nil, 0, map[uint64]time.Time{}, false
+ l.out, l.seq, l.hbSent, l.hasSkew = nil, 0, map[uint64]hbSend{}, false
l.mu.Unlock()
l.mu.Unlock()
+ // A fresh connection has no measured round trip: the first heartbeat must not claim one.
+ l.lastRTT.Store(0)
sessCtx, cancel := context.WithCancel(ctx)
sessCtx, cancel := context.WithCancel(ctx)
var endErr error
var endErr error
@@ -427,11 +438,32 @@ func (l *Link) session(ctx context.Context) error {
@@ -427,11 +438,32 @@ func (l *Link) session(ctx context.Context) error {
l.observe(f.TS, rx)
l.observe(f.TS, rx)
switch m := f.Msg.(type) {
switch m := f.Msg.(type) {
case protocol.HeartbeatAck:
case protocol.HeartbeatAck:
+ // Bot's envelope seq is its own counter, so the ack is matched to a stored send time by the echoed t
+ // (echo, or older Bot's t); a server that only echoes the heartbeat's seq is matched by that. The
+ // stored time is monotonic, so a wall-clock adjustment between send and ack cannot skew the round trip.
+ echo := m.T
+ if echo == 0 && m.Echo != nil {
+ echo = *m.Echo
+ }
l.mu.Lock()
l.mu.Lock()
- sent, ok := l.hbSent[m.Seq]
- delete(l.hbSent, m.Seq)
+ var sent time.Time
+ if echo > 0 {
+ for s, hb := range l.hbSent {
+ if hb.t == echo {
+ sent = hb.at
+ delete(l.hbSent, s)
+ break
+ }
+ }
+ }
+ if sent.IsZero() && m.Seq != 0 {
+ if hb, ok := l.hbSent[m.Seq]; ok {
+ sent = hb.at
+ delete(l.hbSent, m.Seq)
+ }
+ }
l.mu.Unlock()
l.mu.Unlock()
- if ok {
+ if !sent.IsZero() {
// A coarse clock (Windows) can read 0 on a fast loopback; 0 means "not measured yet", so floor at 1 ns.
// A coarse clock (Windows) can read 0 on a fast loopback; 0 means "not measured yet", so floor at 1 ns.
l.lastRTT.Store(int64(max(time.Since(sent), 1)))
l.lastRTT.Store(int64(max(time.Since(sent), 1)))
}
}
@@ -143,6 +143,119 @@ func TestConnectHelloConfigHeartbeat(t *testing.T) {
@@ -143,6 +143,119 @@ func TestConnectHelloConfigHeartbeat(t *testing.T) {
}
}
}
}
+// Bot's heartbeat_ack echoes the device's send time (echo/t) over its own envelope seq, so the Node measures the
+// round trip from that; a fresh connection starts with no measurement.
+func TestHeartbeatRTTReportedAfterAck(t *testing.T) {
+ srv, l, rec, _, _ := setup(t, 60*time.Millisecond)
+ c, err := srv.NextConn(3 * time.Second)
+ if err != nil {
+ t.Fatal(err)
+ }
+ waitFrame(t, rec, protocol.TypeHello)
+ waitFrame(t, rec, protocol.TypeConfig)
+
+ first, err := c.Expect(protocol.TypeHeartbeat, 2*time.Second, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if hb := first.Msg.(protocol.Heartbeat); hb.T == 0 {
+ t.Fatalf("heartbeat carried no t: %+v", hb)
+ } else if hb.RTTMS != nil {
+ t.Fatalf("first heartbeat on a connection carried rtt_ms %d", *hb.RTTMS)
+ }
+
+ // fakebot auto-acks, so a later heartbeat carries the measurement: present, and never negative.
+ for {
+ f, err := c.Expect(protocol.TypeHeartbeat, 3*time.Second, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ hb := f.Msg.(protocol.Heartbeat)
+ if hb.RTTMS == nil {
+ continue
+ }
+ if *hb.RTTMS < 0 {
+ t.Fatalf("rtt_ms %d", *hb.RTTMS)
+ }
+ break
+ }
+ if l.Stats().RTT <= 0 {
+ t.Fatal("no RTT in stats")
+ }
+
+ // A reconnect starts fresh: the first heartbeat on the new connection again carries no rtt_ms.
+ c.Close()
+ <-rec.gotDown
+ c2, err := srv.NextConn(3 * time.Second)
+ if err != nil {
+ t.Fatal(err)
+ }
+ waitFrame(t, rec, protocol.TypeHello)
+ waitFrame(t, rec, protocol.TypeConfig)
+ first2, err := c2.Expect(protocol.TypeHeartbeat, 2*time.Second, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if hb := first2.Msg.(protocol.Heartbeat); hb.RTTMS != nil {
+ t.Fatalf("first heartbeat on a new connection carried rtt_ms %d", *hb.RTTMS)
+ }
+}
+
+// An ack that carries only echo (Bot's newer field, no t) still sets the RTT.
+func TestHeartbeatAckEchoOnlySetsRTT(t *testing.T) {
+ srv, l, rec, _, _ := setup(t, 200*time.Millisecond)
+ c, err := srv.NextConn(3 * time.Second)
+ if err != nil {
+ t.Fatal(err)
+ }
+ waitFrame(t, rec, protocol.TypeConfig)
+ c.Mute() // stop fakebot's automatic t+echo answers; this test sends the ack itself
+ f, err := c.Expect(protocol.TypeHeartbeat, 2*time.Second, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ echo := f.Msg.(protocol.Heartbeat).T
+ if echo == 0 {
+ t.Fatal("heartbeat carried no t to echo")
+ }
+ if err := c.Send(protocol.HeartbeatAck{Echo: &echo}); err != nil {
+ t.Fatal(err)
+ }
+ deadline := time.Now().Add(2 * time.Second)
+ for l.Stats().RTT <= 0 {
+ if time.Now().After(deadline) {
+ t.Fatal("an echo-only heartbeat_ack did not set the RTT")
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+}
+
+// An ack that carries only seq (older Bot, no t/echo) still sets the RTT from the bounded send-time lookup.
+func TestHeartbeatAckSeqOnlySetsRTT(t *testing.T) {
+ srv, l, rec, _, _ := setup(t, 200*time.Millisecond)
+ c, err := srv.NextConn(3 * time.Second)
+ if err != nil {
+ t.Fatal(err)
+ }
+ waitFrame(t, rec, protocol.TypeConfig)
+ c.Mute() // stop fakebot's automatic t+echo answers; this test sends the ack itself
+ f, err := c.Expect(protocol.TypeHeartbeat, 2*time.Second, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ hb := f.Msg.(protocol.Heartbeat)
+ if err := c.Send(protocol.HeartbeatAck{Seq: hb.Seq}); err != nil {
+ t.Fatal(err)
+ }
+ deadline := time.Now().Add(2 * time.Second)
+ for l.Stats().RTT <= 0 {
+ if time.Now().After(deadline) {
+ t.Fatal("a seq-only heartbeat_ack did not set the RTT")
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+}
+
func TestDeadmanOnSilentServer(t *testing.T) {
func TestDeadmanOnSilentServer(t *testing.T) {
srv, _, rec, _, _ := setup(t, 50*time.Millisecond)
srv, _, rec, _, _ := setup(t, 50*time.Millisecond)
c, _ := srv.NextConn(3 * time.Second)
c, _ := srv.NextConn(3 * time.Second)
@@ -169,10 +169,13 @@ type Estop struct {
@@ -169,10 +169,13 @@ type Estop struct {
At string `json:"at,omitempty"` // RFC 3339
At string `json:"at,omitempty"` // RFC 3339
}
}
-// HeartbeatAck echoes the heartbeat's seq. The server writes it over the envelope's seq, so the decoded frame's
-// Seq is the echoed value too.
+// HeartbeatAck answers a heartbeat. Bot's envelope seq is its own counter and never the device's, so the body
+// carries the device's send time back as t (and, on newer Bot, echo: the same value, or null when the heartbeat
+// had no t) and Bot's clock in server_time. Seq is present only from older servers that echoed the heartbeat's seq.
type HeartbeatAck struct {
type HeartbeatAck struct {
- Seq uint64 `json:"seq"`
+ Seq uint64 `json:"seq,omitempty"`
+ T int64 `json:"t,omitempty"`
+ Echo *int64 `json:"echo,omitempty"`
ServerTime string `json:"server_time,omitempty"`
ServerTime string `json:"server_time,omitempty"`
}
}
@@ -238,9 +241,12 @@ type Nack struct {
@@ -238,9 +241,12 @@ type Nack struct {
Message string `json:"message,omitempty"`
Message string `json:"message,omitempty"`
}
}
-// Heartbeat carries the same seq as its envelope (the link sets it), so the server's echo is unambiguous.
+// Heartbeat is sent every config.heartbeat_ms. Seq mirrors the envelope seq for older servers that echo it; T is the
+// device's send time (unix ms), which Bot echoes back as heartbeat_ack.t/echo and the Node measures RTT from. RTTMS is
+// the last measured round trip, absent until one has been measured (a measured 0 ms is still sent, hence the pointer).
type Heartbeat struct {
type Heartbeat struct {
Seq uint64 `json:"seq"`
Seq uint64 `json:"seq"`
+ T int64 `json:"t,omitempty"`
RTTMS *int64 `json:"rtt_ms,omitempty"`
RTTMS *int64 `json:"rtt_ms,omitempty"`
}
}
@@ -32,7 +32,7 @@ func TestEncodeDecodeRoundTrip(t *testing.T) {
@@ -32,7 +32,7 @@ func TestEncodeDecodeRoundTrip(t *testing.T) {
if err := json.Unmarshal(b, &raw); err != nil {
if err := json.Unmarshal(b, &raw); err != nil {
t.Fatalf("%s: %v: %s", m.MessageType(), err, b)
t.Fatalf("%s: %v: %s", m.MessageType(), err, b)
}
}
- // heartbeat and heartbeat_ack carry their own seq, which wins over the envelope's (as on Bot's wire).
+ // heartbeat carries its own seq (mirroring the envelope's); heartbeat_ack does too only from older servers.
_, ownSeq := m.(Heartbeat)
_, ownSeq := m.(Heartbeat)
if _, ok := m.(HeartbeatAck); ok {
if _, ok := m.(HeartbeatAck); ok {
ownSeq = true
ownSeq = true
@@ -103,6 +103,11 @@ func TestBotFixtures(t *testing.T) {
@@ -103,6 +103,11 @@ func TestBotFixtures(t *testing.T) {
if f := decode("heartbeat_ack"); f.Seq != 41 || f.Msg.(HeartbeatAck).Seq != 41 {
if f := decode("heartbeat_ack"); f.Seq != 41 || f.Msg.(HeartbeatAck).Seq != 41 {
t.Errorf("heartbeat_ack %+v", f)
t.Errorf("heartbeat_ack %+v", f)
}
}
+ // Bot's newer ack: its own envelope seq, the device's send time echoed as echo and t.
+ if f := decode("heartbeat_ack_echo"); f.Seq != 4 || f.Msg.(HeartbeatAck).T != 1738065600500 ||
+ f.Msg.(HeartbeatAck).Echo == nil || *f.Msg.(HeartbeatAck).Echo != 1738065600500 || f.Msg.(HeartbeatAck).ServerTime == "" {
+ t.Errorf("heartbeat_ack echo %+v", f)
+ }
if e := decode("estop").Msg.(Estop); e.Latched || e.By != "usr_01J8…" {
if e := decode("estop").Msg.(Estop); e.Latched || e.By != "usr_01J8…" {
t.Errorf("estop %+v", e)
t.Errorf("estop %+v", e)
}
}
@@ -1,8 +1,8 @@
@@ -1,8 +1,8 @@
# OpenVibe.Bot device-contract fixtures
# OpenVibe.Bot device-contract fixtures
-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.
+Pinned to **OpenVibe.Bot `b48da7c13dbe8e4ab2e3c2d7b3cf058c67e5fd80`**, except the heartbeat exchange: the
+`heartbeat_ack_echo` fixture is Bot `ace1671`, where `heartbeat_ack` echoes the heartbeat's `t` as both `echo` and
+`t` over Bot's own envelope `seq`. The older `heartbeat_ack` fixture (the `seq` shape) stays as a decode case.
Each `.json` file is one frame copied verbatim (whitespace and `…` included) from the JSON examples in Bot's
Each `.json` file is one frame copied verbatim (whitespace and `…` included) from the JSON examples in Bot's
`docs/protocol.md` at that commit:
`docs/protocol.md` at that commit:
@@ -13,6 +13,7 @@ Each `.json` file is one frame copied verbatim (whitespace and `…` included) f
@@ -13,6 +13,7 @@ Each `.json` file is one frame copied verbatim (whitespace and `…` included) f
| `ack`, `nack`, `heartbeat`, `heartbeat_ack`, `telemetry`, `status`, | |
| `ack`, `nack`, `heartbeat`, `heartbeat_ack`, `telemetry`, `status`, | |
| `estop_state`, `estop` | |
| `estop_state`, `estop` | |
| `error` | section 2 (`/control`): the same `error` frame shape Bot sends on `/device` |
| `error` | section 2 (`/control`): the same `error` frame shape Bot sends on `/device` |
+| `heartbeat_ack_echo` | section 1 (`/device`), "Examples" (Bot `ace1671`) |
The REST pairing answer has no example in the doc; tests derive it from `paired.json` the way Bot's
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, whip_url, robot_id`
`POST /api/v1/pair` handler (`server/api/v1.js`) builds it: `device_id, credential, publish_key, whip_url, robot_id`
@@ -20,7 +21,7 @@ The REST pairing answer has no example in the doc; tests derive it from `paired.
@@ -20,7 +21,7 @@ The REST pairing answer has no example in the doc; tests derive it from `paired.
Used by `internal/protocol` (every frame decodes, with the fields the Node reads), `internal/link`
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
(`TestBotFixturesEndToEnd`: pair → hello + config → command → ack → reconnect through `internal/fakebot`) and
-`internal/node` (`TestBotFramesDriveAndAck`, `TestRecordedExpiredDriveNeverReachesPlugin`). The command examples carry a
+`internal/node` (`TestBotFramesDriveAndAck`, `TestExpiredDriveNeverReachesPlugin`). 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
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
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.
let the tests say what moved.
new file mode 100644
new file mode 100644
@@ -0,0 +1 @@
@@ -0,0 +1 @@
+{ "v": 1, "seq": 4, "ts": 1738065600501, "type": "heartbeat_ack", "echo": 1738065600500, "t": 1738065600500, "server_time": "2026-09-29T19:20:00.501Z" }