openvibers/openvibe.link Commits

Allow 10 s for hello, tag estopstate with robotid, lock WHIP key (#10)

OP OpenVibers committed 2 Oct 2026 · d5c7212b09 · 7 files, +106 −13

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

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

Changes

ddiff --git a/docs/protocol.md b/docs/protocol.md
iindex d664cd0..36d4a77 100644
---- a/docs/protocol.md
++++ b/docs/protocol.md
@@@ -126,7 +126,7 @@ applied the Node runs nothing but `halt` (`nack not_ready`), so that latch is in
| `ack` | `id`; Node extra: `latency_ms` (receipt → plugin ack) |
| `nack` | `id`, `fault_code`; Node extra: `message` |
| `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: `robot_id` (when `hello` named exactly one robot), `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
upgrade or `hello`; then `status` whenever a plugin changes state or faults and `estop_state` after every `config`,
@@@ -186,7 +186,9 @@ Commands are never queued while offline and never replayed after a reconnect.
### Heartbeat, deadman, reconnect
--- The device sends `heartbeat` every `config.heartbeat_ms` (default 1000). The server answers `heartbeat_ack`.
++- The device sends `heartbeat` every `config.heartbeat_ms` (default 1000), starting once `hello` has arrived. The
++ server answers `heartbeat_ack`. Until `hello`, a new connection may stay silent for up to 10 s (Bot may still be
++ checking the credential); the deadman below applies from `hello` on.
- If nothing at all arrives from the server for 2 × `heartbeat_ms` (two missed heartbeats), the Node closes the link.
Every close stops every actuator at once. (Bot marks the device offline after `heartbeat_ms × 2 + 3000 ms`.)
- Reconnect: exponential backoff from 0.5 s to 30 s with full jitter; see the close-code table for `4000`, `4002` and
ddiff --git a/internal/link/link.go b/internal/link/link.go
iindex e9b7d4f..573785c 100644
---- a/internal/link/link.go
++++ b/internal/link/link.go
@@@ -1,9 +1,10 @@
// Package link keeps the Node's one outbound control connection to OpenVibe.Bot: a WebSocket to wss://<origin>/device
// authenticated with the device credential in the Authorization header of the upgrade (never in the URL). It sends a
--// heartbeat every interval, declares the link lost when nothing has arrived for the deadman interval (two heartbeats
--// by default), and reconnects with exponential backoff and full jitter. Nothing is queued while offline: Send fails
--// 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.
++// heartbeat every interval from the server's hello on, declares the link lost when nothing has arrived for the deadman
++// interval (two heartbeats by default; HelloAllowance before hello), and reconnects with exponential backoff and full
++// jitter. Nothing is queued while offline: Send fails 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) reconnects with the normal backoff;
// 4002 (credential refused) and 4003 (revoked) keep retrying, never sooner than CredentialRetryMin, and log what to
@@@ -64,6 +65,10 @@ type Options struct {
// the next try; jitter only ever adds to it.
const CredentialRetryMin = 10 * time.Second
++// HelloAllowance is how long a new connection may stay silent before the server's hello when the deadman is shorter:
++// Bot authenticates after the upgrade, and heartbeats (and so the deadman) only start once hello has arrived.
++const HelloAllowance = 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 {
@@@ -365,6 +370,7 @@ func (l *Link) session(ctx context.Context) error {
end(ctx.Err())
}()
++ var up atomic.Bool // the server's hello has arrived
var lastRx atomic.Int64
lastRx.Store(time.Now().UnixNano())
@@@ -399,6 +405,13 @@ func (l *Link) session(ctx context.Context) error {
if hbNow != hb {
hb, next = hbNow, now
}
++ if !up.Load() {
++ if now.Sub(time.Unix(0, lastRx.Load())) > max(deadman, HelloAllowance) {
++ end(fmt.Errorf("no hello from the server within %s", max(deadman, HelloAllowance)))
++ return
++ }
++ continue
++ }
if !now.Before(next) {
next = now.Add(hb)
_ = l.Send(protocol.Heartbeat{})
@@@ -411,7 +424,6 @@ func (l *Link) session(ctx context.Context) error {
}
}()
-- up := false
for {
_, data, err := conn.ReadMessage()
if err != nil {
@@@ -479,8 +491,8 @@ func (l *Link) session(ctx context.Context) error {
}
continue
case protocol.Hello:
-- if !up {
-- up = true
++ if !up.Load() {
++ up.Store(true)
l.mu.Lock()
l.out = out
l.mu.Unlock()
@@@ -490,7 +502,7 @@ func (l *Link) session(ctx context.Context) error {
l.handler.Connected()
}
}
-- if !up {
++ if !up.Load() {
l.log.Warn("ignoring a frame before hello", "type", f.Type)
continue
}
ddiff --git a/internal/link/link_test.go b/internal/link/link_test.go
iindex ba1ee91..cc3e265 100644
---- a/internal/link/link_test.go
++++ b/internal/link/link_test.go
@@@ -477,6 +477,15 @@ func TestNothingSentBeforeHello(t *testing.T) {
if err := l.Send(protocol.Heartbeat{}); !errors.Is(err, ErrOffline) {
t.Fatalf("send before hello: %v", err)
}
++ // The deadman (40 ms) waits for hello: until HelloAllowance a server still checking the credential keeps the
++ // connection, and no heartbeat goes out.
++ time.Sleep(500 * time.Millisecond)
++ if n := srv.Dials(); n != 1 {
++ t.Fatalf("dialed %d times while waiting for hello", n)
++ }
++ if len(c.Frames()) != 0 {
++ t.Fatalf("sent %d frames before hello", len(c.Frames()))
++ }
}
func TestDeviceURL(t *testing.T) {
ddiff --git a/internal/node/node.go b/internal/node/node.go
iindex 2fc0033..2ed526b 100644
---- a/internal/node/node.go
++++ b/internal/node/node.go
@@@ -76,6 +76,7 @@ type Node struct {
// 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
++ robotIDs []string // from this connection's hello
}
// New builds a Node. Run starts it.
@@@ -193,7 +194,7 @@ func (n *Node) Run(ctx context.Context) error {
// has been applied (applyConfig), so they report the latch the owner may have set while the link was down.
func (n *Node) Connected() {
n.mu.Lock()
-- n.configured = false
++ n.configured, n.robotIDs = false, nil
n.mu.Unlock()
}
@@@ -208,6 +209,9 @@ func (n *Node) Disconnected(err error) {
func (n *Node) Frame(f protocol.Frame) {
switch m := f.Msg.(type) {
case protocol.Hello:
++ n.mu.Lock()
++ n.robotIDs = m.RobotIDs
++ n.mu.Unlock()
n.log.Info("server hello", "device", m.DeviceID, "robots", m.RobotIDs, "session", m.SessionID)
case protocol.Config:
n.applyConfig(m)
@@@ -445,8 +449,15 @@ func (n *Node) sendEstopState() {
return
}
st := n.latch.State()
++ // robot_id names the robot only when the device serves exactly one; a bridge for several leaves it out.
++ var robot string
++ n.mu.Lock()
++ if len(n.robotIDs) == 1 {
++ robot = n.robotIDs[0]
++ }
++ n.mu.Unlock()
n.send(protocol.EstopState{Latched: st.Stopped(), By: "device", At: time.Now().UTC().Format(time.RFC3339Nano),
-- LocalStop: st.Local, Reason: st.RemoteReason})
++ RobotID: robot, LocalStop: st.Local, Reason: st.RemoteReason})
}
// ---- plugin output ----
ddiff --git a/internal/node/node_test.go b/internal/node/node_test.go
iindex b99694c..f6ee8e2 100644
---- a/internal/node/node_test.go
++++ b/internal/node/node_test.go
@@@ -230,11 +230,20 @@ func TestPairConnectAndStatus(t *testing.T) {
if f.Msg.(protocol.EstopState).Latched {
t.Fatal("latched on a fresh device")
}
++ if r := f.Msg.(protocol.EstopState).RobotID; r != fakebot.RobotID {
++ t.Fatalf("estop_state robot_id %q on a one-robot device", r)
++ }
}
}
if !sawEstopState {
t.Fatal("no estop_state after connect")
}
++ // A device serving several robots cannot name one: robot_id is left out.
++ e.conn.Send(protocol.Hello{SessionID: "sess_2", DeviceID: "dev_1", RobotIDs: []string{fakebot.RobotID, "rob_other"}})
++ e.conn.Send(protocol.Estop{Latched: false, By: "usr_owner"})
++ if _, err := e.conn.Expect(protocol.TypeEstopState, 2*time.Second, func(m protocol.Message) bool { return m.(protocol.EstopState).RobotID == "" }); err != nil {
++ t.Fatal(err)
++ }
if e.srv.CredentialSeenInURL() {
t.Fatal("credential in URL")
}
ddiff --git a/internal/video/whip_test.go b/internal/video/whip_test.go
iindex 4131299..266c8c2 100644
---- a/internal/video/whip_test.go
++++ b/internal/video/whip_test.go
@@@ -4,7 +4,9 @@ import (
"context"
"io"
"log/slog"
++ "net/http"
"net/http/httptest"
++ "sync"
"testing"
"time"
@@@ -64,6 +66,43 @@ func TestPublishRetriesOnBadKey(t *testing.T) {
}
}
++// The receiver's key can change while it serves (a fake Bot rotating the publish key): under -race this is the check
++// that ServeHTTP reads it under the lock.
++func TestWHIPReceiverSetPublishKeyWhileServing(t *testing.T) {
++ rx := &WHIPReceiver{PublishKey: "old"}
++ srv := httptest.NewServer(rx)
++ defer srv.Close()
++ del := func(key string) int {
++ req, _ := http.NewRequest(http.MethodDelete, srv.URL+"/whip/x/session", nil)
++ req.Header.Set("Authorization", "Bearer "+key)
++ resp, err := http.DefaultClient.Do(req)
++ if err != nil {
++ t.Error(err)
++ return 0
++ }
++ resp.Body.Close()
++ return resp.StatusCode
++ }
++ var wg sync.WaitGroup
++ for i := 0; i < 4; i++ {
++ wg.Add(1)
++ go func() {
++ defer wg.Done()
++ for j := 0; j < 10; j++ {
++ del("new")
++ }
++ }()
++ }
++ rx.SetPublishKey("new")
++ wg.Wait()
++ if c := del("old"); c != http.StatusUnauthorized {
++ t.Fatalf("old key after SetPublishKey: HTTP %d", c)
++ }
++ if c := del("new"); c == http.StatusUnauthorized {
++ t.Fatal("new key refused after SetPublishKey")
++ }
++}
++
func contains(s, sub string) bool {
for i := 0; i+len(sub) <= len(s); i++ {
if s[i:i+len(sub)] == sub {
ddiff --git a/internal/video/whiptest.go b/internal/video/whiptest.go
iindex 8f9299e..5b47bca 100644
---- a/internal/video/whiptest.go
++++ b/internal/video/whiptest.go
@@@ -26,8 +26,19 @@ type WHIPReceiver struct {
n int
}
++// SetPublishKey changes the bearer key the receiver demands. It is safe to call while the receiver is serving; set
++// PublishKey directly only before then.
++func (w *WHIPReceiver) SetPublishKey(k string) {
++ w.mu.Lock()
++ defer w.mu.Unlock()
++ w.PublishKey = k
++}
++
func (w *WHIPReceiver) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
-- if r.Header.Get("Authorization") != "Bearer "+w.PublishKey {
++ w.mu.Lock()
++ key := w.PublishKey
++ w.mu.Unlock()
++ if r.Header.Get("Authorization") != "Bearer "+key {
w.BadAuth.Add(1)
http.Error(rw, "unauthorized", http.StatusUnauthorized)
return