openvibers/openrestream Commits
Worker runtime: hooks can no longer reject unhandled, and an idle draining worker records its stop before exiting. Hooks (onDrain, onDrainDeadline, onEndRequested, onHeartbeat) run beside the heartbeat as before (they stop publishers and wait for them to leave; awaiting them in the beat stalled the end-to-end flow), now through one call() that logs a failure whether the hook is sync or async. exitWhenIdle() writes the 'stopped' row and only then exits: before, the write could be lost, leaving the worker ready/draining until its lease expired and new sessions routed to a dead generation. The coordinator test waits for the exit instead of one tick. npm test 15/18 (3 need PostgreSQL), npm run test:pg 18/18.
Changes
@@@ -154,7 +154,8 @@ t('the worker runtime exits by itself when a draining generation is idle, and onassert.deepStrictEqual(exits, [], 'still carrying an output');active = 0;await wr.beat();-- await new Promise(r => setImmediate(r));++ // The exit comes after the stopped row is written (a real query), not on the next tick.++ for (let i = 0; i < 100 && !exits.length; i++) await new Promise(r => setTimeout(r, 10));assert.deepStrictEqual(exits, [0]);assert.strictEqual((await rt.store.workers.get(wr.me.id)).state, 'stopped');@@@ -164,7 +165,7 @@ t('the worker runtime exits by itself when a draining generation is idle, and onawait wr2.ready();await rt.store.workers.lose(wr2.me.id, 'test');await wr2.beat();-- await new Promise(r => setImmediate(r));++ for (let i = 0; i < 100 && !exits2.length; i++) await new Promise(r => setTimeout(r, 10));assert.deepStrictEqual(exits2, [1]);});@@@ -65,25 +65,28 @@ function createWorkerRuntime({ rt, kind, log = console, hooks = {}, exit = (codeif (row.state === 'draining' && !draining) {draining = true;log.log(`[${kind}] generation ${row.generation} draining (deadline ${new Date(row.drain_deadline).toISOString()})`);-- try { hooks.onDrain && hooks.onDrain(row); } catch (err) { log.error(`[${kind}] onDrain: ${err.message}`); }++ call('onDrain', row);}if (draining && row.drain_deadline && row.drain_deadline <= Date.now() && !deadlineHandled) {deadlineHandled = true;log.warn(`[${kind}] drain deadline reached; ending remaining transports`);-- try { hooks.onDrainDeadline && hooks.onDrainDeadline(row); } catch (err) { log.error(`[${kind}] onDrainDeadline: ${err.message}`); }++ call('onDrainDeadline', row);}if (hooks.onEndRequested) {-- for (const s of await store.sessions.endRequestsFor(me.id)) {-- try { hooks.onEndRequested(s); } catch (err) { log.error(`[${kind}] onEndRequested: ${err.message}`); }-- }-- }-- if (hooks.onHeartbeat) {-- try { hooks.onHeartbeat(row); } catch (err) { log.error(`[${kind}] onHeartbeat: ${err.message}`); }++ for (const s of await store.sessions.endRequestsFor(me.id)) call('onEndRequested', s);}++ call('onHeartbeat', row);if (draining) exitWhenIdle();return row;}++ /** A hook, run beside the heartbeat (they stop publishers and wait for them to leave, so the beat must not wait on++ * them); async ones used to reject unhandled — a failure is now logged, sync or async. */++ function call(name, ...args) {++ if (typeof hooks[name] !== 'function') return;++ Promise.resolve().then(() => hooks[name](...args)).catch((err) => log.error(`[${kind}] ${name}: ${err.message}`));++ }++function schedule() {clearTimeout(timer);if (stopped) return;@@@ -108,9 +111,12 @@ function createWorkerRuntime({ rt, kind, log = console, hooks = {}, exit = (codeif (active > 0 || stopped) return false;stopped = true;clearTimeout(timer);-- store.workers.stop(me.id, draining ? 'drained' : 'exit');log.log(`[${kind}] generation ${me.generation} idle after drain; exiting`);-- settle(hooks.onExit).finally(() => exit(0));++ // The stopped row is written before the process goes: otherwise the worker stays ready/draining until its lease++ // expires and new sessions can be routed to a generation that is gone.++ Promise.resolve().then(() => store.workers.stop(me.id, draining ? 'drained' : 'exit'))++ .catch((err) => log.error(`[${kind}] could not record the stop: ${err.message}`))++ .finally(() => settle(hooks.onExit).finally(() => exit(0)));return true;}
@@ -154,7 +154,8 @@ t('the worker runtime exits by itself when a draining generation is idle, and on
@@ -154,7 +154,8 @@ t('the worker runtime exits by itself when a draining generation is idle, and on
assert.deepStrictEqual(exits, [], 'still carrying an output');
assert.deepStrictEqual(exits, [], 'still carrying an output');
active = 0;
active = 0;
await wr.beat();
await wr.beat();
- await new Promise(r => setImmediate(r));
+ // The exit comes after the stopped row is written (a real query), not on the next tick.
+ for (let i = 0; i < 100 && !exits.length; i++) await new Promise(r => setTimeout(r, 10));
assert.deepStrictEqual(exits, [0]);
assert.deepStrictEqual(exits, [0]);
assert.strictEqual((await rt.store.workers.get(wr.me.id)).state, 'stopped');
assert.strictEqual((await rt.store.workers.get(wr.me.id)).state, 'stopped');
@@ -164,7 +165,7 @@ t('the worker runtime exits by itself when a draining generation is idle, and on
@@ -164,7 +165,7 @@ t('the worker runtime exits by itself when a draining generation is idle, and on
await wr2.ready();
await wr2.ready();
await rt.store.workers.lose(wr2.me.id, 'test');
await rt.store.workers.lose(wr2.me.id, 'test');
await wr2.beat();
await wr2.beat();
- await new Promise(r => setImmediate(r));
+ for (let i = 0; i < 100 && !exits2.length; i++) await new Promise(r => setTimeout(r, 10));
assert.deepStrictEqual(exits2, [1]);
assert.deepStrictEqual(exits2, [1]);
});
});
@@ -65,25 +65,28 @@ function createWorkerRuntime({ rt, kind, log = console, hooks = {}, exit = (code
@@ -65,25 +65,28 @@ function createWorkerRuntime({ rt, kind, log = console, hooks = {}, exit = (code
if (row.state === 'draining' && !draining) {
if (row.state === 'draining' && !draining) {
draining = true;
draining = true;
log.log(`[${kind}] generation ${row.generation} draining (deadline ${new Date(row.drain_deadline).toISOString()})`);
log.log(`[${kind}] generation ${row.generation} draining (deadline ${new Date(row.drain_deadline).toISOString()})`);
- try { hooks.onDrain && hooks.onDrain(row); } catch (err) { log.error(`[${kind}] onDrain: ${err.message}`); }
+ call('onDrain', row);
}
}
if (draining && row.drain_deadline && row.drain_deadline <= Date.now() && !deadlineHandled) {
if (draining && row.drain_deadline && row.drain_deadline <= Date.now() && !deadlineHandled) {
deadlineHandled = true;
deadlineHandled = true;
log.warn(`[${kind}] drain deadline reached; ending remaining transports`);
log.warn(`[${kind}] drain deadline reached; ending remaining transports`);
- try { hooks.onDrainDeadline && hooks.onDrainDeadline(row); } catch (err) { log.error(`[${kind}] onDrainDeadline: ${err.message}`); }
+ call('onDrainDeadline', row);
}
}
if (hooks.onEndRequested) {
if (hooks.onEndRequested) {
- for (const s of await store.sessions.endRequestsFor(me.id)) {
- try { hooks.onEndRequested(s); } catch (err) { log.error(`[${kind}] onEndRequested: ${err.message}`); }
- }
- }
- if (hooks.onHeartbeat) {
- try { hooks.onHeartbeat(row); } catch (err) { log.error(`[${kind}] onHeartbeat: ${err.message}`); }
+ for (const s of await store.sessions.endRequestsFor(me.id)) call('onEndRequested', s);
}
}
+ call('onHeartbeat', row);
if (draining) exitWhenIdle();
if (draining) exitWhenIdle();
return row;
return row;
}
}
+ /** A hook, run beside the heartbeat (they stop publishers and wait for them to leave, so the beat must not wait on
+ * them); async ones used to reject unhandled — a failure is now logged, sync or async. */
+ function call(name, ...args) {
+ if (typeof hooks[name] !== 'function') return;
+ Promise.resolve().then(() => hooks[name](...args)).catch((err) => log.error(`[${kind}] ${name}: ${err.message}`));
+ }
+
function schedule() {
function schedule() {
clearTimeout(timer);
clearTimeout(timer);
if (stopped) return;
if (stopped) return;
@@ -108,9 +111,12 @@ function createWorkerRuntime({ rt, kind, log = console, hooks = {}, exit = (code
@@ -108,9 +111,12 @@ function createWorkerRuntime({ rt, kind, log = console, hooks = {}, exit = (code
if (active > 0 || stopped) return false;
if (active > 0 || stopped) return false;
stopped = true;
stopped = true;
clearTimeout(timer);
clearTimeout(timer);
- store.workers.stop(me.id, draining ? 'drained' : 'exit');
log.log(`[${kind}] generation ${me.generation} idle after drain; exiting`);
log.log(`[${kind}] generation ${me.generation} idle after drain; exiting`);
- settle(hooks.onExit).finally(() => exit(0));
+ // The stopped row is written before the process goes: otherwise the worker stays ready/draining until its lease
+ // expires and new sessions can be routed to a generation that is gone.
+ Promise.resolve().then(() => store.workers.stop(me.id, draining ? 'drained' : 'exit'))
+ .catch((err) => log.error(`[${kind}] could not record the stop: ${err.message}`))
+ .finally(() => settle(hooks.onExit).finally(() => exit(0)));
return true;
return true;
}
}