openvibers/openvibe.bot Commits
Protect active robot sessions during Bot deploys (#382)
Changes
@@@ -39,8 +39,8 @@ function createBotReadiness({ db, valkey = null, keys, config, outbox, hub = nul{ name: 'events', required: false, cacheMs: outbox.enabled ? PING_TTL_MS : 0, timeoutMs: 2500, check: events },],details: async (body) => (body.checks.db.status === 'ok'-- ? { events_outbox: await outbox.status(), devices_online: hub ? hub.onlineCount() : 0, queue_waiting: await queueWaiting(db) }-- : { events_outbox: null, devices_online: null, queue_waiting: null }),++ ? { events_outbox: await outbox.status(), devices_online: hub ? hub.onlineCount() : 0, queue_waiting: await queueWaiting(db), protected_sessions: hub ? hub.protectedSessions() : 0 }++ : { events_outbox: null, devices_online: null, queue_waiting: null, protected_sessions: null }),});}@@@ -136,6 +136,8 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Datconst pendingCmds = new Map(); // device-facing command id (server-minted) → pendingconst results = new Map(); // operator key (who + their id) → { result, reason, at }const inflight = new Map(); // operator key → device-facing id, while the command is pending++ const lastRobotCommandAt = new Map();++ const movementLeaseUntil = new Map();// The person always wins a screen (machines): `${device}|${main|private}` → until when a person's input holds it.const personTurns = new Map();const personLast = new Map(); // the same key → when a person last used that screen (ms), for desk turns@@@ -664,6 +666,8 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Datawait domain.audit.record({ robotId, deviceId: device.device.id, subject: conn.subject, operatorKind: conn.principal.kind, role: decision.role, kind: decision.kind, value: decision.value, result: 'refused', reason: 'bot.device_offline' });return sendFrame(conn, 'command_result', { id, result: 'refused', code: 'bot.device_offline', reason: 'the device is offline' });}++ lastRobotCommandAt.set(robotId, now());++ if (decision.deadlineMs != null) movementLeaseUntil.set(robotId, Math.max(movementLeaseUntil.get(robotId) || 0, decision.deadlineMs));const p = { id: deviceCmdId, opId: id, key, deviceId: device.device.id, conn, robotId, subject: conn.subject, operatorKind: conn.principal.kind, role: decision.role, kind: decision.kind, value: decision.value, at: now(), timer: null };pendingCmds.set(deviceCmdId, p);inflight.set(key, deviceCmdId);@@@ -1117,6 +1121,16 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => DatmachineCommand, machineScreen, computerOf, machineChanged,// Real devices only: a simulator is not a machine anyone runs.onlineCount() { let n = 0; for (const c of deviceConns.values()) if (c.online && !c.sim) n++; return n; },++ protectedSessions() {++ const at = now();++ const protectedRobots = new Set();++ for (const [robotId, set] of subsByRobot) {++ const hasOperator = [...set].some((c) => !c.watcher && c.robotId === robotId && c.ws.readyState === OPEN);++ if ((hasOperator && at - (lastRobotCommandAt.get(robotId) || 0) < 120_000) || (movementLeaseUntil.get(robotId) || 0) > at) protectedRobots.add(robotId);++ }++ for (const [robotId, until] of movementLeaseUntil) if (until > at) protectedRobots.add(robotId);++ return protectedRobots.size;++ },devices() { return [...deviceConns.values()].filter((c) => !c.sim).map((c) => ({ device_id: c.device.id, online: c.online })); },async close() {closed = true;@@@ -46,11 +46,13 @@ const plugins = require('./helpers/plugins');const b = await op(bob);const j = await joinResult(b, robot.id);assert.strictEqual(j.role, 'operator');++ assert.strictEqual(t.app.locals.hub.protectedSessions(), 0, 'an idle connected operator is not protected');cmd(b, { id: 'k1', kind: 'drive', value: { throttle: 5, steer: 5 } });const c = await dev.waitFor((m) => m.type === 'command' && m.ref === 'k1');assert.deepStrictEqual(c.value, { throttle: 0.4, steer: 0.3 });dev.send({ type: 'ack', id: c.id });await result(b, 'k1');++ assert.strictEqual(t.app.locals.hub.protectedSessions(), 1, 'an accepted command protects the robot while its control socket remains open');const audit = await t.domain.audit.list(robot.id, { limit: 1 });assert.deepStrictEqual(audit[0].value, { throttle: 0.4, steer: 0.3 });b.close();
@@ -39,8 +39,8 @@ function createBotReadiness({ db, valkey = null, keys, config, outbox, hub = nul
@@ -39,8 +39,8 @@ function createBotReadiness({ db, valkey = null, keys, config, outbox, hub = nul
{ name: 'events', required: false, cacheMs: outbox.enabled ? PING_TTL_MS : 0, timeoutMs: 2500, check: events },
{ name: 'events', required: false, cacheMs: outbox.enabled ? PING_TTL_MS : 0, timeoutMs: 2500, check: events },
],
],
details: async (body) => (body.checks.db.status === 'ok'
details: async (body) => (body.checks.db.status === 'ok'
- ? { events_outbox: await outbox.status(), devices_online: hub ? hub.onlineCount() : 0, queue_waiting: await queueWaiting(db) }
- : { events_outbox: null, devices_online: null, queue_waiting: null }),
+ ? { events_outbox: await outbox.status(), devices_online: hub ? hub.onlineCount() : 0, queue_waiting: await queueWaiting(db), protected_sessions: hub ? hub.protectedSessions() : 0 }
+ : { events_outbox: null, devices_online: null, queue_waiting: null, protected_sessions: null }),
});
});
}
}
@@ -136,6 +136,8 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
@@ -136,6 +136,8 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
const pendingCmds = new Map(); // device-facing command id (server-minted) → pending
const pendingCmds = new Map(); // device-facing command id (server-minted) → pending
const results = new Map(); // operator key (who + their id) → { result, reason, at }
const results = new Map(); // operator key (who + their id) → { result, reason, at }
const inflight = new Map(); // operator key → device-facing id, while the command is pending
const inflight = new Map(); // operator key → device-facing id, while the command is pending
+ const lastRobotCommandAt = new Map();
+ const movementLeaseUntil = new Map();
// The person always wins a screen (machines): `${device}|${main|private}` → until when a person's input holds it.
// The person always wins a screen (machines): `${device}|${main|private}` → until when a person's input holds it.
const personTurns = new Map();
const personTurns = new Map();
const personLast = new Map(); // the same key → when a person last used that screen (ms), for desk turns
const personLast = new Map(); // the same key → when a person last used that screen (ms), for desk turns
@@ -664,6 +666,8 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
@@ -664,6 +666,8 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
await domain.audit.record({ robotId, deviceId: device.device.id, subject: conn.subject, operatorKind: conn.principal.kind, role: decision.role, kind: decision.kind, value: decision.value, result: 'refused', reason: 'bot.device_offline' });
await domain.audit.record({ robotId, deviceId: device.device.id, subject: conn.subject, operatorKind: conn.principal.kind, role: decision.role, kind: decision.kind, value: decision.value, result: 'refused', reason: 'bot.device_offline' });
return sendFrame(conn, 'command_result', { id, result: 'refused', code: 'bot.device_offline', reason: 'the device is offline' });
return sendFrame(conn, 'command_result', { id, result: 'refused', code: 'bot.device_offline', reason: 'the device is offline' });
}
}
+ lastRobotCommandAt.set(robotId, now());
+ if (decision.deadlineMs != null) movementLeaseUntil.set(robotId, Math.max(movementLeaseUntil.get(robotId) || 0, decision.deadlineMs));
const p = { id: deviceCmdId, opId: id, key, deviceId: device.device.id, conn, robotId, subject: conn.subject, operatorKind: conn.principal.kind, role: decision.role, kind: decision.kind, value: decision.value, at: now(), timer: null };
const p = { id: deviceCmdId, opId: id, key, deviceId: device.device.id, conn, robotId, subject: conn.subject, operatorKind: conn.principal.kind, role: decision.role, kind: decision.kind, value: decision.value, at: now(), timer: null };
pendingCmds.set(deviceCmdId, p);
pendingCmds.set(deviceCmdId, p);
inflight.set(key, deviceCmdId);
inflight.set(key, deviceCmdId);
@@ -1117,6 +1121,16 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
@@ -1117,6 +1121,16 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
machineCommand, machineScreen, computerOf, machineChanged,
machineCommand, machineScreen, computerOf, machineChanged,
// Real devices only: a simulator is not a machine anyone runs.
// Real devices only: a simulator is not a machine anyone runs.
onlineCount() { let n = 0; for (const c of deviceConns.values()) if (c.online && !c.sim) n++; return n; },
onlineCount() { let n = 0; for (const c of deviceConns.values()) if (c.online && !c.sim) n++; return n; },
+ protectedSessions() {
+ const at = now();
+ const protectedRobots = new Set();
+ for (const [robotId, set] of subsByRobot) {
+ const hasOperator = [...set].some((c) => !c.watcher && c.robotId === robotId && c.ws.readyState === OPEN);
+ if ((hasOperator && at - (lastRobotCommandAt.get(robotId) || 0) < 120_000) || (movementLeaseUntil.get(robotId) || 0) > at) protectedRobots.add(robotId);
+ }
+ for (const [robotId, until] of movementLeaseUntil) if (until > at) protectedRobots.add(robotId);
+ return protectedRobots.size;
+ },
devices() { return [...deviceConns.values()].filter((c) => !c.sim).map((c) => ({ device_id: c.device.id, online: c.online })); },
devices() { return [...deviceConns.values()].filter((c) => !c.sim).map((c) => ({ device_id: c.device.id, online: c.online })); },
async close() {
async close() {
closed = true;
closed = true;
@@ -46,11 +46,13 @@ const plugins = require('./helpers/plugins');
@@ -46,11 +46,13 @@ const plugins = require('./helpers/plugins');
const b = await op(bob);
const b = await op(bob);
const j = await joinResult(b, robot.id);
const j = await joinResult(b, robot.id);
assert.strictEqual(j.role, 'operator');
assert.strictEqual(j.role, 'operator');
+ assert.strictEqual(t.app.locals.hub.protectedSessions(), 0, 'an idle connected operator is not protected');
cmd(b, { id: 'k1', kind: 'drive', value: { throttle: 5, steer: 5 } });
cmd(b, { id: 'k1', kind: 'drive', value: { throttle: 5, steer: 5 } });
const c = await dev.waitFor((m) => m.type === 'command' && m.ref === 'k1');
const c = await dev.waitFor((m) => m.type === 'command' && m.ref === 'k1');
assert.deepStrictEqual(c.value, { throttle: 0.4, steer: 0.3 });
assert.deepStrictEqual(c.value, { throttle: 0.4, steer: 0.3 });
dev.send({ type: 'ack', id: c.id });
dev.send({ type: 'ack', id: c.id });
await result(b, 'k1');
await result(b, 'k1');
+ assert.strictEqual(t.app.locals.hub.protectedSessions(), 1, 'an accepted command protects the robot while its control socket remains open');
const audit = await t.domain.audit.list(robot.id, { limit: 1 });
const audit = await t.domain.audit.list(robot.id, { limit: 1 });
assert.deepStrictEqual(audit[0].value, { throttle: 0.4, steer: 0.3 });
assert.deepStrictEqual(audit[0].value, { throttle: 0.4, steer: 0.3 });
b.close();
b.close();