diff --git a/apps/hosting-service/src/index.ts b/apps/hosting-service/src/index.ts index 39cdb9e..b1550c2 100644 --- a/apps/hosting-service/src/index.ts +++ b/apps/hosting-service/src/index.ts @@ -11,6 +11,7 @@ import { closeDatabase } from './lib/db' import { closePrivateSitesDatabase } from './lib/private-sites-db' import { closeRevalidateQueue } from './lib/revalidate-queue' import { getStorageConfig, storage } from './lib/storage' +import { unavailableLog } from './lib/unavailable-log' import app from './server' import { onceAsync, stopHttpServerWithGrace } from './shutdown' @@ -52,6 +53,9 @@ if (!existsSync(CACHE_DIR)) { // half-hour sweep left expired site manifests resident for up to 30 minutes. cache.startCleanup(CACHE_SWEEP_INTERVAL_MS) +// Summarize fail-closed 503s once a window instead of logging each request. +unavailableLog.start() + // Start cache invalidation subscriber (listens for firehose-service updates via Redis pub/sub) startCacheInvalidationSubscriber() @@ -107,6 +111,8 @@ const shutdown = onceAsync(async (signal: 'SIGINT' | 'SIGTERM') => { if (httpStop.forceStopFailed) logger.error('HTTP server force stop failed') cache.stopCleanup() + // Flush the partial window while the log exporters are still running. + unavailableLog.stop() const tasks = [ { name: 'cache invalidation subscriber', promise: stopCacheInvalidationSubscriber() }, { name: 'revalidation queue', promise: closeRevalidateQueue() }, 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 1d190e0..1b239a1 100644 --- a/apps/hosting-service/src/lib/file-serving.integration.test.ts +++ b/apps/hosting-service/src/lib/file-serving.integration.test.ts @@ -1,4 +1,4 @@ -import { afterEach, beforeEach, describe, expect, mock, test } from 'bun:test' +import { afterEach, beforeEach, describe, expect, mock, spyOn, test } from 'bun:test' import { gunzipSync, gzipSync } from 'node:zlib' import { computeCID } from '@wispplace/atproto-utils' import { MAX_BLOB_SIZE } from '@wispplace/constants' @@ -274,7 +274,10 @@ mock.module('./revalidate-queue', () => ({ })) const { cache } = await import('./cache-manager') const { resetHtmlHotCacheWarmupForTests } = await import('./html-prewarm') -const { applyCacheInvalidationForTests } = await import('./cache-invalidation') +const { applyCacheInvalidationForTests, markSiteUpdating, resetUpdatingSitesForTests } = await import( + './cache-invalidation' +) +const { unavailableLog } = await import('./unavailable-log') const { serveFileInternal, serveFileInternalWithRewrite, @@ -1611,3 +1614,100 @@ describe('manifest source CID validation', () => { expect(revalidateCalls).toEqual([{ did: DID, rkey: RKEY, reason: 'storage-miss:mismatched.txt' }]) }) }) + +describe('fail-closed 503 visibility', () => { + beforeEach(() => { + resetServingState() + resetUpdatingSitesForTests() + }) + afterEach(() => { + resetUpdatingSitesForTests() + mock.restore() + }) + + async function failClosedCauses(serve: () => Promise) { + const record = spyOn(unavailableLog, 'record').mockImplementation(() => {}) + try { + expect((await serve()).status).toBe(503) + return [...record.mock.calls] + } finally { + record.mockRestore() + } + } + + const serve = (path: string, headers?: Record) => + serveFromCache(DID, RKEY, path, `https://example.com/${path}`, headers) + + test('names a site that is mid-update, for the HTML page too', async () => { + siteFileCids = { 'index.html': 'cid' } + markSiteUpdating(DID, RKEY) + + expect(await failClosedCauses(() => serve('index.html'))).toEqual([[DID, RKEY, 'updating', 'none']]) + expect(await failClosedCauses(() => serve('index.html', { accept: 'text/html' }))).toEqual([ + [DID, RKEY, 'updating', 'none'], + ]) + }) + + test('names a missing manifest with its repair outcome', async () => { + expect(await failClosedCauses(() => serve('index.html'))).toEqual([[DID, RKEY, 'manifest-miss', 'enqueued']]) + }) + + test('names a manifest file absent from every tier', async () => { + siteFileCids = { 'missing.txt': 'expected-cid' } + expect(await failClosedCauses(() => serve('missing.txt'))).toEqual([[DID, RKEY, 'cid-miss', 'enqueued']]) + }) + + test('names a source CID mismatch, including behind a redirect 404', async () => { + siteFileCids = { _redirects: 'redirects-cid', 'stale.txt': 'expected-cid' } + storeFile('_redirects', '/old /stale.txt 404', 'text/plain') + const staleReads = (): FakeReadOutcome[] => [ + { data: new TextEncoder().encode('warm'), source: 'warm', customMetadata: { sourceCid: 'old-cid' } }, + { data: new TextEncoder().encode('cold'), source: 'cold', customMetadata: { sourceCid: 'older-cid' } }, + ] + + queueStorageReads('stale.txt', staleReads()) + expect(await failClosedCauses(() => serve('stale.txt'))).toEqual([[DID, RKEY, 'cid-mismatch', 'enqueued']]) + + cache.clear('sourceCidMismatches') + queueStorageReads('stale.txt', staleReads()) + expect(await failClosedCauses(() => serve('old'))).toEqual([[DID, RKEY, 'cid-mismatch', 'enqueued']]) + }) + + test('names a storage outage on a file read and on a _redirects read', async () => { + siteFileCids = { 'unavailable.txt': 'expected-cid' } + queueStorageReads('unavailable.txt', [new TestStorageUnavailableError('getWithMetadata', 'timeout')]) + expect(await failClosedCauses(() => serve('unavailable.txt'))).toEqual([[DID, RKEY, 'storage-unavailable', 'none']]) + + siteFileCids = { _redirects: 'redirects-cid' } + failStorageGet('_redirects', new TestStorageUnavailableError('get', 'timeout')) + expect(await failClosedCauses(() => serve('old'))).toEqual([[DID, RKEY, 'storage-unavailable', 'none']]) + }) + + test('records nothing for responses that are not fail-closed', async () => { + siteFileCids = { 'index.html': 'cid-of-index.html' } + storeFile('index.html', 'ok') + const record = spyOn(unavailableLog, 'record') + + expect((await serve('index.html')).status).toBe(200) + expect((await serve('nope.txt')).status).toBe(404) + expect(record).not.toHaveBeenCalled() + }) + + test('logs a burst of 1000 failures for one site as one line', async () => { + const info = spyOn(console, 'info').mockImplementation(() => {}) + unavailableLog.flush() + info.mockClear() + + for (let i = 0; i < 1000; i++) await serve('index.html') + expect(info).not.toHaveBeenCalled() + + expect(unavailableLog.flush()).toBe(1) + const lines = info.mock.calls.map(([line]) => String(line)).filter((line) => line.includes('[FailClosed]')) + expect(lines).toHaveLength(1) + expect(lines[0]).toContain(`"site":"${DID}/${RKEY}"`) + expect(lines[0]).toContain('"reason":"manifest-miss"') + expect(lines[0]).toContain('"count":1000') + expect(lines[0]).toContain('"revalidate":{"enqueued":1000}') + expect(lines[0]).not.toContain('index.html') + }) +}) diff --git a/apps/hosting-service/src/lib/file-serving.ts b/apps/hosting-service/src/lib/file-serving.ts index ac45ab6..534193d 100644 --- a/apps/hosting-service/src/lib/file-serving.ts +++ b/apps/hosting-service/src/lib/file-serving.ts @@ -43,6 +43,7 @@ import { enqueueRevalidate } from './revalidate-queue' import { resolveStorageKey, type SiteManifest } from './site-storage-keys' import { addPublicSourceCidIfChecksumMatches, evictPublicCacheKey, isStorageUnavailableError, storage } from './storage' import { createTrace, logTrace, type RequestTrace, span } from './trace' +import { type RevalidateOutcome, type UnavailableReason, unavailableLog } from './unavailable-log' import { getCachedSettings } from './utils' const logger = createLogger('file-serving') @@ -426,28 +427,40 @@ function shouldServeUpdatingPage(requestHeaders?: Record): boole return fetchDest === 'document' || fetchDest === 'iframe' || fetchDest === 'frame' } -function buildUnavailableResponse(message: string, requestHeaders?: Record): Response { - if (shouldServeUpdatingPage(requestHeaders)) { - return siteUpdatingResponse() - } +type UnavailableCause = { reason: UnavailableReason; revalidate: RevalidateOutcome } - return new Response(message, { - status: 503, - headers: { - 'Cache-Control': 'no-store', - 'Retry-After': '5', - }, - }) +// Why each fail-closed 503 was built. Only the response a request actually +// returns is counted, so an intermediate 503 that a caller discards is not. +const unavailableCauses = new WeakMap() + +function buildUnavailableResponse( + message: string, + cause: UnavailableCause, + requestHeaders?: Record, +): Response { + const response = shouldServeUpdatingPage(requestHeaders) + ? siteUpdatingResponse() + : new Response(message, { + status: 503, + headers: { + 'Cache-Control': 'no-store', + 'Retry-After': '5', + }, + }) + unavailableCauses.set(response, cause) + return response } function buildSiteUpdatingResponse(requestHeaders?: Record): Response { - return buildUnavailableResponse('Site is updating', requestHeaders) + return buildUnavailableResponse('Site is updating', { reason: 'updating', revalidate: 'none' }, requestHeaders) } -function buildStorageMissResponse(requestHeaders?: Record): Response { - return buildUnavailableResponse('Storage temporarily unavailable', requestHeaders) +function buildStorageMissResponse(cause: UnavailableCause, requestHeaders?: Record): Response { + return buildUnavailableResponse('Storage temporarily unavailable', cause, requestHeaders) } +const STORAGE_UNAVAILABLE: UnavailableCause = { reason: 'storage-unavailable', revalidate: 'none' } + type DirectoryEntryMap = Map function directoryPrefix(requestPath: string): string { @@ -1370,8 +1383,8 @@ function createExpectedManifestMissTracker( async response(): Promise { if (!expectedMissPath) return null recordStorageMiss(expectedMissPath) - await enqueueRevalidate(did, rkey, `storage-miss:${expectedMissPath}`) - return buildStorageMissResponse(requestHeaders) + const { result } = await enqueueRevalidate(did, rkey, `storage-miss:${expectedMissPath}`) + return buildStorageMissResponse({ reason: 'cid-miss', revalidate: result }, requestHeaders) }, } } @@ -1700,8 +1713,8 @@ async function serveRedirectResponse(options: RedirectRequestOptions): Promise { const sourcePath = sourceManifestPath(filePath) recordStorageMiss(sourcePath) - await enqueueRevalidate(options.did, options.rkey, `storage-miss:${sourcePath}`) - return buildStorageMissResponse(options.requestHeaders) + const { result } = await enqueueRevalidate(options.did, options.rkey, `storage-miss:${sourcePath}`) + return buildStorageMissResponse({ reason: 'cid-mismatch', revalidate: result }, options.requestHeaders) } async function serveFileRequest(options: FileRequestOptions): Promise { @@ -1709,7 +1722,7 @@ async function serveFileRequest(options: FileRequestOptions): Promise return await resolveFileRequest(options) } catch (error) { if (isStorageUnavailableError(error)) { - return buildStorageMissResponse(options.requestHeaders) + return buildStorageMissResponse(STORAGE_UNAVAILABLE, options.requestHeaders) } if (error instanceof SourceCidValidationError) { return await buildSourceCidStorageMissResponse(options, error.filePath) @@ -1754,8 +1767,8 @@ async function resolveCachedRequest(options: CachedRequestOptions, trace: Reques const manifest = await getSiteManifest(did, rkey, trace) if (manifest === null) { recordStorageMiss('manifest') - await enqueueRevalidate(did, rkey, 'storage-miss:manifest') - return buildStorageMissResponse(requestHeaders) + const { result } = await enqueueRevalidate(did, rkey, 'storage-miss:manifest') + return buildStorageMissResponse({ reason: 'manifest-miss', revalidate: result }, requestHeaders) } const settings = await span(trace, 'db:settings', () => getCachedSettings(did, rkey)) const indexFiles = getIndexFiles(settings) @@ -1788,7 +1801,7 @@ async function resolveCachedRequest(options: CachedRequestOptions, trace: Reques return response } -async function serveCachedRequest(options: CachedRequestOptions): Promise { +async function resolveCachedResponse(options: CachedRequestOptions): Promise { const { did, filePath, requestHeaders, rkey } = options if (isSiteUpdating(did, rkey)) { return buildSiteUpdatingResponse(requestHeaders) @@ -1799,12 +1812,19 @@ async function serveCachedRequest(options: CachedRequestOptions): Promise { + const response = await resolveCachedResponse(options) + const cause = unavailableCauses.get(response) + if (cause) unavailableLog.record(options.did, options.rkey, cause.reason, cause.revalidate) + return response +} + /** * Serve files for custom domains and subdomains. */ diff --git a/apps/hosting-service/src/lib/unavailable-log.test.ts b/apps/hosting-service/src/lib/unavailable-log.test.ts new file mode 100644 index 0000000..cc09e4e --- /dev/null +++ b/apps/hosting-service/src/lib/unavailable-log.test.ts @@ -0,0 +1,97 @@ +import { describe, expect, jest, test } from 'bun:test' +import { createUnavailableLog, OVERFLOW_SITE, type UnavailableSummary } from './unavailable-log' + +const DID = 'did:plc:test' + +function collectingLog(options: { maxSites?: number; now?: () => number } = {}) { + const lines: UnavailableSummary[] = [] + const log = createUnavailableLog({ ...options, emit: (summary) => lines.push(summary) }) + return { log, lines } +} + +describe('fail-closed 503 summary log', () => { + test('folds 1000 failures of one site and reason into one line per window', () => { + let clock = Date.parse('2026-10-08T00:18:00.000Z') + const { log, lines } = collectingLog({ now: () => clock++ }) + + for (let i = 0; i < 1000; i++) { + log.record(DID, 'blog', 'cid-mismatch', i === 0 ? 'enqueued' : 'deduped') + } + expect(lines).toHaveLength(0) + + expect(log.flush()).toBe(1) + expect(lines).toEqual([ + { + site: `${DID}/blog`, + reason: 'cid-mismatch', + count: 1000, + firstAt: '2026-10-08T00:18:00.000Z', + lastAt: '2026-10-08T00:18:00.999Z', + revalidate: { enqueued: 1, deduped: 999 }, + }, + ]) + }) + + test('keeps one line per reason and starts a fresh window after each flush', () => { + const { log, lines } = collectingLog() + log.record(DID, 'blog', 'updating', 'none') + log.record(DID, 'blog', 'storage-unavailable', 'none') + log.record(DID, 'docs', 'updating', 'none') + expect(log.flush()).toBe(3) + expect(lines.map(({ site, reason, count }) => [site, reason, count])).toEqual([ + [`${DID}/blog`, 'updating', 1], + [`${DID}/blog`, 'storage-unavailable', 1], + [`${DID}/docs`, 'updating', 1], + ]) + + expect(log.flush()).toBe(0) + expect(lines).toHaveLength(3) + }) + + test('logs nothing when no request failed closed', () => { + const { log, lines } = collectingLog() + expect(log.flush()).toBe(0) + expect(log.stop()).toBe(0) + expect(lines).toEqual([]) + }) + + test('caps distinct sites per window and counts the rest under other', () => { + const { log, lines } = collectingLog({ maxSites: 3 }) + for (let site = 0; site < 10_000; site++) log.record(DID, `site-${site}`, 'cid-miss', 'deduped') + // Sites already tracked keep their own bucket after the cap is reached. + log.record(DID, 'site-0', 'cid-miss', 'deduped') + + expect(log.trackedSites()).toBe(4) + log.flush() + expect(lines.map(({ site, count }) => [site, count])).toEqual([ + [`${DID}/site-0`, 2], + [`${DID}/site-1`, 1], + [`${DID}/site-2`, 1], + [OVERFLOW_SITE, 9_997], + ]) + }) + + test('flushes on its window timer and once more on stop', () => { + jest.useFakeTimers() + try { + const lines: UnavailableSummary[] = [] + const log = createUnavailableLog({ windowMs: 60_000, emit: (summary) => lines.push(summary) }) + log.start() + log.start() + + log.record(DID, 'blog', 'manifest-miss', 'enqueued') + jest.advanceTimersByTime(60_000) + expect(lines.map(({ count }) => count)).toEqual([1]) + + log.record(DID, 'blog', 'manifest-miss', 'deduped') + expect(log.stop()).toBe(1) + expect(lines.map(({ count }) => count)).toEqual([1, 1]) + + log.record(DID, 'blog', 'manifest-miss', 'deduped') + jest.advanceTimersByTime(120_000) + expect(lines).toHaveLength(2) + } finally { + jest.useRealTimers() + } + }) +}) diff --git a/apps/hosting-service/src/lib/unavailable-log.ts b/apps/hosting-service/src/lib/unavailable-log.ts new file mode 100644 index 0000000..a97a504 --- /dev/null +++ b/apps/hosting-service/src/lib/unavailable-log.ts @@ -0,0 +1,118 @@ +import { createLogger } from '@wispplace/observability' +import type { EnqueueResult } from './revalidate-queue' + +/** + * Aggregated visibility for hosting's deliberate fail-closed 503s. + * + * Each 503 is counted under its site and reason, and one summary line per + * (site, reason) is logged when the window flushes, never one per request. + * Distinct sites per window are capped; the rest are counted under `other`. + * Lines carry the site's did/rkey and nothing from the request itself. + */ + +export type UnavailableReason = 'updating' | 'manifest-miss' | 'cid-miss' | 'cid-mismatch' | 'storage-unavailable' + +/** What the 503 did about repair: the queue's answer, or `none` when it does not request one. */ +export type RevalidateOutcome = EnqueueResult | 'none' + +export type UnavailableSummary = { + site: string + reason: UnavailableReason + count: number + firstAt: string + lastAt: string + revalidate: Partial> +} + +type Bucket = Omit & { firstAt: number; lastAt: number } + +export type UnavailableLogOptions = { + maxSites?: number + windowMs?: number + emit?: (summary: UnavailableSummary) => void + now?: () => number +} + +const DEFAULT_WINDOW_MS = 60_000 +const DEFAULT_MAX_SITES = 500 +export const OVERFLOW_SITE = 'other' + +const logger = createLogger('hosting-service') + +function logSummary(summary: UnavailableSummary): void { + logger.info('[FailClosed] 503 summary', summary) +} + +function toSummary(bucket: Bucket): UnavailableSummary { + return { + site: bucket.site, + reason: bucket.reason, + count: bucket.count, + firstAt: new Date(bucket.firstAt).toISOString(), + lastAt: new Date(bucket.lastAt).toISOString(), + revalidate: bucket.revalidate, + } +} + +export function createUnavailableLog(options: UnavailableLogOptions = {}) { + const maxSites = options.maxSites ?? DEFAULT_MAX_SITES + const windowMs = options.windowMs ?? DEFAULT_WINDOW_MS + const emit = options.emit ?? logSummary + const now = options.now ?? Date.now + // Keyed by site then reason; the reason set is closed, so the cap on sites bounds the whole window. + let sites = new Map>() + let timer: ReturnType | null = null + + function siteKey(site: string): string { + return sites.has(site) || sites.size < maxSites ? site : OVERFLOW_SITE + } + + function record(did: string, rkey: string, reason: UnavailableReason, revalidate: RevalidateOutcome): void { + const at = now() + const site = siteKey(`${did}/${rkey}`) + let buckets = sites.get(site) + if (!buckets) { + buckets = new Map() + sites.set(site, buckets) + } + const bucket = buckets.get(reason) + if (!bucket) { + buckets.set(reason, { site, reason, count: 1, firstAt: at, lastAt: at, revalidate: { [revalidate]: 1 } }) + return + } + bucket.count++ + bucket.lastAt = at + bucket.revalidate[revalidate] = (bucket.revalidate[revalidate] ?? 0) + 1 + } + + /** Emit one line per (site, reason) seen since the last flush and start a new window. */ + function flush(): number { + const finished = sites + sites = new Map() + let lines = 0 + for (const buckets of finished.values()) { + for (const bucket of buckets.values()) { + emit(toSummary(bucket)) + lines++ + } + } + return lines + } + + function start(): void { + if (timer) return + timer = setInterval(flush, windowMs) + // Visibility must never keep the process alive. + timer.unref?.() + } + + function stop(): number { + if (timer) clearInterval(timer) + timer = null + return flush() + } + + return { record, flush, start, stop, trackedSites: () => sites.size } +} + +export const unavailableLog = createUnavailableLog()