openvibers/openvibe.deals Commits
The worker stops after its running tick, between offers (#204)
Changes
@@@ -17,6 +17,11 @@function createWorker({ config, store, offers, hotness, indexing, listings, importer, campaigns = null, limits, log = console, timers = { setInterval, clearInterval, setTimeout } }) {let timer = null;let running = false;++ // stop(): no new tick starts, the one running ends at its next step, and stop() waits for it, so nothing touches++ // the database after the service closes its pool ("worker tick failed: Cannot use a pool after calling end on++ // the pool" on each deploy, 2026-11-02: the re-index loop runs offer by offer for minutes).++ let stopping = false;++ let current = null;let lastError = null;let lastTickAt = null;// Shared page builds to keep warm (http/pages.js): the front page and /hot's first page. Their@@@ -39,21 +44,35 @@ function createWorker({ config, store, offers, hotness, indexing, listings, impo}async function tick() {-- if (running) return null;++ if (running || stopping) return null;running = true;++ const run = runTick();++ current = run;++ try { return await run; } finally { current = null; }++ }++++ async function runTick() {const summary = {};try {summary.expired = await offers.expireDue();++ if (stopping) return summary;summary.snapshots = await hotness.tick({ windowDays: config.worker.hotWindowDays });let reindexed = 0;for (const o of await listings.indexable()) {++ if (stopping) return summary;await store.tx(async () => { if (await indexing.indexOffer(o)) reindexed++; });}-- for (const p of await listings.products()) await store.tx(async () => { if (await indexing.indexProduct(p)) reindexed++; });++ for (const p of await listings.products()) {++ if (stopping) return summary;++ await store.tx(async () => { if (await indexing.indexProduct(p)) reindexed++; });++ }summary.reindexed = reindexed;++ if (stopping) return summary;summary.import = await importer.pull();++ if (stopping) return summary;summary.refresh = await importer.refresh();-- if (campaigns) summary.campaigns = await campaigns.pull();++ if (campaigns && !stopping) summary.campaigns = await campaigns.pull();++ if (stopping) return summary;summary.pruned = await limits.prune();lastError = null;} catch (err) {@@@ -88,10 +107,13 @@ function createWorker({ config, store, offers, hotness, indexing, listings, impowarmTimer.unref();}},++ /** Stop: no new tick, the running one ends at its next step; resolves once it has. */stop() {++ stopping = true;timers.clearInterval(timer);timers.clearInterval(warmTimer);warmTimer = null;++ return current ? current.then(() => undefined, () => undefined) : Promise.resolve();},status: () => ({ enabled: config.worker.enabled, warming: warmTimer != null, last_error: lastError, last_tick_at: lastTickAt ? new Date(lastTickAt).toISOString() : null }),};nnew file mode 100644@@@ -0,0 +1,41 @@++'use strict';++/**++ * The worker stops cleanly: stop() ends the running tick at its next step (between offers in the re-index loop) and++ * resolves only once it has, and no tick starts after it. Before, stop() only cleared the timers, and a tick still++ * re-indexing went on after the service closed its database pool ("worker tick failed: Cannot use a pool after++ * calling end on the pool" on each deploy, 2026-11-02).++ *++ * node test/worker-stop.test.js++ */++const assert = require('assert');++const { createWorker } = require('../server/worker');++++(async () => {++ const OFFERS = Array.from({ length: 50 }, (_, i) => ({ id: `off_${i}` }));++ let indexed = 0;++ let closed = false;++ const later = () => new Promise((r) => setTimeout(r, 5));++ const store = { now: () => Date.now(), tx: async (fn) => { assert.ok(!closed, 'nothing touches the database after it closes'); return fn(); } };++ const worker = createWorker({++ config: { worker: { enabled: false, intervalMs: 60_000, hotWindowDays: 14 } },++ store,++ offers: { expireDue: async () => 0 },++ hotness: { tick: async () => 0 },++ indexing: { indexOffer: async () => { await later(); indexed += 1; return false; }, indexProduct: async () => false },++ listings: { indexable: async () => OFFERS, products: async () => [] },++ importer: { pull: async () => ({}), refresh: async () => ({}) },++ limits: { prune: async () => 0 },++ log: { error() {}, log() {}, warn() {} },++ });++ const ticking = worker.tick();++ while (indexed < 3) await later();++ const stopped = worker.stop();++ await stopped;++ closed = true; // what the service does next: close the pool++ const summary = await ticking;++ assert.ok(indexed < OFFERS.length, `the tick ended early (${indexed} of ${OFFERS.length})`);++ assert.ok(!('import' in summary), 'it stopped before the importer');++ assert.strictEqual(await worker.tick(), null, 'no tick starts after stop()');++ assert.strictEqual(await worker.stop(), undefined, 'stop() with nothing running resolves at once');++ console.log(`worker stop: the tick ended after ${indexed} of ${OFFERS.length} offers; nothing ran after the pool closed`);++})().catch((err) => { console.error(err); process.exit(1); });
@@ -17,6 +17,11 @@
@@ -17,6 +17,11 @@
function createWorker({ config, store, offers, hotness, indexing, listings, importer, campaigns = null, limits, log = console, timers = { setInterval, clearInterval, setTimeout } }) {
function createWorker({ config, store, offers, hotness, indexing, listings, importer, campaigns = null, limits, log = console, timers = { setInterval, clearInterval, setTimeout } }) {
let timer = null;
let timer = null;
let running = false;
let running = false;
+ // stop(): no new tick starts, the one running ends at its next step, and stop() waits for it, so nothing touches
+ // the database after the service closes its pool ("worker tick failed: Cannot use a pool after calling end on
+ // the pool" on each deploy, 2026-11-02: the re-index loop runs offer by offer for minutes).
+ let stopping = false;
+ let current = null;
let lastError = null;
let lastError = null;
let lastTickAt = null;
let lastTickAt = null;
// Shared page builds to keep warm (http/pages.js): the front page and /hot's first page. Their
// Shared page builds to keep warm (http/pages.js): the front page and /hot's first page. Their
@@ -39,21 +44,35 @@ function createWorker({ config, store, offers, hotness, indexing, listings, impo
@@ -39,21 +44,35 @@ function createWorker({ config, store, offers, hotness, indexing, listings, impo
}
}
async function tick() {
async function tick() {
- if (running) return null;
+ if (running || stopping) return null;
running = true;
running = true;
+ const run = runTick();
+ current = run;
+ try { return await run; } finally { current = null; }
+ }
+
+ async function runTick() {
const summary = {};
const summary = {};
try {
try {
summary.expired = await offers.expireDue();
summary.expired = await offers.expireDue();
+ if (stopping) return summary;
summary.snapshots = await hotness.tick({ windowDays: config.worker.hotWindowDays });
summary.snapshots = await hotness.tick({ windowDays: config.worker.hotWindowDays });
let reindexed = 0;
let reindexed = 0;
for (const o of await listings.indexable()) {
for (const o of await listings.indexable()) {
+ if (stopping) return summary;
await store.tx(async () => { if (await indexing.indexOffer(o)) reindexed++; });
await store.tx(async () => { if (await indexing.indexOffer(o)) reindexed++; });
}
}
- for (const p of await listings.products()) await store.tx(async () => { if (await indexing.indexProduct(p)) reindexed++; });
+ for (const p of await listings.products()) {
+ if (stopping) return summary;
+ await store.tx(async () => { if (await indexing.indexProduct(p)) reindexed++; });
+ }
summary.reindexed = reindexed;
summary.reindexed = reindexed;
+ if (stopping) return summary;
summary.import = await importer.pull();
summary.import = await importer.pull();
+ if (stopping) return summary;
summary.refresh = await importer.refresh();
summary.refresh = await importer.refresh();
- if (campaigns) summary.campaigns = await campaigns.pull();
+ if (campaigns && !stopping) summary.campaigns = await campaigns.pull();
+ if (stopping) return summary;
summary.pruned = await limits.prune();
summary.pruned = await limits.prune();
lastError = null;
lastError = null;
} catch (err) {
} catch (err) {
@@ -88,10 +107,13 @@ function createWorker({ config, store, offers, hotness, indexing, listings, impo
@@ -88,10 +107,13 @@ function createWorker({ config, store, offers, hotness, indexing, listings, impo
warmTimer.unref();
warmTimer.unref();
}
}
},
},
+ /** Stop: no new tick, the running one ends at its next step; resolves once it has. */
stop() {
stop() {
+ stopping = true;
timers.clearInterval(timer);
timers.clearInterval(timer);
timers.clearInterval(warmTimer);
timers.clearInterval(warmTimer);
warmTimer = null;
warmTimer = null;
+ return current ? current.then(() => undefined, () => undefined) : Promise.resolve();
},
},
status: () => ({ enabled: config.worker.enabled, warming: warmTimer != null, last_error: lastError, last_tick_at: lastTickAt ? new Date(lastTickAt).toISOString() : null }),
status: () => ({ enabled: config.worker.enabled, warming: warmTimer != null, last_error: lastError, last_tick_at: lastTickAt ? new Date(lastTickAt).toISOString() : null }),
};
};
new file mode 100644
new file mode 100644
@@ -0,0 +1,41 @@
@@ -0,0 +1,41 @@
+'use strict';
+/**
+ * The worker stops cleanly: stop() ends the running tick at its next step (between offers in the re-index loop) and
+ * resolves only once it has, and no tick starts after it. Before, stop() only cleared the timers, and a tick still
+ * re-indexing went on after the service closed its database pool ("worker tick failed: Cannot use a pool after
+ * calling end on the pool" on each deploy, 2026-11-02).
+ *
+ * node test/worker-stop.test.js
+ */
+const assert = require('assert');
+const { createWorker } = require('../server/worker');
+
+(async () => {
+ const OFFERS = Array.from({ length: 50 }, (_, i) => ({ id: `off_${i}` }));
+ let indexed = 0;
+ let closed = false;
+ const later = () => new Promise((r) => setTimeout(r, 5));
+ const store = { now: () => Date.now(), tx: async (fn) => { assert.ok(!closed, 'nothing touches the database after it closes'); return fn(); } };
+ const worker = createWorker({
+ config: { worker: { enabled: false, intervalMs: 60_000, hotWindowDays: 14 } },
+ store,
+ offers: { expireDue: async () => 0 },
+ hotness: { tick: async () => 0 },
+ indexing: { indexOffer: async () => { await later(); indexed += 1; return false; }, indexProduct: async () => false },
+ listings: { indexable: async () => OFFERS, products: async () => [] },
+ importer: { pull: async () => ({}), refresh: async () => ({}) },
+ limits: { prune: async () => 0 },
+ log: { error() {}, log() {}, warn() {} },
+ });
+ const ticking = worker.tick();
+ while (indexed < 3) await later();
+ const stopped = worker.stop();
+ await stopped;
+ closed = true; // what the service does next: close the pool
+ const summary = await ticking;
+ assert.ok(indexed < OFFERS.length, `the tick ended early (${indexed} of ${OFFERS.length})`);
+ assert.ok(!('import' in summary), 'it stopped before the importer');
+ assert.strictEqual(await worker.tick(), null, 'no tick starts after stop()');
+ assert.strictEqual(await worker.stop(), undefined, 'stop() with nothing running resolves at once');
+ console.log(`worker stop: the tick ended after ${indexed} of ${OFFERS.length} offers; nothing ran after the pool closed`);
+})().catch((err) => { console.error(err); process.exit(1); });