mirror of
https://github.com/PR0M3TH3AN/TORCH.git
synced 2026-10-27 16:01:54 +00:00
Add bounded retry logic for lock publish quorum
This commit is contained in:
1 parent
3c5bf0ba9e
commit
f22bcfb713
2 files changed
+239
-47
No files matched your search
+138
-23
@@ -65,23 +65,155 @@ describe('lock-ops', () => {
|
||||
});
|
||||
|
||||
describe('publishLock', () => {
|
||||
it('succeeds with partial primary success when minimum successes is 1', async () => {
|
||||
it('retries transient failures and succeeds on a later attempt', async () => {
|
||||
const telemetry = [];
|
||||
const sleepCalls = [];
|
||||
let publishCalls = 0;
|
||||
|
||||
const mockPool = {
|
||||
publish: (relays) => relays.map((relay) => (
|
||||
relay.includes('ok') ? Promise.resolve('ok') : Promise.reject(new Error('offline'))
|
||||
)),
|
||||
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-1' };
|
||||
const result = await publishLock(['wss://ok-1', 'wss://bad-1'], event, {
|
||||
const event = { id: 'evt-retry' };
|
||||
const result = await publishLock(['wss://primary-a', 'wss://primary-b'], event, {
|
||||
poolFactory: () => mockPool,
|
||||
getPublishTimeoutMsFn: () => 200,
|
||||
getMinSuccessfulRelayPublishesFn: () => 1,
|
||||
getRelayFallbacksFn: () => [],
|
||||
retryAttempts: 3,
|
||||
retryBaseDelayMs: 500,
|
||||
retryCapDelayMs: 8_000,
|
||||
randomFn: () => 0.5,
|
||||
sleepFn: async (ms) => {
|
||||
sleepCalls.push(ms);
|
||||
},
|
||||
telemetryLogger: (line) => {
|
||||
telemetry.push(JSON.parse(line));
|
||||
},
|
||||
});
|
||||
|
||||
assert.strictEqual(result, event);
|
||||
assert.strictEqual(publishCalls, 2);
|
||||
assert.deepStrictEqual(sleepCalls, [250]);
|
||||
assert.strictEqual(telemetry.length, 2);
|
||||
assert.deepStrictEqual(telemetry[0], {
|
||||
event: 'lock_publish_retry',
|
||||
attempt: 1,
|
||||
relayUrl: 'wss://primary-a',
|
||||
errorCategory: 'connection_reset',
|
||||
elapsedMs: telemetry[0].elapsedMs,
|
||||
nextDelayMs: 250,
|
||||
});
|
||||
assert.deepStrictEqual(telemetry[1], {
|
||||
event: 'lock_publish_retry',
|
||||
attempt: 1,
|
||||
relayUrl: 'wss://primary-b',
|
||||
errorCategory: 'connection_reset',
|
||||
elapsedMs: telemetry[1].elapsedMs,
|
||||
nextDelayMs: 250,
|
||||
});
|
||||
assert.ok(Number.isInteger(telemetry[0].elapsedMs));
|
||||
assert.ok(Number.isInteger(telemetry[1].elapsedMs));
|
||||
});
|
||||
|
||||
|
||||
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: () => [],
|
||||
retryAttempts: 3,
|
||||
randomFn: () => 1,
|
||||
sleepFn: async (ms) => {
|
||||
sleepCalls.push(ms);
|
||||
},
|
||||
telemetryLogger: () => {},
|
||||
}),
|
||||
/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: () => [],
|
||||
retryAttempts: 4,
|
||||
sleepFn: async () => {
|
||||
throw new Error('should not sleep for permanent failures');
|
||||
},
|
||||
telemetryLogger: () => {
|
||||
throw new Error('should not emit retry telemetry for permanent failures');
|
||||
},
|
||||
}),
|
||||
/Failed relay publish quorum in publish phase: 0\/1 successful \(required=1, timeout=200ms, attempts=4\)/,
|
||||
);
|
||||
|
||||
assert.strictEqual(publishCalls, 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: () => [],
|
||||
retryAttempts: 3,
|
||||
sleepFn: async () => {
|
||||
throw new Error('should not sleep when quorum is satisfied');
|
||||
},
|
||||
});
|
||||
|
||||
assert.strictEqual(result, event);
|
||||
assert.strictEqual(publishCalls, 1);
|
||||
});
|
||||
|
||||
it('uses fallback relays to satisfy min success quorum', async () => {
|
||||
@@ -102,22 +234,5 @@ describe('lock-ops', () => {
|
||||
|
||||
assert.strictEqual(result, event);
|
||||
});
|
||||
|
||||
it('fails when successful publishes stay below configured minimum', async () => {
|
||||
const mockPool = {
|
||||
publish: (relays) => relays.map(() => Promise.reject(new Error('offline'))),
|
||||
close: () => {},
|
||||
};
|
||||
|
||||
await assert.rejects(
|
||||
() => publishLock(['wss://primary-bad'], { id: 'evt-3' }, {
|
||||
poolFactory: () => mockPool,
|
||||
getPublishTimeoutMsFn: () => 200,
|
||||
getMinSuccessfulRelayPublishesFn: () => 2,
|
||||
getRelayFallbacksFn: () => ['wss://fallback-bad'],
|
||||
}),
|
||||
/Failed relay publish quorum in publish phase: 0\/2 successful \(required=2, timeout=200ms\)/,
|
||||
);
|
||||
});
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user