/**
* Firehose Service - Ingests AT Protocol firehose events and caches sites to S3
*
* Modes:
* - Normal: Watch firehose for place.wisp.fs events
* - Backfill: Process existing sites discovered from known DIDs
* - DB Fill Only: Legacy mode; no-op now that site_cache is the canonical projection
*/
import { type ServerType, serve } from '@hono/node-server'
import { onceAsync, settleWithTimeout } from '@wispplace/graceful-shutdown'
import { createLogger, initializeGrafanaExporters, shutdownGrafanaExporters } from '@wispplace/observability'
import { observabilityErrorHandler, observabilityMiddleware } from '@wispplace/observability/middleware/hono'
import { safeFetchJson } from '@wispplace/safe-fetch'
import { Hono } from 'hono'
import { config } from './config'
import {
BackfillAbortedError,
runCancellableWindow,
scanHydrantDids,
throwIfBackfillAborted,
} from './lib/backfill-control'
import { closeCacheInvalidationPublisher } from './lib/cache-invalidation'
import { fetchSiteRecord, handleSiteCreateOrUpdate, listSiteRecordsForDid } from './lib/cache-writer'
import { closeDatabase, getSiteCache, listAllKnownDids } from './lib/db'
import {
getActiveService,
getCurrentSeq,
getFirehoseHealth,
startFirehose,
stopAndDrainFirehose,
stopFirehose,
} from './lib/firehose'
import { resolveRevalidationHealth } from './lib/health-policy'
import {
closeLeaderRedis,
getLeaderInfo,
readCursor,
releaseLeadership,
runLeaderElection,
saveDurableCursor,
} from './lib/leader'
import { createRevalidationResourceContext } from './lib/revalidate-resources'
import { getRevalidateWorkerState, startRevalidateWorker, stopRevalidateWorker } from './lib/revalidate-worker'
import { createStartupGate } from './lib/startup-gate'
import { getStorageStatsSnapshot, startStorageStatsRefresh, stopStorageStatsRefresh } from './lib/storage'
import { SupervisorClient } from './lib/supervisor-client'
import { assertSignallableWorkerPid } from './lib/supervisor-config'
// Initialize Grafana exporters if configured
initializeGrafanaExporters({
serviceName: 'firehose-service',
serviceVersion: '1.0.0',
})
const logger = createLogger('firehose-service')
const app = new Hono()
// Add observability middleware
app.use('*', observabilityMiddleware('firehose-service'))
// Error handler
app.onError(observabilityErrorHandler('firehose-service'))
// Lifecycle state is intentionally separate from the worker's health so the
// health endpoint can report an in-progress drain before the listener closes.
type ServiceLifecycle = 'starting' | 'running' | 'draining' | 'stopped'
let serviceLifecycle: ServiceLifecycle = 'starting'
let isShuttingDown = false
let healthServer: ServerType | null = null
let leaderAbortController: AbortController | null = null
let leaderElectionPromise: Promise<void> | null = null
let supervisorClient: SupervisorClient | null = null
let leadershipState: 'disabled' | 'starting' | 'standby' | 'acquired' | 'releasing' | 'released' | 'fatal' = 'disabled'
let cursorSaveTimer: ReturnType<typeof setInterval> | null = null
let cursorSaveInFlight: Promise<boolean> | null = null
let cursorSavingEnabled = false
let revalidateWorkerStartPromise: Promise<void> | null = null
let backfillPromise: Promise<void> | null = null
let backfillAbortController: AbortController | null = null
const firehoseStartGate = createStartupGate()
const BACKFILL_OPERATION_DEADLINE_MS = 5 * 60_000
const BACKFILL_TRANSFER_BUDGET_BYTES = 1024 * 1024 * 1024
function errorKind(error: unknown): string {
return error instanceof Error && error.name ? error.name : 'UnknownError'
}
function cursorSaveIntervalMs(): number {
const value = config.cursorSaveIntervalMs
return Number.isSafeInteger(value) && value > 0 ? value : 5_000
}
/** Serialize periodic cursor writes so shutdown can wait for a stable final save. */
function saveLatestCursor(force = false): Promise<boolean> {
if (!force && !cursorSavingEnabled) return Promise.resolve(true)
if (cursorSaveInFlight) {
const inFlight = cursorSaveInFlight
if (!force) return inFlight
// A forced save is the post-drain checkpoint. Wait for any earlier periodic
// write, then take a fresh cursor snapshot instead of releasing on an old one.
return inFlight.then(() => {
if (cursorSaveInFlight === inFlight) cursorSaveInFlight = null
return saveLatestCursor(true)
})
}
const save = (async () => {
const seq = getCurrentSeq()
if (seq === undefined) return true
return (await saveDurableCursor(seq, getActiveService())).kind === 'saved'
})().catch((error) => {
// Do not include the Redis endpoint or error message in application logs.
logger.warn('[Shutdown] Cursor save failed', { errorKind: errorKind(error) })
return false
})
cursorSaveInFlight = save
void save.then(() => {
if (cursorSaveInFlight === save) cursorSaveInFlight = null
})
return save
}
function startCursorSaving(): void {
cursorSavingEnabled = true
if (cursorSaveTimer) clearInterval(cursorSaveTimer)
cursorSaveTimer = setInterval(() => {
void saveLatestCursor()
}, cursorSaveIntervalMs())
}
async function stopCursorSaving(): Promise<void> {
cursorSavingEnabled = false
if (cursorSaveTimer) {
clearInterval(cursorSaveTimer)
cursorSaveTimer = null
}
const inFlight = cursorSaveInFlight
await inFlight
if (cursorSaveInFlight === inFlight) cursorSaveInFlight = null
}
/** Stop accepting new health checks immediately; existing requests may finish. */
function closeHealthServer(): Promise<void> {
const server = healthServer
healthServer = null
if (!server) return Promise.resolve()
return new Promise((resolve) => {
try {
server.close((error) => {
if (error) logger.warn('[Shutdown] Health server close failed', { errorKind: errorKind(error) })
resolve()
})
// Health has no long-lived protocol. End existing HTTP/1 keep-alive
// connections too, so close cannot delay resource teardown indefinitely.
if ('closeAllConnections' in server && typeof server.closeAllConnections === 'function') {
server.closeAllConnections()
}
} catch (error) {
logger.warn('[Shutdown] Health server close failed', { errorKind: errorKind(error) })
resolve()
}
})
}
// Health endpoint
app.get('/health', (c) => {
const firehoseHealth = getFirehoseHealth()
const storageStats = getStorageStatsSnapshot()
const revalidationState = getRevalidateWorkerState()
const revalidationConfigured = Boolean(config.redisUrl)
const workerExpected = !config.leadershipSupervisorEnabled || leadershipState === 'acquired'
const revalidation = resolveRevalidationHealth(workerExpected, revalidationConfigured, revalidationState)
const draining = isShuttingDown || firehoseHealth.draining
// Standby is healthy liveness, but it is never ready to serve as the active
// ingest worker until the independent supervisor owns both authority locks.
const standbyHealthy = config.leadershipSupervisorEnabled && leadershipState === 'standby'
const activeLive = workerExpected && firehoseHealth.healthy && revalidation.live
const activeReady = activeLive && revalidation.ready
// Capacity-stat scans are diagnostic. A slow first S3 listing is not an
// ingest liveness failure. A bounded revalidation reconnect is live but not ready.
const healthy = !draining && (standbyHealthy || activeLive)
const ready = !draining && activeReady
const payload = {
status: healthy ? ('healthy' as const) : ('degraded' as const),
lifecycle: draining ? ('draining' as const) : serviceLifecycle,
ready,
readiness: ready ? ('ready' as const) : standbyHealthy ? ('standby' as const) : ('not-ready' as const),
mode: config.isDbFillOnly
? ('db-fill-only' as const)
: config.isBackfill
? ('backfill' as const)
: ('firehose' as const),
firehose: firehoseHealth,
revalidation: {
enabled: revalidationConfigured && workerExpected,
configured: revalidationConfigured,
healthy: revalidation.live,
ready: revalidation.ready,
reconnecting: revalidation.reconnecting,
...revalidationState,
},
// Keep the former storage-stat fields at this level for existing callers.
storage: {
...(storageStats.stats ?? {}),
lastSuccessAgeMs: storageStats.lastSuccessAgeMs,
lastErrorKind: storageStats.lastErrorKind,
stale: storageStats.stale,
refreshing: storageStats.refreshing,
},
...(config.leaderElection &&
!config.leadershipSupervisorEnabled && {
leader: getLeaderInfo(),
}),
...(config.leadershipSupervisorEnabled && {
leadership: {
state: leadershipState,
epoch: supervisorClient?.state?.epoch,
supervisorPid: supervisorClient?.state?.pid,
},
}),
}
return c.json(payload, healthy ? 200 : 503)
})
async function runShutdownStep(name: string, operation: () => Promise<void>): Promise<void> {
try {
await operation()
} catch (error) {
// Avoid serializing dependency errors, which can contain endpoints or credentials.
logger.error(`[Shutdown] ${name} failed`, undefined, { errorKind: errorKind(error) })
}
}
function settlesWithinShutdownGrace(work: Promise<void> | null): Promise<boolean> {
if (!work) return Promise.resolve(true)
return settleWithTimeout(work, config.firehoseDrainGraceMs).then((result) => result.status === 'settled')
}
async function shutdownInternal(signal: string): Promise<void> {
isShuttingDown = true
firehoseStartGate.cancel()
serviceLifecycle = 'draining'
leadershipState = config.leadershipSupervisorEnabled ? 'releasing' : leadershipState
logger.info(`Received ${signal}; stopping intake and draining work`)
stopStorageStatsRefresh()
// Calling close() synchronously stops new accepts before any potentially long
// drain. Its completion is awaited before process exit.
const healthServerClosed = closeHealthServer()
const firehoseDrain = stopAndDrainFirehose()
// Stop admitting backfill work before waiting for its fixed worker pool. The
// signal is also passed to network work that supports RequestInit.signal.
backfillAbortController?.abort()
const backfillDrain = settlesWithinShutdownGrace(backfillPromise)
// No interval can race the final post-drain save. Wait for an already-started
// save before taking the final cursor snapshot below.
await stopCursorSaving()
const [drain, backfillDrained] = await Promise.all([firehoseDrain, backfillDrain])
// Revalidation is accepted work too. Stop it before releasing the supervisor,
// so no callback can touch shared dependencies after authority is handed off.
if (revalidateWorkerStartPromise) {
await runShutdownStep('revalidation worker startup', () => revalidateWorkerStartPromise ?? Promise.resolve())
}
let revalidationWorkerDrained = true
try {
revalidationWorkerDrained = (await stopRevalidateWorker({ gracePeriodMs: config.firehoseDrainGraceMs })).stopped
} catch (error) {
revalidationWorkerDrained = false
logger.error('[Shutdown] revalidation worker stop failed', undefined, { errorKind: errorKind(error) })
}
// The checkpoint is intentionally saved after accepted site work drains and
// before we ask the independent supervisor to release either lock.
const authorityEnabled = config.leaderElection || config.leadershipSupervisorEnabled
const finalCursorSaved = !authorityEnabled || (await saveLatestCursor(true))
// The old in-process election loop is retained only for local compatibility.
// Production uses the child supervisor, which keeps renewing independently.
leaderAbortController?.abort()
if (leaderElectionPromise) {
await leaderElectionPromise.catch((error) => {
logger.error('[Leader] Election loop stopped with an error', undefined, { errorKind: errorKind(error) })
})
}
if (drain.forced || !backfillDrained || !revalidationWorkerDrained || !finalCursorSaved) {
logger.error(
'[Shutdown] Work drain or final cursor save did not complete safely; terminating without closing shared dependencies',
undefined,
{
remainingWork: drain.remainingWork,
activeHandlers: drain.activeHandlers,
pendingCursorEvents: drain.pendingCursorEvents,
backfillDrained,
revalidationWorkerDrained,
finalCursorSaved,
},
)
// Do not release the supervisor while a handler can still use shared
// dependencies. The worker exits; the child observes EOF, fences the (now gone)
// parent, and releases authority safely.
serviceLifecycle = 'stopped'
await runShutdownStep('observability final flush', shutdownGrafanaExporters)
process.exitCode = 1
process.exit(1)
return
}
let supervisorReleased = true
if (config.leadershipSupervisorEnabled && supervisorClient) {
try {
const acknowledged = await settlesWithinShutdownGrace(supervisorClient.requestRelease())
if (!acknowledged) throw new Error('Leadership supervisor release acknowledgement timed out')
leadershipState = 'released'
} catch (error) {
supervisorReleased = false
logger.error('[Supervisor] Release acknowledgement failed', undefined, { errorKind: errorKind(error) })
}
} else if (config.leaderElection) {
await runShutdownStep('leader release', releaseLeadership)
}
await healthServerClosed
await runShutdownStep('cache publisher close', closeCacheInvalidationPublisher)
await runShutdownStep('leader Redis close', closeLeaderRedis)
await runShutdownStep('database close', closeDatabase)
serviceLifecycle = 'stopped'
logger.info('Shutdown complete')
await runShutdownStep('observability final flush', shutdownGrafanaExporters)
const exitCode = supervisorReleased ? 0 : 1
process.exitCode = exitCode
process.exit(exitCode)
}
/** Idempotent signal entry point. Every caller observes the same teardown. */
const shutdown = onceAsync(shutdownInternal)
process.on('SIGINT', () => {
void shutdown('SIGINT')
})
process.on('SIGTERM', () => {
void shutdown('SIGTERM')
})
/**
* Enumerate DIDs from a hydrant indexer.
*
* Unlike listAllKnownDids(), which can only rediscover DIDs we already have a
* row for, hydrant tracks every repo in the network that publishes a
* place.wisp.* record, so this also surfaces sites we have never seen.
*/
async function listDidsFromHydrant(hydrantUrl: string, signal: AbortSignal): Promise<string[]> {
const base = hydrantUrl.replace(/\/$/, '')
return await scanHydrantDids(
async (cursor, limit, requestSignal) => {
const params = new URLSearchParams({ limit: String(limit) })
if (cursor) params.set('cursor', cursor)
return await safeFetchJson<Array<{ did?: string }>>(`${base}/repos?${params.toString()}`, {
headers: { Accept: 'application/json' },
signal: requestSignal,
})
},
{ signal },
)
}
/**
* Backfill phase 1+2:
* - Collect all known DIDs from DB
* - Discover each DID's place.wisp.fs records directly from the PDS
*/
async function collectSitesFromKnownDids(signal: AbortSignal): Promise<Array<{ did: string; rkey: string }>> {
let dids: string[]
if (config.hydrantUrl) {
logger.info('Phase 1/3: Collecting DIDs from configured hydrant')
dids = await listDidsFromHydrant(config.hydrantUrl, signal)
logger.info(`Collected ${dids.length} DIDs from hydrant`)
} else {
logger.info('Phase 1/3: Collecting known DIDs')
dids = await listAllKnownDids()
throwIfBackfillAborted(signal)
logger.info(`Collected ${dids.length} known DIDs`)
}
if (dids.length === 0) {
logger.warn('No known DIDs found; skipping site discovery')
return []
}
logger.info('Phase 2/3: Discovering place.wisp.fs records from known DIDs')
let didsProcessed = 0
let didsFailed = 0
let sitesDiscovered = 0
let sitesFailed = 0
const discoveredSites = new Map<string, { did: string; rkey: string }>()
const concurrency = config.backfillConcurrency
const processDid = async (did: string, workerSignal: AbortSignal) => {
const resources = createRevalidationResourceContext(
BACKFILL_OPERATION_DEADLINE_MS,
BACKFILL_TRANSFER_BUDGET_BYTES,
workerSignal,
)
try {
throwIfBackfillAborted(workerSignal)
const records = await listSiteRecordsForDid(did, undefined, resources)
throwIfBackfillAborted(workerSignal)
for (const row of records) {
try {
discoveredSites.set(`${did}:${row.rkey}`, { did, rkey: row.rkey })
sitesDiscovered++
} catch (err) {
logger.error(`[Backfill:sites] Failed to register site ${did}/${row.rkey}`, undefined, {
errorKind: errorKind(err),
})
sitesFailed++
}
}
didsProcessed++
} catch (err) {
if (err instanceof BackfillAbortedError || workerSignal.aborted) return
logger.error(`[Backfill:sites] Failed to list records for DID ${did}`, undefined, { errorKind: errorKind(err) })
didsFailed++
} finally {
resources.close()
}
logger.info(
`[Backfill:sites] Progress ${didsProcessed + didsFailed}/${dids.length} DIDs (${sitesDiscovered} sites discovered, ${sitesFailed} sites failed)`,
)
}
await runCancellableWindow(dids, concurrency, signal, processDid)
throwIfBackfillAborted(signal)
logger.info(
`Phase 2/3 complete: ${didsProcessed} DIDs processed, ${didsFailed} DIDs failed, ${discoveredSites.size} unique sites discovered, ${sitesFailed} sites failed`,
)
return [...discoveredSites.values()]
}
/**
* Backfill phase 3:
* - process discovered sites and backfill blobs into S3
*/
async function runBackfill(signal: AbortSignal): Promise<void> {
throwIfBackfillAborted(signal)
logger.info('Starting backfill mode')
const startTime = Date.now()
const forceRewriteHtml = process.env.BACKFILL_FORCE_REWRITE_HTML === 'true'
const forceDownload = process.env.BACKFILL_FORCE_DOWNLOAD === 'true'
if (forceRewriteHtml) {
logger.info('Forcing HTML rewrite for all sites')
}
if (forceDownload) {
logger.info('Forcing full file download/write for all backfilled sites')
}
const sites = await collectSitesFromKnownDids(signal)
throwIfBackfillAborted(signal)
if (config.isDbFillOnly) {
logger.info('DB fill only mode enabled; skipping phase 3/3 cache backfill')
return
}
logger.info('Phase 3/3: Backfilling site blobs into S3')
const concurrency = config.backfillConcurrency
logger.info(`Found ${sites.length} sites to process (concurrency: ${concurrency})`)
let processed = 0
let skipped = 0
let failed = 0
const processSite = async (site: { did: string; rkey: string }, workerSignal: AbortSignal) => {
const resources = createRevalidationResourceContext(
BACKFILL_OPERATION_DEADLINE_MS,
BACKFILL_TRANSFER_BUDGET_BYTES,
workerSignal,
)
try {
throwIfBackfillAborted(workerSignal)
const result = await fetchSiteRecord(site.did, site.rkey, resources)
throwIfBackfillAborted(workerSignal)
if (!result) {
logger.info(`Site not found on PDS: ${site.did}/${site.rkey}`)
skipped++
return
}
const existingCache = await getSiteCache(site.did, site.rkey)
throwIfBackfillAborted(workerSignal)
if (
!forceRewriteHtml &&
!forceDownload &&
existingCache &&
result.cid === existingCache.record_cid &&
existingCache.cold_synced === true
) {
logger.info(`Site already up to date: ${site.did}/${site.rkey}`)
skipped++
return
}
throwIfBackfillAborted(workerSignal)
await handleSiteCreateOrUpdate(site.did, site.rkey, result.record, result.cid, {
forceRewriteHtml,
forceDownload,
resources,
})
throwIfBackfillAborted(workerSignal)
processed++
} catch (err) {
if (err instanceof BackfillAbortedError || workerSignal.aborted) return
logger.error(`Failed to process ${site.did}/${site.rkey}`, undefined, { errorKind: errorKind(err) })
failed++
} finally {
resources.close()
}
logger.info(
`Progress: ${processed + skipped + failed}/${sites.length} (${processed} processed, ${skipped} skipped, ${failed} failed)`,
)
}
await runCancellableWindow(sites, concurrency, signal, processSite)
if (signal.aborted) {
logger.info('Backfill cancelled after admitted work drained')
return
}
const elapsedMs = Date.now() - startTime
const elapsedSec = Math.round(elapsedMs / 1000)
const elapsedMin = Math.floor(elapsedSec / 60)
const elapsedRemSec = elapsedSec % 60
const elapsedLabel = elapsedMin > 0 ? `${elapsedMin}m ${elapsedRemSec}s` : `${elapsedSec}s`
logger.info(`Complete: ${processed} processed, ${skipped} skipped, ${failed} failed (${elapsedLabel} elapsed)`)
}
/**
* Start worker-side intervals only after the supervisor/election callback grants
* authority. The optional term guard fences an in-flight cursor read or worker
* startup from a lease which was lost before the await completed.
*/
async function startProtectedWork(
cursor: number | undefined,
onFailure: () => void,
isActive: () => boolean = () => true,
): Promise<boolean> {
if (isShuttingDown || !isActive()) return false
revalidateWorkerStartPromise = startRevalidateWorker()
await revalidateWorkerStartPromise
if (isShuttingDown || !isActive()) {
await stopRevalidateWorker({ gracePeriodMs: config.firehoseDrainGraceMs })
return false
}
if (config.leaderElection || config.leadershipSupervisorEnabled) {
if (!isActive()) {
await stopRevalidateWorker({ gracePeriodMs: config.firehoseDrainGraceMs })
return false
}
startCursorSaving()
}
if (!isActive()) {
await stopCursorSaving()
await stopRevalidateWorker({ gracePeriodMs: config.firehoseDrainGraceMs })
return false
}
startFirehose(cursor, onFailure)
firehoseStartGate.open()
return true
}
// Main entry point
async function main() {
logger.info('Starting firehose-service')
logger.info(`Mode: ${config.isDbFillOnly ? 'db-fill-only' : config.isBackfill ? 'backfill' : 'firehose'}`)
logger.info('Storage configured', { usingDiskFallback: !config.s3Bucket })
// Keep the returned Node server so shutdown can stop accepting health checks
// before it starts its potentially long worker drain.
healthServer = serve({
fetch: app.fetch,
port: config.healthPort,
})
startStorageStatsRefresh()
logger.info('Health server listening', { port: config.healthPort })
if (config.leadershipSupervisorEnabled) {
assertSignallableWorkerPid(process.pid)
leadershipState = 'starting'
supervisorClient = new SupervisorClient({
executable: config.supervisorPath,
onState: (message) => {
leadershipState = message.state
if (message.state === 'standby' && !isShuttingDown) serviceLifecycle = 'running'
if (message.state === 'acquired') {
logger.info('[Supervisor] Acquired firehose authority', { epoch: message.epoch })
}
if (message.state === 'fatal' && !isShuttingDown) void shutdown('leadership supervisor failure')
},
onFailure: () => {
leadershipState = 'fatal'
if (!isShuttingDown) void shutdown('leadership supervisor process failure')
},
})
await supervisorClient.start()
const acquired = await supervisorClient.waitForAcquired()
if (isShuttingDown) return
let cursor: number | undefined
try {
cursor = await readCursor(config.firehoseService)
} catch (error) {
logger.error('[Supervisor] Durable cursor read failed', undefined, { errorKind: errorKind(error) })
void shutdown('durable cursor read failed')
return
}
await startProtectedWork(cursor, () => {
logger.warn('[Supervisor] Firehose failure threshold reached; draining before release')
void shutdown('firehose failure threshold reached')
})
// Keep the epoch visible in logs without passing authority into worker code.
logger.info('[Supervisor] Protected worker started', { epoch: acquired.epoch })
} else if (config.leaderElection) {
logger.info('In-process leader election enabled; waiting for leadership before starting firehose')
leaderAbortController = new AbortController()
leaderElectionPromise = runLeaderElection(
async (cursor, stepDown, isActive) => {
await startProtectedWork(
cursor,
() => {
logger.warn('[Leader] Firehose failure threshold reached; requesting drain-aware step-down')
void stepDown()
},
isActive,
)
},
async () => {
const drain = await stopFirehose()
const workerStop = await stopRevalidateWorker({ gracePeriodMs: config.firehoseDrainGraceMs })
await stopCursorSaving()
if (drain.forced || workerStop.forced) {
// Do not leave a force-drained process alive with handlers that may
// still use shared dependencies. shutdown() preserves the safe cursor,
// lets the lease expire, flushes telemetry, and exits nonzero.
void shutdown('firehose drain grace period expired')
return false
}
// The in-process election owns its lease until this callback settles.
return await saveLatestCursor(true)
},
leaderAbortController.signal,
config.firehoseService,
)
void leaderElectionPromise.catch((error) => {
logger.error('[Leader] Election loop fatal error', undefined, { errorKind: errorKind(error) })
firehoseStartGate.cancel()
void shutdown('leader election fatal error')
})
} else {
// Single-instance mode: start firehose directly. There is no authority
// interval to start in this mode.
await startProtectedWork(undefined, () => {
logger.warn('[Firehose] Failure threshold reached; draining service')
void stopAndDrainFirehose().then((drain) => {
if (drain.forced) void shutdown('firehose drain grace period expired')
})
})
}
// A standby must not perform the production backfill or claim that the
// firehose is active before it has acquired leadership and started intake.
const firehoseStarted = await firehoseStartGate.wait()
if (!firehoseStarted || isShuttingDown) return
serviceLifecycle = 'running'
if (config.isBackfill) {
// Keep the promise visible to shutdown so DB resources cannot close under it.
logger.info('Running backfill after firehose intake started')
backfillAbortController = new AbortController()
backfillPromise = runBackfill(backfillAbortController.signal).catch((error) => {
if (backfillAbortController?.signal.aborted && error instanceof BackfillAbortedError) return
throw error
})
await backfillPromise
backfillAbortController = null
if (!isShuttingDown) logger.info('Backfill complete; continuing firehose consumption')
}
}
main().catch((error) => {
logger.error('Fatal startup error', undefined, { errorKind: errorKind(error) })
process.exit(1)
})