openvibers/openvibe.search Commits
Await what the transactions start (PostgreSQL pass follow-up): the outbox and CDN-purge queues' mark() ran each row's async write inside a sync db.tx callback without awaiting it. A sync callback that calls async code returns before the work is done, so writes could land outside or after their transaction. npm test 12/12.
Changes
@@@ -109,7 +109,7 @@ function createRelay({ outbox, eventsUrl, intervalMs = 2000, tokenClient, tokenOreturn { ok: res.ok, status: res.status, text };}-- async function mark(rows, fn) { await db.tx(() => { for (const r of rows) fn(r); }); }++ async function mark(rows, fn) { await db.tx(async () => { for (const r of rows) await fn(r); }); }async function publishRows(rows) {let r;@@@ -161,7 +161,7 @@ function createPurger({ db, config, fetchImpl = globalThis.fetch, log = console,const clip = (s) => (token ? String(s || '').split(token).join('[token]') : String(s || '')).slice(0, DETAIL_MAX);-- async function mark(rows, fn) { await db.tx(() => { for (const r of rows) fn(r); }); }++ async function mark(rows, fn) { await db.tx(async () => { for (const r of rows) await fn(r); }); }async function retryOrFail(rows, detail) {await mark(rows, async (r) => {
@@ -109,7 +109,7 @@ function createRelay({ outbox, eventsUrl, intervalMs = 2000, tokenClient, tokenO
@@ -109,7 +109,7 @@ function createRelay({ outbox, eventsUrl, intervalMs = 2000, tokenClient, tokenO
return { ok: res.ok, status: res.status, text };
return { ok: res.ok, status: res.status, text };
}
}
- async function mark(rows, fn) { await db.tx(() => { for (const r of rows) fn(r); }); }
+ async function mark(rows, fn) { await db.tx(async () => { for (const r of rows) await fn(r); }); }
async function publishRows(rows) {
async function publishRows(rows) {
let r;
let r;
@@ -161,7 +161,7 @@ function createPurger({ db, config, fetchImpl = globalThis.fetch, log = console,
@@ -161,7 +161,7 @@ function createPurger({ db, config, fetchImpl = globalThis.fetch, log = console,
const clip = (s) => (token ? String(s || '').split(token).join('[token]') : String(s || '')).slice(0, DETAIL_MAX);
const clip = (s) => (token ? String(s || '').split(token).join('[token]') : String(s || '')).slice(0, DETAIL_MAX);
- async function mark(rows, fn) { await db.tx(() => { for (const r of rows) fn(r); }); }
+ async function mark(rows, fn) { await db.tx(async () => { for (const r of rows) await fn(r); }); }
async function retryOrFail(rows, detail) {
async function retryOrFail(rows, detail) {
await mark(rows, async (r) => {
await mark(rows, async (r) => {