From 481d77f33c430e9ed9783faa44d46a986c6ec172 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Sat, 21 Feb 2026 18:34:18 +0000 Subject: [PATCH] Refactor lock publisher to use shared error classifier and utils - Extracted `LockPublisher` and `publishLock` to `src/lock-publisher.mjs` (already existed, but cleaned up). - Refactored `src/lock-publisher.mjs` to import error constants and classification logic from `src/lock-error-classifier.mjs` instead of duplicating them. - Refactored `src/lock-publisher.mjs` and `src/lock-utils.mjs` to use `mergeRelayList` from `src/utils.mjs`. - Consolidated and moved publisher tests from `test/lock-ops.test.mjs` and `test/lock-ops-publish-retry.test.mjs` to `test/lock-publisher.test.mjs`. - Cleaned up `src/lock-utils.mjs` to remove duplicate `mergeRelayList`. - Verified changes with tests. Co-authored-by: PR0M3TH3AN <53631862+PR0M3TH3AN@users.noreply.github.com> --- src/lock-publisher.mjs | 99 +----- src/lock-utils.mjs | 8 +- test/lock-ops-publish-retry.test.mjs | 192 ---------- test/lock-ops.test.mjs | 319 ----------------- test/lock-publisher.test.mjs | 508 +++++++++++++++++++++++++++ 5 files changed, 516 insertions(+), 610 deletions(-) delete mode 100644 test/lock-ops-publish-retry.test.mjs create mode 100644 test/lock-publisher.test.mjs diff --git a/src/lock-publisher.mjs b/src/lock-publisher.mjs index 08d0fb5..09c9287 100644 --- a/src/lock-publisher.mjs +++ b/src/lock-publisher.mjs @@ -6,100 +6,15 @@ import { getRelayFallbacks, getMinActiveRelayPool, } from './torch-config.mjs'; -import { mergeRelayList, relayListLabel, withTimeout } from './utils.mjs'; +import { relayListLabel, withTimeout, mergeRelayList } from './utils.mjs'; import { defaultHealthManager, buildRelayHealthConfig } from './relay-health-manager.mjs'; import { secureRandom } from './lock-utils.mjs'; - -const PUBLISH_ERROR_CODES = { - TIMEOUT: 'publish_timeout', - DNS: 'dns_resolution', - TCP: 'tcp_connect_timeout', - TLS: 'tls_handshake', - WEBSOCKET: 'websocket_open_failure', - NETWORK: 'network_timeout', - CONNECTION_RESET: 'connection_reset', - RELAY_UNAVAILABLE: 'relay_unavailable', - PERMANENT: 'permanent_validation_error', -}; - -const PUBLISH_FAILURE_CATEGORIES = { - QUORUM_FAILURE: 'relay_publish_quorum_failure', - NON_RETRYABLE: 'relay_publish_non_retryable', -}; - -/** - * Classifies a raw error message into a standardized publication error code. - * Used to determine if a failure is transient (retryable) or permanent. - * - * @param {string|Error} message - The error message or object to classify. - * @returns {string} One of the PUBLISH_ERROR_CODES constants. - */ -function classifyPublishError(message) { - const normalized = String(message || '').toLowerCase(); - if (normalized.includes('publish timed out after') || normalized.includes('publish timeout')) { - return PUBLISH_ERROR_CODES.TIMEOUT; - } - if ( - normalized.includes('enotfound') - || normalized.includes('eai_again') - || normalized.includes('getaddrinfo') - || (normalized.includes('dns') && normalized.includes('websocket')) - ) { - return PUBLISH_ERROR_CODES.DNS; - } - if ( - normalized.includes('connect etimedout') - || normalized.includes('tcp connect timed out') - || normalized.includes('connect timeout') - ) { - return PUBLISH_ERROR_CODES.TCP; - } - if ( - normalized.includes('tls') - || normalized.includes('ssl') - || normalized.includes('certificate') - || normalized.includes('handshake') - ) { - return PUBLISH_ERROR_CODES.TLS; - } - if ( - normalized.includes('websocket') - || normalized.includes('bad response') - || normalized.includes('unexpected server response') - ) { - return PUBLISH_ERROR_CODES.WEBSOCKET; - } - if (normalized.includes('timed out') || normalized.includes('timeout') || normalized.includes('etimedout')) { - return PUBLISH_ERROR_CODES.NETWORK; - } - if (normalized.includes('econnreset') || normalized.includes('connection reset') || normalized.includes('socket hang up')) { - return PUBLISH_ERROR_CODES.CONNECTION_RESET; - } - if ( - normalized.includes('unavailable') - || normalized.includes('offline') - || normalized.includes('econnrefused') - || normalized.includes('connection refused') - || normalized.includes('enotfound') - || normalized.includes('503') - ) { - return PUBLISH_ERROR_CODES.RELAY_UNAVAILABLE; - } - return PUBLISH_ERROR_CODES.PERMANENT; -} - -function isTransientPublishCategory(category) { - return [ - PUBLISH_ERROR_CODES.TIMEOUT, - PUBLISH_ERROR_CODES.DNS, - PUBLISH_ERROR_CODES.TCP, - PUBLISH_ERROR_CODES.TLS, - PUBLISH_ERROR_CODES.WEBSOCKET, - PUBLISH_ERROR_CODES.NETWORK, - PUBLISH_ERROR_CODES.CONNECTION_RESET, - PUBLISH_ERROR_CODES.RELAY_UNAVAILABLE, - ].includes(category); -} +import { + PUBLISH_ERROR_CODES, + PUBLISH_FAILURE_CATEGORIES, + classifyPublishError, + isTransientPublishCategory, +} from './lock-error-classifier.mjs'; function calculateBackoffDelayMs(attemptNumber, baseMs, capMs, randomFn = secureRandom) { const maxDelay = Math.min(capMs, baseMs * (2 ** Math.max(0, attemptNumber - 1))); diff --git a/src/lock-utils.mjs b/src/lock-utils.mjs index 8293fb2..72ef4c9 100644 --- a/src/lock-utils.mjs +++ b/src/lock-utils.mjs @@ -3,7 +3,7 @@ import path from 'node:path'; import { randomInt } from 'node:crypto'; import { todayDateStr, getIsoWeekStr } from './utils.mjs'; -export { relayListLabel } from './utils.mjs'; +export { relayListLabel, withTimeout, mergeRelayList } from './utils.mjs'; /** * Scans the local log directory to identify agents that have already completed their task @@ -51,12 +51,6 @@ export async function getCompletedAgents(cadence, logDir, deps) { return completed; } -export { withTimeout } from './utils.mjs'; - -export function mergeRelayList(primaryRelays, fallbackRelays) { - return [...new Set([...primaryRelays, ...fallbackRelays])]; -} - const MAX_RANDOM = 281474976710655; // 2**48 - 1 export function secureRandom() { diff --git a/test/lock-ops-publish-retry.test.mjs b/test/lock-ops-publish-retry.test.mjs deleted file mode 100644 index ad35d14..0000000 --- a/test/lock-ops-publish-retry.test.mjs +++ /dev/null @@ -1,192 +0,0 @@ -import { describe, it, beforeEach, afterEach, mock } from 'node:test'; -import assert from 'node:assert'; -import { publishLock, _resetRelayHealthState } from '../src/lock-ops.mjs'; - -function createScriptedPool(scriptedOutcomesByAttempt, relaysSeen) { - let publishCallCount = 0; - return { - publish: (relays) => { - publishCallCount += 1; - relaysSeen.push({ publishAttempt: publishCallCount, relays: [...relays] }); - const attemptOutcomes = scriptedOutcomesByAttempt[publishCallCount - 1] || {}; - return relays.map((relay) => { - const outcome = attemptOutcomes[relay] || { ok: true }; - return outcome.ok ? Promise.resolve('ok') : Promise.reject(new Error(outcome.message)); - }); - }, - close: () => {}, - getPublishCallCount: () => publishCallCount, - }; -} - -const parseTelemetry = (logs) => logs.map((line) => JSON.parse(line)); - -describe('publishLock retry behavior', { concurrency: false }, () => { - beforeEach(() => { - _resetRelayHealthState(); - }); - - afterEach(() => { - mock.timers.reset(); - }); - - it('succeeds immediately on first attempt without retries', async () => { - const telemetryLogs = []; - const relaysSeen = []; - const mockPool = createScriptedPool([ - { 'wss://relay-a': { ok: true } }, - ], relaysSeen); - - const event = { id: 'evt-immediate' }; - const result = await publishLock(['wss://relay-a'], event, { - poolFactory: () => mockPool, - getPublishTimeoutMsFn: () => 200, - getMinSuccessfulRelayPublishesFn: () => 1, - getRelayFallbacksFn: () => [], - getMinActiveRelayPoolFn: () => 1, - retryAttempts: 4, - telemetryLogger: (line) => telemetryLogs.push(line), - healthLogger: () => {}, - }); - - assert.strictEqual(result, event); - assert.strictEqual(mockPool.getPublishCallCount(), 1); - assert.deepStrictEqual(relaysSeen, [{ publishAttempt: 1, relays: ['wss://relay-a'] }]); - - const telemetry = parseTelemetry(telemetryLogs); - assert.strictEqual(telemetry.filter((entry) => entry.event === 'lock_publish_retry').length, 0); - const metEvent = telemetry.find((entry) => entry.event === 'lock_publish_quorum_met'); - assert.ok(metEvent); - assert.strictEqual(metEvent.publishAttempt, 1); - }); - - it('retries a transient failure then succeeds within retry budget with bounded jitter', async () => { - mock.timers.enable({ apis: ['Date', 'setTimeout'] }); - - const telemetryLogs = []; - const relaysSeen = []; - const sleepCalls = []; - const mockPool = createScriptedPool([ - { 'wss://relay-a': { ok: false, message: 'connection reset by peer' } }, - { 'wss://relay-a': { ok: true } }, - ], relaysSeen); - - const result = await publishLock(['wss://relay-a'], { id: 'evt-retry-success' }, { - poolFactory: () => mockPool, - getPublishTimeoutMsFn: () => 200, - getMinSuccessfulRelayPublishesFn: () => 1, - getRelayFallbacksFn: () => [], - getMinActiveRelayPoolFn: () => 1, - retryAttempts: 3, - retryBaseDelayMs: 100, - retryCapDelayMs: 500, - randomFn: () => 0.25, - sleepFn: async (delayMs) => { - sleepCalls.push(delayMs); - mock.timers.tick(delayMs); - }, - telemetryLogger: (line) => telemetryLogs.push(line), - healthLogger: () => {}, - }); - - assert.strictEqual(result.id, 'evt-retry-success'); - assert.deepStrictEqual(sleepCalls, [25]); - assert.strictEqual(mockPool.getPublishCallCount(), 2); - - const telemetry = parseTelemetry(telemetryLogs); - const retryEvents = telemetry.filter((entry) => entry.event === 'lock_publish_retry'); - assert.strictEqual(retryEvents.length, 1); - assert.strictEqual(retryEvents[0].publishAttempt, 1); - assert.strictEqual(retryEvents[0].relayUrl, 'wss://relay-a'); - assert.ok(retryEvents[0].nextDelayMs >= 0 && retryEvents[0].nextDelayMs <= 100); - }); - - it('returns terminal relay_publish_quorum_failure after exhausting retry budget', async () => { - mock.timers.enable({ apis: ['Date', 'setTimeout'] }); - - const telemetryLogs = []; - const relaysSeen = []; - const sleepCalls = []; - const randomValues = [0.1, 0.9]; - const mockPool = createScriptedPool([ - { 'wss://relay-a': { ok: false, message: 'relay unavailable' } }, - { 'wss://relay-a': { ok: false, message: 'relay unavailable' } }, - { 'wss://relay-a': { ok: false, message: 'relay unavailable' } }, - ], relaysSeen); - - await assert.rejects( - () => publishLock(['wss://relay-a'], { id: 'evt-exhausted' }, { - poolFactory: () => mockPool, - getPublishTimeoutMsFn: () => 200, - getMinSuccessfulRelayPublishesFn: () => 1, - getRelayFallbacksFn: () => [], - getMinActiveRelayPoolFn: () => 1, - retryAttempts: 3, - retryBaseDelayMs: 100, - retryCapDelayMs: 250, - randomFn: () => randomValues.shift() ?? 0, - sleepFn: async (delayMs) => { - sleepCalls.push(delayMs); - mock.timers.tick(delayMs); - }, - telemetryLogger: (line) => telemetryLogs.push(line), - healthLogger: () => {}, - }), - /error_category=relay_publish_quorum_failure/, - ); - - assert.strictEqual(mockPool.getPublishCallCount(), 3); - assert.deepStrictEqual(sleepCalls, [10, 180]); - assert.deepStrictEqual(relaysSeen.map((entry) => entry.publishAttempt), [1, 2, 3]); - - const telemetry = parseTelemetry(telemetryLogs); - const retryEvents = telemetry.filter((entry) => entry.event === 'lock_publish_retry'); - assert.strictEqual(retryEvents.length, 2); - assert.strictEqual(retryEvents[0].publishAttempt, 1); - assert.strictEqual(retryEvents[0].relayUrl, 'wss://relay-a'); - assert.ok(retryEvents[0].nextDelayMs >= 0 && retryEvents[0].nextDelayMs <= 100); - assert.strictEqual(retryEvents[1].publishAttempt, 2); - assert.strictEqual(retryEvents[1].relayUrl, 'wss://relay-a'); - assert.ok(retryEvents[1].nextDelayMs >= retryEvents[0].nextDelayMs); - assert.ok(retryEvents[1].nextDelayMs <= 200); - - const failedEvent = telemetry.find((entry) => entry.event === 'lock_publish_quorum_failed'); - assert.ok(failedEvent); - assert.strictEqual(failedEvent.errorCategory, 'relay_publish_quorum_failure'); - assert.strictEqual(failedEvent.attempts, 3); - }); - - it('short-circuits non-retryable failures without extra attempts', async () => { - const telemetryLogs = []; - const relaysSeen = []; - const mockPool = createScriptedPool([ - { 'wss://relay-a': { ok: false, message: 'validation failed: invalid signature' } }, - { 'wss://relay-a': { ok: true } }, - ], relaysSeen); - - await assert.rejects( - () => publishLock(['wss://relay-a'], { id: 'evt-permanent' }, { - poolFactory: () => mockPool, - getPublishTimeoutMsFn: () => 200, - getMinSuccessfulRelayPublishesFn: () => 1, - getRelayFallbacksFn: () => [], - getMinActiveRelayPoolFn: () => 1, - retryAttempts: 4, - telemetryLogger: (line) => telemetryLogs.push(line), - healthLogger: () => {}, - }), - /error_category=relay_publish_non_retryable/, - ); - - assert.strictEqual(mockPool.getPublishCallCount(), 1); - assert.deepStrictEqual(relaysSeen, [{ publishAttempt: 1, relays: ['wss://relay-a'] }]); - - const telemetry = parseTelemetry(telemetryLogs); - assert.strictEqual(telemetry.filter((entry) => entry.event === 'lock_publish_retry').length, 0); - - const failedEvent = telemetry.find((entry) => entry.event === 'lock_publish_quorum_failed'); - assert.ok(failedEvent); - assert.strictEqual(failedEvent.errorCategory, 'relay_publish_non_retryable'); - assert.notStrictEqual(failedEvent.errorCategory, 'relay_publish_quorum_failure'); - }); -}); diff --git a/test/lock-ops.test.mjs b/test/lock-ops.test.mjs index 74d8e1c..907fc13 100644 --- a/test/lock-ops.test.mjs +++ b/test/lock-ops.test.mjs @@ -2,7 +2,6 @@ import { describe, it, beforeEach } from 'node:test'; import assert from 'node:assert'; import { parseLockEvent, - publishLock, queryLocks, _resetRelayHealthState, } from '../src/lock-ops.mjs'; @@ -111,322 +110,4 @@ describe('lock-ops', () => { assert.strictEqual(locks[0].agent, 'agent1'); }); }); - - describe('publishLock', () => { - it('retries transient failures and succeeds on a later attempt', async () => { - const telemetry = []; - const sleepCalls = []; - let publishCalls = 0; - - const mockPool = { - 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-retry' }; - const result = await publishLock(['wss://primary-a', 'wss://primary-b'], event, { - poolFactory: () => mockPool, - getPublishTimeoutMsFn: () => 200, - getMinSuccessfulRelayPublishesFn: () => 1, - getRelayFallbacksFn: () => [], - getMinActiveRelayPoolFn: () => 1, - retryAttempts: 3, - retryBaseDelayMs: 500, - retryCapDelayMs: 8_000, - randomFn: () => 0.5, - sleepFn: async (ms) => { - sleepCalls.push(ms); - }, - telemetryLogger: (line) => { - telemetry.push(JSON.parse(line)); - }, - healthLogger: () => {}, - diagnostics: { - attemptId: 'attempt-1', - correlationId: 'corr-1', - }, - }); - - assert.strictEqual(result, event); - assert.strictEqual(publishCalls, 2); - assert.deepStrictEqual(sleepCalls, [250]); - - const retryEvents = telemetry.filter((entry) => entry.event === 'lock_publish_retry'); - assert.strictEqual(retryEvents.length, 2); - assert.deepStrictEqual(retryEvents[0], { - event: 'lock_publish_retry', - correlationId: 'corr-1', - attemptId: 'attempt-1', - publishAttempt: 1, - relayUrl: 'wss://primary-a', - errorCategory: 'connection_reset', - elapsedMs: retryEvents[0].elapsedMs, - nextDelayMs: 250, - }); - assert.deepStrictEqual(retryEvents[1], { - event: 'lock_publish_retry', - correlationId: 'corr-1', - attemptId: 'attempt-1', - publishAttempt: 1, - relayUrl: 'wss://primary-b', - errorCategory: 'connection_reset', - elapsedMs: retryEvents[1].elapsedMs, - nextDelayMs: 250, - }); - assert.ok(Number.isInteger(retryEvents[0].elapsedMs)); - assert.ok(Number.isInteger(retryEvents[1].elapsedMs)); - - const metEvent = telemetry.find((entry) => entry.event === 'lock_publish_quorum_met'); - assert.ok(metEvent); - assert.equal(metEvent.correlationId, 'corr-1'); - assert.equal(metEvent.attemptId, 'attempt-1'); - assert.equal(metEvent.publishAttempt, 2); - assert.ok(Array.isArray(metEvent.retryTimeline)); - }); - - 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: () => [], - getMinActiveRelayPoolFn: () => 1, - retryAttempts: 3, - randomFn: () => 1, - sleepFn: async (ms) => { - sleepCalls.push(ms); - }, - telemetryLogger: () => {}, - healthLogger: () => {}, - }), - /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: () => [], - getMinActiveRelayPoolFn: () => 1, - retryAttempts: 4, - sleepFn: async () => { - throw new Error('should not sleep for permanent failures'); - }, - telemetryLogger: () => {}, - healthLogger: () => {}, - }), - /Failed relay publish quorum in publish phase: 0\/1 successful \(required=1, timeout=200ms, attempts=4,/, - ); - - assert.strictEqual(publishCalls, 1); - }); - - it('includes per-relay reason categories and retry metadata in failure diagnostics', async () => { - const telemetry = []; - const mockPool = { - publish: (relays) => relays.map((relay) => { - if (relay.includes('dns')) return Promise.reject(new Error('getaddrinfo ENOTFOUND relay.example')); - if (relay.includes('tcp')) return Promise.reject(new Error('connect ETIMEDOUT 1.2.3.4:443')); - if (relay.includes('tls')) return Promise.reject(new Error('TLS handshake failure')); - if (relay.includes('ws')) return Promise.reject(new Error('websocket: unexpected server response')); - return Promise.reject(new Error('[publish:primary] Publish timed out after 200ms')); - }), - close: () => {}, - }; - - await assert.rejects( - () => publishLock(['wss://dns', 'wss://tcp', 'wss://tls', 'wss://ws', 'relay-timeout'], { id: 'evt-diag' }, { - poolFactory: () => mockPool, - getPublishTimeoutMsFn: () => 200, - getMinSuccessfulRelayPublishesFn: () => 1, - getRelayFallbacksFn: () => [], - getMinActiveRelayPoolFn: () => 1, - retryAttempts: 1, - telemetryLogger: (line) => telemetry.push(JSON.parse(line)), - healthLogger: () => {}, - diagnostics: { attemptId: 'attempt-42', correlationId: 'corr-42' }, - }), - ); - - const summaryEvent = telemetry.find((entry) => entry.event === 'lock_publish_quorum_failed'); - assert.ok(summaryEvent); - const failureEvents = telemetry.filter((entry) => entry.event === 'lock_publish_failure'); - const failureText = failureEvents.map((entry) => `${entry.relayUrl}:${entry.reason}`).join('\n'); - - assert.match(failureText, /wss:\/\/dns:dns_resolution/); - assert.match(failureText, /wss:\/\/tcp:tcp_connect_timeout/); - assert.match(failureText, /wss:\/\/tls:tls_handshake/); - assert.match(failureText, /wss:\/\/ws:websocket_open_failure/); - assert.match(failureText, /relay-timeout:publish_timeout/); - - - assert.deepStrictEqual(summaryEvent.reasonDistribution, { - dns_resolution: 1, - tcp_connect_timeout: 1, - tls_handshake: 1, - websocket_open_failure: 1, - publish_timeout: 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: () => [], - getMinActiveRelayPoolFn: () => 1, - retryAttempts: 3, - sleepFn: async () => { - throw new Error('should not sleep when quorum is satisfied'); - }, - healthLogger: () => {}, - }); - - assert.strictEqual(result, event); - assert.strictEqual(publishCalls, 1); - }); - - it('uses fallback relays to satisfy min success quorum', async () => { - const mockPool = { - publish: (relays) => relays.map((relay) => ( - relay.includes('fallback-ok') ? Promise.resolve('ok') : Promise.reject(new Error('offline')) - )), - close: () => {}, - }; - - const event = { id: 'evt-2' }; - const result = await publishLock(['wss://primary-bad'], event, { - poolFactory: () => mockPool, - getPublishTimeoutMsFn: () => 200, - getMinSuccessfulRelayPublishesFn: () => 1, - getRelayFallbacksFn: () => ['wss://fallback-ok'], - getMinActiveRelayPoolFn: () => 1, - healthLogger: () => {}, - }); - - assert.strictEqual(result, event); - }); - - it('quarantines repeatedly failing relays and still reaches quorum with one healthy relay', async () => { - const callOrder = []; - const mockPool = { - publish: (relays) => relays.map((relay) => { - callOrder.push(relay); - if (relay === 'wss://healthy') { - return Promise.resolve('ok'); - } - return Promise.reject(new Error('offline')); - }), - close: () => {}, - }; - - const event = { id: 'evt-health' }; - - for (let i = 0; i < 3; i += 1) { - await publishLock(['wss://bad', 'wss://healthy'], event, { - poolFactory: () => mockPool, - getPublishTimeoutMsFn: () => 200, - getMinSuccessfulRelayPublishesFn: () => 1, - getRelayFallbacksFn: () => [], - getMinActiveRelayPoolFn: () => 1, - retryAttempts: 1, - failureThreshold: 2, - quarantineCooldownMs: 60_000, - healthLogger: () => {}, - }); - } - - callOrder.length = 0; - await publishLock(['wss://bad', 'wss://healthy'], event, { - poolFactory: () => mockPool, - getPublishTimeoutMsFn: () => 200, - getMinSuccessfulRelayPublishesFn: () => 1, - getRelayFallbacksFn: () => [], - getMinActiveRelayPoolFn: () => 1, - retryAttempts: 1, - failureThreshold: 2, - quarantineCooldownMs: 60_000, - healthLogger: () => {}, - }); - - assert.deepStrictEqual(callOrder, ['wss://healthy']); - }); - - it('uses fallback and min active pool to reintroduce quarantined relay when needed', async () => { - const attempts = []; - const mockPool = { - publish: (relays) => relays.map((relay) => { - attempts.push(relay); - if (relay === 'wss://fallback-healthy') { - return Promise.resolve('ok'); - } - return Promise.reject(new Error('relay timeout')); - }), - close: () => {}, - }; - - const event = { id: 'evt-quorum' }; - await publishLock(['wss://primary-a', 'wss://primary-b'], event, { - poolFactory: () => mockPool, - getPublishTimeoutMsFn: () => 200, - getMinSuccessfulRelayPublishesFn: () => 1, - getRelayFallbacksFn: () => ['wss://fallback-healthy'], - getMinActiveRelayPoolFn: () => 2, - retryAttempts: 1, - failureThreshold: 1, - quarantineCooldownMs: 60_000, - healthLogger: () => {}, - }); - - assert.ok(attempts.includes('wss://fallback-healthy')); - }); - }); }); diff --git a/test/lock-publisher.test.mjs b/test/lock-publisher.test.mjs new file mode 100644 index 0000000..e6e0fac --- /dev/null +++ b/test/lock-publisher.test.mjs @@ -0,0 +1,508 @@ +import { describe, it, beforeEach, afterEach, mock } from 'node:test'; +import assert from 'node:assert'; +import { publishLock } from '../src/lock-publisher.mjs'; +import { defaultHealthManager } from '../src/relay-health-manager.mjs'; + +function _resetRelayHealthState() { + defaultHealthManager.reset(); +} + +function createScriptedPool(scriptedOutcomesByAttempt, relaysSeen) { + let publishCallCount = 0; + return { + publish: (relays) => { + publishCallCount += 1; + relaysSeen.push({ publishAttempt: publishCallCount, relays: [...relays] }); + const attemptOutcomes = scriptedOutcomesByAttempt[publishCallCount - 1] || {}; + return relays.map((relay) => { + const outcome = attemptOutcomes[relay] || { ok: true }; + return outcome.ok ? Promise.resolve('ok') : Promise.reject(new Error(outcome.message)); + }); + }, + close: () => {}, + getPublishCallCount: () => publishCallCount, + }; +} + +const parseTelemetry = (logs) => logs.map((line) => JSON.parse(line)); + +describe('LockPublisher', () => { + beforeEach(() => { + _resetRelayHealthState(); + }); + + afterEach(() => { + mock.timers.reset(); + }); + + describe('publishLock basic and retry logic', () => { + it('retries transient failures and succeeds on a later attempt', async () => { + const telemetry = []; + const sleepCalls = []; + let publishCalls = 0; + + const mockPool = { + 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-retry' }; + const result = await publishLock(['wss://primary-a', 'wss://primary-b'], event, { + poolFactory: () => mockPool, + getPublishTimeoutMsFn: () => 200, + getMinSuccessfulRelayPublishesFn: () => 1, + getRelayFallbacksFn: () => [], + getMinActiveRelayPoolFn: () => 1, + retryAttempts: 3, + retryBaseDelayMs: 500, + retryCapDelayMs: 8_000, + randomFn: () => 0.5, + sleepFn: async (ms) => { + sleepCalls.push(ms); + }, + telemetryLogger: (line) => { + telemetry.push(JSON.parse(line)); + }, + healthLogger: () => {}, + diagnostics: { + attemptId: 'attempt-1', + correlationId: 'corr-1', + }, + }); + + assert.strictEqual(result, event); + assert.strictEqual(publishCalls, 2); + assert.deepStrictEqual(sleepCalls, [250]); + + const retryEvents = telemetry.filter((entry) => entry.event === 'lock_publish_retry'); + assert.strictEqual(retryEvents.length, 2); + assert.deepStrictEqual(retryEvents[0], { + event: 'lock_publish_retry', + correlationId: 'corr-1', + attemptId: 'attempt-1', + publishAttempt: 1, + relayUrl: 'wss://primary-a', + errorCategory: 'connection_reset', + elapsedMs: retryEvents[0].elapsedMs, + nextDelayMs: 250, + }); + assert.deepStrictEqual(retryEvents[1], { + event: 'lock_publish_retry', + correlationId: 'corr-1', + attemptId: 'attempt-1', + publishAttempt: 1, + relayUrl: 'wss://primary-b', + errorCategory: 'connection_reset', + elapsedMs: retryEvents[1].elapsedMs, + nextDelayMs: 250, + }); + }); + + 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: () => [], + getMinActiveRelayPoolFn: () => 1, + retryAttempts: 3, + randomFn: () => 1, + sleepFn: async (ms) => { + sleepCalls.push(ms); + }, + telemetryLogger: () => {}, + healthLogger: () => {}, + }), + /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: () => [], + getMinActiveRelayPoolFn: () => 1, + retryAttempts: 4, + sleepFn: async () => { + throw new Error('should not sleep for permanent failures'); + }, + telemetryLogger: () => {}, + healthLogger: () => {}, + }), + /Failed relay publish quorum in publish phase: 0\/1 successful \(required=1, timeout=200ms, attempts=4,/, + ); + + assert.strictEqual(publishCalls, 1); + }); + + it('includes per-relay reason categories and retry metadata in failure diagnostics', async () => { + const telemetry = []; + const mockPool = { + publish: (relays) => relays.map((relay) => { + if (relay.includes('dns')) return Promise.reject(new Error('getaddrinfo ENOTFOUND relay.example')); + if (relay.includes('tcp')) return Promise.reject(new Error('connect ETIMEDOUT 1.2.3.4:443')); + if (relay.includes('tls')) return Promise.reject(new Error('TLS handshake failure')); + if (relay.includes('ws')) return Promise.reject(new Error('websocket: unexpected server response')); + return Promise.reject(new Error('[publish:primary] Publish timed out after 200ms')); + }), + close: () => {}, + }; + + await assert.rejects( + () => publishLock(['wss://dns', 'wss://tcp', 'wss://tls', 'wss://ws', 'relay-timeout'], { id: 'evt-diag' }, { + poolFactory: () => mockPool, + getPublishTimeoutMsFn: () => 200, + getMinSuccessfulRelayPublishesFn: () => 1, + getRelayFallbacksFn: () => [], + getMinActiveRelayPoolFn: () => 1, + retryAttempts: 1, + telemetryLogger: (line) => telemetry.push(JSON.parse(line)), + healthLogger: () => {}, + diagnostics: { attemptId: 'attempt-42', correlationId: 'corr-42' }, + }), + ); + + const summaryEvent = telemetry.find((entry) => entry.event === 'lock_publish_quorum_failed'); + assert.ok(summaryEvent); + const failureEvents = telemetry.filter((entry) => entry.event === 'lock_publish_failure'); + const failureText = failureEvents.map((entry) => `${entry.relayUrl}:${entry.reason}`).join('\n'); + + assert.match(failureText, /wss:\/\/dns:dns_resolution/); + assert.match(failureText, /wss:\/\/tcp:tcp_connect_timeout/); + assert.match(failureText, /wss:\/\/tls:tls_handshake/); + assert.match(failureText, /wss:\/\/ws:websocket_open_failure/); + assert.match(failureText, /relay-timeout:publish_timeout/); + + + assert.deepStrictEqual(summaryEvent.reasonDistribution, { + dns_resolution: 1, + tcp_connect_timeout: 1, + tls_handshake: 1, + websocket_open_failure: 1, + publish_timeout: 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: () => [], + getMinActiveRelayPoolFn: () => 1, + retryAttempts: 3, + sleepFn: async () => { + throw new Error('should not sleep when quorum is satisfied'); + }, + healthLogger: () => {}, + }); + + assert.strictEqual(result, event); + assert.strictEqual(publishCalls, 1); + }); + + it('uses fallback relays to satisfy min success quorum', async () => { + const mockPool = { + publish: (relays) => relays.map((relay) => ( + relay.includes('fallback-ok') ? Promise.resolve('ok') : Promise.reject(new Error('offline')) + )), + close: () => {}, + }; + + const event = { id: 'evt-2' }; + const result = await publishLock(['wss://primary-bad'], event, { + poolFactory: () => mockPool, + getPublishTimeoutMsFn: () => 200, + getMinSuccessfulRelayPublishesFn: () => 1, + getRelayFallbacksFn: () => ['wss://fallback-ok'], + getMinActiveRelayPoolFn: () => 1, + healthLogger: () => {}, + }); + + assert.strictEqual(result, event); + }); + + it('quarantines repeatedly failing relays and still reaches quorum with one healthy relay', async () => { + const callOrder = []; + const mockPool = { + publish: (relays) => relays.map((relay) => { + callOrder.push(relay); + if (relay === 'wss://healthy') { + return Promise.resolve('ok'); + } + return Promise.reject(new Error('offline')); + }), + close: () => {}, + }; + + const event = { id: 'evt-health' }; + + for (let i = 0; i < 3; i += 1) { + await publishLock(['wss://bad', 'wss://healthy'], event, { + poolFactory: () => mockPool, + getPublishTimeoutMsFn: () => 200, + getMinSuccessfulRelayPublishesFn: () => 1, + getRelayFallbacksFn: () => [], + getMinActiveRelayPoolFn: () => 1, + retryAttempts: 1, + failureThreshold: 2, + quarantineCooldownMs: 60_000, + healthLogger: () => {}, + }); + } + + callOrder.length = 0; + await publishLock(['wss://bad', 'wss://healthy'], event, { + poolFactory: () => mockPool, + getPublishTimeoutMsFn: () => 200, + getMinSuccessfulRelayPublishesFn: () => 1, + getRelayFallbacksFn: () => [], + getMinActiveRelayPoolFn: () => 1, + retryAttempts: 1, + failureThreshold: 2, + quarantineCooldownMs: 60_000, + healthLogger: () => {}, + }); + + assert.deepStrictEqual(callOrder, ['wss://healthy']); + }); + + it('uses fallback and min active pool to reintroduce quarantined relay when needed', async () => { + const attempts = []; + const mockPool = { + publish: (relays) => relays.map((relay) => { + attempts.push(relay); + if (relay === 'wss://fallback-healthy') { + return Promise.resolve('ok'); + } + return Promise.reject(new Error('relay timeout')); + }), + close: () => {}, + }; + + const event = { id: 'evt-quorum' }; + await publishLock(['wss://primary-a', 'wss://primary-b'], event, { + poolFactory: () => mockPool, + getPublishTimeoutMsFn: () => 200, + getMinSuccessfulRelayPublishesFn: () => 1, + getRelayFallbacksFn: () => ['wss://fallback-healthy'], + getMinActiveRelayPoolFn: () => 2, + retryAttempts: 1, + failureThreshold: 1, + quarantineCooldownMs: 60_000, + healthLogger: () => {}, + }); + + assert.ok(attempts.includes('wss://fallback-healthy')); + }); + }); + + describe('scripted retry scenarios', () => { + it('succeeds immediately on first attempt without retries', async () => { + const telemetryLogs = []; + const relaysSeen = []; + const mockPool = createScriptedPool([ + { 'wss://relay-a': { ok: true } }, + ], relaysSeen); + + const event = { id: 'evt-immediate' }; + const result = await publishLock(['wss://relay-a'], event, { + poolFactory: () => mockPool, + getPublishTimeoutMsFn: () => 200, + getMinSuccessfulRelayPublishesFn: () => 1, + getRelayFallbacksFn: () => [], + getMinActiveRelayPoolFn: () => 1, + retryAttempts: 4, + telemetryLogger: (line) => telemetryLogs.push(line), + healthLogger: () => {}, + }); + + assert.strictEqual(result, event); + assert.strictEqual(mockPool.getPublishCallCount(), 1); + assert.deepStrictEqual(relaysSeen, [{ publishAttempt: 1, relays: ['wss://relay-a'] }]); + + const telemetry = parseTelemetry(telemetryLogs); + assert.strictEqual(telemetry.filter((entry) => entry.event === 'lock_publish_retry').length, 0); + const metEvent = telemetry.find((entry) => entry.event === 'lock_publish_quorum_met'); + assert.ok(metEvent); + assert.strictEqual(metEvent.publishAttempt, 1); + }); + + it('retries a transient failure then succeeds within retry budget with bounded jitter', async () => { + mock.timers.enable({ apis: ['Date', 'setTimeout'] }); + + const telemetryLogs = []; + const relaysSeen = []; + const sleepCalls = []; + const mockPool = createScriptedPool([ + { 'wss://relay-a': { ok: false, message: 'connection reset by peer' } }, + { 'wss://relay-a': { ok: true } }, + ], relaysSeen); + + const result = await publishLock(['wss://relay-a'], { id: 'evt-retry-success' }, { + poolFactory: () => mockPool, + getPublishTimeoutMsFn: () => 200, + getMinSuccessfulRelayPublishesFn: () => 1, + getRelayFallbacksFn: () => [], + getMinActiveRelayPoolFn: () => 1, + retryAttempts: 3, + retryBaseDelayMs: 100, + retryCapDelayMs: 500, + randomFn: () => 0.25, + sleepFn: async (delayMs) => { + sleepCalls.push(delayMs); + mock.timers.tick(delayMs); + }, + telemetryLogger: (line) => telemetryLogs.push(line), + healthLogger: () => {}, + }); + + assert.strictEqual(result.id, 'evt-retry-success'); + assert.deepStrictEqual(sleepCalls, [25]); + assert.strictEqual(mockPool.getPublishCallCount(), 2); + + const telemetry = parseTelemetry(telemetryLogs); + const retryEvents = telemetry.filter((entry) => entry.event === 'lock_publish_retry'); + assert.strictEqual(retryEvents.length, 1); + assert.strictEqual(retryEvents[0].publishAttempt, 1); + assert.strictEqual(retryEvents[0].relayUrl, 'wss://relay-a'); + assert.ok(retryEvents[0].nextDelayMs >= 0 && retryEvents[0].nextDelayMs <= 100); + }); + + it('returns terminal relay_publish_quorum_failure after exhausting retry budget', async () => { + mock.timers.enable({ apis: ['Date', 'setTimeout'] }); + + const telemetryLogs = []; + const relaysSeen = []; + const sleepCalls = []; + const randomValues = [0.1, 0.9]; + const mockPool = createScriptedPool([ + { 'wss://relay-a': { ok: false, message: 'relay unavailable' } }, + { 'wss://relay-a': { ok: false, message: 'relay unavailable' } }, + { 'wss://relay-a': { ok: false, message: 'relay unavailable' } }, + ], relaysSeen); + + await assert.rejects( + () => publishLock(['wss://relay-a'], { id: 'evt-exhausted' }, { + poolFactory: () => mockPool, + getPublishTimeoutMsFn: () => 200, + getMinSuccessfulRelayPublishesFn: () => 1, + getRelayFallbacksFn: () => [], + getMinActiveRelayPoolFn: () => 1, + retryAttempts: 3, + retryBaseDelayMs: 100, + retryCapDelayMs: 250, + randomFn: () => randomValues.shift() ?? 0, + sleepFn: async (delayMs) => { + sleepCalls.push(delayMs); + mock.timers.tick(delayMs); + }, + telemetryLogger: (line) => telemetryLogs.push(line), + healthLogger: () => {}, + }), + /error_category=relay_publish_quorum_failure/, + ); + + assert.strictEqual(mockPool.getPublishCallCount(), 3); + assert.deepStrictEqual(sleepCalls, [10, 180]); + assert.deepStrictEqual(relaysSeen.map((entry) => entry.publishAttempt), [1, 2, 3]); + + const telemetry = parseTelemetry(telemetryLogs); + const retryEvents = telemetry.filter((entry) => entry.event === 'lock_publish_retry'); + assert.strictEqual(retryEvents.length, 2); + assert.strictEqual(retryEvents[0].publishAttempt, 1); + assert.strictEqual(retryEvents[0].relayUrl, 'wss://relay-a'); + assert.ok(retryEvents[0].nextDelayMs >= 0 && retryEvents[0].nextDelayMs <= 100); + assert.strictEqual(retryEvents[1].publishAttempt, 2); + assert.strictEqual(retryEvents[1].relayUrl, 'wss://relay-a'); + assert.ok(retryEvents[1].nextDelayMs >= retryEvents[0].nextDelayMs); + assert.ok(retryEvents[1].nextDelayMs <= 200); + + const failedEvent = telemetry.find((entry) => entry.event === 'lock_publish_quorum_failed'); + assert.ok(failedEvent); + assert.strictEqual(failedEvent.errorCategory, 'relay_publish_quorum_failure'); + assert.strictEqual(failedEvent.attempts, 3); + }); + + it('short-circuits non-retryable failures without extra attempts', async () => { + const telemetryLogs = []; + const relaysSeen = []; + const mockPool = createScriptedPool([ + { 'wss://relay-a': { ok: false, message: 'validation failed: invalid signature' } }, + { 'wss://relay-a': { ok: true } }, + ], relaysSeen); + + await assert.rejects( + () => publishLock(['wss://relay-a'], { id: 'evt-permanent' }, { + poolFactory: () => mockPool, + getPublishTimeoutMsFn: () => 200, + getMinSuccessfulRelayPublishesFn: () => 1, + getRelayFallbacksFn: () => [], + getMinActiveRelayPoolFn: () => 1, + retryAttempts: 4, + telemetryLogger: (line) => telemetryLogs.push(line), + healthLogger: () => {}, + }), + /error_category=relay_publish_non_retryable/, + ); + + assert.strictEqual(mockPool.getPublishCallCount(), 1); + assert.deepStrictEqual(relaysSeen, [{ publishAttempt: 1, relays: ['wss://relay-a'] }]); + + const telemetry = parseTelemetry(telemetryLogs); + assert.strictEqual(telemetry.filter((entry) => entry.event === 'lock_publish_retry').length, 0); + + const failedEvent = telemetry.find((entry) => entry.event === 'lock_publish_quorum_failed'); + assert.ok(failedEvent); + assert.strictEqual(failedEvent.errorCategory, 'relay_publish_non_retryable'); + assert.notStrictEqual(failedEvent.errorCategory, 'relay_publish_quorum_failure'); + }); + }); +});