openvibers/openvibe.sdk Commits
v0.21.2: an outbox flush() during a relay pass gets the next pass; test containers keep PostgreSQL within its connections.
Changes
@@@ -8,6 +8,11 @@ release may change an API and says so here.++## 0.21.2 (2026-09-28)++++- `openvibe-sdk/events` outboxes (SQLite and PostgreSQL): a `flush()` called while a relay pass runs now gets the next pass, which starts when that one ends. It used to share the running pass, so it could return before the rows committed after that pass claimed were published (a caller's "flush, then assert both went out" saw one). Calls made during a pass share that next pass; `stop()` waits for both.++- `scripts/test-services.sh`: PgBouncer closes idle server connections after 5 s and takes at most 120 per database, and PostgreSQL allows 200. Each test process has a role of its own, so the pools of a suite with many files ran PostgreSQL out of connections.++## 0.21.1 (2026-09-28)- `openvibe-sdk/events` `createPgOutbox`: the relay publishes the rows it claimed in the order they were written. `UPDATE … RETURNING` gives no order, so a batch could reach Events out of order (a `media.clip.failed` before the `media.vod.ready` written before it). Found converting OpenVibe.Media.@@@ -3,7 +3,7 @@"stage": "alpha","code": true,"wave": "W2",-- "version": "0.21.1",++ "version": "0.21.2","plan": "OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026), §3.2; roadmap §30","deployed": false,"deployment": "none: a package released as git tags; services and apps pin a tag",@@@ -1,4 +1,4 @@--// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.21.1, MIT). Do not edit.++// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.21.2, MIT). Do not edit.// One self-contained ES module: core at the top level plus the browser-safe subpaths as namespaces// (auth, registry, modules, realtime, media, community, search, chat, ai, jobs, tools, projects, geo). No server-only code: no service tokens, JWT verification, events, identity./* eslint-disable */@@@ -41,7 +41,7 @@ const { paginate, offsetPager } = __load(7);const semver = __load(6);/** The package version (test/version.test.js keeps it equal to package.json). */--const SDK_VERSION = '0.21.1';++const SDK_VERSION = '0.21.2';module.exports = {createClient,@@@ -1,12 +1,12 @@{"name": "openvibe-sdk",-- "version": "0.21.1",++ "version": "0.21.2","lockfileVersion": 3,"requires": true,"packages": {"": {"name": "openvibe-sdk",-- "version": "0.21.1",++ "version": "0.21.2","license": "MIT","devDependencies": {"@electric-sql/pglite": "^0.5.8",@@@ -1,6 +1,6 @@{"name": "openvibe-sdk",-- "version": "0.21.1",++ "version": "0.21.2","description": "Supported browser and server clients for the OpenVibe platform: auth (incl. developer apps), registry discovery, identity, user modules, events, realtime, media, community, search, chat, AI runs, Tools (registry, run API and jobs), VIP memberships and developer projects, the async PostgreSQL data layer (db) and the Valkey cache, limits store, queue and pub/sub, with deadlines, safe retries, problem errors and trace propagation.","main": "index.js","types": "types/index.d.ts",@@@ -7,6 +7,8 @@## Tests read OV_TEST_PG_URL (through PgBouncer), OV_TEST_PG_DIRECT_URL and OV_TEST_VALKEY_URL; without them the# integration cases print "<label>: skipped (…)" and the suite still runs on PGlite and the memory stores.++# Every test process gets a role of its own, so PgBouncer holds a pool per role: idle server connections close after++# 5 s and one database takes at most 120, or a service with many test files runs PostgreSQL out of connections.set -euo pipefailNET=ovsdk-test; PW=ovtestpwcase "${1:-up}" in@@@ -16,12 +18,13 @@ up)docker rm -f ovsdk-pg >/dev/null 2>&1 || truedocker run -d --name ovsdk-pg --network $NET -p 127.0.0.1:55432:5432 \-e POSTGRES_USER=ov -e POSTGRES_PASSWORD=$PW -e POSTGRES_DB=ovtest -e POSTGRES_HOST_AUTH_METHOD=scram-sha-256 -e POSTGRES_INITDB_ARGS=--auth-host=scram-sha-256 \-- postgres:18-alpine >/dev/null++ postgres:18-alpine -c max_connections=200 >/dev/nullfiif ! docker ps --format '{{.Names}}' | grep -qx ovsdk-pgbouncer; thendocker rm -f ovsdk-pgbouncer >/dev/null 2>&1 || truedocker run -d --name ovsdk-pgbouncer --network $NET -p 127.0.0.1:56432:5432 \-e DATABASE_URL="postgres://ov:$PW@ovsdk-pg:5432/ovtest" -e POOL_MODE=transaction -e AUTH_TYPE=scram-sha-256 -e MAX_CLIENT_CONN=500 -e DEFAULT_POOL_SIZE=10 \++ -e MAX_DB_CONNECTIONS=120 -e SERVER_IDLE_TIMEOUT=5 \edoburu/pgbouncer:latest >/dev/nullfiif ! docker ps --format '{{.Names}}' | grep -qx ovsdk-valkey; then@@@ -8,7 +8,7 @@ const { paginate, offsetPager } = require('./paginate');const semver = require('./semver');/** The package version (test/version.test.js keeps it equal to package.json). */--const SDK_VERSION = '0.21.1';++const SDK_VERSION = '0.21.2';module.exports = {createClient,@@@ -28,6 +28,21 @@ const RETRYABLE_4XX = new Set([401, 408, 409, 425, 429]);// Only Events refusing the envelope is permanent. A failure to get a token (no grant yet, Network// restarting) says nothing about the event, so it is always retried.++/**++ * One relay pass at a time. A flush() while a pass runs gets the next pass, which starts after it (so it also covers rows++ * committed after that pass claimed); every call in the meantime shares that next pass. flush.idle() settles with both.++ */++function serialFlush(pass) {++ let current = null; let next = null;++ const flush = () => {++ if (!current) { current = pass().finally(() => { current = null; }); return current; }++ if (!next) next = current.catch(() => {}).then(() => { next = null; return flush(); });++ return next;++ };++ flush.idle = () => (next || current || Promise.resolve()).catch(() => {});++ return flush;++}++function isPermanent(err) {const s = err && err.status;if (err && typeof err.url === 'string' && /\/oauth\/token(\?|$)/.test(err.url)) return false;@@@ -46,7 +61,6 @@ function createOutbox(db, {let stmts = null;let timer = null;let running = false;-- let flushing = null;function ensureSchema() {db.exec(`CREATE TABLE IF NOT EXISTS ${table} (@@@ -140,11 +154,8 @@ function createOutbox(db, {return total;}-- /** Publish due rows once. Concurrent callers share the flush in progress. */-- function flush() {-- if (!flushing) flushing = doFlush().finally(() => { flushing = null; });-- return flushing;-- }++ /** Publish due rows once (serialFlush: a call during a pass also covers rows committed since that pass claimed). */++ const flush = serialFlush(doFlush);function schedule(ms) {if (!running) return;@@@ -162,7 +173,7 @@ function createOutbox(db, {enqueue,flush,start() { if (!running) { running = true; schedule(0); } },-- stop() { running = false; clearTimeout(timer); timer = null; return flushing || Promise.resolve(); },++ stop() { running = false; clearTimeout(timer); timer = null; return flush.idle(); },/** Wake the relay now (e.g. right after the transaction commits). */kick() { schedule(0); },pending: () => q().pending.get().n,@@@ -264,7 +275,7 @@ function createPgOutbox(db, {throw new TypeError('events must be an openvibe-sdk events client (createEventsClient)');}if (!TABLE_RE.test(table)) throw new TypeError('bad outbox table name');-- let timer = null; let running = false; let flushing = null;++ let timer = null; let running = false;const backoff = (attempts) => backoffMs[Math.min(attempts, backoffMs.length - 1)];async function enqueue(t, envelope, { traceparent } = {}) {@@@ -327,7 +338,7 @@ function createPgOutbox(db, {}return total;}-- function flush() { if (!flushing) flushing = doFlush().finally(() => { flushing = null; }); return flushing; }++ const flush = serialFlush(doFlush);function schedule(ms) {if (!running) return;clearTimeout(timer);@@@ -344,7 +355,7 @@ function createPgOutbox(db, {ensureSchema: () => db.query(outboxSchema(table)),enqueue, flush,start() { if (!running) { running = true; schedule(0); } },-- stop() { running = false; clearTimeout(timer); timer = null; return flushing || Promise.resolve(); },++ stop() { running = false; clearTimeout(timer); timer = null; return flush.idle(); },kick() { schedule(0); },pending: () => db.value(`SELECT count(*) FROM ${table} WHERE sent_at IS NULL AND rejected_at IS NULL`),rejected: () => db.value(`SELECT count(*) FROM ${table} WHERE rejected_at IS NOT NULL`),@@@ -107,6 +107,24 @@ run([assert.equal(outbox.rejected(), 0);assert.equal(outbox.pending(), 1);}],++ ['a flush called during a pass also publishes what was committed after that pass claimed', async () => {++ const { events, db } = setup();++ let release; const gate = new Promise((r) => { release = r; });++ const sent = [];++ const slow = { prepare: events.prepare, publish: async (input, o) => { await gate; sent.push(input.subject.id); return await events.publish(input, o); } };++ const outbox = createOutbox(db, { events: slow });++ outbox.ensureSchema();++ db.transaction(() => outbox.enqueue({ event_type: 'live.stream.started', actor, subject }))();++ const first = outbox.flush();++ await new Promise((r) => setImmediate(r));++ db.transaction(() => outbox.enqueue({ event_type: 'live.stream.started', actor, subject: { type: 'stream', id: '13', revision: 1 } }))();++ const second = outbox.flush();++ assert.equal(outbox.flush(), second, 'calls during a pass share the next one');++ release();++ await first; await second;++ assert.deepEqual(sent.sort(), ['12', '13']);++ assert.equal(outbox.pending(), 0);++ }],['inbox runs a handler exactly once and rolls back with it', async () => {const { db } = setup();const inbox = createInbox(db);
@@ -8,6 +8,11 @@ release may change an API and says so here.
@@ -8,6 +8,11 @@ release may change an API and says so here.
+## 0.21.2 (2026-09-28)
+
+- `openvibe-sdk/events` outboxes (SQLite and PostgreSQL): a `flush()` called while a relay pass runs now gets the next pass, which starts when that one ends. It used to share the running pass, so it could return before the rows committed after that pass claimed were published (a caller's "flush, then assert both went out" saw one). Calls made during a pass share that next pass; `stop()` waits for both.
+- `scripts/test-services.sh`: PgBouncer closes idle server connections after 5 s and takes at most 120 per database, and PostgreSQL allows 200. Each test process has a role of its own, so the pools of a suite with many files ran PostgreSQL out of connections.
+
## 0.21.1 (2026-09-28)
## 0.21.1 (2026-09-28)
- `openvibe-sdk/events` `createPgOutbox`: the relay publishes the rows it claimed in the order they were written. `UPDATE … RETURNING` gives no order, so a batch could reach Events out of order (a `media.clip.failed` before the `media.vod.ready` written before it). Found converting OpenVibe.Media.
- `openvibe-sdk/events` `createPgOutbox`: the relay publishes the rows it claimed in the order they were written. `UPDATE … RETURNING` gives no order, so a batch could reach Events out of order (a `media.clip.failed` before the `media.vod.ready` written before it). Found converting OpenVibe.Media.
@@ -3,7 +3,7 @@
@@ -3,7 +3,7 @@
"stage": "alpha",
"stage": "alpha",
"code": true,
"code": true,
"wave": "W2",
"wave": "W2",
- "version": "0.21.1",
+ "version": "0.21.2",
"plan": "OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026), §3.2; roadmap §30",
"plan": "OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026), §3.2; roadmap §30",
"deployed": false,
"deployed": false,
"deployment": "none: a package released as git tags; services and apps pin a tag",
"deployment": "none: a package released as git tags; services and apps pin a tag",
@@ -1,4 +1,4 @@
@@ -1,4 +1,4 @@
-// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.21.1, MIT). Do not edit.
+// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.21.2, MIT). Do not edit.
// One self-contained ES module: core at the top level plus the browser-safe subpaths as namespaces
// One self-contained ES module: core at the top level plus the browser-safe subpaths as namespaces
// (auth, registry, modules, realtime, media, community, search, chat, ai, jobs, tools, projects, geo). No server-only code: no service tokens, JWT verification, events, identity.
// (auth, registry, modules, realtime, media, community, search, chat, ai, jobs, tools, projects, geo). No server-only code: no service tokens, JWT verification, events, identity.
/* eslint-disable */
/* eslint-disable */
@@ -41,7 +41,7 @@ const { paginate, offsetPager } = __load(7);
@@ -41,7 +41,7 @@ const { paginate, offsetPager } = __load(7);
const semver = __load(6);
const semver = __load(6);
/** The package version (test/version.test.js keeps it equal to package.json). */
/** The package version (test/version.test.js keeps it equal to package.json). */
-const SDK_VERSION = '0.21.1';
+const SDK_VERSION = '0.21.2';
module.exports = {
module.exports = {
createClient,
createClient,
@@ -1,12 +1,12 @@
@@ -1,12 +1,12 @@
{
{
"name": "openvibe-sdk",
"name": "openvibe-sdk",
- "version": "0.21.1",
+ "version": "0.21.2",
"lockfileVersion": 3,
"lockfileVersion": 3,
"requires": true,
"requires": true,
"packages": {
"packages": {
"": {
"": {
"name": "openvibe-sdk",
"name": "openvibe-sdk",
- "version": "0.21.1",
+ "version": "0.21.2",
"license": "MIT",
"license": "MIT",
"devDependencies": {
"devDependencies": {
"@electric-sql/pglite": "^0.5.8",
"@electric-sql/pglite": "^0.5.8",
@@ -1,6 +1,6 @@
@@ -1,6 +1,6 @@
{
{
"name": "openvibe-sdk",
"name": "openvibe-sdk",
- "version": "0.21.1",
+ "version": "0.21.2",
"description": "Supported browser and server clients for the OpenVibe platform: auth (incl. developer apps), registry discovery, identity, user modules, events, realtime, media, community, search, chat, AI runs, Tools (registry, run API and jobs), VIP memberships and developer projects, the async PostgreSQL data layer (db) and the Valkey cache, limits store, queue and pub/sub, with deadlines, safe retries, problem errors and trace propagation.",
"description": "Supported browser and server clients for the OpenVibe platform: auth (incl. developer apps), registry discovery, identity, user modules, events, realtime, media, community, search, chat, AI runs, Tools (registry, run API and jobs), VIP memberships and developer projects, the async PostgreSQL data layer (db) and the Valkey cache, limits store, queue and pub/sub, with deadlines, safe retries, problem errors and trace propagation.",
"main": "index.js",
"main": "index.js",
"types": "types/index.d.ts",
"types": "types/index.d.ts",
@@ -7,6 +7,8 @@
@@ -7,6 +7,8 @@
#
#
# Tests read OV_TEST_PG_URL (through PgBouncer), OV_TEST_PG_DIRECT_URL and OV_TEST_VALKEY_URL; without them the
# Tests read OV_TEST_PG_URL (through PgBouncer), OV_TEST_PG_DIRECT_URL and OV_TEST_VALKEY_URL; without them the
# integration cases print "<label>: skipped (…)" and the suite still runs on PGlite and the memory stores.
# integration cases print "<label>: skipped (…)" and the suite still runs on PGlite and the memory stores.
+# Every test process gets a role of its own, so PgBouncer holds a pool per role: idle server connections close after
+# 5 s and one database takes at most 120, or a service with many test files runs PostgreSQL out of connections.
set -euo pipefail
set -euo pipefail
NET=ovsdk-test; PW=ovtestpw
NET=ovsdk-test; PW=ovtestpw
case "${1:-up}" in
case "${1:-up}" in
@@ -16,12 +18,13 @@ up)
@@ -16,12 +18,13 @@ up)
docker rm -f ovsdk-pg >/dev/null 2>&1 || true
docker rm -f ovsdk-pg >/dev/null 2>&1 || true
docker run -d --name ovsdk-pg --network $NET -p 127.0.0.1:55432:5432 \
docker run -d --name ovsdk-pg --network $NET -p 127.0.0.1:55432:5432 \
-e POSTGRES_USER=ov -e POSTGRES_PASSWORD=$PW -e POSTGRES_DB=ovtest -e POSTGRES_HOST_AUTH_METHOD=scram-sha-256 -e POSTGRES_INITDB_ARGS=--auth-host=scram-sha-256 \
-e POSTGRES_USER=ov -e POSTGRES_PASSWORD=$PW -e POSTGRES_DB=ovtest -e POSTGRES_HOST_AUTH_METHOD=scram-sha-256 -e POSTGRES_INITDB_ARGS=--auth-host=scram-sha-256 \
- postgres:18-alpine >/dev/null
+ postgres:18-alpine -c max_connections=200 >/dev/null
fi
fi
if ! docker ps --format '{{.Names}}' | grep -qx ovsdk-pgbouncer; then
if ! docker ps --format '{{.Names}}' | grep -qx ovsdk-pgbouncer; then
docker rm -f ovsdk-pgbouncer >/dev/null 2>&1 || true
docker rm -f ovsdk-pgbouncer >/dev/null 2>&1 || true
docker run -d --name ovsdk-pgbouncer --network $NET -p 127.0.0.1:56432:5432 \
docker run -d --name ovsdk-pgbouncer --network $NET -p 127.0.0.1:56432:5432 \
-e DATABASE_URL="postgres://ov:$PW@ovsdk-pg:5432/ovtest" -e POOL_MODE=transaction -e AUTH_TYPE=scram-sha-256 -e MAX_CLIENT_CONN=500 -e DEFAULT_POOL_SIZE=10 \
-e DATABASE_URL="postgres://ov:$PW@ovsdk-pg:5432/ovtest" -e POOL_MODE=transaction -e AUTH_TYPE=scram-sha-256 -e MAX_CLIENT_CONN=500 -e DEFAULT_POOL_SIZE=10 \
+ -e MAX_DB_CONNECTIONS=120 -e SERVER_IDLE_TIMEOUT=5 \
edoburu/pgbouncer:latest >/dev/null
edoburu/pgbouncer:latest >/dev/null
fi
fi
if ! docker ps --format '{{.Names}}' | grep -qx ovsdk-valkey; then
if ! docker ps --format '{{.Names}}' | grep -qx ovsdk-valkey; then
@@ -8,7 +8,7 @@ const { paginate, offsetPager } = require('./paginate');
@@ -8,7 +8,7 @@ const { paginate, offsetPager } = require('./paginate');
const semver = require('./semver');
const semver = require('./semver');
/** The package version (test/version.test.js keeps it equal to package.json). */
/** The package version (test/version.test.js keeps it equal to package.json). */
-const SDK_VERSION = '0.21.1';
+const SDK_VERSION = '0.21.2';
module.exports = {
module.exports = {
createClient,
createClient,
@@ -28,6 +28,21 @@ const RETRYABLE_4XX = new Set([401, 408, 409, 425, 429]);
@@ -28,6 +28,21 @@ const RETRYABLE_4XX = new Set([401, 408, 409, 425, 429]);
// Only Events refusing the envelope is permanent. A failure to get a token (no grant yet, Network
// Only Events refusing the envelope is permanent. A failure to get a token (no grant yet, Network
// restarting) says nothing about the event, so it is always retried.
// restarting) says nothing about the event, so it is always retried.
+/**
+ * One relay pass at a time. A flush() while a pass runs gets the next pass, which starts after it (so it also covers rows
+ * committed after that pass claimed); every call in the meantime shares that next pass. flush.idle() settles with both.
+ */
+function serialFlush(pass) {
+ let current = null; let next = null;
+ const flush = () => {
+ if (!current) { current = pass().finally(() => { current = null; }); return current; }
+ if (!next) next = current.catch(() => {}).then(() => { next = null; return flush(); });
+ return next;
+ };
+ flush.idle = () => (next || current || Promise.resolve()).catch(() => {});
+ return flush;
+}
+
function isPermanent(err) {
function isPermanent(err) {
const s = err && err.status;
const s = err && err.status;
if (err && typeof err.url === 'string' && /\/oauth\/token(\?|$)/.test(err.url)) return false;
if (err && typeof err.url === 'string' && /\/oauth\/token(\?|$)/.test(err.url)) return false;
@@ -46,7 +61,6 @@ function createOutbox(db, {
@@ -46,7 +61,6 @@ function createOutbox(db, {
let stmts = null;
let stmts = null;
let timer = null;
let timer = null;
let running = false;
let running = false;
- let flushing = null;
function ensureSchema() {
function ensureSchema() {
db.exec(`CREATE TABLE IF NOT EXISTS ${table} (
db.exec(`CREATE TABLE IF NOT EXISTS ${table} (
@@ -140,11 +154,8 @@ function createOutbox(db, {
@@ -140,11 +154,8 @@ function createOutbox(db, {
return total;
return total;
}
}
- /** Publish due rows once. Concurrent callers share the flush in progress. */
- function flush() {
- if (!flushing) flushing = doFlush().finally(() => { flushing = null; });
- return flushing;
- }
+ /** Publish due rows once (serialFlush: a call during a pass also covers rows committed since that pass claimed). */
+ const flush = serialFlush(doFlush);
function schedule(ms) {
function schedule(ms) {
if (!running) return;
if (!running) return;
@@ -162,7 +173,7 @@ function createOutbox(db, {
@@ -162,7 +173,7 @@ function createOutbox(db, {
enqueue,
enqueue,
flush,
flush,
start() { if (!running) { running = true; schedule(0); } },
start() { if (!running) { running = true; schedule(0); } },
- stop() { running = false; clearTimeout(timer); timer = null; return flushing || Promise.resolve(); },
+ stop() { running = false; clearTimeout(timer); timer = null; return flush.idle(); },
/** Wake the relay now (e.g. right after the transaction commits). */
/** Wake the relay now (e.g. right after the transaction commits). */
kick() { schedule(0); },
kick() { schedule(0); },
pending: () => q().pending.get().n,
pending: () => q().pending.get().n,
@@ -264,7 +275,7 @@ function createPgOutbox(db, {
@@ -264,7 +275,7 @@ function createPgOutbox(db, {
throw new TypeError('events must be an openvibe-sdk events client (createEventsClient)');
throw new TypeError('events must be an openvibe-sdk events client (createEventsClient)');
}
}
if (!TABLE_RE.test(table)) throw new TypeError('bad outbox table name');
if (!TABLE_RE.test(table)) throw new TypeError('bad outbox table name');
- let timer = null; let running = false; let flushing = null;
+ let timer = null; let running = false;
const backoff = (attempts) => backoffMs[Math.min(attempts, backoffMs.length - 1)];
const backoff = (attempts) => backoffMs[Math.min(attempts, backoffMs.length - 1)];
async function enqueue(t, envelope, { traceparent } = {}) {
async function enqueue(t, envelope, { traceparent } = {}) {
@@ -327,7 +338,7 @@ function createPgOutbox(db, {
@@ -327,7 +338,7 @@ function createPgOutbox(db, {
}
}
return total;
return total;
}
}
- function flush() { if (!flushing) flushing = doFlush().finally(() => { flushing = null; }); return flushing; }
+ const flush = serialFlush(doFlush);
function schedule(ms) {
function schedule(ms) {
if (!running) return;
if (!running) return;
clearTimeout(timer);
clearTimeout(timer);
@@ -344,7 +355,7 @@ function createPgOutbox(db, {
@@ -344,7 +355,7 @@ function createPgOutbox(db, {
ensureSchema: () => db.query(outboxSchema(table)),
ensureSchema: () => db.query(outboxSchema(table)),
enqueue, flush,
enqueue, flush,
start() { if (!running) { running = true; schedule(0); } },
start() { if (!running) { running = true; schedule(0); } },
- stop() { running = false; clearTimeout(timer); timer = null; return flushing || Promise.resolve(); },
+ stop() { running = false; clearTimeout(timer); timer = null; return flush.idle(); },
kick() { schedule(0); },
kick() { schedule(0); },
pending: () => db.value(`SELECT count(*) FROM ${table} WHERE sent_at IS NULL AND rejected_at IS NULL`),
pending: () => db.value(`SELECT count(*) FROM ${table} WHERE sent_at IS NULL AND rejected_at IS NULL`),
rejected: () => db.value(`SELECT count(*) FROM ${table} WHERE rejected_at IS NOT NULL`),
rejected: () => db.value(`SELECT count(*) FROM ${table} WHERE rejected_at IS NOT NULL`),
@@ -107,6 +107,24 @@ run([
@@ -107,6 +107,24 @@ run([
assert.equal(outbox.rejected(), 0);
assert.equal(outbox.rejected(), 0);
assert.equal(outbox.pending(), 1);
assert.equal(outbox.pending(), 1);
}],
}],
+ ['a flush called during a pass also publishes what was committed after that pass claimed', async () => {
+ const { events, db } = setup();
+ let release; const gate = new Promise((r) => { release = r; });
+ const sent = [];
+ const slow = { prepare: events.prepare, publish: async (input, o) => { await gate; sent.push(input.subject.id); return await events.publish(input, o); } };
+ const outbox = createOutbox(db, { events: slow });
+ outbox.ensureSchema();
+ db.transaction(() => outbox.enqueue({ event_type: 'live.stream.started', actor, subject }))();
+ const first = outbox.flush();
+ await new Promise((r) => setImmediate(r));
+ db.transaction(() => outbox.enqueue({ event_type: 'live.stream.started', actor, subject: { type: 'stream', id: '13', revision: 1 } }))();
+ const second = outbox.flush();
+ assert.equal(outbox.flush(), second, 'calls during a pass share the next one');
+ release();
+ await first; await second;
+ assert.deepEqual(sent.sort(), ['12', '13']);
+ assert.equal(outbox.pending(), 0);
+ }],
['inbox runs a handler exactly once and rolls back with it', async () => {
['inbox runs a handler exactly once and rolls back with it', async () => {
const { db } = setup();
const { db } = setup();
const inbox = createInbox(db);
const inbox = createInbox(db);