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.
54 kB · 1494 lines
TypeScript
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495import { afterAll, afterEach, beforeEach, describe, expect, test } from 'bun:test'import { readFile, rm, writeFile } from 'node:fs/promises'import { tmpdir } from 'node:os'import { join } from 'node:path'import { Effect } from 'effect'import type Redis from 'ioredis'import type { CacheInvalidationDependenciesForTests, CacheInvalidationGapRecoveryDependencies, CacheInvalidationMessage,} from './cache-invalidation'import { invalidationIo } from './cache-invalidation-runtime'
const DID = 'did:plc:test'const RKEY = 'site'const cursorFile = join(tmpdir(), `wisp-cache-invalidation-${process.pid}.lastid`)const originalCursorFile = process.env.WISP_CACHE_INVALIDATION_CURSOR_FILEconst originalRedisUrl = process.env.REDIS_URL
const hotPrefixes: string[] = []const warmPrefixes: string[] = []const cacheDeletes: Array<[string, string]> = []const cachePrefixDeletes: Array<[string, string]> = []const htmlWarmupResets: Array<[string, string]> = []const redisClients: FakeRedis[] = []let hotFailuresRemaining = 0let xinfoFirstEntryId: string | null = nulllet xinfoResponse: unknown | null = nulllet xinfoError: Error | null = nulllet notifyReplayRead: (() => void) | null = null
type ReplayResponse = Array<[string, Array<[string, string[]]>]> | null
class FakeRedis { readonly xreadArguments: unknown[][] = [] readonly options: Record<string, unknown> | undefined status = 'ready' pingCount = 0 pingError: Error | null = null deferredPing: Promise<string> | undefined disconnected = false private readonly listeners = new Map<string, Array<(...args: unknown[]) => void>>() private readonly pendingReads: Array<{ resolve(response: ReplayResponse): void; reject(reason?: unknown): void }> = []
constructor(...args: unknown[]) { this.options = args[1] as Record<string, unknown> | undefined redisClients.push(this) }
on(event: string, listener: (...args: unknown[]) => void): this { const listeners = this.listeners.get(event) ?? [] listeners.push(listener) this.listeners.set(event, listeners) return this }
emit(event: string, ...args: unknown[]): void { for (const listener of this.listeners.get(event) ?? []) { listener(...args) } }
deferredSubscribeAck = false subscribeError: Error | null = null subscribeCalls: string[] = [] private subscribeCallback: ((error: Error | null) => void) | null = null
subscribe(channel: string, callback: (error: Error | null) => void): void { this.subscribeCalls.push(channel) if (this.deferredSubscribeAck) { this.subscribeCallback = callback return } callback(this.subscribeError) }
ackSubscribe(): void { this.subscribeCallback?.(this.subscribeError) this.subscribeCallback = null }
async ping(): Promise<string> { this.pingCount += 1 if (this.pingError) throw this.pingError return this.deferredPing ?? 'PONG' }
async xinfo(..._args: unknown[]): Promise<unknown> { if (xinfoError) throw xinfoError if (xinfoResponse !== null) return xinfoResponse return xinfoFirstEntryId ? ['first-entry', [xinfoFirstEntryId, []]] : [] }
xread(...args: unknown[]): Promise<ReplayResponse> { this.xreadArguments.push(args) // A real ioredis client rejects commands after 'end'; mirror that so a // replay loop reaching XREAD after disconnect cannot hang on a promise // that never settles. if (this.disconnected) { return Promise.reject(new Error('Connection is closed.')) } notifyReplayRead?.() notifyReplayRead = null const { promise, resolve, reject } = Promise.withResolvers<ReplayResponse>() this.pendingReads.push({ resolve, reject }) return promise }
deliver(response: ReplayResponse): void { const pending = this.pendingReads.shift() if (!pending) throw new Error('No pending replay read') pending.resolve(response) }
disconnect(): void { this.disconnected = true const error = new Error('Redis disconnected') for (const { reject } of this.pendingReads) { reject(error) } this.pendingReads.length = 0 }}
/** * Captures heartbeat interval requests instead of running real timers. Each * scheduled entry stays fireable until its cancel handle is called, mirroring a * real clearInterval without needing fake time. */class FakeHeartbeatScheduler { readonly requestedIntervalMs: number[] = [] cancelledCount = 0 private readonly entries: Array<{ intervalMs: number; cancelled: boolean; tick: () => void }> = []
schedule = (callback: () => void, intervalMs: number): (() => void) => { const entry = { intervalMs, cancelled: false, tick: callback } this.entries.push(entry) this.requestedIntervalMs.push(intervalMs) return () => { entry.cancelled = true this.cancelledCount += 1 } }
get activeCount(): number { return this.entries.filter((entry) => !entry.cancelled).length }
fireActiveTicks(): void { for (const entry of [...this.entries]) { if (!entry.cancelled) entry.tick() } }}
const fakeHotTier = { async deletePrefix(prefix: string): Promise<number> { hotPrefixes.push(prefix) if (hotFailuresRemaining > 0) { hotFailuresRemaining -= 1 throw new Error('hot tier unavailable') } return 3 },}
const fakeWarmTier = { async deletePrefix(prefix: string): Promise<number> { warmPrefixes.push(prefix) return 2 },}
const fakeCache = { delete(namespace: string, key: string): void { cacheDeletes.push([namespace, key]) }, deletePrefix(namespace: string, prefix: string): void { cachePrefixDeletes.push([namespace, prefix]) },}
// Keep replay cursor test writes out of the repository while exercising the// real persistence queue.process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE = cursorFile
const { applyCacheInvalidationForTests, clearSiteUpdating, compareStreamIds, getCacheInvalidationHealthSnapshot, getLastProcessedStreamIdForTests, getUpdatingSiteCountForTests, isSiteUpdating, markSiteUpdating, parseCacheInvalidationMessage, parseCacheInvalidationStreamEntry, resetCacheInvalidationReplayForTests, stopCacheInvalidationSubscriber, resetUpdatingSitesForTests, startCacheInvalidationSubscriberForTests, getCacheInvalidationQueueSizeForTests, isCacheInvalidationOverloaded,} = await import('./cache-invalidation')
const testDependencies: CacheInvalidationDependenciesForTests = { hotTier: fakeHotTier as unknown as CacheInvalidationDependenciesForTests['hotTier'], warmTier: fakeWarmTier as unknown as CacheInvalidationDependenciesForTests['warmTier'], cache: fakeCache as unknown as CacheInvalidationDependenciesForTests['cache'], resetSiteHtmlHotCacheWarmup(did: string, rkey: string): void { htmlWarmupResets.push([did, rkey]) },}
function applyForTest(parsed: CacheInvalidationMessage, source: 'pubsub' | 'replay' = 'pubsub'): Promise<void> { return applyCacheInvalidationForTests(parsed, source, testDependencies)}
function siteMessage( action: Extract<CacheInvalidationMessage['action'], 'updating' | 'update' | 'delete' | 'settings'>, overrides: Partial<CacheInvalidationMessage> = {},): CacheInvalidationMessage { return { did: DID, rkey: RKEY, action, ...overrides }}
async function flushAsyncWork(): Promise<void> { await new Promise<void>((resolve) => setImmediate(resolve))}
function waitForReplayRead(): Promise<void> { return new Promise((resolve) => { notifyReplayRead = resolve })}
beforeEach(async () => { await resetCacheInvalidationReplayForTests() await rm(cursorFile, { force: true }) resetUpdatingSitesForTests() hotPrefixes.length = 0 warmPrefixes.length = 0 cacheDeletes.length = 0 cachePrefixDeletes.length = 0 htmlWarmupResets.length = 0 redisClients.length = 0 hotFailuresRemaining = 0 xinfoFirstEntryId = null xinfoResponse = null xinfoError = null notifyReplayRead = null})
afterEach(async () => { await stopCacheInvalidationSubscriber() for (const client of redisClients) { client.disconnect() } redisClients.length = 0})
afterAll(async () => { await stopCacheInvalidationSubscriber() for (const client of redisClients) { client.disconnect() } redisClients.length = 0 await resetCacheInvalidationReplayForTests() await rm(cursorFile, { force: true }) if (originalCursorFile === undefined) { delete process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE } else { process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE = originalCursorFile } if (originalRedisUrl === undefined) { delete process.env.REDIS_URL } else { process.env.REDIS_URL = originalRedisUrl }})
describe('cache invalidation updating state', () => { test('stale token cannot clear a newer update', () => { markSiteUpdating(DID, RKEY, 'token-a') markSiteUpdating(DID, RKEY, 'token-b')
expect(clearSiteUpdating(DID, RKEY, 'token-a')).toBe(false) expect(isSiteUpdating(DID, RKEY)).toBe(true) })
test('matching token clears the active update', () => { markSiteUpdating(DID, RKEY, 'token-a')
expect(clearSiteUpdating(DID, RKEY, 'token-a')).toBe(true) expect(isSiteUpdating(DID, RKEY)).toBe(false) })
test('unversioned clear remains backward compatible', () => { markSiteUpdating(DID, RKEY, 'token-a')
expect(clearSiteUpdating(DID, RKEY)).toBe(true) expect(isSiteUpdating(DID, RKEY)).toBe(false) })
test('a newer terminal stream clears an older marker despite a different token', () => { markSiteUpdating(DID, RKEY, 'token-a', '1-0')
expect(clearSiteUpdating(DID, RKEY, 'token-b', '2-0')).toBe(true) expect(isSiteUpdating(DID, RKEY)).toBe(false) })
test('an older terminal stream cannot clear a newer marker despite a different token', () => { markSiteUpdating(DID, RKEY, 'token-b', '2-0')
expect(clearSiteUpdating(DID, RKEY, 'token-a', '1-0')).toBe(false) expect(isSiteUpdating(DID, RKEY)).toBe(true) })
test('legacy unversioned token mismatch cannot clear a marker', () => { markSiteUpdating(DID, RKEY, 'token-a')
expect(clearSiteUpdating(DID, RKEY, 'token-b')).toBe(false) expect(isSiteUpdating(DID, RKEY)).toBe(true) })
test('legacy token matching remains compatible when only the start has a stream id', () => { markSiteUpdating(DID, RKEY, 'token-a', '2-0')
expect(clearSiteUpdating(DID, RKEY, 'token-a')).toBe(true) expect(isSiteUpdating(DID, RKEY)).toBe(false) })
test('tokenized final update invalidates when the updating marker is absent', async () => { await applyForTest(siteMessage('update', { token: 'token-a' }))
expect(hotPrefixes).toEqual([`${DID}/${RKEY}/`]) expect(warmPrefixes).toEqual([`${DID}/${RKEY}/`]) expect(isSiteUpdating(DID, RKEY)).toBe(false) })
test('tokenized final update clears a matching marker and invalidates', async () => { markSiteUpdating(DID, RKEY, 'token-a')
await applyForTest(siteMessage('update', { token: 'token-a' }))
expect(isSiteUpdating(DID, RKEY)).toBe(false) expect(hotPrefixes).toEqual([`${DID}/${RKEY}/`]) expect(warmPrefixes).toEqual([`${DID}/${RKEY}/`]) })
test('tokenized final update keeps a newer marker and still invalidates', async () => { markSiteUpdating(DID, RKEY, 'token-b')
await applyForTest(siteMessage('update', { token: 'token-a' }))
expect(isSiteUpdating(DID, RKEY)).toBe(true) expect(clearSiteUpdating(DID, RKEY, 'token-b')).toBe(true) expect(hotPrefixes).toEqual([`${DID}/${RKEY}/`]) expect(warmPrefixes).toEqual([`${DID}/${RKEY}/`]) })
test('marking a new site prunes expired updating entries that were never requested', () => { const originalNow = Date.now let now = 1_000_000 Date.now = () => now
try { markSiteUpdating(DID, 'expired-a') markSiteUpdating(DID, 'expired-b')
now += 10 * 60 * 1000 + 1 markSiteUpdating(DID, 'active')
expect(getUpdatingSiteCountForTests()).toBe(1) expect(isSiteUpdating(DID, 'active')).toBe(true) expect(isSiteUpdating(DID, 'expired-a')).toBe(false) expect(isSiteUpdating(DID, 'expired-b')).toBe(false) } finally { Date.now = originalNow } })
test('updating state is capped even when entries have not expired', () => { for (let i = 0; i < 10_005; i++) { markSiteUpdating(DID, `site-${i}`) }
expect(getUpdatingSiteCountForTests()).toBe(10_000) expect(isSiteUpdating(DID, 'site-0')).toBe(false) expect(isSiteUpdating(DID, 'site-5')).toBe(true) expect(isSiteUpdating(DID, 'site-10004')).toBe(true) })})
describe('cache invalidation actions', () => { test('settings only clears settings-dependent memory caches', async () => { markSiteUpdating(DID, RKEY, 'active-update')
await applyForTest(siteMessage('settings'))
expect(hotPrefixes).toEqual([]) expect(warmPrefixes).toEqual([]) expect(cacheDeletes).toEqual([ ['redirectRules', `${DID}:${RKEY}`], ['settings', `${DID}:${RKEY}`], ]) expect(cachePrefixDeletes).toEqual([ ['siteFiles', `${DID}:${RKEY}:`], ['sourceCidMismatches', `${DID}:${RKEY}:`], ]) expect(htmlWarmupResets).toEqual([]) expect(isSiteUpdating(DID, RKEY)).toBe(true) })
test('site update purges file tiers and related memory caches', async () => { await applyForTest(siteMessage('update'))
expect(hotPrefixes).toEqual([`${DID}/${RKEY}/`]) expect(warmPrefixes).toEqual([`${DID}/${RKEY}/`]) expect(cacheDeletes).toEqual([ ['redirectRules', `${DID}:${RKEY}`], ['settings', `${DID}:${RKEY}`], ['siteCache', `${DID}:${RKEY}`], ]) expect(cachePrefixDeletes).toEqual([ ['siteFiles', `${DID}:${RKEY}:`], ['sourceCidMismatches', `${DID}:${RKEY}:`], ]) expect(htmlWarmupResets).toEqual([[DID, RKEY]]) })
test('site delete purges file tiers and related memory caches', async () => { await applyForTest(siteMessage('delete'))
expect(hotPrefixes).toEqual([`${DID}/${RKEY}/`]) expect(warmPrefixes).toEqual([`${DID}/${RKEY}/`]) expect(cacheDeletes).toEqual([ ['redirectRules', `${DID}:${RKEY}`], ['settings', `${DID}:${RKEY}`], ['siteCache', `${DID}:${RKEY}`], ]) expect(cachePrefixDeletes).toEqual([ ['siteFiles', `${DID}:${RKEY}:`], ['sourceCidMismatches', `${DID}:${RKEY}:`], ]) expect(htmlWarmupResets).toEqual([[DID, RKEY]]) })
test('uses the TieredStorage upper-cache invalidation fence when no tiers are injected', async () => { const upperCachePrefixes: string[] = [] const productionStyleDependencies: CacheInvalidationDependenciesForTests = { storage: { async invalidateUpperCaches(prefix: string) { upperCachePrefixes.push(prefix) return { hotDeleted: 3, warmDeleted: 2, failures: [] } }, }, cache: fakeCache, resetSiteHtmlHotCacheWarmup(did: string, rkey: string): void { htmlWarmupResets.push([did, rkey]) }, }
await applyCacheInvalidationForTests(siteMessage('update'), 'pubsub', productionStyleDependencies)
expect(upperCachePrefixes).toEqual([`${DID}/${RKEY}/`]) expect(hotPrefixes).toEqual([]) expect(warmPrefixes).toEqual([]) expect(htmlWarmupResets).toEqual([[DID, RKEY]]) })})
describe('cache invalidation replay', () => { test('a higher live stream id cannot skip an earlier replay event for another site', async () => { const liveTarget = { did: DID, rkey: 'live-site', action: 'update' as const, streamId: '2-0' } const earlierReplayTarget = { did: DID, rkey: 'replay-site', action: 'update' as const, streamId: '1-0' }
await applyForTest(liveTarget, 'pubsub') expect(getLastProcessedStreamIdForTests()).toBe('0-0')
await applyForTest(earlierReplayTarget, 'replay') expect(getLastProcessedStreamIdForTests()).toBe('1-0')
await applyForTest(liveTarget, 'replay') expect(getLastProcessedStreamIdForTests()).toBe('2-0') expect(hotPrefixes).toEqual([`${DID}/live-site/`, `${DID}/replay-site/`, `${DID}/live-site/`]) })
test('a live terminal update suppresses an older replayed updating marker', async () => { const terminal = siteMessage('update', { token: 'token-a', streamId: '2-0' }) const olderUpdating = siteMessage('updating', { token: 'token-a', streamId: '1-0' })
await applyForTest(terminal, 'pubsub') await applyForTest(olderUpdating, 'replay')
expect(isSiteUpdating(DID, RKEY)).toBe(false) expect(getLastProcessedStreamIdForTests()).toBe('1-0') expect(hotPrefixes).toEqual([`${DID}/${RKEY}/`])
// Replay still reapplies the terminal event idempotently and advances only in order. await applyForTest(terminal, 'replay') expect(getLastProcessedStreamIdForTests()).toBe('2-0') })
test('an older tokenless delete retains a newer live updating marker but still purges stale caches', async () => { await applyForTest(siteMessage('updating', { token: 'token-new', streamId: '2-0' }), 'pubsub') await applyForTest(siteMessage('delete', { streamId: '1-0' }), 'replay')
expect(isSiteUpdating(DID, RKEY)).toBe(true) expect(getLastProcessedStreamIdForTests()).toBe('1-0') expect(hotPrefixes).toEqual([`${DID}/${RKEY}/`]) expect(warmPrefixes).toEqual([`${DID}/${RKEY}/`]) })
test('settings and terminal invalidations remain independent across out-of-order delivery', async () => { await applyForTest(siteMessage('settings', { streamId: '2-0' }), 'pubsub') await applyForTest(siteMessage('delete', { streamId: '1-0' }), 'replay')
// A newer settings event cannot suppress an older terminal file purge. expect(hotPrefixes).toEqual([`${DID}/${RKEY}/`]) expect(warmPrefixes).toEqual([`${DID}/${RKEY}/`]) expect(getLastProcessedStreamIdForTests()).toBe('1-0')
const otherTerminal = { did: DID, rkey: 'mixed-site', action: 'update' as const, streamId: '3-0' } const olderSettings = { did: DID, rkey: 'mixed-site', action: 'settings' as const, streamId: '2-0' } await applyForTest(otherTerminal, 'pubsub') cacheDeletes.length = 0 cachePrefixDeletes.length = 0 await applyForTest(olderSettings, 'replay')
// Settings cleanup is also safe after a terminal event for the same site. expect(cacheDeletes).toEqual([ ['redirectRules', `${DID}:mixed-site`], ['settings', `${DID}:mixed-site`], ]) expect(cachePrefixDeletes).toEqual([ ['siteFiles', `${DID}:mixed-site:`], ['sourceCidMismatches', `${DID}:mixed-site:`], ]) expect(getLastProcessedStreamIdForTests()).toBe('2-0') })
test('tier failure does not advance the replay cursor and retries the same event', async () => { hotFailuresRemaining = 1 const event = siteMessage('update', { streamId: '3-0' })
await expect(applyForTest(event, 'replay')).rejects.toThrow('Failed to invalidate file caches') expect(getLastProcessedStreamIdForTests()).toBe('0-0') expect(hotPrefixes).toEqual([`${DID}/${RKEY}/`]) // Warm cleanup still ran despite the hot-tier failure. expect(warmPrefixes).toEqual([`${DID}/${RKEY}/`])
await applyForTest(event, 'replay') expect(getLastProcessedStreamIdForTests()).toBe('3-0') expect(hotPrefixes).toEqual([`${DID}/${RKEY}/`, `${DID}/${RKEY}/`]) expect(warmPrefixes).toEqual([`${DID}/${RKEY}/`, `${DID}/${RKEY}/`]) })
test('TieredStorage upper-cache failures also retain the replay cursor for retry', async () => { const failure = new Error('hot tier unavailable') let attempts = 0 const dependencies: CacheInvalidationDependenciesForTests = { storage: { async invalidateUpperCaches() { attempts++ return attempts === 1 ? { hotDeleted: 0, warmDeleted: 2, failures: [{ tier: 'hot' as const, reason: failure }] } : { hotDeleted: 3, warmDeleted: 2, failures: [] } }, }, cache: fakeCache, resetSiteHtmlHotCacheWarmup(did: string, rkey: string): void { htmlWarmupResets.push([did, rkey]) }, } const event = siteMessage('update', { streamId: '3-0' })
await expect(applyCacheInvalidationForTests(event, 'replay', dependencies)).rejects.toThrow( 'Failed to invalidate file caches', ) expect(getLastProcessedStreamIdForTests()).toBe('0-0')
await applyCacheInvalidationForTests(event, 'replay', dependencies) expect(attempts).toBe(2) expect(getLastProcessedStreamIdForTests()).toBe('3-0') })
test('tier failure retains a matching updating marker until the retry succeeds', async () => { markSiteUpdating(DID, RKEY, 'token-a', '3-0') hotFailuresRemaining = 1 const event = siteMessage('update', { token: 'token-a', streamId: '3-0' })
await expect(applyForTest(event, 'replay')).rejects.toThrow('Failed to invalidate file caches') expect(isSiteUpdating(DID, RKEY)).toBe(true) expect(getLastProcessedStreamIdForTests()).toBe('0-0')
await applyForTest(event, 'replay') expect(isSiteUpdating(DID, RKEY)).toBe(false) expect(getLastProcessedStreamIdForTests()).toBe('3-0') })})
describe('cache invalidation message parsing', () => { test('message parsing preserves token', () => { expect( parseCacheInvalidationMessage(JSON.stringify({ did: DID, rkey: RKEY, action: 'update', token: 'token-a' })), ).toEqual({ did: DID, rkey: RKEY, action: 'update', token: 'token-a', }) })
test('message parsing preserves stream id', () => { expect( parseCacheInvalidationMessage( JSON.stringify({ did: DID, rkey: RKEY, action: 'update', token: 'token-a', streamId: '1713811200000-2' }), ), ).toEqual({ did: DID, rkey: RKEY, action: 'update', token: 'token-a', streamId: '1713811200000-2', }) })
test('stream entry parsing reconstructs invalidation messages', () => { expect( parseCacheInvalidationStreamEntry('1713811200000-5', [ 'did', DID, 'rkey', RKEY, 'action', 'updating', 'token', 'token-a', 'ts', '1713811200000', ]), ).toEqual({ did: DID, rkey: RKEY, action: 'updating', token: 'token-a', streamId: '1713811200000-5', }) })
test('domain invalidation message parsing preserves domain keys', () => { expect( parseCacheInvalidationMessage( JSON.stringify({ action: 'domain', domain: 'example.wisp.place', domainKind: 'wisp', streamId: '1713811200000-6', }), ), ).toEqual({ action: 'domain', domain: 'example.wisp.place', domainKind: 'wisp', streamId: '1713811200000-6', }) })
test('domain stream entry parsing reconstructs domain invalidation messages', () => { expect( parseCacheInvalidationStreamEntry('1713811200000-7', [ 'action', 'domain', 'domain', 'example.com', 'domainKind', 'custom', 'customDomainId', 'abc123', 'ts', '1713811200000', ]), ).toEqual({ action: 'domain', domain: 'example.com', domainKind: 'custom', customDomainId: 'abc123', streamId: '1713811200000-7', }) })
test('stream ids sort by timestamp and sequence', () => { expect(compareStreamIds('1713811200000-1', '1713811200000-2')).toBeLessThan(0) expect(compareStreamIds('1713811200001-0', '1713811200000-999')).toBeGreaterThan(0) expect(compareStreamIds('1713811200001-3', '1713811200001-3')).toBe(0) })})
describe('cache invalidation replay health', () => { test('stopping disconnects a blocking replay read promptly', async () => { process.env.REDIS_URL = 'redis://cache-invalidation-test' await startCacheInvalidationSubscriberForTests( (redisUrl, options) => new FakeRedis(redisUrl, options) as unknown as Redis, ) await flushAsyncWork()
expect(redisClients).toHaveLength(2) const [subscriber, replay] = redisClients expect(replay?.xreadArguments).toHaveLength(1) expect(replay?.options?.blockingTimeout).toEqual(expect.any(Number)) expect(replay?.options?.socketTimeout as number).toBeGreaterThan(replay?.options?.blockingTimeout as number)
await stopCacheInvalidationSubscriber()
expect(subscriber?.disconnected).toBe(true) expect(replay?.disconnected).toBe(true) expect(getCacheInvalidationHealthSnapshot()).toMatchObject({ subscriberConnected: false, replayConnected: false, replayState: 'stopped', retrying: false, }) })
test('recovers a retained-stream gap by clearing local state before replaying from zero', async () => { process.env.REDIS_URL = 'redis://cache-invalidation-test' xinfoFirstEntryId = '2-0' await writeFile(cursorFile, '1-0\n') markSiteUpdating(DID, RKEY, 'stale-update', '1-0') const recoveryOrder: string[] = [] const replayRead = waitForReplayRead() const recovery: CacheInvalidationGapRecoveryDependencies = { clearMutableCaches() { recoveryOrder.push('cache-manager') }, clearUpdatingAndTerminalState() { recoveryOrder.push('site-state') resetUpdatingSitesForTests() }, resetHtmlPrewarmState() { recoveryOrder.push('html-prewarm') }, async evictHotTier() { recoveryOrder.push('hot-tier') }, async validateWarmCacheSourceCids() { recoveryOrder.push('warm-cid-validation') }, }
await startCacheInvalidationSubscriberForTests( (redisUrl, options) => new FakeRedis(redisUrl, options) as unknown as Redis, recovery, ) await replayRead
expect(recoveryOrder).toEqual(['cache-manager', 'site-state', 'html-prewarm', 'hot-tier', 'warm-cid-validation']) expect(isSiteUpdating(DID, RKEY)).toBe(false) expect(getLastProcessedStreamIdForTests()).toBe('0-0') expect(await readFile(cursorFile, 'utf8')).toBe('0-0\n') const replayReadArguments = redisClients[1]?.xreadArguments[0] ?? [] expect(replayReadArguments[replayReadArguments.length - 1]).toBe('0-0')
const health = getCacheInvalidationHealthSnapshot() expect(health.replayState).toBe('starting') expect(health.gapCount).toBe(1) expect(health.lastGapAt).not.toBeNull() expect(health.lastGapRecoveryAt).not.toBeNull() expect(health.lastErrorAt).not.toBeNull()
await stopCacheInvalidationSubscriber() })
test('retries malformed XINFO without clearing state or resetting the cursor', async () => { process.env.REDIS_URL = 'redis://cache-invalidation-test' xinfoResponse = ['first-entry', ['not-a-stream-id', []]] await writeFile(cursorFile, '1-0\n') const recoveryOrder: string[] = []
await startCacheInvalidationSubscriberForTests( (redisUrl, options) => new FakeRedis(redisUrl, options) as unknown as Redis, { clearMutableCaches: () => { recoveryOrder.push('cache-manager') }, }, ) await flushAsyncWork()
expect(recoveryOrder).toEqual([]) expect(getLastProcessedStreamIdForTests()).toBe('1-0') expect(redisClients[1]?.xreadArguments).toEqual([]) expect(getCacheInvalidationHealthSnapshot()).toMatchObject({ replayState: 'degraded', gapCount: 0, lastGapAt: null, lastGapRecoveryAt: null, retrying: true, })
await stopCacheInvalidationSubscriber() })
test('retries an XINFO Redis error without clearing state or resetting the cursor', async () => { process.env.REDIS_URL = 'redis://cache-invalidation-test' xinfoError = new Error('Redis temporarily unavailable') await writeFile(cursorFile, '1-0\n') const recoveryOrder: string[] = []
await startCacheInvalidationSubscriberForTests( (redisUrl, options) => new FakeRedis(redisUrl, options) as unknown as Redis, { clearMutableCaches: () => { recoveryOrder.push('cache-manager') }, }, ) await flushAsyncWork()
expect(recoveryOrder).toEqual([]) expect(getLastProcessedStreamIdForTests()).toBe('1-0') expect(redisClients[1]?.xreadArguments).toEqual([]) expect(getCacheInvalidationHealthSnapshot()).toMatchObject({ replayState: 'degraded', gapCount: 0, lastGapAt: null, lastGapRecoveryAt: null, retrying: true, })
await stopCacheInvalidationSubscriber() })})
describe('cache invalidation subscriber heartbeat', () => { async function startHeartbeatSubscriber(scheduler: FakeHeartbeatScheduler): Promise<FakeRedis> { process.env.REDIS_URL = 'redis://cache-invalidation-test' await startCacheInvalidationSubscriberForTests( (redisUrl, options) => new FakeRedis(redisUrl, options) as unknown as Redis, undefined, { scheduleInterval: scheduler.schedule }, ) // Park the replay loop in its blocking read before any lifecycle changes, // matching the rhythm of the replay-health tests above. await flushAsyncWork() const subscriberClient = redisClients[0] if (!subscriberClient) throw new Error('subscriber client was not created') return subscriberClient }
test('requests the 60 second production keepalive interval', async () => { const scheduler = new FakeHeartbeatScheduler() const subscriberClient = await startHeartbeatSubscriber(scheduler) subscriberClient.emit('ready')
expect(scheduler.requestedIntervalMs).toEqual([60_000]) expect(subscriberClient.pingCount).toBe(0)
await stopCacheInvalidationSubscriber() })
test('pings the subscriber on each heartbeat interval tick while connected', async () => { const scheduler = new FakeHeartbeatScheduler() const subscriberClient = await startHeartbeatSubscriber(scheduler) subscriberClient.emit('ready')
scheduler.fireActiveTicks() await flushAsyncWork() expect(subscriberClient.pingCount).toBe(1) scheduler.fireActiveTicks() await flushAsyncWork() expect(subscriberClient.pingCount).toBe(2) expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(true)
await stopCacheInvalidationSubscriber() })
test('repeated ready events replace the heartbeat interval instead of stacking', async () => { const scheduler = new FakeHeartbeatScheduler() const subscriberClient = await startHeartbeatSubscriber(scheduler) subscriberClient.emit('ready') subscriberClient.emit('ready') subscriberClient.emit('ready')
expect(scheduler.requestedIntervalMs).toEqual([60_000, 60_000, 60_000]) expect(scheduler.cancelledCount).toBe(2) expect(scheduler.activeCount).toBe(1) scheduler.fireActiveTicks() await flushAsyncWork() expect(subscriberClient.pingCount).toBe(1) expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(true)
await stopCacheInvalidationSubscriber() })
test('heartbeat stops on close/end and restarts on reconnect ready', async () => { const scheduler = new FakeHeartbeatScheduler() const subscriberClient = await startHeartbeatSubscriber(scheduler) subscriberClient.emit('ready') scheduler.fireActiveTicks() await flushAsyncWork() scheduler.fireActiveTicks() await flushAsyncWork() expect(subscriberClient.pingCount).toBe(2)
subscriberClient.emit('close') scheduler.fireActiveTicks() await flushAsyncWork() expect(subscriberClient.pingCount).toBe(2) expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(false)
subscriberClient.emit('ready') scheduler.fireActiveTicks() await flushAsyncWork() scheduler.fireActiveTicks() await flushAsyncWork() expect(subscriberClient.pingCount).toBe(4) expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(true)
subscriberClient.emit('end') scheduler.fireActiveTicks() await flushAsyncWork() expect(subscriberClient.pingCount).toBe(4) expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(false)
await stopCacheInvalidationSubscriber() })
test('a failed keepalive ping recreates the wedged subscriber and fences stale clients', async () => { const scheduler = new FakeHeartbeatScheduler() const subscriberClient = await startHeartbeatSubscriber(scheduler) subscriberClient.emit('ready') scheduler.fireActiveTicks() await flushAsyncWork() expect(subscriberClient.pingCount).toBe(1)
subscriberClient.pingError = new Error('subscriber keepalive failed') scheduler.fireActiveTicks() await flushAsyncWork() await flushAsyncWork()
expect(getCacheInvalidationHealthSnapshot().lastErrorAt).not.toBeNull() expect(getCacheInvalidationHealthSnapshot().subscriberRecreations).toBe(1) expect(subscriberClient.disconnected).toBe(true) expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(false)
// The replacement client restores health on ready + SUBSCRIBE ack, and // the heartbeat continues against the new client. const replacement = redisClients[2] if (!replacement) throw new Error('replacement subscriber was not created') replacement.emit('ready') scheduler.fireActiveTicks() await flushAsyncWork() expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(true) expect(replacement.pingCount).toBe(1)
// A stale client's late events are generation-fenced: a close from the // old client cannot flip the new client's health. subscriberClient.emit('close') expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(true)
await stopCacheInvalidationSubscriber() })
test('autoResubscribe is disabled so each ready issues its own acked SUBSCRIBE', async () => { const scheduler = new FakeHeartbeatScheduler() const subscriberClient = await startHeartbeatSubscriber(scheduler) expect(subscriberClient.options?.autoResubscribe).toBe(false)
subscriberClient.emit('ready') subscriberClient.emit('ready') expect(subscriberClient.subscribeCalls).toEqual(['wisp:cache-invalidate', 'wisp:cache-invalidate']) expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(true)
await stopCacheInvalidationSubscriber() })
test('ready alone is not connected until the SUBSCRIBE ack arrives', async () => { const scheduler = new FakeHeartbeatScheduler() const subscriberClient = await startHeartbeatSubscriber(scheduler) subscriberClient.deferredSubscribeAck = true subscriberClient.emit('ready')
expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(false)
subscriberClient.ackSubscribe() expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(true)
await stopCacheInvalidationSubscriber() })
test('a failed SUBSCRIBE ack leaves the subscriber unconnected and records an error', async () => { const scheduler = new FakeHeartbeatScheduler() const subscriberClient = await startHeartbeatSubscriber(scheduler) subscriberClient.subscribeError = new Error('NOPERM denied') subscriberClient.emit('ready')
expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(false) expect(getCacheInvalidationHealthSnapshot().lastErrorAt).not.toBeNull()
await stopCacheInvalidationSubscriber() })
test('supervisor recreates a subscriber that stays ready without a SUBSCRIBE ack', async () => { const scheduler = new FakeHeartbeatScheduler() const supervisorScheduler = new FakeHeartbeatScheduler() process.env.REDIS_URL = 'redis://cache-invalidation-test' await startCacheInvalidationSubscriberForTests( (redisUrl, options) => new FakeRedis(redisUrl, options) as unknown as Redis, undefined, { scheduleInterval: scheduler.schedule }, { scheduleInterval: supervisorScheduler.schedule, recreateAfterMs: 1 }, ) await flushAsyncWork() const subscriberClient = redisClients[0] if (!subscriberClient) throw new Error('subscriber client was not created') subscriberClient.deferredSubscribeAck = true subscriberClient.emit('ready')
// recreateAfterMs=1: the first supervisor tick records the unhealthy // start; a tick after the bound recreates the client. supervisorScheduler.fireActiveTicks() await new Promise((resolve) => setTimeout(resolve, 2)) supervisorScheduler.fireActiveTicks() await flushAsyncWork() expect(getCacheInvalidationHealthSnapshot().subscriberRecreations).toBe(1) expect(subscriberClient.disconnected).toBe(true)
const replacement = redisClients[2] if (!replacement) throw new Error('replacement subscriber was not created') replacement.emit('ready') expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(true)
await stopCacheInvalidationSubscriber() })
test('start-once: a repeated full start is ignored and creates no duplicate clients', async () => { const scheduler = new FakeHeartbeatScheduler() const subscriberClient = await startHeartbeatSubscriber(scheduler) await startCacheInvalidationSubscriberForTests( (redisUrl, options) => new FakeRedis(redisUrl, options) as unknown as Redis, undefined, { scheduleInterval: scheduler.schedule }, ) await flushAsyncWork() expect(redisClients).toHaveLength(2)
subscriberClient.emit('ready') expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(true) await stopCacheInvalidationSubscriber() })
test('stop-once: concurrent stops share one idempotent teardown', async () => { const scheduler = new FakeHeartbeatScheduler() const subscriberClient = await startHeartbeatSubscriber(scheduler) subscriberClient.emit('ready')
const [first, second] = [stopCacheInvalidationSubscriber(), stopCacheInvalidationSubscriber()] await Promise.all([first, second]) await stopCacheInvalidationSubscriber()
expect(subscriberClient.disconnected).toBe(true) expect(redisClients[1]?.disconnected).toBe(true) expect(getCacheInvalidationHealthSnapshot()).toMatchObject({ subscriberConnected: false, replayConnected: false, replayState: 'stopped', }) })
test('stopping the subscriber cancels the heartbeat so no further pings fire', async () => { const scheduler = new FakeHeartbeatScheduler() const subscriberClient = await startHeartbeatSubscriber(scheduler) subscriberClient.emit('ready') scheduler.fireActiveTicks() await flushAsyncWork() expect(subscriberClient.pingCount).toBe(1)
await stopCacheInvalidationSubscriber() scheduler.fireActiveTicks() await flushAsyncWork()
expect(subscriberClient.pingCount).toBe(1) expect(scheduler.activeCount).toBe(0) expect(subscriberClient.disconnected).toBe(true) expect(redisClients[1]?.disconnected).toBe(true) expect(getCacheInvalidationHealthSnapshot()).toMatchObject({ subscriberConnected: false, replayConnected: false, replayState: 'stopped', }) })})
describe('bounded invalidation dispatch', () => { async function start(dependencies?: CacheInvalidationDependenciesForTests) { process.env.REDIS_URL = 'redis://cache-invalidation-test' process.env.WISP_CACHE_INVALIDATION_QUEUE_CAPACITY = '2' await startCacheInvalidationSubscriberForTests( (url, options) => new FakeRedis(url, options) as unknown as Redis, undefined, undefined, undefined, dependencies ?? testDependencies, ) delete process.env.WISP_CACHE_INVALIDATION_QUEUE_CAPACITY await flushAsyncWork() const subscriber = redisClients[redisClients.length - 2]! const replay = redisClients[redisClients.length - 1]! subscriber.emit('ready') return { subscriber, replay } } const send = (subscriber: FakeRedis, message: CacheInvalidationMessage) => subscriber.emit('message', 'wisp:cache-invalidate', JSON.stringify(message)) const entry = (id: string, action = 'settings'): [string, string[]] => [ id, ['action', action, 'did', DID, 'rkey', RKEY], ]
test('durable overflow has a finite queue and recovers only through earlier ordered replay', async () => { const { subscriber, replay } = await start() for (let i = 1; i <= 20; i++) send(subscriber, siteMessage('settings', { streamId: `${i}-0` })) expect(getCacheInvalidationQueueSizeForTests()).toBeLessThanOrEqual(2) expect(getCacheInvalidationHealthSnapshot()).toMatchObject({ replayState: 'degraded', overloadLatched: false }) expect(getLastProcessedStreamIdForTests()).toBe('0-0') const earlierRead = waitForReplayRead() replay.deliver([['wisp:cache-invalidate-stream', [entry('1-0')]]]) await earlierRead expect(getLastProcessedStreamIdForTests()).toBe('1-0') expect(getCacheInvalidationHealthSnapshot().replayState).toBe('degraded') const caughtUp = waitForReplayRead() replay.deliver([['wisp:cache-invalidate-stream', [entry('20-0')]]]) await caughtUp expect(getLastProcessedStreamIdForTests()).toBe('20-0') expect(getCacheInvalidationHealthSnapshot().replayState).toBe('healthy') expect(await readFile(cursorFile, 'utf8')).toBe('20-0\n') })
test('legacy overflow latches fail-closed even after healthy replay', async () => { const { subscriber, replay } = await start() for (let i = 0; i < 20; i++) send(subscriber, siteMessage('settings')) expect(getCacheInvalidationQueueSizeForTests()).toBeLessThanOrEqual(2) expect(isCacheInvalidationOverloaded()).toBe(true) const nextRead = waitForReplayRead() replay.deliver(null) await nextRead expect(getCacheInvalidationHealthSnapshot().replayState).toBe('degraded') expect(isCacheInvalidationOverloaded()).toBe(true) const { default: app } = await import('../server') const response = await app.request('https://sites.wisp.place/did:plc:test/site/') expect(response.status).toBe(503) expect(response.headers.get('cache-control')).toBe('no-store') expect(await response.text()).toBe('Hosting temporarily unavailable') expect((await app.request('https://wisp.place/live')).status).toBe(200) expect((await app.request('https://wisp.place/health')).status).toBe(503) await stopCacheInvalidationSubscriber() await start() expect(isCacheInvalidationOverloaded()).toBe(true) })
test('oversized input is never retained or parsed and latches conservatively', async () => { const { subscriber } = await start() subscriber.emit( 'message', 'wisp:cache-invalidate', JSON.stringify({ action: 'settings', did: DID, rkey: RKEY, streamId: '1-0', extra: 'x'.repeat(1024 * 1024) }), ) expect(getCacheInvalidationQueueSizeForTests()).toBe(0) expect(isCacheInvalidationOverloaded()).toBe(true) expect(getCacheInvalidationHealthSnapshot().overflowCount).toBe(1) })
test('malformed bounded input is discarded without a false legacy-overflow latch', async () => { const { subscriber } = await start() subscriber.emit('message', 'wisp:cache-invalidate', '{bad json') expect(getCacheInvalidationQueueSizeForTests()).toBe(0) expect(isCacheInvalidationOverloaded()).toBe(false) expect(getCacheInvalidationHealthSnapshot().lastErrorAt).not.toBeNull() })
test('partial replay-client acquisition failure disconnects subscriber and cancels timers', async () => { process.env.REDIS_URL = 'redis://cache-invalidation-test' const scheduler = new FakeHeartbeatScheduler() let creations = 0 await expect( startCacheInvalidationSubscriberForTests( (url, options) => { if (++creations === 2) throw new Error('secret connection string') return new FakeRedis(url, options) as unknown as Redis }, undefined, { scheduleInterval: scheduler.schedule }, { scheduleInterval: scheduler.schedule }, ), ).rejects.toBeDefined() expect(redisClients[0]?.disconnected).toBe(true) expect(scheduler.activeCount).toBe(0) expect(getCacheInvalidationHealthSnapshot().replayState).toBe('degraded') })
test('stop waits for uncancelable accepted tier writes before finishing', async () => { const started = Promise.withResolvers<void>() const release = Promise.withResolvers<void>() const dependencies = { ...testDependencies, storage: { async invalidateUpperCaches() { started.resolve() await release.promise return { hotDeleted: 0, warmDeleted: 0, failures: [] } }, }, hotTier: undefined, warmTier: undefined, } delete dependencies.hotTier delete dependencies.warmTier const { subscriber, replay } = await start(dependencies) send(subscriber, siteMessage('update', { streamId: '1-0' })) await started.promise for (let i = 2; i <= 20; i++) send(subscriber, siteMessage('settings', { streamId: `${i}-0` })) expect(getCacheInvalidationQueueSizeForTests()).toBe(2) expect(getCacheInvalidationHealthSnapshot().replayState).toBe('degraded') let stopped = false const closing = stopCacheInvalidationSubscriber().then(() => { stopped = true }) await flushAsyncWork() expect(replay.disconnected).toBe(true) expect(stopped).toBe(false) release.resolve() await closing expect(stopped).toBe(true) })
test('cursor persistence failure retains cursor and terminal marker, then retry persists before clearing', async () => { const original = process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE const impossible = join(cursorFile, 'child') await writeFile(cursorFile, 'not a directory') process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE = impossible markSiteUpdating(DID, RKEY, 'token-a', '1-0') try { await expect( applyForTest(siteMessage('update', { token: 'token-a', streamId: '1-0' }), 'replay'), ).rejects.toBeDefined() expect(getLastProcessedStreamIdForTests()).toBe('0-0') expect(isSiteUpdating(DID, RKEY)).toBe(true) } finally { process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE = original } await applyForTest(siteMessage('update', { token: 'token-a', streamId: '1-0' }), 'replay') expect(getLastProcessedStreamIdForTests()).toBe('1-0') expect(isSiteUpdating(DID, RKEY)).toBe(false) expect(await readFile(cursorFile, 'utf8')).toBe('1-0\n') })
test('stale subscription ack cannot make a replacement subscriber healthy', async () => { const scheduler = new FakeHeartbeatScheduler() process.env.REDIS_URL = 'redis://cache-invalidation-test' await startCacheInvalidationSubscriberForTests( (url, options) => new FakeRedis(url, options) as unknown as Redis, undefined, undefined, { scheduleInterval: scheduler.schedule, recreateAfterMs: 1 }, ) const old = redisClients[0]! old.deferredSubscribeAck = true old.emit('ready') scheduler.fireActiveTicks() await new Promise((resolve) => setTimeout(resolve, 2)) scheduler.fireActiveTicks() expect(old.disconnected).toBe(true) old.ackSubscribe() expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(false) redisClients[2]!.emit('ready') expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(true) })
test('replay storage failure remains degraded until the same entry is retried', async () => { const { subscriber, replay } = await start() markSiteUpdating(DID, RKEY, 'token-a', '1-0') hotFailuresRemaining = 1 for (let i = 1; i <= 5; i++) send(subscriber, siteMessage('settings', { streamId: `${i}-0` })) replay.deliver([ ['wisp:cache-invalidate-stream', [['1-0', ['action', 'update', 'did', DID, 'rkey', RKEY, 'token', 'token-a']]]], ]) await flushAsyncWork() expect(getLastProcessedStreamIdForTests()).toBe('0-0') expect(isSiteUpdating(DID, RKEY)).toBe(true) expect(getCacheInvalidationHealthSnapshot().replayState).toBe('degraded') const retryRead = waitForReplayRead() await retryRead const caughtUp = waitForReplayRead() replay.deliver([ [ 'wisp:cache-invalidate-stream', [['1-0', ['action', 'update', 'did', DID, 'rkey', RKEY, 'token', 'token-a']], entry('5-0')], ], ]) await caughtUp expect(getLastProcessedStreamIdForTests()).toBe('5-0') expect(isSiteUpdating(DID, RKEY)).toBe(false) expect(getCacheInvalidationHealthSnapshot().replayState).toBe('healthy') })
test('restart replays safely from the last persisted cursor, not a later live ID', async () => { const { subscriber, replay } = await start() const nextRead = waitForReplayRead() replay.deliver([['wisp:cache-invalidate-stream', [entry('1-0')]]]) await nextRead send(subscriber, siteMessage('settings', { streamId: '9-0' })) await stopCacheInvalidationSubscriber() const restarted = await start() const args = restarted.replay.xreadArguments[0]! expect(args[args.length - 1]).toBe('1-0') const duplicate = waitForReplayRead() restarted.replay.deliver([['wisp:cache-invalidate-stream', [entry('1-0'), entry('2-0')]]]) await duplicate expect(getLastProcessedStreamIdForTests()).toBe('2-0') expect(await readFile(cursorFile, 'utf8')).toBe('2-0\n') })
test('a cache callback defect latches rather than entering a network retry loop', async () => { const { subscriber } = await start({ ...testDependencies, cache: { delete() { throw new Error('secret defect') }, deletePrefix() {}, }, }) send(subscriber, siteMessage('settings', { streamId: '1-0' })) await flushAsyncWork() expect(isCacheInvalidationOverloaded()).toBe(true) expect(getLastProcessedStreamIdForTests()).toBe('0-0') expect(getCacheInvalidationHealthSnapshot().replayState).toBe('degraded') })
test('an unsettled PING cannot accumulate a command or fiber on each timer tick', async () => { const scheduler = new FakeHeartbeatScheduler() process.env.REDIS_URL = 'redis://cache-invalidation-test' await startCacheInvalidationSubscriberForTests( (url, options) => new FakeRedis(url, options) as unknown as Redis, undefined, { scheduleInterval: scheduler.schedule }, ) const client = redisClients[0]! const pong = Promise.withResolvers<string>() client.deferredPing = pong.promise client.emit('ready') for (let i = 0; i < 100; i++) scheduler.fireActiveTicks() expect(client.pingCount).toBe(1) client.emit('ready') scheduler.fireActiveTicks() expect(client.pingCount).toBe(1) await stopCacheInvalidationSubscriber() pong.resolve('PONG') await flushAsyncWork() expect(getCacheInvalidationHealthSnapshot().subscriberConnected).toBe(false) expect(isCacheInvalidationOverloaded()).toBe(false) })
test('oversized replay fields latch without advancing or retaining an event', async () => { const { replay } = await start() replay.deliver([ [ 'wisp:cache-invalidate-stream', [['1-0', ['action', 'settings', 'did', DID, 'rkey', RKEY, 'extra', 'x'.repeat(1024 * 1024)]]], ], ]) await flushAsyncWork() expect(isCacheInvalidationOverloaded()).toBe(true) expect(getLastProcessedStreamIdForTests()).toBe('0-0') expect(getCacheInvalidationQueueSizeForTests()).toBe(0) })
test('fallback tier deletion batches keys and closes the iterator on failure', async () => { let closed = false const batches: string[][] = [] const tier = { async *listKeys() { try { for (let i = 0; i < 200; i++) yield `key-${i}` } finally { closed = true } }, async deleteMany(keys: string[]) { batches.push(keys) if (batches.length === 2) throw new Error('disk unavailable') }, } await expect( applyCacheInvalidationForTests(siteMessage('update', { streamId: '1-0' }), 'replay', { ...testDependencies, hotTier: tier as unknown as CacheInvalidationDependenciesForTests['hotTier'], }), ).rejects.toThrow('Failed to invalidate file caches') expect(batches.map((batch) => batch.length)).toEqual([64, 64]) expect(closed).toBe(true) expect(warmPrefixes).toEqual([`${DID}/${RKEY}/`]) expect(getLastProcessedStreamIdForTests()).toBe('0-0') })
test('runtime persistence failure is visible in readiness and retains replay cursor/update marker', async () => { const { replay } = await start() const original = process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE await writeFile(cursorFile, 'not a directory') process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE = join(cursorFile, 'child') markSiteUpdating(DID, RKEY, 'token-a', '1-0') const retried = waitForReplayRead() replay.deliver([ ['wisp:cache-invalidate-stream', [['1-0', ['action', 'update', 'did', DID, 'rkey', RKEY, 'token', 'token-a']]]], ]) try { await retried expect(getCacheInvalidationHealthSnapshot().replayState).toBe('degraded') expect(getLastProcessedStreamIdForTests()).toBe('0-0') expect(isSiteUpdating(DID, RKEY)).toBe(true) } finally { process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE = original } const caughtUp = waitForReplayRead() replay.deliver([ ['wisp:cache-invalidate-stream', [['1-0', ['action', 'update', 'did', DID, 'rkey', RKEY, 'token', 'token-a']]]], ]) await caughtUp expect(getCacheInvalidationHealthSnapshot().replayState).toBe('healthy') expect(getLastProcessedStreamIdForTests()).toBe('1-0') expect(isSiteUpdating(DID, RKEY)).toBe(false) })
test('gap recovery is ordered with live work and does not issue file-tier purges', async () => { process.env.REDIS_URL = 'redis://cache-invalidation-test' await writeFile(cursorFile, '1-0\n') xinfoFirstEntryId = '2-0' const began = Promise.withResolvers<void>() const release = Promise.withResolvers<void>() const read = waitForReplayRead() await startCacheInvalidationSubscriberForTests( (url, options) => new FakeRedis(url, options) as unknown as Redis, { clearMutableCaches() {}, clearUpdatingAndTerminalState: resetUpdatingSitesForTests, resetHtmlPrewarmState() {}, async evictHotTier() { began.resolve() await release.promise }, validateWarmCacheSourceCids() {}, }, undefined, undefined, testDependencies, ) await began.promise redisClients[0]!.emit( 'message', 'wisp:cache-invalidate', JSON.stringify(siteMessage('updating', { token: 'new', streamId: '3-0' })), ) await flushAsyncWork() expect(isSiteUpdating(DID, RKEY)).toBe(false) expect(getCacheInvalidationQueueSizeForTests()).toBe(1) release.resolve() await read await flushAsyncWork() expect(isSiteUpdating(DID, RKEY)).toBe(true) expect(hotPrefixes).toEqual([]) expect(warmPrefixes).toEqual([]) })
test('expected IO errors carry no serializable foreign causes or credentials', async () => { const hostile = { message: 'redis://user:secret@private-host', body: { token: 'secret' }, path: '/private/secret' } const result = await Effect.runPromise(Effect.result(invalidationIo('xread', () => Promise.reject(hostile)))) expect(result._tag).toBe('Failure') if (result._tag !== 'Failure') throw new Error('Expected IO failure') expect(JSON.stringify(result.failure)).not.toContain('secret') expect(JSON.stringify(result.failure)).not.toContain('private-host') expect('cause' in result.failure).toBe(false) expect(result.failure.operation).toBe('xread') })
test('missing-stream recovery preserves a safe discriminator, not the Redis error body', async () => { const result = await Effect.runPromise( Effect.result(invalidationIo('xinfo', () => Promise.reject(new Error('ERR no such key secret path')))), ) expect(result._tag).toBe('Failure') if (result._tag !== 'Failure') throw new Error('Expected IO failure') expect(result.failure.missingStream).toBe(true) expect(JSON.stringify(result.failure)).not.toContain('secret') })})