fix(health): detect stopped ingestion and stale chain data — liveness, freshness, backlog exposure (#186)

.env.example

@@ -8,10 +8,21 @@ DATABASE_URL=postgresql://bigname:bigname@127.0.0.1:5432/bigname
 BIGNAME_DATABASE_URL=postgresql://bigname:bigname@127.0.0.1:5432/bigname
 # Per-service primary pool ceiling; the API reserves one additional readiness connection.
 BIGNAME_DATABASE_MAX_CONNECTIONS=10
+# Stable service identities used by docker-compose.server.yml.
+BIGNAME_INDEXER_HEARTBEAT_INSTANCE_ID=indexer
+BIGNAME_WORKER_HEARTBEAT_INSTANCE_ID=worker
 
 # API bind address for local bigname-api serve.
 BIGNAME_API_BIND_ADDR=127.0.0.1:3000
-# API live execution RPC entries, as comma-delimited <chain>=<url> pairs.
+BIGNAME_API_HEARTBEAT_MAX_AGE_SECS=20
+BIGNAME_API_WORKER_REBUILD_PHASE_MAX_AGE_SECS=43200
+BIGNAME_API_STATUS_PROVIDER_TIMEOUT_MS=750
+BIGNAME_API_STATUS_PROVIDER_REFRESH_SECS=5
+BIGNAME_API_STATUS_PROVIDER_CACHE_TTL_SECS=30
+BIGNAME_API_STATUS_MAX_BLOCK_LAG=30
+BIGNAME_API_STATUS_MAX_LAG_SECS=60
+# API live execution and status-readiness RPC entries, as comma-delimited <chain>=<url> pairs.
+# Status remains degraded for every expected status chain omitted here.
 # BIGNAME_API_CHAIN_RPC_URLS=ethereum-mainnet=http://127.0.0.1:8545
 # Maximum addresses accepted by the native identity batch endpoint.
 BIGNAME_API_IDENTITY_BATCH_LIMIT=1000
@@ -45,6 +56,7 @@ BIGNAME_API_TRUST_X_FORWARDED_FOR=false
 BIGNAME_INDEXER_MANIFESTS_ROOT=manifests/mainnet
 # Poll interval in seconds for live indexer loops.
 BIGNAME_INDEXER_POLL_INTERVAL_SECS=5
+BIGNAME_INDEXER_HEARTBEAT_MAX_AGE_SECS=20
 # Maximum watched addresses verified per chain by one code-observation sweep tick.
 BIGNAME_INDEXER_RAW_CODE_BASELINE_MAX_ADDRESSES_PER_TICK=2048
 # Block span for hash-pinned backfill ranges.
@@ -145,6 +157,8 @@ BIGNAME_INDEXER_LIVE_ADAPTER_BACKLOG_BLOCK_BATCH_SIZE=1
 
 # Worker loop interval in seconds.
 BIGNAME_WORKER_POLL_INTERVAL_SECS=5
+BIGNAME_WORKER_HEARTBEAT_MAX_AGE_SECS=20
+BIGNAME_WORKER_REBUILD_PHASE_MAX_AGE_SECS=43200
 # Optional worker RPC for ENSv1 legacy text hydration during projection replay/apply.
 # BIGNAME_WORKER_CHAIN_RPC_URLS=ethereum-mainnet=http://127.0.0.1:8545
 # Multicall3 address for projection-owned text hydration.

.env.server.example

@@ -8,12 +8,22 @@ POSTGRES_PASSWORD=change-me
 BIGNAME_DATABASE_URL=postgresql://bigname:change-me@postgres:5432/bigname
 # Per-service primary pool ceiling; the API reserves one additional readiness connection.
 BIGNAME_DATABASE_MAX_CONNECTIONS=10
+# Stable process identities shared by each service and its healthcheck command.
+BIGNAME_INDEXER_HEARTBEAT_INSTANCE_ID=indexer
+BIGNAME_WORKER_HEARTBEAT_INSTANCE_ID=worker
 
 # API binding. The container listens on 3000; these control the host binding.
 # Keep the API's host-published port private; Caddy is the public path.
 BIGNAME_API_HOST=127.0.0.1
 BIGNAME_API_PORT=3000
 BIGNAME_API_BIND_ADDR=0.0.0.0:3000
+BIGNAME_API_HEARTBEAT_MAX_AGE_SECS=20
+BIGNAME_API_WORKER_REBUILD_PHASE_MAX_AGE_SECS=43200
+BIGNAME_API_STATUS_PROVIDER_TIMEOUT_MS=750
+BIGNAME_API_STATUS_PROVIDER_REFRESH_SECS=5
+BIGNAME_API_STATUS_PROVIDER_CACHE_TTL_SECS=30
+BIGNAME_API_STATUS_MAX_BLOCK_LAG=30
+BIGNAME_API_STATUS_MAX_LAG_SECS=60
 # Maximum addresses accepted by the native identity batch endpoint.
 BIGNAME_API_IDENTITY_BATCH_LIMIT=1000
 # Maximum inputs accepted by the v2 lookup endpoint.
@@ -37,7 +47,8 @@ BIGNAME_PUBLIC_SITE_ADDRESS=:80
 BIGNAME_PUBLIC_HTTP_PORT=80
 BIGNAME_PUBLIC_HTTPS_PORT=443
 
-# Comma-delimited <chain>=<url> entries for API-triggered live execution.
+# Comma-delimited <chain>=<url> entries for API-triggered live execution and
+# status readiness. Every expected status chain must be present in production.
 # ENS verified resolution currently needs ethereum-mainnet when no matching
 # persisted execution output exists.
 # API JSON-RPC support accepts http:// and https:// endpoints.
@@ -46,6 +57,7 @@ BIGNAME_PUBLIC_HTTPS_PORT=443
 # Runtime profile selection. Use /app/manifests/sepolia for the ENSv2 Sepolia profile.
 BIGNAME_INDEXER_MANIFESTS_ROOT=/app/manifests/mainnet
 BIGNAME_INDEXER_POLL_INTERVAL_SECS=5
+BIGNAME_INDEXER_HEARTBEAT_MAX_AGE_SECS=20
 BIGNAME_INDEXER_HASH_PINNED_BACKFILL_CHUNK_BLOCKS=1024
 BIGNAME_INDEXER_HASH_PINNED_BACKFILL_MAX_LOGS_PER_PUSH=10000
 # Legacy alias for BIGNAME_INDEXER_HASH_PINNED_BACKFILL_MAX_LOGS_PER_PUSH.
@@ -115,6 +127,8 @@ BIGNAME_INDEXER_LIVE_ADAPTER_BACKLOG_BLOCK_BATCH_SIZE=1
 # Use the same extra address list as BIGNAME_WORKER_PRIMARY_NAME_LEGACY_REVERSE_RESOLVER_ADDRESSES.
 # BIGNAME_INDEXER_EVENT_SILENT_REVERSE_RESOLVER_ADDRESSES=0x...
 BIGNAME_WORKER_POLL_INTERVAL_SECS=5
+BIGNAME_WORKER_HEARTBEAT_MAX_AGE_SECS=20
+BIGNAME_WORKER_REBUILD_PHASE_MAX_AGE_SECS=43200
 
 # Optional worker RPC for projection-owned ENSv1 legacy text hydration and
 # legacy event-silent reverse resolver primary-name hydration.

apps/api/src/cli.rs

@@ -19,7 +19,7 @@ pub(crate) struct Cli {
 
 #[derive(Subcommand, Debug)]
 pub(crate) enum Command {
-    Serve(ServeArgs),
+    Serve(Box<ServeArgs>),
     PrintOpenapi,
 }
 
@@ -47,6 +47,48 @@ pub(crate) struct ServeArgs {
     pub(crate) rpc_timeout_ms: u64,
     #[command(flatten)]
     pub(crate) bounds: ApiBoundsConfig,
+    #[arg(
+        long,
+        env = "BIGNAME_API_HEARTBEAT_MAX_AGE_SECS",
+        default_value_t = 20_i64
+    )]
+    pub(crate) heartbeat_max_age_secs: i64,
+    #[arg(
+        long,
+        env = "BIGNAME_API_WORKER_REBUILD_PHASE_MAX_AGE_SECS",
+        default_value_t = bigname_storage::DEFAULT_WORKER_REBUILD_PHASE_MAX_AGE_SECS
+    )]
+    pub(crate) worker_rebuild_phase_max_age_secs: i64,
+    #[arg(
+        long,
+        env = "BIGNAME_API_STATUS_PROVIDER_TIMEOUT_MS",
+        default_value_t = crate::status_freshness::DEFAULT_PROVIDER_TIMEOUT_MS
+    )]
+    pub(crate) status_provider_timeout_ms: u64,
+    #[arg(
+        long,
+        env = "BIGNAME_API_STATUS_PROVIDER_REFRESH_SECS",
+        default_value_t = crate::status_freshness::DEFAULT_PROVIDER_REFRESH_SECS
+    )]
+    pub(crate) status_provider_refresh_secs: u64,
+    #[arg(
+        long,
+        env = "BIGNAME_API_STATUS_PROVIDER_CACHE_TTL_SECS",
+        default_value_t = crate::status_freshness::DEFAULT_PROVIDER_CACHE_TTL_SECS
+    )]
+    pub(crate) status_provider_cache_ttl_secs: u64,
+    #[arg(
+        long,
+        env = "BIGNAME_API_STATUS_MAX_BLOCK_LAG",
+        default_value_t = crate::status_freshness::DEFAULT_MAX_BLOCK_LAG
+    )]
+    pub(crate) status_max_block_lag: i64,
+    #[arg(
+        long,
+        env = "BIGNAME_API_STATUS_MAX_LAG_SECS",
+        default_value_t = crate::status_freshness::DEFAULT_MAX_LAG_SECS
+    )]
+    pub(crate) status_max_lag_secs: i64,
     #[command(flatten)]
     pub(crate) database: DatabaseConfig,
 }

apps/api/src/handlers/app_facing/identity/mod.rs

@@ -24,5 +24,5 @@ async fn load_indexing_status_response(state: &AppState) -> ApiResult<IndexingSt
             ApiError::internal_error("failed to load indexing status")
         })?;
 
-    Ok(build_indexing_status_response(&read))
+    Ok(build_indexing_status_response(&read, state).await)
 }

apps/api/src/handlers/health.rs

@@ -7,6 +7,7 @@ pub(crate) const HEALTH_DATABASE_CHECK_TIMEOUT: std::time::Duration =
 pub(crate) struct HealthDatabasePool(pub(crate) PgPool);
 
 pub(super) async fn health(
+    State(state): State<AppState>,
     axum::Extension(health_pool): axum::Extension<HealthDatabasePool>,
 ) -> (StatusCode, Json<HealthResponse>) {
     let database_reachable = match tokio::time::timeout(
@@ -36,37 +37,154 @@ pub(super) async fn health(
         }
     };
 
-    let (http_status, status, database) = match database_reachable {
-        true => (
-            StatusCode::OK,
-            "ready",
-            HealthDatabaseResponse {
+    let (database, loops, loops_ready) = match database_reachable {
+        true => {
+            let database = HealthDatabaseResponse {
                 status: "reachable",
                 reachable: true,
                 check: "select_1",
                 error: None,
-            },
-        ),
-        false => (
-            StatusCode::SERVICE_UNAVAILABLE,
-            "degraded",
-            HealthDatabaseResponse {
+            };
+            match bigname_storage::load_preferred_service_loop_heartbeats(
+                &health_pool.0,
+                &[
+                    bigname_storage::INDEXER_SERVICE_NAME,
+                    bigname_storage::WORKER_SERVICE_NAME,
+                ],
+                state.heartbeat_max_age_secs,
+                state.worker_rebuild_phase_max_age_secs,
+            )
+            .await
+            {
+                Ok(heartbeats) => {
+                    let indexer = loop_health_response(
+                        heartbeats.iter().find(|heartbeat| {
+                            heartbeat.service_name == bigname_storage::INDEXER_SERVICE_NAME
+                        }),
+                        state.heartbeat_max_age_secs,
+                        state.heartbeat_max_age_secs,
+                    );
+                    let worker = loop_health_response(
+                        heartbeats.iter().find(|heartbeat| {
+                            heartbeat.service_name == bigname_storage::WORKER_SERVICE_NAME
+                        }),
+                        state.heartbeat_max_age_secs,
+                        state.worker_rebuild_phase_max_age_secs,
+                    );
+                    let loops_ready = indexer.status == "running" && worker.status == "running";
+                    (database, HealthLoopsResponse { indexer, worker }, loops_ready)
+                }
+                Err(readiness_error) => {
+                    warn!(
+                        service = "api",
+                        build_sha = BUILD_SHA,
+                        error = ?readiness_error,
+                        "service loop heartbeat readiness probe failed"
+                    );
+                    (
+                        database,
+                        unavailable_loop_health(state.heartbeat_max_age_secs),
+                        false,
+                    )
+                }
+            }
+        }
+        false => {
+            let database = HealthDatabaseResponse {
                 status: "unreachable",
                 reachable: false,
                 check: "select_1",
                 error: Some("database readiness query failed"),
-            },
-        ),
+            };
+            (
+                database,
+                unavailable_loop_health(state.heartbeat_max_age_secs),
+                false,
+            )
+        }
     };
+    let api_ready = database.reachable;
+    let aggregate_ready = api_ready && loops_ready;
+    let http_status = if api_ready {
+        StatusCode::OK
+    } else {
+        StatusCode::SERVICE_UNAVAILABLE
+    };
+    let status = if aggregate_ready {
+        "ready"
+    } else {
+        "degraded"
+    };
+    let api_status = if api_ready { "ready" } else { "degraded" };
 
     (
         http_status,
         Json(HealthResponse {
             service: "api",
             identity: HealthIdentityResponse::current(),
             status,
+            api_status,
             process: HealthProcessResponse { status: "running" },
             database,
+            loops,
         }),
     )
 }
+
+fn loop_health_response(
+    heartbeat: Option<&bigname_storage::ServiceLoopHeartbeat>,
+    max_age_seconds: i64,
+    phase_max_age_seconds: i64,
+) -> HealthLoopResponse {
+    let Some(heartbeat) = heartbeat else {
+        return HealthLoopResponse {
+            status: "not_started",
+            phase: None,
+            started_at: None,
+            heartbeat_at: None,
+            heartbeat_age_seconds: None,
+            max_age_seconds,
+        };
+    };
+    if let Some(phase) = heartbeat.active_phase.as_ref() {
+        return HealthLoopResponse {
+            status: if phase.age_seconds <= phase_max_age_seconds {
+                "running"
+            } else {
+                "stale"
+            },
+            phase: Some(phase.phase.clone()),
+            started_at: Some(format_timestamp(phase.started_at)),
+            heartbeat_at: Some(format_timestamp(phase.heartbeat_at)),
+            heartbeat_age_seconds: Some(phase.age_seconds),
+            max_age_seconds: phase_max_age_seconds,
+        };
+    }
+    HealthLoopResponse {
+        status: if heartbeat.age_seconds <= max_age_seconds {
+            "running"
+        } else {
+            "stale"
+        },
+        phase: None,
+        started_at: Some(format_timestamp(heartbeat.started_at)),
+        heartbeat_at: Some(format_timestamp(heartbeat.heartbeat_at)),
+        heartbeat_age_seconds: Some(heartbeat.age_seconds),
+        max_age_seconds,
+    }
+}
+
+fn unavailable_loop_health(max_age_seconds: i64) -> HealthLoopsResponse {
+    let unavailable = || HealthLoopResponse {
+        status: "unavailable",
+        phase: None,
+        started_at: None,
+        heartbeat_at: None,
+        heartbeat_age_seconds: None,
+        max_age_seconds,
+    };
+    HealthLoopsResponse {
+        indexer: unavailable(),
+        worker: unavailable(),
+    }
+}

apps/api/src/main.rs

@@ -46,6 +46,7 @@ mod pagination;
 mod query;
 mod routes;
 mod state;
+mod status_freshness;
 mod types;
 mod v2;
 
@@ -87,7 +88,7 @@ async fn main() -> Result<()> {
     match Cli::parse().command {
         Command::Serve(args) => {
             init_tracing("bigname-api");
-            serve(args).await
+            serve(*args).await
         }
         Command::PrintOpenapi => {
             print!("{}", render_openapi_document());

apps/api/src/openapi/schemas.rs

@@ -25,8 +25,8 @@ use identity_native::{
     native_identity_record_schema, normalization_info_schema,
 };
 use health::{
-    health_database_schema, health_identity_schema, health_process_schema,
-    health_projection_publication_versions_schema, health_response_schema,
+    health_database_schema, health_identity_schema, health_loop_schema, health_loops_schema,
+    health_process_schema, health_projection_publication_versions_schema, health_response_schema,
 };
 use primary_name::{
     primary_name_claimed_result_schema, primary_name_route_provenance_schema,
@@ -356,6 +356,8 @@ pub(super) fn openapi_components() -> JsonValue {
             "HealthIdentity": health_identity_schema(),
             "HealthProcess": health_process_schema(),
             "HealthDatabase": health_database_schema(),
+            "HealthLoop": health_loop_schema(),
+            "HealthLoops": health_loops_schema(),
             "HealthResponse": health_response_schema(),
             "ErrorBody": json!({
                 "type": "object",