openvibers/openvibe.deals Commits
Ticks re-index what time changed; everything at boot and once a day (#206)
Changes
@@@ -39,6 +39,11 @@ function createListings({ store }) {bySeller: db.prepare(`SELECT * FROM deal_offers WHERE seller = ? AND merged_into IS NULL AND status <> 'disabled' ORDER BY status = 'active' DESC, created_at DESC LIMIT ? OFFSET ?`),countSeller: db.prepare(`SELECT COUNT(*) AS n FROM deal_offers WHERE seller = ? AND merged_into IS NULL AND status <> 'disabled'`),indexable: db.prepare(`SELECT * FROM deal_offers WHERE merged_into IS NULL AND status <> 'disabled' ORDER BY created_at DESC LIMIT 50000`),++ // The canonical offers with an observation that came of age (freshnessMs) in a window: their freshness, and so++ // their index document, may have changed with time alone. A superset is fine: an unchanged document sends nothing.++ agedInto: db.prepare(`SELECT r.* FROM deal_offers r WHERE r.merged_into IS NULL AND r.status <> 'disabled' AND r.id IN (++ SELECT COALESCE(o.merged_into, o.id) FROM deal_price_observations p JOIN deal_offers o ON o.id = p.offer_id++ WHERE p.observed_at > ? AND p.observed_at <= ?) LIMIT 50000`),products: db.prepare(`SELECT DISTINCT p.* FROM deal_products p JOIN deal_offers o ON o.product_id = p.id WHERE o.merged_into IS NULL AND o.status <> 'disabled' LIMIT 50000`),duplicates: db.prepare(`SELECT a.id AS a, b.id AS b FROM deal_offers a JOIN deal_offers bON b.store_id = a.store_id AND b.id > a.id@@@ -90,6 +95,8 @@ function createListings({ store }) {bySeller: async (seller, limit, offset) => await q.bySeller.all(seller, limit, offset),countSeller: async (seller) => (await q.countSeller.get(seller)).n,indexable: async () => await q.indexable.all(),++ /** Offers whose freshness may have turned between `since` and `now` (an observation crossed freshnessMs). */++ agedInto: async (since, now, freshnessMs) => await q.agedInto.all(since - freshnessMs, now - freshnessMs),products: async () => await q.products.all(),possibleDuplicates: async (limit = 50) => await q.duplicates.all(limit),pendingReview: async (limit = 50) => await q.pendingReview.all(limit),@@@ -22,6 +22,15 @@ function createWorker({ config, store, offers, hotness, indexing, listings, impo// the pool" on each deploy, 2026-11-02: the re-index loop runs offer by offer for minutes).let stopping = false;let current = null;++ // The re-index (plan T9's freshness): every write re-indexes its own offer (offers.js), and the document changes++ // with time alone only when the latest price comes of age (freshnessMs, 48 h). So a tick re-indexes the offers++ // whose observation crossed that line since the last pass, and the whole set only at boot and once a day (a++ // safety net, and how a change to the document's shape reaches every offer). Rebuilding all ~40,000 documents++ // every ten minutes cost about half a core and 26 reads a second (2026-11-02).++ const FULL_EVERY_MS = 24 * 3_600_000;++ const MARGIN_MS = 60_000;++ let lastFullAt = null;++ let lastPassAt = 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@@@ -58,15 +67,29 @@ function createWorker({ config, store, offers, hotness, indexing, listings, impoif (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()) {-- if (stopping) return summary;-- await store.tx(async () => { if (await indexing.indexProduct(p)) reindexed++; });++ const passAt = store.now();++ const full = lastFullAt == null || lastPassAt == null || passAt - lastFullAt >= FULL_EVERY_MS || typeof listings.agedInto !== 'function';++ if (full) {++ 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()) {++ if (stopping) return summary;++ await store.tx(async () => { if (await indexing.indexProduct(p)) reindexed++; });++ }++ lastFullAt = passAt;++ } else {++ // The offers that came of age since the last pass, and their products (reindex does both).++ const aged = await listings.agedInto(lastPassAt - MARGIN_MS, passAt, config.freshnessMs || 48 * 3_600_000);++ for (const o of aged) {++ if (stopping) return summary;++ await store.tx(async () => { await indexing.reindex(o); reindexed++; });++ }}++ lastPassAt = passAt;summary.reindexed = reindexed;++ summary.reindex = full ? 'full' : 'aged';if (stopping) return summary;summary.import = await importer.pull();if (stopping) return summary;nnew file mode 100644@@@ -0,0 +1,58 @@++'use strict';++/**++ * What a tick re-indexes: every offer and product at the first tick and once a day; in between, only the offers whose++ * price came of age (an observation crossed freshnessMs since the last pass), through reindex (offer and product).++ * Rebuilding all ~40,000 documents every ten minutes cost about half a core (2026-11-02); a write re-indexes its own++ * offer, so time alone is what the ticks are for.++ *++ * node test/worker-reindex.test.js++ */++const assert = require('assert');++const { createWorker } = require('../server/worker');++++(async () => {++ const HOUR = 3_600_000;++ let t = 1_000_000_000_000;++ const calls = { indexable: 0, products: 0, indexOffer: 0, indexProduct: 0, reindex: 0, aged: [] };++ const worker = createWorker({++ config: { freshnessMs: 48 * HOUR, worker: { enabled: false, intervalMs: 600_000, hotWindowDays: 14 } },++ store: { now: () => t, tx: async (fn) => fn() },++ offers: { expireDue: async () => 0 },++ hotness: { tick: async () => 0 },++ indexing: {++ indexOffer: async () => { calls.indexOffer += 1; return false; },++ indexProduct: async () => { calls.indexProduct += 1; return false; },++ reindex: async () => { calls.reindex += 1; },++ },++ listings: {++ indexable: async () => { calls.indexable += 1; return [{ id: 'a' }, { id: 'b' }, { id: 'c' }]; },++ products: async () => { calls.products += 1; return [{ id: 'p' }]; },++ agedInto: async (since, now, freshnessMs) => { calls.aged.push({ since, now, freshnessMs }); return [{ id: 'b' }]; },++ },++ importer: { pull: async () => ({}), refresh: async () => ({}) },++ limits: { prune: async () => 0 },++ log: { error() {}, log() {}, warn() {} },++ });++++ let s = await worker.tick();++ assert.strictEqual(s.reindex, 'full', 'the first tick re-indexes everything');++ assert.deepStrictEqual([calls.indexable, calls.indexOffer, calls.indexProduct, calls.aged.length], [1, 3, 1, 0]);++++ const first = t;++ t += 10 * 60_000;++ s = await worker.tick();++ assert.strictEqual(s.reindex, 'aged', 'ten minutes later: only what came of age');++ assert.strictEqual(calls.indexable, 1, 'not the whole set');++ assert.deepStrictEqual(calls.aged[0], { since: first - 60_000, now: t, freshnessMs: 48 * HOUR }, 'since the last pass, with a minute of margin');++ assert.strictEqual(calls.reindex, 1, 'each aged offer through reindex (its product too)');++++ t += 10 * 60_000;++ await worker.tick();++ assert.strictEqual(calls.aged[1].since, t - 10 * 60_000 - 60_000, 'the window follows the passes');++++ t = first + 24 * HOUR;++ s = await worker.tick();++ assert.strictEqual(s.reindex, 'full', 'once a day, everything again');++ assert.strictEqual(calls.indexable, 2);++ console.log('worker re-index: full at the first tick and daily, otherwise only offers whose price came of age');++})().catch((err) => { console.error(err); process.exit(1); });
@@ -39,6 +39,11 @@ function createListings({ store }) {
@@ -39,6 +39,11 @@ function createListings({ store }) {
bySeller: db.prepare(`SELECT * FROM deal_offers WHERE seller = ? AND merged_into IS NULL AND status <> 'disabled' ORDER BY status = 'active' DESC, created_at DESC LIMIT ? OFFSET ?`),
bySeller: db.prepare(`SELECT * FROM deal_offers WHERE seller = ? AND merged_into IS NULL AND status <> 'disabled' ORDER BY status = 'active' DESC, created_at DESC LIMIT ? OFFSET ?`),
countSeller: db.prepare(`SELECT COUNT(*) AS n FROM deal_offers WHERE seller = ? AND merged_into IS NULL AND status <> 'disabled'`),
countSeller: db.prepare(`SELECT COUNT(*) AS n FROM deal_offers WHERE seller = ? AND merged_into IS NULL AND status <> 'disabled'`),
indexable: db.prepare(`SELECT * FROM deal_offers WHERE merged_into IS NULL AND status <> 'disabled' ORDER BY created_at DESC LIMIT 50000`),
indexable: db.prepare(`SELECT * FROM deal_offers WHERE merged_into IS NULL AND status <> 'disabled' ORDER BY created_at DESC LIMIT 50000`),
+ // The canonical offers with an observation that came of age (freshnessMs) in a window: their freshness, and so
+ // their index document, may have changed with time alone. A superset is fine: an unchanged document sends nothing.
+ agedInto: db.prepare(`SELECT r.* FROM deal_offers r WHERE r.merged_into IS NULL AND r.status <> 'disabled' AND r.id IN (
+ SELECT COALESCE(o.merged_into, o.id) FROM deal_price_observations p JOIN deal_offers o ON o.id = p.offer_id
+ WHERE p.observed_at > ? AND p.observed_at <= ?) LIMIT 50000`),
products: db.prepare(`SELECT DISTINCT p.* FROM deal_products p JOIN deal_offers o ON o.product_id = p.id WHERE o.merged_into IS NULL AND o.status <> 'disabled' LIMIT 50000`),
products: db.prepare(`SELECT DISTINCT p.* FROM deal_products p JOIN deal_offers o ON o.product_id = p.id WHERE o.merged_into IS NULL AND o.status <> 'disabled' LIMIT 50000`),
duplicates: db.prepare(`SELECT a.id AS a, b.id AS b FROM deal_offers a JOIN deal_offers b
duplicates: db.prepare(`SELECT a.id AS a, b.id AS b FROM deal_offers a JOIN deal_offers b
ON b.store_id = a.store_id AND b.id > a.id
ON b.store_id = a.store_id AND b.id > a.id
@@ -90,6 +95,8 @@ function createListings({ store }) {
@@ -90,6 +95,8 @@ function createListings({ store }) {
bySeller: async (seller, limit, offset) => await q.bySeller.all(seller, limit, offset),
bySeller: async (seller, limit, offset) => await q.bySeller.all(seller, limit, offset),
countSeller: async (seller) => (await q.countSeller.get(seller)).n,
countSeller: async (seller) => (await q.countSeller.get(seller)).n,
indexable: async () => await q.indexable.all(),
indexable: async () => await q.indexable.all(),
+ /** Offers whose freshness may have turned between `since` and `now` (an observation crossed freshnessMs). */
+ agedInto: async (since, now, freshnessMs) => await q.agedInto.all(since - freshnessMs, now - freshnessMs),
products: async () => await q.products.all(),
products: async () => await q.products.all(),
possibleDuplicates: async (limit = 50) => await q.duplicates.all(limit),
possibleDuplicates: async (limit = 50) => await q.duplicates.all(limit),
pendingReview: async (limit = 50) => await q.pendingReview.all(limit),
pendingReview: async (limit = 50) => await q.pendingReview.all(limit),
@@ -22,6 +22,15 @@ function createWorker({ config, store, offers, hotness, indexing, listings, impo
@@ -22,6 +22,15 @@ function createWorker({ config, store, offers, hotness, indexing, listings, impo
// the pool" on each deploy, 2026-11-02: the re-index loop runs offer by offer for minutes).
// the pool" on each deploy, 2026-11-02: the re-index loop runs offer by offer for minutes).
let stopping = false;
let stopping = false;
let current = null;
let current = null;
+ // The re-index (plan T9's freshness): every write re-indexes its own offer (offers.js), and the document changes
+ // with time alone only when the latest price comes of age (freshnessMs, 48 h). So a tick re-indexes the offers
+ // whose observation crossed that line since the last pass, and the whole set only at boot and once a day (a
+ // safety net, and how a change to the document's shape reaches every offer). Rebuilding all ~40,000 documents
+ // every ten minutes cost about half a core and 26 reads a second (2026-11-02).
+ const FULL_EVERY_MS = 24 * 3_600_000;
+ const MARGIN_MS = 60_000;
+ let lastFullAt = null;
+ let lastPassAt = 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
@@ -58,15 +67,29 @@ function createWorker({ config, store, offers, hotness, indexing, listings, impo
@@ -58,15 +67,29 @@ function createWorker({ config, store, offers, hotness, indexing, listings, impo
if (stopping) return summary;
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()) {
- if (stopping) return summary;
- await store.tx(async () => { if (await indexing.indexOffer(o)) reindexed++; });
- }
- for (const p of await listings.products()) {
- if (stopping) return summary;
- await store.tx(async () => { if (await indexing.indexProduct(p)) reindexed++; });
+ const passAt = store.now();
+ const full = lastFullAt == null || lastPassAt == null || passAt - lastFullAt >= FULL_EVERY_MS || typeof listings.agedInto !== 'function';
+ if (full) {
+ 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()) {
+ if (stopping) return summary;
+ await store.tx(async () => { if (await indexing.indexProduct(p)) reindexed++; });
+ }
+ lastFullAt = passAt;
+ } else {
+ // The offers that came of age since the last pass, and their products (reindex does both).
+ const aged = await listings.agedInto(lastPassAt - MARGIN_MS, passAt, config.freshnessMs || 48 * 3_600_000);
+ for (const o of aged) {
+ if (stopping) return summary;
+ await store.tx(async () => { await indexing.reindex(o); reindexed++; });
+ }
}
}
+ lastPassAt = passAt;
summary.reindexed = reindexed;
summary.reindexed = reindexed;
+ summary.reindex = full ? 'full' : 'aged';
if (stopping) return summary;
if (stopping) return summary;
summary.import = await importer.pull();
summary.import = await importer.pull();
if (stopping) return summary;
if (stopping) return summary;
new file mode 100644
new file mode 100644
@@ -0,0 +1,58 @@
@@ -0,0 +1,58 @@
+'use strict';
+/**
+ * What a tick re-indexes: every offer and product at the first tick and once a day; in between, only the offers whose
+ * price came of age (an observation crossed freshnessMs since the last pass), through reindex (offer and product).
+ * Rebuilding all ~40,000 documents every ten minutes cost about half a core (2026-11-02); a write re-indexes its own
+ * offer, so time alone is what the ticks are for.
+ *
+ * node test/worker-reindex.test.js
+ */
+const assert = require('assert');
+const { createWorker } = require('../server/worker');
+
+(async () => {
+ const HOUR = 3_600_000;
+ let t = 1_000_000_000_000;
+ const calls = { indexable: 0, products: 0, indexOffer: 0, indexProduct: 0, reindex: 0, aged: [] };
+ const worker = createWorker({
+ config: { freshnessMs: 48 * HOUR, worker: { enabled: false, intervalMs: 600_000, hotWindowDays: 14 } },
+ store: { now: () => t, tx: async (fn) => fn() },
+ offers: { expireDue: async () => 0 },
+ hotness: { tick: async () => 0 },
+ indexing: {
+ indexOffer: async () => { calls.indexOffer += 1; return false; },
+ indexProduct: async () => { calls.indexProduct += 1; return false; },
+ reindex: async () => { calls.reindex += 1; },
+ },
+ listings: {
+ indexable: async () => { calls.indexable += 1; return [{ id: 'a' }, { id: 'b' }, { id: 'c' }]; },
+ products: async () => { calls.products += 1; return [{ id: 'p' }]; },
+ agedInto: async (since, now, freshnessMs) => { calls.aged.push({ since, now, freshnessMs }); return [{ id: 'b' }]; },
+ },
+ importer: { pull: async () => ({}), refresh: async () => ({}) },
+ limits: { prune: async () => 0 },
+ log: { error() {}, log() {}, warn() {} },
+ });
+
+ let s = await worker.tick();
+ assert.strictEqual(s.reindex, 'full', 'the first tick re-indexes everything');
+ assert.deepStrictEqual([calls.indexable, calls.indexOffer, calls.indexProduct, calls.aged.length], [1, 3, 1, 0]);
+
+ const first = t;
+ t += 10 * 60_000;
+ s = await worker.tick();
+ assert.strictEqual(s.reindex, 'aged', 'ten minutes later: only what came of age');
+ assert.strictEqual(calls.indexable, 1, 'not the whole set');
+ assert.deepStrictEqual(calls.aged[0], { since: first - 60_000, now: t, freshnessMs: 48 * HOUR }, 'since the last pass, with a minute of margin');
+ assert.strictEqual(calls.reindex, 1, 'each aged offer through reindex (its product too)');
+
+ t += 10 * 60_000;
+ await worker.tick();
+ assert.strictEqual(calls.aged[1].since, t - 10 * 60_000 - 60_000, 'the window follows the passes');
+
+ t = first + 24 * HOUR;
+ s = await worker.tick();
+ assert.strictEqual(s.reindex, 'full', 'once a day, everything again');
+ assert.strictEqual(calls.indexable, 2);
+ console.log('worker re-index: full at the first tick and daily, otherwise only offers whose price came of age');
+})().catch((err) => { console.error(err); process.exit(1); });