From 17eb4988847766717523ea67daef298ca766ea85 Mon Sep 17 00:00:00 2001 From: "nekomimi.pet" Date: Thu, 24 Sep 2026 16:19:28 -0400 Subject: [PATCH] mark sites absent on confirmed RecordNotFound; stop serving, purge after grace --- .../src/lib/absent-site-sweeper.test.ts | 111 +++++++++++++ .../src/lib/absent-site-sweeper.ts | 149 ++++++++++++++++++ .../src/lib/cache-writer.test.ts | 26 ++- apps/firehose-service/src/lib/cache-writer.ts | 61 ++++--- apps/firehose-service/src/lib/db.ts | 49 +++++- .../src/lib/revalidate-worker.test.ts | 54 +++++++ .../src/lib/revalidate-worker.ts | 40 ++++- apps/hosting-service/src/lib/db.ts | 2 +- .../src/lib/file-serving.integration.test.ts | 17 ++ apps/hosting-service/src/lib/file-serving.ts | 3 + apps/main-app/src/lib/migrations.ts | 9 ++ packages/@wispplace/database/src/types.ts | 3 + 12 files changed, 491 insertions(+), 33 deletions(-) create mode 100644 apps/firehose-service/src/lib/absent-site-sweeper.test.ts create mode 100644 apps/firehose-service/src/lib/absent-site-sweeper.ts diff --git a/apps/firehose-service/src/lib/absent-site-sweeper.test.ts b/apps/firehose-service/src/lib/absent-site-sweeper.test.ts new file mode 100644 index 0000000..6ac63a8 --- /dev/null +++ b/apps/firehose-service/src/lib/absent-site-sweeper.test.ts @@ -0,0 +1,111 @@ +import { describe, expect, test } from 'bun:test' +import { + ABSENT_SITE_PURGE_MIN_AGE_SECONDS, + type AbsentSiteSweeperDependencies, + decideAbsentSiteAction, + sweepAbsentSites, +} from './absent-site-sweeper' +import type { SiteRecordFetchOutcome } from './cache-writer' +import type { AbsentSiteMark } from './db' + +const NOW = 2_000_000_000 +const OLD = NOW - ABSENT_SITE_PURGE_MIN_AGE_SECONDS +const confirmed: SiteRecordFetchOutcome = { kind: 'absent', confirmed: true } + +describe('decideAbsentSiteAction', () => { + test('restores a site whose record is back', () => { + const present = { kind: 'present', record: {} as never, cid: 'bafy' } as const + expect(decideAbsentSiteAction(present, null, NOW)).toBe('restore') + }) + + test('never acts on an unreachable PDS or an unconfirmed 404', () => { + const mark = { absent_since: OLD - 1, absent_checks: 99 } + expect(decideAbsentSiteAction({ kind: 'retryable', error: 'FETCH_FAILED' }, mark, NOW)).toBe('skip') + expect(decideAbsentSiteAction({ kind: 'retryable', error: 'PDS_UNRESOLVED' }, mark, NOW)).toBe('skip') + expect(decideAbsentSiteAction({ kind: 'absent', confirmed: false }, mark, NOW)).toBe('skip') + expect(decideAbsentSiteAction({ kind: 'absent' }, mark, NOW)).toBe('skip') + }) + + test('keeps counting inside the grace period', () => { + expect(decideAbsentSiteAction(confirmed, { absent_since: NOW - 60, absent_checks: 10 }, NOW)).toBe('count') + expect(decideAbsentSiteAction(confirmed, { absent_since: OLD - 1, absent_checks: 2 }, NOW)).toBe('count') + }) + + test('purges only after enough confirmations over the full grace period', () => { + expect(decideAbsentSiteAction(confirmed, { absent_since: OLD, absent_checks: 3 }, NOW)).toBe('purge') + expect(decideAbsentSiteAction(confirmed, { absent_since: String(OLD) as never, absent_checks: 3 }, NOW)).toBe( + 'purge', + ) + }) +}) + +describe('sweepAbsentSites', () => { + function fakeDependencies( + outcomes: Record, + marks: Record, + ) { + const calls: string[] = [] + const dependencies: AbsentSiteSweeperDependencies = { + listAbsentSites: async () => Object.values(marks), + fetchSiteRecordOutcome: async (did) => { + const outcome = outcomes[did]! + if (outcome instanceof Error) throw outcome + return outcome + }, + markSiteAbsent: async (did) => { + const mark = marks[did]! + mark.absent_checks++ + calls.push(`mark ${did}`) + return mark + }, + clearSiteAbsent: async (did) => { + calls.push(`clear ${did}`) + return true + }, + handleSiteDelete: async (did) => { + calls.push(`delete ${did}`) + }, + publishCacheInvalidation: async (did) => { + calls.push(`invalidate ${did}`) + }, + now: () => NOW, + } + return { dependencies, calls } + } + + const mark = (did: string, absent_since: number, absent_checks: number): AbsentSiteMark => ({ + did, + rkey: 'site', + absent_since, + absent_checks, + }) + + test('restores, counts, purges and skips per site, and never deletes an unreachable one', async () => { + const { dependencies, calls } = fakeDependencies( + { + back: { kind: 'present', record: {} as never, cid: 'bafy' }, + young: confirmed, + old: confirmed, + away: { kind: 'retryable', error: 'FETCH_FAILED' }, + broken: new Error('boom'), + }, + { + back: mark('back', OLD, 5), + young: mark('young', NOW - 3600, 0), + old: mark('old', OLD, 2), + away: mark('away', OLD - 999_999, 50), + broken: mark('broken', OLD, 9), + }, + ) + + const result = await sweepAbsentSites(undefined, dependencies) + + expect(result).toEqual({ checked: 5, restored: 1, counted: 1, purged: 1, skipped: 2 }) + expect(calls).toContain('clear back') + expect(calls).toContain('invalidate back') + expect(calls).toContain('mark young') + expect(calls).toContain('delete old') + expect(calls.filter((call) => call.endsWith(' away'))).toEqual([]) + expect(calls.filter((call) => call.startsWith('delete') && !call.endsWith(' old'))).toEqual([]) + }) +}) diff --git a/apps/firehose-service/src/lib/absent-site-sweeper.ts b/apps/firehose-service/src/lib/absent-site-sweeper.ts new file mode 100644 index 0000000..67818d8 --- /dev/null +++ b/apps/firehose-service/src/lib/absent-site-sweeper.ts @@ -0,0 +1,149 @@ +/** + * Re-checks cached sites the owner's PDS reported as `RecordNotFound`. + * + * A mark only ever comes from a confirmed RecordNotFound on the owner's current + * PDS (resolved through PLC). An unreachable, deactivated or taken-down PDS is a + * retryable outcome and leaves the site untouched, so an account that is away + * for months comes back to its site. Purging reuses `handleSiteDelete`, which + * re-reads the record under the per-site lock before removing anything, and it + * only ever removes cached files: domain claims are never touched here. + */ +import { createLogger } from '@wispplace/observability' +import { publishCacheInvalidation } from './cache-invalidation' +import { fetchSiteRecordOutcome, handleSiteDelete, type SiteRecordFetchOutcome } from './cache-writer' +import { type AbsentSiteMark, clearSiteAbsent, listAbsentSites, markSiteAbsent } from './db' +import { createRevalidationResourceContext } from './revalidate-resources' + +const logger = createLogger('firehose-service') + +export const ABSENT_SITE_PURGE_MIN_CHECKS = 3 +export const ABSENT_SITE_PURGE_MIN_AGE_SECONDS = 30 * 24 * 60 * 60 +const SWEEP_INTERVAL_MS = 6 * 60 * 60 * 1000 +const FIRST_SWEEP_DELAY_MS = 10 * 60 * 1000 +const SWEEP_BATCH = 100 +const LOOKUP_DEADLINE_MS = 60_000 +const LOOKUP_BYTE_BUDGET = 4 * 1024 * 1024 + +export type AbsentSiteAction = 'restore' | 'count' | 'purge' | 'skip' + +/** What one confirmed or unconfirmed lookup means for a marked site. */ +export function decideAbsentSiteAction( + outcome: SiteRecordFetchOutcome, + markAfterCount: Pick | null, + nowSeconds: number, +): AbsentSiteAction { + if (outcome.kind === 'present') return 'restore' + if (outcome.kind !== 'absent' || outcome.confirmed !== true) return 'skip' + if (!markAfterCount) return 'skip' + const age = nowSeconds - Number(markAfterCount.absent_since) + return markAfterCount.absent_checks >= ABSENT_SITE_PURGE_MIN_CHECKS && age >= ABSENT_SITE_PURGE_MIN_AGE_SECONDS + ? 'purge' + : 'count' +} + +export interface AbsentSiteSweeperDependencies { + listAbsentSites: typeof listAbsentSites + fetchSiteRecordOutcome: typeof fetchSiteRecordOutcome + markSiteAbsent: typeof markSiteAbsent + clearSiteAbsent: typeof clearSiteAbsent + handleSiteDelete: (did: string, rkey: string) => Promise + publishCacheInvalidation: typeof publishCacheInvalidation + now: () => number +} + +const defaultDependencies: AbsentSiteSweeperDependencies = { + listAbsentSites, + fetchSiteRecordOutcome: (did, rkey) => { + const resources = createRevalidationResourceContext(LOOKUP_DEADLINE_MS, LOOKUP_BYTE_BUDGET) + return fetchSiteRecordOutcome(did, rkey, resources).finally(() => resources.close()) + }, + markSiteAbsent, + clearSiteAbsent, + handleSiteDelete: (did, rkey) => handleSiteDelete(did, rkey), + publishCacheInvalidation, + now: () => Math.floor(Date.now() / 1000), +} + +export interface AbsentSiteSweepResult { + checked: number + restored: number + counted: number + purged: number + skipped: number +} + +/** One pass over marked sites, oldest first. Errors on one site never stop the pass. */ +export async function sweepAbsentSites( + signal?: AbortSignal, + dependencies: AbsentSiteSweeperDependencies = defaultDependencies, +): Promise { + const result: AbsentSiteSweepResult = { checked: 0, restored: 0, counted: 0, purged: 0, skipped: 0 } + for (const site of await dependencies.listAbsentSites(SWEEP_BATCH)) { + if (signal?.aborted) break + result.checked++ + const { did, rkey } = site + try { + const outcome = await dependencies.fetchSiteRecordOutcome(did, rkey) + const confirmedAbsent = outcome.kind === 'absent' && outcome.confirmed === true + const mark = confirmedAbsent ? await dependencies.markSiteAbsent(did, rkey) : null + const action = decideAbsentSiteAction(outcome, mark, dependencies.now()) + if (action === 'restore') { + if (await dependencies.clearSiteAbsent(did, rkey)) { + await dependencies.publishCacheInvalidation(did, rkey, 'update') + logger.info(`[AbsentSites] Record is back; serving again: ${did}/${rkey}`) + } + result.restored++ + } else if (action === 'purge') { + await dependencies.handleSiteDelete(did, rkey) + logger.info(`[AbsentSites] Purged cached files after grace period: ${did}/${rkey}`, { + absentChecks: mark?.absent_checks, + }) + result.purged++ + } else if (action === 'count') { + result.counted++ + } else { + result.skipped++ + } + } catch (error) { + result.skipped++ + logger.warn(`[AbsentSites] Check failed; leaving ${did}/${rkey} as is`, { + errorKind: error instanceof Error ? error.constructor.name || 'Error' : 'UnknownError', + }) + } + } + return result +} + +let timer: ReturnType | null = null +let controller: AbortController | null = null +let activeSweep: Promise | null = null + +/** Leader-only: started and stopped with the revalidation worker. */ +export function startAbsentSiteSweeper(): void { + if (controller) return + const own = new AbortController() + controller = own + const schedule = (delayMs: number) => { + timer = setTimeout(async () => { + if (own.signal.aborted) return + activeSweep = sweepAbsentSites(own.signal) + .then((result) => { + if (result.checked > 0) logger.info('[AbsentSites] Sweep complete', { ...result }) + }) + .catch(() => logger.warn('[AbsentSites] Sweep failed')) + await activeSweep + activeSweep = null + if (!own.signal.aborted) schedule(SWEEP_INTERVAL_MS) + }, delayMs) + timer.unref?.() + } + schedule(FIRST_SWEEP_DELAY_MS) +} + +export async function stopAbsentSiteSweeper(): Promise { + controller?.abort() + controller = null + if (timer) clearTimeout(timer) + timer = null + await activeSweep?.catch(() => undefined) +} diff --git a/apps/firehose-service/src/lib/cache-writer.test.ts b/apps/firehose-service/src/lib/cache-writer.test.ts index 0e129e6..c98f136 100644 --- a/apps/firehose-service/src/lib/cache-writer.test.ts +++ b/apps/firehose-service/src/lib/cache-writer.test.ts @@ -95,12 +95,22 @@ function storageMetadata(customMetadata: Record): StorageMetadat } describe('PDS getRecord absence responses', () => { - test('accepts the ATProto RecordNotFound XRPC error as authoritative absence', async () => { + test('accepts the PDS HTTP 400 RecordNotFound as confirmed absence', async () => { const response = new Response(JSON.stringify({ error: 'RecordNotFound', message: 'not found' }), { status: 400, headers: { 'content-type': 'application/json' }, }) - expect(await readPdsRecordJsonResponse(response)).toEqual({ kind: 'absent' }) + expect(await readPdsRecordJsonResponse(response)).toEqual({ kind: 'absent', confirmed: true }) + }) + + test('keeps a deactivated or taken-down repo retryable, not absent', async () => { + for (const error of ['RepoDeactivated', 'RepoTakendown', 'RepoNotFound']) { + const response = new Response(JSON.stringify({ error }), { + status: 400, + headers: { 'content-type': 'application/json' }, + }) + await expect(readPdsRecordJsonResponse(response)).rejects.toMatchObject({ status: 400 }) + } }) test('keeps an unrelated HTTP 400 retryable', async () => { @@ -111,8 +121,16 @@ describe('PDS getRecord absence responses', () => { await expect(readPdsRecordJsonResponse(response)).rejects.toMatchObject({ status: 400 }) }) - test('accepts a gateway HTTP 404 as authoritative absence', async () => { - expect(await readPdsRecordJsonResponse(new Response(null, { status: 404 }))).toEqual({ kind: 'absent' }) + test('treats a bare gateway HTTP 404 as unconfirmed absence', async () => { + expect(await readPdsRecordJsonResponse(new Response(null, { status: 404 }))).toEqual({ + kind: 'absent', + confirmed: false, + }) + }) + + test('confirms a 404 that carries the RecordNotFound body', async () => { + const response = new Response(JSON.stringify({ error: 'RecordNotFound' }), { status: 404 }) + expect(await readPdsRecordJsonResponse(response)).toEqual({ kind: 'absent', confirmed: true }) }) test('returns a successful bounded JSON response', async () => { diff --git a/apps/firehose-service/src/lib/cache-writer.ts b/apps/firehose-service/src/lib/cache-writer.ts index 5387144..4183d50 100644 --- a/apps/firehose-service/src/lib/cache-writer.ts +++ b/apps/firehose-service/src/lib/cache-writer.ts @@ -176,34 +176,35 @@ const pdsRequestOptions = allowDevLocalPdsFetch ? { allowLocalhost: true } : und */ export const MAX_PDS_RECORD_RESPONSE_BYTES = 1024 * 1024 -export type PdsRecordJsonResponse = { kind: 'present'; value: T } | { kind: 'absent' } +/** + * `confirmed` is true only when the PDS itself answered with the typed XRPC + * error `RecordNotFound`. That is the bar for marking a cached site absent; + * a bare 404 is still absence for non-destructive callers but may come from a + * gateway or proxy in front of the PDS rather than the PDS. + */ +export type PdsRecordJsonResponse = { kind: 'present'; value: T } | { kind: 'absent'; confirmed: boolean } + +const isRecordNotFoundBody = (body: unknown): boolean => + typeof body === 'object' && body !== null && (body as { error?: unknown }).error === 'RecordNotFound' /** * Interpret the ATProto getRecord response without treating every HTTP 400 as - * absence. Conformant PDS implementations report a missing record as the typed - * XRPC error `RecordNotFound` with status 400; some gateways use 404 instead. - * The response body is bounded by safeFetch before this function reads it. + * absence. A PDS reports a missing record as HTTP 400 with the typed XRPC error + * `RecordNotFound`; other 400s (`RepoDeactivated`, `RepoTakendown`, ...) are + * not absence and stay retryable. Some gateways answer 404 instead, which is + * treated as unconfirmed absence. The body is bounded by safeFetch first. */ export async function readPdsRecordJsonResponse(response: Response): Promise> { if (response.ok) return { kind: 'present', value: (await response.json()) as T } - if (response.status === 404) { - void response.body?.cancel().catch(() => undefined) - return { kind: 'absent' } - } - if (response.status === 400) { + if (response.status === 400 || response.status === 404) { let errorBody: unknown try { errorBody = await response.json() } catch { // Preserve the typed HTTP error for malformed or oversized bodies. } - if ( - typeof errorBody === 'object' && - errorBody !== null && - (errorBody as { error?: unknown }).error === 'RecordNotFound' - ) { - return { kind: 'absent' } - } + if (isRecordNotFoundBody(errorBody)) return { kind: 'absent', confirmed: true } + if (response.status === 404) return { kind: 'absent', confirmed: false } } void response.body?.cancel().catch(() => undefined) throw new SafeFetchHttpError(response) @@ -369,14 +370,23 @@ export class AuthoritativeSiteRecordError extends Error { /** * Read current PDS state without conflating absence with transport or validation - * failure. Only a typed HTTP 404 is authoritative absence; every other failure - * is retryable by the caller. + * failure. Absence is the PDS's HTTP 400 `RecordNotFound` (or a gateway 404); + * every other failure is retryable by the caller. */ export async function fetchAuthoritativeSiteRecord( did: string, rkey: string, resources?: RevalidationResources, ): Promise { + const outcome = await lookupAuthoritativeSiteRecord(did, rkey, resources) + return outcome.kind === 'present' ? { record: outcome.record, cid: outcome.cid } : null +} + +async function lookupAuthoritativeSiteRecord( + did: string, + rkey: string, + resources?: RevalidationResources, +): Promise<{ kind: 'present'; record: WispFsRecord; cid: string } | { kind: 'absent'; confirmed: boolean }> { assertRevalidationActive(resources) const resolvedPdsEndpoint = await resolvePdsEndpoint(did, resources) const pdsEndpoint = resolvedPdsEndpoint ? rewritePdsEndpoint(resolvedPdsEndpoint) : null @@ -387,7 +397,7 @@ export async function fetchAuthoritativeSiteRecord( `${pdsEndpoint}/xrpc/com.atproto.repo.getRecord?${query.toString()}`, resources, ) - if (response.kind === 'absent') return null + if (response.kind === 'absent') return response const data = response.value const record = parseLexiconJson(data.value) @@ -395,23 +405,26 @@ export async function fetchAuthoritativeSiteRecord( if (typeof data.cid !== 'string' || data.cid.length === 0 || data.cid.length > 256) { throw new AuthoritativeSiteRecordError('MISSING_CID') } - return { record, cid: data.cid } + return { kind: 'present', record, cid: data.cid } } export type SiteRecordFetchOutcome = | { kind: 'present'; record: WispFsRecord; cid: string } - | { kind: 'absent' } + /** `confirmed`: the PDS answered `RecordNotFound`, not just a bare 404. */ + | { kind: 'absent'; confirmed?: boolean } | { kind: 'retryable'; error: 'PDS_UNRESOLVED' | 'INVALID_RECORD' | 'MISSING_CID' | 'FETCH_FAILED' } -/** A revalidation-safe PDS lookup. Only an actual HTTP 404 is `absent`. */ +/** + * A revalidation-safe PDS lookup. `absent` means the PDS answered HTTP 400 + * `RecordNotFound` (confirmed) or a gateway answered 404 (unconfirmed). + */ export async function fetchSiteRecordOutcome( did: string, rkey: string, resources?: RevalidationResources, ): Promise { try { - const result = await fetchAuthoritativeSiteRecord(did, rkey, resources) - return result ? { kind: 'present', ...result } : { kind: 'absent' } + return await lookupAuthoritativeSiteRecord(did, rkey, resources) } catch (error) { if (error instanceof AuthoritativeSiteRecordError) return { kind: 'retryable', error: error.code } return { kind: 'retryable', error: 'FETCH_FAILED' } diff --git a/apps/firehose-service/src/lib/db.ts b/apps/firehose-service/src/lib/db.ts index 765f2b4..bab3d04 100644 --- a/apps/firehose-service/src/lib/db.ts +++ b/apps/firehose-service/src/lib/db.ts @@ -971,7 +971,9 @@ export async function upsertSiteCache( record_cid = EXCLUDED.record_cid, file_cids = EXCLUDED.file_cids, updated_at = EXTRACT(EPOCH FROM NOW()), - cold_synced = EXCLUDED.cold_synced + cold_synced = EXCLUDED.cold_synced, + absent_since = NULL, + absent_checks = 0 ` logger.debug(`[DB] upsertSiteCache completed for ${did}/${rkey}`) } catch (err) { @@ -980,6 +982,51 @@ export async function upsertSiteCache( } } +export interface AbsentSiteMark { + did: string + rkey: string + absent_since: number + absent_checks: number +} + +/** + * Record one confirmed `RecordNotFound` for a cached site. The first mark + * stops hosting from serving it; files stay until the sweeper's grace period + * passes. Deleted tombstone rows are never marked. Returns null when there is + * no live row to mark. + */ +export async function markSiteAbsent(did: string, rkey: string): Promise { + const rows = await sql` + UPDATE site_cache + SET absent_since = COALESCE(absent_since, EXTRACT(EPOCH FROM NOW())::bigint), + absent_checks = absent_checks + 1 + WHERE did = ${did} AND rkey = ${rkey} AND record_cid <> ${DELETED_SITE_RECORD_CID} + RETURNING did, rkey, absent_since, absent_checks + ` + return rows[0] ?? null +} + +/** The record is back: serve the cached site again. */ +export async function clearSiteAbsent(did: string, rkey: string): Promise { + const rows = await sql` + UPDATE site_cache SET absent_since = NULL, absent_checks = 0 + WHERE did = ${did} AND rkey = ${rkey} AND absent_since IS NOT NULL + RETURNING did + ` + return rows.length > 0 +} + +/** Marked sites, oldest first, for the absent-site sweeper. */ +export async function listAbsentSites(limit: number): Promise { + return await sql` + SELECT did, rkey, absent_since, absent_checks + FROM site_cache + WHERE absent_since IS NOT NULL AND record_cid <> ${DELETED_SITE_RECORD_CID} + ORDER BY absent_since ASC + LIMIT ${limit} + ` +} + /** Keep an empty durable manifest so hosting can distinguish a confirmed delete from a projection/storage outage. */ export async function markSiteCacheDeleted(did: string, rkey: string): Promise { await upsertSiteCache(did, rkey, DELETED_SITE_RECORD_CID, {}) diff --git a/apps/firehose-service/src/lib/revalidate-worker.test.ts b/apps/firehose-service/src/lib/revalidate-worker.test.ts index 109bc6c..18cc4d9 100644 --- a/apps/firehose-service/src/lib/revalidate-worker.test.ts +++ b/apps/firehose-service/src/lib/revalidate-worker.test.ts @@ -333,6 +333,60 @@ describe('shouldSkipInvalidationForReason', () => { }) }) +describe('absent site marking', () => { + function markingDependencies( + outcome: Awaited>>, + ) { + const marks: string[] = [] + const invalidations: string[] = [] + const deletes: string[] = [] + const dependencies = createDependencies({ + fetchSiteRecordOutcome: async () => outcome, + markSiteAbsent: async (did, rkey) => { + marks.push(`${did}/${rkey}`) + return { did, rkey, absent_since: 1, absent_checks: 1 } + }, + publishCacheInvalidation: async (did, rkey, action) => { + invalidations.push(`${did}/${rkey}:${action}`) + }, + handleSiteDelete: async (did, rkey) => { + deletes.push(`${did}/${rkey}`) + }, + }) + return { dependencies, marks, invalidations, deletes } + } + + test('marks a site the PDS confirms RecordNotFound, stops serving it, and deletes nothing', async () => { + const { redis, acks } = createRedis() + const { dependencies, marks, invalidations, deletes } = markingDependencies({ kind: 'absent', confirmed: true }) + + await processRevalidationMessage(MESSAGE_ID, messageFields('storage-miss:index.html'), redis, dependencies) + + expect(marks).toEqual([`${DID}/${RKEY}`]) + expect(invalidations).toEqual([`${DID}/${RKEY}:update`]) + expect(deletes).toEqual([]) + expect(acks).toEqual([MESSAGE_ID]) + }) + + test('does not mark on a bare gateway 404', async () => { + const { redis } = createRedis() + const { dependencies, marks } = markingDependencies({ kind: 'absent', confirmed: false }) + + await processRevalidationMessage(MESSAGE_ID, messageFields('storage-miss:index.html'), redis, dependencies) + + expect(marks).toEqual([]) + }) + + test('does not mark when the PDS is unreachable', async () => { + const { redis } = createRedis() + const { dependencies, marks } = markingDependencies({ kind: 'retryable', error: 'FETCH_FAILED' }) + + await processRevalidationMessage(MESSAGE_ID, messageFields('storage-miss:index.html'), redis, dependencies) + + expect(marks).toEqual([]) + }) +}) + describe('delete tombstone revalidation', () => { test('deletes a missing PDS record and ACKs only after delete succeeds', async () => { const { redis, acks, streamDeletes } = createRedis(60) diff --git a/apps/firehose-service/src/lib/revalidate-worker.ts b/apps/firehose-service/src/lib/revalidate-worker.ts index 1dee8ea..c3a5e73 100644 --- a/apps/firehose-service/src/lib/revalidate-worker.ts +++ b/apps/firehose-service/src/lib/revalidate-worker.ts @@ -6,7 +6,9 @@ import { SafeFetchHttpError } from '@wispplace/safe-fetch' import { DecompressionLimitError } from '@wispplace/tiered-storage' import Redis from 'ioredis' import { config } from '../config' +import { startAbsentSiteSweeper, stopAbsentSiteSweeper } from './absent-site-sweeper' import { BlobIntegrityError } from './blob-integrity' +import { publishCacheInvalidation } from './cache-invalidation' import { AuthoritativeSettingsRecordError, AuthoritativeSiteRecordError, @@ -22,6 +24,7 @@ import { SiteBlobBackoffError, SiteLogicalQuotaExceededError, } from './cache-writer' +import { markSiteAbsent } from './db' import { isSettingsFailureRevalidationReason, isSiteDeleteTombstoneReason, @@ -388,6 +391,9 @@ export interface RevalidateWorkerDependencies { handleSettingsUpdate: typeof handleSettingsUpdate handleSiteCreateOrUpdate: typeof handleSiteCreateOrUpdate handleSiteDelete: typeof handleSiteDelete + /** Mark a cached site absent after a confirmed RecordNotFound. */ + markSiteAbsent?: typeof markSiteAbsent + publishCacheInvalidation?: typeof publishCacheInvalidation } export interface RevalidateRedisPipeline { @@ -494,6 +500,8 @@ const defaultRevalidateWorkerDependencies: RevalidateWorkerDependencies = { handleSettingsUpdate, handleSiteCreateOrUpdate, handleSiteDelete, + markSiteAbsent, + publishCacheInvalidation, } /** @@ -798,9 +806,20 @@ async function handleMissingSiteRecord( redisClient: RevalidateRedisClient, dependencies: RevalidateWorkerDependencies, resources?: RevalidationResourceContext, + confirmed = false, ): Promise { if (!isDeleteTombstone) { - logger.warn(`[Revalidate] Site record not found on PDS: ${did}/${rkey}`) + logger.warn(`[Revalidate] Site record not found on PDS: ${did}/${rkey}`, { confirmed }) + // Only the owner's current PDS saying RecordNotFound marks a site. A missed + // delete event otherwise leaves it served forever; unreachable or ambiguous + // PDS answers never get here as `confirmed`, so an away account is safe. + if (confirmed && dependencies.markSiteAbsent) { + const mark = await dependencies.markSiteAbsent(did, rkey) + if (mark) { + logger.info(`[Revalidate] Marked site absent: ${did}/${rkey}`, { absentChecks: mark.absent_checks }) + await dependencies.publishCacheInvalidation?.(did, rkey, 'update') + } + } assertRevalidationActive(resources) await acknowledgeCompletedMessage(redisClient, id) return @@ -868,7 +887,16 @@ async function prepareSiteRevalidation( } if (outcome.kind === 'absent') { if (verifiedRepair) throw new RevalidationProcessingError('REPAIR_SOURCE_ABSENT', 'permanent') - await handleMissingSiteRecord(id, did, rkey, isDeleteTombstone, redisClient, dependencies, resources) + await handleMissingSiteRecord( + id, + did, + rkey, + isDeleteTombstone, + redisClient, + dependencies, + resources, + 'confirmed' in outcome && outcome.confirmed === true, + ) return null } const record = { record: outcome.record, cid: outcome.cid } @@ -1704,6 +1732,8 @@ function startRevalidateWorkerWithFactory( export async function startRevalidateWorker(): Promise { startRevalidateWorkerWithFactory(config.redisUrl, defaultRedisClientFactory, revalidateWorkerRuntimeConfig) + // Leader-only like this worker, so two nodes never sweep the same site. + startAbsentSiteSweeper() } export function startRevalidateWorkerForTests( @@ -1767,6 +1797,7 @@ export async function stopRevalidateWorker( ): Promise { running = false cancelLoopRetryWait?.() + const sweeperStopped = stopAbsentSiteSweeper() // Abort active PDS/blob work before disconnecting Redis. The resource context // treats this as lifecycle cancellation, not a delivery failure, so the PEL @@ -1783,7 +1814,10 @@ export async function stopRevalidateWorker( clientToClose?.disconnect() const gracePeriodMs = options.gracePeriodMs ?? config.firehoseDrainGraceMs - const stopped = await settlesWithinWorkerGrace(loopToWait, gracePeriodMs) + const stopped = await settlesWithinWorkerGrace( + Promise.all([loopToWait, sweeperStopped]).then(() => undefined), + gracePeriodMs, + ) if (!stopped) return { stopped: false, forced: true } if (workerGeneration === generation && workerAbortController === abortController) { workerAbortController = null diff --git a/apps/hosting-service/src/lib/db.ts b/apps/hosting-service/src/lib/db.ts index 2655d7c..9701d52 100644 --- a/apps/hosting-service/src/lib/db.ts +++ b/apps/hosting-service/src/lib/db.ts @@ -131,7 +131,7 @@ export async function getSiteCache(did: string, rkey: string): Promise { const result = await sql` - SELECT did, rkey, record_cid, file_cids, cached_at, updated_at + SELECT did, rkey, record_cid, file_cids, cached_at, updated_at, absent_since FROM site_cache WHERE did = ${did} AND rkey = ${rkey} LIMIT 1 diff --git a/apps/hosting-service/src/lib/file-serving.integration.test.ts b/apps/hosting-service/src/lib/file-serving.integration.test.ts index 56fa2cf..796c35b 100644 --- a/apps/hosting-service/src/lib/file-serving.integration.test.ts +++ b/apps/hosting-service/src/lib/file-serving.integration.test.ts @@ -47,6 +47,7 @@ let evictPublicCacheKeyStarted: (() => void) | null = null let legacyMetadataHealResult: boolean | Error = true const legacyMetadataHealCalls: Array<{ key: string; checksum: string; sourceCid: string }> = [] let siteFileCids: Record | null = null +let siteAbsentSince: number | null = null let gatedStorageReadKey: string | null = null let gatedStorageReadStarted: (() => void) | null = null let gatedStorageReadGate: Promise | null = null @@ -219,6 +220,7 @@ mock.module('./db', () => ({ file_cids: siteFileCids, cached_at: 0, updated_at: 0, + absent_since: siteAbsentSince, } : null, getSiteSettingsCache: async () => null, @@ -314,6 +316,7 @@ function resetServingState() { legacyMetadataHealResult = true legacyMetadataHealCalls.length = 0 siteFileCids = null + siteAbsentSince = null gatedStorageReadKey = null gatedStorageReadStarted = null gatedStorageReadGate = null @@ -392,6 +395,20 @@ describe('serveFileInternal directory-index fallback for extensioned paths', () expect(await response.text()).toContain('visible.txt') }) + test('serves a site marked absent like a deleted one, without leaking its files', async () => { + siteFileCids = { 'visible.txt': 'visible-cid' } + siteAbsentSince = 1_790_000_000 + + const response = await serveFileInternal(DID, RKEY, 'missing.md', { + $type: 'place.wisp.settings', + directoryListing: true, + cleanUrls: false, + }) + + expect(response.status).toBe(404) + expect(await response.text()).not.toContain('visible.txt') + }) + test('does not re-probe failed direct and index candidates after directory fallback', async () => { const response = await serveFileInternal(DID, RKEY, 'missing') diff --git a/apps/hosting-service/src/lib/file-serving.ts b/apps/hosting-service/src/lib/file-serving.ts index 4999c69..5128161 100644 --- a/apps/hosting-service/src/lib/file-serving.ts +++ b/apps/hosting-service/src/lib/file-serving.ts @@ -384,6 +384,9 @@ async function getExpectedFileCidsForSite( ): Promise | null> { const siteCache = await span(trace, 'db:siteCache', () => getSiteCache(did, rkey)) if (!siteCache) return null + // The owner's PDS confirmed the record is gone: serve it like a deleted site + // (an empty manifest, so every path 404s) while the files await the sweeper. + if (siteCache.absent_since !== null && siteCache.absent_since !== undefined) return {} return normalizeFileCids(siteCache.file_cids).value } diff --git a/apps/main-app/src/lib/migrations.ts b/apps/main-app/src/lib/migrations.ts index e259d86..37be03c 100644 --- a/apps/main-app/src/lib/migrations.ts +++ b/apps/main-app/src/lib/migrations.ts @@ -405,6 +405,15 @@ export const runDatabaseMigrations = async (primaryDb: SQL): Promise => { await db`ALTER TABLE site_cache ADD COLUMN IF NOT EXISTS cold_synced BOOLEAN NOT NULL DEFAULT true` }) + // Set when the owner's current PDS answers RecordNotFound for a cached site. + // Hosting stops serving marked rows; the firehose sweeper purges cached files + // only after repeated confirmations over a grace period. Never touches domains. + await runMigration('add site_cache.absent_since', async () => { + await db`ALTER TABLE site_cache ADD COLUMN IF NOT EXISTS absent_since BIGINT` + await db`ALTER TABLE site_cache ADD COLUMN IF NOT EXISTS absent_checks INTEGER NOT NULL DEFAULT 0` + await db`CREATE INDEX IF NOT EXISTS idx_site_cache_absent_since ON site_cache(absent_since) WHERE absent_since IS NOT NULL` + }) + // Remove the unique constraint on domains.did to allow multiple domains per user await runMigration('drop legacy domains_did_key', async () => { await db`ALTER TABLE domains DROP CONSTRAINT IF EXISTS domains_did_key` diff --git a/packages/@wispplace/database/src/types.ts b/packages/@wispplace/database/src/types.ts index 984f593..bfbc1de 100644 --- a/packages/@wispplace/database/src/types.ts +++ b/packages/@wispplace/database/src/types.ts @@ -72,6 +72,9 @@ export interface SiteCache { // source of truth and the firehose must do a full (re)download rather than // trusting record_cid/file_cids as a proxy for "already in S3". cold_synced?: boolean + // Epoch seconds when the owner's PDS first confirmed RecordNotFound for this + // site. Non-null means hosting must not serve it (see the absent-site sweeper). + absent_since?: number | string | null } /** -- 2.54.0