From f22bcfb7130da79bc2ff88eb4b1e0faf0803ec79 Mon Sep 17 00:00:00 2001 From: thePR0M3TH3AN <53631862+PR0M3TH3AN@users.noreply.github.com> Date: Sat, 14 Feb 2026 19:08:53 -0500 Subject: [PATCH] Add bounded retry logic for lock publish quorum --- src/lock-ops.mjs | 125 ++++++++++++++++++++++++++------ test/lock-ops.test.mjs | 161 +++++++++++++++++++++++++++++++++++------ 2 files changed, 239 insertions(+), 47 deletions(-) diff --git a/src/lock-ops.mjs b/src/lock-ops.mjs index 8326f08..8c853c8 100644 --- a/src/lock-ops.mjs +++ b/src/lock-ops.mjs @@ -61,6 +61,36 @@ function relayListLabel(relays) { return relays.join(', '); } +function classifyPublishError(message) { + const normalized = String(message || '').toLowerCase(); + if (normalized.includes('timed out') || normalized.includes('timeout') || normalized.includes('etimedout')) { + return 'network_timeout'; + } + if (normalized.includes('econnreset') || normalized.includes('connection reset') || normalized.includes('socket hang up')) { + return 'connection_reset'; + } + if ( + normalized.includes('unavailable') || + normalized.includes('offline') || + normalized.includes('econnrefused') || + normalized.includes('connection refused') || + normalized.includes('enotfound') || + normalized.includes('503') + ) { + return 'relay_unavailable'; + } + return 'permanent_validation_error'; +} + +function isTransientPublishCategory(category) { + return category === 'network_timeout' || category === 'connection_reset' || category === 'relay_unavailable'; +} + +function calculateBackoffDelayMs(attemptNumber, baseMs, capMs, randomFn = Math.random) { + const maxDelay = Math.min(capMs, baseMs * (2 ** Math.max(0, attemptNumber - 1))); + return Math.floor(randomFn() * maxDelay); +} + export async function queryLocks(relays, cadence, dateStr, namespace, deps = {}) { const { poolFactory = () => new SimplePool(), @@ -137,26 +167,33 @@ export async function publishLock(relays, event, deps = {}) { getPublishTimeoutMsFn = getPublishTimeoutMs, getMinSuccessfulRelayPublishesFn = getMinSuccessfulRelayPublishes, getRelayFallbacksFn = getRelayFallbacks, + retryAttempts = 4, + retryBaseDelayMs = 500, + retryCapDelayMs = 8_000, + sleepFn = (ms) => new Promise((resolve) => setTimeout(resolve, ms)), + randomFn = Math.random, + telemetryLogger = console.error, } = deps; const pool = poolFactory(); const publishTimeoutMs = getPublishTimeoutMsFn(); const minSuccesses = getMinSuccessfulRelayPublishesFn(); const fallbackRelays = getRelayFallbacksFn().filter((relay) => !relays.includes(relay)); + const maxAttempts = Math.max(1, Math.floor(retryAttempts)); - const attempted = new Set(); - const publishResults = []; + const publishOnce = async () => { + const attempted = new Set(); + const publishResults = []; - const attemptPhase = async (phaseRelays, phaseName) => { - if (!phaseRelays.length) return; - for (const relay of phaseRelays) { - attempted.add(relay); - } - const phaseResults = await publishToRelays(pool, phaseRelays, event, publishTimeoutMs, phaseName); - publishResults.push(...phaseResults); - }; + const attemptPhase = async (phaseRelays, phaseName) => { + if (!phaseRelays.length) return; + for (const relay of phaseRelays) { + attempted.add(relay); + } + const phaseResults = await publishToRelays(pool, phaseRelays, event, publishTimeoutMs, phaseName); + publishResults.push(...phaseResults); + }; - try { await attemptPhase(relays, 'publish:primary'); let successCount = publishResults.filter((result) => result.success).length; @@ -165,23 +202,63 @@ export async function publishLock(relays, event, deps = {}) { successCount = publishResults.filter((result) => result.success).length; } - if (successCount < minSuccesses) { - const failureLines = publishResults - .filter((result) => !result.success) - .map((result) => `${result.relay} (${result.phase}): ${result.message}`); + const failures = publishResults + .filter((result) => !result.success) + .map((result) => ({ + ...result, + category: classifyPublishError(result.message), + })); - throw new Error( - `Failed relay publish quorum in publish phase: ${successCount}/${attempted.size} successful ` + - `(required=${minSuccesses}, timeout=${publishTimeoutMs}ms)\n ${failureLines.join('\n ')}`, - ); + return { attempted, publishResults, successCount, failures }; + }; + + try { + let lastAttemptState = null; + for (let attemptNumber = 1; attemptNumber <= maxAttempts; attemptNumber += 1) { + const startedAtMs = Date.now(); + lastAttemptState = await publishOnce(); + + if (lastAttemptState.successCount >= minSuccesses) { + console.error( + ` Published to ${lastAttemptState.successCount}/${lastAttemptState.attempted.size} relays ` + + `(required=${minSuccesses}, timeout=${publishTimeoutMs}ms)`, + ); + return event; + } + + const hasTransientFailure = lastAttemptState.failures.some((failure) => isTransientPublishCategory(failure.category)); + const hasPermanentFailure = lastAttemptState.failures.some((failure) => !isTransientPublishCategory(failure.category)); + const canRetry = attemptNumber < maxAttempts && hasTransientFailure && !hasPermanentFailure; + + if (!canRetry) { + break; + } + + const elapsedMs = Date.now() - startedAtMs; + const nextDelayMs = calculateBackoffDelayMs(attemptNumber, retryBaseDelayMs, retryCapDelayMs, randomFn); + for (const failure of lastAttemptState.failures) { + if (!isTransientPublishCategory(failure.category)) continue; + telemetryLogger(JSON.stringify({ + event: 'lock_publish_retry', + attempt: attemptNumber, + relayUrl: failure.relay, + errorCategory: failure.category, + elapsedMs, + nextDelayMs, + })); + } + + await sleepFn(nextDelayMs); } - console.error( - ` Published to ${successCount}/${attempted.size} relays ` + - `(required=${minSuccesses}, timeout=${publishTimeoutMs}ms)`, + const failureLines = lastAttemptState.failures + .map((result) => `${result.relay} (${result.phase}, category=${result.category}): ${result.message}`); + + throw new Error( + `Failed relay publish quorum in publish phase: ${lastAttemptState.successCount}/${lastAttemptState.attempted.size} successful ` + + `(required=${minSuccesses}, timeout=${publishTimeoutMs}ms, attempts=${maxAttempts})\n ${failureLines.join('\n ')}`, ); - return event; } finally { - pool.close([...attempted]); + pool.close([...new Set([...relays, ...fallbackRelays])]); } } diff --git a/test/lock-ops.test.mjs b/test/lock-ops.test.mjs index b79a7e4..70bd577 100644 --- a/test/lock-ops.test.mjs +++ b/test/lock-ops.test.mjs @@ -65,23 +65,155 @@ describe('lock-ops', () => { }); describe('publishLock', () => { - it('succeeds with partial primary success when minimum successes is 1', async () => { + it('retries transient failures and succeeds on a later attempt', async () => { + const telemetry = []; + const sleepCalls = []; + let publishCalls = 0; + const mockPool = { - publish: (relays) => relays.map((relay) => ( - relay.includes('ok') ? Promise.resolve('ok') : Promise.reject(new Error('offline')) - )), + publish: (relays) => { + publishCalls += 1; + if (publishCalls === 1) { + return relays.map(() => Promise.reject(new Error('connection reset by peer'))); + } + return relays.map(() => Promise.resolve('ok')); + }, close: () => {}, }; - const event = { id: 'evt-1' }; - const result = await publishLock(['wss://ok-1', 'wss://bad-1'], event, { + const event = { id: 'evt-retry' }; + const result = await publishLock(['wss://primary-a', 'wss://primary-b'], event, { poolFactory: () => mockPool, getPublishTimeoutMsFn: () => 200, getMinSuccessfulRelayPublishesFn: () => 1, getRelayFallbacksFn: () => [], + retryAttempts: 3, + retryBaseDelayMs: 500, + retryCapDelayMs: 8_000, + randomFn: () => 0.5, + sleepFn: async (ms) => { + sleepCalls.push(ms); + }, + telemetryLogger: (line) => { + telemetry.push(JSON.parse(line)); + }, }); assert.strictEqual(result, event); + assert.strictEqual(publishCalls, 2); + assert.deepStrictEqual(sleepCalls, [250]); + assert.strictEqual(telemetry.length, 2); + assert.deepStrictEqual(telemetry[0], { + event: 'lock_publish_retry', + attempt: 1, + relayUrl: 'wss://primary-a', + errorCategory: 'connection_reset', + elapsedMs: telemetry[0].elapsedMs, + nextDelayMs: 250, + }); + assert.deepStrictEqual(telemetry[1], { + event: 'lock_publish_retry', + attempt: 1, + relayUrl: 'wss://primary-b', + errorCategory: 'connection_reset', + elapsedMs: telemetry[1].elapsedMs, + nextDelayMs: 250, + }); + assert.ok(Number.isInteger(telemetry[0].elapsedMs)); + assert.ok(Number.isInteger(telemetry[1].elapsedMs)); + }); + + + it('fails after retry budget is exhausted for persistent transient failures', async () => { + let publishCalls = 0; + const sleepCalls = []; + + const mockPool = { + publish: (relays) => { + publishCalls += 1; + return relays.map(() => Promise.reject(new Error('relay unavailable'))); + }, + close: () => {}, + }; + + await assert.rejects( + () => publishLock(['wss://primary-down'], { id: 'evt-exhausted' }, { + poolFactory: () => mockPool, + getPublishTimeoutMsFn: () => 200, + getMinSuccessfulRelayPublishesFn: () => 1, + getRelayFallbacksFn: () => [], + retryAttempts: 3, + randomFn: () => 1, + sleepFn: async (ms) => { + sleepCalls.push(ms); + }, + telemetryLogger: () => {}, + }), + /Failed relay publish quorum in publish phase: 0\/1 successful \(required=1, timeout=200ms, attempts=3\)/, + ); + + assert.strictEqual(publishCalls, 3); + assert.deepStrictEqual(sleepCalls, [500, 1000]); + }); + + it('fails without retry for persistent validation failures', async () => { + let publishCalls = 0; + const mockPool = { + publish: (relays) => { + publishCalls += 1; + return relays.map(() => Promise.reject(new Error('validation failed: invalid signature'))); + }, + close: () => {}, + }; + + await assert.rejects( + () => publishLock(['wss://primary-bad'], { id: 'evt-permanent' }, { + poolFactory: () => mockPool, + getPublishTimeoutMsFn: () => 200, + getMinSuccessfulRelayPublishesFn: () => 1, + getRelayFallbacksFn: () => [], + retryAttempts: 4, + sleepFn: async () => { + throw new Error('should not sleep for permanent failures'); + }, + telemetryLogger: () => { + throw new Error('should not emit retry telemetry for permanent failures'); + }, + }), + /Failed relay publish quorum in publish phase: 0\/1 successful \(required=1, timeout=200ms, attempts=4\)/, + ); + + assert.strictEqual(publishCalls, 1); + }); + + it('returns success when mixed relay outcomes still satisfy quorum', async () => { + let publishCalls = 0; + const mockPool = { + publish: (relays) => { + publishCalls += 1; + return relays.map((relay) => ( + relay === 'wss://relay-ok' + ? Promise.resolve('ok') + : Promise.reject(new Error('validation failed: rejected event')) + )); + }, + close: () => {}, + }; + + const event = { id: 'evt-mixed' }; + const result = await publishLock(['wss://relay-ok', 'wss://relay-bad'], event, { + poolFactory: () => mockPool, + getPublishTimeoutMsFn: () => 200, + getMinSuccessfulRelayPublishesFn: () => 1, + getRelayFallbacksFn: () => [], + retryAttempts: 3, + sleepFn: async () => { + throw new Error('should not sleep when quorum is satisfied'); + }, + }); + + assert.strictEqual(result, event); + assert.strictEqual(publishCalls, 1); }); it('uses fallback relays to satisfy min success quorum', async () => { @@ -102,22 +234,5 @@ describe('lock-ops', () => { assert.strictEqual(result, event); }); - - it('fails when successful publishes stay below configured minimum', async () => { - const mockPool = { - publish: (relays) => relays.map(() => Promise.reject(new Error('offline'))), - close: () => {}, - }; - - await assert.rejects( - () => publishLock(['wss://primary-bad'], { id: 'evt-3' }, { - poolFactory: () => mockPool, - getPublishTimeoutMsFn: () => 200, - getMinSuccessfulRelayPublishesFn: () => 2, - getRelayFallbacksFn: () => ['wss://fallback-bad'], - }), - /Failed relay publish quorum in publish phase: 0\/2 successful \(required=2, timeout=200ms\)/, - ); - }); }); });