openvibers/openvibe.reviews Commits
Reviews: the event import queue re-indexes each entity once per drain, like a pulled page (12 queued reviews of one game sent 12 index documents in one trace; Search's follow-ups were refused as a loop) (#74)
Changes
@@@ -111,18 +111,23 @@ function createSourcesSync({ db, svc, sources, config, now = () => Date.now(), lif (!sources.enabled) return { ok: false, reason: 'OpenVibe.Sources is not configured', done: 0 };let done = 0;let failed = 0;-- for (const row of await q.due.all(now(), limit)) {-- try {-- await svc.importItem(row.item_id, sources, system);-- await q.done.run(row.item_id);-- done++;-- } catch (err) {-- // A 404/422 is final (the item is gone or unusable); anything else is retried.-- if (err && (err.status === 404 || err.status === 422)) { await q.done.run(row.item_id); log.warn(`[Reviews] queued item ${row.item_id} dropped: ${err.message}`); continue; }-- await q.failed.run(now() + BACKOFF_MS[Math.min(row.attempts, BACKOFF_MS.length - 1)], String(err && err.message).slice(0, 500), row.item_id);-- failed++;++ // The queued items in one svc.batch(), like a pulled page: 12 events for reviews of one game re-index it once.++ // Item by item, its index documents shared one trace and Search's follow-ups were refused as a loop (2026-10-17).++ // An item that fails rolls back alone (applyItem's transaction is a savepoint inside the batch).++ await svc.batch(async () => {++ for (const row of await q.due.all(now(), limit)) {++ try {++ await svc.importItem(row.item_id, sources, system);++ await q.done.run(row.item_id);++ done++;++ } catch (err) {++ // A 404/422 is final (the item is gone or unusable); anything else is retried.++ if (err && (err.status === 404 || err.status === 422)) { await q.done.run(row.item_id); log.warn(`[Reviews] queued item ${row.item_id} dropped: ${err.message}`); continue; }++ await q.failed.run(now() + BACKOFF_MS[Math.min(row.attempts, BACKOFF_MS.length - 1)], String(err && err.message).slice(0, 500), row.item_id);++ failed++;++ }}-- }++ });return { ok: true, done, failed };}@@@ -204,6 +204,20 @@ t('one page of many reviews of a game re-indexes the game once (the Search follo} finally { await h.stop(); }});++t('a queue of twelve reviews of one game (events that arrived while a drain ran) re-indexes it once', async () => {++ const h = await boot();++ try {++ const e = await createEntity(h, { name: 'Portal 2', kind: 'game', aliases: [{ type: 'source', value: 'steam-reviews-portal-2' }] });++ const upserts = async () => (await outbox(h)).filter((x) => x.event_type === 'reviews.index_document.upserted' && x.subject.id.endsWith(e.id)).length;++ const before = await upserts();++ for (let i = 0; i < 12; i++) await h.sync.enqueue(h.sources.put(steamItem({ votedUp: i % 3 !== 0 })).id, 'sources.item.created');++ const drained = await h.sync.drain();++ assert.deepStrictEqual([drained.done, drained.failed], [12, 0]);++ assert.strictEqual(await upserts() - before, 1, 'one index document for the drained queue, not one per review');++ assert.strictEqual((await aggregateOf(h, e.slug)).aggregate.result.components.recommendation.percent, 66.7, 'all 12 reviews counted');++ } finally { await h.stop(); }++});++t('a Sources outage changes nothing and invents nothing', async () => {const h = await boot();try {
@@ -111,18 +111,23 @@ function createSourcesSync({ db, svc, sources, config, now = () => Date.now(), l
@@ -111,18 +111,23 @@ function createSourcesSync({ db, svc, sources, config, now = () => Date.now(), l
if (!sources.enabled) return { ok: false, reason: 'OpenVibe.Sources is not configured', done: 0 };
if (!sources.enabled) return { ok: false, reason: 'OpenVibe.Sources is not configured', done: 0 };
let done = 0;
let done = 0;
let failed = 0;
let failed = 0;
- for (const row of await q.due.all(now(), limit)) {
- try {
- await svc.importItem(row.item_id, sources, system);
- await q.done.run(row.item_id);
- done++;
- } catch (err) {
- // A 404/422 is final (the item is gone or unusable); anything else is retried.
- if (err && (err.status === 404 || err.status === 422)) { await q.done.run(row.item_id); log.warn(`[Reviews] queued item ${row.item_id} dropped: ${err.message}`); continue; }
- await q.failed.run(now() + BACKOFF_MS[Math.min(row.attempts, BACKOFF_MS.length - 1)], String(err && err.message).slice(0, 500), row.item_id);
- failed++;
+ // The queued items in one svc.batch(), like a pulled page: 12 events for reviews of one game re-index it once.
+ // Item by item, its index documents shared one trace and Search's follow-ups were refused as a loop (2026-10-17).
+ // An item that fails rolls back alone (applyItem's transaction is a savepoint inside the batch).
+ await svc.batch(async () => {
+ for (const row of await q.due.all(now(), limit)) {
+ try {
+ await svc.importItem(row.item_id, sources, system);
+ await q.done.run(row.item_id);
+ done++;
+ } catch (err) {
+ // A 404/422 is final (the item is gone or unusable); anything else is retried.
+ if (err && (err.status === 404 || err.status === 422)) { await q.done.run(row.item_id); log.warn(`[Reviews] queued item ${row.item_id} dropped: ${err.message}`); continue; }
+ await q.failed.run(now() + BACKOFF_MS[Math.min(row.attempts, BACKOFF_MS.length - 1)], String(err && err.message).slice(0, 500), row.item_id);
+ failed++;
+ }
}
}
- }
+ });
return { ok: true, done, failed };
return { ok: true, done, failed };
}
}
@@ -204,6 +204,20 @@ t('one page of many reviews of a game re-indexes the game once (the Search follo
@@ -204,6 +204,20 @@ t('one page of many reviews of a game re-indexes the game once (the Search follo
} finally { await h.stop(); }
} finally { await h.stop(); }
});
});
+t('a queue of twelve reviews of one game (events that arrived while a drain ran) re-indexes it once', async () => {
+ const h = await boot();
+ try {
+ const e = await createEntity(h, { name: 'Portal 2', kind: 'game', aliases: [{ type: 'source', value: 'steam-reviews-portal-2' }] });
+ const upserts = async () => (await outbox(h)).filter((x) => x.event_type === 'reviews.index_document.upserted' && x.subject.id.endsWith(e.id)).length;
+ const before = await upserts();
+ for (let i = 0; i < 12; i++) await h.sync.enqueue(h.sources.put(steamItem({ votedUp: i % 3 !== 0 })).id, 'sources.item.created');
+ const drained = await h.sync.drain();
+ assert.deepStrictEqual([drained.done, drained.failed], [12, 0]);
+ assert.strictEqual(await upserts() - before, 1, 'one index document for the drained queue, not one per review');
+ assert.strictEqual((await aggregateOf(h, e.slug)).aggregate.result.components.recommendation.percent, 66.7, 'all 12 reviews counted');
+ } finally { await h.stop(); }
+});
+
t('a Sources outage changes nothing and invents nothing', async () => {
t('a Sources outage changes nothing and invents nothing', async () => {
const h = await boot();
const h = await boot();
try {
try {