diff --git a/src/lock-ops.mjs b/src/lock-ops.mjs index 9611779..e3fbcad 100644 --- a/src/lock-ops.mjs +++ b/src/lock-ops.mjs @@ -8,7 +8,7 @@ import { KIND_APP_DATA, MS_PER_SECOND, } from './constants.mjs'; -import { nowUnix, mergeRelayList } from './utils.mjs'; +import { nowUnix, relayListLabel } from './utils.mjs'; import { defaultHealthManager, buildRelayHealthConfig, RelayHealthManager } from './relay-health-manager.mjs'; import { publishLock, LockPublisher } from './lock-publisher.mjs'; import { @@ -61,6 +61,20 @@ function filterActiveLocks(locks) { return locks.filter((lock) => !lock.expiresAt || lock.expiresAt > now); } +function withTimeout(promise, timeoutMs, timeoutMessage) { + let timeoutHandle; + const timeoutPromise = new Promise((_, reject) => { + timeoutHandle = setTimeout(() => reject(new Error(timeoutMessage)), timeoutMs); + }); + return Promise.race([promise, timeoutPromise]).finally(() => { + if (timeoutHandle) clearTimeout(timeoutHandle); + }); +} + +function mergeRelayList(primaryRelays, fallbackRelays) { + return [...new Set([...primaryRelays, ...fallbackRelays])]; +} + /** * Queries relays for active lock events matching the criteria. * Uses a tiered approach: tries primary relays first, then falls back to others if needed. diff --git a/src/lock-publisher.mjs b/src/lock-publisher.mjs index b7fa817..61fba13 100644 --- a/src/lock-publisher.mjs +++ b/src/lock-publisher.mjs @@ -8,17 +8,118 @@ import { } from './torch-config.mjs'; import { mergeRelayList } from './utils.mjs'; import { defaultHealthManager, buildRelayHealthConfig } from './relay-health-manager.mjs'; -import { - withTimeout, - relayListLabel, - mergeRelayList, - secureRandom, -} from './lock-utils.mjs'; -import { - PUBLISH_FAILURE_CATEGORIES, - classifyPublishError, - isTransientPublishCategory, -} from './lock-error-classifier.mjs'; +import { relayListLabel } from './utils.mjs'; + +function withTimeout(promise, timeoutMs, timeoutMessage) { + let timeoutHandle; + const timeoutPromise = new Promise((_, reject) => { + timeoutHandle = setTimeout(() => reject(new Error(timeoutMessage)), timeoutMs); + }); + return Promise.race([promise, timeoutPromise]).finally(() => { + if (timeoutHandle) clearTimeout(timeoutHandle); + }); +} + +function mergeRelayList(primaryRelays, fallbackRelays) { + return [...new Set([...primaryRelays, ...fallbackRelays])]; +} + +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); +} + +const MAX_RANDOM = 281474976710655; // 2**48 - 1 + +export function secureRandom() { + return randomInt(0, MAX_RANDOM) / MAX_RANDOM; +} 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 a064014..dbe1820 100644 --- a/src/lock-utils.mjs +++ b/src/lock-utils.mjs @@ -3,6 +3,8 @@ import path from 'node:path'; import { randomInt } from 'node:crypto'; import { todayDateStr, getIsoWeekStr } from './utils.mjs'; +export { relayListLabel } from './utils.mjs'; + /** * Scans the local log directory to identify agents that have already completed their task * for the current period (daily or weekly). @@ -51,10 +53,6 @@ export async function getCompletedAgents(cadence, logDir, deps) { export { withTimeout }; -export function relayListLabel(relays) { - return relays.join(', '); -} - export function mergeRelayList(primaryRelays, fallbackRelays) { return [...new Set([...primaryRelays, ...fallbackRelays])]; } diff --git a/src/utils.mjs b/src/utils.mjs index abd9f9d..4397000 100644 --- a/src/utils.mjs +++ b/src/utils.mjs @@ -52,12 +52,6 @@ export function ensureDir(dir) { } } -export function withTimeout(promise, timeoutMs, timeoutMessage) { - let timeoutHandle; - const timeoutPromise = new Promise((_, reject) => { - timeoutHandle = setTimeout(() => reject(new Error(timeoutMessage)), timeoutMs); - }); - return Promise.race([promise, timeoutPromise]).finally(() => { - if (timeoutHandle) clearTimeout(timeoutHandle); - }); +export function relayListLabel(relays) { + return relays.join(', '); } diff --git a/test/utils.test.mjs b/test/utils.test.mjs index 0c59f6a..7051687 100644 --- a/test/utils.test.mjs +++ b/test/utils.test.mjs @@ -2,7 +2,7 @@ import { describe, it, after } from 'node:test'; import assert from 'node:assert'; import fs from 'node:fs'; import path from 'node:path'; -import { getIsoWeekStr, todayDateStr, nowUnix, ensureDir, detectPlatform, mergeRelayList } from '../src/utils.mjs'; +import { getIsoWeekStr, todayDateStr, nowUnix, ensureDir, detectPlatform, relayListLabel } from '../src/utils.mjs'; describe('Date Utilities', () => { describe('todayDateStr', () => { @@ -201,3 +201,22 @@ describe('File Utilities', () => { }); }); }); + +describe('String Utilities', () => { + describe('relayListLabel', () => { + it('joins relay URLs with comma and space', () => { + const relays = ['ws://relay1.com', 'ws://relay2.com']; + assert.strictEqual(relayListLabel(relays), 'ws://relay1.com, ws://relay2.com'); + }); + + it('handles single relay', () => { + const relays = ['ws://relay1.com']; + assert.strictEqual(relayListLabel(relays), 'ws://relay1.com'); + }); + + it('handles empty array', () => { + const relays = []; + assert.strictEqual(relayListLabel(relays), ''); + }); + }); +});