openvibers/openvibe.events Commits
The worker stops after the pass it is in and what that pass started (#217)
Changes
@@@ -138,6 +138,7 @@ function createWorker({ store, config, clock = { now: () => Date.now() }, fetchIlet finished = 0; // deliveries settled so far (drain)let passing = null;let queued = null;++ let stopping = false; // stop() was called: no new pass claims anythingfunction dispatch() {if (!passing) {passing = dispatchPass().finally(() => { passing = null; });@@@ -150,6 +151,7 @@ function createWorker({ store, config, clock = { now: () => Date.now() }, fetchI/** Start every due delivery there is room for. Returns how many were started. */async function dispatchPass() {lastTickAt = Date.now();++ if (stopping) return 0;const room = opts.maxInflight - inflight.size;if (room <= 0) return 0;const now = clock.now();@@@ -198,8 +200,17 @@ function createWorker({ store, config, clock = { now: () => Date.now() }, fetchI}return {-- start() { if (!running) { running = true; schedule(0); } },-- async stop() { running = false; clearTimeout(timer); timer = null; await Promise.allSettled([...inflight]); },++ start() { if (!running) { running = true; stopping = false; schedule(0); } },++ // A pass that is reading the due rows when the service stops still starts what it claimed (those rows are++ // leased to this worker), so stop() waits for the pass and then for every delivery it started: the database++ // pool is closed after this returns, and a delivery recording its attempt on a closed pool crashed and was++ // sent again by the next worker (2026-11-02, both deploys of the day). No new pass claims anything once++ // stopping; a later start() lifts that.++ async stop() {++ running = false; stopping = true; clearTimeout(timer); timer = null;++ for (let i = 0; i < 3 && (passing || queued); i++) await Promise.allSettled([queued, passing].filter(Boolean));++ while (inflight.size) await Promise.allSettled([...inflight]);++ },kick() { schedule(0); },// The carrier path (ADR-042 decision 5): claim one row (store.claimDelivery) and send it through the same send().deliver(delivery) { return send(delivery); },nnew file mode 100644@@@ -0,0 +1,75 @@++'use strict';++// The worker stops cleanly: a pass that is reading the due rows when the service stops still starts what it claimed++// (those rows are leased to this worker), and stop() resolves only after every one of those deliveries has recorded its++// attempt, because the database pool is closed right after. No new pass claims anything once stopping. On 2026-11-02++// both deploys logged "delivery … crashed: Cannot use a pool after calling end on the pool": the deliveries started by++// the pass outlived the pool, and the next worker sent them again.++const assert = require('assert');++const { boot, request, serviceToken, envelope, subscriber, suite, sleep } = require('./helpers');++const { createWorker } = require('../server/worker');++++const t = suite('worker-stop');++let h;++const silent = { log() {}, warn() {}, error() {} };++++t('boot', async () => { h = await boot(); });++++t('stop() waits for the pass that is claiming and for what it started; nothing is claimed after', async () => {++ const stub = await subscriber(() => 204);++ const sub = await request(h.base, 'POST', '/api/v1/subscriptions', { token: serviceToken('stopper', ['events.subscription.manage']), body: { topic_pattern: 'live.stopper.*', endpoint: stub.url } });++ assert.strictEqual(sub.status, 201, sub.text);++ for (let i = 0; i < 3; i++) {++ const r = await request(h.base, 'POST', '/api/v1/events', { token: serviceToken('live', ['events.event.publish']), body: envelope('live', { event_type: 'live.stopper.x' }) });++ assert.ok(r.status === 201 || r.status === 200, r.text);++ }++ // The store, with the claim held until the test lets it go, and every recorded attempt counted.++ let release;++ const gate = new Promise((r) => { release = r; });++ let claims = 0;++ let claimedRows = 0;++ let recorded = 0;++ let closed = false;++ const store = new Proxy(h.store, {++ get(target, key) {++ if (key === 'claimDeliveries') {++ return async (...args) => {++ claims += 1;++ if (claims === 1) await gate;++ const rows = await target.claimDeliveries(...args);++ claimedRows += rows.length;++ return rows;++ };++ }++ if (key === 'recordAttempt') {++ return async (...args) => {++ assert.ok(!closed, 'an attempt is recorded only while the pool is open');++ await sleep(5);++ const out = await target.recordAttempt(...args);++ recorded += 1;++ return out;++ };++ }++ const v = target[key];++ return typeof v === 'function' ? v.bind(target) : v;++ },++ });++ const fetchImpl = async () => { await sleep(10); return new Response(null, { status: 204 }); };++ const w = createWorker({ store, config: { ...h.config, worker: { ...(h.config.worker || {}), intervalMs: 5 } }, clock: h.clock, fetchImpl, log: silent });++ w.start();++ for (let i = 0; i < 100 && claims === 0; i++) await sleep(5);++ assert.strictEqual(claims, 1, 'the first pass is claiming');++ const stopping = w.stop();++ release();++ await stopping;++ closed = true; // what index.js does next: close the pool++ assert.ok(claimedRows >= 1, 'the pass claimed the due rows');++ assert.strictEqual(recorded, claimedRows, 'every delivery the pass started recorded its attempt before stop() returned');++ const before = claims;++ await sleep(60);++ assert.strictEqual(claims, before, 'no pass claims anything after stop()');++ await stub.close();++});++++t('close', async () => { await h.close(); });++++t.run();
@@ -138,6 +138,7 @@ function createWorker({ store, config, clock = { now: () => Date.now() }, fetchI
@@ -138,6 +138,7 @@ function createWorker({ store, config, clock = { now: () => Date.now() }, fetchI
let finished = 0; // deliveries settled so far (drain)
let finished = 0; // deliveries settled so far (drain)
let passing = null;
let passing = null;
let queued = null;
let queued = null;
+ let stopping = false; // stop() was called: no new pass claims anything
function dispatch() {
function dispatch() {
if (!passing) {
if (!passing) {
passing = dispatchPass().finally(() => { passing = null; });
passing = dispatchPass().finally(() => { passing = null; });
@@ -150,6 +151,7 @@ function createWorker({ store, config, clock = { now: () => Date.now() }, fetchI
@@ -150,6 +151,7 @@ function createWorker({ store, config, clock = { now: () => Date.now() }, fetchI
/** Start every due delivery there is room for. Returns how many were started. */
/** Start every due delivery there is room for. Returns how many were started. */
async function dispatchPass() {
async function dispatchPass() {
lastTickAt = Date.now();
lastTickAt = Date.now();
+ if (stopping) return 0;
const room = opts.maxInflight - inflight.size;
const room = opts.maxInflight - inflight.size;
if (room <= 0) return 0;
if (room <= 0) return 0;
const now = clock.now();
const now = clock.now();
@@ -198,8 +200,17 @@ function createWorker({ store, config, clock = { now: () => Date.now() }, fetchI
@@ -198,8 +200,17 @@ function createWorker({ store, config, clock = { now: () => Date.now() }, fetchI
}
}
return {
return {
- start() { if (!running) { running = true; schedule(0); } },
- async stop() { running = false; clearTimeout(timer); timer = null; await Promise.allSettled([...inflight]); },
+ start() { if (!running) { running = true; stopping = false; schedule(0); } },
+ // A pass that is reading the due rows when the service stops still starts what it claimed (those rows are
+ // leased to this worker), so stop() waits for the pass and then for every delivery it started: the database
+ // pool is closed after this returns, and a delivery recording its attempt on a closed pool crashed and was
+ // sent again by the next worker (2026-11-02, both deploys of the day). No new pass claims anything once
+ // stopping; a later start() lifts that.
+ async stop() {
+ running = false; stopping = true; clearTimeout(timer); timer = null;
+ for (let i = 0; i < 3 && (passing || queued); i++) await Promise.allSettled([queued, passing].filter(Boolean));
+ while (inflight.size) await Promise.allSettled([...inflight]);
+ },
kick() { schedule(0); },
kick() { schedule(0); },
// The carrier path (ADR-042 decision 5): claim one row (store.claimDelivery) and send it through the same send().
// The carrier path (ADR-042 decision 5): claim one row (store.claimDelivery) and send it through the same send().
deliver(delivery) { return send(delivery); },
deliver(delivery) { return send(delivery); },
new file mode 100644
new file mode 100644
@@ -0,0 +1,75 @@
@@ -0,0 +1,75 @@
+'use strict';
+// The worker stops cleanly: a pass that is reading the due rows when the service stops still starts what it claimed
+// (those rows are leased to this worker), and stop() resolves only after every one of those deliveries has recorded its
+// attempt, because the database pool is closed right after. No new pass claims anything once stopping. On 2026-11-02
+// both deploys logged "delivery … crashed: Cannot use a pool after calling end on the pool": the deliveries started by
+// the pass outlived the pool, and the next worker sent them again.
+const assert = require('assert');
+const { boot, request, serviceToken, envelope, subscriber, suite, sleep } = require('./helpers');
+const { createWorker } = require('../server/worker');
+
+const t = suite('worker-stop');
+let h;
+const silent = { log() {}, warn() {}, error() {} };
+
+t('boot', async () => { h = await boot(); });
+
+t('stop() waits for the pass that is claiming and for what it started; nothing is claimed after', async () => {
+ const stub = await subscriber(() => 204);
+ const sub = await request(h.base, 'POST', '/api/v1/subscriptions', { token: serviceToken('stopper', ['events.subscription.manage']), body: { topic_pattern: 'live.stopper.*', endpoint: stub.url } });
+ assert.strictEqual(sub.status, 201, sub.text);
+ for (let i = 0; i < 3; i++) {
+ const r = await request(h.base, 'POST', '/api/v1/events', { token: serviceToken('live', ['events.event.publish']), body: envelope('live', { event_type: 'live.stopper.x' }) });
+ assert.ok(r.status === 201 || r.status === 200, r.text);
+ }
+ // The store, with the claim held until the test lets it go, and every recorded attempt counted.
+ let release;
+ const gate = new Promise((r) => { release = r; });
+ let claims = 0;
+ let claimedRows = 0;
+ let recorded = 0;
+ let closed = false;
+ const store = new Proxy(h.store, {
+ get(target, key) {
+ if (key === 'claimDeliveries') {
+ return async (...args) => {
+ claims += 1;
+ if (claims === 1) await gate;
+ const rows = await target.claimDeliveries(...args);
+ claimedRows += rows.length;
+ return rows;
+ };
+ }
+ if (key === 'recordAttempt') {
+ return async (...args) => {
+ assert.ok(!closed, 'an attempt is recorded only while the pool is open');
+ await sleep(5);
+ const out = await target.recordAttempt(...args);
+ recorded += 1;
+ return out;
+ };
+ }
+ const v = target[key];
+ return typeof v === 'function' ? v.bind(target) : v;
+ },
+ });
+ const fetchImpl = async () => { await sleep(10); return new Response(null, { status: 204 }); };
+ const w = createWorker({ store, config: { ...h.config, worker: { ...(h.config.worker || {}), intervalMs: 5 } }, clock: h.clock, fetchImpl, log: silent });
+ w.start();
+ for (let i = 0; i < 100 && claims === 0; i++) await sleep(5);
+ assert.strictEqual(claims, 1, 'the first pass is claiming');
+ const stopping = w.stop();
+ release();
+ await stopping;
+ closed = true; // what index.js does next: close the pool
+ assert.ok(claimedRows >= 1, 'the pass claimed the due rows');
+ assert.strictEqual(recorded, claimedRows, 'every delivery the pass started recorded its attempt before stop() returned');
+ const before = claims;
+ await sleep(60);
+ assert.strictEqual(claims, before, 'no pass claims anything after stop()');
+ await stub.close();
+});
+
+t('close', async () => { await h.close(); });
+
+t.run();