diff --git a/apps/firehose-service/src/lib/firehose.ts b/apps/firehose-service/src/lib/firehose.ts index dae9b34..0d46017 100644 --- a/apps/firehose-service/src/lib/firehose.ts +++ b/apps/firehose-service/src/lib/firehose.ts @@ -24,15 +24,16 @@ const logger = createLogger('firehose-service'); // Track firehose health let lastEventTime = Date.now(); let isConnected = false; -let eventQueue: Array<() => Promise> = []; let activeHandlers = 0; +let queuedHandlers = 0; +const siteQueues = new Map>(); export function getFirehoseHealth() { return { connected: isConnected, lastEventTime, timeSinceLastEvent: Date.now() - lastEventTime, - queueSize: eventQueue.length, + queueSize: queuedHandlers, activeHandlers, healthy: isConnected && (Date.now() - lastEventTime < 60000), }; @@ -55,6 +56,30 @@ async function processWithConcurrencyLimit(handler: () => Promise): Promis } } +/** + * Schedule work so each site (did/rkey) is processed in event order. + * This prevents stale writes when multiple updates arrive quickly. + */ +function scheduleSiteWork(siteKey: string, handler: () => Promise): void { + const previous = siteQueues.get(siteKey) ?? Promise.resolve(); + queuedHandlers++; + + const next = previous + .catch(() => undefined) + .then(() => processWithConcurrencyLimit(handler)) + .catch((err) => { + logger.error(`[firehose] Unhandled site work error for ${siteKey}`, err); + }) + .finally(() => { + queuedHandlers = Math.max(0, queuedHandlers - 1); + if (siteQueues.get(siteKey) === next) { + siteQueues.delete(siteKey); + } + }); + + siteQueues.set(siteKey, next); +} + /** * Handle a firehose event */ @@ -77,7 +102,8 @@ async function handleEvent(evt: Event | CommitEvt): Promise { // Handle place.wisp.fs events if (collection === 'place.wisp.fs') { logger.info(`[place.wisp.fs] Received ${commitEvt.event} event`, { did, rkey, cid: cid?.toString() || 'unknown' }); - processWithConcurrencyLimit(async () => { + const siteKey = `${did}/${rkey}`; + scheduleSiteWork(siteKey, async () => { try { logger.debug(`[place.wisp.fs] Processing ${commitEvt.event} event`, { did, rkey }); if (commitEvt.event === 'delete') { @@ -96,14 +122,13 @@ async function handleEvent(evt: Event | CommitEvt): Promise { } catch (err) { logger.error(`[place.wisp.fs] Error handling event`, err, { did, rkey, event: commitEvt.event }); } - }).catch(err => { - logger.error(`[place.wisp.fs] Error processing event`, err, { did, rkey, event: commitEvt.event }); }); } // Handle place.wisp.settings events if (collection === 'place.wisp.settings') { - processWithConcurrencyLimit(async () => { + const siteKey = `${did}/${rkey}`; + scheduleSiteWork(siteKey, async () => { try { if (commitEvt.event === 'delete') { await handleSettingsDelete(did, rkey); @@ -114,8 +139,6 @@ async function handleEvent(evt: Event | CommitEvt): Promise { } catch (err) { logger.error(`[place.wisp.settings] Error handling event`, err, { did, rkey, event: commitEvt.event }); } - }).catch(err => { - logger.error(`[place.wisp.settings] Error processing event`, err, { did, rkey, event: commitEvt.event }); }); } } catch (err) { diff --git a/apps/hosting-service/src/lib/file-serving.ts b/apps/hosting-service/src/lib/file-serving.ts index 4600469..9c80b93 100644 --- a/apps/hosting-service/src/lib/file-serving.ts +++ b/apps/hosting-service/src/lib/file-serving.ts @@ -10,7 +10,7 @@ import { shouldCompressMimeType } from '@wispplace/atproto-utils/compression'; import { getCachedSettings } from './utils'; import { loadRedirectRules, matchRedirectRule, parseCookies, parseQueryString } from './redirects'; import { isHtmlContent, rewriteHtmlPaths } from './html-rewriter'; -import { generate404Page, generateDirectoryListing } from './page-generators'; +import { generate404Page, generateDirectoryListing, siteUpdatingResponse } from './page-generators'; import { getIndexFiles, applyCustomHeaders } from './request-utils'; import { cache } from './cache-manager'; import { storage } from './storage'; @@ -65,7 +65,21 @@ async function storageExists(did: string, rkey: string, filePath: string): Promi return storage.exists(key); } -function buildStorageMissResponse(): Response { +function shouldServeUpdatingPage(requestHeaders?: Record): boolean { + const accept = (requestHeaders?.accept ?? '').toLowerCase(); + if (accept.includes('text/html') || accept.includes('application/xhtml+xml')) { + return true; + } + + const fetchDest = (requestHeaders?.['sec-fetch-dest'] ?? '').toLowerCase(); + return fetchDest === 'document' || fetchDest === 'iframe' || fetchDest === 'frame'; +} + +function buildStorageMissResponse(requestHeaders?: Record): Response { + if (shouldServeUpdatingPage(requestHeaders)) { + return siteUpdatingResponse(); + } + return new Response('Storage temporarily unavailable', { status: 503, headers: { @@ -368,7 +382,7 @@ export async function serveFileInternal( if (!expectedMissPath) return null; recordStorageMiss(expectedMissPath); await enqueueRevalidate(did, rkey, `storage-miss:${expectedMissPath}`); - return buildStorageMissResponse(); + return buildStorageMissResponse(requestHeaders); }; const indexFiles = getIndexFiles(settings);