import { afterEach, describe, expect, test } from 'bun:test'
import type { Record as WispFsRecord } from '@wispplace/lexicons/types/place/wisp/fs'
import type { Record as WispSettings } from '@wispplace/lexicons/types/place/wisp/settings'
import type Redis from 'ioredis'
import { AuthoritativeSettingsRecordError, type SettingsUpdateOptions, SiteBlobBackoffError } from './cache-writer'
import {
SETTINGS_DELETE_FAILURE_REASON,
SETTINGS_UPDATE_FAILURE_REASON,
SITE_DELETE_TOMBSTONE_REASON,
} from './revalidate-queue'
import { createRevalidationResourceContext } from './revalidate-resources'
import {
createRevalidateWorkerRedisOptions,
getRevalidateWorkerStateForTests,
processRevalidationMessage,
type RevalidateRedisClient,
type RevalidateWorkerDependencies,
type RevalidateWorkerRedisFactory,
type RevalidateWorkerRedisOptions,
type RevalidateWorkerRuntimeConfig,
resolveRevalidateWorkerRuntimeConfig,
scanStaleRevalidationMessagesForTests,
shouldSkipInvalidationForReason,
startRevalidateWorkerForTests,
stopRevalidateWorker,
trimDrainedRevalidationEntries,
} from './revalidate-worker'
const DID = 'did:plc:test'
const RKEY = 'site'
const MESSAGE_ID = '1-0'
function messageFields(reason: string): string[] {
return ['did', DID, 'rkey', RKEY, 'reason', reason]
}
function createRedis(
ttl = -1,
completionOutcome: [number, number] | Error = [1, 1],
): {
redis: RevalidateRedisClient
acks: string[]
streamDeletes: string[]
calls: string[]
sets: Array<[string, string, 'EX', number]>
} {
const acks: string[] = []
const streamDeletes: string[] = []
const calls: string[] = []
const sets: Array<[string, string, 'EX', number]> = []
return {
redis: {
async ttl() {
return ttl
},
async eval(_script, keyCount, ...args) {
const id = args[keyCount + 1] as string
calls.push(`eval-complete:${id}`)
if (completionOutcome instanceof Error) throw completionOutcome
if (completionOutcome[0] === 1) acks.push(id)
if (completionOutcome[0] === 1 && completionOutcome[1] === 1) streamDeletes.push(id)
return completionOutcome
},
async set(key, value, expirationMode, ttlSeconds) {
sets.push([key, value, expirationMode, ttlSeconds])
return 'OK'
},
},
acks,
streamDeletes,
calls,
sets,
}
}
function createDependencies(overrides: Partial<RevalidateWorkerDependencies> = {}): RevalidateWorkerDependencies {
return {
fetchSettingsRecord: async () => null,
fetchSiteRecord: async () => null,
handleSettingsDelete: async () => undefined,
handleSettingsUpdate: async () => undefined,
handleSiteCreateOrUpdate: async () => undefined,
handleSiteDelete: async () => undefined,
...overrides,
}
}
const reappearedRecord: WispFsRecord = {
$type: 'place.wisp.fs',
site: RKEY,
root: { type: 'directory', entries: [] },
createdAt: '2024-01-01T00:00:00.000Z',
}
const reappearedSettings: WispSettings = {
$type: 'place.wisp.settings',
directoryListing: false,
cleanUrls: true,
}
async function runSettingsReconciliationForTest(
options: SettingsUpdateOptions | undefined,
onPresent: (record: WispSettings, cid: string) => void,
onAbsent: () => void,
): Promise<void> {
const outcome = await options?.fetchSettingsRecordOutcome?.(DID, RKEY, undefined, options.resources)
if (!outcome) throw new Error('Expected the locked settings lookup seam')
if (outcome.kind === 'retryable') throw new AuthoritativeSettingsRecordError(outcome.error)
if (outcome.kind === 'present') onPresent(outcome.record, outcome.cid)
else onAbsent()
}
async function flushAsyncWork(): Promise<void> {
for (let i = 0; i < 8; i++) {
await Promise.resolve()
}
}
function deferred<T = void>() {
let resolve!: (value: T | PromiseLike<T>) => void
const promise = new Promise<T>((resolvePromise) => {
resolve = resolvePromise
})
return { promise, resolve }
}
class FakeLifecycleRedis {
readonly xreadArguments: unknown[][] = []
claimCalls = 0
evalCalls = 0
disconnected = false
private rejectPendingRead: ((reason?: unknown) => void) | null = null
constructor(
readonly options: RevalidateWorkerRedisOptions,
private readonly xgroupFailure?: Error,
private readonly xautoclaimFailure?: Error,
) {}
on(_event: string, _listener: (..._args: unknown[]) => void): this {
return this
}
async xgroup(..._args: unknown[]): Promise<unknown> {
if (this.xgroupFailure) throw this.xgroupFailure
return 'OK'
}
async xautoclaim(..._args: unknown[]): Promise<unknown> {
this.claimCalls++
if (this.xautoclaimFailure) throw this.xautoclaimFailure
return ['0-0', []]
}
async ttl(_key: string): Promise<number> {
return -1
}
async eval(..._args: unknown[]): Promise<unknown> {
this.evalCalls++
return [1, 1]
}
xreadgroup(...args: unknown[]): Promise<null | [string, Array<[string, string[]]>][]> {
this.xreadArguments.push(args)
return new Promise((_, reject) => {
this.rejectPendingRead = reject
})
}
disconnect(): void {
this.disconnected = true
this.rejectPendingRead?.(new Error('Redis disconnected'))
this.rejectPendingRead = null
}
}
class FakeHealthyThenFailureRedis extends FakeLifecycleRedis {
override async xautoclaim(..._args: unknown[]): Promise<unknown> {
this.claimCalls++
if (this.claimCalls === 1) return ['0-0', []]
throw new Error('connection failed after healthy read')
}
override xreadgroup(...args: unknown[]): Promise<null> {
this.xreadArguments.push(args)
return Promise.resolve(null)
}
override async eval(..._args: unknown[]): Promise<number> {
return 0
}
}
class FakeMessageRedis extends FakeLifecycleRedis {
override xreadgroup(...args: unknown[]): Promise<[string, Array<[string, string[]]>][]> {
this.xreadArguments.push(args)
return Promise.resolve([['wisp:revalidate', [['1-0', messageFields(SITE_DELETE_TOMBSTONE_REASON)]]]])
}
}
function lifecycleFactory(
clients: FakeLifecycleRedis[],
options: { xgroupFailure?: Error; xautoclaimFailure?: Error } = {},
): RevalidateWorkerRedisFactory {
return (_redisUrl, redisOptions) => {
const client = new FakeLifecycleRedis(redisOptions, options.xgroupFailure, options.xautoclaimFailure)
clients.push(client)
return client as unknown as Redis
}
}
interface FakePipeline {
ttl(key: string): FakePipeline
exec(): Promise<unknown>
}
class FakeClaimRedis {
readonly calls: Array<{ command: string; args: unknown[] }> = []
readonly deliveryCounts = new Map<string, number>()
readonly quarantineAttempts: number[] = []
readonly ids: string[]
readonly id: string
readonly fields: string[]
readonly retryTtlOverrides = new Map<string, number>()
private pending = true
retryTtlSeconds = 0
siteBackoffTtlSeconds = -1
pipelineResultOverride: unknown = undefined
constructor(id: string | readonly string[], fields: string[]) {
this.ids = typeof id === 'string' ? [id] : [...id]
const firstId = this.ids[0]
if (!firstId) throw new Error('FakeClaimRedis needs at least one ID')
this.id = firstId
this.fields = fields
}
private ttlForKey(key: string): number {
if (key.startsWith('revalidate:retry:')) {
const id = key.slice('revalidate:retry:'.length)
return this.retryTtlOverrides.get(id) ?? this.retryTtlSeconds
}
if (key.startsWith('revalidate:site:failure-backoff:')) return this.siteBackoffTtlSeconds
return -1
}
async ttl(key: string): Promise<number> {
this.calls.push({ command: 'TTL', args: [key] })
return this.ttlForKey(key)
}
pipeline(): FakePipeline {
const keys: string[] = []
const pipeline: FakePipeline = {
ttl: (key) => {
keys.push(key)
return pipeline
},
exec: async () => {
this.calls.push({ command: 'PIPELINE_EXEC', args: [...keys] })
return this.pipelineResultOverride ?? keys.map((key) => [null, this.ttlForKey(key)])
},
}
return pipeline
}
async xautoclaim(...args: unknown[]): Promise<unknown> {
this.calls.push({ command: 'XAUTOCLAIM', args })
return ['0-0', this.pending ? this.ids : [], []]
}
async xclaim(...args: unknown[]): Promise<unknown> {
this.calls.push({ command: 'XCLAIM', args })
const ids = args.slice(4).filter((value): value is string => typeof value === 'string')
return ids
.filter((id) => this.ids.includes(id) && this.pending)
.map((id) => {
this.deliveryCounts.set(id, (this.deliveryCounts.get(id) ?? 0) + 1)
return [id, this.fields]
})
}
async xpending(...args: unknown[]): Promise<unknown> {
this.calls.push({ command: 'XPENDING', args })
const id = String(args[2])
return [[id, 'consumer', 0, this.deliveryCounts.get(id) ?? 0]]
}
async set(key: string, _value: string, _mode: 'EX', ttlSeconds: number): Promise<'OK'> {
this.calls.push({ command: 'SET', args: [key, ttlSeconds] })
if (key === `revalidate:retry:${this.id}`) this.retryTtlSeconds = ttlSeconds
else if (key.startsWith('revalidate:retry:'))
this.retryTtlOverrides.set(key.slice('revalidate:retry:'.length), ttlSeconds)
else if (key.startsWith('revalidate:site:failure-backoff:')) this.siteBackoffTtlSeconds = ttlSeconds
return 'OK'
}
async eval(_script: string, keyCount: number, ...args: string[]): Promise<unknown> {
this.calls.push({ command: 'EVAL', args: [keyCount, ...args] })
if (keyCount === 4) {
this.quarantineAttempts.push(Number(args[4 + 7]))
this.pending = false
return [1, 'dlq-1', 1]
}
this.pending = false
return [1, 1]
}
}
afterEach(async () => {
await stopRevalidateWorker()
})
describe('shouldSkipInvalidationForReason', () => {
test('skips invalidation for rewrite repair jobs', () => {
expect(shouldSkipInvalidationForReason('rewrite-miss:docs/w/~/index.html')).toBe(true)
})
test('does not skip invalidation for storage misses', () => {
expect(shouldSkipInvalidationForReason('storage-miss:docs/raw/README.md')).toBe(false)
})
test('does not skip invalidation for other revalidate reasons', () => {
expect(shouldSkipInvalidationForReason('manual')).toBe(false)
})
})
describe('delete tombstone revalidation', () => {
test('deletes a missing PDS record and ACKs only after delete succeeds', async () => {
const { redis, acks, streamDeletes } = createRedis(60)
const deletes: Array<[string, string]> = []
const dependencies = createDependencies({
handleSiteDelete: async (did, rkey) => {
deletes.push([did, rkey])
},
})
await processRevalidationMessage(MESSAGE_ID, messageFields(SITE_DELETE_TOMBSTONE_REASON), redis, dependencies)
expect(deletes).toEqual([[DID, RKEY]])
expect(acks).toEqual([MESSAGE_ID])
expect(streamDeletes).toEqual([MESSAGE_ID])
})
test('fully materializes a reappeared PDS record instead of deleting it', async () => {
const { redis, acks, streamDeletes } = createRedis()
const deletes: Array<[string, string]> = []
const updates: Array<{ did: string; rkey: string; cid: string; forceDownload?: boolean }> = []
const dependencies = createDependencies({
fetchSiteRecord: async () => ({ record: reappearedRecord, cid: 'new-cid' }),
handleSiteCreateOrUpdate: async (did, rkey, _record, cid, options) => {
updates.push({ did, rkey, cid, forceDownload: options?.forceDownload })
},
handleSiteDelete: async (did, rkey) => {
deletes.push([did, rkey])
},
})
await processRevalidationMessage(MESSAGE_ID, messageFields(SITE_DELETE_TOMBSTONE_REASON), redis, dependencies)
expect(updates).toEqual([{ did: DID, rkey: RKEY, cid: 'new-cid', forceDownload: true }])
expect(deletes).toEqual([])
expect(acks).toEqual([MESSAGE_ID])
expect(streamDeletes).toEqual([MESSAGE_ID])
})
test('leaves a tombstone pending when delete cleanup has a transient failure', async () => {
const { redis, acks, streamDeletes } = createRedis()
const dependencies = createDependencies({
handleSiteDelete: async () => {
throw new Error('database temporarily unavailable')
},
})
await expect(
processRevalidationMessage(MESSAGE_ID, messageFields(SITE_DELETE_TOMBSTONE_REASON), redis, dependencies),
).rejects.toThrow('database temporarily unavailable')
expect(acks).toEqual([])
expect(streamDeletes).toEqual([])
})
test('aborts a blocked PDS fetch without acknowledging or retrying the pending entry', async () => {
const { redis, acks, streamDeletes, sets, calls } = createRedis()
const controller = new AbortController()
const fetchStarted = deferred<void>()
const dependencies = createDependencies({
fetchSiteRecord: async (_did, _rkey, resources) => {
fetchStarted.resolve()
await new Promise<void>((resolve) =>
resources?.signal.addEventListener('abort', () => resolve(), { once: true }),
)
throw resources?.signal.reason instanceof Error ? resources.signal.reason : new Error('aborted')
},
})
const processing = processRevalidationMessage(
MESSAGE_ID,
messageFields(SITE_DELETE_TOMBSTONE_REASON),
redis,
dependencies,
{ upstreamSignal: controller.signal, enforceAttemptPolicy: true },
)
await fetchStarted.promise
controller.abort()
await processing
expect(acks).toEqual([])
expect(streamDeletes).toEqual([])
expect(sets).toEqual([])
expect(calls).toEqual([])
})
test('aborts a blocked blob operation without acknowledging or retrying the pending entry', async () => {
const { redis, acks, streamDeletes, sets, calls } = createRedis()
const controller = new AbortController()
const materializationStarted = deferred<void>()
const dependencies = createDependencies({
fetchSiteRecord: async () => ({ record: reappearedRecord, cid: 'new-cid' }),
handleSiteCreateOrUpdate: async (_did, _rkey, _record, _cid, options) => {
materializationStarted.resolve()
await new Promise<void>((resolve) =>
options?.resources?.signal.addEventListener('abort', () => resolve(), { once: true }),
)
throw options?.resources?.signal.reason instanceof Error
? options.resources.signal.reason
: new Error('aborted')
},
})
const processing = processRevalidationMessage(
MESSAGE_ID,
messageFields(SITE_DELETE_TOMBSTONE_REASON),
redis,
dependencies,
{ upstreamSignal: controller.signal, enforceAttemptPolicy: true },
)
await materializationStarted.promise
controller.abort()
await processing
expect(acks).toEqual([])
expect(streamDeletes).toEqual([])
expect(sets).toEqual([])
expect(calls).toEqual([])
})
test('treats a supplied resource deadline as a retryable failure, not lifecycle cancellation', async () => {
const { redis, acks, streamDeletes, sets } = createRedis()
const resources = createRevalidationResourceContext(1, 1024)
const dependencies = createDependencies({
fetchSiteRecord: async () => {
await new Promise((resolve) => setTimeout(resolve, 5))
return { record: reappearedRecord, cid: 'new-cid' }
},
})
try {
await processRevalidationMessage(MESSAGE_ID, messageFields(SITE_DELETE_TOMBSTONE_REASON), redis, dependencies, {
resourceContext: resources,
deliveryAttempt: 1,
enforceAttemptPolicy: true,
})
} finally {
resources.close()
}
expect(acks).toEqual([])
expect(streamDeletes).toEqual([])
expect(sets.map(([key]) => key)).toEqual([`revalidate:retry:${MESSAGE_ID}`])
})
test('keeps a reappeared tombstone pending while a blob backoff is active', async () => {
const { redis, acks, streamDeletes } = createRedis(60)
let updates = 0
const dependencies = createDependencies({
fetchSiteRecord: async () => ({ record: reappearedRecord, cid: 'new-cid' }),
handleSiteCreateOrUpdate: async () => {
updates++
},
})
await processRevalidationMessage(MESSAGE_ID, messageFields(SITE_DELETE_TOMBSTONE_REASON), redis, dependencies)
expect(updates).toBe(0)
expect(acks).toEqual([])
expect(streamDeletes).toEqual([])
})
test('keeps a tombstone pending when its replacement record hits blob backoff', async () => {
const { redis, acks, sets, streamDeletes } = createRedis()
const dependencies = createDependencies({
fetchSiteRecord: async () => ({ record: reappearedRecord, cid: 'new-cid' }),
handleSiteCreateOrUpdate: async () => {
throw new SiteBlobBackoffError(DID, RKEY, Date.now() + 60_000, 1)
},
})
await processRevalidationMessage(MESSAGE_ID, messageFields(SITE_DELETE_TOMBSTONE_REASON), redis, dependencies)
expect(sets).toHaveLength(2)
expect(sets.map(([key]) => key)).toEqual([
`revalidate:site:failure-backoff:${DID}:${RKEY}`,
`revalidate:retry:${MESSAGE_ID}`,
])
expect(acks).toEqual([])
expect(streamDeletes).toEqual([])
})
})
describe('settings failure revalidation', () => {
test('applies a current settings record after a failed settings update', async () => {
const { redis, acks, streamDeletes } = createRedis(60)
const updates: Array<{ did: string; rkey: string; cid: string; settings: WispSettings }> = []
let lookups = 0
let deletes = 0
const dependencies = createDependencies({
fetchSettingsRecord: async () => {
lookups++
return { record: reappearedSettings, cid: 'settings-cid' }
},
handleSettingsUpdate: async (did, rkey, _settings, _cid, options) => {
await runSettingsReconciliationForTest(
options,
(settings, cid) => {
updates.push({ did, rkey, settings, cid })
},
() => {
deletes++
},
)
},
})
await processRevalidationMessage(MESSAGE_ID, messageFields(SETTINGS_UPDATE_FAILURE_REASON), redis, dependencies)
expect(lookups).toBe(1)
expect(updates).toEqual([{ did: DID, rkey: RKEY, settings: reappearedSettings, cid: 'settings-cid' }])
expect(deletes).toBe(0)
expect(acks).toEqual([MESSAGE_ID])
expect(streamDeletes).toEqual([MESSAGE_ID])
})
test('keeps settings pending for transient, invalid, or missing-CID PDS outcomes', async () => {
const { redis, acks, streamDeletes } = createRedis()
let deletes = 0
const dependencies = createDependencies({
fetchSettingsRecordOutcome: async () => ({ kind: 'retryable', error: 'MISSING_CID' }),
handleSettingsDelete: async (_did, _rkey, _dependencies, _resources, options) => {
await runSettingsReconciliationForTest(
options,
() => undefined,
() => {
deletes++
},
)
},
})
await processRevalidationMessage(MESSAGE_ID, messageFields(SETTINGS_DELETE_FAILURE_REASON), redis, dependencies)
expect(deletes).toBe(0)
expect(acks).toEqual([])
expect(streamDeletes).toEqual([])
})
test('applies an idempotent settings delete after a failed delete remains absent', async () => {
const { redis, acks, streamDeletes } = createRedis()
const deletes: Array<[string, string]> = []
const dependencies = createDependencies({
fetchSettingsRecordOutcome: async () => ({ kind: 'absent' }),
handleSettingsDelete: async (did, rkey, _dependencies, _resources, options) => {
await runSettingsReconciliationForTest(
options,
() => undefined,
() => {
deletes.push([did, rkey])
},
)
},
})
await processRevalidationMessage(MESSAGE_ID, messageFields(SETTINGS_DELETE_FAILURE_REASON), redis, dependencies)
expect(deletes).toEqual([[DID, RKEY]])
expect(acks).toEqual([MESSAGE_ID])
expect(streamDeletes).toEqual([MESSAGE_ID])
})
test('restores a reappeared settings record instead of deleting it for an old failed delete', async () => {
const { redis, acks, streamDeletes } = createRedis()
let updates = 0
let deletes = 0
const dependencies = createDependencies({
fetchSettingsRecord: async () => ({ record: reappearedSettings, cid: 'settings-cid' }),
handleSettingsDelete: async (_did, _rkey, _dependencies, _resources, options) => {
await runSettingsReconciliationForTest(
options,
() => {
updates++
},
() => {
deletes++
},
)
},
})
await processRevalidationMessage(MESSAGE_ID, messageFields(SETTINGS_DELETE_FAILURE_REASON), redis, dependencies)
expect(updates).toBe(1)
expect(deletes).toBe(0)
expect(acks).toEqual([MESSAGE_ID])
expect(streamDeletes).toEqual([MESSAGE_ID])
})
test('leaves settings failures pending when reconciliation fails', async () => {
const { redis, acks, streamDeletes } = createRedis()
const dependencies = createDependencies({
fetchSettingsRecord: async () => ({ record: reappearedSettings, cid: 'settings-cid' }),
handleSettingsUpdate: async () => {
throw new Error('settings database temporarily unavailable')
},
})
await expect(
processRevalidationMessage(MESSAGE_ID, messageFields(SETTINGS_UPDATE_FAILURE_REASON), redis, dependencies),
).rejects.toThrow('settings database temporarily unavailable')
expect(acks).toEqual([])
expect(streamDeletes).toEqual([])
})
})
describe('ordinary revalidation', () => {
test('uses no aggregate deadline or transfer cap for filesystem repair', async () => {
const { redis, acks } = createRedis()
let observedResources: Parameters<RevalidateWorkerDependencies['handleSiteCreateOrUpdate']>[4] extends infer Options
? Options
: never
const dependencies = createDependencies({
fetchSiteRecord: async () => ({ record: reappearedRecord, cid: 'cid' }),
handleSiteCreateOrUpdate: async (_did, _rkey, _record, _cid, options) => {
observedResources = options
},
})
await processRevalidationMessage(MESSAGE_ID, messageFields('storage-miss:index.html'), redis, dependencies)
expect(observedResources?.resources?.deadlineAt).toBeNull()
expect(observedResources?.resources?.transferBudget.maxBytes).toBe(Number.POSITIVE_INFINITY)
observedResources?.resources?.transferBudget.consume(1_000_000_000)
expect(observedResources?.resources?.signal.aborted).toBe(false)
expect(acks).toEqual([MESSAGE_ID])
})
test('acknowledges queued duplicates without repair while a DLQ fence exists', async () => {
const { redis, acks, streamDeletes } = createRedis()
;(redis as RevalidateRedisClient).get = async () => 'dlq-9'
let lookups = 0
const dependencies = createDependencies({
fetchSiteRecord: async () => {
lookups++
return { record: reappearedRecord, cid: 'cid' }
},
})
await processRevalidationMessage(MESSAGE_ID, messageFields('storage-miss:index.html'), redis, dependencies)
expect(lookups).toBe(0)
expect(acks).toEqual([MESSAGE_ID])
expect(streamDeletes).toEqual([MESSAGE_ID])
})
test('leaves transient or invalid typed PDS outcomes pending', async () => {
const { redis, acks, streamDeletes } = createRedis()
const dependencies = createDependencies({
fetchSiteRecordOutcome: async () => ({ kind: 'retryable', error: 'INVALID_RECORD' }),
})
await processRevalidationMessage(MESSAGE_ID, messageFields('storage-miss:index.html'), redis, dependencies)
expect(acks).toEqual([])
expect(streamDeletes).toEqual([])
})
test('ACKs a missing non-delete record without invoking delete cleanup', async () => {
const { redis, acks, streamDeletes } = createRedis()
let deletes = 0
const dependencies = createDependencies({
handleSiteDelete: async () => {
deletes++
},
})
await processRevalidationMessage(MESSAGE_ID, messageFields('storage-miss:index.html'), redis, dependencies)
expect(deletes).toBe(0)
expect(acks).toEqual([MESSAGE_ID])
expect(streamDeletes).toEqual([MESSAGE_ID])
})
test('uses one atomic completion command rather than separate ACK and delete calls', async () => {
const { redis, acks, calls, streamDeletes } = createRedis()
await processRevalidationMessage(MESSAGE_ID, messageFields('storage-miss:index.html'), redis, createDependencies())
expect(acks).toEqual([MESSAGE_ID])
expect(streamDeletes).toEqual([MESSAGE_ID])
expect(calls).toEqual([`eval-complete:${MESSAGE_ID}`])
})
test('leaves an entry pending when atomic completion reports zero ACK', async () => {
const { redis, acks, calls, streamDeletes } = createRedis(-1, [0, 0])
await processRevalidationMessage(MESSAGE_ID, messageFields('storage-miss:index.html'), redis, createDependencies())
expect(acks).toEqual([])
expect(streamDeletes).toEqual([])
expect(calls).toEqual([`eval-complete:${MESSAGE_ID}`])
})
test('propagates a disconnect-ambiguous completion response without claiming completion', async () => {
const { redis, calls, streamDeletes } = createRedis(-1, new Error('connection dropped after script write'))
await expect(
processRevalidationMessage(MESSAGE_ID, messageFields('storage-miss:index.html'), redis, createDependencies()),
).rejects.toThrow('connection dropped')
expect(streamDeletes).toEqual([])
expect(calls).toEqual([`eval-complete:${MESSAGE_ID}`])
})
test('rejects malformed atomic completion replies', async () => {
const { redis } = createRedis()
redis.eval = async () => ['bad']
await expect(
processRevalidationMessage(MESSAGE_ID, messageFields('storage-miss:index.html'), redis, createDependencies()),
).rejects.toThrow('malformed')
})
})
describe('stale revalidation delivery accounting', () => {
const runtimeConfig: RevalidateWorkerRuntimeConfig = {
batchSize: 10,
claimIdleMs: 60_000,
blockMs: 5_000,
blockingGraceMs: 1_000,
socketTimeoutMs: 16_000,
failureBackoffSeconds: 600,
reconnectMinMs: 10,
reconnectMaxMs: 10,
maxAttempts: 2,
revalidationDeadlineMs: 5_000,
transferBudgetBytes: 1_000_000,
retryBackoffBaseMs: 100,
retryBackoffMaxMs: 100,
}
test('uses JUSTID and does not consume deliveries while retry TTL is active', async () => {
const id = '900001-0'
const fake = new FakeClaimRedis(id, messageFields('storage-miss:index.html'))
const dependencies = createDependencies({
fetchSiteRecordOutcome: async () => {
throw new Error('temporary PDS failure')
},
})
await scanStaleRevalidationMessagesForTests(fake as unknown as Redis, runtimeConfig, dependencies)
expect(fake.deliveryCounts.get(id)).toBe(1)
const autoClaimArgs = fake.calls.find((call) => call.command === 'XAUTOCLAIM')?.args ?? []
expect(autoClaimArgs[autoClaimArgs.length - 1]).toBe('JUSTID')
expect(fake.calls.filter((call) => call.command === 'XCLAIM')).toHaveLength(1)
// A scan during backoff may claim ownership, but must not XCLAIM or read
// XPENDING. This models repeated scans by this worker and after a restart.
const pendingReadsBeforeBackoffScan = fake.calls.filter((call) => call.command === 'XPENDING').length
await scanStaleRevalidationMessagesForTests(fake as unknown as Redis, runtimeConfig, dependencies)
expect(fake.deliveryCounts.get(id)).toBe(1)
expect(fake.calls.filter((call) => call.command === 'XCLAIM')).toHaveLength(1)
expect(fake.calls.filter((call) => call.command === 'XPENDING')).toHaveLength(pendingReadsBeforeBackoffScan)
})
test('suppresses site-backoff scans for processing failures and reappeared tombstones', async () => {
const reasons = ['firehose-processing-failed:update', SITE_DELETE_TOMBSTONE_REASON]
for (const [index, reason] of reasons.entries()) {
const id = `90001${index}-0`
const fake = new FakeClaimRedis(id, messageFields(reason))
fake.siteBackoffTtlSeconds = 60
const dependencies = createDependencies({
fetchSiteRecordOutcome: async () => ({ kind: 'present' as const, record: reappearedRecord, cid: 'cid' }),
})
await scanStaleRevalidationMessagesForTests(fake as unknown as Redis, runtimeConfig, dependencies)
expect(fake.deliveryCounts.get(id)).toBe(1)
expect(fake.calls.filter((call) => call.command === 'XCLAIM')).toHaveLength(1)
expect(fake.retryTtlSeconds).toBe(60)
// Fresh scans use a fresh cursor. These calls model repeated scans and a
// worker restart; the suppression key is the durable Redis state.
await scanStaleRevalidationMessagesForTests(fake as unknown as Redis, runtimeConfig, dependencies)
await scanStaleRevalidationMessagesForTests(fake as unknown as Redis, runtimeConfig, dependencies)
expect(fake.deliveryCounts.get(id)).toBe(1)
expect(fake.calls.filter((call) => call.command === 'XCLAIM')).toHaveLength(1)
expect(fake.calls.filter((call) => call.command === 'XPENDING')).toHaveLength(1)
expect(fake.calls.filter((call) => call.command === 'PIPELINE_EXEC')).toHaveLength(3)
}
})
test('persists suppression after blob backoff for tombstone and firehose failures', async () => {
const reasons = [SITE_DELETE_TOMBSTONE_REASON, 'firehose-processing-failed:update']
for (const [index, reason] of reasons.entries()) {
const id = `90002${index}-0`
const fake = new FakeClaimRedis(id, messageFields(reason))
const dependencies = createDependencies({
fetchSiteRecordOutcome: async () => ({ kind: 'present' as const, record: reappearedRecord, cid: 'cid' }),
handleSiteCreateOrUpdate: async () => {
throw new SiteBlobBackoffError(DID, RKEY, Date.now() + 60_000, 1)
},
})
await scanStaleRevalidationMessagesForTests(fake as unknown as Redis, runtimeConfig, dependencies)
expect(fake.deliveryCounts.get(id)).toBe(1)
expect(fake.calls.filter((call) => call.command === 'SET').map((call) => call.args[0])).toEqual([
`revalidate:site:failure-backoff:${DID}:${RKEY}`,
`revalidate:retry:${id}`,
])
expect(fake.retryTtlSeconds).toBeGreaterThan(0)
expect(fake.siteBackoffTtlSeconds).toBe(fake.retryTtlSeconds)
// The persisted per-message key suppresses both the TTL filter and
// XPENDING/XCLAIM accounting until the site backoff expires.
await scanStaleRevalidationMessagesForTests(fake as unknown as Redis, runtimeConfig, dependencies)
await scanStaleRevalidationMessagesForTests(fake as unknown as Redis, runtimeConfig, dependencies)
expect(fake.deliveryCounts.get(id)).toBe(1)
expect(fake.calls.filter((call) => call.command === 'XCLAIM')).toHaveLength(1)
expect(fake.calls.filter((call) => call.command === 'XPENDING')).toHaveLength(1)
}
})
test('uses one bounded TTL pipeline and caps an oversized JUSTID response before XCLAIM', async () => {
const ids = Array.from({ length: 150 }, (_, index) => `${910000 + index}-0`)
const fake = new FakeClaimRedis(ids, messageFields('firehose-processing-failed:update'))
for (const id of ids.slice(0, 50)) fake.retryTtlOverrides.set(id, 120)
const boundedRuntime = { ...runtimeConfig, batchSize: 100 }
await scanStaleRevalidationMessagesForTests(fake as unknown as Redis, boundedRuntime)
const pipelineExecutions = fake.calls.filter((call) => call.command === 'PIPELINE_EXEC')
expect(pipelineExecutions).toHaveLength(1)
expect(pipelineExecutions[0]?.args).toHaveLength(100)
const xclaim = fake.calls.find((call) => call.command === 'XCLAIM')
expect(xclaim?.args.slice(4)).toEqual(ids.slice(50, 100))
expect(fake.deliveryCounts.get(ids[0] ?? '')).toBeUndefined()
expect(fake.deliveryCounts.get(ids[50] ?? '')).toBe(1)
expect(fake.deliveryCounts.get(ids[100] ?? '')).toBeUndefined()
})
test('fails closed on a malformed batched TTL response without XCLAIM', async () => {
const ids = ['920000-0', '920001-0']
const fake = new FakeClaimRedis(ids, messageFields('firehose-processing-failed:update'))
fake.pipelineResultOverride = [[null, -1]]
await expect(scanStaleRevalidationMessagesForTests(fake as unknown as Redis, runtimeConfig)).rejects.toThrow(
'pipeline returned',
)
expect(fake.calls.filter((call) => call.command === 'XCLAIM')).toHaveLength(0)
expect(fake.deliveryCounts.size).toBe(0)
})
test('keeps compatibility with a minimal Redis seam without pipeline', async () => {
const id = '920010-0'
const fake = new FakeClaimRedis(id, messageFields('firehose-processing-failed:update'))
;(fake as unknown as { pipeline?: unknown }).pipeline = undefined
await scanStaleRevalidationMessagesForTests(fake as unknown as Redis, runtimeConfig)
expect(fake.deliveryCounts.get(id)).toBe(1)
expect(fake.calls.filter((call) => call.command === 'XCLAIM')).toHaveLength(1)
})
test('counts only real processing deliveries toward maxAttempts', async () => {
const id = '900002-0'
const fake = new FakeClaimRedis(id, messageFields('storage-miss:index.html'))
const dependencies = createDependencies({
fetchSiteRecordOutcome: async () => {
throw new Error('temporary PDS failure')
},
})
await scanStaleRevalidationMessagesForTests(fake as unknown as Redis, runtimeConfig, dependencies)
await scanStaleRevalidationMessagesForTests(fake as unknown as Redis, runtimeConfig, dependencies)
expect(fake.deliveryCounts.get(id)).toBe(1)
// Let the retry become eligible. The second actual processing attempt is
// the one that reaches the quarantine threshold.
fake.retryTtlSeconds = 0
await scanStaleRevalidationMessagesForTests(fake as unknown as Redis, runtimeConfig, dependencies)
expect(fake.deliveryCounts.get(id)).toBe(2)
expect(fake.quarantineAttempts).toEqual([2])
})
})
describe('drained acknowledged-entry maintenance', () => {
async function maintenance(result: unknown): Promise<{ scripts: string[] }> {
const scripts: string[] = []
await trimDrainedRevalidationEntries({
eval: async (script) => {
scripts.push(script)
return result
},
})
return { scripts }
}
test('does not trim when any consumer group has pending entries or lag', async () => {
const { scripts } = await maintenance(0)
const script = scripts[0] ?? ''
expect(script).toContain('if pending ~= 0 or lag ~= 0 then return 0 end')
expect(script).toContain('if not configuredFound then return 0 end')
expect(script).toContain('XTRIM')
})
test('fails closed when any group omits safety metadata', async () => {
const { scripts } = await maintenance(0)
expect(scripts[0]).toContain('if not name or pending == nil or lag == nil then return 0 end')
})
test('does nothing when the configured group is absent', async () => {
const { scripts } = await maintenance(0)
expect(scripts[0]).toContain('return 0')
})
test('allows atomic cleanup only when every group is drained', async () => {
const { scripts } = await maintenance(4)
expect(scripts[0]).toContain('pending ~= 0 or lag ~= 0')
expect(scripts[0]).toContain("'MAXLEN', '=', 0")
})
})
describe('worker runtime configuration', () => {
test('falls back for malformed, unsafe, and out-of-range values', () => {
const defaults = resolveRevalidateWorkerRuntimeConfig({})
expect(
resolveRevalidateWorkerRuntimeConfig({
WISP_REVALIDATE_BATCH_SIZE: '0',
WISP_REVALIDATE_CLAIM_IDLE_MS: '1',
WISP_REVALIDATE_BLOCK_MS: 'not-a-number',
WISP_REVALIDATE_BLOCKING_GRACE_MS: '999999',
WISP_REVALIDATE_SOCKET_TIMEOUT_MS: '-1',
WISP_REVALIDATE_FAILURE_BACKOFF_SECONDS: '1.5',
}),
).toEqual(defaults)
})
test('uses bounded runtime values to configure Redis blocking and socket timeouts', () => {
const runtimeConfig = resolveRevalidateWorkerRuntimeConfig({
WISP_REVALIDATE_BATCH_SIZE: '50',
WISP_REVALIDATE_CLAIM_IDLE_MS: '120000',
WISP_REVALIDATE_BLOCK_MS: '10000',
WISP_REVALIDATE_BLOCKING_GRACE_MS: '2000',
WISP_REVALIDATE_SOCKET_TIMEOUT_MS: '25000',
WISP_REVALIDATE_FAILURE_BACKOFF_SECONDS: '300',
})
expect(runtimeConfig).toEqual({
batchSize: 50,
claimIdleMs: 120000,
blockMs: 10000,
blockingGraceMs: 2000,
socketTimeoutMs: 25000,
failureBackoffSeconds: 300,
reconnectMinMs: 250,
reconnectMaxMs: 30_000,
})
expect(createRevalidateWorkerRedisOptions(runtimeConfig)).toEqual({
maxRetriesPerRequest: 2,
enableReadyCheck: true,
blockingTimeout: 12000,
blockingTimeoutGrace: 2000,
socketTimeout: 25000,
})
})
})
describe('worker lifecycle', () => {
const runtimeConfig: RevalidateWorkerRuntimeConfig = {
batchSize: 10,
claimIdleMs: 60_000,
blockMs: 5_000,
blockingGraceMs: 1_000,
socketTimeoutMs: 16_000,
failureBackoffSeconds: 600,
reconnectMinMs: 10,
reconnectMaxMs: 10,
}
test('disconnects a blocking XREADGROUP before awaiting the loop', async () => {
const clients: FakeLifecycleRedis[] = []
startRevalidateWorkerForTests(lifecycleFactory(clients), 'redis://worker-test', runtimeConfig)
await flushAsyncWork()
const client = clients[0]
expect(client?.xreadArguments).toHaveLength(1)
expect(client?.options).toEqual(createRevalidateWorkerRedisOptions(runtimeConfig))
await stopRevalidateWorker()
expect(client?.disconnected).toBe(true)
expect(getRevalidateWorkerStateForTests()).toEqual({
running: false,
hasRedisClient: false,
hasLoop: false,
})
})
test('aborts an active handler before disconnect so the pending entry is left untouched', async () => {
const clients: FakeLifecycleRedis[] = []
const handlerStarted = deferred<void>()
const dependencies = createDependencies({
fetchSiteRecord: async (_did, _rkey, resources) => {
handlerStarted.resolve()
await new Promise<void>((resolve) =>
resources?.signal.addEventListener('abort', () => resolve(), { once: true }),
)
throw resources?.signal.reason instanceof Error ? resources.signal.reason : new Error('aborted')
},
})
const factory: RevalidateWorkerRedisFactory = (_redisUrl, redisOptions) => {
const client = new FakeMessageRedis(redisOptions)
clients.push(client)
return client as unknown as Redis
}
startRevalidateWorkerForTests(factory, 'redis://worker-test', runtimeConfig, { dependencies })
await handlerStarted.promise
await stopRevalidateWorker()
expect(clients[0]?.disconnected).toBe(true)
expect(clients[0]?.evalCalls).toBe(0)
expect(getRevalidateWorkerStateForTests()).toEqual({
running: false,
hasRedisClient: false,
hasLoop: false,
})
})
test('reports an unsafe stop when storage ignores lifecycle cancellation', async () => {
const clients: FakeLifecycleRedis[] = []
const handlerStarted = deferred<void>()
const releaseHandler = deferred<void>()
const dependencies = createDependencies({
fetchSiteRecord: async () => {
handlerStarted.resolve()
await releaseHandler.promise
return null
},
})
const factory: RevalidateWorkerRedisFactory = (_redisUrl, redisOptions) => {
const client = new FakeMessageRedis(redisOptions)
clients.push(client)
return client as unknown as Redis
}
startRevalidateWorkerForTests(factory, 'redis://worker-test', runtimeConfig, { dependencies })
await handlerStarted.promise
await expect(stopRevalidateWorker({ gracePeriodMs: 0 })).resolves.toEqual({ stopped: false, forced: true })
expect(clients[0]?.disconnected).toBe(true)
expect(getRevalidateWorkerStateForTests().hasLoop).toBe(true)
// Release the fake handler so the detached loop can settle before the test
// ends. A real caller must exit nonzero instead of releasing authority.
releaseHandler.resolve()
await stopRevalidateWorker()
expect(getRevalidateWorkerStateForTests()).toMatchObject({ running: false, hasLoop: false })
})
test('supervises initial group failure and reconnects while reporting no live Redis client', async () => {
const clients: FakeLifecycleRedis[] = []
const failedFactory = lifecycleFactory(clients, { xgroupFailure: new Error('group unavailable') })
startRevalidateWorkerForTests(failedFactory, 'redis://worker-test', runtimeConfig)
await new Promise((resolve) => setTimeout(resolve, 30))
expect(clients.length).toBeGreaterThan(1)
expect(clients[0]?.disconnected).toBe(true)
expect(getRevalidateWorkerStateForTests().running).toBe(true)
})
test('cancels retry backoff promptly when stop is requested', async () => {
const clients: FakeLifecycleRedis[] = []
startRevalidateWorkerForTests(
lifecycleFactory(clients, { xautoclaimFailure: new Error('temporary Redis error') }),
'redis://worker-test',
runtimeConfig,
)
await flushAsyncWork()
expect(clients[0]?.claimCalls).toBe(1)
let stopped = false
const stopping = stopRevalidateWorker().then(() => {
stopped = true
})
await flushAsyncWork()
expect(stopped).toBe(true)
await stopping
})
test('resets reconnect backoff after a healthy read iteration', async () => {
const clients: FakeLifecycleRedis[] = []
const delays: number[] = []
const releaseWaiters: Array<() => void> = []
let resolveSecondDelay: (() => void) | undefined
const secondDelaySeen = new Promise<void>((resolve) => {
resolveSecondDelay = resolve
})
const reconnectRuntime: RevalidateWorkerRuntimeConfig = {
...runtimeConfig,
reconnectMinMs: 100,
reconnectMaxMs: 500,
}
const factory: RevalidateWorkerRedisFactory = (_redisUrl, redisOptions) => {
const client =
clients.length === 0
? new FakeLifecycleRedis(redisOptions, undefined, new Error('initial connection failed'))
: new FakeHealthyThenFailureRedis(redisOptions)
clients.push(client)
return client as unknown as Redis
}
const waitForReconnect = (delayMs: number) =>
new Promise<void>((resolve) => {
delays.push(delayMs)
releaseWaiters.push(resolve)
if (delays.length === 2) resolveSecondDelay?.()
})
startRevalidateWorkerForTests(factory, 'redis://worker-test', reconnectRuntime, {
random: () => 0.5,
waitForReconnect,
})
await flushAsyncWork()
expect(delays).toEqual([50])
releaseWaiters.shift()?.()
await secondDelaySeen
expect(clients).toHaveLength(2)
// Without the healthy-read reset this second failure would use attempt=1
// and produce a 100ms delay; it should be the first 50ms delay again.
expect(delays).toEqual([50, 50])
releaseWaiters.shift()?.()
await stopRevalidateWorker()
})
})