diff --git a/src/constants.mjs b/src/constants.mjs index a9b2440..7605c8e 100644 --- a/src/constants.mjs +++ b/src/constants.mjs @@ -10,5 +10,6 @@ export const DEFAULT_NAMESPACE = 'torch'; export const DEFAULT_QUERY_TIMEOUT_MS = 15_000; export const DEFAULT_PUBLISH_TIMEOUT_MS = 10_000; export const DEFAULT_MIN_SUCCESSFUL_PUBLISHES = 1; +export const DEFAULT_MIN_ACTIVE_RELAY_POOL = 1; export const VALID_CADENCES = new Set(['daily', 'weekly']); export const KIND_APP_DATA = 30078; diff --git a/src/docs/TORCH.md b/src/docs/TORCH.md index dff6b90..b37d4ef 100644 --- a/src/docs/TORCH.md +++ b/src/docs/TORCH.md @@ -35,6 +35,7 @@ node bin/torch-lock.mjs list - `NOSTR_LOCK_PUBLISH_TIMEOUT_MS` - `NOSTR_LOCK_MIN_SUCCESSFUL_PUBLISHES` - `NOSTR_LOCK_RELAY_FALLBACKS` +- `NOSTR_LOCK_MIN_ACTIVE_RELAY_POOL` - `NOSTR_LOCK_DAILY_ROSTER` - `NOSTR_LOCK_WEEKLY_ROSTER` - `TORCH_CONFIG_PATH` @@ -57,6 +58,7 @@ Common settings: - `nostrLock.publishTimeoutMs` — per-relay publish timeout (ms, valid range: 100..120000). - `nostrLock.minSuccessfulRelayPublishes` — minimum successful publishes required before lock acquisition continues (default: `1`). - `nostrLock.relayFallbacks` — optional fallback relay URLs used when primary query/publish attempts fail quorum. +- `nostrLock.minActiveRelayPool` — minimum number of relays kept active even when lower-ranked relays are quarantined (default: `1`). - `nostrLock.dailyRoster` / `nostrLock.weeklyRoster` — optional per-project roster overrides. - `dashboard.defaultCadenceView` — default dashboard view (`daily`, `weekly`, `all`). - `dashboard.defaultStatusView` — default dashboard status filter (`active`, `all`). @@ -80,12 +82,14 @@ Recommended baseline for production scheduler runs: - `nostrLock.queryTimeoutMs`: `10000` - `nostrLock.publishTimeoutMs`: `8000` - `nostrLock.minSuccessfulRelayPublishes`: `2` +- `nostrLock.minActiveRelayPool`: `2` Validation behavior: - Relay URLs must be absolute `ws://` or `wss://` URLs. - Invalid relay URLs or invalid timeout/count ranges are fatal startup errors. - Lock backend errors include phase (`query:primary`, `query:fallback`, `publish:primary`, `publish:fallback`), relay endpoint, and timeout value used. +- Relay health snapshots are emitted periodically and whenever lock publish/query fails; snapshots include success rate, timeout rate, rolling latency, and quarantine state per relay. ## Scheduler lock reliability reporting diff --git a/src/lock-ops.mjs b/src/lock-ops.mjs index 8c853c8..112be88 100644 --- a/src/lock-ops.mjs +++ b/src/lock-ops.mjs @@ -4,9 +4,15 @@ import { getPublishTimeoutMs, getMinSuccessfulRelayPublishes, getRelayFallbacks, + getMinActiveRelayPool, } from './torch-config.mjs'; import { KIND_APP_DATA } from './constants.mjs'; +const relayHealthState = { + metricsByRelay: new Map(), + lastSnapshotAt: 0, +}; + function nowUnix() { return Math.floor(Date.now() / 1000); } @@ -70,12 +76,12 @@ function classifyPublishError(message) { return 'connection_reset'; } if ( - normalized.includes('unavailable') || - normalized.includes('offline') || - normalized.includes('econnrefused') || - normalized.includes('connection refused') || - normalized.includes('enotfound') || - normalized.includes('503') + normalized.includes('unavailable') + || normalized.includes('offline') + || normalized.includes('econnrefused') + || normalized.includes('connection refused') + || normalized.includes('enotfound') + || normalized.includes('503') ) { return 'relay_unavailable'; } @@ -91,44 +97,219 @@ function calculateBackoffDelayMs(attemptNumber, baseMs, capMs, randomFn = Math.r return Math.floor(randomFn() * maxDelay); } +function ensureRelayMetrics(relay, config) { + let metrics = relayHealthState.metricsByRelay.get(relay); + if (!metrics) { + metrics = { + relay, + recentOutcomes: [], + failureStreak: 0, + quarantineUntil: 0, + cooldownMs: config.quarantineCooldownMs, + lastLatencyMs: null, + lastResultAt: null, + }; + relayHealthState.metricsByRelay.set(relay, metrics); + } + return metrics; +} + +function summarizeRelayMetrics(metrics, nowMs) { + const total = metrics.recentOutcomes.length; + const successCount = metrics.recentOutcomes.filter((entry) => entry.success).length; + const timeoutCount = metrics.recentOutcomes.filter((entry) => entry.timedOut).length; + const latencyEntries = metrics.recentOutcomes.map((entry) => entry.latencyMs).filter((v) => Number.isFinite(v)); + const averageLatencyMs = latencyEntries.length + ? Math.round(latencyEntries.reduce((sum, value) => sum + value, 0) / latencyEntries.length) + : null; + const successRate = total > 0 ? successCount / total : 0.5; + const timeoutRate = total > 0 ? timeoutCount / total : 0; + const quarantineRemainingMs = metrics.quarantineUntil > nowMs ? metrics.quarantineUntil - nowMs : 0; + return { + relay: metrics.relay, + sampleSize: total, + successRate, + timeoutRate, + averageLatencyMs, + failureStreak: metrics.failureStreak, + quarantined: quarantineRemainingMs > 0, + quarantineRemainingMs, + cooldownMs: metrics.cooldownMs, + lastResultAt: metrics.lastResultAt, + }; +} + +function computeRelayScore(summary) { + const latencyPenalty = summary.averageLatencyMs === null + ? 0.1 + : Math.min(0.35, summary.averageLatencyMs / 10_000); + const quarantinePenalty = summary.quarantined ? 1 : 0; + return (summary.successRate * 1.4) - (summary.timeoutRate * 0.9) - latencyPenalty - quarantinePenalty; +} + +function rankRelaysByHealth(relays, config, nowMs = Date.now()) { + const summaries = relays.map((relay) => { + const metrics = ensureRelayMetrics(relay, config); + const summary = summarizeRelayMetrics(metrics, nowMs); + return { + relay, + summary, + score: computeRelayScore(summary), + }; + }); + + return summaries.sort((a, b) => { + if (b.score !== a.score) return b.score - a.score; + if (a.summary.quarantined !== b.summary.quarantined) return a.summary.quarantined ? 1 : -1; + if (a.summary.averageLatencyMs !== b.summary.averageLatencyMs) { + if (a.summary.averageLatencyMs === null) return 1; + if (b.summary.averageLatencyMs === null) return -1; + return a.summary.averageLatencyMs - b.summary.averageLatencyMs; + } + return a.relay.localeCompare(b.relay); + }); +} + +function prioritizeRelays(relays, config, nowMs = Date.now()) { + const minActive = Math.max(1, Math.min(config.minActiveRelayPool, relays.length || 1)); + const ranked = rankRelaysByHealth([...new Set(relays)], config, nowMs); + const active = ranked.filter((entry) => !entry.summary.quarantined); + const quarantined = ranked + .filter((entry) => entry.summary.quarantined) + .sort((a, b) => a.summary.quarantineRemainingMs - b.summary.quarantineRemainingMs); + + const selected = [...active]; + if (selected.length < minActive) { + const additionalNeeded = Math.min(minActive - selected.length, quarantined.length); + selected.push(...quarantined.slice(0, additionalNeeded)); + } + + return { + prioritized: selected.map((entry) => entry.relay), + ranked, + }; +} + +function recordRelayOutcome(relay, success, errorMessage, latencyMs, config, nowMs = Date.now()) { + const metrics = ensureRelayMetrics(relay, config); + const timedOut = String(errorMessage || '').toLowerCase().includes('timeout') + || String(errorMessage || '').toLowerCase().includes('timed out'); + metrics.recentOutcomes.push({ success, timedOut, latencyMs, atMs: nowMs }); + if (metrics.recentOutcomes.length > config.rollingWindowSize) { + metrics.recentOutcomes.splice(0, metrics.recentOutcomes.length - config.rollingWindowSize); + } + metrics.lastResultAt = nowMs; + metrics.lastLatencyMs = Number.isFinite(latencyMs) ? latencyMs : null; + + if (success) { + metrics.failureStreak = 0; + if (metrics.quarantineUntil > nowMs) { + metrics.quarantineUntil = 0; + metrics.cooldownMs = config.quarantineCooldownMs; + } + return; + } + + metrics.failureStreak += 1; + if (metrics.failureStreak >= config.failureThreshold) { + metrics.quarantineUntil = nowMs + metrics.cooldownMs; + metrics.cooldownMs = Math.min(config.maxQuarantineCooldownMs, Math.floor(metrics.cooldownMs * 1.5)); + } +} + +function collectHealthSnapshot(relays, config, nowMs = Date.now()) { + const uniqueRelays = [...new Set(relays)]; + return rankRelaysByHealth(uniqueRelays, config, nowMs).map((entry) => ({ + relay: entry.relay, + score: Number(entry.score.toFixed(4)), + ...entry.summary, + })); +} + +function maybeLogHealthSnapshot(relays, config, logger, reason, force = false, nowMs = Date.now()) { + const intervalReached = nowMs - relayHealthState.lastSnapshotAt >= config.snapshotIntervalMs; + if (!force && !intervalReached) return; + relayHealthState.lastSnapshotAt = nowMs; + logger(JSON.stringify({ + event: 'relay_health_snapshot', + reason, + relays: collectHealthSnapshot(relays, config, nowMs), + })); +} + +function buildRelayHealthConfig(deps) { + return { + rollingWindowSize: deps.rollingWindowSize ?? 25, + failureThreshold: deps.failureThreshold ?? 3, + quarantineCooldownMs: deps.quarantineCooldownMs ?? 30_000, + maxQuarantineCooldownMs: deps.maxQuarantineCooldownMs ?? 5 * 60_000, + snapshotIntervalMs: deps.snapshotIntervalMs ?? 60_000, + minActiveRelayPool: Math.max(1, deps.minActiveRelayPool), + }; +} + +function mergeRelayList(primaryRelays, fallbackRelays) { + return [...new Set([...primaryRelays, ...fallbackRelays])]; +} + export async function queryLocks(relays, cadence, dateStr, namespace, deps = {}) { const { poolFactory = () => new SimplePool(), getQueryTimeoutMsFn = getQueryTimeoutMs, getRelayFallbacksFn = getRelayFallbacks, + getMinActiveRelayPoolFn = getMinActiveRelayPool, errorLogger = console.error, + healthLogger = console.error, } = deps; const pool = poolFactory(); const tagFilter = `${namespace}-lock-${cadence}-${dateStr}`; const queryTimeoutMs = getQueryTimeoutMsFn(); - const fallbackRelays = getRelayFallbacksFn(); + const fallbackRelays = getRelayFallbacksFn().filter((relay) => !relays.includes(relay)); + const healthConfig = buildRelayHealthConfig({ + ...deps, + minActiveRelayPool: getMinActiveRelayPoolFn(), + }); const runQuery = async (relaySet, phase) => { + const { prioritized } = prioritizeRelays(relaySet, healthConfig); + const startedAtMs = Date.now(); try { const events = await withTimeout( - pool.querySync(relaySet, { + pool.querySync(prioritized, { kinds: [KIND_APP_DATA], '#t': [tagFilter], }), queryTimeoutMs, - `[${phase}] Relay query timed out after ${queryTimeoutMs}ms (relays: ${relayListLabel(relaySet)})`, + `[${phase}] Relay query timed out after ${queryTimeoutMs}ms (relays: ${relayListLabel(prioritized)})`, ); + const elapsedMs = Date.now() - startedAtMs; + for (const relay of prioritized) { + recordRelayOutcome(relay, true, null, elapsedMs, healthConfig); + } return filterActiveLocks(events.map(parseLockEvent)); } catch (err) { + const elapsedMs = Date.now() - startedAtMs; const message = err instanceof Error ? err.message : String(err); + for (const relay of prioritized) { + recordRelayOutcome(relay, false, message, elapsedMs, healthConfig); + } throw new Error( - `[${phase}] Relay query failed (timeout=${queryTimeoutMs}ms, relays=${relayListLabel(relaySet)}): ${message}`, + `[${phase}] Relay query failed (timeout=${queryTimeoutMs}ms, relays=${relayListLabel(prioritized)}): ${message}`, { cause: err }, ); } }; + const allRelays = mergeRelayList(relays, fallbackRelays); + try { + maybeLogHealthSnapshot(allRelays, healthConfig, healthLogger, 'query:periodic'); try { return await runQuery(relays, 'query:primary'); } catch (primaryErr) { if (!fallbackRelays.length) { + maybeLogHealthSnapshot(allRelays, healthConfig, healthLogger, 'query:failure', true); throw primaryErr; } errorLogger(`WARN: ${primaryErr.message}`); @@ -136,7 +317,6 @@ export async function queryLocks(relays, cadence, dateStr, namespace, deps = {}) return await runQuery(fallbackRelays, 'query:fallback'); } } finally { - const allRelays = [...new Set([...relays, ...fallbackRelays])]; pool.close(allRelays); } } @@ -153,11 +333,22 @@ async function publishToRelays(pool, relays, event, publishTimeoutMs, phase) { return settled.map((result, index) => { if (result.status === 'fulfilled') { - return { relay: relays[index], success: true, phase }; + return { + relay: relays[index], + success: true, + phase, + latencyMs: null, + }; } const reason = result.reason; const message = reason instanceof Error ? reason.message : String(reason ?? 'unknown'); - return { relay: relays[index], success: false, phase, message }; + return { + relay: relays[index], + success: false, + phase, + message, + latencyMs: null, + }; }); } @@ -167,12 +358,14 @@ export async function publishLock(relays, event, deps = {}) { getPublishTimeoutMsFn = getPublishTimeoutMs, getMinSuccessfulRelayPublishesFn = getMinSuccessfulRelayPublishes, getRelayFallbacksFn = getRelayFallbacks, + getMinActiveRelayPoolFn = getMinActiveRelayPool, retryAttempts = 4, retryBaseDelayMs = 500, retryCapDelayMs = 8_000, sleepFn = (ms) => new Promise((resolve) => setTimeout(resolve, ms)), randomFn = Math.random, telemetryLogger = console.error, + healthLogger = console.error, } = deps; const pool = poolFactory(); @@ -180,6 +373,10 @@ export async function publishLock(relays, event, deps = {}) { const minSuccesses = getMinSuccessfulRelayPublishesFn(); const fallbackRelays = getRelayFallbacksFn().filter((relay) => !relays.includes(relay)); const maxAttempts = Math.max(1, Math.floor(retryAttempts)); + const healthConfig = buildRelayHealthConfig({ + ...deps, + minActiveRelayPool: getMinActiveRelayPoolFn(), + }); const publishOnce = async () => { const attempted = new Set(); @@ -187,10 +384,19 @@ export async function publishLock(relays, event, deps = {}) { const attemptPhase = async (phaseRelays, phaseName) => { if (!phaseRelays.length) return; - for (const relay of phaseRelays) { + const { prioritized } = prioritizeRelays(phaseRelays, healthConfig); + if (!prioritized.length) return; + + const startedAtMs = Date.now(); + for (const relay of prioritized) { attempted.add(relay); } - const phaseResults = await publishToRelays(pool, phaseRelays, event, publishTimeoutMs, phaseName); + const phaseResults = await publishToRelays(pool, prioritized, event, publishTimeoutMs, phaseName); + const elapsedMs = Date.now() - startedAtMs; + for (const result of phaseResults) { + result.latencyMs = elapsedMs; + recordRelayOutcome(result.relay, result.success, result.message, elapsedMs, healthConfig); + } publishResults.push(...phaseResults); }; @@ -212,7 +418,10 @@ export async function publishLock(relays, event, deps = {}) { return { attempted, publishResults, successCount, failures }; }; + const allRelays = mergeRelayList(relays, fallbackRelays); + try { + maybeLogHealthSnapshot(allRelays, healthConfig, healthLogger, 'publish:periodic'); let lastAttemptState = null; for (let attemptNumber = 1; attemptNumber <= maxAttempts; attemptNumber += 1) { const startedAtMs = Date.now(); @@ -220,8 +429,8 @@ export async function publishLock(relays, event, deps = {}) { if (lastAttemptState.successCount >= minSuccesses) { console.error( - ` Published to ${lastAttemptState.successCount}/${lastAttemptState.attempted.size} relays ` + - `(required=${minSuccesses}, timeout=${publishTimeoutMs}ms)`, + ` Published to ${lastAttemptState.successCount}/${lastAttemptState.attempted.size} relays ` + + `(required=${minSuccesses}, timeout=${publishTimeoutMs}ms)`, ); return event; } @@ -254,11 +463,17 @@ export async function publishLock(relays, event, deps = {}) { const failureLines = lastAttemptState.failures .map((result) => `${result.relay} (${result.phase}, category=${result.category}): ${result.message}`); + maybeLogHealthSnapshot(allRelays, healthConfig, healthLogger, 'publish:failure', true); 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 ')}`, + `Failed relay publish quorum in publish phase: ${lastAttemptState.successCount}/${lastAttemptState.attempted.size} successful ` + + `(required=${minSuccesses}, timeout=${publishTimeoutMs}ms, attempts=${maxAttempts})\n ${failureLines.join('\n ')}`, ); } finally { - pool.close([...new Set([...relays, ...fallbackRelays])]); + pool.close(allRelays); } } + +export function _resetRelayHealthState() { + relayHealthState.metricsByRelay.clear(); + relayHealthState.lastSnapshotAt = 0; +} diff --git a/src/torch-config.mjs b/src/torch-config.mjs index d7de51f..cd8204e 100644 --- a/src/torch-config.mjs +++ b/src/torch-config.mjs @@ -7,6 +7,7 @@ import { DEFAULT_QUERY_TIMEOUT_MS, DEFAULT_PUBLISH_TIMEOUT_MS, DEFAULT_MIN_SUCCESSFUL_PUBLISHES, + DEFAULT_MIN_ACTIVE_RELAY_POOL, } from './constants.mjs'; const DEFAULT_CONFIG_PATH = 'torch-config.json'; @@ -96,6 +97,7 @@ export function parseTorchConfig(raw, configPath = null) { queryTimeoutMs: parsePositiveInteger(nostrLock.queryTimeoutMs), publishTimeoutMs: parsePositiveInteger(nostrLock.publishTimeoutMs), minSuccessfulRelayPublishes: parsePositiveInteger(nostrLock.minSuccessfulRelayPublishes), + minActiveRelayPool: parsePositiveInteger(nostrLock.minActiveRelayPool), dailyRoster: parseRoster(nostrLock.dailyRoster), weeklyRoster: parseRoster(nostrLock.weeklyRoster), }, @@ -165,6 +167,9 @@ function validateLockBackendConfig(config) { if (config.nostrLock.minSuccessfulRelayPublishes !== null) { assertPositiveCount(config.nostrLock.minSuccessfulRelayPublishes, 'nostrLock.minSuccessfulRelayPublishes'); } + if (config.nostrLock.minActiveRelayPool !== null) { + assertPositiveCount(config.nostrLock.minActiveRelayPool, 'nostrLock.minActiveRelayPool'); + } } function parseEnvRelayList(envValue, envName) { @@ -264,3 +269,16 @@ export function getMinSuccessfulRelayPublishes() { assertPositiveCount(value, 'effective min successful relay publishes'); return value; } + +export function getMinActiveRelayPool() { + const config = loadTorchConfig(); + const envValue = process.env.NOSTR_LOCK_MIN_ACTIVE_RELAY_POOL; + if (envValue) { + const parsed = parseEnvInteger(envValue, 'NOSTR_LOCK_MIN_ACTIVE_RELAY_POOL'); + assertPositiveCount(parsed, 'NOSTR_LOCK_MIN_ACTIVE_RELAY_POOL'); + return parsed; + } + const value = config.nostrLock.minActiveRelayPool || DEFAULT_MIN_ACTIVE_RELAY_POOL; + assertPositiveCount(value, 'effective min active relay pool'); + return value; +} diff --git a/test/lock-ops.test.mjs b/test/lock-ops.test.mjs index 70bd577..50d3d73 100644 --- a/test/lock-ops.test.mjs +++ b/test/lock-ops.test.mjs @@ -1,8 +1,17 @@ -import { describe, it } from 'node:test'; +import { describe, it, beforeEach } from 'node:test'; import assert from 'node:assert'; -import { parseLockEvent, publishLock, queryLocks } from '../src/lock-ops.mjs'; +import { + parseLockEvent, + publishLock, + queryLocks, + _resetRelayHealthState, +} from '../src/lock-ops.mjs'; describe('lock-ops', () => { + beforeEach(() => { + _resetRelayHealthState(); + }); + describe('parseLockEvent', () => { it('parses a valid lock event correctly', () => { const event = { @@ -32,7 +41,7 @@ describe('lock-ops', () => { it('falls back to fallback relays when primary query fails', async () => { const mockPool = { querySync: async (relays) => { - if (relays[0] === 'wss://primary') { + if (relays.includes('wss://primary')) { throw new Error('primary unavailable'); } return [{ @@ -56,6 +65,7 @@ describe('lock-ops', () => { getQueryTimeoutMsFn: () => 200, getRelayFallbacksFn: () => ['wss://fallback'], errorLogger: () => {}, + healthLogger: () => {}, }, ); @@ -87,6 +97,7 @@ describe('lock-ops', () => { getPublishTimeoutMsFn: () => 200, getMinSuccessfulRelayPublishesFn: () => 1, getRelayFallbacksFn: () => [], + getMinActiveRelayPoolFn: () => 1, retryAttempts: 3, retryBaseDelayMs: 500, retryCapDelayMs: 8_000, @@ -97,6 +108,7 @@ describe('lock-ops', () => { telemetryLogger: (line) => { telemetry.push(JSON.parse(line)); }, + healthLogger: () => {}, }); assert.strictEqual(result, event); @@ -123,7 +135,6 @@ describe('lock-ops', () => { assert.ok(Number.isInteger(telemetry[1].elapsedMs)); }); - it('fails after retry budget is exhausted for persistent transient failures', async () => { let publishCalls = 0; const sleepCalls = []; @@ -142,12 +153,14 @@ describe('lock-ops', () => { 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\)/, ); @@ -172,6 +185,7 @@ describe('lock-ops', () => { getPublishTimeoutMsFn: () => 200, getMinSuccessfulRelayPublishesFn: () => 1, getRelayFallbacksFn: () => [], + getMinActiveRelayPoolFn: () => 1, retryAttempts: 4, sleepFn: async () => { throw new Error('should not sleep for permanent failures'); @@ -179,6 +193,7 @@ describe('lock-ops', () => { telemetryLogger: () => { throw new Error('should not emit retry telemetry for permanent failures'); }, + healthLogger: () => {}, }), /Failed relay publish quorum in publish phase: 0\/1 successful \(required=1, timeout=200ms, attempts=4\)/, ); @@ -206,10 +221,12 @@ describe('lock-ops', () => { 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); @@ -230,9 +247,86 @@ describe('lock-ops', () => { 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' }; + + // Warm up failure history so bad relay enters quarantine. + 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/torch-config.test.mjs b/test/torch-config.test.mjs index 00b892d..190cf84 100644 --- a/test/torch-config.test.mjs +++ b/test/torch-config.test.mjs @@ -12,6 +12,7 @@ import { getPublishTimeoutMs, getMinSuccessfulRelayPublishes, getRelayFallbacks, + getMinActiveRelayPool, } from '../src/torch-config.mjs'; import { DEFAULT_RELAYS } from '../src/constants.mjs'; @@ -45,6 +46,7 @@ describe('torch-config', () => { queryTimeoutMs: 5000, publishTimeoutMs: 8000, minSuccessfulRelayPublishes: 2, + minActiveRelayPool: 2, }, }); @@ -54,6 +56,7 @@ describe('torch-config', () => { assert.strictEqual(config.nostrLock.queryTimeoutMs, 5000); assert.strictEqual(config.nostrLock.publishTimeoutMs, 8000); assert.strictEqual(config.nostrLock.minSuccessfulRelayPublishes, 2); + assert.strictEqual(config.nostrLock.minActiveRelayPool, 2); }); }); @@ -101,6 +104,7 @@ describe('torch-config', () => { NOSTR_LOCK_PUBLISH_TIMEOUT_MS: process.env.NOSTR_LOCK_PUBLISH_TIMEOUT_MS, NOSTR_LOCK_MIN_SUCCESSFUL_PUBLISHES: process.env.NOSTR_LOCK_MIN_SUCCESSFUL_PUBLISHES, NOSTR_LOCK_RELAY_FALLBACKS: process.env.NOSTR_LOCK_RELAY_FALLBACKS, + NOSTR_LOCK_MIN_ACTIVE_RELAY_POOL: process.env.NOSTR_LOCK_MIN_ACTIVE_RELAY_POOL, }; beforeEach(() => { @@ -110,6 +114,7 @@ describe('torch-config', () => { delete process.env.NOSTR_LOCK_PUBLISH_TIMEOUT_MS; delete process.env.NOSTR_LOCK_MIN_SUCCESSFUL_PUBLISHES; delete process.env.NOSTR_LOCK_RELAY_FALLBACKS; + delete process.env.NOSTR_LOCK_MIN_ACTIVE_RELAY_POOL; }); after(() => { @@ -138,6 +143,10 @@ describe('torch-config', () => { process.env.NOSTR_LOCK_RELAY_FALLBACKS = 'https://bad'; assert.throws(() => getRelayFallbacks(), /NOSTR_LOCK_RELAY_FALLBACKS/); + + process.env.NOSTR_LOCK_RELAY_FALLBACKS = ''; + process.env.NOSTR_LOCK_MIN_ACTIVE_RELAY_POOL = '0'; + assert.throws(() => getMinActiveRelayPool(), /NOSTR_LOCK_MIN_ACTIVE_RELAY_POOL/); }); }); }); diff --git a/torch-config.example.json b/torch-config.example.json index 701e9b3..2794feb 100644 --- a/torch-config.example.json +++ b/torch-config.example.json @@ -10,6 +10,7 @@ "queryTimeoutMs": 15000, "publishTimeoutMs": 8000, "minSuccessfulRelayPublishes": 2, + "minActiveRelayPool": 2, "relayFallbacks": [ "wss://relay.snort.social", "wss://relay.nostr.band"