openvibers/openvibe.community Commits
Community: the relay's Events worker keeps its place as an opaque cursor only (#56)
Changes
@@@ -41,7 +41,7 @@ little progressive JavaScript for pagination, copy buttons and the upload path.- OpenVibe.Media (screenshot and attachment uploads), OpenVibe.VIP (members-only gates), OpenVibe.Chat(a space's room), OpenVibe.Events (the outbox relay and the Pulse and account subscriptions)- OpenVibe.Live only for old VOD/clip comment imports (it reads and writes no pastes here any more)--- `openvibe-contracts` v0.112.0, `openvibe-sdk` v0.40.1 (events outbox, per-actor limits),++- `openvibe-contracts` v0.112.0, `openvibe-sdk` v0.42.0 (events outbox, per-actor limits),`openvibe-shared` v3.0.0, pinned by release tarball## How it fits the network@@@ -466,7 +466,7 @@ token. Setup, owner steps, limits and operations: [docs/discord-relay.md](docs/dthread members-only) DELETE the Discord messages through the **external message map**(`relay_message_map`, unique both ways; WS-J task 5). Creates are queued by the relay's**Events worker** from Community's own `community.thread.*` / `community.post.*` read back from-- OpenVibe.Events' pull API with a stored cursor (`relay_cursors`; WS-J task 6), or by the forum++ OpenVibe.Events' pull API with a stored opaque cursor (`relay_cursors.position`; WS-J task 6), or by the forumitself when the worker is off (no `EVENTS_URL` / client secret, or `DISCORD_RELAY_EVENTS=off`).Dedupe keys in `relay_deliveries` mean nothing is queued or posted twice.- **In** (`DISCORD_RELAY_INBOUND=on`, `DISCORD_BOT_TOKEN`, `inbound: true` on the mapping): a@@@ -484,7 +484,7 @@ token. Setup, owner steps, limits and operations: [docs/discord-relay.md](docs/dretried with exponential backoff up to `DISCORD_RELAY_MAX_ATTEMPTS`, then the delivery is`failed` (the dead letter); other 4xx fail at once. Mentions are disabled on every message.- **Staff** (admin/global_mod browsers, or services with `community.comment.moderate`):-- `GET /api/v1/relay/status` (queue, Events worker cursor/lag/errors/gaps, gateway state),++ `GET /api/v1/relay/status` (queue, Events worker position/caught-up/errors/gaps, gateway state),`GET /api/v1/relay/deliveries?status=failed`, `POST /api/v1/relay/deliveries/:id/retry|drop`,`GET /api/v1/relay/inbound` and `POST /api/v1/relay/inbound/:id/dismiss` (inbound failures),`GET|POST /api/v1/relay/mappings`, `PUT /api/v1/relay/mappings/:id`.@@@ -825,6 +825,6 @@ release after paste comments moved onto the typed comment threads. [Submissions]<!-- versions:start -->- openvibe-contracts: v0.129.0--- openvibe-sdk: v0.40.1++- openvibe-sdk: v0.42.0- openvibe-shared: v3.0.0<!-- versions:end -->@@@ -8,7 +8,7 @@"port": 4200,"domain": "openvibe.community","contracts": "openvibe-contracts v0.129.0",-- "sdk": "openvibe-sdk v0.40.1",++ "sdk": "openvibe-sdk v0.42.0","shared": "openvibe-shared v3.0.0","capabilities": ["community.paste.create",@@@ -42,7 +42,7 @@ for the inbound half, a bot token.| --- | --- | --- || Queue and sender | `server/relay/discord.js` | `relay_deliveries`: one row per dedupe key, actions `create`, `edit`, `delete`. Creates are sent with `?wait=true`, so Discord answers with the message and its id goes into the map. || External message map (task 5) | `relay_message_map` | Every Discord message the relay knows, both ways (`out`: a thread or reply it sent; `in`: a reply that came from Discord). Unique both ways: a Discord message is one local object, and a local object is one message per mapping. Edits and deletes on either side find their counterpart here. |--| Events worker (task 6) | `server/relay/events-worker.js` | Reads Community's own `community.thread.*` and `community.post.*` from OpenVibe.Events (`GET /api/v1/events`, capability `events.event.read`) and queues the creates. Its cursor (`relay_cursors`) moves in the same transaction as the deliveries a page queued; a replay queues nothing twice (dedupe keys), and nothing already in the map is sent twice. The first start begins at Events' head: older threads are never announced. |++| Events worker (task 6) | `server/relay/events-worker.js` | Reads Community's own `community.thread.*` and `community.post.*` from OpenVibe.Events (`GET /api/v1/events`, capability `events.event.read`) and queues the creates. Its cursor (`relay_cursors.position`, Events' opaque cursor, never a number) moves in the same transaction as the deliveries a page queued; a replay queues nothing twice (dedupe keys), and nothing already in the map is sent twice. The first start begins at Events' head: older threads are never announced. || Gateway | `server/relay/discord-gateway.js` | Discord gateway v10 over Node 22's built-in WebSocket: IDENTIFY/RESUME, heartbeats (a missing ACK means a zombied connection: resume), RECONNECT, INVALID_SESSION, backoff. Intents `GUILD_MESSAGES` and `MESSAGE_CONTENT`. || Inbound | `server/relay/inbound.js` | `MESSAGE_CREATE`, `MESSAGE_UPDATE`, `MESSAGE_DELETE`, `MESSAGE_DELETE_BULK` for mapped channels only. || Staff API | `server/relay/api.js` | `/api/v1/relay/*`, below. |@@@ -74,7 +74,7 @@ Staff only: an admin/global_mod browser session, or a service holding `community| Route | What || --- | --- |--| `GET /api/v1/relay/status` | The queue by status (`failed` = dead letters), the Events worker (cursor, latest seq, lag, last error, last gap), the gateway (state, reconnects, last error) and open inbound failures |++| `GET /api/v1/relay/status` | The queue by status (`failed` = dead letters), the Events worker (position, whether the last read reached the head, last error, last gap), the gateway (state, reconnects, last error) and open inbound failures || `GET /api/v1/relay/deliveries?status=&action=&limit=` | Deliveries; `status` pending, delivered, failed, dropped or skipped; `action` create, edit or delete || `POST /api/v1/relay/deliveries/:id/retry` | Queue it again now, with a fresh attempt budget || `POST /api/v1/relay/deliveries/:id/drop` | Give up on a pending or failed one (kept as `dropped`) |@@@ -128,7 +128,7 @@ Staff only: an admin/global_mod browser session, or a service holding `communityreply to it on Discord (Reply, or in a thread on the message) → the reply appears in thethread with a Discord badge; edit and delete it on Discord → the post follows; reply and editon the site → Discord follows; delete the thread → its messages leave Discord. Watch-- `GET /api/v1/relay/status` (no dead letters, lag 0, gateway `ready`).++ `GET /api/v1/relay/status` (no dead letters, the Events worker `caught_up`, gateway `ready`).**Off again**: unset `DISCORD_RELAY_ENABLED` and restart (nothing is sent or read; the tables stay),or per mapping `PUT /api/v1/relay/mappings/:id { "enabled": false }` (deletes still go out) ornnew file mode 100644@@@ -0,0 +1,5 @@++-- phase: expand++-- The Discord relay's Events worker keeps its place only as Events' opaque cursor (`position`; ADR-042 decision 7,++-- plan T7: Events no longer answers numeric positions). It stops writing the numeric `cursor` and `latest_seq`, so++-- `cursor` may be NULL from now on; a later contract migration (after this one) drops both columns.++ALTER TABLE relay_cursors ALTER COLUMN cursor DROP NOT NULL;@@@ -18,7 +18,7 @@"iovalkey": "^0.4.0","multer": "^2.4.0","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.129.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.40.1",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.42.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v3.0.0","pg": "^8.23.0"},@@@ -933,9 +933,9 @@}},"node_modules/openvibe-sdk": {-- "version": "0.40.1",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.40.1",-- "integrity": "sha512-QqJKJu5cujCD+TVY/u+0a1K23wAJq4G9ZIuHjfNpwxccQw+IAHiFqEyyFbIXZHaNE4stDW3pN/+9JZQ9BQP1yw==",++ "version": "0.42.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.42.0",++ "integrity": "sha512-MpUPLW8iBviKYcnjvNffQ6pshDCMT9ETj70hYoh6VFtwu1rxCs2DwSs8B84yPLRIyTKNxiMFchTd7tLnOUZAtg==","license": "MIT","engines": {"node": ">=20"@@@ -29,7 +29,7 @@"iovalkey": "^0.4.0","multer": "^2.4.0","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.129.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.40.1",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.42.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v3.0.0","pg": "^8.23.0"},@@@ -55,7 +55,7 @@ function createCommunityReadiness({ db, auth, config, relay = null, release = nuif (relay && relay.enabled && body.checks.db.status === 'ok') {// The queue by status (failed = dead letters), and whether the Events worker and the inbound gateway run.const st = await relay.status();-- const brief = (x) => (x.enabled ? { enabled: true, state: x.state || (x.running ? 'running' : 'stopped'), last_error: x.last_error || null, ...(x.lag != null ? { lag: x.lag } : {}) } : { enabled: false, reason: x.reason });++ const brief = (x) => (x.enabled ? { enabled: true, state: x.state || (x.running ? 'running' : 'stopped'), last_error: x.last_error || null, ...(x.caught_up != null ? { caught_up: x.caught_up } : {}) } : { enabled: false, reason: x.reason });out.discord_relay = { enabled: true, deliveries: st.deliveries, creates_from: st.creates_from, events_worker: brief(st.events_worker), inbound: brief(st.inbound), inbound_failures: st.inbound_failures };}return out;@@@ -9,10 +9,9 @@* - Cursor: the relay_cursors row 'discord-relay:events', moved in the same transaction as the* deliveries its page queued. A crash replays at most the page in progress, and the dedupe keys* make that replay queue nothing twice. `position` is Events' opaque cursor (the page's next_cursor),-- * the place the worker reads on from; `cursor` is the numeric seq, still written while Events sends it-- * (a worker of the previous release, and the lag shown to staff). A row with no position yet (written-- * by the previous release) is read on from its numeric cursor once, and has a position from then on.-- * - First start (no cursor yet): it begins at Events' head (latest_cursor). Threads from before are never++ * the only place the worker reads on from; it is never parsed. Staff see whether the last read reached++ * the head (`caught_up`), not a numeric lag.++ * - First start (no position yet): it begins at Events' head (latest_cursor). Threads from before are never* announced.* - Events carry ids, never content: the worker reads the thread or post as it is now and the relay's* rules decide (public spaces only, never members-only, never origin 'discord'). Redacted events@@@ -49,19 +48,21 @@ function createRelayEventsWorker({ db, relay, events = null, eventsUrl = null, cclient = createEventsClient(core, { source: 'community' });}const on = !reason;-- const stats = { seen: 0, queued: 0, pages: 0, last_error: null, last_error_at: null, last_ok_at: null, last_gap: null, started_at_seq: null };++ const stats = { seen: 0, queued: 0, pages: 0, last_error: null, last_error_at: null, last_ok_at: null, last_gap: null, started_at_head: null, head: null };let timer = null;let running = null;let stopped = false;const iso = () => new Date(now()).toISOString();-- const readCursor = async () => await db.prepare('SELECT cursor, position, latest_seq FROM relay_cursors WHERE name = ?').get(CURSOR) || null;-- async function writeCursor(cursor, latest, position) {-- await db.prepare(`INSERT INTO relay_cursors (name, cursor, latest_seq, position, updated_at) VALUES (?, ?, ?, ?, ov_now())-- ON CONFLICT(name) DO UPDATE SET cursor = excluded.cursor, latest_seq = excluded.latest_seq, position = excluded.position, updated_at = ov_now()`)-- .run(CURSOR, cursor, latest == null ? null : latest, position || null);++ const readPosition = async () => {++ const row = await db.prepare('SELECT position FROM relay_cursors WHERE name = ?').get(CURSOR);++ return row && row.position ? row.position : null;++ };++ async function writePosition(position) {++ await db.prepare(`INSERT INTO relay_cursors (name, position, updated_at) VALUES (?, ?, ov_now())++ ON CONFLICT(name) DO UPDATE SET position = excluded.position, updated_at = ov_now()`)++ .run(CURSOR, position);}-- const num = (v) => (Number.isFinite(Number(v)) && v !== null && v !== undefined ? Number(v) : null);/** One event → deliveries queued (0 when it is not for the relay). */async function handle(ev) {@@@ -90,43 +91,36 @@ function createRelayEventsWorker({ db, relay, events = null, eventsUrl = null, crunning = (async () => {let queuedNow = 0;try {-- let cur = await readCursor();-- if (!cur) {-- // First start: Events' head, as an opaque cursor (the numeric seq beside it while Events sends one).++ let position = await readPosition();++ if (!position) {++ // First start: Events' head, as an opaque cursor.const head = await client.pull({ topic: TOPICS, limit: 1 });-- const latest = num(head.latest_seq) || 0;-- await writeCursor(latest, latest, head.latest_cursor || null);-- stats.started_at_seq = latest;-- console.log(`[Relay] Events worker starts at the head (seq ${latest}; earlier threads are not announced)`);-- cur = { cursor: latest, position: head.latest_cursor || null, latest_seq: latest };++ if (!head.latest_cursor) throw new Error('Events answered no latest_cursor');++ position = head.latest_cursor;++ await writePosition(position);++ stats.started_at_head = iso();++ stats.head = position;++ console.log('[Relay] Events worker starts at the head (earlier threads are not announced)');}-- let cursor = num(cur.cursor) || 0;-- let position = cur.position || null;for (let pages = 0; pages < MAX_PAGES_PER_TICK && !stopped; pages++) {-- // Read on from the opaque position; a row from the previous release has only the numeric cursor.-- const page = position-- ? await client.pull({ topic: TOPICS, after: position, limit: pageLimit })-- : await client.pull({ topic: TOPICS, afterSeq: cursor, limit: pageLimit });++ const page = await client.pull({ topic: TOPICS, after: position, limit: pageLimit });if (stopped) break;if (page.gap) {stats.last_gap = { from_seq: page.gap.from_seq, to_seq: page.gap.to_seq, at: iso() };console.warn(`[Relay] Events gap: seq ${page.gap.from_seq}..${page.gap.to_seq} were pruned before the relay read them; threads and replies in it are not relayed`);}const nextPosition = page.next_cursor || position;-- const next = num(page.next_after_seq);-- const nextCursor = next != null && next > cursor ? next : cursor;-- const latest = num(page.latest_seq);const items = page.events || [];queuedNow += await db.tx(async () => {let n = 0;for (const item of items) n += await handle(item && item.event);-- await writeCursor(nextCursor, latest, nextPosition);++ await writePosition(nextPosition);return n;});stats.pages++;-- const moved = nextPosition !== position || nextCursor !== cursor;++ if (page.latest_cursor) stats.head = page.latest_cursor;++ const moved = nextPosition !== position;position = nextPosition;-- cursor = nextCursor;// A short page means the read reached the head (next_cursor already moved past what did not match).if (!moved || items.length < pageLimit) break;}@@@ -164,12 +158,13 @@ function createRelayEventsWorker({ db, relay, events = null, eventsUrl = null, casync function status() {if (!on) return { enabled: false, reason };-- const cur = await readCursor();++ const position = await readPosition();return {enabled: true, running: !!timer, topics: TOPICS,-- cursor: cur ? cur.cursor : null, position: cur ? cur.position || null : null, latest_seq: cur ? cur.latest_seq : null,-- lag: cur && cur.latest_seq != null ? Math.max(cur.latest_seq - cur.cursor, 0) : null,-- events_seen: stats.seen, deliveries_queued: stats.queued, started_at_seq: stats.started_at_seq,++ position,++ // Whether the last read reached Events' head (both opaque cursors of one epoch: equal strings, equal place).++ caught_up: position && stats.head ? position === stats.head : null,++ events_seen: stats.seen, deliveries_queued: stats.queued, started_at_head: stats.started_at_head,last_ok_at: stats.last_ok_at, last_error: stats.last_error, last_error_at: stats.last_error_at, last_gap: stats.last_gap,};}@@@ -4,8 +4,8 @@* the pull route the relay's Events worker reads.** POST /api/v1/events one envelope or { events: [...] } → { event_id, seq, duplicate } | { results }-- * GET /api/v1/events?topic=a.*,b.*&after=<cursor>|after_seq=&limit=-- * → { events: [{ seq, cursor, event }], next_after_seq, next_cursor, latest_seq, latest_cursor, gap? }++ * GET /api/v1/events?topic=a.*,b.*[&after=<cursor>]&limit=++ * → { events: [{ seq, cursor, event }], next_cursor, latest_cursor, gap? } (after_seq is a 400, as in Events)* Cursors are opaque to the worker; here, as in Events, `c1.<epoch>.<base64url(seq)>` (one epoch, 0).** Topic patterns follow Events (`*` is one or more whole segments). Every call needs a Bearer token.@@@ -47,8 +47,9 @@ function startEvents() {}if (u.pathname === '/api/v1/events' && req.method === 'GET') {const patterns = String(u.searchParams.get('topic') || '*').split(',').map(matcher);++ if (u.searchParams.has('after_seq')) return json(400, { type: 'about:blank', code: 'events.bad_request', status: 400, detail: 'after_seq is retired: pass after=<cursor>' });const cur = u.searchParams.get('after');-- const after = cur ? decode(cur) : Number(u.searchParams.get('after_seq') || 0);++ const after = cur ? decode(cur) : 0;if (!Number.isFinite(after)) return json(400, { type: 'about:blank', code: 'events.bad_request', status: 400, detail: 'after must be an opaque cursor' });const limit = Number(u.searchParams.get('limit') || 100);const out = {};@@@ -63,9 +64,7 @@ function startEvents() {if (rows.length >= limit) break;}out.events = rows.map((r) => ({ ...r, cursor: encode(r.seq) }));-- out.next_after_seq = cursor;out.next_cursor = encode(cursor);-- out.latest_seq = log.length;out.latest_cursor = encode(log.length);return json(200, out);}@@@ -70,10 +70,12 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.creaawait worker.tick();assert.strictEqual((await relay.status()).creates_from, 'events');const st = await worker.status();-- assert.deepStrictEqual([st.enabled, st.cursor, st.latest_seq, st.started_at_seq, st.lag], [true, 3, 3, 3, 0]);++ assert.deepStrictEqual([st.enabled, st.position, st.caught_up], [true, events.encode(3), true]);++ assert.ok(st.started_at_head, 'it says it started at the head');++ assert.ok(!('cursor' in st) && !('lag' in st) && !('latest_seq' in st), 'no numeric position in the status');assert.strictEqual((await deliveries()).length, 0);-- assert.deepStrictEqual(await db.prepare('SELECT name, cursor, position FROM relay_cursors').all(), [{ name: CURSOR, cursor: 3, position: events.encode(3) }], 'the head, stored as the opaque latest_cursor');-- assert.ok(events.calls.every((c) => !/[?&]after_seq=[1-9]/.test(c.path)), 'no numeric position is ever sent once a cursor is stored');++ assert.deepStrictEqual(await db.prepare('SELECT name, cursor, position FROM relay_cursors').all(), [{ name: CURSOR, cursor: null, position: events.encode(3) }], 'the head, stored as the opaque latest_cursor and nothing numeric');++ assert.ok(events.calls.every((c) => !/after_seq/.test(c.path)), 'no numeric position is ever sent');before = events.calls.length;});@@@ -96,7 +98,7 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.creaawait settle(); await relay.drain(); await relay.drain();assert.deepStrictEqual(hook.hits.map((h) => h.method), ['POST', 'POST']);assert.match(hook.hits[1].body.content, /replied to \*\*From the stream\*\*/);-- assert.deepStrictEqual([(await worker.status()).cursor, (await worker.status()).lag], [6, 0]);++ assert.deepStrictEqual([(await worker.status()).position, (await worker.status()).caught_up], [events.encode(6), true]);});await check('a replay (cursor moved back) queues nothing twice and posts nothing twice', async () => {@@@ -104,7 +106,7 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.creaassert.deepStrictEqual(await worker.tick(), { queued: 0 });await relay.drain();assert.strictEqual(hook.hits.length, 2);-- assert.strictEqual((await worker.status()).cursor, 6);++ assert.strictEqual((await worker.status()).position, events.encode(6));});await check('the cursor survives a restart: a new worker carries on from it (no second "start at head")', async () => {@@@ -112,7 +114,7 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.creaevents.add(threadEvent(t2));const again = createRelayEventsWorker({ db, relay, events: client, pollMs: 3_600_000 });assert.deepStrictEqual(await again.tick(), { queued: 1 });-- assert.strictEqual((await again.status()).started_at_seq, null);++ assert.strictEqual((await again.status()).started_at_head, null);assert.strictEqual((await deliveries()).pop().thread_id, t2.id);});@@@ -133,25 +135,25 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.creaawait check('Events down: the error shows, the cursor stays, and it catches up once Events answers', async () => {const t4 = await newThread('After the outage');events.add(threadEvent(t4));-- const cursor = (await worker.status()).cursor;++ const position = (await worker.status()).position;events.state.failNext = 1;assert.deepStrictEqual(await worker.tick(), { queued: 0 });const st = await worker.status();assert.ok(st.last_error, 'the error is shown');-- assert.strictEqual(st.cursor, cursor);-- assert.strictEqual(st.lag, st.latest_seq - cursor);++ assert.strictEqual(st.position, position, 'the cursor stays');assert.deepStrictEqual(await worker.tick(), { queued: 1 });assert.strictEqual((await worker.status()).last_error, null);});-- await check('a row from the previous release (numeric cursor, no position) reads on from it once, then by cursor', async () => {++ await check('a row with no position (a numeric one only) starts at the head: a number is never read as a place', async () => {await db.prepare('UPDATE relay_cursors SET cursor = 3, position = NULL WHERE name = ?').run(CURSOR);++ const t5 = await newThread('Before the restart at the head');++ events.add(threadEvent(t5));const from = events.calls.length;-- assert.deepStrictEqual(await worker.tick(), { queued: 0 }, 'what it read again was already queued');-- const sent = events.calls.slice(from).map((c) => c.path);-- assert.ok(sent.some((p) => /[?&]after_seq=3\b/.test(p)), 'the first read uses the numeric cursor it had');-- const row = await db.prepare('SELECT cursor, position FROM relay_cursors WHERE name = ?').get(CURSOR);-- assert.strictEqual(row.position, events.encode(events.log.length), 'and from then on the opaque cursor');++ assert.deepStrictEqual(await worker.tick(), { queued: 0 }, 'nothing from before the head is announced');++ assert.ok(events.calls.slice(from).every((c) => !/after_seq/.test(c.path)), 'never a numeric position');++ const row = await db.prepare('SELECT position FROM relay_cursors WHERE name = ?').get(CURSOR);++ assert.strictEqual(row.position, events.encode(events.log.length), 'the head, as an opaque cursor');const again = events.calls.length;await worker.tick();assert.ok(events.calls.slice(again).every((c) => /[?&]after=c1\./.test(c.path)), 'later reads send the cursor');@@@ -162,7 +164,7 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.creaevents.state.prunedThrough = 5;await worker.tick();assert.deepStrictEqual([(await worker.status()).last_gap.from_seq, (await worker.status()).last_gap.to_seq], [2, 5]);-- assert.strictEqual((await worker.status()).cursor, events.log.length);++ assert.strictEqual((await worker.status()).position, events.encode(events.log.length));events.state.prunedThrough = 0;assert.strictEqual((await relay.status()).events_worker.last_gap.to_seq, 5);});@@@ -186,7 +188,7 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.creaconst call = (p, { method = 'GET', cookie, json } = {}) => t.get(p, { method, headers: json !== undefined ? { 'content-type': 'application/json' } : {}, body: json !== undefined ? JSON.stringify(json) : undefined, cookies: cookie ? [`ov_token=${cookie}`] : [] });const waitFor = async (fn, what) => { for (let i = 0; i < 200; i++) { if (await fn()) return; await new Promise((r) => setTimeout(r, 20)); } assert.fail(`timed out waiting for ${what}`); };assert.strictEqual((await call('/api/v1/relay/mappings', { method: 'POST', cookie: adminJwt, json: { space: 'general', webhook_url_ref: 'DISCORD_WEBHOOK_GENERAL' } })).status, 201);-- await waitFor(async () => (await t.app.locals.relay.status()).events_worker.cursor != null, 'the worker\'s first read');++ await waitFor(async () => (await t.app.locals.relay.status()).events_worker.position != null, 'the worker\'s first read');const st = (await call('/api/v1/relay/status', { cookie: adminJwt })).json();assert.deepStrictEqual([st.creates_from, st.events_worker.enabled], ['events', true]);const th = await call('/api/v1/spaces/general/threads', { method: 'POST', cookie: samJwt, json: { title: 'Round trip', body: 'through Events' } });
@@ -41,7 +41,7 @@ little progressive JavaScript for pagination, copy buttons and the upload path.
@@ -41,7 +41,7 @@ little progressive JavaScript for pagination, copy buttons and the upload path.
- OpenVibe.Media (screenshot and attachment uploads), OpenVibe.VIP (members-only gates), OpenVibe.Chat
- OpenVibe.Media (screenshot and attachment uploads), OpenVibe.VIP (members-only gates), OpenVibe.Chat
(a space's room), OpenVibe.Events (the outbox relay and the Pulse and account subscriptions)
(a space's room), OpenVibe.Events (the outbox relay and the Pulse and account subscriptions)
- OpenVibe.Live only for old VOD/clip comment imports (it reads and writes no pastes here any more)
- OpenVibe.Live only for old VOD/clip comment imports (it reads and writes no pastes here any more)
-- `openvibe-contracts` v0.112.0, `openvibe-sdk` v0.40.1 (events outbox, per-actor limits),
+- `openvibe-contracts` v0.112.0, `openvibe-sdk` v0.42.0 (events outbox, per-actor limits),
`openvibe-shared` v3.0.0, pinned by release tarball
`openvibe-shared` v3.0.0, pinned by release tarball
## How it fits the network
## How it fits the network
@@ -466,7 +466,7 @@ token. Setup, owner steps, limits and operations: [docs/discord-relay.md](docs/d
@@ -466,7 +466,7 @@ token. Setup, owner steps, limits and operations: [docs/discord-relay.md](docs/d
thread members-only) DELETE the Discord messages through the **external message map**
thread members-only) DELETE the Discord messages through the **external message map**
(`relay_message_map`, unique both ways; WS-J task 5). Creates are queued by the relay's
(`relay_message_map`, unique both ways; WS-J task 5). Creates are queued by the relay's
**Events worker** from Community's own `community.thread.*` / `community.post.*` read back from
**Events worker** from Community's own `community.thread.*` / `community.post.*` read back from
- OpenVibe.Events' pull API with a stored cursor (`relay_cursors`; WS-J task 6), or by the forum
+ OpenVibe.Events' pull API with a stored opaque cursor (`relay_cursors.position`; WS-J task 6), or by the forum
itself when the worker is off (no `EVENTS_URL` / client secret, or `DISCORD_RELAY_EVENTS=off`).
itself when the worker is off (no `EVENTS_URL` / client secret, or `DISCORD_RELAY_EVENTS=off`).
Dedupe keys in `relay_deliveries` mean nothing is queued or posted twice.
Dedupe keys in `relay_deliveries` mean nothing is queued or posted twice.
- **In** (`DISCORD_RELAY_INBOUND=on`, `DISCORD_BOT_TOKEN`, `inbound: true` on the mapping): a
- **In** (`DISCORD_RELAY_INBOUND=on`, `DISCORD_BOT_TOKEN`, `inbound: true` on the mapping): a
@@ -484,7 +484,7 @@ token. Setup, owner steps, limits and operations: [docs/discord-relay.md](docs/d
@@ -484,7 +484,7 @@ token. Setup, owner steps, limits and operations: [docs/discord-relay.md](docs/d
retried with exponential backoff up to `DISCORD_RELAY_MAX_ATTEMPTS`, then the delivery is
retried with exponential backoff up to `DISCORD_RELAY_MAX_ATTEMPTS`, then the delivery is
`failed` (the dead letter); other 4xx fail at once. Mentions are disabled on every message.
`failed` (the dead letter); other 4xx fail at once. Mentions are disabled on every message.
- **Staff** (admin/global_mod browsers, or services with `community.comment.moderate`):
- **Staff** (admin/global_mod browsers, or services with `community.comment.moderate`):
- `GET /api/v1/relay/status` (queue, Events worker cursor/lag/errors/gaps, gateway state),
+ `GET /api/v1/relay/status` (queue, Events worker position/caught-up/errors/gaps, gateway state),
`GET /api/v1/relay/deliveries?status=failed`, `POST /api/v1/relay/deliveries/:id/retry|drop`,
`GET /api/v1/relay/deliveries?status=failed`, `POST /api/v1/relay/deliveries/:id/retry|drop`,
`GET /api/v1/relay/inbound` and `POST /api/v1/relay/inbound/:id/dismiss` (inbound failures),
`GET /api/v1/relay/inbound` and `POST /api/v1/relay/inbound/:id/dismiss` (inbound failures),
`GET|POST /api/v1/relay/mappings`, `PUT /api/v1/relay/mappings/:id`.
`GET|POST /api/v1/relay/mappings`, `PUT /api/v1/relay/mappings/:id`.
@@ -825,6 +825,6 @@ release after paste comments moved onto the typed comment threads. [Submissions]
@@ -825,6 +825,6 @@ release after paste comments moved onto the typed comment threads. [Submissions]
<!-- versions:start -->
<!-- versions:start -->
- openvibe-contracts: v0.129.0
- openvibe-contracts: v0.129.0
-- openvibe-sdk: v0.40.1
+- openvibe-sdk: v0.42.0
- openvibe-shared: v3.0.0
- openvibe-shared: v3.0.0
<!-- versions:end -->
<!-- versions:end -->
@@ -8,7 +8,7 @@
@@ -8,7 +8,7 @@
"port": 4200,
"port": 4200,
"domain": "openvibe.community",
"domain": "openvibe.community",
"contracts": "openvibe-contracts v0.129.0",
"contracts": "openvibe-contracts v0.129.0",
- "sdk": "openvibe-sdk v0.40.1",
+ "sdk": "openvibe-sdk v0.42.0",
"shared": "openvibe-shared v3.0.0",
"shared": "openvibe-shared v3.0.0",
"capabilities": [
"capabilities": [
"community.paste.create",
"community.paste.create",
@@ -42,7 +42,7 @@ for the inbound half, a bot token.
@@ -42,7 +42,7 @@ for the inbound half, a bot token.
| --- | --- | --- |
| --- | --- | --- |
| Queue and sender | `server/relay/discord.js` | `relay_deliveries`: one row per dedupe key, actions `create`, `edit`, `delete`. Creates are sent with `?wait=true`, so Discord answers with the message and its id goes into the map. |
| Queue and sender | `server/relay/discord.js` | `relay_deliveries`: one row per dedupe key, actions `create`, `edit`, `delete`. Creates are sent with `?wait=true`, so Discord answers with the message and its id goes into the map. |
| External message map (task 5) | `relay_message_map` | Every Discord message the relay knows, both ways (`out`: a thread or reply it sent; `in`: a reply that came from Discord). Unique both ways: a Discord message is one local object, and a local object is one message per mapping. Edits and deletes on either side find their counterpart here. |
| External message map (task 5) | `relay_message_map` | Every Discord message the relay knows, both ways (`out`: a thread or reply it sent; `in`: a reply that came from Discord). Unique both ways: a Discord message is one local object, and a local object is one message per mapping. Edits and deletes on either side find their counterpart here. |
-| Events worker (task 6) | `server/relay/events-worker.js` | Reads Community's own `community.thread.*` and `community.post.*` from OpenVibe.Events (`GET /api/v1/events`, capability `events.event.read`) and queues the creates. Its cursor (`relay_cursors`) moves in the same transaction as the deliveries a page queued; a replay queues nothing twice (dedupe keys), and nothing already in the map is sent twice. The first start begins at Events' head: older threads are never announced. |
+| Events worker (task 6) | `server/relay/events-worker.js` | Reads Community's own `community.thread.*` and `community.post.*` from OpenVibe.Events (`GET /api/v1/events`, capability `events.event.read`) and queues the creates. Its cursor (`relay_cursors.position`, Events' opaque cursor, never a number) moves in the same transaction as the deliveries a page queued; a replay queues nothing twice (dedupe keys), and nothing already in the map is sent twice. The first start begins at Events' head: older threads are never announced. |
| Gateway | `server/relay/discord-gateway.js` | Discord gateway v10 over Node 22's built-in WebSocket: IDENTIFY/RESUME, heartbeats (a missing ACK means a zombied connection: resume), RECONNECT, INVALID_SESSION, backoff. Intents `GUILD_MESSAGES` and `MESSAGE_CONTENT`. |
| Gateway | `server/relay/discord-gateway.js` | Discord gateway v10 over Node 22's built-in WebSocket: IDENTIFY/RESUME, heartbeats (a missing ACK means a zombied connection: resume), RECONNECT, INVALID_SESSION, backoff. Intents `GUILD_MESSAGES` and `MESSAGE_CONTENT`. |
| Inbound | `server/relay/inbound.js` | `MESSAGE_CREATE`, `MESSAGE_UPDATE`, `MESSAGE_DELETE`, `MESSAGE_DELETE_BULK` for mapped channels only. |
| Inbound | `server/relay/inbound.js` | `MESSAGE_CREATE`, `MESSAGE_UPDATE`, `MESSAGE_DELETE`, `MESSAGE_DELETE_BULK` for mapped channels only. |
| Staff API | `server/relay/api.js` | `/api/v1/relay/*`, below. |
| Staff API | `server/relay/api.js` | `/api/v1/relay/*`, below. |
@@ -74,7 +74,7 @@ Staff only: an admin/global_mod browser session, or a service holding `community
@@ -74,7 +74,7 @@ Staff only: an admin/global_mod browser session, or a service holding `community
| Route | What |
| Route | What |
| --- | --- |
| --- | --- |
-| `GET /api/v1/relay/status` | The queue by status (`failed` = dead letters), the Events worker (cursor, latest seq, lag, last error, last gap), the gateway (state, reconnects, last error) and open inbound failures |
+| `GET /api/v1/relay/status` | The queue by status (`failed` = dead letters), the Events worker (position, whether the last read reached the head, last error, last gap), the gateway (state, reconnects, last error) and open inbound failures |
| `GET /api/v1/relay/deliveries?status=&action=&limit=` | Deliveries; `status` pending, delivered, failed, dropped or skipped; `action` create, edit or delete |
| `GET /api/v1/relay/deliveries?status=&action=&limit=` | Deliveries; `status` pending, delivered, failed, dropped or skipped; `action` create, edit or delete |
| `POST /api/v1/relay/deliveries/:id/retry` | Queue it again now, with a fresh attempt budget |
| `POST /api/v1/relay/deliveries/:id/retry` | Queue it again now, with a fresh attempt budget |
| `POST /api/v1/relay/deliveries/:id/drop` | Give up on a pending or failed one (kept as `dropped`) |
| `POST /api/v1/relay/deliveries/:id/drop` | Give up on a pending or failed one (kept as `dropped`) |
@@ -128,7 +128,7 @@ Staff only: an admin/global_mod browser session, or a service holding `community
@@ -128,7 +128,7 @@ Staff only: an admin/global_mod browser session, or a service holding `community
reply to it on Discord (Reply, or in a thread on the message) → the reply appears in the
reply to it on Discord (Reply, or in a thread on the message) → the reply appears in the
thread with a Discord badge; edit and delete it on Discord → the post follows; reply and edit
thread with a Discord badge; edit and delete it on Discord → the post follows; reply and edit
on the site → Discord follows; delete the thread → its messages leave Discord. Watch
on the site → Discord follows; delete the thread → its messages leave Discord. Watch
- `GET /api/v1/relay/status` (no dead letters, lag 0, gateway `ready`).
+ `GET /api/v1/relay/status` (no dead letters, the Events worker `caught_up`, gateway `ready`).
**Off again**: unset `DISCORD_RELAY_ENABLED` and restart (nothing is sent or read; the tables stay),
**Off again**: unset `DISCORD_RELAY_ENABLED` and restart (nothing is sent or read; the tables stay),
or per mapping `PUT /api/v1/relay/mappings/:id { "enabled": false }` (deletes still go out) or
or per mapping `PUT /api/v1/relay/mappings/:id { "enabled": false }` (deletes still go out) or
new file mode 100644
new file mode 100644
@@ -0,0 +1,5 @@
@@ -0,0 +1,5 @@
+-- phase: expand
+-- The Discord relay's Events worker keeps its place only as Events' opaque cursor (`position`; ADR-042 decision 7,
+-- plan T7: Events no longer answers numeric positions). It stops writing the numeric `cursor` and `latest_seq`, so
+-- `cursor` may be NULL from now on; a later contract migration (after this one) drops both columns.
+ALTER TABLE relay_cursors ALTER COLUMN cursor DROP NOT NULL;
@@ -18,7 +18,7 @@
@@ -18,7 +18,7 @@
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"multer": "^2.4.0",
"multer": "^2.4.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.129.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.129.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.40.1",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.42.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v3.0.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v3.0.0",
"pg": "^8.23.0"
"pg": "^8.23.0"
},
},
@@ -933,9 +933,9 @@
@@ -933,9 +933,9 @@
}
}
},
},
"node_modules/openvibe-sdk": {
"node_modules/openvibe-sdk": {
- "version": "0.40.1",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.40.1",
- "integrity": "sha512-QqJKJu5cujCD+TVY/u+0a1K23wAJq4G9ZIuHjfNpwxccQw+IAHiFqEyyFbIXZHaNE4stDW3pN/+9JZQ9BQP1yw==",
+ "version": "0.42.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.42.0",
+ "integrity": "sha512-MpUPLW8iBviKYcnjvNffQ6pshDCMT9ETj70hYoh6VFtwu1rxCs2DwSs8B84yPLRIyTKNxiMFchTd7tLnOUZAtg==",
"license": "MIT",
"license": "MIT",
"engines": {
"engines": {
"node": ">=20"
"node": ">=20"
@@ -29,7 +29,7 @@
@@ -29,7 +29,7 @@
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"multer": "^2.4.0",
"multer": "^2.4.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.129.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.129.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.40.1",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.42.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v3.0.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v3.0.0",
"pg": "^8.23.0"
"pg": "^8.23.0"
},
},
@@ -55,7 +55,7 @@ function createCommunityReadiness({ db, auth, config, relay = null, release = nu
@@ -55,7 +55,7 @@ function createCommunityReadiness({ db, auth, config, relay = null, release = nu
if (relay && relay.enabled && body.checks.db.status === 'ok') {
if (relay && relay.enabled && body.checks.db.status === 'ok') {
// The queue by status (failed = dead letters), and whether the Events worker and the inbound gateway run.
// The queue by status (failed = dead letters), and whether the Events worker and the inbound gateway run.
const st = await relay.status();
const st = await relay.status();
- const brief = (x) => (x.enabled ? { enabled: true, state: x.state || (x.running ? 'running' : 'stopped'), last_error: x.last_error || null, ...(x.lag != null ? { lag: x.lag } : {}) } : { enabled: false, reason: x.reason });
+ const brief = (x) => (x.enabled ? { enabled: true, state: x.state || (x.running ? 'running' : 'stopped'), last_error: x.last_error || null, ...(x.caught_up != null ? { caught_up: x.caught_up } : {}) } : { enabled: false, reason: x.reason });
out.discord_relay = { enabled: true, deliveries: st.deliveries, creates_from: st.creates_from, events_worker: brief(st.events_worker), inbound: brief(st.inbound), inbound_failures: st.inbound_failures };
out.discord_relay = { enabled: true, deliveries: st.deliveries, creates_from: st.creates_from, events_worker: brief(st.events_worker), inbound: brief(st.inbound), inbound_failures: st.inbound_failures };
}
}
return out;
return out;
@@ -9,10 +9,9 @@
@@ -9,10 +9,9 @@
* - Cursor: the relay_cursors row 'discord-relay:events', moved in the same transaction as the
* - Cursor: the relay_cursors row 'discord-relay:events', moved in the same transaction as the
* deliveries its page queued. A crash replays at most the page in progress, and the dedupe keys
* deliveries its page queued. A crash replays at most the page in progress, and the dedupe keys
* make that replay queue nothing twice. `position` is Events' opaque cursor (the page's next_cursor),
* make that replay queue nothing twice. `position` is Events' opaque cursor (the page's next_cursor),
- * the place the worker reads on from; `cursor` is the numeric seq, still written while Events sends it
- * (a worker of the previous release, and the lag shown to staff). A row with no position yet (written
- * by the previous release) is read on from its numeric cursor once, and has a position from then on.
- * - First start (no cursor yet): it begins at Events' head (latest_cursor). Threads from before are never
+ * the only place the worker reads on from; it is never parsed. Staff see whether the last read reached
+ * the head (`caught_up`), not a numeric lag.
+ * - First start (no position yet): it begins at Events' head (latest_cursor). Threads from before are never
* announced.
* announced.
* - Events carry ids, never content: the worker reads the thread or post as it is now and the relay's
* - Events carry ids, never content: the worker reads the thread or post as it is now and the relay's
* rules decide (public spaces only, never members-only, never origin 'discord'). Redacted events
* rules decide (public spaces only, never members-only, never origin 'discord'). Redacted events
@@ -49,19 +48,21 @@ function createRelayEventsWorker({ db, relay, events = null, eventsUrl = null, c
@@ -49,19 +48,21 @@ function createRelayEventsWorker({ db, relay, events = null, eventsUrl = null, c
client = createEventsClient(core, { source: 'community' });
client = createEventsClient(core, { source: 'community' });
}
}
const on = !reason;
const on = !reason;
- const stats = { seen: 0, queued: 0, pages: 0, last_error: null, last_error_at: null, last_ok_at: null, last_gap: null, started_at_seq: null };
+ const stats = { seen: 0, queued: 0, pages: 0, last_error: null, last_error_at: null, last_ok_at: null, last_gap: null, started_at_head: null, head: null };
let timer = null;
let timer = null;
let running = null;
let running = null;
let stopped = false;
let stopped = false;
const iso = () => new Date(now()).toISOString();
const iso = () => new Date(now()).toISOString();
- const readCursor = async () => await db.prepare('SELECT cursor, position, latest_seq FROM relay_cursors WHERE name = ?').get(CURSOR) || null;
- async function writeCursor(cursor, latest, position) {
- await db.prepare(`INSERT INTO relay_cursors (name, cursor, latest_seq, position, updated_at) VALUES (?, ?, ?, ?, ov_now())
- ON CONFLICT(name) DO UPDATE SET cursor = excluded.cursor, latest_seq = excluded.latest_seq, position = excluded.position, updated_at = ov_now()`)
- .run(CURSOR, cursor, latest == null ? null : latest, position || null);
+ const readPosition = async () => {
+ const row = await db.prepare('SELECT position FROM relay_cursors WHERE name = ?').get(CURSOR);
+ return row && row.position ? row.position : null;
+ };
+ async function writePosition(position) {
+ await db.prepare(`INSERT INTO relay_cursors (name, position, updated_at) VALUES (?, ?, ov_now())
+ ON CONFLICT(name) DO UPDATE SET position = excluded.position, updated_at = ov_now()`)
+ .run(CURSOR, position);
}
}
- const num = (v) => (Number.isFinite(Number(v)) && v !== null && v !== undefined ? Number(v) : null);
/** One event → deliveries queued (0 when it is not for the relay). */
/** One event → deliveries queued (0 when it is not for the relay). */
async function handle(ev) {
async function handle(ev) {
@@ -90,43 +91,36 @@ function createRelayEventsWorker({ db, relay, events = null, eventsUrl = null, c
@@ -90,43 +91,36 @@ function createRelayEventsWorker({ db, relay, events = null, eventsUrl = null, c
running = (async () => {
running = (async () => {
let queuedNow = 0;
let queuedNow = 0;
try {
try {
- let cur = await readCursor();
- if (!cur) {
- // First start: Events' head, as an opaque cursor (the numeric seq beside it while Events sends one).
+ let position = await readPosition();
+ if (!position) {
+ // First start: Events' head, as an opaque cursor.
const head = await client.pull({ topic: TOPICS, limit: 1 });
const head = await client.pull({ topic: TOPICS, limit: 1 });
- const latest = num(head.latest_seq) || 0;
- await writeCursor(latest, latest, head.latest_cursor || null);
- stats.started_at_seq = latest;
- console.log(`[Relay] Events worker starts at the head (seq ${latest}; earlier threads are not announced)`);
- cur = { cursor: latest, position: head.latest_cursor || null, latest_seq: latest };
+ if (!head.latest_cursor) throw new Error('Events answered no latest_cursor');
+ position = head.latest_cursor;
+ await writePosition(position);
+ stats.started_at_head = iso();
+ stats.head = position;
+ console.log('[Relay] Events worker starts at the head (earlier threads are not announced)');
}
}
- let cursor = num(cur.cursor) || 0;
- let position = cur.position || null;
for (let pages = 0; pages < MAX_PAGES_PER_TICK && !stopped; pages++) {
for (let pages = 0; pages < MAX_PAGES_PER_TICK && !stopped; pages++) {
- // Read on from the opaque position; a row from the previous release has only the numeric cursor.
- const page = position
- ? await client.pull({ topic: TOPICS, after: position, limit: pageLimit })
- : await client.pull({ topic: TOPICS, afterSeq: cursor, limit: pageLimit });
+ const page = await client.pull({ topic: TOPICS, after: position, limit: pageLimit });
if (stopped) break;
if (stopped) break;
if (page.gap) {
if (page.gap) {
stats.last_gap = { from_seq: page.gap.from_seq, to_seq: page.gap.to_seq, at: iso() };
stats.last_gap = { from_seq: page.gap.from_seq, to_seq: page.gap.to_seq, at: iso() };
console.warn(`[Relay] Events gap: seq ${page.gap.from_seq}..${page.gap.to_seq} were pruned before the relay read them; threads and replies in it are not relayed`);
console.warn(`[Relay] Events gap: seq ${page.gap.from_seq}..${page.gap.to_seq} were pruned before the relay read them; threads and replies in it are not relayed`);
}
}
const nextPosition = page.next_cursor || position;
const nextPosition = page.next_cursor || position;
- const next = num(page.next_after_seq);
- const nextCursor = next != null && next > cursor ? next : cursor;
- const latest = num(page.latest_seq);
const items = page.events || [];
const items = page.events || [];
queuedNow += await db.tx(async () => {
queuedNow += await db.tx(async () => {
let n = 0;
let n = 0;
for (const item of items) n += await handle(item && item.event);
for (const item of items) n += await handle(item && item.event);
- await writeCursor(nextCursor, latest, nextPosition);
+ await writePosition(nextPosition);
return n;
return n;
});
});
stats.pages++;
stats.pages++;
- const moved = nextPosition !== position || nextCursor !== cursor;
+ if (page.latest_cursor) stats.head = page.latest_cursor;
+ const moved = nextPosition !== position;
position = nextPosition;
position = nextPosition;
- cursor = nextCursor;
// A short page means the read reached the head (next_cursor already moved past what did not match).
// A short page means the read reached the head (next_cursor already moved past what did not match).
if (!moved || items.length < pageLimit) break;
if (!moved || items.length < pageLimit) break;
}
}
@@ -164,12 +158,13 @@ function createRelayEventsWorker({ db, relay, events = null, eventsUrl = null, c
@@ -164,12 +158,13 @@ function createRelayEventsWorker({ db, relay, events = null, eventsUrl = null, c
async function status() {
async function status() {
if (!on) return { enabled: false, reason };
if (!on) return { enabled: false, reason };
- const cur = await readCursor();
+ const position = await readPosition();
return {
return {
enabled: true, running: !!timer, topics: TOPICS,
enabled: true, running: !!timer, topics: TOPICS,
- cursor: cur ? cur.cursor : null, position: cur ? cur.position || null : null, latest_seq: cur ? cur.latest_seq : null,
- lag: cur && cur.latest_seq != null ? Math.max(cur.latest_seq - cur.cursor, 0) : null,
- events_seen: stats.seen, deliveries_queued: stats.queued, started_at_seq: stats.started_at_seq,
+ position,
+ // Whether the last read reached Events' head (both opaque cursors of one epoch: equal strings, equal place).
+ caught_up: position && stats.head ? position === stats.head : null,
+ events_seen: stats.seen, deliveries_queued: stats.queued, started_at_head: stats.started_at_head,
last_ok_at: stats.last_ok_at, last_error: stats.last_error, last_error_at: stats.last_error_at, last_gap: stats.last_gap,
last_ok_at: stats.last_ok_at, last_error: stats.last_error, last_error_at: stats.last_error_at, last_gap: stats.last_gap,
};
};
}
}
@@ -4,8 +4,8 @@
@@ -4,8 +4,8 @@
* the pull route the relay's Events worker reads.
* the pull route the relay's Events worker reads.
*
*
* POST /api/v1/events one envelope or { events: [...] } → { event_id, seq, duplicate } | { results }
* POST /api/v1/events one envelope or { events: [...] } → { event_id, seq, duplicate } | { results }
- * GET /api/v1/events?topic=a.*,b.*&after=<cursor>|after_seq=&limit=
- * → { events: [{ seq, cursor, event }], next_after_seq, next_cursor, latest_seq, latest_cursor, gap? }
+ * GET /api/v1/events?topic=a.*,b.*[&after=<cursor>]&limit=
+ * → { events: [{ seq, cursor, event }], next_cursor, latest_cursor, gap? } (after_seq is a 400, as in Events)
* Cursors are opaque to the worker; here, as in Events, `c1.<epoch>.<base64url(seq)>` (one epoch, 0).
* Cursors are opaque to the worker; here, as in Events, `c1.<epoch>.<base64url(seq)>` (one epoch, 0).
*
*
* Topic patterns follow Events (`*` is one or more whole segments). Every call needs a Bearer token.
* Topic patterns follow Events (`*` is one or more whole segments). Every call needs a Bearer token.
@@ -47,8 +47,9 @@ function startEvents() {
@@ -47,8 +47,9 @@ function startEvents() {
}
}
if (u.pathname === '/api/v1/events' && req.method === 'GET') {
if (u.pathname === '/api/v1/events' && req.method === 'GET') {
const patterns = String(u.searchParams.get('topic') || '*').split(',').map(matcher);
const patterns = String(u.searchParams.get('topic') || '*').split(',').map(matcher);
+ if (u.searchParams.has('after_seq')) return json(400, { type: 'about:blank', code: 'events.bad_request', status: 400, detail: 'after_seq is retired: pass after=<cursor>' });
const cur = u.searchParams.get('after');
const cur = u.searchParams.get('after');
- const after = cur ? decode(cur) : Number(u.searchParams.get('after_seq') || 0);
+ const after = cur ? decode(cur) : 0;
if (!Number.isFinite(after)) return json(400, { type: 'about:blank', code: 'events.bad_request', status: 400, detail: 'after must be an opaque cursor' });
if (!Number.isFinite(after)) return json(400, { type: 'about:blank', code: 'events.bad_request', status: 400, detail: 'after must be an opaque cursor' });
const limit = Number(u.searchParams.get('limit') || 100);
const limit = Number(u.searchParams.get('limit') || 100);
const out = {};
const out = {};
@@ -63,9 +64,7 @@ function startEvents() {
@@ -63,9 +64,7 @@ function startEvents() {
if (rows.length >= limit) break;
if (rows.length >= limit) break;
}
}
out.events = rows.map((r) => ({ ...r, cursor: encode(r.seq) }));
out.events = rows.map((r) => ({ ...r, cursor: encode(r.seq) }));
- out.next_after_seq = cursor;
out.next_cursor = encode(cursor);
out.next_cursor = encode(cursor);
- out.latest_seq = log.length;
out.latest_cursor = encode(log.length);
out.latest_cursor = encode(log.length);
return json(200, out);
return json(200, out);
}
}
@@ -70,10 +70,12 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.crea
@@ -70,10 +70,12 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.crea
await worker.tick();
await worker.tick();
assert.strictEqual((await relay.status()).creates_from, 'events');
assert.strictEqual((await relay.status()).creates_from, 'events');
const st = await worker.status();
const st = await worker.status();
- assert.deepStrictEqual([st.enabled, st.cursor, st.latest_seq, st.started_at_seq, st.lag], [true, 3, 3, 3, 0]);
+ assert.deepStrictEqual([st.enabled, st.position, st.caught_up], [true, events.encode(3), true]);
+ assert.ok(st.started_at_head, 'it says it started at the head');
+ assert.ok(!('cursor' in st) && !('lag' in st) && !('latest_seq' in st), 'no numeric position in the status');
assert.strictEqual((await deliveries()).length, 0);
assert.strictEqual((await deliveries()).length, 0);
- assert.deepStrictEqual(await db.prepare('SELECT name, cursor, position FROM relay_cursors').all(), [{ name: CURSOR, cursor: 3, position: events.encode(3) }], 'the head, stored as the opaque latest_cursor');
- assert.ok(events.calls.every((c) => !/[?&]after_seq=[1-9]/.test(c.path)), 'no numeric position is ever sent once a cursor is stored');
+ assert.deepStrictEqual(await db.prepare('SELECT name, cursor, position FROM relay_cursors').all(), [{ name: CURSOR, cursor: null, position: events.encode(3) }], 'the head, stored as the opaque latest_cursor and nothing numeric');
+ assert.ok(events.calls.every((c) => !/after_seq/.test(c.path)), 'no numeric position is ever sent');
before = events.calls.length;
before = events.calls.length;
});
});
@@ -96,7 +98,7 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.crea
@@ -96,7 +98,7 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.crea
await settle(); await relay.drain(); await relay.drain();
await settle(); await relay.drain(); await relay.drain();
assert.deepStrictEqual(hook.hits.map((h) => h.method), ['POST', 'POST']);
assert.deepStrictEqual(hook.hits.map((h) => h.method), ['POST', 'POST']);
assert.match(hook.hits[1].body.content, /replied to \*\*From the stream\*\*/);
assert.match(hook.hits[1].body.content, /replied to \*\*From the stream\*\*/);
- assert.deepStrictEqual([(await worker.status()).cursor, (await worker.status()).lag], [6, 0]);
+ assert.deepStrictEqual([(await worker.status()).position, (await worker.status()).caught_up], [events.encode(6), true]);
});
});
await check('a replay (cursor moved back) queues nothing twice and posts nothing twice', async () => {
await check('a replay (cursor moved back) queues nothing twice and posts nothing twice', async () => {
@@ -104,7 +106,7 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.crea
@@ -104,7 +106,7 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.crea
assert.deepStrictEqual(await worker.tick(), { queued: 0 });
assert.deepStrictEqual(await worker.tick(), { queued: 0 });
await relay.drain();
await relay.drain();
assert.strictEqual(hook.hits.length, 2);
assert.strictEqual(hook.hits.length, 2);
- assert.strictEqual((await worker.status()).cursor, 6);
+ assert.strictEqual((await worker.status()).position, events.encode(6));
});
});
await check('the cursor survives a restart: a new worker carries on from it (no second "start at head")', async () => {
await check('the cursor survives a restart: a new worker carries on from it (no second "start at head")', async () => {
@@ -112,7 +114,7 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.crea
@@ -112,7 +114,7 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.crea
events.add(threadEvent(t2));
events.add(threadEvent(t2));
const again = createRelayEventsWorker({ db, relay, events: client, pollMs: 3_600_000 });
const again = createRelayEventsWorker({ db, relay, events: client, pollMs: 3_600_000 });
assert.deepStrictEqual(await again.tick(), { queued: 1 });
assert.deepStrictEqual(await again.tick(), { queued: 1 });
- assert.strictEqual((await again.status()).started_at_seq, null);
+ assert.strictEqual((await again.status()).started_at_head, null);
assert.strictEqual((await deliveries()).pop().thread_id, t2.id);
assert.strictEqual((await deliveries()).pop().thread_id, t2.id);
});
});
@@ -133,25 +135,25 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.crea
@@ -133,25 +135,25 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.crea
await check('Events down: the error shows, the cursor stays, and it catches up once Events answers', async () => {
await check('Events down: the error shows, the cursor stays, and it catches up once Events answers', async () => {
const t4 = await newThread('After the outage');
const t4 = await newThread('After the outage');
events.add(threadEvent(t4));
events.add(threadEvent(t4));
- const cursor = (await worker.status()).cursor;
+ const position = (await worker.status()).position;
events.state.failNext = 1;
events.state.failNext = 1;
assert.deepStrictEqual(await worker.tick(), { queued: 0 });
assert.deepStrictEqual(await worker.tick(), { queued: 0 });
const st = await worker.status();
const st = await worker.status();
assert.ok(st.last_error, 'the error is shown');
assert.ok(st.last_error, 'the error is shown');
- assert.strictEqual(st.cursor, cursor);
- assert.strictEqual(st.lag, st.latest_seq - cursor);
+ assert.strictEqual(st.position, position, 'the cursor stays');
assert.deepStrictEqual(await worker.tick(), { queued: 1 });
assert.deepStrictEqual(await worker.tick(), { queued: 1 });
assert.strictEqual((await worker.status()).last_error, null);
assert.strictEqual((await worker.status()).last_error, null);
});
});
- await check('a row from the previous release (numeric cursor, no position) reads on from it once, then by cursor', async () => {
+ await check('a row with no position (a numeric one only) starts at the head: a number is never read as a place', async () => {
await db.prepare('UPDATE relay_cursors SET cursor = 3, position = NULL WHERE name = ?').run(CURSOR);
await db.prepare('UPDATE relay_cursors SET cursor = 3, position = NULL WHERE name = ?').run(CURSOR);
+ const t5 = await newThread('Before the restart at the head');
+ events.add(threadEvent(t5));
const from = events.calls.length;
const from = events.calls.length;
- assert.deepStrictEqual(await worker.tick(), { queued: 0 }, 'what it read again was already queued');
- const sent = events.calls.slice(from).map((c) => c.path);
- assert.ok(sent.some((p) => /[?&]after_seq=3\b/.test(p)), 'the first read uses the numeric cursor it had');
- const row = await db.prepare('SELECT cursor, position FROM relay_cursors WHERE name = ?').get(CURSOR);
- assert.strictEqual(row.position, events.encode(events.log.length), 'and from then on the opaque cursor');
+ assert.deepStrictEqual(await worker.tick(), { queued: 0 }, 'nothing from before the head is announced');
+ assert.ok(events.calls.slice(from).every((c) => !/after_seq/.test(c.path)), 'never a numeric position');
+ const row = await db.prepare('SELECT position FROM relay_cursors WHERE name = ?').get(CURSOR);
+ assert.strictEqual(row.position, events.encode(events.log.length), 'the head, as an opaque cursor');
const again = events.calls.length;
const again = events.calls.length;
await worker.tick();
await worker.tick();
assert.ok(events.calls.slice(again).every((c) => /[?&]after=c1\./.test(c.path)), 'later reads send the cursor');
assert.ok(events.calls.slice(again).every((c) => /[?&]after=c1\./.test(c.path)), 'later reads send the cursor');
@@ -162,7 +164,7 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.crea
@@ -162,7 +164,7 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.crea
events.state.prunedThrough = 5;
events.state.prunedThrough = 5;
await worker.tick();
await worker.tick();
assert.deepStrictEqual([(await worker.status()).last_gap.from_seq, (await worker.status()).last_gap.to_seq], [2, 5]);
assert.deepStrictEqual([(await worker.status()).last_gap.from_seq, (await worker.status()).last_gap.to_seq], [2, 5]);
- assert.strictEqual((await worker.status()).cursor, events.log.length);
+ assert.strictEqual((await worker.status()).position, events.encode(events.log.length));
events.state.prunedThrough = 0;
events.state.prunedThrough = 0;
assert.strictEqual((await relay.status()).events_worker.last_gap.to_seq, 5);
assert.strictEqual((await relay.status()).events_worker.last_gap.to_seq, 5);
});
});
@@ -186,7 +188,7 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.crea
@@ -186,7 +188,7 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.crea
const call = (p, { method = 'GET', cookie, json } = {}) => t.get(p, { method, headers: json !== undefined ? { 'content-type': 'application/json' } : {}, body: json !== undefined ? JSON.stringify(json) : undefined, cookies: cookie ? [`ov_token=${cookie}`] : [] });
const call = (p, { method = 'GET', cookie, json } = {}) => t.get(p, { method, headers: json !== undefined ? { 'content-type': 'application/json' } : {}, body: json !== undefined ? JSON.stringify(json) : undefined, cookies: cookie ? [`ov_token=${cookie}`] : [] });
const waitFor = async (fn, what) => { for (let i = 0; i < 200; i++) { if (await fn()) return; await new Promise((r) => setTimeout(r, 20)); } assert.fail(`timed out waiting for ${what}`); };
const waitFor = async (fn, what) => { for (let i = 0; i < 200; i++) { if (await fn()) return; await new Promise((r) => setTimeout(r, 20)); } assert.fail(`timed out waiting for ${what}`); };
assert.strictEqual((await call('/api/v1/relay/mappings', { method: 'POST', cookie: adminJwt, json: { space: 'general', webhook_url_ref: 'DISCORD_WEBHOOK_GENERAL' } })).status, 201);
assert.strictEqual((await call('/api/v1/relay/mappings', { method: 'POST', cookie: adminJwt, json: { space: 'general', webhook_url_ref: 'DISCORD_WEBHOOK_GENERAL' } })).status, 201);
- await waitFor(async () => (await t.app.locals.relay.status()).events_worker.cursor != null, 'the worker\'s first read');
+ await waitFor(async () => (await t.app.locals.relay.status()).events_worker.position != null, 'the worker\'s first read');
const st = (await call('/api/v1/relay/status', { cookie: adminJwt })).json();
const st = (await call('/api/v1/relay/status', { cookie: adminJwt })).json();
assert.deepStrictEqual([st.creates_from, st.events_worker.enabled], ['events', true]);
assert.deepStrictEqual([st.creates_from, st.events_worker.enabled], ['events', true]);
const th = await call('/api/v1/spaces/general/threads', { method: 'POST', cookie: samJwt, json: { title: 'Round trip', body: 'through Events' } });
const th = await call('/api/v1/spaces/general/threads', { method: 'POST', cookie: samJwt, json: { title: 'Round trip', body: 'through Events' } });