openvibers/openvibe.chat Commits
/ready tells the truth: a real chat.db read (required) and the last successful Live sync (optional)
Changes
@@@ -35,6 +35,8 @@ OV_OAUTH_CLIENT_SECRET=written-by-service-principal-js# OpenVibe.Live — every Live-owned read and side effect (server/live-context.js)OV_LIVE_INTERNAL_URL=http://127.0.0.1:3000#LIVE_TIMEOUT_MS=4000++# /ready reads the Live sync degraded once its last clean pass is older than this (ms)++#LIVE_SYNC_STALE_MS=60000# Copy Chat's changes into Live's chat tables (read mirror + rollback path). ON from the cutover;# OFF for rehearsals (and Live refuses mirror writes unless it runs with CHAT_AUTHORITY=chat).LIVE_MIRROR=0@@@ -88,7 +88,10 @@ node scripts/parity-check.js --live https://openvibe.live --chat http://127.0.0.node scripts/mirror-flush.js # rollback helper: push queued mirror rows to Live```--`/ready` reports the last Live sync, the mirror queue and whether events are relayed.++`/ready` (openvibe-shared/ready shape: `status` ready/degraded/not_ready and `checks`) is 503 only++when the `db` check (a read of `chat_messages`) fails. `live_sync` is optional: it reads degraded when++the last clean Live sync is older than `LIVE_SYNC_STALE_MS` (default 60 s) or a sync step has missed++its own interval by that much. It also reports the mirror queue and whether events are relayed.Production: `deploy/systemd/openvibe-chat.service`, `deploy/nginx/openvibe.live-chat.locations.conf`,`/etc/openvibe/chat.env`. The whole switch-over — rehearsal, import, parity checks, nginx, the flag,rollback — is `docs/cutover.md`.@@@ -109,7 +109,7 @@ report. Run the import again: every `inserted` must be 0 (idempotent).# 3. a rehearsal Chat on 4401 against production Live, mirror OFFsudo -u ubuntu env $(sudo cat /etc/openvibe/chat.env | xargs) PORT=4401 CHAT_DB_PATH=/tmp/chat-rehearsal.db \CHAT_CACHE_DIR=/tmp/chat-rehearsal-cache LIVE_MIRROR=0 EVENTS_URL= node server/index.js &--curl -s http://127.0.0.1:4401/ready | jq # ready: true after the first Live sync++curl -s http://127.0.0.1:4401/ready | jq # status: "ready" once a Live sync pass succeeded# 4. parity: the same reads answer the same (history pinned before the snapshot)node scripts/parity-check.js --live https://openvibe.live --chat http://127.0.0.1:4401 --before "$SNAP_AT" \@@@ -153,7 +153,7 @@ streamers; `deploy.sh --wait-idle` waits for none).Chat's own new rows are already in Live (mirror) and count as `identical`; `chat_kept` counts rowsChat edited since pass 1 (Chat wins); `held` must be 0.4. **Checks.**-- - `curl -s 127.0.0.1:4400/ready | jq` — `live.last_sync_at` recent, `mirror.pending` near 0,++ - `curl -s 127.0.0.1:4400/ready | jq` — `status` "ready" (`live.last_success_at` recent), `mirror.pending` near 0,`mirror.last_error` null.- Live: `sqlite3 …/live.db "select count(*) from chat_bridge_outbox"` stays near 0 (its forwardedwrites are acknowledged); `journalctl -u openvibe-live | grep ChatRemote` shows no refusals.@@@ -15,6 +15,7 @@ const cors = require('cors');const rateLimit = require('express-rate-limit');const config = require('./config');const ctx = require('./live-context');++const db = require('./db/database');const { extractToken, extractWsToken, authenticateWs } = require('./auth/auth');const { trustProxy } = require('./net/client-ip');@@@ -74,13 +75,52 @@ function createApp({ chatServer, bridge, mirror, relay }) {// Alert sounds travel as base64 in broadcasts (uploads up to MAX_SOUND_SIZE_KB): room for them.app.use('/internal/live', express.json({ limit: '16mb' }), bridge);app.get('/health', (req, res) => res.json({ ok: true, service: 'chat' }));++ // Readiness in the openvibe-shared/ready shape (status ready/degraded/not_ready, named checks):++ // 503 only when the required check fails. `db` is Chat's own database, which it cannot serve++ // without; `live_sync` is optional, because chat keeps flowing from warm caches while Live is++ // down (live-context.js), so a stale sync degrades the service instead of taking it out.++ function timed(fn) {++ const t0 = process.hrtime.bigint();++ let out;++ try { out = fn(); } catch (err) { out = { ok: false, error: String((err && err.message) || err).split('\n')[0].slice(0, 200) }; }++ const { ok, error, detail } = out;++ return {++ status: ok ? 'ok' : 'fail', required: false,++ ...(error ? { error } : {}), ...(detail !== undefined ? { detail } : {}),++ latency_ms: Math.round(Number(process.hrtime.bigint() - t0) / 1e5) / 10,++ checked_at: new Date().toISOString(),++ };++ }app.get('/ready', (req, res) => {-- const ready = !!ctx.stats.lastSyncAt;++ const checks = {++ // A real read of a chat table (MAX of the rowid is an index seek, not a scan).++ db: { ...timed(() => ({ ok: true, detail: { max_message_id: db.get('SELECT MAX(id) AS id FROM chat_messages').id } })), required: true },++ live_sync: timed(() => {++ const s = ctx.syncStatus(config.live.syncStaleMs);++ const detail = { ...s, threshold_ms: config.live.syncStaleMs };++ if (s.last_success_at === null) return { ok: false, error: 'no successful Live sync since start', detail };++ if (s.age_ms > config.live.syncStaleMs) return { ok: false, error: `last successful Live sync ${Math.round(s.age_ms / 1000)}s ago`, detail };++ if (s.late_steps.length) return { ok: false, error: `Live sync late: ${s.late_steps.join(', ')}`, detail };++ return { ok: true, detail };++ }),++ };++ const failed = Object.keys(checks).filter((k) => checks[k].required && checks[k].status !== 'ok');++ const degraded = Object.keys(checks).filter((k) => !checks[k].required && checks[k].status !== 'ok');++ const ready = failed.length === 0;++ let pending = null;++ if (mirror && checks.db.status === 'ok') { try { pending = mirror.pending(); } catch { /* reported by db */ } }++ res.set('Cache-Control', 'no-store');res.status(ready ? 200 : 503).json({ready,++ status: !ready ? 'not_ready' : degraded.length ? 'degraded' : 'ready',++ service: 'chat',++ checked_at: new Date().toISOString(),++ failed,++ degraded,++ checks,connections: chatServer.getTotalConnections(),-- live: { last_sync_at: ctx.stats.lastSyncAt, failures: ctx.stats.failures, last_error: ctx.stats.lastError },-- mirror: mirror ? { enabled: config.live.mirror, pending: mirror.pending(), last_error: mirror.lastError() } : null,++ live: { last_sync_at: ctx.stats.lastSyncAt, last_success_at: ctx.stats.lastSuccessAt, failures: ctx.stats.failures, last_error: ctx.stats.lastError },++ mirror: mirror ? { enabled: config.live.mirror, pending, last_error: mirror.lastError() } : null,events: { enabled: !!config.events.url },});});@@@ -54,6 +54,14 @@ module.exports = {mirror: bool(process.env.LIVE_MIRROR, false),mirrorIntervalMs: int(process.env.LIVE_MIRROR_INTERVAL_MS, 1000),requestTimeoutMs: int(process.env.LIVE_TIMEOUT_MS, 4000),++ // /ready reports the Live sync degraded once its last clean pass is older than this.++ // The loop ticks every 2 s and the bans and stream projections are due every 10 s, so a++ // healthy Chat finishes a clean pass at least every ~10 s; 60 s is six missed ban/stream++ // refreshes: long enough to ride out a Live restart (about 6 s) or a slow pass of 4 s++ // timeouts, short enough that bans and live state served from stale caches show within++ // a minute. The same margin applies to each step on top of its own interval++ // (live-context.js SCHEDULE), so the 5- and 15-minute full syncs are judged fairly too.++ syncStaleMs: int(process.env.LIVE_SYNC_STALE_MS, 60_000),},// OpenVibe.Events — the outbox relays only when EVENTS_URL is set.@@@ -118,7 +118,9 @@ function defaultModerationSettings(channelId) {// ── Transport ────────────────────────────────────────────────────────────────────────────────let _fetch = (...a) => globalThis.fetch(...a);--const stats = { reads: 0, effects: 0, failures: 0, lastError: null, lastSyncAt: null };++// lastSyncAt: the last sync pass, whatever its outcome. lastSuccessAt: the last pass in which every++// projection it ran came back from Live (what /ready judges freshness by).++const stats = { reads: 0, effects: 0, failures: 0, lastError: null, lastSyncAt: null, lastSuccessAt: null };class LiveError extends Error {constructor(status, message, body) { super(message); this.status = status; this.body = body; }@@@ -655,36 +657,72 @@ const SCHEDULE = [['streams*', 30 * 60_000, () => syncTable('streams', 'ctx_streams', STREAM_COLS, { full: true })],['managed*', 5 * 60_000, () => syncTable('managed-streams', 'ctx_managed_streams', MS_COLS, { full: true })],['channels*', 5 * 60_000, () => syncTable('channels', 'ctx_channels', CHANNEL_COLS, { full: true })],-- ['bans', TTL.bans, refreshBans],-- ['settings', TTL.settings, () => ensureSettings(true)],++ // refreshBans and ensureSettings keep the cached value on a Live error instead of throwing++ // (callers want the stale value); for the schedule a refresh that did not land is a failure.++ ['bans', TTL.bans, async () => { const t0 = Date.now(); await refreshBans(); if (_bans.at < t0) throw new Error('bans not refreshed'); }],++ ['settings', TTL.settings, async () => { const t0 = Date.now(); await ensureSettings(true); if (_settings.at < t0) throw new Error('settings not refreshed'); }],];const _lastRun = new Map();++const _lastOk = new Map(); // step name -> ms of its last success++let _since = null; // ms the sync loop began (a step that never succeeded is late from here)let _timer = null;let _running = false;++/** Run one scheduled step; true when it succeeded. */++async function runStep(name, fn, label) {++ try { await fn(); _lastOk.set(name, Date.now()); return true; } catch (err) { console.warn(`[LiveContext] ${label} ${name}: ${err.message}`); return false; }++}++function endPass(ran, failed) {++ stats.lastSyncAt = new Date().toISOString();++ if (ran && !failed) stats.lastSuccessAt = stats.lastSyncAt;++}++++/**++ * How current the projections are, for /ready: the last clean pass and its age, and the steps++ * that have not succeeded for longer than their own interval plus `staleMs` (a step that fails++ * on every run of its own — say the 5-minute channel sync — never spoils a whole pass for long,++ * because the 10-second steps pass in between, so it is judged on its own).++ */++function syncStatus(staleMs, now = Date.now()) {++ const lastOk = stats.lastSuccessAt ? Date.parse(stats.lastSuccessAt) : null;++ const late = [];++ if (_since !== null) {++ for (const [name, every] of SCHEDULE) {++ const at = _lastOk.get(name) || _since;++ if (now - at > every + staleMs) late.push(name);++ }++ }++ return { last_success_at: stats.lastSuccessAt, age_ms: lastOk === null ? null : now - lastOk, late_steps: late };++}++async function tick() {if (_running) return;_running = true;try {++ let ran = 0, failed = 0;for (const [name, every, fn] of SCHEDULE) {if (Date.now() - (_lastRun.get(name) || 0) < every) continue;_lastRun.set(name, Date.now());-- try { await fn(); } catch (err) { console.warn(`[LiveContext] sync ${name}: ${err.message}`); }++ ran++;++ if (!(await runStep(name, fn, 'sync'))) failed++;}-- stats.lastSyncAt = new Date().toISOString();++ endPass(ran, failed);} finally { _running = false; }}/** One full pass over every projection (boot, tests, `npm run` tools). */async function sync() {++ if (_since === null) _since = Date.now();_lastRun.clear();-- for (const [, , fn] of SCHEDULE) { try { await fn(); } catch (err) { console.warn('[LiveContext] initial sync:', err.message); } }++ let failed = 0;++ for (const [name, , fn] of SCHEDULE) if (!(await runStep(name, fn, 'initial sync'))) failed++;for (const [name] of SCHEDULE) _lastRun.set(name, Date.now());-- stats.lastSyncAt = new Date().toISOString();++ endPass(SCHEDULE.length, failed);}function start({ intervalMs = 2000 } = {}) {if (_timer) return;++ if (_since === null) _since = Date.now();_timer = setInterval(() => { tick().catch(() => {}); }, intervalMs);if (_timer.unref) _timer.unref();_batchTimer = setInterval(flushBatches, 2000);@@@ -751,7 +789,7 @@ module.exports = {LiveError,stats,// lifecycle-- start, stop, sync, tick, warm,++ start, stop, sync, tick, warm, syncStatus,// identity + usersauthenticate, authFailureReason, upsertUser, invalidateUser, subjectFor,getUserById, getUserByUsername, getUserByDisplayName, ensureUsers, ensureUserByUsername,nnew file mode 100644@@@ -0,0 +1,91 @@++'use strict';++/**++ * /ready tells the truth about Chat's backing store and its Live sync: the database is required++ * (a real read of a chat table; 503 when it fails), the Live sync is optional and reads degraded++ * (still 200) once its last clean pass is older than LIVE_SYNC_STALE_MS — a pass where Live++ * failed does not count as a sync — or when one step has missed its own schedule.++ */++const assert = require('assert');++const { boot, suite } = require('./helpers');++++const t = suite('ready');++const STALE_MS = 2500;++let h;++++async function ready() {++ const r = await h.http('GET', '/ready');++ return { status: r.status, body: r.body };++}++++t('boot', async () => {++ h = await boot({ env: { LIVE_SYNC_STALE_MS: String(STALE_MS) } });++ await h.ctx.sync();++});++++t('healthy: 200 ready with a db and a live_sync check', async () => {++ const { status, body } = await ready();++ assert.strictEqual(status, 200);++ assert.strictEqual(body.status, 'ready');++ assert.strictEqual(body.ready, true);++ assert.deepStrictEqual(body.failed, []);++ assert.deepStrictEqual(body.degraded, []);++ assert.strictEqual(body.checks.db.status, 'ok');++ assert.strictEqual(body.checks.db.required, true);++ assert.strictEqual(body.checks.live_sync.status, 'ok');++ assert.strictEqual(body.checks.live_sync.required, false);++ assert.strictEqual(body.checks.live_sync.detail.threshold_ms, STALE_MS);++ assert.ok(body.live.last_success_at, 'last successful sync reported');++ assert.strictEqual(body.service, 'chat');++});++++t('Live failing: passes keep running but do not count; degraded (still 200) after the threshold', async () => {++ h.live.down = true;++ const failuresBefore = h.ctx.stats.failures;++ await h.sleep(STALE_MS + 1500);++ // Force a pass so the attempt is recorded whatever the 10 s schedule says.++ await h.ctx.sync();++ assert.ok(h.ctx.stats.failures > failuresBefore, 'Live calls failed');++ const { status, body } = await ready();++ assert.strictEqual(status, 200, 'Live is optional: still ready');++ assert.strictEqual(body.status, 'degraded');++ assert.deepStrictEqual(body.degraded, ['live_sync']);++ assert.strictEqual(body.checks.live_sync.status, 'fail');++ assert.match(body.checks.live_sync.error, /last successful Live sync \d+s ago/);++ assert.ok(Date.parse(body.live.last_sync_at) > Date.parse(body.live.last_success_at), 'an attempt is not a success');++});++++t('Live back: one clean pass makes it ready again', async () => {++ h.live.down = false;++ await h.ctx.sync();++ const { status, body } = await ready();++ assert.strictEqual(status, 200);++ assert.strictEqual(body.status, 'ready');++ assert.strictEqual(body.live.last_sync_at, body.live.last_success_at);++});++++t('a step that misses its own schedule is late on its own (not masked by the fast steps)', async () => {++ // 12 s from now the 10 s steps are past interval + 1 s; the 30 s and slower ones are not.++ const s = h.ctx.syncStatus(1000, Date.now() + 12_000);++ assert.ok(s.late_steps.includes('bans') && s.late_steps.includes('streams~'), JSON.stringify(s));++ assert.ok(!s.late_steps.includes('users+') && !s.late_steps.includes('users*'), JSON.stringify(s));++ assert.deepStrictEqual(h.ctx.syncStatus(1000).late_steps, []);++});++++t('database failing: 503 not_ready naming the db check', async () => {++ h.db.run('ALTER TABLE chat_messages RENAME TO chat_messages_away');++ try {++ const { status, body } = await ready();++ assert.strictEqual(status, 503);++ assert.strictEqual(body.ready, false);++ assert.strictEqual(body.status, 'not_ready');++ assert.deepStrictEqual(body.failed, ['db']);++ assert.strictEqual(body.checks.db.status, 'fail');++ assert.match(body.checks.db.error, /no such table/);++ } finally {++ h.db.run('ALTER TABLE chat_messages_away RENAME TO chat_messages');++ }++ const { status } = await ready();++ assert.strictEqual(status, 200);++});++++t.run(() => h && h.close());
@@ -35,6 +35,8 @@ OV_OAUTH_CLIENT_SECRET=written-by-service-principal-js
@@ -35,6 +35,8 @@ OV_OAUTH_CLIENT_SECRET=written-by-service-principal-js
# OpenVibe.Live — every Live-owned read and side effect (server/live-context.js)
# OpenVibe.Live — every Live-owned read and side effect (server/live-context.js)
OV_LIVE_INTERNAL_URL=http://127.0.0.1:3000
OV_LIVE_INTERNAL_URL=http://127.0.0.1:3000
#LIVE_TIMEOUT_MS=4000
#LIVE_TIMEOUT_MS=4000
+# /ready reads the Live sync degraded once its last clean pass is older than this (ms)
+#LIVE_SYNC_STALE_MS=60000
# Copy Chat's changes into Live's chat tables (read mirror + rollback path). ON from the cutover;
# Copy Chat's changes into Live's chat tables (read mirror + rollback path). ON from the cutover;
# OFF for rehearsals (and Live refuses mirror writes unless it runs with CHAT_AUTHORITY=chat).
# OFF for rehearsals (and Live refuses mirror writes unless it runs with CHAT_AUTHORITY=chat).
LIVE_MIRROR=0
LIVE_MIRROR=0
@@ -88,7 +88,10 @@ node scripts/parity-check.js --live https://openvibe.live --chat http://127.0.0.
@@ -88,7 +88,10 @@ node scripts/parity-check.js --live https://openvibe.live --chat http://127.0.0.
node scripts/mirror-flush.js # rollback helper: push queued mirror rows to Live
node scripts/mirror-flush.js # rollback helper: push queued mirror rows to Live
```
```
-`/ready` reports the last Live sync, the mirror queue and whether events are relayed.
+`/ready` (openvibe-shared/ready shape: `status` ready/degraded/not_ready and `checks`) is 503 only
+when the `db` check (a read of `chat_messages`) fails. `live_sync` is optional: it reads degraded when
+the last clean Live sync is older than `LIVE_SYNC_STALE_MS` (default 60 s) or a sync step has missed
+its own interval by that much. It also reports the mirror queue and whether events are relayed.
Production: `deploy/systemd/openvibe-chat.service`, `deploy/nginx/openvibe.live-chat.locations.conf`,
Production: `deploy/systemd/openvibe-chat.service`, `deploy/nginx/openvibe.live-chat.locations.conf`,
`/etc/openvibe/chat.env`. The whole switch-over — rehearsal, import, parity checks, nginx, the flag,
`/etc/openvibe/chat.env`. The whole switch-over — rehearsal, import, parity checks, nginx, the flag,
rollback — is `docs/cutover.md`.
rollback — is `docs/cutover.md`.
@@ -109,7 +109,7 @@ report. Run the import again: every `inserted` must be 0 (idempotent).
@@ -109,7 +109,7 @@ report. Run the import again: every `inserted` must be 0 (idempotent).
# 3. a rehearsal Chat on 4401 against production Live, mirror OFF
# 3. a rehearsal Chat on 4401 against production Live, mirror OFF
sudo -u ubuntu env $(sudo cat /etc/openvibe/chat.env | xargs) PORT=4401 CHAT_DB_PATH=/tmp/chat-rehearsal.db \
sudo -u ubuntu env $(sudo cat /etc/openvibe/chat.env | xargs) PORT=4401 CHAT_DB_PATH=/tmp/chat-rehearsal.db \
CHAT_CACHE_DIR=/tmp/chat-rehearsal-cache LIVE_MIRROR=0 EVENTS_URL= node server/index.js &
CHAT_CACHE_DIR=/tmp/chat-rehearsal-cache LIVE_MIRROR=0 EVENTS_URL= node server/index.js &
-curl -s http://127.0.0.1:4401/ready | jq # ready: true after the first Live sync
+curl -s http://127.0.0.1:4401/ready | jq # status: "ready" once a Live sync pass succeeded
# 4. parity: the same reads answer the same (history pinned before the snapshot)
# 4. parity: the same reads answer the same (history pinned before the snapshot)
node scripts/parity-check.js --live https://openvibe.live --chat http://127.0.0.1:4401 --before "$SNAP_AT" \
node scripts/parity-check.js --live https://openvibe.live --chat http://127.0.0.1:4401 --before "$SNAP_AT" \
@@ -153,7 +153,7 @@ streamers; `deploy.sh --wait-idle` waits for none).
@@ -153,7 +153,7 @@ streamers; `deploy.sh --wait-idle` waits for none).
Chat's own new rows are already in Live (mirror) and count as `identical`; `chat_kept` counts rows
Chat's own new rows are already in Live (mirror) and count as `identical`; `chat_kept` counts rows
Chat edited since pass 1 (Chat wins); `held` must be 0.
Chat edited since pass 1 (Chat wins); `held` must be 0.
4. **Checks.**
4. **Checks.**
- - `curl -s 127.0.0.1:4400/ready | jq` — `live.last_sync_at` recent, `mirror.pending` near 0,
+ - `curl -s 127.0.0.1:4400/ready | jq` — `status` "ready" (`live.last_success_at` recent), `mirror.pending` near 0,
`mirror.last_error` null.
`mirror.last_error` null.
- Live: `sqlite3 …/live.db "select count(*) from chat_bridge_outbox"` stays near 0 (its forwarded
- Live: `sqlite3 …/live.db "select count(*) from chat_bridge_outbox"` stays near 0 (its forwarded
writes are acknowledged); `journalctl -u openvibe-live | grep ChatRemote` shows no refusals.
writes are acknowledged); `journalctl -u openvibe-live | grep ChatRemote` shows no refusals.
@@ -15,6 +15,7 @@ const cors = require('cors');
@@ -15,6 +15,7 @@ const cors = require('cors');
const rateLimit = require('express-rate-limit');
const rateLimit = require('express-rate-limit');
const config = require('./config');
const config = require('./config');
const ctx = require('./live-context');
const ctx = require('./live-context');
+const db = require('./db/database');
const { extractToken, extractWsToken, authenticateWs } = require('./auth/auth');
const { extractToken, extractWsToken, authenticateWs } = require('./auth/auth');
const { trustProxy } = require('./net/client-ip');
const { trustProxy } = require('./net/client-ip');
@@ -74,13 +75,52 @@ function createApp({ chatServer, bridge, mirror, relay }) {
@@ -74,13 +75,52 @@ function createApp({ chatServer, bridge, mirror, relay }) {
// Alert sounds travel as base64 in broadcasts (uploads up to MAX_SOUND_SIZE_KB): room for them.
// Alert sounds travel as base64 in broadcasts (uploads up to MAX_SOUND_SIZE_KB): room for them.
app.use('/internal/live', express.json({ limit: '16mb' }), bridge);
app.use('/internal/live', express.json({ limit: '16mb' }), bridge);
app.get('/health', (req, res) => res.json({ ok: true, service: 'chat' }));
app.get('/health', (req, res) => res.json({ ok: true, service: 'chat' }));
+ // Readiness in the openvibe-shared/ready shape (status ready/degraded/not_ready, named checks):
+ // 503 only when the required check fails. `db` is Chat's own database, which it cannot serve
+ // without; `live_sync` is optional, because chat keeps flowing from warm caches while Live is
+ // down (live-context.js), so a stale sync degrades the service instead of taking it out.
+ function timed(fn) {
+ const t0 = process.hrtime.bigint();
+ let out;
+ try { out = fn(); } catch (err) { out = { ok: false, error: String((err && err.message) || err).split('\n')[0].slice(0, 200) }; }
+ const { ok, error, detail } = out;
+ return {
+ status: ok ? 'ok' : 'fail', required: false,
+ ...(error ? { error } : {}), ...(detail !== undefined ? { detail } : {}),
+ latency_ms: Math.round(Number(process.hrtime.bigint() - t0) / 1e5) / 10,
+ checked_at: new Date().toISOString(),
+ };
+ }
app.get('/ready', (req, res) => {
app.get('/ready', (req, res) => {
- const ready = !!ctx.stats.lastSyncAt;
+ const checks = {
+ // A real read of a chat table (MAX of the rowid is an index seek, not a scan).
+ db: { ...timed(() => ({ ok: true, detail: { max_message_id: db.get('SELECT MAX(id) AS id FROM chat_messages').id } })), required: true },
+ live_sync: timed(() => {
+ const s = ctx.syncStatus(config.live.syncStaleMs);
+ const detail = { ...s, threshold_ms: config.live.syncStaleMs };
+ if (s.last_success_at === null) return { ok: false, error: 'no successful Live sync since start', detail };
+ if (s.age_ms > config.live.syncStaleMs) return { ok: false, error: `last successful Live sync ${Math.round(s.age_ms / 1000)}s ago`, detail };
+ if (s.late_steps.length) return { ok: false, error: `Live sync late: ${s.late_steps.join(', ')}`, detail };
+ return { ok: true, detail };
+ }),
+ };
+ const failed = Object.keys(checks).filter((k) => checks[k].required && checks[k].status !== 'ok');
+ const degraded = Object.keys(checks).filter((k) => !checks[k].required && checks[k].status !== 'ok');
+ const ready = failed.length === 0;
+ let pending = null;
+ if (mirror && checks.db.status === 'ok') { try { pending = mirror.pending(); } catch { /* reported by db */ } }
+ res.set('Cache-Control', 'no-store');
res.status(ready ? 200 : 503).json({
res.status(ready ? 200 : 503).json({
ready,
ready,
+ status: !ready ? 'not_ready' : degraded.length ? 'degraded' : 'ready',
+ service: 'chat',
+ checked_at: new Date().toISOString(),
+ failed,
+ degraded,
+ checks,
connections: chatServer.getTotalConnections(),
connections: chatServer.getTotalConnections(),
- live: { last_sync_at: ctx.stats.lastSyncAt, failures: ctx.stats.failures, last_error: ctx.stats.lastError },
- mirror: mirror ? { enabled: config.live.mirror, pending: mirror.pending(), last_error: mirror.lastError() } : null,
+ live: { last_sync_at: ctx.stats.lastSyncAt, last_success_at: ctx.stats.lastSuccessAt, failures: ctx.stats.failures, last_error: ctx.stats.lastError },
+ mirror: mirror ? { enabled: config.live.mirror, pending, last_error: mirror.lastError() } : null,
events: { enabled: !!config.events.url },
events: { enabled: !!config.events.url },
});
});
});
});
@@ -54,6 +54,14 @@ module.exports = {
@@ -54,6 +54,14 @@ module.exports = {
mirror: bool(process.env.LIVE_MIRROR, false),
mirror: bool(process.env.LIVE_MIRROR, false),
mirrorIntervalMs: int(process.env.LIVE_MIRROR_INTERVAL_MS, 1000),
mirrorIntervalMs: int(process.env.LIVE_MIRROR_INTERVAL_MS, 1000),
requestTimeoutMs: int(process.env.LIVE_TIMEOUT_MS, 4000),
requestTimeoutMs: int(process.env.LIVE_TIMEOUT_MS, 4000),
+ // /ready reports the Live sync degraded once its last clean pass is older than this.
+ // The loop ticks every 2 s and the bans and stream projections are due every 10 s, so a
+ // healthy Chat finishes a clean pass at least every ~10 s; 60 s is six missed ban/stream
+ // refreshes: long enough to ride out a Live restart (about 6 s) or a slow pass of 4 s
+ // timeouts, short enough that bans and live state served from stale caches show within
+ // a minute. The same margin applies to each step on top of its own interval
+ // (live-context.js SCHEDULE), so the 5- and 15-minute full syncs are judged fairly too.
+ syncStaleMs: int(process.env.LIVE_SYNC_STALE_MS, 60_000),
},
},
// OpenVibe.Events — the outbox relays only when EVENTS_URL is set.
// OpenVibe.Events — the outbox relays only when EVENTS_URL is set.
@@ -118,7 +118,9 @@ function defaultModerationSettings(channelId) {
@@ -118,7 +118,9 @@ function defaultModerationSettings(channelId) {
// ── Transport ────────────────────────────────────────────────────────────────────────────────
// ── Transport ────────────────────────────────────────────────────────────────────────────────
let _fetch = (...a) => globalThis.fetch(...a);
let _fetch = (...a) => globalThis.fetch(...a);
-const stats = { reads: 0, effects: 0, failures: 0, lastError: null, lastSyncAt: null };
+// lastSyncAt: the last sync pass, whatever its outcome. lastSuccessAt: the last pass in which every
+// projection it ran came back from Live (what /ready judges freshness by).
+const stats = { reads: 0, effects: 0, failures: 0, lastError: null, lastSyncAt: null, lastSuccessAt: null };
class LiveError extends Error {
class LiveError extends Error {
constructor(status, message, body) { super(message); this.status = status; this.body = body; }
constructor(status, message, body) { super(message); this.status = status; this.body = body; }
@@ -655,36 +657,72 @@ const SCHEDULE = [
@@ -655,36 +657,72 @@ const SCHEDULE = [
['streams*', 30 * 60_000, () => syncTable('streams', 'ctx_streams', STREAM_COLS, { full: true })],
['streams*', 30 * 60_000, () => syncTable('streams', 'ctx_streams', STREAM_COLS, { full: true })],
['managed*', 5 * 60_000, () => syncTable('managed-streams', 'ctx_managed_streams', MS_COLS, { full: true })],
['managed*', 5 * 60_000, () => syncTable('managed-streams', 'ctx_managed_streams', MS_COLS, { full: true })],
['channels*', 5 * 60_000, () => syncTable('channels', 'ctx_channels', CHANNEL_COLS, { full: true })],
['channels*', 5 * 60_000, () => syncTable('channels', 'ctx_channels', CHANNEL_COLS, { full: true })],
- ['bans', TTL.bans, refreshBans],
- ['settings', TTL.settings, () => ensureSettings(true)],
+ // refreshBans and ensureSettings keep the cached value on a Live error instead of throwing
+ // (callers want the stale value); for the schedule a refresh that did not land is a failure.
+ ['bans', TTL.bans, async () => { const t0 = Date.now(); await refreshBans(); if (_bans.at < t0) throw new Error('bans not refreshed'); }],
+ ['settings', TTL.settings, async () => { const t0 = Date.now(); await ensureSettings(true); if (_settings.at < t0) throw new Error('settings not refreshed'); }],
];
];
const _lastRun = new Map();
const _lastRun = new Map();
+const _lastOk = new Map(); // step name -> ms of its last success
+let _since = null; // ms the sync loop began (a step that never succeeded is late from here)
let _timer = null;
let _timer = null;
let _running = false;
let _running = false;
+/** Run one scheduled step; true when it succeeded. */
+async function runStep(name, fn, label) {
+ try { await fn(); _lastOk.set(name, Date.now()); return true; } catch (err) { console.warn(`[LiveContext] ${label} ${name}: ${err.message}`); return false; }
+}
+function endPass(ran, failed) {
+ stats.lastSyncAt = new Date().toISOString();
+ if (ran && !failed) stats.lastSuccessAt = stats.lastSyncAt;
+}
+
+/**
+ * How current the projections are, for /ready: the last clean pass and its age, and the steps
+ * that have not succeeded for longer than their own interval plus `staleMs` (a step that fails
+ * on every run of its own — say the 5-minute channel sync — never spoils a whole pass for long,
+ * because the 10-second steps pass in between, so it is judged on its own).
+ */
+function syncStatus(staleMs, now = Date.now()) {
+ const lastOk = stats.lastSuccessAt ? Date.parse(stats.lastSuccessAt) : null;
+ const late = [];
+ if (_since !== null) {
+ for (const [name, every] of SCHEDULE) {
+ const at = _lastOk.get(name) || _since;
+ if (now - at > every + staleMs) late.push(name);
+ }
+ }
+ return { last_success_at: stats.lastSuccessAt, age_ms: lastOk === null ? null : now - lastOk, late_steps: late };
+}
+
async function tick() {
async function tick() {
if (_running) return;
if (_running) return;
_running = true;
_running = true;
try {
try {
+ let ran = 0, failed = 0;
for (const [name, every, fn] of SCHEDULE) {
for (const [name, every, fn] of SCHEDULE) {
if (Date.now() - (_lastRun.get(name) || 0) < every) continue;
if (Date.now() - (_lastRun.get(name) || 0) < every) continue;
_lastRun.set(name, Date.now());
_lastRun.set(name, Date.now());
- try { await fn(); } catch (err) { console.warn(`[LiveContext] sync ${name}: ${err.message}`); }
+ ran++;
+ if (!(await runStep(name, fn, 'sync'))) failed++;
}
}
- stats.lastSyncAt = new Date().toISOString();
+ endPass(ran, failed);
} finally { _running = false; }
} finally { _running = false; }
}
}
/** One full pass over every projection (boot, tests, `npm run` tools). */
/** One full pass over every projection (boot, tests, `npm run` tools). */
async function sync() {
async function sync() {
+ if (_since === null) _since = Date.now();
_lastRun.clear();
_lastRun.clear();
- for (const [, , fn] of SCHEDULE) { try { await fn(); } catch (err) { console.warn('[LiveContext] initial sync:', err.message); } }
+ let failed = 0;
+ for (const [name, , fn] of SCHEDULE) if (!(await runStep(name, fn, 'initial sync'))) failed++;
for (const [name] of SCHEDULE) _lastRun.set(name, Date.now());
for (const [name] of SCHEDULE) _lastRun.set(name, Date.now());
- stats.lastSyncAt = new Date().toISOString();
+ endPass(SCHEDULE.length, failed);
}
}
function start({ intervalMs = 2000 } = {}) {
function start({ intervalMs = 2000 } = {}) {
if (_timer) return;
if (_timer) return;
+ if (_since === null) _since = Date.now();
_timer = setInterval(() => { tick().catch(() => {}); }, intervalMs);
_timer = setInterval(() => { tick().catch(() => {}); }, intervalMs);
if (_timer.unref) _timer.unref();
if (_timer.unref) _timer.unref();
_batchTimer = setInterval(flushBatches, 2000);
_batchTimer = setInterval(flushBatches, 2000);
@@ -751,7 +789,7 @@ module.exports = {
@@ -751,7 +789,7 @@ module.exports = {
LiveError,
LiveError,
stats,
stats,
// lifecycle
// lifecycle
- start, stop, sync, tick, warm,
+ start, stop, sync, tick, warm, syncStatus,
// identity + users
// identity + users
authenticate, authFailureReason, upsertUser, invalidateUser, subjectFor,
authenticate, authFailureReason, upsertUser, invalidateUser, subjectFor,
getUserById, getUserByUsername, getUserByDisplayName, ensureUsers, ensureUserByUsername,
getUserById, getUserByUsername, getUserByDisplayName, ensureUsers, ensureUserByUsername,
new file mode 100644
new file mode 100644
@@ -0,0 +1,91 @@
@@ -0,0 +1,91 @@
+'use strict';
+/**
+ * /ready tells the truth about Chat's backing store and its Live sync: the database is required
+ * (a real read of a chat table; 503 when it fails), the Live sync is optional and reads degraded
+ * (still 200) once its last clean pass is older than LIVE_SYNC_STALE_MS — a pass where Live
+ * failed does not count as a sync — or when one step has missed its own schedule.
+ */
+const assert = require('assert');
+const { boot, suite } = require('./helpers');
+
+const t = suite('ready');
+const STALE_MS = 2500;
+let h;
+
+async function ready() {
+ const r = await h.http('GET', '/ready');
+ return { status: r.status, body: r.body };
+}
+
+t('boot', async () => {
+ h = await boot({ env: { LIVE_SYNC_STALE_MS: String(STALE_MS) } });
+ await h.ctx.sync();
+});
+
+t('healthy: 200 ready with a db and a live_sync check', async () => {
+ const { status, body } = await ready();
+ assert.strictEqual(status, 200);
+ assert.strictEqual(body.status, 'ready');
+ assert.strictEqual(body.ready, true);
+ assert.deepStrictEqual(body.failed, []);
+ assert.deepStrictEqual(body.degraded, []);
+ assert.strictEqual(body.checks.db.status, 'ok');
+ assert.strictEqual(body.checks.db.required, true);
+ assert.strictEqual(body.checks.live_sync.status, 'ok');
+ assert.strictEqual(body.checks.live_sync.required, false);
+ assert.strictEqual(body.checks.live_sync.detail.threshold_ms, STALE_MS);
+ assert.ok(body.live.last_success_at, 'last successful sync reported');
+ assert.strictEqual(body.service, 'chat');
+});
+
+t('Live failing: passes keep running but do not count; degraded (still 200) after the threshold', async () => {
+ h.live.down = true;
+ const failuresBefore = h.ctx.stats.failures;
+ await h.sleep(STALE_MS + 1500);
+ // Force a pass so the attempt is recorded whatever the 10 s schedule says.
+ await h.ctx.sync();
+ assert.ok(h.ctx.stats.failures > failuresBefore, 'Live calls failed');
+ const { status, body } = await ready();
+ assert.strictEqual(status, 200, 'Live is optional: still ready');
+ assert.strictEqual(body.status, 'degraded');
+ assert.deepStrictEqual(body.degraded, ['live_sync']);
+ assert.strictEqual(body.checks.live_sync.status, 'fail');
+ assert.match(body.checks.live_sync.error, /last successful Live sync \d+s ago/);
+ assert.ok(Date.parse(body.live.last_sync_at) > Date.parse(body.live.last_success_at), 'an attempt is not a success');
+});
+
+t('Live back: one clean pass makes it ready again', async () => {
+ h.live.down = false;
+ await h.ctx.sync();
+ const { status, body } = await ready();
+ assert.strictEqual(status, 200);
+ assert.strictEqual(body.status, 'ready');
+ assert.strictEqual(body.live.last_sync_at, body.live.last_success_at);
+});
+
+t('a step that misses its own schedule is late on its own (not masked by the fast steps)', async () => {
+ // 12 s from now the 10 s steps are past interval + 1 s; the 30 s and slower ones are not.
+ const s = h.ctx.syncStatus(1000, Date.now() + 12_000);
+ assert.ok(s.late_steps.includes('bans') && s.late_steps.includes('streams~'), JSON.stringify(s));
+ assert.ok(!s.late_steps.includes('users+') && !s.late_steps.includes('users*'), JSON.stringify(s));
+ assert.deepStrictEqual(h.ctx.syncStatus(1000).late_steps, []);
+});
+
+t('database failing: 503 not_ready naming the db check', async () => {
+ h.db.run('ALTER TABLE chat_messages RENAME TO chat_messages_away');
+ try {
+ const { status, body } = await ready();
+ assert.strictEqual(status, 503);
+ assert.strictEqual(body.ready, false);
+ assert.strictEqual(body.status, 'not_ready');
+ assert.deepStrictEqual(body.failed, ['db']);
+ assert.strictEqual(body.checks.db.status, 'fail');
+ assert.match(body.checks.db.error, /no such table/);
+ } finally {
+ h.db.run('ALTER TABLE chat_messages_away RENAME TO chat_messages');
+ }
+ const { status } = await ready();
+ assert.strictEqual(status, 200);
+});
+
+t.run(() => h && h.close());