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.
35 kB · 971 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972/** * Cache invalidation subscriber * * Uses Redis pub/sub for low-latency invalidation and a Redis stream for replay. * When a site is updated/deleted, clears the hosting-service's local caches * (tiered storage hot+warm tiers, redirect rules) so stale data isn't served. * * Also tracks sites that are actively being downloaded ('updating' action) so * the serving layer can show a "site updating" page instead of stale/partial content. */
import { existsSync, readFileSync } from 'node:fs'import { mkdir, rename, writeFile } from 'node:fs/promises'import { basename, dirname, resolve } from 'node:path'import { type CacheInvalidationMessage, DEFAULT_CACHE_INVALIDATION_CHANNEL, DEFAULT_CACHE_INVALIDATION_STREAM, decodeCacheInvalidationMessage,} from '@wispplace/constants'import { createLogger } from '@wispplace/observability'import type { StorageTier } from '@wispplace/tiered-storage'import { Effect, Exit, Scope, Stream } from 'effect'import Redis from 'ioredis'import { InvalidationIoError, type InvalidationRuntime, invalidationIo, makeInvalidationRuntime,} from './cache-invalidation-runtime'import { cache } from './cache-manager'import { resetAllHtmlHotCacheWarmups, resetSiteHtmlHotCacheWarmup } from './html-prewarm'import { hotTier, storage, warmTier } from './storage'
const logger = createLogger('cache-invalidation')const CHANNEL = DEFAULT_CACHE_INVALIDATION_CHANNELconst STREAM = process.env.WISP_CACHE_INVALIDATION_STREAM || DEFAULT_CACHE_INVALIDATION_STREAMconst STREAM_BLOCK_MS = parsePositiveInt(process.env.WISP_CACHE_INVALIDATION_BLOCK_MS, 5000)const STREAM_BATCH_COUNT = parsePositiveInt(process.env.WISP_CACHE_INVALIDATION_BATCH_COUNT, 100)const STREAM_BLOCKING_GRACE_MS = parsePositiveInt(process.env.WISP_CACHE_INVALIDATION_BLOCKING_GRACE_MS, 1000)const STREAM_SOCKET_TIMEOUT_MS = Math.max( parsePositiveInt( process.env.WISP_CACHE_INVALIDATION_SOCKET_TIMEOUT_MS, STREAM_BLOCK_MS + STREAM_BLOCKING_GRACE_MS + 10_000, ), STREAM_BLOCK_MS + STREAM_BLOCKING_GRACE_MS + 1000,)const SUBSCRIBER_RECREATE_AFTER_MS = 180_000
function getCursorFile(): string { return ( process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE || resolve(process.env.CACHE_DIR || './cache/sites', '..', 'cache-invalidation.lastid') )}const STREAM_ID_PATTERN = /^\d+-\d+$/
export type { CacheInvalidationMessage } from '@wispplace/constants'
// Sites currently being downloaded by the firehose-service.// Maps `${did}/${rkey}` → current update token and timestamp.// Used to show an "updating" page instead of serving stale files.const UPDATING_TTL_MS = 10 * 60 * 1000 // 10 minutes safety timeoutconst MAX_UPDATING_SITES = parsePositiveInt(process.env.WISP_MAX_UPDATING_SITES, 10_000)type UpdatingSiteState = { since: number; token?: string; streamId?: string }type TerminalSiteState = { since: number; streamId: string }
const updatingSites = new Map<string, UpdatingSiteState>()// This local, bounded high-water mark prevents live terminal events from being// undone by older replayed starts. Replay remains the only durable cursor owner.const latestTerminalSiteStreamIds = new Map<string, TerminalSiteState>()
type InvalidationCache = Pick<typeof cache, 'delete' | 'deletePrefix'>type UpperCacheInvalidator = Pick<typeof storage, 'invalidateUpperCaches'>
export interface CacheInvalidationDependenciesForTests { /** Use the real TieredStorage upper-cache fence in production or inject it in focused tests. */ storage?: UpperCacheInvalidator /** Supplying either tier keeps deterministic legacy injected-tier tests on the direct fallback. */ hotTier?: StorageTier warmTier?: StorageTier cache?: InvalidationCache resetSiteHtmlHotCacheWarmup?: typeof resetSiteHtmlHotCacheWarmup}
interface CacheInvalidationDependencies { storage: UpperCacheInvalidator | undefined hotTier: StorageTier warmTier: StorageTier | undefined cache: InvalidationCache resetSiteHtmlHotCacheWarmup: typeof resetSiteHtmlHotCacheWarmup}
function getCacheInvalidationDependencies( overrides?: CacheInvalidationDependenciesForTests,): CacheInvalidationDependencies { const hasInjectedFileTiers = overrides !== undefined && ('hotTier' in overrides || 'warmTier' in overrides) return { storage: hasInjectedFileTiers ? undefined : (overrides?.storage ?? storage), hotTier: overrides?.hotTier ?? hotTier, warmTier: overrides?.warmTier ?? warmTier, cache: overrides?.cache ?? cache, resetSiteHtmlHotCacheWarmup: overrides?.resetSiteHtmlHotCacheWarmup ?? resetSiteHtmlHotCacheWarmup, }}
/** * Recovery hooks for a Redis stream retention gap. The warm-cache validation * hook is intentionally separate: file serving validates entries against the * current source CID, so recovery must not wipe up to 10GB of warm disk data. */export interface CacheInvalidationGapRecoveryDependencies { clearMutableCaches?: () => void | Promise<void> clearUpdatingAndTerminalState?: () => void | Promise<void> resetHtmlPrewarmState?: () => void | Promise<void> evictHotTier?: () => void | Promise<void> validateWarmCacheSourceCids?: () => void | Promise<void>}
interface ResolvedCacheInvalidationGapRecoveryDependencies { clearMutableCaches: () => void | Promise<void> clearUpdatingAndTerminalState: () => void | Promise<void> resetHtmlPrewarmState: () => void | Promise<void> evictHotTier: () => void | Promise<void> validateWarmCacheSourceCids: () => void | Promise<void>}
function clearUpdatingAndTerminalState(): void { updatingSites.clear() latestTerminalSiteStreamIds.clear()}
function getCacheInvalidationGapRecoveryDependencies( overrides?: CacheInvalidationGapRecoveryDependencies,): ResolvedCacheInvalidationGapRecoveryDependencies { return { clearMutableCaches: overrides?.clearMutableCaches ?? (() => cache.clearAll()), clearUpdatingAndTerminalState: overrides?.clearUpdatingAndTerminalState ?? clearUpdatingAndTerminalState, resetHtmlPrewarmState: overrides?.resetHtmlPrewarmState ?? resetAllHtmlHotCacheWarmups, evictHotTier: overrides?.evictHotTier ?? (() => hotTier.clear()), // File serving validates retained warm entries against the source CID. This // seam permits deployments to perform an additional validation pass before // replay resumes, without an unsafe broad warm-tier wipe. validateWarmCacheSourceCids: overrides?.validateWarmCacheSourceCids ?? (() => undefined), }}
let cacheInvalidationGapRecoveryDependencies = getCacheInvalidationGapRecoveryDependencies()
function getSiteInvalidationKey(did: string, rkey: string): string { return `${did}/${rkey}`}
function pruneExpiredUpdatingSites(now = Date.now()): void { for (const [key, state] of updatingSites) { if (now - state.since > UPDATING_TTL_MS) { updatingSites.delete(key) } }}
function pruneExpiredTerminalSiteStreamIds(now = Date.now()): void { for (const [key, state] of latestTerminalSiteStreamIds) { if (now - state.since > UPDATING_TTL_MS) { latestTerminalSiteStreamIds.delete(key) } }}
function enforceUpdatingSitesLimit(): void { while (updatingSites.size > MAX_UPDATING_SITES) { const oldestKey = updatingSites.keys().next().value if (oldestKey === undefined) return updatingSites.delete(oldestKey) }}
function enforceTerminalSiteStreamIdsLimit(): void { while (latestTerminalSiteStreamIds.size > MAX_UPDATING_SITES) { const oldestKey = latestTerminalSiteStreamIds.keys().next().value if (oldestKey === undefined) return latestTerminalSiteStreamIds.delete(oldestKey) }}
function recordTerminalSiteStreamId(did: string, rkey: string, streamId: string | undefined): void { const normalized = normalizeStreamId(streamId) if (!normalized) return
const now = Date.now() pruneExpiredTerminalSiteStreamIds(now) const key = getSiteInvalidationKey(did, rkey) const previous = latestTerminalSiteStreamIds.get(key) if (previous && compareStreamIds(normalized, previous.streamId) < 0) return
latestTerminalSiteStreamIds.delete(key) latestTerminalSiteStreamIds.set(key, { since: now, streamId: normalized }) enforceTerminalSiteStreamIdsLimit()}
function isUpdatingSupersededByTerminal(did: string, rkey: string, streamId: string | undefined): boolean { const normalized = normalizeStreamId(streamId) if (!normalized) return false
pruneExpiredTerminalSiteStreamIds() const terminal = latestTerminalSiteStreamIds.get(getSiteInvalidationKey(did, rkey)) return terminal !== undefined && compareStreamIds(terminal.streamId, normalized) >= 0}
export function isSiteUpdating(did: string, rkey: string): boolean { const key = getSiteInvalidationKey(did, rkey) const state = updatingSites.get(key) if (state === undefined) return false if (Date.now() - state.since > UPDATING_TTL_MS) { // Firehose must have crashed; remove the stale entry updatingSites.delete(key) return false } return true}
export function markSiteUpdating(did: string, rkey: string, token?: string, streamId?: string): void { const now = Date.now() const key = getSiteInvalidationKey(did, rkey) pruneExpiredUpdatingSites(now) // Refresh insertion order so the cap evicts the oldest active update. updatingSites.delete(key) updatingSites.set(key, { since: now, token, streamId: normalizeStreamId(streamId) }) enforceUpdatingSitesLimit()}
function canClearSiteUpdating(did: string, rkey: string, token?: string, streamId?: string): boolean { const state = updatingSites.get(getSiteInvalidationKey(did, rkey)) if (!state) return false
const terminalStreamId = normalizeStreamId(streamId) if (state.streamId && terminalStreamId) { // Comparable stream IDs are authoritative. A later terminal event wins even // if its updating hint was missed or carried a different token. return compareStreamIds(terminalStreamId, state.streamId) >= 0 }
// Without comparable stream IDs, retain the legacy token guard. This keeps // tokenized producers interoperable when only one side carries a stream ID. return !(token && state.token && state.token !== token)}
export function clearSiteUpdating(did: string, rkey: string, token?: string, streamId?: string): boolean { if (!canClearSiteUpdating(did, rkey, token, streamId)) return false updatingSites.delete(getSiteInvalidationKey(did, rkey)) return true}
export function resetUpdatingSitesForTests(): void { clearUpdatingAndTerminalState()}
export function getUpdatingSiteCountForTests(): number { return updatingSites.size}
export interface CacheInvalidationHealthSnapshot { subscriberConnected: boolean replayConnected: boolean replayState: 'stopped' | 'starting' | 'healthy' | 'degraded' | 'gap' cursor: string lastEventAt: number | null lastErrorAt: number | null lastGapAt: number | null gapCount: number lastGapRecoveryAt: number | null retrying: boolean subscriberRecreations: number overloadLatched?: boolean overflowCount?: number}
export interface CacheInvalidationRedisOptions { maxRetriesPerRequest: number enableReadyCheck: boolean blockingTimeout?: number blockingTimeoutGrace?: number socketTimeout?: number autoResubscribe?: boolean}
export type CacheInvalidationRedisFactory = (redisUrl: string, options: CacheInvalidationRedisOptions) => Redis
/** * Schedules the repeating subscriber keepalive PING and returns a cancel handle. * Production uses an unref'd interval; focused tests inject a fake that records * the requested delay and fires ticks manually instead of waiting on real time. */export type CacheInvalidationHeartbeatIntervalScheduler = (callback: () => void, intervalMs: number) => () => void
/** Overrides for the subscriber keepalive PING; focused tests inject a fake scheduler. */export interface CacheInvalidationSubscriberHeartbeatOptions { scheduleInterval?: CacheInvalidationHeartbeatIntervalScheduler}
export interface CacheInvalidationSubscriberSupervisorOptions { scheduleInterval?: CacheInvalidationHeartbeatIntervalScheduler recreateAfterMs?: number}
const defaultRedisClientFactory: CacheInvalidationRedisFactory = (redisUrl, options) => new Redis(redisUrl, options)
// The keepalive must never hold the event loop open during shutdown.const scheduleRealHeartbeatInterval: CacheInvalidationHeartbeatIntervalScheduler = (callback, intervalMs) => { const timer = setInterval(callback, intervalMs) timer.unref() return () => clearInterval(timer)}
let runtimeScope: Scope.Closeable | undefinedlet runtime: InvalidationRuntime | undefinedlet startPromise: Promise<void> | undefinedlet stopPromise: Promise<void> | undefinedlet replayCursorIsEstablished = falselet lastProcessedStreamId = '0-0'let cacheInvalidationHealth: CacheInvalidationHealthSnapshot = { subscriberConnected: false, replayConnected: false, replayState: 'stopped', cursor: lastProcessedStreamId, lastEventAt: null, lastErrorAt: null, lastGapAt: null, gapCount: 0, lastGapRecoveryAt: null, retrying: false, subscriberRecreations: 0,}
export function getCacheInvalidationHealthSnapshot(): CacheInvalidationHealthSnapshot { return { ...cacheInvalidationHealth, cursor: lastProcessedStreamId }}
function setReplayState(state: CacheInvalidationHealthSnapshot['replayState']): void { cacheInvalidationHealth.replayState = state}
function recordCacheInvalidationError(): void { cacheInvalidationHealth.lastErrorAt = Date.now()}
function singleLineLogValue(value: string, maxLength = 256): string { return Array.from(value, (character) => { const code = character.charCodeAt(0) return code <= 0x1f || code === 0x7f || character === '\u2028' || character === '\u2029' ? ' ' : character }) .join('') .slice(0, maxLength)}
function safeCursorFileBasename(cursorFile: string): string { const name = basename(cursorFile) return /^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$/.test(name) ? name : '<unsafe-basename>'}
function cursorLogContext(cursorFile: string): { cursorFile: string; cursorFileConfigured: boolean } { return { cursorFile: safeCursorFileBasename(cursorFile), cursorFileConfigured: Boolean(process.env.WISP_CACHE_INVALIDATION_CURSOR_FILE), }}
function errorKind(error: unknown): string { if (error instanceof Error && error.name) return singleLineLogValue(error.name, 80) return 'UnknownError'}
function siteLogContext(did: string, rkey: string): { did: string; rkey: string } { return { did: singleLineLogValue(did), rkey: singleLineLogValue(rkey) }}
function invalidationLogContext(parsed: CacheInvalidationMessage): Record<string, string | boolean | undefined> { if (parsed.action === 'domain') { return { action: parsed.action, domain: parsed.domain === undefined ? undefined : singleLineLogValue(parsed.domain), domainKind: parsed.domainKind, hasCustomDomainId: Boolean(parsed.customDomainId), } }
return { action: parsed.action, ...siteLogContext(parsed.did ?? '(missing)', parsed.rkey ?? '(missing)'), tokenized: Boolean(parsed.token), }}
function parsePositiveInt(value: string | undefined, fallback: number): number { if (!value) return fallback const parsed = Number.parseInt(value, 10) return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback}
function normalizeStreamId(streamId: string | undefined): string | undefined { if (!streamId) return undefined const trimmed = streamId.trim() return STREAM_ID_PATTERN.test(trimmed) ? trimmed : undefined}
function parseStreamIdParts(streamId: string): [bigint, bigint] { const normalized = normalizeStreamId(streamId) if (!normalized) { throw new Error('Invalid Redis stream ID') }
const [msRaw, seqRaw] = normalized.split('-') as [string, string] return [BigInt(msRaw), BigInt(seqRaw)]}
export function compareStreamIds(a: string, b: string): number { const [aMs, aSeq] = parseStreamIdParts(a) const [bMs, bSeq] = parseStreamIdParts(b) if (aMs < bMs) return -1 if (aMs > bMs) return 1 if (aSeq < bSeq) return -1 if (aSeq > bSeq) return 1 return 0}
function loadCursorFromDisk(): void { replayCursorIsEstablished = false const cursorFile = getCursorFile() if (!existsSync(cursorFile)) return
try { const stored = normalizeStreamId(readFileSync(cursorFile, 'utf8')) if (!stored) { logger.warn('[CacheInvalidation] Invalid cursor file ignored', { operation: 'cursor-load', ...cursorLogContext(cursorFile), }) return } lastProcessedStreamId = stored replayCursorIsEstablished = true logger.info('[CacheInvalidation] Replay cursor loaded', { operation: 'cursor-load', streamId: stored, ...cursorLogContext(cursorFile), }) } catch (err) { recordCacheInvalidationError() logger.error('[CacheInvalidation] Cursor load failed', undefined, { operation: 'cursor-load', ...cursorLogContext(cursorFile), errorKind: errorKind(err), }) }}
const persistCursor = Effect.fnUntraced(function* (streamId: string) { const cursorFile = getCursorFile() yield* invalidationIo('cursor-persist', async () => { await mkdir(dirname(cursorFile), { recursive: true }) const temporary = `${cursorFile}.tmp` await writeFile(temporary, `${streamId}\n`, 'utf8') await rename(temporary, cursorFile) })})
const advanceStreamCursor = Effect.fnUntraced(function* (streamId: string | undefined) { const normalized = normalizeStreamId(streamId) if (!normalized || compareStreamIds(normalized, lastProcessedStreamId) <= 0) return yield* persistCursor(normalized) lastProcessedStreamId = normalized replayCursorIsEstablished = true})
function shouldSkipReplayMessage(streamId: string | undefined): boolean { const normalized = normalizeStreamId(streamId) if (!normalized) return false return compareStreamIds(normalized, lastProcessedStreamId) <= 0}
type XInfoField = { found: boolean; value: unknown }
function findXInfoField(streamInfo: unknown[], wantedField: string): XInfoField { for (let index = 0; index < streamInfo.length - 1; index += 2) { if (streamInfo[index] === wantedField) return { found: true, value: streamInfo[index + 1] } } return { found: false, value: undefined }}
function parseXInfoLength(value: unknown): number | undefined { if (typeof value === 'number') return value if (typeof value === 'string' && /^\d+$/.test(value)) return Number(value) return undefined}
function parseXInfoFirstEntry(value: unknown): string | undefined { if (value === null) return undefined if (!Array.isArray(value) || typeof value[0] !== 'string') { throw new Error('Malformed Redis XINFO first-entry response') } const firstStreamId = normalizeStreamId(value[0]) if (!firstStreamId) throw new Error('Malformed Redis XINFO first-entry ID') return firstStreamId}
function extractFirstStreamEntryId(streamInfo: unknown): string | undefined { if (!Array.isArray(streamInfo)) throw new Error('Malformed Redis XINFO STREAM response') if (streamInfo.length === 0) return undefined
const firstEntry = findXInfoField(streamInfo, 'first-entry') if (firstEntry.found) return parseXInfoFirstEntry(firstEntry.value)
if (parseXInfoLength(findXInfoField(streamInfo, 'length').value) === 0) return undefined throw new Error('Malformed Redis XINFO STREAM response')}
const recoverReplayCursorGap = Effect.fnUntraced(function* (firstStreamId: string) { const priorCursor = lastProcessedStreamId cacheInvalidationHealth.lastGapAt = Date.now() cacheInvalidationHealth.gapCount += 1 cacheInvalidationHealth.retrying = false setReplayState('degraded') recordCacheInvalidationError() logger.warn('[CacheInvalidation] Replay retention gap detected; starting fail-closed recovery', { operation: 'replay-gap-detected', priorCursor, firstRetainedStreamId: firstStreamId, gapCount: cacheInvalidationHealth.gapCount, })
const recovery = cacheInvalidationGapRecoveryDependencies // All local state must be cleared before the cursor can move. A failure in // any step leaves the old cursor intact and makes the loop retry recovery. yield* invalidationIo('gap-clearMutableCaches', async () => recovery.clearMutableCaches()) yield* invalidationIo('gap-clearUpdatingAndTerminalState', async () => recovery.clearUpdatingAndTerminalState()) yield* invalidationIo('gap-resetHtmlPrewarmState', async () => recovery.resetHtmlPrewarmState()) yield* invalidationIo('gap-evictHotTier', async () => recovery.evictHotTier()) yield* invalidationIo('gap-validateWarmCacheSourceCids', async () => recovery.validateWarmCacheSourceCids())
// Persist the reset before exposing it in memory. If this fails, retain the // old cursor and retry rather than replaying with an unrecorded safe point. yield* persistCursor('0-0') lastProcessedStreamId = '0-0' replayCursorIsEstablished = false cacheInvalidationHealth.lastGapRecoveryAt = Date.now() setReplayState('starting') logger.info('[CacheInvalidation] Replay retention gap recovered; replaying retained stream', { operation: 'replay-gap-recovered', priorCursor, firstRetainedStreamId: firstStreamId, resetCursor: lastProcessedStreamId, gapCount: cacheInvalidationHealth.gapCount, })})
const ensureReplayCursorIsRetained = Effect.fnUntraced(function* (client: Redis) { // A first boot has no established cursor to prove was behind a trimmed range. // Once replay has a cursor, a cursor that predates first-entry needs a local // cache reset before replay can safely resume from the retained history. if (!replayCursorIsEstablished) return true
const streamInfo = yield* invalidationIo('xinfo', () => client.xinfo('STREAM', STREAM)).pipe( Effect.catch((error) => { if (error.missingStream) return Effect.succeed([]) return Effect.fail(error) }), )
const firstStreamId = yield* Effect.try({ try: () => extractFirstStreamEntryId(streamInfo), catch: () => new InvalidationIoError({ operation: 'xinfo-decode' }), }) if (!firstStreamId || compareStreamIds(lastProcessedStreamId, firstStreamId) >= 0) { return true }
yield* recoverReplayCursorGap(firstStreamId) return true})
export function parseCacheInvalidationMessage(message: string): CacheInvalidationMessage | null { const parsed = decodeCacheInvalidationMessage(message) if (!parsed) return null parsed.streamId = normalizeStreamId(typeof parsed.streamId === 'string' ? parsed.streamId : undefined) return parsed}
export function parseCacheInvalidationStreamEntry(streamId: string, fields: string[]): CacheInvalidationMessage | null { const payload: Record<string, string> = {} for (let index = 0; index < fields.length - 1; index += 2) { payload[fields[index]!] = fields[index + 1]! } delete payload.ts return parseCacheInvalidationMessage(JSON.stringify({ ...payload, streamId }))}
const invalidateTier = (tier: StorageTier, prefix: string) => { if (tier.deletePrefix) return invalidationIo('tier-delete-prefix', () => tier.deletePrefix!(prefix)) return Stream.fromAsyncIterable( tier.listKeys(prefix), () => new InvalidationIoError({ operation: 'tier-list' }), ).pipe( Stream.grouped(64), Stream.mapEffect((keys) => invalidationIo('tier-delete-many', async () => { await tier.deleteMany([...keys]) return keys.length }), ), Stream.runFold( () => 0, (total, count) => total + count, ), )}
const invalidateSiteFileTiers = Effect.fnUntraced(function* ( did: string, rkey: string, dependencies: CacheInvalidationDependencies,) { const prefix = `${did}/${rkey}/` const logFailure = (tier: string, reason: unknown) => logger.error('[CacheInvalidation] Tier invalidation failed', undefined, { operation: 'site-tier-invalidate', tier, ...siteLogContext(did, rkey), errorKind: errorKind(reason), }) // Retain the shared promotion fence: direct hot/warm deletion is only an injected-test fallback. if (dependencies.storage) { const result = yield* invalidationIo('invalidate-upper-caches', () => dependencies.storage!.invalidateUpperCaches(prefix), ) for (const failure of result.failures) logFailure(failure.tier, failure.reason) return { hotDeleted: result.hotDeleted, warmDeleted: result.warmDeleted, failures: result.failures.map((failure) => failure.reason), } } const tiers: Array<{ tier: StorageTier; name: 'hot' | 'warm' }> = [{ tier: dependencies.hotTier, name: 'hot' }] if (dependencies.warmTier) tiers.push({ tier: dependencies.warmTier, name: 'warm' }) // Attempt both tiers even if one fails; replay retries partial purges idempotently. const results = yield* Effect.forEach(tiers, ({ tier }) => Effect.result(invalidateTier(tier, prefix)), { concurrency: 2, }) let hotDeleted = 0 let warmDeleted = 0 const failures: unknown[] = [] for (const [index, result] of results.entries()) { const { name } = tiers[index]! if (result._tag === 'Success') { if (name === 'hot') hotDeleted = result.success else warmDeleted = result.success } else { logFailure(name, result.failure) failures.push(result.failure) } } return { hotDeleted, warmDeleted, failures }})
const advanceReplayCursor = (source: 'pubsub' | 'replay', streamId: string | undefined) => source === 'replay' ? advanceStreamCursor(streamId) : Effect.void
const applyCacheInvalidation = Effect.fnUntraced(function* ( parsed: CacheInvalidationMessage, source: 'pubsub' | 'replay', dependencies = getCacheInvalidationDependencies(),) { const { did, rkey, action, token, streamId } = parsed cacheInvalidationHealth.lastEventAt = Date.now()
// Pub/sub is only a low-latency hint. It must not move the durable cursor because // a later live message can otherwise skip earlier ordered stream entries. if (source === 'replay' && shouldSkipReplayMessage(streamId)) { logger.info('[CacheInvalidation] Duplicate replay event skipped', { operation: 'event-skip-duplicate', source, streamId, ...invalidationLogContext(parsed), }) return }
logger.info('[CacheInvalidation] Invalidation event received', { operation: 'event-received', source, streamId, ...invalidationLogContext(parsed), })
if (action === 'domain') { applyDomainCacheInvalidation(parsed, dependencies.cache) yield* advanceReplayCursor(source, streamId) return }
if (!did || !rkey) { logger.warn('[CacheInvalidation] Site invalidation missing identifier', { operation: 'site-event-invalid', source, streamId, action, hasDid: Boolean(did), hasRkey: Boolean(rkey), }) yield* advanceReplayCursor(source, streamId) return }
if (action === 'updating') { if (isUpdatingSupersededByTerminal(did, rkey, streamId)) { logger.info('[CacheInvalidation] Superseded updating marker ignored', { operation: 'updating-marker-skip', source, streamId, ...siteLogContext(did, rkey), }) yield* advanceReplayCursor(source, streamId) return }
markSiteUpdating(did, rkey, token, streamId) yield* advanceReplayCursor(source, streamId) logger.info('[CacheInvalidation] Site marked updating', { operation: 'updating-marker-set', source, streamId, ...siteLogContext(did, rkey), tokenized: Boolean(token), }) return }
if (action === 'settings') { // Settings affect routing and fallback paths, but do not change stored site files. dependencies.cache.delete('redirectRules', `${did}:${rkey}`) dependencies.cache.delete('settings', `${did}:${rkey}`) dependencies.cache.deletePrefix('siteFiles', `${did}:${rkey}:`) dependencies.cache.deletePrefix('sourceCidMismatches', `${did}:${rkey}:`) yield* advanceReplayCursor(source, streamId) return }
// Do not remove a matching updating marker until every file-tier purge has // succeeded. A replay retry must keep serving the updating state, not stale // or partially refreshed cache content, while a terminal invalidation fails. const canClearUpdating = canClearSiteUpdating(did, rkey, token, streamId) if (!canClearUpdating && action === 'update' && token) { // The marker may be absent after a restart/TTL expiry, or belong to a newer // update. In either case, final invalidation is still required and idempotent. logger.info('[CacheInvalidation] Updating marker absent; invalidating anyway', { operation: 'updating-marker-missing', reason: 'missing-or-replaced', source, streamId, ...siteLogContext(did, rkey), }) } else if (!canClearUpdating && streamId) { // A newer start may stay visible, but terminal cache cleanup is still safe. logger.info('[CacheInvalidation] Updating marker not clearable; invalidating anyway', { operation: 'updating-marker-missing', reason: 'newer-marker', source, streamId, ...siteLogContext(did, rkey), }) }
const { hotDeleted, warmDeleted, failures } = yield* invalidateSiteFileTiers(did, rkey, dependencies)
logger.info('[CacheInvalidation] Site file tiers invalidated', { operation: 'site-tier-invalidate', hotDeleted, warmDeleted, source, streamId, ...siteLogContext(did, rkey), })
dependencies.cache.delete('redirectRules', `${did}:${rkey}`) dependencies.cache.delete('settings', `${did}:${rkey}`) dependencies.cache.delete('siteCache', `${did}:${rkey}`) dependencies.cache.deletePrefix('siteFiles', `${did}:${rkey}:`) dependencies.cache.deletePrefix('sourceCidMismatches', `${did}:${rkey}:`) dependencies.resetSiteHtmlHotCacheWarmup(did, rkey)
if (failures.length > 0) { recordCacheInvalidationError() return yield* new InvalidationIoError({ operation: 'invalidate-tiers' }) }
yield* advanceReplayCursor(source, streamId) if (canClearUpdating) { clearSiteUpdating(did, rkey, token, streamId) }
// Only a completed terminal invalidation can supersede an older replayed start. recordTerminalSiteStreamId(did, rkey, streamId)})
export async function applyCacheInvalidationForTests( parsed: CacheInvalidationMessage, source: 'pubsub' | 'replay' = 'pubsub', dependencies?: CacheInvalidationDependenciesForTests,): Promise<void> { await Effect.runPromise(applyCacheInvalidation(parsed, source, getCacheInvalidationDependencies(dependencies)))}
export function getLastProcessedStreamIdForTests(): string { return lastProcessedStreamId}
export async function resetCacheInvalidationReplayForTests(): Promise<void> { await stopCacheInvalidationSubscriber() lastProcessedStreamId = '0-0' replayCursorIsEstablished = false cacheInvalidationHealth = initialHealth() cacheInvalidationHealth.overloadLatched = false cacheInvalidationHealth.overflowCount = 0}
function applyDomainCacheInvalidation(parsed: CacheInvalidationMessage, invalidationCache: InvalidationCache): void { const domain = parsed.domain?.trim().toLowerCase() if (!domain) return
if (parsed.domainKind !== 'custom') { invalidationCache.delete('domains', domain) }
if (parsed.domainKind !== 'wisp') { invalidationCache.delete('customDomains', domain) if (parsed.customDomainId) { invalidationCache.delete('customDomains', parsed.customDomainId) invalidationCache.delete('customDomains', `hash:${parsed.customDomainId}`) } }
logger.info('[CacheInvalidation] Domain lookup cache invalidated', { operation: 'domain-cache-clear', ...invalidationLogContext({ ...parsed, domain }), })}
function initialHealth(): CacheInvalidationHealthSnapshot { return { subscriberConnected: false, replayConnected: false, replayState: 'starting', cursor: lastProcessedStreamId, lastEventAt: null, lastErrorAt: null, lastGapAt: null, gapCount: 0, lastGapRecoveryAt: null, retrying: false, subscriberRecreations: 0, overloadLatched: cacheInvalidationHealth.overloadLatched ?? false, overflowCount: cacheInvalidationHealth.overflowCount ?? 0, }}
export function isCacheInvalidationOverloaded(): boolean { return cacheInvalidationHealth.overloadLatched === true}
export function getCacheInvalidationQueueSizeForTests(): number { return runtime?.queued() ?? 0}
function boundedConfiguration(value: string | undefined, fallback: number, maximum: number): number { const parsed = Number(value) return Number.isSafeInteger(parsed) && parsed > 0 && parsed <= maximum ? parsed : fallback}
/** Process integration boundary. Resources are acquired once and owned by one scope. */export function startCacheInvalidationSubscriber( gapRecoveryOverrides?: CacheInvalidationGapRecoveryDependencies,): Promise<void> { return startCacheInvalidationSubscriberForTests(defaultRedisClientFactory, gapRecoveryOverrides)}
export function startCacheInvalidationSubscriberForTests( factory: CacheInvalidationRedisFactory, gapRecoveryOverrides?: CacheInvalidationGapRecoveryDependencies, heartbeatOptions?: CacheInvalidationSubscriberHeartbeatOptions, supervisorOptions?: CacheInvalidationSubscriberSupervisorOptions, dependencies?: CacheInvalidationDependenciesForTests,): Promise<void> { if (startPromise) return startPromise if (runtimeScope || stopPromise) return Promise.resolve() const redisUrl = process.env.REDIS_URL if (!redisUrl) { cacheInvalidationHealth.replayState = 'stopped' return Promise.resolve() } cacheInvalidationHealth = initialHealth() cacheInvalidationGapRecoveryDependencies = getCacheInvalidationGapRecoveryDependencies(gapRecoveryOverrides) loadCursorFromDisk() const scope = Scope.makeUnsafe() runtimeScope = scope startPromise = Effect.runPromise( Scope.provide(scope)( makeInvalidationRuntime({ redisUrl, factory, channel: CHANNEL, stream: STREAM, blockMs: STREAM_BLOCK_MS, batchCount: STREAM_BATCH_COUNT, blockingGraceMs: STREAM_BLOCKING_GRACE_MS, socketTimeoutMs: STREAM_SOCKET_TIMEOUT_MS, capacity: boundedConfiguration(process.env.WISP_CACHE_INVALIDATION_QUEUE_CAPACITY, 1024, 65_536), maxMessageBytes: boundedConfiguration( process.env.WISP_CACHE_INVALIDATION_MAX_MESSAGE_BYTES, 1024 * 1024, 1024 * 1024, ), heartbeatScheduler: heartbeatOptions?.scheduleInterval ?? scheduleRealHeartbeatInterval, supervisorScheduler: supervisorOptions?.scheduleInterval ?? scheduleRealHeartbeatInterval, recreateAfterMs: supervisorOptions?.recreateAfterMs ?? SUBSCRIBER_RECREATE_AFTER_MS, health: cacheInvalidationHealth, parse: parseCacheInvalidationMessage, parseEntry: parseCacheInvalidationStreamEntry, apply: (message, source) => applyCacheInvalidation(message, source, getCacheInvalidationDependencies(dependencies)), ensureRetained: ensureReplayCursorIsRetained, advance: advanceStreamCursor, cursor: () => lastProcessedStreamId, compare: compareStreamIds, error: (operation, error) => { recordCacheInvalidationError() logger.error('[CacheInvalidation] Invalidation operation failed', undefined, { operation, errorKind: errorKind(error), }) }, }), ), ) .then((acquired) => { runtime = acquired }) .catch(async (error: unknown) => { await Effect.runPromise(Scope.close(scope, Exit.void)) runtimeScope = undefined cacheInvalidationHealth.replayState = 'degraded' throw error }) .finally(() => { startPromise = undefined }) return startPromise}
export function stopCacheInvalidationSubscriber(): Promise<void> { if (stopPromise) return stopPromise if (!runtimeScope) return Promise.resolve() const scope = runtimeScope stopPromise = (async () => { await startPromise?.catch(() => undefined) await Effect.runPromise(Scope.close(scope, Exit.void)) runtime = undefined runtimeScope = undefined })().finally(() => { stopPromise = undefined }) return stopPromise}