import { describe, expect, test } from 'bun:test'
import { config } from '../config'
import {
boundedCursorFromStoredValue,
type CursorRedisClient,
DELETE_MIGRATED_CURSOR_SCRIPT,
isValidCursor,
type LeaderElectionDependencies,
legacyCursorKey,
MIGRATE_CURSOR_SCRIPT,
parseStoredCursor,
readDurableCursorWithRedis,
relayFingerprint,
runLeaderElection,
SAVE_CURSOR_SCRIPT,
shouldAdvanceStoredCursor,
} from './leader'
describe('stored leader cursors', () => {
test('accepts only nonnegative decimal safe integers', () => {
expect(parseStoredCursor('0')).toBe(0)
expect(parseStoredCursor('9007199254740991')).toBe(Number.MAX_SAFE_INTEGER)
expect(isValidCursor(42)).toBe(true)
for (const value of ['-1', '+1', '1.5', '1e3', ' 1', '9007199254740992', 'not-a-number']) {
expect(parseStoredCursor(value)).toBeUndefined()
expect(boundedCursorFromStoredValue(value)).toBeUndefined()
}
expect(boundedCursorFromStoredValue(null)).toBeUndefined()
expect(isValidCursor(-1)).toBe(false)
expect(isValidCursor(Number.MAX_SAFE_INTEGER + 1)).toBe(false)
})
test('keeps a relay cursor monotonic and repairs corrupt stored values', () => {
expect(shouldAdvanceStoredCursor('41', 42)).toBe(true)
expect(shouldAdvanceStoredCursor('42', 42)).toBe(false)
expect(shouldAdvanceStoredCursor('43', 42)).toBe(false)
expect(shouldAdvanceStoredCursor('corrupt', 42)).toBe(true)
expect(shouldAdvanceStoredCursor(null, 42)).toBe(true)
expect(shouldAdvanceStoredCursor('1', -1)).toBe(false)
})
test('derives the exact same-relay legacy checkpoint key without credentials', () => {
expect(legacyCursorKey('wss://user:secret@relay.example.invalid:443/path?token=secret')).toBe(
'wisp:firehose-cursor:relay.example.invalid',
)
expect(legacyCursorKey('not a URL')).toBeUndefined()
})
test('keeps legacy host compatibility while separating hashed relay paths', () => {
const first = 'wss://relay.example.invalid/first'
const second = 'wss://relay.example.invalid/second'
expect(legacyCursorKey(first)).toBe(legacyCursorKey(second))
expect(relayFingerprint(first)).not.toBe(relayFingerprint(second))
})
test('uses a normalized non-secret relay fingerprint', () => {
const clean = relayFingerprint('wss://relay.example.invalid/path')
const credentialed = relayFingerprint('wss://user:secret@relay.example.invalid/path?token=secret')
expect(clean).toBe(credentialed)
expect(clean).toMatch(/^[a-f0-9]{64}$/)
expect(clean).not.toContain('relay.example')
expect(clean).not.toContain('secret')
})
})
const MIGRATION_SERVICE = 'wss://relay.example.invalid'
const MIGRATION_HASHED_KEY = `wisp:firehose-cursor:${relayFingerprint(MIGRATION_SERVICE)}`
const MIGRATION_LEGACY_KEY = legacyCursorKey(MIGRATION_SERVICE) as string
const AMBIGUOUS_SERVICE = 'wss://relay.example.invalid/independent'
const AMBIGUOUS_HASHED_KEY = `wisp:firehose-cursor:${relayFingerprint(AMBIGUOUS_SERVICE)}`
const AMBIGUOUS_LEGACY_KEY = legacyCursorKey(AMBIGUOUS_SERVICE) as string
type EvalCall = { script: string; keyCount: number; args: Array<string | number> }
class FakeCursorRedis implements CursorRedisClient {
readonly values = new Map<string, string>()
readonly getCalls: string[] = []
readonly evalCalls: EvalCall[] = []
migrationFailure: Error | undefined
deletionFailure: Error | undefined
beforeHashedRead: (() => void) | undefined
beforeDeletion: (() => void) | undefined
async get(key: string): Promise<string | null> {
this.getCalls.push(key)
if (key === MIGRATION_HASHED_KEY && this.beforeHashedRead) {
const hook = this.beforeHashedRead
this.beforeHashedRead = undefined
hook()
}
return this.values.get(key) ?? null
}
async eval(script: string, keyCount: number, ...args: Array<string | number>): Promise<unknown> {
this.evalCalls.push({ script, keyCount, args })
if (script === SAVE_CURSOR_SCRIPT) {
const [key, nextValue] = args
if (typeof key !== 'string' || typeof nextValue !== 'string') throw new Error('invalid fake save args')
const current = parseStoredCursor(this.values.get(key) ?? null)
const next = parseStoredCursor(nextValue)
if (next === undefined) throw new Error('invalid fake cursor')
if (current !== undefined && current >= next) return 0
this.values.set(key, nextValue)
return 1
}
if (script === MIGRATE_CURSOR_SCRIPT) {
if (this.migrationFailure) throw this.migrationFailure
const [hashedKey, legacyKey] = args
if (typeof hashedKey !== 'string' || typeof legacyKey !== 'string') throw new Error('invalid fake keys')
const currentValue = this.values.get(hashedKey)
const current = parseStoredCursor(currentValue ?? null)
const legacy = parseStoredCursor(this.values.get(legacyKey) ?? null)
if (legacy === undefined) return [0, currentValue ?? '']
const selected = current !== undefined && current >= legacy ? current : legacy
if (current === undefined || current !== selected) this.values.set(hashedKey, String(selected))
return [1, String(selected)]
}
if (script === DELETE_MIGRATED_CURSOR_SCRIPT) {
if (this.deletionFailure) throw this.deletionFailure
this.beforeDeletion?.()
this.beforeDeletion = undefined
const [hashedKey, legacyKey, expectedValue] = args
if (typeof hashedKey !== 'string' || typeof legacyKey !== 'string' || typeof expectedValue !== 'string') {
throw new Error('invalid fake deletion args')
}
const current = parseStoredCursor(this.values.get(hashedKey) ?? null)
const legacy = parseStoredCursor(this.values.get(legacyKey) ?? null)
const expected = parseStoredCursor(expectedValue)
if (current === undefined) return -2
if (expected === undefined) return -4
if (legacy !== undefined && current < legacy) return -3
if (current < expected) return -4
if (!this.values.has(legacyKey)) return 0
if (legacy === undefined) return -1
this.values.delete(legacyKey)
return 1
}
throw new Error('unknown fake script')
}
}
describe('atomic cursor saves', () => {
test('repairs a stored decimal above Number.MAX_SAFE_INTEGER', async () => {
const redis = new FakeCursorRedis()
redis.values.set(MIGRATION_HASHED_KEY, '9007199254740992')
await redis.eval(SAVE_CURSOR_SCRIPT, 1, MIGRATION_HASHED_KEY, '42')
expect(redis.values.get(MIGRATION_HASHED_KEY)).toBe('42')
})
})
describe('same-relay cursor migration', () => {
test('atomically migrates, re-reads, and removes the legacy key', async () => {
const redis = new FakeCursorRedis()
redis.values.set(MIGRATION_LEGACY_KEY, '41')
const result = await readDurableCursorWithRedis(redis, MIGRATION_SERVICE)
expect(result).toEqual({ read: { kind: 'found', cursor: 41 }, migrated: true, corrupt: false })
expect(redis.values.get(MIGRATION_HASHED_KEY)).toBe('41')
expect(redis.values.has(MIGRATION_LEGACY_KEY)).toBe(false)
expect(redis.getCalls).toEqual([MIGRATION_HASHED_KEY])
expect(redis.evalCalls.map(({ script }) => script)).toEqual([MIGRATE_CURSOR_SCRIPT, DELETE_MIGRATED_CURSOR_SCRIPT])
expect(redis.evalCalls[1]?.args).toEqual([MIGRATION_HASHED_KEY, MIGRATION_LEGACY_KEY, '41'])
})
test('keeps a newer hashed cursor and still removes the older legacy key', async () => {
const redis = new FakeCursorRedis()
redis.values.set(MIGRATION_HASHED_KEY, '80')
redis.values.set(MIGRATION_LEGACY_KEY, '41')
const result = await readDurableCursorWithRedis(redis, MIGRATION_SERVICE)
expect(result.read).toEqual({ kind: 'found', cursor: 80 })
expect(redis.values.has(MIGRATION_LEGACY_KEY)).toBe(false)
expect(redis.evalCalls[1]?.args).toEqual([MIGRATION_HASHED_KEY, MIGRATION_LEGACY_KEY, '80'])
})
test('returns missing and preserves a malformed legacy value', async () => {
const redis = new FakeCursorRedis()
redis.values.set(MIGRATION_LEGACY_KEY, 'not-a-cursor')
const result = await readDurableCursorWithRedis(redis, MIGRATION_SERVICE)
expect(result).toEqual({ read: { kind: 'missing' }, migrated: false, corrupt: false })
expect(redis.values.get(MIGRATION_LEGACY_KEY)).toBe('not-a-cursor')
expect(redis.evalCalls).toHaveLength(1)
})
test('does not delete malformed legacy data when the hashed key is valid', async () => {
const redis = new FakeCursorRedis()
redis.values.set(MIGRATION_HASHED_KEY, '80')
redis.values.set(MIGRATION_LEGACY_KEY, 'not-a-cursor')
const result = await readDurableCursorWithRedis(redis, MIGRATION_SERVICE)
expect(result.read).toEqual({ kind: 'found', cursor: 80 })
expect(redis.values.get(MIGRATION_LEGACY_KEY)).toBe('not-a-cursor')
expect(redis.evalCalls).toHaveLength(1)
})
test('fails closed and keeps the legacy value when migration cannot run', async () => {
const redis = new FakeCursorRedis()
redis.values.set(MIGRATION_LEGACY_KEY, '41')
redis.migrationFailure = new Error('migration unavailable')
await expect(readDurableCursorWithRedis(redis, MIGRATION_SERVICE)).rejects.toThrow('migration unavailable')
expect(redis.values.has(MIGRATION_HASHED_KEY)).toBe(false)
expect(redis.values.get(MIGRATION_LEGACY_KEY)).toBe('41')
})
test('fails closed and keeps the legacy value when deletion cannot run', async () => {
const redis = new FakeCursorRedis()
redis.values.set(MIGRATION_LEGACY_KEY, '41')
redis.deletionFailure = new Error('deletion unavailable')
await expect(readDurableCursorWithRedis(redis, MIGRATION_SERVICE)).rejects.toThrow('deletion unavailable')
expect(redis.values.get(MIGRATION_HASHED_KEY)).toBe('41')
expect(redis.values.get(MIGRATION_LEGACY_KEY)).toBe('41')
})
test('returns the newer hashed cursor when it wins during the re-read', async () => {
const redis = new FakeCursorRedis()
redis.values.set(MIGRATION_LEGACY_KEY, '41')
redis.beforeHashedRead = () => redis.values.set(MIGRATION_HASHED_KEY, '77')
const result = await readDurableCursorWithRedis(redis, MIGRATION_SERVICE)
expect(result.read).toEqual({ kind: 'found', cursor: 77 })
expect(redis.values.has(MIGRATION_LEGACY_KEY)).toBe(false)
})
test('retries when a newer legacy value wins the delete race', async () => {
const redis = new FakeCursorRedis()
redis.values.set(MIGRATION_LEGACY_KEY, '41')
let raced = false
redis.beforeDeletion = () => {
if (!raced) {
raced = true
redis.values.set(MIGRATION_LEGACY_KEY, '80')
}
}
const result = await readDurableCursorWithRedis(redis, MIGRATION_SERVICE)
expect(result.read).toEqual({ kind: 'found', cursor: 80 })
expect(redis.values.get(MIGRATION_HASHED_KEY)).toBe('80')
expect(redis.values.has(MIGRATION_LEGACY_KEY)).toBe(false)
expect(redis.evalCalls.map(({ script }) => script)).toEqual([
MIGRATE_CURSOR_SCRIPT,
DELETE_MIGRATED_CURSOR_SCRIPT,
MIGRATE_CURSOR_SCRIPT,
DELETE_MIGRATED_CURSOR_SCRIPT,
])
})
test('keeps the legacy value when the hashed value becomes malformed after migration', async () => {
const redis = new FakeCursorRedis()
redis.values.set(MIGRATION_LEGACY_KEY, '41')
redis.beforeHashedRead = () => redis.values.set(MIGRATION_HASHED_KEY, 'corrupt')
await expect(readDurableCursorWithRedis(redis, MIGRATION_SERVICE)).rejects.toThrow('legacy migration was confirmed')
expect(redis.values.get(MIGRATION_LEGACY_KEY)).toBe('41')
})
test('does not trust a stale re-read if both keys disappear before deletion', async () => {
const redis = new FakeCursorRedis()
redis.values.set(MIGRATION_LEGACY_KEY, '41')
redis.beforeDeletion = () => {
redis.values.delete(MIGRATION_HASHED_KEY)
redis.values.delete(MIGRATION_LEGACY_KEY)
}
await expect(readDurableCursorWithRedis(redis, MIGRATION_SERVICE)).rejects.toThrow('legacy migration was confirmed')
})
test('fails closed instead of importing an ambiguous non-root legacy cursor', async () => {
const redis = new FakeCursorRedis()
redis.values.set(AMBIGUOUS_LEGACY_KEY, '41')
await expect(readDurableCursorWithRedis(redis, AMBIGUOUS_SERVICE)).rejects.toThrow(/preseed .*correct checkpoint/i)
expect(redis.values.has(AMBIGUOUS_HASHED_KEY)).toBe(false)
expect(redis.values.get(AMBIGUOUS_LEGACY_KEY)).toBe('41')
expect(redis.evalCalls).toHaveLength(0)
})
test('fails closed when a corrupt path cursor meets an ambiguous legacy cursor', async () => {
const redis = new FakeCursorRedis()
redis.values.set(AMBIGUOUS_HASHED_KEY, 'corrupt')
redis.values.set(AMBIGUOUS_LEGACY_KEY, '41')
await expect(readDurableCursorWithRedis(redis, AMBIGUOUS_SERVICE)).rejects.toThrow(/ambiguous legacy cursor/i)
expect(redis.values.get(AMBIGUOUS_HASHED_KEY)).toBe('corrupt')
expect(redis.values.get(AMBIGUOUS_LEGACY_KEY)).toBe('41')
})
test('does not consult or delete ambiguous legacy data when a path cursor is valid', async () => {
const redis = new FakeCursorRedis()
redis.values.set(AMBIGUOUS_HASHED_KEY, '80')
redis.values.set(AMBIGUOUS_LEGACY_KEY, '41')
const result = await readDurableCursorWithRedis(redis, AMBIGUOUS_SERVICE)
expect(result).toEqual({ read: { kind: 'found', cursor: 80 }, migrated: false, corrupt: false })
expect(redis.values.get(AMBIGUOUS_LEGACY_KEY)).toBe('41')
expect(redis.getCalls).toEqual([AMBIGUOUS_HASHED_KEY])
})
})
function deferred<T>() {
let resolve!: (value: T | PromiseLike<T>) => void
const promise = new Promise<T>((resolvePromise) => {
resolve = resolvePromise
})
return { promise, resolve }
}
function lifecycleDependencies(overrides: Partial<LeaderElectionDependencies> = {}): LeaderElectionDependencies {
return {
tryBecomeLeader: async () => true,
renewLeadership: async () => true,
releaseLeadership: async () => undefined,
readCursor: async () => undefined,
sleep: async (_ms, signal) => {
if (signal.aborted) return
await new Promise<void>((resolve) => signal.addEventListener('abort', () => resolve(), { once: true }))
},
...overrides,
}
}
describe('leader drain lifecycle', () => {
test('keeps the lease until the drain-aware loss callback settles', async () => {
const controller = new AbortController()
const started = deferred<void>()
const drain = deferred<boolean>()
let stepDown: (() => Promise<void>) | undefined
let receivedCursor: number | undefined = 42
let releases = 0
const election = runLeaderElection(
(cursor, receivedStepDown) => {
receivedCursor = cursor
stepDown = receivedStepDown
started.resolve()
},
async () => await drain.promise,
controller.signal,
'wss://relay.example.invalid',
lifecycleDependencies({
releaseLeadership: async () => {
releases++
},
}),
)
await started.promise
// A confirmed missing checkpoint starts live rather than being rewritten
// as a synthetic zero cursor.
expect(receivedCursor).toBeUndefined()
const stopping = stepDown?.()
expect(stopping).toBeDefined()
await Promise.resolve()
expect(releases).toBe(0)
drain.resolve(true)
await stopping
expect(releases).toBe(1)
controller.abort()
await election
})
test('does not start relay work when the durable cursor read fails', async () => {
const controller = new AbortController()
let starts = 0
let losses = 0
let releases = 0
const election = runLeaderElection(
() => {
starts++
},
async () => {
losses++
return true
},
controller.signal,
'wss://relay.example.invalid',
lifecycleDependencies({
readCursor: async () => {
throw new Error('cursor store unavailable')
},
releaseLeadership: async () => {
releases++
},
sleep: async () => {
controller.abort()
},
}),
)
await election
expect(starts).toBe(0)
expect(losses).toBe(1)
expect(releases).toBe(1)
})
test('does not start stale leader work after the lease is lost during cursor read', async () => {
const mutableConfig = config as unknown as {
leaderTtlMs: number
leaderRenewIntervalMs: number
leaderPollIntervalMs: number
}
const originalConfig = { ...mutableConfig }
Object.assign(mutableConfig, { leaderTtlMs: 110, leaderRenewIntervalMs: 1000, leaderPollIntervalMs: 5 })
const controller = new AbortController()
let fakeNow = 0
const readStarted = deferred<void>()
const readResult = deferred<number | undefined>()
type ScheduledTimer = { callback: () => void; delay: number; cancelled: boolean }
const timers: ScheduledTimer[] = []
let starts = 0
let losses = 0
let releases = 0
let election: Promise<void> | undefined
try {
election = runLeaderElection(
() => {
starts++
},
async () => {
losses++
controller.abort()
return true
},
controller.signal,
'wss://relay.example.invalid',
lifecycleDependencies({
readCursor: async () => {
readStarted.resolve()
return readResult.promise
},
releaseLeadership: async () => {
releases++
},
now: () => fakeNow,
setTimeout: (callback, delay) => {
const timer: ScheduledTimer = { callback, delay, cancelled: false }
timers.push(timer)
return timer as unknown as ReturnType<typeof setTimeout>
},
clearTimeout: (timer) => {
;(timer as unknown as ScheduledTimer).cancelled = true
},
}),
)
await readStarted.promise
expect(timers[0]?.delay).toBe(10)
fakeNow = 10
timers[0]?.callback()
expect(losses).toBe(1)
readResult.resolve(42)
await election
expect(starts).toBe(0)
expect(releases).toBe(0)
} finally {
readResult.resolve(undefined)
controller.abort()
await election?.catch(() => undefined)
Object.assign(mutableConfig, originalConfig)
}
})
test('does not release a lease after a forced drain result', async () => {
const controller = new AbortController()
const started = deferred<void>()
let stepDown: (() => Promise<void>) | undefined
let receivedCursor: number | undefined = 42
let releases = 0
const election = runLeaderElection(
(cursor, receivedStepDown) => {
receivedCursor = cursor
stepDown = receivedStepDown
started.resolve()
},
async () => false,
controller.signal,
'wss://relay.example.invalid',
lifecycleDependencies({
releaseLeadership: async () => {
releases++
},
}),
)
await started.promise
// A confirmed missing checkpoint starts live rather than being rewritten
// as a synthetic zero cursor.
expect(receivedCursor).toBeUndefined()
await stepDown?.()
expect(releases).toBe(0)
await expect(election).rejects.toThrow('forced leadership drain')
controller.abort()
})
test('fences a hung renewal at the monotonic lease safety deadline', async () => {
const mutableConfig = config as unknown as {
leaderTtlMs: number
leaderRenewIntervalMs: number
leaderPollIntervalMs: number
}
const originalConfig = { ...mutableConfig }
Object.assign(mutableConfig, { leaderTtlMs: 400, leaderRenewIntervalMs: 25, leaderPollIntervalMs: 5 })
const controller = new AbortController()
const started = deferred<void>()
const renewalStarted = deferred<void>()
const hangingRenewal = deferred<boolean>()
let fakeNow = 0
let losses = 0
let starts = 0
let renewals = 0
let election: Promise<void> | undefined
try {
election = runLeaderElection(
() => {
starts++
if (starts === 1) started.resolve()
},
async () => {
losses++
return true
},
controller.signal,
'wss://relay.example.invalid',
lifecycleDependencies({
renewLeadership: async () => {
renewals++
if (renewals === 1) {
renewalStarted.resolve()
return hangingRenewal.promise
}
controller.abort()
return true
},
sleep: async (_ms, signal) => {
await new Promise<void>((resolve) => setTimeout(resolve, 2))
fakeNow += 10
if (signal.aborted) return
},
now: () => fakeNow,
}),
)
await started.promise
await renewalStarted.promise
await election
expect(losses).toBe(2)
expect(starts).toBe(2)
expect(renewals).toBe(2)
expect(fakeNow).toBeGreaterThanOrEqual(300)
} finally {
hangingRenewal.resolve(true)
controller.abort()
await election?.catch(() => undefined)
Object.assign(mutableConfig, originalConfig)
}
})
})