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.
13 kB · 404 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405import type { CacheInvalidationMessage } from '@wispplace/constants'import { Cause, Deferred, Effect, Exit, Fiber, Queue, Schema, Scope } from 'effect'import type Redis from 'ioredis'import type { CacheInvalidationHealthSnapshot, CacheInvalidationHeartbeatIntervalScheduler, CacheInvalidationRedisFactory,} from './cache-invalidation'
export class InvalidationIoError extends Schema.TaggedError<InvalidationIoError>()('InvalidationIoError', { operation: Schema.Literals([ 'cursor-persist', 'xinfo', 'xinfo-decode', 'invalidate-upper-caches', 'tier-list', 'tier-delete-many', 'tier-delete-prefix', 'invalidate-tiers', 'gap-clearMutableCaches', 'gap-clearUpdatingAndTerminalState', 'gap-resetHtmlPrewarmState', 'gap-evictHotTier', 'gap-validateWarmCacheSourceCids', 'ping', 'subscriber-create', 'supervisor-create', 'replay-create', 'xread', 'replay-message-size', ]), missingStream: Schema.optionalKey(Schema.Literal(true)),}) { override get message(): string { return this.operation === 'invalidate-tiers' ? 'Failed to invalidate file caches' : `Invalidation ${this.operation} failed` }}
export const invalidationIo = <A>(operation: InvalidationIoError['operation'], run: () => PromiseLike<A>) => Effect.tryPromise({ try: () => Promise.resolve(run()), catch: (cause) => new InvalidationIoError({ operation, ...(operation === 'xinfo' && cause instanceof Error && /no such key/i.test(cause.message) ? { missingStream: true as const } : {}), }), })
interface RuntimeOptions { redisUrl: string factory: CacheInvalidationRedisFactory channel: string stream: string blockMs: number batchCount: number blockingGraceMs: number socketTimeoutMs: number capacity: number maxMessageBytes: number heartbeatScheduler: CacheInvalidationHeartbeatIntervalScheduler supervisorScheduler: CacheInvalidationHeartbeatIntervalScheduler recreateAfterMs: number health: CacheInvalidationHealthSnapshot parse(message: string): CacheInvalidationMessage | null parseEntry(id: string, fields: string[]): CacheInvalidationMessage | null apply(message: CacheInvalidationMessage, source: 'pubsub' | 'replay'): Effect.Effect<void, InvalidationIoError> ensureRetained(client: Redis): Effect.Effect<boolean, InvalidationIoError> advance(id: string): Effect.Effect<void, InvalidationIoError> cursor(): string compare(a: string, b: string): number error(operation: string, error?: unknown): void}
type Work = | { kind: 'event' message: CacheInvalidationMessage source: 'pubsub' | 'replay' ack?: Deferred.Deferred<void, InvalidationIoError> } | { kind: 'retention' ack: Deferred.Deferred<void, InvalidationIoError> }
export interface InvalidationRuntime { queued(): number}
/** One ordered worker owns accepted work; callbacks never fork suspended offers. */export const makeInvalidationRuntime = Effect.fn('hosting.makeInvalidationRuntime')(function* ( options: RuntimeOptions,): Effect.fn.Return<InvalidationRuntime, InvalidationIoError, Scope.Scope> { const taskScope = yield* Scope.make() const queue = yield* Queue.bounded<Work, Cause.Done>(options.capacity) const health = options.health let stopping = false let subscriber: Redis | undefined let replay: Redis | undefined let generation = 0 let attempt = 0 let acked = false let lastPong: number | null = null let unhealthySince: number | null = null let cancelHeartbeat: (() => void) | undefined let cancelSupervisor: (() => void) | undefined let recoveringThrough: string | undefined const heartbeatFibers = new Set<Fiber.Fiber<void>>()
const degraded = (operation: string, error?: unknown) => { health.replayState = 'degraded' options.error(operation, error) } const requireReplay = (message?: CacheInvalidationMessage) => { if (!message?.streamId) health.overloadLatched = true else if (!recoveringThrough || options.compare(message.streamId, recoveringThrough) > 0) recoveringThrough = message.streamId } const overflow = (message?: CacheInvalidationMessage) => { health.overflowCount = Math.min(Number.MAX_SAFE_INTEGER, (health.overflowCount ?? 0) + 1) requireReplay(message) if (health.overflowCount === 1) degraded('admission-overflow') else health.replayState = 'degraded' } const current = (client: Redis, version: number) => !stopping && subscriber === client && version === generation const stopHeartbeat = () => { cancelHeartbeat?.() cancelHeartbeat = undefined }
// Register before any acquisition: a later factory/scheduler failure unwinds everything. let worker: Fiber.Fiber<void> | undefined let replayFiber: Fiber.Fiber<void> | undefined yield* Effect.addFinalizer(() => Effect.gen(function* () { stopping = true generation += 1 stopHeartbeat() cancelSupervisor?.() // Disconnect first: interruption alone cannot cancel an ioredis BLOCK command. replay?.disconnect() subscriber?.disconnect() if (replayFiber) yield* Fiber.interrupt(replayFiber) for (const fiber of heartbeatFibers) yield* Fiber.interrupt(fiber) yield* Queue.end(queue) // Storage writes are not cancellable. Join accepted work before backing services close. if (worker) yield* Fiber.join(worker) health.subscriberConnected = false health.replayConnected = false health.retrying = false health.replayState = 'stopped' yield* Scope.close(taskScope, Exit.void) }), )
const process = Effect.fnUntraced( function* (work: Work) { const result = yield* Effect.result( Effect.uninterruptible( work.kind === 'retention' ? options.ensureRetained(replay!).pipe(Effect.asVoid) : options.apply(work.message, work.source), ), ) if (result._tag === 'Failure') { degraded('apply', result.failure) if (work.kind === 'event' && work.source === 'pubsub') { requireReplay(work.message) } } if (work.ack) yield* Deferred.done(work.ack, result._tag === 'Failure' ? Exit.fail(result.failure) : Exit.void) }, (effect, work) => effect.pipe( Effect.catchCause((cause) => Effect.gen(function* () { health.overloadLatched = true degraded('worker-defect') if (work.ack) yield* Deferred.done(work.ack, Exit.failCause(cause)) }), ), ), ) worker = yield* Effect.gen(function* () { while (true) yield* process(yield* Queue.take(queue)) }).pipe( Effect.catch((error) => (Cause.isDone(error) ? Effect.void : Effect.fail(error))), Effect.forkIn(taskScope, { startImmediately: true }), )
const createSubscriber = () => { const version = ++generation const client = options.factory(options.redisUrl, { maxRetriesPerRequest: 2, enableReadyCheck: true, autoResubscribe: false, }) subscriber = client acked = false lastPong = null unhealthySince = null const disconnected = () => { if (!current(client, version)) return attempt += 1 acked = false health.subscriberConnected = false stopHeartbeat() } client.on('close', disconnected) client.on('end', disconnected) client.on('error', (error) => { disconnected() if (current(client, version)) options.error('subscriber', error) }) let pinging = false client.on('ready', () => { if (!current(client, version)) return const readyAttempt = ++attempt acked = false health.subscriberConnected = false stopHeartbeat() const heartbeat = options.heartbeatScheduler(() => { if (!current(client, version) || cancelHeartbeat !== heartbeat || pinging) return pinging = true // This is an external timer callback boundary; at most one PING per generation. const fiber = Effect.runFork( invalidationIo('ping', () => client.ping()).pipe( Effect.tap(() => Effect.sync(() => { if (current(client, version) && cancelHeartbeat === heartbeat) lastPong = Date.now() }), ), Effect.catch((error) => Effect.sync(() => { if (current(client, version) && cancelHeartbeat === heartbeat) recreate('ping', error) }), ), Effect.ensuring( Effect.sync(() => { pinging = false }), ), Effect.asVoid, ), ) heartbeatFibers.add(fiber) fiber.addObserver(() => heartbeatFibers.delete(fiber)) }, 60_000) cancelHeartbeat = heartbeat client.subscribe(options.channel, (error) => { if (!current(client, version) || attempt !== readyAttempt) return acked = !error health.subscriberConnected = acked if (error) options.error('subscribe', error) else unhealthySince = null }) }) client.on('message', (channel: string, raw: string) => { if (!current(client, version) || channel !== options.channel) return // Oversize cannot be safely classified without parsing an unbounded payload. // Latch conservatively, even if a forged streamId might claim durability. if (Buffer.byteLength(raw, 'utf8') > options.maxMessageBytes) { overflow() return } const message = options.parse(raw) if (!message) { options.error('message-invalid') return } if (!Queue.offerUnsafe(queue, { kind: 'event', message, source: 'pubsub' })) overflow(message) }) } const recreate = (operation: string, error?: unknown) => { if (stopping) return stopHeartbeat() for (const fiber of heartbeatFibers) fiber.interruptUnsafe() generation += 1 health.subscriberConnected = false health.subscriberRecreations += 1 subscriber?.disconnect() subscriber = undefined acked = false lastPong = null unhealthySince = null options.error(operation, error) try { createSubscriber() } catch (cause) { degraded('subscriber-create', cause) } } yield* Effect.try({ try: createSubscriber, catch: () => new InvalidationIoError({ operation: 'subscriber-create' }), }) cancelSupervisor = yield* Effect.try({ try: () => options.supervisorScheduler(() => { if (stopping) return const now = Date.now() if (subscriber?.status === 'ready' && acked && lastPong !== null && now - lastPong < options.recreateAfterMs) { unhealthySince = null } else if (unhealthySince === null) unhealthySince = now else if (now - unhealthySince >= options.recreateAfterMs) recreate('supervisor') }, 30_000), catch: () => new InvalidationIoError({ operation: 'supervisor-create' }), }) replay = yield* Effect.try({ try: () => options.factory(options.redisUrl, { maxRetriesPerRequest: 2, enableReadyCheck: true, blockingTimeout: options.blockMs + options.blockingGraceMs, blockingTimeoutGrace: options.blockingGraceMs, socketTimeout: options.socketTimeoutMs, }), catch: () => new InvalidationIoError({ operation: 'replay-create' }), }) const reader = replay reader.on('error', (error) => { if (!stopping) { health.replayConnected = false degraded('replay-client', error) } }) reader.on('close', () => { if (!stopping) health.replayConnected = false }) reader.on('end', () => { if (!stopping) health.replayConnected = false }) reader.on('ready', () => { if (!stopping) { health.replayConnected = true health.replayState = 'starting' } })
const iteration = Effect.fnUntraced(function* () { const retained = yield* Deferred.make<void, InvalidationIoError>() yield* Queue.offer(queue, { kind: 'retention', ack: retained }) yield* Deferred.await(retained) const response = yield* invalidationIo('xread', () => reader.xread('COUNT', options.batchCount, 'BLOCK', options.blockMs, 'STREAMS', options.stream, options.cursor()), ) if (response) { for (const [, entries] of response) { for (const [id, fields] of entries) { let bytes = Buffer.byteLength(id, 'utf8') for (const field of fields) { bytes += Buffer.byteLength(field, 'utf8') if (bytes > options.maxMessageBytes) { overflow() return yield* new InvalidationIoError({ operation: 'replay-message-size' }) } } const message = options.parseEntry(id, fields) if (!message) { yield* options.advance(id) continue } const ack = yield* Deferred.make<void, InvalidationIoError>() yield* Queue.offer(queue, { kind: 'event', message, source: 'replay', ack }) yield* Deferred.await(ack) } } } if (recoveringThrough && options.compare(options.cursor(), recoveringThrough) >= 0) { recoveringThrough = undefined } health.replayConnected = true health.retrying = false health.replayState = health.overloadLatched || recoveringThrough ? 'degraded' : 'healthy' }) replayFiber = yield* Effect.gen(function* () { while (!stopping) { yield* Effect.uninterruptible(iteration()).pipe( Effect.catch((error) => Effect.gen(function* () { if (stopping) return degraded('replay', error) health.retrying = true yield* Effect.sleep('1 second') }), ), ) } }).pipe( Effect.catchCause(() => Effect.sync(() => { if (stopping) return health.overloadLatched = true degraded('replay-defect') }), ), Effect.forkIn(taskScope, { startImmediately: true }), ) return { queued: () => Queue.sizeUnsafe(queue) }})