diff --git a/test/relay-fanout-quorum.integration.test.mjs b/test/relay-fanout-quorum.integration.test.mjs new file mode 100644 index 0000000..ca2769e --- /dev/null +++ b/test/relay-fanout-quorum.integration.test.mjs @@ -0,0 +1,239 @@ +import { afterEach, beforeEach, describe, it } from 'node:test'; +import assert from 'node:assert'; +import net from 'node:net'; +import { WebSocketServer } from 'ws'; + +import { publishLock, _resetRelayHealthState } from '../src/lock-ops.mjs'; + +async function getFreePort() { + return new Promise((resolve, reject) => { + const server = net.createServer(); + server.listen(0, '127.0.0.1', () => { + const address = server.address(); + const port = typeof address === 'object' && address ? address.port : null; + server.close((closeErr) => { + if (closeErr) { + reject(closeErr); + return; + } + resolve(port); + }); + }); + server.on('error', reject); + }); +} + +async function createRelaySimulatorFixture(modes) { + if (modes.length !== 3) { + throw new Error(`Expected exactly three relays, got ${modes.length}`); + } + + const servers = []; + const relays = []; + + for (const mode of modes) { + const port = await getFreePort(); + const relayUrl = `ws://127.0.0.1:${port}`; + + if (mode.type === 'connection-fail') { + relays.push(relayUrl); + continue; + } + + const wss = new WebSocketServer({ host: '127.0.0.1', port }); + wss.on('connection', (socket) => { + socket.on('message', (rawData) => { + let payload; + try { + payload = JSON.parse(rawData.toString()); + } catch { + return; + } + + if (!Array.isArray(payload) || payload[0] !== 'EVENT') { + return; + } + + const event = payload[1] || {}; + if (mode.type === 'success') { + const delayMs = mode.delayMs ?? 0; + setTimeout(() => { + if (socket.readyState === socket.OPEN) { + socket.send(JSON.stringify(['OK', event.id, true, 'accepted'])); + } + }, delayMs); + } + }); + }); + + servers.push(wss); + relays.push(relayUrl); + } + + return { + relays, + async stop() { + await Promise.all(servers.map((server) => new Promise((resolve, reject) => { + server.close((err) => { + if (err) reject(err); + else resolve(); + }); + }))); + }, + }; +} + +function parseTelemetry(lines) { + return lines.map((line) => JSON.parse(line)); +} + +function makeEvent(id) { + return { + id, + kind: 1, + pubkey: 'f'.repeat(64), + created_at: Math.floor(Date.now() / 1000), + tags: [], + content: 'relay-test', + sig: 'e'.repeat(128), + }; +} + +describe('relay fanout quorum integration', { concurrency: false }, () => { + let fixture = null; + + beforeEach(() => { + _resetRelayHealthState(); + }); + + afterEach(async () => { + if (fixture) { + await fixture.stop(); + fixture = null; + } + }); + + it('passes with required=1 when one relay succeeds and two fail', async () => { + fixture = await createRelaySimulatorFixture([ + { type: 'success' }, + { type: 'timeout' }, + { type: 'timeout' }, + ]); + + const telemetryLogs = []; + const result = await publishLock(fixture.relays, makeEvent('evt-required1-pass'), { + getPublishTimeoutMsFn: () => 80, + getMinSuccessfulRelayPublishesFn: () => 1, + getRelayFallbacksFn: () => [], + getMinActiveRelayPoolFn: () => 1, + retryAttempts: 1, + telemetryLogger: (line) => telemetryLogs.push(line), + healthLogger: () => {}, + }); + + assert.strictEqual(result.id, 'evt-required1-pass'); + + const telemetry = parseTelemetry(telemetryLogs); + const quorumSummary = telemetry.find((entry) => entry.event === 'lock_publish_quorum_met'); + assert.ok(quorumSummary); + assert.strictEqual(quorumSummary.successCount, 1); + assert.strictEqual(quorumSummary.requiredSuccesses, 1); + assert.strictEqual(quorumSummary.relayAttemptedCount, 3); + }); + + it('fails with quorum error for required=1 when all relays fail', async () => { + fixture = await createRelaySimulatorFixture([ + { type: 'timeout' }, + { type: 'timeout' }, + { type: 'timeout' }, + ]); + + const telemetryLogs = []; + await assert.rejects( + () => publishLock(fixture.relays, makeEvent('evt-required1-all-fail'), { + getPublishTimeoutMsFn: () => 80, + getMinSuccessfulRelayPublishesFn: () => 1, + getRelayFallbacksFn: () => [], + getMinActiveRelayPoolFn: () => 1, + retryAttempts: 1, + telemetryLogger: (line) => telemetryLogs.push(line), + healthLogger: () => {}, + }), + /error_category=relay_publish_quorum_failure/, + ); + + const telemetry = parseTelemetry(telemetryLogs); + const relayOutcomes = telemetry.filter((entry) => entry.event === 'lock_publish_failure'); + assert.strictEqual(relayOutcomes.length, 3); + assert.ok(relayOutcomes.every((entry) => fixture.relays.includes(entry.relayUrl))); + + const quorumSummary = telemetry.find((entry) => entry.event === 'lock_publish_quorum_failed'); + assert.ok(quorumSummary); + assert.strictEqual(quorumSummary.successCount, 0); + assert.strictEqual(quorumSummary.requiredSuccesses, 1); + assert.strictEqual(quorumSummary.relayAttemptedCount, 3); + }); + + it('fails with required=2 when only one relay succeeds', async () => { + fixture = await createRelaySimulatorFixture([ + { type: 'success' }, + { type: 'timeout' }, + { type: 'timeout' }, + ]); + + const telemetryLogs = []; + await assert.rejects( + () => publishLock(fixture.relays, makeEvent('evt-required2-fail'), { + getPublishTimeoutMsFn: () => 80, + getMinSuccessfulRelayPublishesFn: () => 2, + getRelayFallbacksFn: () => [], + getMinActiveRelayPoolFn: () => 1, + retryAttempts: 1, + telemetryLogger: (line) => telemetryLogs.push(line), + healthLogger: () => {}, + }), + /error_category=relay_publish_quorum_failure/, + ); + + const telemetry = parseTelemetry(telemetryLogs); + const relayOutcomes = telemetry.filter((entry) => entry.event === 'lock_publish_failure'); + assert.strictEqual(relayOutcomes.length, 2); + + const quorumSummary = telemetry.find((entry) => entry.event === 'lock_publish_quorum_failed'); + assert.ok(quorumSummary); + assert.strictEqual(quorumSummary.successCount, 1); + assert.strictEqual(quorumSummary.requiredSuccesses, 2); + assert.strictEqual(quorumSummary.relayAttemptedCount, 3); + }); + + it('passes when delayed successes reach quorum before timeout', async () => { + fixture = await createRelaySimulatorFixture([ + { type: 'success', delayMs: 5 }, + { type: 'success', delayMs: 25 }, + { type: 'timeout' }, + ]); + + const telemetryLogs = []; + const startedAt = Date.now(); + const result = await publishLock(fixture.relays, makeEvent('evt-delayed-quorum'), { + getPublishTimeoutMsFn: () => 80, + getMinSuccessfulRelayPublishesFn: () => 2, + getRelayFallbacksFn: () => [], + getMinActiveRelayPoolFn: () => 1, + retryAttempts: 1, + telemetryLogger: (line) => telemetryLogs.push(line), + healthLogger: () => {}, + }); + const elapsedMs = Date.now() - startedAt; + + assert.strictEqual(result.id, 'evt-delayed-quorum'); + assert.ok(elapsedMs < 160); + + const telemetry = parseTelemetry(telemetryLogs); + const quorumSummary = telemetry.find((entry) => entry.event === 'lock_publish_quorum_met'); + assert.ok(quorumSummary); + assert.strictEqual(quorumSummary.successCount, 2); + assert.strictEqual(quorumSummary.requiredSuccesses, 2); + assert.strictEqual(quorumSummary.relayAttemptedCount, 3); + }); +});