Something went wrong. Try again.
Monorepo for wisp.place. A static site hosting service built on top of the AT Protocol.
Something went wrong. Try again.
8.4 kB · 332 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333import { describe, expect, test } from 'bun:test'import { createDatabaseReadCircuit } from './database-read-circuit'
describe('database read circuit', () => { test('uses the primary directly when no separate read endpoint is configured', async () => { const circuit = createDatabaseReadCircuit({ configured: false, maxReplayLagMs: 2_000, probeIntervalMs: 5_000, cooldownMs: 5_000, probe: async () => { throw new Error('should not probe') }, })
expect( await circuit.withRead( async () => 'replica', async () => 'primary', ), ).toBe('primary') expect(circuit.snapshot()).toMatchObject({ configured: false, mode: 'primary', usingPrimaryFallback: false, }) })
test('falls back to primary when the replica probe cannot execute', async () => { const circuit = createDatabaseReadCircuit({ configured: true, maxReplayLagMs: 100, probeIntervalMs: 5_000, cooldownMs: 5_000, probe: async () => { throw new Error('receiver probe permission denied') }, })
expect((await circuit.probeNow()).mode).toBe('unavailable') expect( await circuit.withRead( async () => 'replica', async () => 'primary', ), ).toBe('primary') })
test('uses a quiet caught-up read-only replica endpoint', async () => { let now = 1_000 const circuit = createDatabaseReadCircuit({ configured: true, maxReplayLagMs: 2_000, probeIntervalMs: 5_000, cooldownMs: 5_000, now: () => now, probe: async () => ({ transactionReadOnly: true, sensitiveDataRestricted: true, writePrivilegesRestricted: true, inRecovery: true, replicationReceiverHealthy: true, replayLagMs: 0, }), })
expect((await circuit.probeNow()).mode).toBe('healthy') expect( await circuit.withRead( async () => 'replica', async () => 'primary', ), ).toBe('replica') now += 1 expect(circuit.snapshot().usingPrimaryFallback).toBe(false) })
test('falls back to primary when replica replay lag exceeds the configured bound', async () => { const circuit = createDatabaseReadCircuit({ configured: true, maxReplayLagMs: 100, probeIntervalMs: 5_000, cooldownMs: 5_000, probe: async () => ({ transactionReadOnly: true, sensitiveDataRestricted: true, writePrivilegesRestricted: true, inRecovery: true, replicationReceiverHealthy: true, replayLagMs: 101, }), })
expect((await circuit.probeNow()).mode).toBe('lagging') expect( await circuit.withRead( async () => 'replica', async () => 'primary', ), ).toBe('primary') })
test('falls back when a replica has WAL backlog but no replay timestamp', async () => { const circuit = createDatabaseReadCircuit({ configured: true, maxReplayLagMs: 100, probeIntervalMs: 5_000, cooldownMs: 5_000, probe: async () => ({ transactionReadOnly: true, sensitiveDataRestricted: true, writePrivilegesRestricted: true, inRecovery: true, replicationReceiverHealthy: true, replayLagMs: null, }), })
expect((await circuit.probeNow()).mode).toBe('lagging') expect( await circuit.withRead( async () => 'replica', async () => 'primary', ), ).toBe('primary') })
test('falls back when a caught-up replica receiver is disconnected or stale', async () => { const circuit = createDatabaseReadCircuit({ configured: true, maxReplayLagMs: 100, probeIntervalMs: 5_000, cooldownMs: 5_000, probe: async () => ({ transactionReadOnly: true, sensitiveDataRestricted: true, writePrivilegesRestricted: true, inRecovery: true, replicationReceiverHealthy: false, replayLagMs: 0, }), })
expect((await circuit.probeNow()).mode).toBe('receiver_unhealthy') expect( await circuit.withRead( async () => 'replica', async () => 'primary', ), ).toBe('primary') })
test('accepts a local primary endpoint only when its session is read-only', async () => { const circuit = createDatabaseReadCircuit({ configured: true, maxReplayLagMs: 100, probeIntervalMs: 5_000, cooldownMs: 5_000, probe: async () => ({ transactionReadOnly: true, sensitiveDataRestricted: true, writePrivilegesRestricted: true, inRecovery: false, replicationReceiverHealthy: true, replayLagMs: null, }), })
expect((await circuit.probeNow()).mode).toBe('healthy') expect( await circuit.withRead( async () => 'read-only-primary', async () => 'primary', ), ).toBe('read-only-primary') })
test('falls back when the configured role can read sensitive tables', async () => { const circuit = createDatabaseReadCircuit({ configured: true, maxReplayLagMs: 100, probeIntervalMs: 5_000, cooldownMs: 5_000, probe: async () => ({ transactionReadOnly: true, sensitiveDataRestricted: false, writePrivilegesRestricted: true, inRecovery: false, replicationReceiverHealthy: true, replayLagMs: null, }), })
expect((await circuit.probeNow()).mode).toBe('unsafe') expect( await circuit.withRead( async () => 'replica', async () => 'primary', ), ).toBe('primary') })
test('falls back when the configured role has write privileges', async () => { const circuit = createDatabaseReadCircuit({ configured: true, maxReplayLagMs: 100, probeIntervalMs: 5_000, cooldownMs: 5_000, probe: async () => ({ transactionReadOnly: true, sensitiveDataRestricted: true, writePrivilegesRestricted: false, inRecovery: false, replicationReceiverHealthy: true, replayLagMs: null, }), })
expect((await circuit.probeNow()).mode).toBe('unsafe') expect( await circuit.withRead( async () => 'replica', async () => 'primary', ), ).toBe('primary') })
test('opens the circuit after a replica query failure and does not retry during cooldown', async () => { let now = 1_000 let probeCalls = 0 let replicaCalls = 0 const circuit = createDatabaseReadCircuit({ configured: true, maxReplayLagMs: 100, probeIntervalMs: 5_000, cooldownMs: 5_000, now: () => now, probe: async () => { probeCalls++ return { transactionReadOnly: true, sensitiveDataRestricted: true, writePrivilegesRestricted: true, inRecovery: true, replicationReceiverHealthy: true, replayLagMs: 1, } }, })
await circuit.probeNow() expect( await circuit.withRead( async () => { replicaCalls++ throw new Error('postgres://secret@replica/wisp') }, async () => 'primary', ), ).toBe('primary') expect(circuit.snapshot().mode).toBe('unavailable')
expect( await circuit.withRead( async () => 'replica', async () => 'primary', ), ).toBe('primary') expect({ probeCalls, replicaCalls }).toEqual({ probeCalls: 1, replicaCalls: 1 }) now += 5_000 expect( await circuit.withRead( async () => 'replica', async () => 'primary', ), ).toBe('replica') expect(probeCalls).toBe(2) })
test('shares concurrent probes and retains no driver error in health', async () => { let resolveProbe: | ((value: { transactionReadOnly: boolean sensitiveDataRestricted: boolean writePrivilegesRestricted: boolean inRecovery: boolean replicationReceiverHealthy: boolean replayLagMs: number }) => void) | undefined let probeCalls = 0 const circuit = createDatabaseReadCircuit({ configured: true, maxReplayLagMs: 100, probeIntervalMs: 5_000, cooldownMs: 5_000, probe: async () => { probeCalls++ return await new Promise<{ transactionReadOnly: boolean sensitiveDataRestricted: boolean writePrivilegesRestricted: boolean inRecovery: boolean replicationReceiverHealthy: boolean replayLagMs: number }>((resolve) => { resolveProbe = resolve }) }, })
const first = circuit.withRead( async () => 'replica-one', async () => 'primary-one', ) const second = circuit.withRead( async () => 'replica-two', async () => 'primary-two', ) await Promise.resolve() expect(probeCalls).toBe(1) resolveProbe?.({ transactionReadOnly: true, sensitiveDataRestricted: true, writePrivilegesRestricted: true, inRecovery: true, replicationReceiverHealthy: true, replayLagMs: 1, }) expect(await Promise.all([first, second])).toEqual(['replica-one', 'replica-two']) })})