openvibers/openvibe.sdk Commits
0.63.0: outbox relays pause on a 429; service outboxes prune sent rows (10-24 batch) (#75)
Changes
@@@ -3,6 +3,21 @@All notable changes to `openvibe-sdk`. The package follows semver; while it is `0.x`, a minorrelease may change an API and says so here.++## 0.63.0 (2026-10-24)++++- **Outbox relays pause on a 429.** `createPgOutbox` (and so `createServiceOutbox` and the usage relay) used to hit++ Events' per-actor limit again on every tick after a 429, because the rows it claims next are under the same limit.++ Search's re-index bursts were refused 935 times on 2026-10-23, and every event was delivered only on a retry.++ - A 429 now pauses the whole relay until the limit's window reopens: its `retry_after_seconds`, else a minute, at++ most an hour.++ - Passes before then send nothing. The refused rows back off as before, and a 429 never rejects a row.++ - Nothing changes for a service that stays under its limit.++- **`createServiceOutbox` prunes sent rows.** Rows sent more than `pruneAfterMs` ago (default 7 days; `false` keeps++ them) are deleted every 6 hours while the relay runs, and once a minute after `start()`; `out.prune()` runs it now.++ - Services that never called `outbox.prune()` kept every event they ever sent: on 2026-10-23, Work had 52,130 rows++ (119 MB) since 10-10, and Trade, Deals and Help since September or early October.++ - A service that already prunes itself is unaffected (a second prune finds nothing).++## 0.62.0 (2026-10-23)- **`openvibe-sdk/no-looks`: pronouns.** A pronoun is a guessed gender too, so the 27 Community summaries repaired on@@@ -2,14 +2,14 @@> Supported browser and server clients for the OpenVibe platform.--**Status:** alpha, v0.62.0 (roadmap Wave 2; developer apps from Wave 20; Media objects v2 from Wave 4; the Tools platform API; Search, Chat and AI clients from WS-F task 4; per-actor limits from WS-R task 4). Every release is tagged (see [CHANGELOG.md](CHANGELOG.md)). Tested against local stub servers and the built-in mock platform only, never against the live platform. Services pin release tags; each repository's `STATUS.json` names the one it uses.++**Status:** alpha, v0.63.0 (roadmap Wave 2; developer apps from Wave 20; Media objects v2 from Wave 4; the Tools platform API; Search, Chat and AI clients from WS-F task 4; per-actor limits from WS-R task 4). Every release is tagged (see [CHANGELOG.md](CHANGELOG.md)). Tested against local stub servers and the built-in mock platform only, never against the live platform. Services pin release tags; each repository's `STATUS.json` names the one it uses.**Plan:** OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026), §3.2; roadmap §30.**License:** MIT ([LICENSE](LICENSE)). This package is a library that apps outside the network embed, so it uses MIT. The OpenVibe services themselves stay AGPL-3.0.**If a capability is not in the SDK, it is not public.** Apps call services through `openvibe-sdk` and never build internal routes themselves. A route with no SDK wrapper is internal, even when you can reach it, and it can change without notice. To make a capability public, first define it in OpenVibe.Contracts, then wrap it here.```bash--npm install https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.62.0++npm install https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.63.0```It has no runtime dependencies. It needs Node ≥ 20, or any browser with `fetch`, Web Crypto and `TextDecoder`. There is no build step. The package is CommonJS with ESM entry points (`import` works). Each subpath has its own `.d.ts`. For a page with no bundler, `browser/openvibe-sdk.mjs` is one self-contained ES module of the browser-safe subpaths (see [Browser without a bundler](#browser-without-a-bundler)). It does not depend on `openvibe-contracts` at runtime: it copies the contract types it uses (from Contracts v0.86.0, a devDependency the tests check them against).@@@ -3,7 +3,7 @@"stage": "alpha","code": true,"wave": "W2",-- "version": "0.62.0",++ "version": "0.63.0","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.62.0, MIT). Do not edit.++// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.63.0, 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, bot, search, chat, ai, jobs, tools, projects, geo). No server-only code: no service tokens, JWT verification, events, identity./* eslint-disable */@@@ -42,7 +42,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.62.0';++const SDK_VERSION = '0.63.0';module.exports = {createClient,@@@ -1,12 +1,12 @@{"name": "openvibe-sdk",-- "version": "0.62.0",++ "version": "0.63.0","lockfileVersion": 3,"requires": true,"packages": {"": {"name": "openvibe-sdk",-- "version": "0.62.0",++ "version": "0.63.0","license": "MIT","bin": {"openvibe-download": "bin/openvibe-download.js"@@@ -1,6 +1,6 @@{"name": "openvibe-sdk",-- "version": "0.62.0",++ "version": "0.63.0","description": "Supported browser and server clients for the OpenVibe platform: auth (incl. developer apps), registry discovery, identity, user modules, events, realtime, media, community, Bot robots, 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",@@@ -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.62.0';++const SDK_VERSION = '0.63.0';module.exports = {createClient,@@@ -107,6 +107,14 @@ function createPgOutbox(db, {if (!TABLE_RE.test(table)) throw new TypeError('bad outbox table name');let timer = null; let running = false;const backoff = (attempts) => backoffMs[Math.min(attempts, backoffMs.length - 1)];++ // A 429 (Events' per-actor limit) pauses the whole relay until the limit's window reopens (its++ // retry_after_seconds, else a minute): every row it would claim next is under the same limit, so a burst (a++ // re-index) was refused on every tick for the rest of the minute (Search, 935 refusals on 2026-10-23).++ let pausedUntil = 0;++ const pauseMs = (err) => {++ const s = Number(err && err.problem && err.problem.retry_after_seconds);++ return (Number.isFinite(s) && s > 0 ? Math.min(s, 3600) : 60) * 1000;++ };async function enqueue(t, envelope, { traceparent } = {}) {if (!t || typeof t.query !== 'function' || typeof t.tx !== 'function') throw new TypeError('outbox.enqueue(t, …): pass the transaction handle db.tx gives you');@@@ -144,6 +152,7 @@ function createPgOutbox(db, {[now(), rows.map((r) => r.id)]);stats.sent += rows.length;} catch (err) {++ if (err && err.status === 429) pausedUntil = now() + pauseMs(err);if (rows.length > 1 && isPermanent(err)) {for (const row of rows) { const s = await publishRows([row]); stats.sent += s.sent; stats.failed += s.failed; stats.rejected += s.rejected; }} else {@@@ -157,6 +166,7 @@ function createPgOutbox(db, {async function doFlush() {const total = { sent: 0, failed: 0, rejected: 0 };for (;;) {++ if (now() < pausedUntil) break;const rows = await claim();if (!rows.length) break;const s = await publishRows(rows);@@@ -19,6 +19,10 @@* await out.emitIn(t, envelope, { traceparent }); // or on an explicit transaction handle* await out.moderationAction({ action, target, actorSubject, reason, details }); // <source>.moderation.action* out.start(); out.stop(); await out.kick(); await out.status(); // → { enabled, pending, rejected, last_error }++ *++ * Sent rows are pruned once they are `pruneAfterMs` old (default 7 days; false keeps them), every 6 hours while the++ * relay runs and once soon after start(). Services that never called outbox.prune() kept every event they ever sent++ * (Work 52,130 rows / 119 MB, Trade, Deals and Help since September or early October, on 2026-10-23).*/const { createClient } = require('./core');const { createServiceTokenClient } = require('./auth/tokens');@@@ -26,7 +30,7 @@ const { createServiceTokenClient } = require('./auth/tokens');function createServiceOutbox({db, source, eventsUrl = null, networkInternalUrl = 'http://127.0.0.1:4000', clientId = null, clientSecret = null,table = 'event_outbox', intervalMs, now, fetch: fetchImpl, log = console, eventTypes = null, validate = null,-- autoDiscover, moderationOwnerSubject = true,++ autoDiscover, moderationOwnerSubject = true, pruneAfterMs = 7 * 24 * 60 * 60 * 1000,} = {}) {if (!db) throw new TypeError('createServiceOutbox: db is required');if (!source || !/^[a-z][a-z0-9_-]*$/.test(source)) throw new TypeError('createServiceOutbox: source is the service id (e.g. "trade")');@@@ -95,10 +99,26 @@ function createServiceOutbox({}, { traceparent });}++ const PRUNE_EVERY_MS = 6 * 60 * 60 * 1000;++ let pruneTimers = [];++ const prune = () => {++ if (!(pruneAfterMs > 0)) return Promise.resolve(0);++ return outbox.prune(pruneAfterMs).catch((err) => { log.warn(`[outbox:${source}] prune failed: ${err && err.message}`); return 0; });++ };return {-- emit, emitIn, moderationAction, outbox, events, enabled,-- start() { if (enabled) outbox.start(); },-- stop: () => outbox.stop(),++ emit, emitIn, moderationAction, outbox, events, enabled, prune,++ start() {++ if (!enabled) return;++ outbox.start();++ if (pruneAfterMs > 0 && !pruneTimers.length) {++ const first = setTimeout(prune, 60 * 1000);++ const every = setInterval(prune, PRUNE_EVERY_MS);++ if (first.unref) first.unref();++ if (every.unref) every.unref();++ pruneTimers = [first, every];++ }++ },++ stop: () => { for (const t of pruneTimers) { clearTimeout(t); clearInterval(t); } pruneTimers = []; return outbox.stop(); },async kick() { if (enabled) await outbox.kick(); },status: async () => ({ enabled, pending: await outbox.pending(), rejected: await outbox.rejected(), last_error: lastError }),};@@@ -110,6 +110,31 @@ function cases(label, open) {assert.equal(await o2.rejected(), 1);} finally { await done(); }}],++ [`${label}: a 429 pauses the whole relay for its retry_after_seconds (else a minute), then it drains`, async () => {++ const { db, done } = await open();++ try {++ const { platform, events } = eventsClient();++ let t = 2_000_000; let limited = 1; let calls = 0;++ const tooMany = (s) => Object.assign(new Error('rate_limited'), { status: 429, problem: s == null ? null : { code: 'rate_limited', status: 429, retry_after_seconds: s } });++ const busy = { prepare: events.prepare, publish: (...a) => { calls++; return limited ? Promise.reject(tooMany(limited === 1 ? 20 : null)) : events.publish(...a); } };++ const outbox = createPgOutbox(db, { events: busy, batchSize: 2, now: () => t });++ await db.tx(async (tx) => { for (let i = 0; i < 6; i++) await outbox.enqueue(tx, { event_type: 'live.stream.started', actor, subject: { type: 'stream', id: `r${i}` } }); });++ assert.deepEqual(await outbox.flush(), { sent: 0, failed: 2, rejected: 0 }, 'the first batch is refused');++ t += 1000;++ assert.deepEqual(await outbox.flush(), { sent: 0, failed: 0, rejected: 0 }, 'paused: the next rows are not even tried');++ assert.equal(calls, 1, 'one refused request, not one per tick');++ t += 20000;++ limited = 2; // refused again, with no retry_after_seconds: a minute++ await outbox.flush();++ t += 59000;++ assert.deepEqual(await outbox.flush(), { sent: 0, failed: 0, rejected: 0 }, 'still within the minute');++ t += 2000; limited = 0;++ let sent = 0; for (let i = 0; i < 5; i++) sent += (await outbox.flush()).sent;++ assert.equal(sent, 6, 'every row went out once the window reopened');++ assert.equal(platform.state.events.length, 6);++ assert.equal(await outbox.rejected(), 0, 'a 429 never rejects');++ } finally { await done(); }++ }],[`${label}: a token-endpoint refusal is retried, never rejected`, async () => {const { db, done } = await open();try {@@@ -75,6 +75,25 @@ run([assert.deepEqual([b.payload.target.owner_subject, b.actor], [null, { type: 'service', id: 'trade' }]);} finally { await db.close(); }}],++ ['sent rows older than pruneAfterMs go (prune()), unsent and recent ones stay; false keeps everything', async () => {++ const db = await open(); const p = platform();++ try {++ let t = 1_000_000_000;++ const out = make(db, p, { now: () => t, pruneAfterMs: 1000 });++ await db.tx((tx) => out.emitIn(tx, { event_type: 'trade.observation.created', actor, subject }));++ await out.outbox.flush();++ t += 500;++ await db.tx((tx) => out.emitIn(tx, { event_type: 'trade.observation.created', actor, subject: { ...subject, id: 'later' } }));++ await out.outbox.flush();++ await db.tx((tx) => out.emitIn(tx, { event_type: 'trade.observation.created', actor, subject: { ...subject, id: 'unsent' } }));++ t += 800; // the first sent row is 1,300 ms old, the second 800 ms++ assert.equal(await out.prune(), 1);++ assert.equal(await db.value(sql`SELECT count(*)::int FROM event_outbox`), 2, 'the recent sent row and the unsent one stay');++ const keep = make(db, p, { now: () => t + 10_000, pruneAfterMs: false });++ assert.equal(await keep.prune(), 0, 'false: nothing is pruned');++ out.start(); await out.stop(); // start() schedules the prune timers and stop() clears them (the test would hang otherwise)++ } finally { await db.close(); }++ }],['a service with its own table name; misuse is a TypeError', async () => {const db = await open('tips_event_outbox'); const p = platform();try {@@@ -171,6 +171,8 @@ export interface ServiceOutboxOptions {autoDiscover?: boolean;/** false: <source>.moderation.action withholds target.owner_subject. */moderationOwnerSubject?: boolean;++ /** Sent rows older than this are pruned every 6 hours while the relay runs (default 7 days); false keeps them. */++ pruneAfterMs?: number | false;}export interface ServiceOutbox {/** Inside the change's ambient openvibe-sdk/db transaction. */@@@ -181,6 +183,8 @@ export interface ServiceOutbox {outbox: PgOutbox;events: EventsClient;enabled: boolean;++ /** Prunes sent rows older than pruneAfterMs now; resolves with the count (0 when off or on failure). */++ prune(): Promise<number>;start(): void;stop(): Promise<unknown>;kick(): Promise<void>;
@@ -3,6 +3,21 @@
@@ -3,6 +3,21 @@
All notable changes to `openvibe-sdk`. The package follows semver; while it is `0.x`, a minor
All notable changes to `openvibe-sdk`. The package follows semver; while it is `0.x`, a minor
release may change an API and says so here.
release may change an API and says so here.
+## 0.63.0 (2026-10-24)
+
+- **Outbox relays pause on a 429.** `createPgOutbox` (and so `createServiceOutbox` and the usage relay) used to hit
+ Events' per-actor limit again on every tick after a 429, because the rows it claims next are under the same limit.
+ Search's re-index bursts were refused 935 times on 2026-10-23, and every event was delivered only on a retry.
+ - A 429 now pauses the whole relay until the limit's window reopens: its `retry_after_seconds`, else a minute, at
+ most an hour.
+ - Passes before then send nothing. The refused rows back off as before, and a 429 never rejects a row.
+ - Nothing changes for a service that stays under its limit.
+- **`createServiceOutbox` prunes sent rows.** Rows sent more than `pruneAfterMs` ago (default 7 days; `false` keeps
+ them) are deleted every 6 hours while the relay runs, and once a minute after `start()`; `out.prune()` runs it now.
+ - Services that never called `outbox.prune()` kept every event they ever sent: on 2026-10-23, Work had 52,130 rows
+ (119 MB) since 10-10, and Trade, Deals and Help since September or early October.
+ - A service that already prunes itself is unaffected (a second prune finds nothing).
+
## 0.62.0 (2026-10-23)
## 0.62.0 (2026-10-23)
- **`openvibe-sdk/no-looks`: pronouns.** A pronoun is a guessed gender too, so the 27 Community summaries repaired on
- **`openvibe-sdk/no-looks`: pronouns.** A pronoun is a guessed gender too, so the 27 Community summaries repaired on
@@ -2,14 +2,14 @@
@@ -2,14 +2,14 @@
> Supported browser and server clients for the OpenVibe platform.
> Supported browser and server clients for the OpenVibe platform.
-**Status:** alpha, v0.62.0 (roadmap Wave 2; developer apps from Wave 20; Media objects v2 from Wave 4; the Tools platform API; Search, Chat and AI clients from WS-F task 4; per-actor limits from WS-R task 4). Every release is tagged (see [CHANGELOG.md](CHANGELOG.md)). Tested against local stub servers and the built-in mock platform only, never against the live platform. Services pin release tags; each repository's `STATUS.json` names the one it uses.
+**Status:** alpha, v0.63.0 (roadmap Wave 2; developer apps from Wave 20; Media objects v2 from Wave 4; the Tools platform API; Search, Chat and AI clients from WS-F task 4; per-actor limits from WS-R task 4). Every release is tagged (see [CHANGELOG.md](CHANGELOG.md)). Tested against local stub servers and the built-in mock platform only, never against the live platform. Services pin release tags; each repository's `STATUS.json` names the one it uses.
**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.
**License:** MIT ([LICENSE](LICENSE)). This package is a library that apps outside the network embed, so it uses MIT. The OpenVibe services themselves stay AGPL-3.0.
**License:** MIT ([LICENSE](LICENSE)). This package is a library that apps outside the network embed, so it uses MIT. The OpenVibe services themselves stay AGPL-3.0.
**If a capability is not in the SDK, it is not public.** Apps call services through `openvibe-sdk` and never build internal routes themselves. A route with no SDK wrapper is internal, even when you can reach it, and it can change without notice. To make a capability public, first define it in OpenVibe.Contracts, then wrap it here.
**If a capability is not in the SDK, it is not public.** Apps call services through `openvibe-sdk` and never build internal routes themselves. A route with no SDK wrapper is internal, even when you can reach it, and it can change without notice. To make a capability public, first define it in OpenVibe.Contracts, then wrap it here.
```bash
```bash
-npm install https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.62.0
+npm install https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.63.0
```
```
It has no runtime dependencies. It needs Node ≥ 20, or any browser with `fetch`, Web Crypto and `TextDecoder`. There is no build step. The package is CommonJS with ESM entry points (`import` works). Each subpath has its own `.d.ts`. For a page with no bundler, `browser/openvibe-sdk.mjs` is one self-contained ES module of the browser-safe subpaths (see [Browser without a bundler](#browser-without-a-bundler)). It does not depend on `openvibe-contracts` at runtime: it copies the contract types it uses (from Contracts v0.86.0, a devDependency the tests check them against).
It has no runtime dependencies. It needs Node ≥ 20, or any browser with `fetch`, Web Crypto and `TextDecoder`. There is no build step. The package is CommonJS with ESM entry points (`import` works). Each subpath has its own `.d.ts`. For a page with no bundler, `browser/openvibe-sdk.mjs` is one self-contained ES module of the browser-safe subpaths (see [Browser without a bundler](#browser-without-a-bundler)). It does not depend on `openvibe-contracts` at runtime: it copies the contract types it uses (from Contracts v0.86.0, a devDependency the tests check them against).
@@ -3,7 +3,7 @@
@@ -3,7 +3,7 @@
"stage": "alpha",
"stage": "alpha",
"code": true,
"code": true,
"wave": "W2",
"wave": "W2",
- "version": "0.62.0",
+ "version": "0.63.0",
"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.62.0, MIT). Do not edit.
+// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.63.0, 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, bot, search, chat, ai, jobs, tools, projects, geo). No server-only code: no service tokens, JWT verification, events, identity.
// (auth, registry, modules, realtime, media, community, bot, search, chat, ai, jobs, tools, projects, geo). No server-only code: no service tokens, JWT verification, events, identity.
/* eslint-disable */
/* eslint-disable */
@@ -42,7 +42,7 @@ const { paginate, offsetPager } = __load(7);
@@ -42,7 +42,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.62.0';
+const SDK_VERSION = '0.63.0';
module.exports = {
module.exports = {
createClient,
createClient,
@@ -1,12 +1,12 @@
@@ -1,12 +1,12 @@
{
{
"name": "openvibe-sdk",
"name": "openvibe-sdk",
- "version": "0.62.0",
+ "version": "0.63.0",
"lockfileVersion": 3,
"lockfileVersion": 3,
"requires": true,
"requires": true,
"packages": {
"packages": {
"": {
"": {
"name": "openvibe-sdk",
"name": "openvibe-sdk",
- "version": "0.62.0",
+ "version": "0.63.0",
"license": "MIT",
"license": "MIT",
"bin": {
"bin": {
"openvibe-download": "bin/openvibe-download.js"
"openvibe-download": "bin/openvibe-download.js"
@@ -1,6 +1,6 @@
@@ -1,6 +1,6 @@
{
{
"name": "openvibe-sdk",
"name": "openvibe-sdk",
- "version": "0.62.0",
+ "version": "0.63.0",
"description": "Supported browser and server clients for the OpenVibe platform: auth (incl. developer apps), registry discovery, identity, user modules, events, realtime, media, community, Bot robots, 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, Bot robots, 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",
@@ -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.62.0';
+const SDK_VERSION = '0.63.0';
module.exports = {
module.exports = {
createClient,
createClient,
@@ -107,6 +107,14 @@ function createPgOutbox(db, {
@@ -107,6 +107,14 @@ function createPgOutbox(db, {
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 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)];
+ // A 429 (Events' per-actor limit) pauses the whole relay until the limit's window reopens (its
+ // retry_after_seconds, else a minute): every row it would claim next is under the same limit, so a burst (a
+ // re-index) was refused on every tick for the rest of the minute (Search, 935 refusals on 2026-10-23).
+ let pausedUntil = 0;
+ const pauseMs = (err) => {
+ const s = Number(err && err.problem && err.problem.retry_after_seconds);
+ return (Number.isFinite(s) && s > 0 ? Math.min(s, 3600) : 60) * 1000;
+ };
async function enqueue(t, envelope, { traceparent } = {}) {
async function enqueue(t, envelope, { traceparent } = {}) {
if (!t || typeof t.query !== 'function' || typeof t.tx !== 'function') throw new TypeError('outbox.enqueue(t, …): pass the transaction handle db.tx gives you');
if (!t || typeof t.query !== 'function' || typeof t.tx !== 'function') throw new TypeError('outbox.enqueue(t, …): pass the transaction handle db.tx gives you');
@@ -144,6 +152,7 @@ function createPgOutbox(db, {
@@ -144,6 +152,7 @@ function createPgOutbox(db, {
[now(), rows.map((r) => r.id)]);
[now(), rows.map((r) => r.id)]);
stats.sent += rows.length;
stats.sent += rows.length;
} catch (err) {
} catch (err) {
+ if (err && err.status === 429) pausedUntil = now() + pauseMs(err);
if (rows.length > 1 && isPermanent(err)) {
if (rows.length > 1 && isPermanent(err)) {
for (const row of rows) { const s = await publishRows([row]); stats.sent += s.sent; stats.failed += s.failed; stats.rejected += s.rejected; }
for (const row of rows) { const s = await publishRows([row]); stats.sent += s.sent; stats.failed += s.failed; stats.rejected += s.rejected; }
} else {
} else {
@@ -157,6 +166,7 @@ function createPgOutbox(db, {
@@ -157,6 +166,7 @@ function createPgOutbox(db, {
async function doFlush() {
async function doFlush() {
const total = { sent: 0, failed: 0, rejected: 0 };
const total = { sent: 0, failed: 0, rejected: 0 };
for (;;) {
for (;;) {
+ if (now() < pausedUntil) break;
const rows = await claim();
const rows = await claim();
if (!rows.length) break;
if (!rows.length) break;
const s = await publishRows(rows);
const s = await publishRows(rows);
@@ -19,6 +19,10 @@
@@ -19,6 +19,10 @@
* await out.emitIn(t, envelope, { traceparent }); // or on an explicit transaction handle
* await out.emitIn(t, envelope, { traceparent }); // or on an explicit transaction handle
* await out.moderationAction({ action, target, actorSubject, reason, details }); // <source>.moderation.action
* await out.moderationAction({ action, target, actorSubject, reason, details }); // <source>.moderation.action
* out.start(); out.stop(); await out.kick(); await out.status(); // → { enabled, pending, rejected, last_error }
* out.start(); out.stop(); await out.kick(); await out.status(); // → { enabled, pending, rejected, last_error }
+ *
+ * Sent rows are pruned once they are `pruneAfterMs` old (default 7 days; false keeps them), every 6 hours while the
+ * relay runs and once soon after start(). Services that never called outbox.prune() kept every event they ever sent
+ * (Work 52,130 rows / 119 MB, Trade, Deals and Help since September or early October, on 2026-10-23).
*/
*/
const { createClient } = require('./core');
const { createClient } = require('./core');
const { createServiceTokenClient } = require('./auth/tokens');
const { createServiceTokenClient } = require('./auth/tokens');
@@ -26,7 +30,7 @@ const { createServiceTokenClient } = require('./auth/tokens');
@@ -26,7 +30,7 @@ const { createServiceTokenClient } = require('./auth/tokens');
function createServiceOutbox({
function createServiceOutbox({
db, source, eventsUrl = null, networkInternalUrl = 'http://127.0.0.1:4000', clientId = null, clientSecret = null,
db, source, eventsUrl = null, networkInternalUrl = 'http://127.0.0.1:4000', clientId = null, clientSecret = null,
table = 'event_outbox', intervalMs, now, fetch: fetchImpl, log = console, eventTypes = null, validate = null,
table = 'event_outbox', intervalMs, now, fetch: fetchImpl, log = console, eventTypes = null, validate = null,
- autoDiscover, moderationOwnerSubject = true,
+ autoDiscover, moderationOwnerSubject = true, pruneAfterMs = 7 * 24 * 60 * 60 * 1000,
} = {}) {
} = {}) {
if (!db) throw new TypeError('createServiceOutbox: db is required');
if (!db) throw new TypeError('createServiceOutbox: db is required');
if (!source || !/^[a-z][a-z0-9_-]*$/.test(source)) throw new TypeError('createServiceOutbox: source is the service id (e.g. "trade")');
if (!source || !/^[a-z][a-z0-9_-]*$/.test(source)) throw new TypeError('createServiceOutbox: source is the service id (e.g. "trade")');
@@ -95,10 +99,26 @@ function createServiceOutbox({
@@ -95,10 +99,26 @@ function createServiceOutbox({
}, { traceparent });
}, { traceparent });
}
}
+ const PRUNE_EVERY_MS = 6 * 60 * 60 * 1000;
+ let pruneTimers = [];
+ const prune = () => {
+ if (!(pruneAfterMs > 0)) return Promise.resolve(0);
+ return outbox.prune(pruneAfterMs).catch((err) => { log.warn(`[outbox:${source}] prune failed: ${err && err.message}`); return 0; });
+ };
return {
return {
- emit, emitIn, moderationAction, outbox, events, enabled,
- start() { if (enabled) outbox.start(); },
- stop: () => outbox.stop(),
+ emit, emitIn, moderationAction, outbox, events, enabled, prune,
+ start() {
+ if (!enabled) return;
+ outbox.start();
+ if (pruneAfterMs > 0 && !pruneTimers.length) {
+ const first = setTimeout(prune, 60 * 1000);
+ const every = setInterval(prune, PRUNE_EVERY_MS);
+ if (first.unref) first.unref();
+ if (every.unref) every.unref();
+ pruneTimers = [first, every];
+ }
+ },
+ stop: () => { for (const t of pruneTimers) { clearTimeout(t); clearInterval(t); } pruneTimers = []; return outbox.stop(); },
async kick() { if (enabled) await outbox.kick(); },
async kick() { if (enabled) await outbox.kick(); },
status: async () => ({ enabled, pending: await outbox.pending(), rejected: await outbox.rejected(), last_error: lastError }),
status: async () => ({ enabled, pending: await outbox.pending(), rejected: await outbox.rejected(), last_error: lastError }),
};
};
@@ -110,6 +110,31 @@ function cases(label, open) {
@@ -110,6 +110,31 @@ function cases(label, open) {
assert.equal(await o2.rejected(), 1);
assert.equal(await o2.rejected(), 1);
} finally { await done(); }
} finally { await done(); }
}],
}],
+ [`${label}: a 429 pauses the whole relay for its retry_after_seconds (else a minute), then it drains`, async () => {
+ const { db, done } = await open();
+ try {
+ const { platform, events } = eventsClient();
+ let t = 2_000_000; let limited = 1; let calls = 0;
+ const tooMany = (s) => Object.assign(new Error('rate_limited'), { status: 429, problem: s == null ? null : { code: 'rate_limited', status: 429, retry_after_seconds: s } });
+ const busy = { prepare: events.prepare, publish: (...a) => { calls++; return limited ? Promise.reject(tooMany(limited === 1 ? 20 : null)) : events.publish(...a); } };
+ const outbox = createPgOutbox(db, { events: busy, batchSize: 2, now: () => t });
+ await db.tx(async (tx) => { for (let i = 0; i < 6; i++) await outbox.enqueue(tx, { event_type: 'live.stream.started', actor, subject: { type: 'stream', id: `r${i}` } }); });
+ assert.deepEqual(await outbox.flush(), { sent: 0, failed: 2, rejected: 0 }, 'the first batch is refused');
+ t += 1000;
+ assert.deepEqual(await outbox.flush(), { sent: 0, failed: 0, rejected: 0 }, 'paused: the next rows are not even tried');
+ assert.equal(calls, 1, 'one refused request, not one per tick');
+ t += 20000;
+ limited = 2; // refused again, with no retry_after_seconds: a minute
+ await outbox.flush();
+ t += 59000;
+ assert.deepEqual(await outbox.flush(), { sent: 0, failed: 0, rejected: 0 }, 'still within the minute');
+ t += 2000; limited = 0;
+ let sent = 0; for (let i = 0; i < 5; i++) sent += (await outbox.flush()).sent;
+ assert.equal(sent, 6, 'every row went out once the window reopened');
+ assert.equal(platform.state.events.length, 6);
+ assert.equal(await outbox.rejected(), 0, 'a 429 never rejects');
+ } finally { await done(); }
+ }],
[`${label}: a token-endpoint refusal is retried, never rejected`, async () => {
[`${label}: a token-endpoint refusal is retried, never rejected`, async () => {
const { db, done } = await open();
const { db, done } = await open();
try {
try {
@@ -75,6 +75,25 @@ run([
@@ -75,6 +75,25 @@ run([
assert.deepEqual([b.payload.target.owner_subject, b.actor], [null, { type: 'service', id: 'trade' }]);
assert.deepEqual([b.payload.target.owner_subject, b.actor], [null, { type: 'service', id: 'trade' }]);
} finally { await db.close(); }
} finally { await db.close(); }
}],
}],
+ ['sent rows older than pruneAfterMs go (prune()), unsent and recent ones stay; false keeps everything', async () => {
+ const db = await open(); const p = platform();
+ try {
+ let t = 1_000_000_000;
+ const out = make(db, p, { now: () => t, pruneAfterMs: 1000 });
+ await db.tx((tx) => out.emitIn(tx, { event_type: 'trade.observation.created', actor, subject }));
+ await out.outbox.flush();
+ t += 500;
+ await db.tx((tx) => out.emitIn(tx, { event_type: 'trade.observation.created', actor, subject: { ...subject, id: 'later' } }));
+ await out.outbox.flush();
+ await db.tx((tx) => out.emitIn(tx, { event_type: 'trade.observation.created', actor, subject: { ...subject, id: 'unsent' } }));
+ t += 800; // the first sent row is 1,300 ms old, the second 800 ms
+ assert.equal(await out.prune(), 1);
+ assert.equal(await db.value(sql`SELECT count(*)::int FROM event_outbox`), 2, 'the recent sent row and the unsent one stay');
+ const keep = make(db, p, { now: () => t + 10_000, pruneAfterMs: false });
+ assert.equal(await keep.prune(), 0, 'false: nothing is pruned');
+ out.start(); await out.stop(); // start() schedules the prune timers and stop() clears them (the test would hang otherwise)
+ } finally { await db.close(); }
+ }],
['a service with its own table name; misuse is a TypeError', async () => {
['a service with its own table name; misuse is a TypeError', async () => {
const db = await open('tips_event_outbox'); const p = platform();
const db = await open('tips_event_outbox'); const p = platform();
try {
try {
@@ -171,6 +171,8 @@ export interface ServiceOutboxOptions {
@@ -171,6 +171,8 @@ export interface ServiceOutboxOptions {
autoDiscover?: boolean;
autoDiscover?: boolean;
/** false: <source>.moderation.action withholds target.owner_subject. */
/** false: <source>.moderation.action withholds target.owner_subject. */
moderationOwnerSubject?: boolean;
moderationOwnerSubject?: boolean;
+ /** Sent rows older than this are pruned every 6 hours while the relay runs (default 7 days); false keeps them. */
+ pruneAfterMs?: number | false;
}
}
export interface ServiceOutbox {
export interface ServiceOutbox {
/** Inside the change's ambient openvibe-sdk/db transaction. */
/** Inside the change's ambient openvibe-sdk/db transaction. */
@@ -181,6 +183,8 @@ export interface ServiceOutbox {
@@ -181,6 +183,8 @@ export interface ServiceOutbox {
outbox: PgOutbox;
outbox: PgOutbox;
events: EventsClient;
events: EventsClient;
enabled: boolean;
enabled: boolean;
+ /** Prunes sent rows older than pruneAfterMs now; resolves with the count (0 when off or on failure). */
+ prune(): Promise<number>;
start(): void;
start(): void;
stop(): Promise<unknown>;
stop(): Promise<unknown>;
kick(): Promise<void>;
kick(): Promise<void>;