openvibers/openvibe.live Commits

PowerChat: honour 429 Retry-After with capped jittered backoff on every API call

OP OpenVibers committed 27 Aug 2026 · 53ecf903bc · 5 files, +234 −20

No rate-limit handling existed: a chat burst or a currency-earn flush that
tripped PowerChat's per-app window (120/min reads+chat, 60/min intake,
30/min alerts) just failed silently for the rest of the window.

powerchat-retry.js is a dependency-free policy (unit-tested): 429 waits
Retry-After (delta-seconds or HTTP-date) when present, else equal-jitter
exponential backoff capped at 8s; 5xx is replayed only for idempotent calls
(all intake POSTs carry messageId/externalId, display-only alert triggers
don't and are marked idempotent:false); other 4xx never retry; 3 attempts;
a Retry-After beyond a 30s budget ends the call rather than parking a relay.
apiRequest keeps its 401 refresh-and-replay-once inside each attempt and
logs a retry at most once per minute per (user, endpoint).

Co-Authored-By: Claude Fable 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_017AyeSFu7qdLdR64Zv99QNm

Changes

ddiff --git a/server/integrations/powerchat-oauth.js b/server/integrations/powerchat-oauth.js
iindex a02897c6..10c8b1bb 100644
---- a/server/integrations/powerchat-oauth.js
++++ b/server/integrations/powerchat-oauth.js
@@@ -246,14 +246,26 @@ async function getValidAccessToken(userId, { force = false } = {}) {
}
// ── Authenticated REST client ────────────────────────────────────────────────
--// path is relative to /streamers/:username, e.g. '/profile'. `username` defaults to the
--// streamer's stored PowerChat username (sandbox: the app owner's).
--async function apiRequest(userId, { method = 'GET', path, username, body, query } = {}) {
++// path is relative to /streamers/:username, e.g. '/profile' (or absolute under the
++// API root when `root: true`, e.g. '/me'). `username` defaults to the streamer's
++// stored PowerChat username (sandbox: the app owner's).
++//
++// Failure policy (see powerchat-retry.js): one authenticated attempt; a 401 forces a
++// token refresh and replays exactly once; 429 honours Retry-After / backs off with
++// jitter; 5xx is replayed only for idempotent calls (`idempotent: false` for
++// display-only alert triggers, which carry no idempotency key); every other 4xx is
++// terminal. `retry: false` disables the 429/5xx loop for callers that must not block.
++const retryPolicy = require('./powerchat-retry');
++const _rateWarned = new Map(); // `${userId}:${path}` → last warn (1/min so a burst logs once)
++async function apiRequest(userId, { method = 'GET', path, username, body, query, root = false, idempotent = true, retry = true } = {}) {
const c = getConfig();
const conn = db.getPowerchatConnection(userId);
const uname = username || (conn && conn.powerchat_username) || c.sandboxUsername;
-- let url = `${c.apiBase}/streamers/${encodeURIComponent(uname)}${path}`;
-- if (query) { const qs = new URLSearchParams(query).toString(); if (qs) url += `?${qs}`; }
++ let url = root ? `${c.apiBase}${path}` : `${c.apiBase}/streamers/${encodeURIComponent(uname)}${path}`;
++ if (query) {
++ const qs = new URLSearchParams(Object.entries(query).filter(([, v]) => v !== undefined && v !== null)).toString();
++ if (qs) url += `?${qs}`;
++ }
// One authenticated attempt; on a 401 (expired / rotated / "older credential
// generation") force a token refresh and retry exactly once.
@@@ -272,19 +284,34 @@ async function apiRequest(userId, { method = 'GET', path, username, body, query
return { res, json };
};
-- let { res, json } = await attempt(false);
-- if (res.status === 401) {
-- // Refresh-and-retry. If the refresh itself is rejected, getValidAccessToken
-- // clears the tokens (reconnect needed) and throws.
-- ({ res, json } = await attempt(true));
-- }
-- if (!res.ok) {
-- const e = new Error((json.error && json.error.message) || `PowerChat API ${res.status}`);
-- e.status = res.status;
-- e.code = json.error && json.error.code;
-- throw e;
-- }
-- return json;
++ const once = async () => {
++ let { res, json } = await attempt(false);
++ if (res.status === 401) {
++ // Refresh-and-retry. If the refresh itself is rejected, getValidAccessToken
++ // clears the tokens (reconnect needed) and throws.
++ ({ res, json } = await attempt(true));
++ }
++ if (!res.ok) {
++ const e = new Error((json.error && json.error.message) || `PowerChat API ${res.status}`);
++ e.status = res.status;
++ e.code = json.error && json.error.code;
++ if (res.status === 429) e.retryAfterMs = retryPolicy.parseRetryAfter(res.headers && res.headers.get && res.headers.get('retry-after'));
++ throw e;
++ }
++ return json;
++ };
++
++ return retryPolicy.withRetry(once, {
++ attempts: retry ? retryPolicy.DEFAULTS.attempts : 1,
++ idempotent,
++ onRetry: ({ attempt: n, status, delay }) => {
++ const key = `${userId}:${method} ${path}`;
++ const now = Date.now();
++ if ((_rateWarned.get(key) || 0) > now - 60000) return;
++ _rateWarned.set(key, now);
++ console.warn(`[PowerChat] ${status === 429 ? 'rate limited' : `HTTP ${status}`} on ${method} ${path} (user ${userId}) — retry ${n} in ${delay}ms`);
++ },
++ });
}
// Public profile + live status + tipPageUrl (scope profile:read).
ddiff --git a/server/integrations/powerchat-platform.js b/server/integrations/powerchat-platform.js
iindex 3d94baa7..a85ef402 100644
---- a/server/integrations/powerchat-platform.js
++++ b/server/integrations/powerchat-platform.js
@@@ -458,7 +458,8 @@ async function sendCustomAlert(streamerUserId, { actorName, message, amountCents
if (!conn) return false;
try {
await oauth.apiRequest(streamerUserId, {
-- method: 'POST', path: '/alerts/custom',
++ // No idempotency key on custom alerts → a replayed 5xx could double-alert.
++ method: 'POST', path: '/alerts/custom', idempotent: false,
// Schema: actorName 1-32, message ≤250, amountCents 0-100000000.
body: {
actorName: String(actorName || 'Test').slice(0, 32),
ddiff --git a/server/integrations/powerchat-retry.js b/server/integrations/powerchat-retry.js
nnew file mode 100644
iindex 00000000..23593ba1
---- /dev/null
++++ b/server/integrations/powerchat-retry.js
@@@ -0,0 +1,94 @@
++/**
++ * powerchat-retry.js — retry policy for PowerChat Developer API calls.
++ *
++ * PowerChat rate-limits per app with sliding windows (roughly 120/min reads + chat,
++ * 60/min subs/currency/tips/view-count, 30/min alerts) and answers 429 with a
++ * Retry-After header. Before this helper nothing honoured it: a chat burst or a
++ * currency-earn flush that tripped the limit just failed every call for the rest
++ * of the window (and, being best-effort, failed silently).
++ *
++ * Policy (deliberately small):
++ * - 429 → wait Retry-After when the server sends one, else exponential backoff
++ * with jitter (so a burst that got limited together doesn't retry together).
++ * - 5xx → same backoff, but ONLY when the caller says the call is idempotent
++ * (every intake POST carries an idempotency key — messageId / externalId — so
++ * a replay dedupes; display-only alert triggers have no key and must not be
++ * replayed into a double alert).
++ * - any other 4xx → never retried: a 400/401/403/404 fails identically forever.
++ * - capped attempts, and a Retry-After longer than we're willing to block on
++ * ends the call instead of parking a relay for a minute.
++ *
++ * Dependency-free on purpose: this file is unit-tested without the database.
++ */
++'use strict';
++
++const DEFAULTS = Object.freeze({
++ attempts: 3, // total tries, including the first
++ baseMs: 500, // first backoff step
++ capMs: 8000, // longest computed backoff
++ maxWaitMs: 30000, // longest Retry-After we honour before giving up instead
++});
++
++/** Retry-After → milliseconds. Accepts delta-seconds or an HTTP-date; null if absent/garbage. */
++function parseRetryAfter(value, now = Date.now()) {
++ if (value == null) return null;
++ const s = String(value).trim();
++ if (!s) return null;
++ if (/^\d+(\.\d+)?$/.test(s)) return Math.max(0, Math.round(Number(s) * 1000));
++ const t = Date.parse(s);
++ return Number.isFinite(t) ? Math.max(0, t - now) : null;
++}
++
++/** Is this HTTP status worth another try? 429 always; 5xx only for idempotent calls. */
++function isRetryable(status, { idempotent = true } = {}) {
++ if (status === 429) return true;
++ return idempotent && status >= 500 && status <= 599;
++}
++
++/**
++ * Delay after the given (1-based) failed attempt. A server Retry-After always wins;
++ * otherwise "equal jitter" backoff: uniformly inside [step/2, step] where
++ * step = min(cap, base·2^(attempt-1)).
++ */
++function backoffDelay({ attempt, retryAfterMs = null, baseMs = DEFAULTS.baseMs, capMs = DEFAULTS.capMs, random = Math.random } = {}) {
++ if (retryAfterMs != null && Number.isFinite(retryAfterMs)) return Math.max(0, Math.round(retryAfterMs));
++ const step = Math.min(capMs, baseMs * Math.pow(2, Math.max(0, attempt - 1)));
++ return Math.round(step / 2 + random() * (step / 2));
++}
++
++/**
++ * Run `fn(attempt)` with the policy above. `fn` must throw an Error carrying
++ * `.status` (HTTP status) and optionally `.retryAfterMs` for the policy to apply;
++ * anything else (network failure without a status) is not retried — the callers
++ * are best-effort relays and a dead network is not a rate limit.
++ *
++ * `sleep` / `random` / `now` are injectable for tests. `onRetry({attempt, status,
++ * delay})` is called before each wait.
++ */
++async function withRetry(fn, opts = {}) {
++ const attempts = Math.max(1, opts.attempts ?? DEFAULTS.attempts);
++ const maxWaitMs = opts.maxWaitMs ?? DEFAULTS.maxWaitMs;
++ const sleep = opts.sleep || ((ms) => new Promise((r) => setTimeout(r, ms)));
++ let lastErr;
++ for (let attempt = 1; attempt <= attempts; attempt++) {
++ try {
++ return await fn(attempt);
++ } catch (err) {
++ lastErr = err;
++ const status = err && Number(err.status);
++ if (!isRetryable(status, { idempotent: opts.idempotent !== false })) throw err;
++ if (attempt === attempts) throw err;
++ const delay = backoffDelay({
++ attempt,
++ retryAfterMs: err.retryAfterMs ?? null,
++ baseMs: opts.baseMs, capMs: opts.capMs, random: opts.random,
++ });
++ if (delay > maxWaitMs) { err.gaveUp = `Retry-After ${Math.round(delay / 1000)}s exceeds the ${Math.round(maxWaitMs / 1000)}s wait budget`; throw err; }
++ if (opts.onRetry) { try { opts.onRetry({ attempt, status, delay }); } catch { /* observer only */ } }
++ await sleep(delay);
++ }
++ }
++ throw lastErr;
++}
++
++module.exports = { DEFAULTS, parseRetryAfter, isRetryable, backoffDelay, withRetry };
ddiff --git a/server/integrations/powerchat-routes.js b/server/integrations/powerchat-routes.js
iindex f7c96d50..9af9ad47 100644
---- a/server/integrations/powerchat-routes.js
++++ b/server/integrations/powerchat-routes.js
@@@ -317,7 +317,7 @@ router.post('/test-alert', requireAuth, async (req, res) => {
host: { fromChannel: actorName, viewers: 42 },
};
await oauth.apiRequest(req.user.id, {
-- method: 'POST', path: '/test-alerts',
++ method: 'POST', path: '/test-alerts', idempotent: false,
body: { kind, payload: payloads[kind] },
});
}
ddiff --git a/test/powerchat-retry.test.js b/test/powerchat-retry.test.js
nnew file mode 100644
iindex 00000000..a2859e47
---- /dev/null
++++ b/test/powerchat-retry.test.js
@@@ -0,0 +1,92 @@
++// Unit tests for the PowerChat 429 / Retry-After policy (no DB, no network).
++const assert = require('assert');
++const { parseRetryAfter, isRetryable, backoffDelay, withRetry } = require('../server/integrations/powerchat-retry');
++
++const httpErr = (status, extra = {}) => Object.assign(new Error(`HTTP ${status}`), { status, ...extra });
++
++(async () => {
++ // ── parseRetryAfter ────────────────────────────────────────────
++ assert.strictEqual(parseRetryAfter('2'), 2000);
++ assert.strictEqual(parseRetryAfter(' 0.5 '), 500);
++ assert.strictEqual(parseRetryAfter('0'), 0);
++ const now = Date.parse('2026-08-27T10:00:00Z');
++ assert.strictEqual(parseRetryAfter('Thu, 27 Aug 2026 10:00:07 GMT', now), 7000);
++ assert.strictEqual(parseRetryAfter('Thu, 27 Aug 2026 09:59:00 GMT', now), 0, 'past date clamps to 0');
++ assert.strictEqual(parseRetryAfter(undefined), null);
++ assert.strictEqual(parseRetryAfter(''), null);
++ assert.strictEqual(parseRetryAfter('soon'), null);
++
++ // ── isRetryable ────────────────────────────────────────────────
++ assert.strictEqual(isRetryable(429), true);
++ assert.strictEqual(isRetryable(503), true);
++ assert.strictEqual(isRetryable(503, { idempotent: false }), false, '5xx is not replayed for non-idempotent calls');
++ assert.strictEqual(isRetryable(429, { idempotent: false }), true, '429 is always safe: nothing was processed');
++ for (const s of [400, 401, 403, 404, 409, 422]) assert.strictEqual(isRetryable(s), false, `${s} must never retry`);
++ assert.strictEqual(isRetryable(undefined), false);
++
++ // ── backoffDelay ───────────────────────────────────────────────
++ assert.strictEqual(backoffDelay({ attempt: 1, retryAfterMs: 1234, random: () => 0 }), 1234, 'Retry-After wins');
++ assert.strictEqual(backoffDelay({ attempt: 1, baseMs: 500, random: () => 0 }), 250, 'lower jitter bound is step/2');
++ assert.strictEqual(backoffDelay({ attempt: 1, baseMs: 500, random: () => 0.999999 }), 500);
++ assert.strictEqual(backoffDelay({ attempt: 3, baseMs: 500, random: () => 1 }), 2000, 'doubles per attempt');
++ assert.strictEqual(backoffDelay({ attempt: 10, baseMs: 500, capMs: 8000, random: () => 1 }), 8000, 'capped');
++ let lo = Infinity, hi = -Infinity;
++ for (let i = 0; i < 200; i++) { const d = backoffDelay({ attempt: 2 }); lo = Math.min(lo, d); hi = Math.max(hi, d); }
++ assert.ok(lo >= 500 && hi <= 1000, `jitter stays inside [500,1000] (got ${lo}..${hi})`);
++
++ // ── withRetry ──────────────────────────────────────────────────
++ const sleeps = [];
++ const sleep = async (ms) => { sleeps.push(ms); };
++
++ // 429 with Retry-After: waits exactly what the server said, then succeeds.
++ let calls = 0;
++ const r1 = await withRetry(async () => { calls++; if (calls < 3) throw httpErr(429, { retryAfterMs: 1500 }); return 'ok'; }, { sleep, attempts: 3 });
++ assert.strictEqual(r1, 'ok');
++ assert.strictEqual(calls, 3);
++ assert.deepStrictEqual(sleeps, [1500, 1500]);
++
++ // 429 without Retry-After: exponential backoff with jitter (deterministic random here).
++ sleeps.length = 0; calls = 0;
++ await withRetry(async () => { calls++; if (calls < 3) throw httpErr(429); return 1; }, { sleep, attempts: 3, baseMs: 100, random: () => 1 });
++ assert.deepStrictEqual(sleeps, [100, 200]);
++
++ // Attempts are capped: the last failure propagates unchanged.
++ sleeps.length = 0; calls = 0;
++ await assert.rejects(withRetry(async () => { calls++; throw httpErr(429); }, { sleep, attempts: 3, random: () => 0 }), (e) => e.status === 429);
++ assert.strictEqual(calls, 3);
++ assert.strictEqual(sleeps.length, 2, 'no sleep after the final attempt');
++
++ // A non-429 4xx is never retried.
++ for (const s of [400, 401, 403, 404]) {
++ sleeps.length = 0; calls = 0;
++ await assert.rejects(withRetry(async () => { calls++; throw httpErr(s); }, { sleep }), (e) => e.status === s);
++ assert.strictEqual(calls, 1, `${s} tried once`);
++ assert.strictEqual(sleeps.length, 0);
++ }
++
++ // 5xx: retried for idempotent calls, not for non-idempotent ones.
++ calls = 0;
++ assert.strictEqual(await withRetry(async () => { calls++; if (calls === 1) throw httpErr(502); return 'up'; }, { sleep, random: () => 0 }), 'up');
++ assert.strictEqual(calls, 2);
++ calls = 0;
++ await assert.rejects(withRetry(async () => { calls++; throw httpErr(502); }, { sleep, idempotent: false }), (e) => e.status === 502);
++ assert.strictEqual(calls, 1, 'non-idempotent 5xx is not replayed');
++
++ // A Retry-After beyond the wait budget ends the call instead of parking it.
++ sleeps.length = 0; calls = 0;
++ await assert.rejects(withRetry(async () => { calls++; throw httpErr(429, { retryAfterMs: 120000 }); }, { sleep, maxWaitMs: 30000 }), (e) => /wait budget/.test(e.gaveUp));
++ assert.strictEqual(calls, 1);
++ assert.strictEqual(sleeps.length, 0);
++
++ // Errors without an HTTP status (network) are not retried.
++ calls = 0;
++ await assert.rejects(withRetry(async () => { calls++; throw new Error('ECONNRESET'); }, { sleep }), /ECONNRESET/);
++ assert.strictEqual(calls, 1);
++
++ // onRetry observer sees attempt/status/delay and cannot break the loop.
++ const seen = [];
++ await withRetry(async (a) => { if (a < 2) throw httpErr(429, { retryAfterMs: 5 }); return a; }, { sleep, onRetry: (i) => { seen.push(i); throw new Error('observer bug'); } });
++ assert.deepStrictEqual(seen, [{ attempt: 1, status: 429, delay: 5 }]);
++
++ console.log('✅ powerchat retry policy tests passed');
++})().catch((e) => { console.error('❌', e); process.exit(1); });