stabilize distributed reads and revalidation
11 unmodified lines1213141516171819202122232425262 unmodified lines29303132333435363780 unmodified lines118119120121122123124125126127128129130131132133134135136417 unmodified linesExpand all55455555655755855956056156250 unmodified lines11 unmodified linesimport { promptAdminSetup } from './lib/admin-auth'import { csrfProtection } from './lib/csrf'import { closeDatabase, connectionWarmingIntervalMs, getCookieSecret, getDatabaseReadHealth, getWebhookSecretEncryptionHealth, hasSeparateDatabaseReadPool, pruneAnalyticsData, warmPrimaryConnections,} from './lib/db'import { type DNSVerificationLogLevel, DNSVerificationWorker } from './lib/dns-verification-worker'import { startPeriodicSingleFlightTask, stopServerWithGracePeriod } from './lib/lifecycle'import {2 unmodified lines createClientMetadata, getCurrentKeys, getOAuthClient, rotateKeysIfNeeded, warmOAuthLockConnection,} from './lib/oauth-client'import { startPrivateSiteReaper } from './lib/private-site-reaper'import { pruneHandoffs, pruneSessions } from './lib/private-sites-db'import { getPublicUploadRequestGateStats } from './lib/public-upload-gate'80 unmodified linesconst maintenance = startPeriodicSingleFlightTask(runMaintenance, 60 * 60 * 1000, () => logger.error('[Maintenance] Periodic maintenance failed'),)
// Keeping a connection open is not an optimisation here so much as removing a// fixed cost: a cold primary connection from a remote region costs more than// every query in a sign-in put together.const connectionWarming = startPeriodicSingleFlightTask( async () => { await Promise.all([warmPrimaryConnections(), warmOAuthLockConnection()]) }, connectionWarmingIntervalMs, () => logger.error('[Database] Connection warming failed'),)
const privateSiteReaper = startPrivateSiteReaper()
// Start DNS verification worker (runs every 10 minutes)// Can be disabled via DISABLE_DNS_WORKER=true environment variable417 unmodified linesExpand all
// Stop periodic scheduling now. The returned promises wait for only the // already-active maintenance and reaper passes. const backgroundTasks = [ { name: 'connection warming', promise: connectionWarming.stop() }, { name: 'maintenance', promise: maintenance.stop() }, { name: 'private-site reaper', promise: privateSiteReaper.stop() }, { name: 'public uploads', promise: stopAndDrainPublicUploads(GRACEFUL_SHUTDOWN_TIMEOUT_MS) }, ]50 unmodified lines75 unmodified lines76777879808081828384858687888990919293949596979811 unmodified lines75 unmodified lines expect(redis.calls).toHaveLength(2) expect(redis.calls[0]?.script).toContain('XRANGE') expect(redis.calls[0]?.script).toContain('XLEN') expect(redis.calls[0]?.script).not.toContain('MAXLEN') expect(redis.calls[0]?.keys).toEqual([`revalidate:site:delete-tombstone:${DID}:${RKEY}`, config.revalidateStream]) expect(redis.calls[0]?.args.slice(-2)[0]).toBe('1') expect(redis.calls[0]?.keys).toEqual([ `revalidate:site:delete-tombstone:${DID}:${RKEY}`, config.revalidateStream, `wisp:revalidate:quarantine:${encodeURIComponent(DID)}/${encodeURIComponent(RKEY)}`, ]) })
test('does not recreate repair work while a DLQ fence exists', async () => { const redis = new FakeQueue() redis.response = [-2, 'dlq-9'] expect(await enqueueSiteRevalidationWithRedis(redis, DID, RKEY, 'storage-miss:index.html')).toBe('quarantined') expect(redis.calls[0]?.args.slice(-2)[0]).toBe('1') })
test('does not report deduplicated when the atomic XADD transaction fails or has no proven job', async () => { const redis = new FakeQueue() redis.response = new Error('XADD unavailable') expect(await enqueueSiteRevalidationWithRedis(redis, DID, RKEY, 'storage-miss:index.html')).toBe('unavailable')11 unmodified lines4 unmodified lines567899101112131415161718101 unmodified linesExpand all1201211221231191241251261271282 unmodified lines13113213313413513613713813914014114214314414514613714714814915015119 unmodified lines17117217317417517617717817918018118217218318418518618718818919019119219319419519619719819920020120220320420520616 unmodified lines2232242252262272282292302152312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732743 unmodified lines4 unmodified lines * can clear its local caches (tiered storage, redirect rules) when a site * is updated or deleted via the firehose. */
import { DEFAULT_CACHE_INVALIDATION_CHANNEL, publishCacheInvalidationEvent } from '@wispplace/constants'import { DEFAULT_CACHE_INVALIDATION_CHANNEL, publishCacheInvalidationEvent, revalidationQuarantineKey, revalidationSiteVersionKey,} from '@wispplace/constants'import { createLogger } from '@wispplace/observability'import Redis from 'ioredis'import { config } from '../config'import { isSiteDeleteTombstoneReason } from './revalidate-queue'101 unmodified linesExpand all * Whether direct firehose work is durably covered by the revalidation stream. * Only `enqueued` and `deduplicated` permit a caller to advance its own cursor. * `capacity` is explicit backpressure and must be retried without dropping work. */export type SiteRevalidationEnqueueResult = 'enqueued' | 'deduplicated' | 'capacity' | 'unavailable'export type SiteRevalidationEnqueueResult = 'enqueued' | 'deduplicated' | 'quarantined' | 'capacity' | 'unavailable'
export interface RevalidationQueueClient { /** Runs the revalidation enqueue script atomically. */ eval(script: string, keyCount: number, ...args: string[]): PromiseLike<unknown>2 unmodified lines// This script deliberately has no MAXLEN. A producer must never remove a// pending consumer-group entry. Capacity is a hard backpressure boundary; ACK// plus XDEL is the only normal removal path.const ENQUEUE_REVALIDATION_SCRIPT = `if ARGV[7] == '1' then local quarantine = redis.call('GET', KEYS[3]) if quarantine and (ARGV[8] == '' or ARGV[8] <= quarantine) then return {-2, quarantine} endend
local existing = redis.call('GET', KEYS[1])if existing then local found = redis.call('XRANGE', KEYS[2], existing, existing, 'COUNT', 1) if #found == 1 and found[1][1] == existing then return {0, existing} end redis.call('DEL', KEYS[1])endif redis.call('XLEN', KEYS[2]) >= tonumber(ARGV[1]) then return {-1, ''} endlocal id = redis.call('XADD', KEYS[2], '*', 'did', ARGV[2], 'rkey', ARGV[3], 'reason', ARGV[4], 'ts', ARGV[5])local id = redis.call('XADD', KEYS[2], '*', 'did', ARGV[2], 'rkey', ARGV[3], 'reason', ARGV[4], 'ts', ARGV[5], 'sourceVersion', ARGV[8])redis.call('SET', KEYS[1], id, 'EX', ARGV[6])return {1, id}`
19 unmodified lines redis: RevalidationQueueClient, did: string, rkey: string, reason: string, sourceVersion = '',): Promise<SiteRevalidationEnqueueResult> { const dedupeCategory = getRevalidationDedupeCategory(reason) const dedupeKey = `revalidate:site:${dedupeCategory}:${did}:${rkey}` try { const outcome = parseEnqueueResult( await redis.eval( ENQUEUE_REVALIDATION_SCRIPT, 2, 3, dedupeKey, config.revalidateStream, revalidationQuarantineKey(did, rkey), config.revalidateStreamMaxLen.toString(), did, rkey, reason, Date.now().toString(), config.revalidateDedupeTtlSeconds.toString(), dedupeCategory === 'settings' ? '0' : '1', sourceVersion, ), ) if (!outcome) { logger.error('[Revalidate] Redis enqueue script returned an invalid result', undefined, { did, rkey, reason }) return 'unavailable' } if (outcome.status === 0 && outcome.streamId) return 'deduplicated' if (outcome.status === -2 && outcome.streamId) return 'quarantined' if (outcome.status === 1 && outcome.streamId) { logger.info(`[Revalidate] Enqueued ${did}/${rkey} after firehose failure`, { reason, streamId: outcome.streamId }) return 'enqueued' }16 unmodified linesexport async function enqueueSiteRevalidation( did: string, rkey: string, reason: string, sourceVersion = '',): Promise<SiteRevalidationEnqueueResult> { const redis = getPublisher() if (!redis) return 'unavailable' return enqueueSiteRevalidationWithRedis(redis, did, rkey, reason) return enqueueSiteRevalidationWithRedis(redis, did, rkey, reason, sourceVersion)}
/** * Record a successfully reconciled repo revision and clear only an older DLQ * fence. ATProto repo revisions are monotonic TIDs, so relay replays of the * same or an older event cannot reopen hosting repair work. */export async function recordSiteReconciliation(did: string, rkey: string, revision: string): Promise<boolean> { if (!config.redisUrl) return true const redis = getPublisher() if (!redis || !/^[a-z2-7]{13}$/.test(revision)) return false try { await waitForCacheInvalidationPublisherReady(redis, publisherReadyTimeoutMs) const result = await redis.eval( `local current = redis.call('GET', KEYS[1])if current and current >= ARGV[1] then return {0, current, 0} endredis.call('SET', KEYS[1], ARGV[1])local quarantine = redis.call('GET', KEYS[2])if quarantine and ARGV[1] > quarantine then redis.call('DEL', KEYS[2]) return {1, ARGV[1], 1}endreturn {1, ARGV[1], 0}`, 2, revalidationSiteVersionKey(did, rkey), revalidationQuarantineKey(did, rkey), revision, ) return Array.isArray(result) && (result[0] === 0 || result[0] === 1) } catch (error) { logger.error('[Revalidate] Failed to record successful site reconciliation', undefined, { did, rkey, errorKind: revalidationErrorKind(error), }) return false }}
export async function closeCacheInvalidationPublisher(): Promise<void> { if (publisher) { const toClose = publisher publisher = null3 unmodified lines693 unmodified linesExpand all6946956966976986986997007017027037027037047057067076 unmodified lines714715716717718718719720721722161 unmodified linesExpand all693 unmodified linesExpand all resolveCacheWriterResourceConfig({ BLOB_500_BACKOFF_MS: 'not-a-number', FIREHOSE_DOWNLOAD_CONCURRENCY: '20', }), ).toEqual({ blob500BackoffMs: 10 * 60 * 1000, downloadConcurrency: 1 }) ).toEqual({ blob500BackoffMs: 10 * 60 * 1000, downloadConcurrency: 3 }) expect(resolveCacheWriterResourceConfig({ BLOB_500_BACKOFF_MS: '1000' }).blob500BackoffMs).toBe(1000) })
test('serializes twenty simulated huge HTML fetch/process operations', async () => { const gate = new AsyncWorkGate(1) test('bounds concurrent blob work at three operations', async () => { const gate = new AsyncWorkGate(3) let active = 0 let maxActive = 0
await Promise.all(6 unmodified lines }), ), )
expect(maxActive).toBe(1) expect(maxActive).toBe(3) })
test('releases a heavy-work permit after an aborted operation', async () => { const gate = new AsyncWorkGate(1)161 unmodified linesExpand all85 unmodified lines8687888990919293909192939495969735 unmodified lines1331341351361371381391371381391401411421431477 unmodified linesExpand all162116221623162416251625162616271628162916301631163216331634163316341635163616371637163816391640163816391640164116421643164416451646164716481649165016511652165316541655165616571658658 unmodified linesExpand all85 unmodified lines 10 * 60 * 1000, 1_000, 24 * 60 * 60 * 1000, ), // Blob reads deliberately remain serial within a site update. A value // other than one is rejected rather than allowing several 200 MiB buffers // to queue before a quota violation can stop the batch. downloadConcurrency: getBoundedEnvironmentInteger(environment, 'FIREHOSE_DOWNLOAD_CONCURRENCY', 1, 1, 1), // Three independent blob reads keep large sites moving without making // concurrency unbounded. Manifest and actual logical-size checks remain the // admission boundary; network retries do not count as additional site data. downloadConcurrency: getBoundedEnvironmentInteger(environment, 'FIREHOSE_DOWNLOAD_CONCURRENCY', 3, 1, 3), }}
const resourceConfig = resolveCacheWriterResourceConfig()35 unmodified lines } }}
// Do not make this configurable above one: it starts before safeFetchBlob and// therefore bounds buffered binary downloads as well as gzip/HTML work.const blobProcessingGate = new AsyncWorkGate(1)// The permit begins before safeFetchBlob, so this bounds buffered binary// downloads as well as gzip/HTML work to the configured three-way ceiling.const blobProcessingGate = new AsyncWorkGate(DOWNLOAD_CONCURRENCY)
/** * Localhost PDS access requires all three dev-only gates. Production remains * HTTPS-only, and callers still opt in explicitly at both the identity and1477 unmodified linesExpand all plan.update.resources, )}
/** Download serial bounded batches; terminal limit failures stop the entire update. *//** Download bounded parallel batches; terminal limit failures stop the update after the active batch settles. */async function downloadSiteFiles( plan: AccountedSiteFilePlan, files: FileInfo[] = plan.filesToDownload,): Promise<DownloadFailure[]> { const failures: DownloadFailure[] = [] for (let index = 0; index < files.length; index += DOWNLOAD_CONCURRENCY) { assertRevalidationActive(plan.update.resources) for (const file of files.slice(index, index + DOWNLOAD_CONCURRENCY)) { try { const batch = files.slice(index, index + DOWNLOAD_CONCURRENCY) const results = await Promise.allSettled( batch.map(async (file) => { assertRevalidationActive(plan.update.resources) const result = await downloadPlannedFile(plan, file) plan.uncompressedSizes.set(file.path, result.uncompressedSize) } catch (error) { if (isTerminalIngestError(error)) throw error failures.push({ path: file.path, error }) return { file, result } }), )
let terminalError: unknown for (const [batchIndex, result] of results.entries()) { const file = batch[batchIndex] if (!file) continue if (result.status === 'fulfilled') { plan.uncompressedSizes.set(file.path, result.value.result.uncompressedSize) } else if (isTerminalIngestError(result.reason)) { terminalError ??= result.reason } else { failures.push({ path: file.path, error: result.reason }) } } if (terminalError !== undefined) throw terminalError } return failures}
658 unmodified linesExpand all12 unmodified lines1314151617171819202122232425262728252930313233343536373820 unmodified lines59606162636465556667686970217 unmodified linesExpand all288289290291292293294295296285286287288297298299300301302303304305306307120 unmodified linesExpand all4284294304314324334344354364374384394404414424435 unmodified lines449450451452431453454455456457495 unmodified linesExpand all12 unmodified linesimport type { Record as WispSettings } from '@wispplace/lexicons/types/place/wisp/settings'import { validateRecord as validateSettingsRecord } from '@wispplace/lexicons/types/place/wisp/settings'import { createLogger } from '@wispplace/observability'import { config } from '../config'import { enqueueSiteRevalidation, type SiteRevalidationEnqueueResult } from './cache-invalidation'import { enqueueSiteRevalidation, recordSiteReconciliation, type SiteRevalidationEnqueueResult,} from './cache-invalidation'import { handleSettingsDelete, handleSettingsUpdate, handleSiteCreateOrUpdate, handleSiteDelete } from './cache-writer'import { readDurableCursor, relayFingerprint, saveDurableCursor } from './leader'import { SETTINGS_DELETE_FAILURE_REASON, SETTINGS_UPDATE_FAILURE_REASON, SITE_DELETE_TOMBSTONE_REASON,} from './revalidate-queue'import { createRevalidationResourceContext, type RevalidationResourceContext } from './revalidate-resources'import { assertRevalidationActive, createRevalidationResourceContext, type RevalidationResourceContext,} from './revalidate-resources'import { DEFAULT_REVALIDATE_DEADLINE_MS, DEFAULT_REVALIDATE_TRANSFER_BUDGET_BYTES, resolveRevalidateWorkerRuntimeConfig,} from './revalidate-worker'20 unmodified linesconst firehoseTransferBudgetBytes = firehoseResourceConfig.transferBudgetBytes ?? DEFAULT_REVALIDATE_TRANSFER_BUDGET_BYTES
function createFilesystemEventResources(upstreamSignal?: AbortSignal): RevalidationResourceContext { // Site updates keep per-request fetch timeouts and manifest/file limits, but // have no aggregate wall or transfer deadline. A large valid publication must // not fail because serial/parallel retries counted its bytes twice. return createRevalidationResourceContext(firehoseOperationDeadlineMs, firehoseTransferBudgetBytes, upstreamSignal) return createRevalidationResourceContext(null, null, upstreamSignal)}
function createSettingsEventResources(upstreamSignal?: AbortSignal): RevalidationResourceContext { return createRevalidationResourceContext(217 unmodified linesExpand all did: string, rkey: string, reason: string, lifecycleSignal: AbortSignal, sourceVersion = '',): Promise<void> { // Do not route a full stream through requestReplay: that path invokes the // leader failure callback and stops this process's only consumer. Capacity // waits here, while the independently started worker remains alive to drain. const result = await retryDurableRevalidationUntilAvailable(() => enqueueSiteRevalidation(did, rkey, reason), { shouldContinue: () => acceptingEvents, signal: lifecycleSignal, }) const result = await retryDurableRevalidationUntilAvailable( () => enqueueSiteRevalidation(did, rkey, reason, sourceVersion), { shouldContinue: () => acceptingEvents, signal: lifecycleSignal, }, ) if (isDurableRevalidation(result)) { durableReplayController.recordDurableSuccess() return }120 unmodified linesExpand all const { did, rkey } = event let recoveryReason: string | undefined try { logger.debug('[place.wisp.fs] Processing event', { did, rkey, event: event.event }) assertRevalidationActive(resources) recoveryReason = await reconcileFilesystemEvent(event, resources) assertRevalidationActive(resources) // Record only a completed reconciliation. The monotonic repo revision keeps // a replay of this same event from clearing a later DLQ fence. if (!(await recordSiteReconciliation(did, rkey, event.rev))) { throw new Error('Successful site reconciliation could not be recorded') } } catch (error) { logger.error('[place.wisp.fs] Error handling event', undefined, { did, rkey,5 unmodified lines } finally { resources.close() }
if (recoveryReason) await requireDurableRevalidation(did, rkey, recoveryReason, lifecycleSignal) if (recoveryReason) await requireDurableRevalidation(did, rkey, recoveryReason, lifecycleSignal, event.rev) logger.debug('[place.wisp.fs] Completed event', { did, rkey, event: event.event })}
async function reconcileSettingsEvent(495 unmodified linesExpand all70 unmodified lines7172737475757677787980818283848586878889909192939495969798991001011021033 unmodified lines107108109110919211111211311495961151161171181191201219812212310012412512610212712812913013110413213313413513613711013813914014114214314414514611914714812114915015115215312 unmodified lines70 unmodified lines}
export interface RevalidationResourceContext { readonly signal: AbortSignal /** Null for filesystem work, which is bounded per request and by lifecycle rather than wall time. */ readonly deadlineAt: number readonly deadlineAt: number | null readonly transferBudget: TransferByteBudgetLike close(): void}
/** Counts transfer bytes without treating retries as quota consumption. */class UnboundedTransferByteBudget implements TransferByteBudgetLike { readonly maxBytes = Number.POSITIVE_INFINITY private consumed = 0
get consumedBytes(): number { return this.consumed }
get remainingBytes(): number { return Number.POSITIVE_INFINITY }
consume(bytes: number): void { if (!Number.isSafeInteger(bytes) || bytes < 0) throw new RangeError('Transfer byte charge must be non-negative') this.consumed = Math.min(Number.MAX_SAFE_INTEGER, this.consumed + bytes) }}
function abortReason(signal: AbortSignal): Error { if (signal.reason instanceof Error) return signal.reason return new Error(signal.reason ? String(signal.reason) : 'Revalidation aborted')}3 unmodified lines * returned close method is idempotent and must be called by the message * boundary after all work has stopped. */export function createRevalidationResourceContext( deadlineMs: number, transferBudgetBytes: number, deadlineMs: number | null, transferBudgetBytes: number | null, upstreamSignal?: AbortSignal,): RevalidationResourceContext { if (!Number.isSafeInteger(deadlineMs) || deadlineMs < 1) { throw new RangeError('Revalidation deadline must be a positive safe integer') if (deadlineMs !== null && (!Number.isSafeInteger(deadlineMs) || deadlineMs < 1)) { throw new RangeError('Revalidation deadline must be null or a positive safe integer') } if (transferBudgetBytes !== null && (!Number.isSafeInteger(transferBudgetBytes) || transferBudgetBytes < 1)) { throw new RangeError('Revalidation transfer budget must be null or a positive safe integer') }
const deadlineAt = Date.now() + deadlineMs const deadlineAt = deadlineMs === null ? null : Date.now() + deadlineMs const controller = new AbortController() const budget = new TransferByteBudget(transferBudgetBytes) const boundedBudget = transferBudgetBytes === null ? null : new TransferByteBudget(transferBudgetBytes) const budget: TransferByteBudgetLike = boundedBudget ?? new UnboundedTransferByteBudget() let closed = false const timer = setTimeout(() => controller.abort(new RevalidationDeadlineError(deadlineAt)), deadlineMs) const timer = deadlineAt === null ? undefined : setTimeout(() => controller.abort(new RevalidationDeadlineError(deadlineAt)), deadlineMs as number) const abortFromUpstream = () => controller.abort(abortReason(upstreamSignal as AbortSignal)) const abortFromBudget = () => controller.abort(abortReason(budget.signal)) const abortFromBudget = () => controller.abort(abortReason(boundedBudget!.signal))
if (upstreamSignal) { if (upstreamSignal.aborted) abortFromUpstream() else upstreamSignal.addEventListener('abort', abortFromUpstream, { once: true }) } budget.signal.addEventListener('abort', abortFromBudget, { once: true }) boundedBudget?.signal.addEventListener('abort', abortFromBudget, { once: true })
return { signal: controller.signal, deadlineAt, transferBudget: budget, close: () => { if (closed) return closed = true clearTimeout(timer) if (timer !== undefined) clearTimeout(timer) upstreamSignal?.removeEventListener('abort', abortFromUpstream) budget.signal.removeEventListener('abort', abortFromBudget) boundedBudget?.signal.removeEventListener('abort', abortFromBudget) }, }}
12 unmodified lines126 unmodified linesExpand all1271281291301311321311321331341351361371381391381401411421431441451461451471481491501511521 unmodified line1541551561571551581591601611621631641651661671685 unmodified lines1741751761771691701781791801811821831841851861871884 unmodified lines19319419519619719819920020122 unmodified lines22422522622721322822923023123223330 unmodified lines126 unmodified linesExpand all
expect(evals).toHaveLength(1) const [call] = evals if (!call) throw new Error('Expected quarantine script call') expect(call.keyCount).toBe(2) expect(call.args.slice(0, 4)).toEqual([ expect(call.keyCount).toBe(4) expect(call.args.slice(0, 6)).toEqual([ config.revalidateStream, config.revalidateDlqStream, 'wisp:revalidate:quarantine:/site', 'wisp:revalidate:version:/site', config.revalidateGroup, id, ]) expect(call.args.slice(4, 9)).toEqual([ expect(call.args.slice(6, 11)).toEqual([ '', 'site', 'malformed', 'MALFORMED_MESSAGE', 'Revalidation failed: MALFORMED_MESSAGE', ]) expect(call.script.indexOf("redis.call('XADD'")).toBeLessThan(call.script.indexOf("redis.call('XACK'")) expect(call.script.indexOf("redis.call('XADD'")).toBeLessThan(call.script.indexOf("redis.call('SET'")) expect(call.script.indexOf("redis.call('SET'")).toBeLessThan(call.script.indexOf("redis.call('XACK'")) expect(call.script.indexOf("redis.call('XACK'")).toBeLessThan(call.script.indexOf("redis.call('XDEL'")) })
test('maps quarantine ARGVs to every DLQ field and MAXLEN', async () => {1 unmodified line const { redis, evals } = fakeRedis()
await quarantineRevalidationMessage( redis, { fields: {}, did: 'did:plc:contract', rkey: 'docs/index.html', reason: 'storage-miss:docs/index.html' }, { fields: {}, did: 'did:plc:contract', rkey: 'docs/index.html', reason: 'storage-miss:docs/index.html', sourceVersion: '3mzzzzzzzzzzz', }, id, new Error('upstream timeout'), 7, 'transient',5 unmodified lines if (!call) throw new Error('Expected quarantine script call') const quarantinedAt = call.args[call.keyCount + 9] if (!quarantinedAt) throw new Error('Expected a quarantine timestamp argument') expect(quarantinedAt).toMatch(/^\d+$/) expect(call.keyCount).toBe(2) expect(call.args.slice(0, 2)).toEqual([config.revalidateStream, config.revalidateDlqStream]) expect(call.keyCount).toBe(4) expect(call.args.slice(0, 4)).toEqual([ config.revalidateStream, config.revalidateDlqStream, 'wisp:revalidate:quarantine:did%3Aplc%3Acontract/docs%2Findex.html', 'wisp:revalidate:version:did%3Aplc%3Acontract/docs%2Findex.html', ]) expect(call.args.slice(call.keyCount)).toEqual([ config.revalidateGroup, id, 'did:plc:contract',4 unmodified lines '7', 'transient', expect.any(String), String(config.revalidateDlqStreamMaxLen), '3mzzzzzzzzzzz', ])
expect(parseDlqXaddContract(call.script)).toEqual({ maxLenArgv: 11,22 unmodified lines attempts: '7', quarantinedAt, }, }) expect(call.script.indexOf("redis.call('XADD'")).toBeLessThan(call.script.indexOf("redis.call('XACK'")) expect(call.script.indexOf("redis.call('XADD'")).toBeLessThan(call.script.indexOf("redis.call('SET'")) expect(call.script.indexOf("redis.call('SET'")).toBeLessThan(call.script.indexOf("redis.call('XACK'")) expect(call.script.indexOf("redis.call('XACK'")).toBeLessThan(call.script.indexOf("redis.call('XDEL'")) })
test('quarantines a transient failure at the attempt bound without materializing later', async () => {30 unmodified lines297 unmodified linesExpand all298299300301302303302303304305306307324 unmodified linesExpand all632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678466 unmodified linesExpand all297 unmodified linesExpand all }
async eval(_script: string, keyCount: number, ...args: string[]): Promise<unknown> { this.calls.push({ command: 'EVAL', args: [keyCount, ...args] }) if (keyCount === 2) { this.quarantineAttempts.push(Number(args[2 + 7])) if (keyCount === 4) { this.quarantineAttempts.push(Number(args[4 + 7])) this.pending = false return [1, 'dlq-1', 1] } this.pending = false324 unmodified linesExpand all })})
describe('ordinary revalidation', () => { test('uses no aggregate deadline or transfer cap for filesystem repair', async () => { const { redis, acks } = createRedis() let observedResources: Parameters<RevalidateWorkerDependencies['handleSiteCreateOrUpdate']>[4] extends infer Options ? Options : never const dependencies = createDependencies({ fetchSiteRecord: async () => ({ record: reappearedRecord, cid: 'cid' }), handleSiteCreateOrUpdate: async (_did, _rkey, _record, _cid, options) => { observedResources = options }, })
await processRevalidationMessage(MESSAGE_ID, messageFields('storage-miss:index.html'), redis, dependencies)
expect(observedResources?.resources?.deadlineAt).toBeNull() expect(observedResources?.resources?.transferBudget.maxBytes).toBe(Number.POSITIVE_INFINITY) observedResources?.resources?.transferBudget.consume(1_000_000_000) expect(observedResources?.resources?.signal.aborted).toBe(false) expect(acks).toEqual([MESSAGE_ID]) })
test('acknowledges queued duplicates without repair while a DLQ fence exists', async () => { const { redis, acks, streamDeletes } = createRedis() ;(redis as RevalidateRedisClient).get = async () => 'dlq-9' let lookups = 0 const dependencies = createDependencies({ fetchSiteRecord: async () => { lookups++ return { record: reappearedRecord, cid: 'cid' } }, })
await processRevalidationMessage(MESSAGE_ID, messageFields('storage-miss:index.html'), redis, dependencies)
expect(lookups).toBe(0) expect(acks).toEqual([MESSAGE_ID]) expect(streamDeletes).toEqual([MESSAGE_ID]) })
test('leaves transient or invalid typed PDS outcomes pending', async () => { const { redis, acks, streamDeletes } = createRedis() const dependencies = createDependencies({ fetchSiteRecordOutcome: async () => ({ kind: 'retryable', error: 'INVALID_RECORD' }),466 unmodified linesExpand all