openvibers/openvibe.bot Commits
Robot commands and turn halts carry turn_epoch (milliseconds, rising across restarts), so a device can refuse a command from an ended turn (#156)
Changes
@@@ -116,10 +116,16 @@ queued"), `seq` starts again on each connection, and the device sees only serveranswers a repeated id from its dedup cache instead of running it again). A command in flight when thesocket drops is lost, and its operator gets `expired`.--Not carried to a driver today: the turn epoch, Bot's `seq` or `ts`. A plugin sees the command id, kind,--value, the time left and `operator { subject, role }`, so it can tell turns apart only by the operator's--subject. A per-turn epoch in the command frame, checked by Node and passed to plugins, is the planned next--step.++**`turn_epoch`** (contracts 0.213.0) is the robot's control epoch, in milliseconds on Bot's clock, so it keeps rising++across a Bot restart:++- It moves when a turn ends, is left or is taken over. The halt for that change carries the new value, and every++ command carries the epoch current when it is sent. An owner's command right after a takeover carries the epoch its++ halt moved to.++- OpenVibe.Node (Node#55) keeps the newest epoch a halt carried per robot, nacks `expired` any command older than it,++ and passes the epoch to drivers (`cmd.turn_epoch`), so a driver can check it too.++- It is absent until a robot's first turn change since Bot started. A command without it runs as before.++++Bot's `seq` and `ts` are not passed to drivers: `seq` restarts on each connection and orders only that connection.**Owner takeover.** When the owner or an operator sends a command to a `queue` robot while someone holdsthe turn:@@@ -857,7 +857,9 @@ function createDomain({ db, config, outbox, link = null, nodes = null, openre =const epochs = new Map();const epochOf = (robotId) => epochs.get(robotId) || 0;const handedOff = (robotId, from, reason) => {-- epochs.set(robotId, epochOf(robotId) + 1);++ // Milliseconds on Bot's clock (contracts 0.213.0 turn_epoch): it keeps rising across a restart, and two changes in++ // one millisecond still differ.++ epochs.set(robotId, Math.max(epochOf(robotId) + 1, now()));if (epochs.size > 10000) epochs.delete(epochs.keys().next().value);for (const fn of handoffListeners) { try { fn(robotId, { from, reason }); } catch { /* a listener never breaks the caller */ } }};@@@ -560,7 +560,10 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Datreturn sendFrame(conn, 'command_result', { id, result: 'refused', code: 'bot.not_your_turn', reason: 'your turn ended before this command could be sent' });}const deviceCmdId = prefixedId('cmd', now());-- const frame = { v: 1, type: 'command', seq: nextSeq(device), ts: now(), id: deviceCmdId, ref: id, kind: decision.kind, value: decision.value, deadline_ms: decision.deadlineMs, operator: { subject: conn.subject, role: decision.role }, robot_id: robotId };++ // turn_epoch (contracts 0.213.0): the robot's epoch now, which a queue command's check above proved is the one it++ // was decided under; an owner's command that just took a turn over carries the epoch its own halt moved to.++ const epoch = domain.queue.epoch(robotId);++ const frame = { v: 1, type: 'command', seq: nextSeq(device), ts: now(), id: deviceCmdId, ref: id, kind: decision.kind, value: decision.value, deadline_ms: decision.deadlineMs, operator: { subject: conn.subject, role: decision.role }, robot_id: robotId, ...(epoch ? { turn_epoch: epoch } : {}) };if (!send(device.ws, frame)) {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' });@@@ -592,7 +595,9 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Datconst device = deviceForRobot(robotId);if (!device) return;const id = prefixedId('cmd', now());-- if (!send(device.ws, { v: 1, type: 'command', seq: nextSeq(device), ts: now(), id, kind: 'halt', value: {}, operator: { subject: null, role: 'system' }, robot_id: robotId })) return;++ // The halt carries the epoch the change moved to: a device refuses anything older after it (turn_epoch).++ const epoch = domain.queue.epoch(robotId);++ if (!send(device.ws, { v: 1, type: 'command', seq: nextSeq(device), ts: now(), id, kind: 'halt', value: {}, operator: { subject: null, role: 'system' }, robot_id: robotId, ...(epoch ? { turn_epoch: epoch } : {}) })) return;const p = { id, opId: id, key: `system|${id}`, deviceId: device.device.id, conn: SYSTEM_CONN, robotId, subject: null, operatorKind: 'system', role: null, kind: 'halt', value: { reason, turn_of: from || null }, at: now(), timer: null };pendingCmds.set(id, p);p.timer = setTimeout(() => expireCommand(id), 2500);@@@ -250,6 +250,8 @@ const plugins = require('./helpers/plugins');assert.deepStrictEqual(halt.operator, { subject: null, role: 'system' });assert.deepStrictEqual(halt.value, {});assert.strictEqual(halt.deadline_ms, undefined);++ assert.ok(halt.turn_epoch > 0, 'the halt carries the new turn epoch');++ assert.strictEqual(dev.messages.find((m) => m.type === 'command' && m.ref === 'q3').turn_epoch, halt.turn_epoch, 'the next person drives in it');} finally {t.clock.offset -= 120000;}@@@ -350,6 +352,11 @@ const plugins = require('./helpers/plugins');const halt = cmds.find((m) => m.kind === 'halt');dev.send({ type: 'ack', id: halt.id });assert.deepStrictEqual(halt.operator, { subject: null, role: 'system' });++ // turn_epoch (contracts 0.213.0): the halt carries the epoch the takeover moved to, and the owner's command the same++ // one, so a device that refuses commands older than its newest halt still runs the owner's.++ assert.ok(halt.turn_epoch > 0, 'the takeover halt carries its epoch');++ assert.strictEqual(cmds.find((m) => m.ref === 'o2').turn_epoch, halt.turn_epoch, 'the owner\'s command is of the new epoch');++ assert.ok(!(cmds.find((m) => m.ref === 'o1').turn_epoch >= halt.turn_epoch), 'carol\'s earlier command is of an older epoch');await t.wait(30);const row = (await t.domain.audit.list(robot.id, { limit: 6 })).find((x) => x.kind === 'halt');assert.deepStrictEqual(row.value, { reason: 'owner_takeover', turn_of: carol.subject }, 'audited as an owner takeover of carol\'s turn');
@@ -116,10 +116,16 @@ queued"), `seq` starts again on each connection, and the device sees only server
@@ -116,10 +116,16 @@ queued"), `seq` starts again on each connection, and the device sees only server
answers a repeated id from its dedup cache instead of running it again). A command in flight when the
answers a repeated id from its dedup cache instead of running it again). A command in flight when the
socket drops is lost, and its operator gets `expired`.
socket drops is lost, and its operator gets `expired`.
-Not carried to a driver today: the turn epoch, Bot's `seq` or `ts`. A plugin sees the command id, kind,
-value, the time left and `operator { subject, role }`, so it can tell turns apart only by the operator's
-subject. A per-turn epoch in the command frame, checked by Node and passed to plugins, is the planned next
-step.
+**`turn_epoch`** (contracts 0.213.0) is the robot's control epoch, in milliseconds on Bot's clock, so it keeps rising
+across a Bot restart:
+- It moves when a turn ends, is left or is taken over. The halt for that change carries the new value, and every
+ command carries the epoch current when it is sent. An owner's command right after a takeover carries the epoch its
+ halt moved to.
+- OpenVibe.Node (Node#55) keeps the newest epoch a halt carried per robot, nacks `expired` any command older than it,
+ and passes the epoch to drivers (`cmd.turn_epoch`), so a driver can check it too.
+- It is absent until a robot's first turn change since Bot started. A command without it runs as before.
+
+Bot's `seq` and `ts` are not passed to drivers: `seq` restarts on each connection and orders only that connection.
**Owner takeover.** When the owner or an operator sends a command to a `queue` robot while someone holds
**Owner takeover.** When the owner or an operator sends a command to a `queue` robot while someone holds
the turn:
the turn:
@@ -857,7 +857,9 @@ function createDomain({ db, config, outbox, link = null, nodes = null, openre =
@@ -857,7 +857,9 @@ function createDomain({ db, config, outbox, link = null, nodes = null, openre =
const epochs = new Map();
const epochs = new Map();
const epochOf = (robotId) => epochs.get(robotId) || 0;
const epochOf = (robotId) => epochs.get(robotId) || 0;
const handedOff = (robotId, from, reason) => {
const handedOff = (robotId, from, reason) => {
- epochs.set(robotId, epochOf(robotId) + 1);
+ // Milliseconds on Bot's clock (contracts 0.213.0 turn_epoch): it keeps rising across a restart, and two changes in
+ // one millisecond still differ.
+ epochs.set(robotId, Math.max(epochOf(robotId) + 1, now()));
if (epochs.size > 10000) epochs.delete(epochs.keys().next().value);
if (epochs.size > 10000) epochs.delete(epochs.keys().next().value);
for (const fn of handoffListeners) { try { fn(robotId, { from, reason }); } catch { /* a listener never breaks the caller */ } }
for (const fn of handoffListeners) { try { fn(robotId, { from, reason }); } catch { /* a listener never breaks the caller */ } }
};
};
@@ -560,7 +560,10 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
@@ -560,7 +560,10 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
return sendFrame(conn, 'command_result', { id, result: 'refused', code: 'bot.not_your_turn', reason: 'your turn ended before this command could be sent' });
return sendFrame(conn, 'command_result', { id, result: 'refused', code: 'bot.not_your_turn', reason: 'your turn ended before this command could be sent' });
}
}
const deviceCmdId = prefixedId('cmd', now());
const deviceCmdId = prefixedId('cmd', now());
- const frame = { v: 1, type: 'command', seq: nextSeq(device), ts: now(), id: deviceCmdId, ref: id, kind: decision.kind, value: decision.value, deadline_ms: decision.deadlineMs, operator: { subject: conn.subject, role: decision.role }, robot_id: robotId };
+ // turn_epoch (contracts 0.213.0): the robot's epoch now, which a queue command's check above proved is the one it
+ // was decided under; an owner's command that just took a turn over carries the epoch its own halt moved to.
+ const epoch = domain.queue.epoch(robotId);
+ const frame = { v: 1, type: 'command', seq: nextSeq(device), ts: now(), id: deviceCmdId, ref: id, kind: decision.kind, value: decision.value, deadline_ms: decision.deadlineMs, operator: { subject: conn.subject, role: decision.role }, robot_id: robotId, ...(epoch ? { turn_epoch: epoch } : {}) };
if (!send(device.ws, frame)) {
if (!send(device.ws, frame)) {
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' });
@@ -592,7 +595,9 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
@@ -592,7 +595,9 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
const device = deviceForRobot(robotId);
const device = deviceForRobot(robotId);
if (!device) return;
if (!device) return;
const id = prefixedId('cmd', now());
const id = prefixedId('cmd', now());
- if (!send(device.ws, { v: 1, type: 'command', seq: nextSeq(device), ts: now(), id, kind: 'halt', value: {}, operator: { subject: null, role: 'system' }, robot_id: robotId })) return;
+ // The halt carries the epoch the change moved to: a device refuses anything older after it (turn_epoch).
+ const epoch = domain.queue.epoch(robotId);
+ if (!send(device.ws, { v: 1, type: 'command', seq: nextSeq(device), ts: now(), id, kind: 'halt', value: {}, operator: { subject: null, role: 'system' }, robot_id: robotId, ...(epoch ? { turn_epoch: epoch } : {}) })) return;
const p = { id, opId: id, key: `system|${id}`, deviceId: device.device.id, conn: SYSTEM_CONN, robotId, subject: null, operatorKind: 'system', role: null, kind: 'halt', value: { reason, turn_of: from || null }, at: now(), timer: null };
const p = { id, opId: id, key: `system|${id}`, deviceId: device.device.id, conn: SYSTEM_CONN, robotId, subject: null, operatorKind: 'system', role: null, kind: 'halt', value: { reason, turn_of: from || null }, at: now(), timer: null };
pendingCmds.set(id, p);
pendingCmds.set(id, p);
p.timer = setTimeout(() => expireCommand(id), 2500);
p.timer = setTimeout(() => expireCommand(id), 2500);
@@ -250,6 +250,8 @@ const plugins = require('./helpers/plugins');
@@ -250,6 +250,8 @@ const plugins = require('./helpers/plugins');
assert.deepStrictEqual(halt.operator, { subject: null, role: 'system' });
assert.deepStrictEqual(halt.operator, { subject: null, role: 'system' });
assert.deepStrictEqual(halt.value, {});
assert.deepStrictEqual(halt.value, {});
assert.strictEqual(halt.deadline_ms, undefined);
assert.strictEqual(halt.deadline_ms, undefined);
+ assert.ok(halt.turn_epoch > 0, 'the halt carries the new turn epoch');
+ assert.strictEqual(dev.messages.find((m) => m.type === 'command' && m.ref === 'q3').turn_epoch, halt.turn_epoch, 'the next person drives in it');
} finally {
} finally {
t.clock.offset -= 120000;
t.clock.offset -= 120000;
}
}
@@ -350,6 +352,11 @@ const plugins = require('./helpers/plugins');
@@ -350,6 +352,11 @@ const plugins = require('./helpers/plugins');
const halt = cmds.find((m) => m.kind === 'halt');
const halt = cmds.find((m) => m.kind === 'halt');
dev.send({ type: 'ack', id: halt.id });
dev.send({ type: 'ack', id: halt.id });
assert.deepStrictEqual(halt.operator, { subject: null, role: 'system' });
assert.deepStrictEqual(halt.operator, { subject: null, role: 'system' });
+ // turn_epoch (contracts 0.213.0): the halt carries the epoch the takeover moved to, and the owner's command the same
+ // one, so a device that refuses commands older than its newest halt still runs the owner's.
+ assert.ok(halt.turn_epoch > 0, 'the takeover halt carries its epoch');
+ assert.strictEqual(cmds.find((m) => m.ref === 'o2').turn_epoch, halt.turn_epoch, 'the owner\'s command is of the new epoch');
+ assert.ok(!(cmds.find((m) => m.ref === 'o1').turn_epoch >= halt.turn_epoch), 'carol\'s earlier command is of an older epoch');
await t.wait(30);
await t.wait(30);
const row = (await t.domain.audit.list(robot.id, { limit: 6 })).find((x) => x.kind === 'halt');
const row = (await t.domain.audit.list(robot.id, { limit: 6 })).find((x) => x.kind === 'halt');
assert.deepStrictEqual(row.value, { reason: 'owner_takeover', turn_of: carol.subject }, 'audited as an owner takeover of carol\'s turn');
assert.deepStrictEqual(row.value, { reason: 'owner_takeover', turn_of: carol.subject }, 'audited as an owner takeover of carol\'s turn');