thePR0M3TH3AN/TORCH
1
0 Fork 0
mirror of https://github.com/PR0M3TH3AN/TORCH.git synced 2026-10-27 07:51:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor lock publisher to use shared error classifier and utils

Browse Source 
- 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 <[email protected]>
This commit is contained in:
google-labs-jules[bot]andPR0M3TH3AN committed 2026-02-21 18:34:18 +00:00
1 parent 79e261d7b2
commit 481d77f33c
5 files changed
+516 -610
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/lock-publisher.mjs
+7 -92
View File
@@ -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)));
src/lock-utils.mjs
+1 -7
View File
@@ -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() {
test/lock-ops-publish-retry.test.mjs
-192
View File
@@ -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');
});
});
test/lock-ops.test.mjs
-319
View File
@@ -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'));
});
});
});
test/lock-publisher.test.mjs
+508
View File
@@ -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');
});
});
});