feat: add a read-only data-completeness gate for promotion and cutover

apps/indexer/src/main/backfill/reservation_execution.rs

@@ -56,7 +56,8 @@ use scan_all::{
 
 const HASH_PINNED_BACKFILL_SCAN_MODE: &str = "hash_pinned_block";
 pub(crate) const COINBASE_SQL_BACKFILL_SCAN_MODE: &str = "coinbase_sql_hash_pinned_logs_v1";
-pub(crate) const DEFAULT_HASH_PINNED_BACKFILL_CHUNK_BLOCKS: i64 = 1_024;
+pub(crate) const DEFAULT_HASH_PINNED_BACKFILL_CHUNK_BLOCKS: i64 =
+    bigname_storage::MAX_LIVE_CONTIGUOUS_GAP_FILL_BLOCKS;
 pub(crate) const COMPACT_SOURCE_IDENTITY_SELECTED_TARGET_THRESHOLD: usize = 10_000;
 
 pub(crate) async fn create_hash_pinned_backfill_job(

apps/indexer/src/main/normalized_replay_catchup.rs

@@ -25,7 +25,7 @@ mod indexes;
 mod sources;
 
 use cursors::{
-    advance_cursor, ensure_cursor, record_cursor_failure,
+    advance_cursor, clear_cursor_failure, ensure_cursor, record_cursor_failure,
     rewind_cursor_for_newly_observed_older_logs,
 };
 use indexes::{
@@ -276,6 +276,7 @@ pub(crate) async fn run_normalized_replay_catchup_iteration(
             )
             .await?;
         }
+        clear_cursor_failure(pool, &config.deployment_profile, chain).await?;
         return Ok(CatchupIterationStatus::Idle);
     };
     if pending_base_rederive_replay_target.is_some() {
@@ -334,6 +335,7 @@ pub(crate) async fn run_normalized_replay_catchup_iteration(
             )
             .await?;
         }
+        clear_cursor_failure(pool, &config.deployment_profile, chain).await?;
         return Ok(CatchupIterationStatus::Idle);
     }
 

apps/indexer/src/main/normalized_replay_catchup/cursors.rs

@@ -273,6 +273,39 @@ pub(super) async fn record_cursor_failure(
     Ok(())
 }
 
+/// A successful idle iteration is a successful health signal even though there is no cursor
+/// advancement to clear a previous transient failure. The predicate avoids rewriting a healthy
+/// cursor on every poll.
+pub(super) async fn clear_cursor_failure(
+    pool: &PgPool,
+    deployment_profile: &str,
+    chain: &str,
+) -> Result<()> {
+    sqlx::query(
+        r#"
+        UPDATE normalized_replay_cursors
+        SET
+            last_failure_reason = NULL,
+            last_failure_at = NULL,
+            updated_at = now()
+        WHERE deployment_profile = $1
+          AND chain_id = $2
+          AND cursor_kind = $3
+          AND (last_failure_reason IS NOT NULL OR last_failure_at IS NOT NULL)
+        "#,
+    )
+    .bind(deployment_profile)
+    .bind(chain)
+    .bind(CURSOR_KIND_RAW_FACT_NORMALIZED_EVENTS)
+    .execute(pool)
+    .await
+    .with_context(|| {
+        format!("failed to clear normalized replay cursor failure for {deployment_profile}/{chain}")
+    })?;
+
+    Ok(())
+}
+
 fn postgres_text_safe(text: &str) -> String {
     text.replace('\0', "\\u0000")
 }

apps/indexer/src/main/reconciliation/canonical.rs

@@ -1,15 +1,15 @@
 use anyhow::{Result, bail};
 use bigname_storage::{
-    CanonicalityState, ChainCheckpoint, ChainCheckpointUpdate, advance_chain_checkpoints,
-    chain_lineage_contains_ancestor, load_chain_lineage_block, mark_chain_lineage_range_orphaned,
+    CanonicalityState, ChainCheckpoint, ChainCheckpointUpdate, MAX_LIVE_CONTIGUOUS_GAP_FILL_BLOCKS,
+    advance_chain_checkpoints, chain_lineage_contains_ancestor, load_chain_lineage_block,
+    mark_chain_lineage_range_orphaned,
     upsert_chain_lineage_blocks_recanonicalizing_orphaned as upsert_recanonicalized_lineage_blocks,
     upsert_chain_lineage_blocks_without_snapshots,
     upsert_chain_lineage_blocks_without_snapshots_recanonicalizing_orphaned as upsert_recanonicalized_lineage_blocks_without_snapshots,
 };
 use tracing::{info, warn};
 
 use crate::{
-    backfill::DEFAULT_HASH_PINNED_BACKFILL_CHUNK_BLOCKS,
     provider::{ChainProviderOps, ProviderBlock, ProviderHeadSnapshot, ProviderRegistry},
     runtime::{IntakeChainTask, checkpoint_mode},
 };
@@ -48,8 +48,6 @@ use stored_lineage::{
 const MAX_PARENT_FETCH_DEPTH: usize = 131_072;
 // Live polling fails closed before it tries to ingest a large catch-up range.
 // Hash-pinned backfill owns larger bounded gaps.
-const MAX_LIVE_CONTIGUOUS_GAP_FILL_BLOCKS: i64 = DEFAULT_HASH_PINNED_BACKFILL_CHUNK_BLOCKS;
-
 #[allow(dead_code)]
 pub(crate) async fn poll_provider_heads(
     pool: &sqlx::PgPool,

apps/indexer/src/main/reconciliation/canonical/stored_lineage/coverage.rs

@@ -3,22 +3,14 @@ use std::sync::Mutex;
 
 use anyhow::Result;
 use bigname_manifests::{
-    UncoveredWatchedTuple, find_uncovered_watched_tuples,
-    load_active_manifest_abi_events_by_chain_and_source_families, load_discovery_admission_epoch,
-    load_log_producing_source_families, load_watched_contracts_by_addresses,
+    UncoveredWatchedTuple, WATCHED_COVERAGE_VERIFICATION_CHUNK_BLOCKS,
+    find_uncovered_watched_tuples, load_active_manifest_topic0s_by_chain_and_source_families,
+    load_discovery_admission_epoch, load_log_producing_source_families,
+    load_watched_contracts_by_addresses,
 };
-use bigname_storage::ChainLineageBlock;
+use bigname_storage::{ChainLineageBlock, ensure_backfill_family_topic_sets_undrifted};
 use sqlx::Row;
 
-#[path = "coverage/topic_drift.rs"]
-mod topic_drift;
-
-use topic_drift::ensure_family_topic_sets_undrifted;
-
-/// Frontier extensions verify coverage in chunks of this many blocks, so a
-/// deep gap costs a handful of anti-join queries once and every promotion
-/// cycle afterwards is an O(1) in-memory comparison.
-pub(crate) const COVERAGE_FRONTIER_VERIFICATION_CHUNK_BLOCKS: i64 = 131_072;
 const MAX_REPORTED_UNCOVERED_TUPLES: i64 = 20;
 
 /// Process-lifetime, per-chain memo of the block interval whose watched-tuple
@@ -143,7 +135,7 @@ pub(super) async fn stored_path_has_required_raw_fact_coverage(
 
 /// Extend the chain's verified coverage frontier until it contains
 /// `[required_from, required_through]`, verifying in
-/// [`COVERAGE_FRONTIER_VERIFICATION_CHUNK_BLOCKS`] chunks that opportunistically
+/// [`WATCHED_COVERAGE_VERIFICATION_CHUNK_BLOCKS`] chunks that opportunistically
 /// look ahead as far as `verify_ahead_through_block` (the stored promotion
 /// anchor) so subsequent cycles are O(1). A violation in the look-ahead beyond
 /// the required target falls back to verifying exactly up to the target, so an
@@ -159,8 +151,13 @@ async fn ensure_verified_coverage_frontier(
     let log_producing_source_families = load_log_producing_source_families(pool, chain)
         .await
         .map_err(|error| error.to_string())?;
-    let current_topic0s_by_family =
-        load_current_topic0s_by_family(pool, chain, &log_producing_source_families).await?;
+    let current_topic0s_by_family = load_active_manifest_topic0s_by_chain_and_source_families(
+        pool,
+        chain,
+        &log_producing_source_families,
+    )
+    .await
+    .map_err(|error| error.to_string())?;
     let topic_set_fingerprint = topic_set_fingerprint(&current_topic0s_by_family);
     let discovery_admission_epoch = load_discovery_admission_epoch(pool, chain)
         .await
@@ -189,7 +186,7 @@ async fn ensure_verified_coverage_frontier(
     let extension_from = interval
         .as_ref()
         .map_or(required_from, |existing| existing.through_block + 1);
-    if let Err(_look_ahead_drift) = ensure_family_topic_sets_undrifted(
+    if let Err(_look_ahead_drift) = ensure_backfill_family_topic_sets_undrifted(
         pool,
         chain,
         &current_topic0s_by_family,
@@ -202,7 +199,7 @@ async fn ensure_verified_coverage_frontier(
         // must not block promoting the covered prefix: recheck scoped to the
         // target, and if that passes, cap the look-ahead so the memo never
         // covers a span the drift guard did not clear.
-        ensure_family_topic_sets_undrifted(
+        ensure_backfill_family_topic_sets_undrifted(
             pool,
             chain,
             &current_topic0s_by_family,
@@ -222,7 +219,7 @@ async fn ensure_verified_coverage_frontier(
     while through_block < required_through {
         let chunk_from = through_block + 1;
         let chunk_through = chunk_from
-            .saturating_add(COVERAGE_FRONTIER_VERIFICATION_CHUNK_BLOCKS - 1)
+            .saturating_add(WATCHED_COVERAGE_VERIFICATION_CHUNK_BLOCKS - 1)
             .min(verify_ahead_through_block);
         let violations = find_uncovered_watched_tuples(
             pool,
@@ -316,36 +313,6 @@ fn uncovered_tuples_refusal(
     )
 }
 
-/// Current manifest topic0 sets per log-producing family; also the input to
-/// the frontier memo's ABI fingerprint.
-async fn load_current_topic0s_by_family(
-    pool: &sqlx::PgPool,
-    chain: &str,
-    log_producing_source_families: &[String],
-) -> std::result::Result<BTreeMap<String, BTreeSet<String>>, String> {
-    if log_producing_source_families.is_empty() {
-        return Ok(BTreeMap::new());
-    }
-    let events = load_active_manifest_abi_events_by_chain_and_source_families(
-        pool,
-        chain,
-        log_producing_source_families,
-    )
-    .await
-    .map_err(|error| error.to_string())?;
-    let mut current_topic0s_by_family = BTreeMap::<String, BTreeSet<String>>::new();
-    for event in events {
-        let Some(topic0) = event.topic0 else {
-            continue;
-        };
-        current_topic0s_by_family
-            .entry(event.source_family)
-            .or_default()
-            .insert(topic0.to_ascii_lowercase());
-    }
-    Ok(current_topic0s_by_family)
-}
-
 fn topic_set_fingerprint(current_topic0s_by_family: &BTreeMap<String, BTreeSet<String>>) -> String {
     let mut fingerprint = String::new();
     for (source_family, topic0s) in current_topic0s_by_family {

apps/indexer/src/main/reconciliation/replay/classification.rs

@@ -470,14 +470,9 @@ fn closure_source_families_for_contracts(
 }
 
 fn closure_or_dependency_source_families() -> Vec<String> {
-    NORMALIZED_EVENT_REPLAY_CONTRACTS
+    bigname_adapters::CLOSURE_OR_DEPENDENCY_REPLAY_SOURCE_FAMILIES
         .iter()
-        .filter(|contract| contract.raw_fact_replay_participant)
-        .filter(|contract| !contract.model.restricted_replay_supported())
-        .flat_map(|contract| contract.source_families.iter().copied())
-        .collect::<BTreeSet<_>>()
-        .into_iter()
-        .map(str::to_owned)
+        .map(|source_family| (*source_family).to_owned())
         .collect()
 }
 

apps/indexer/src/main/reconciliation/replay/classification/tests.rs

@@ -47,6 +47,21 @@ fn closure_or_dependency_contracts_do_not_claim_restricted_replay_proofs() {
     }
 }
 
+#[test]
+fn shared_latched_target_families_match_replay_contracts() {
+    let contract_families = NORMALIZED_EVENT_REPLAY_CONTRACTS
+        .iter()
+        .filter(|contract| contract.raw_fact_replay_participant)
+        .filter(|contract| contract.model != ReplayDependencyModel::StatelessRawFact)
+        .flat_map(|contract| contract.source_families.iter().copied())
+        .collect::<BTreeSet<_>>();
+    let shared_families = bigname_adapters::CLOSURE_OR_DEPENDENCY_REPLAY_SOURCE_FAMILIES
+        .iter()
+        .copied()
+        .collect::<BTreeSet<_>>();
+    assert_eq!(shared_families, contract_families);
+}
+
 #[test]
 fn implemented_full_closure_contracts_are_enumerated() {
     let actual = NORMALIZED_EVENT_REPLAY_CONTRACTS

apps/indexer/src/main/reconciliation/replay/profile_scope.rs

@@ -3,6 +3,7 @@ use bigname_manifests::{
     WatchedSourceSelector, load_manifest_declared_watched_source_selector_plan,
     load_watched_chain_plan, load_watched_contracts_by_addresses,
 };
+use bigname_storage::load_active_manifest_deployment_profile;
 
 use super::scoped::replay_source_scope_from_requested_scope;
 use crate::{
@@ -19,7 +20,11 @@ pub(super) async fn ensure_replay_matches_deployment_profile_scope(
     request: &RawFactNormalizedEventReplayRequest,
     range: Option<(i64, i64)>,
 ) -> Result<()> {
-    let active_profile = infer_active_manifest_deployment_profile(pool).await?;
+    let Some(active_profile) = load_active_manifest_deployment_profile(pool).await? else {
+        bail!(
+            "deployment_profile cannot be enforced because the active manifest/discovery corpus does not match a supported deployment profile"
+        );
+    };
     if request.deployment_profile != active_profile {
         bail!(
             "deployment_profile {} does not match active manifest/discovery corpus profile {active_profile}",
@@ -276,39 +281,3 @@ async fn ensure_active_watched_chain_for_replay_profile(
 
     Ok(())
 }
-
-async fn infer_active_manifest_deployment_profile(pool: &sqlx::PgPool) -> Result<String> {
-    let rows = sqlx::query_as::<_, (String, String)>(
-        r#"
-        SELECT DISTINCT chain, deployment_epoch
-        FROM manifest_versions
-        WHERE rollout_status = 'active'
-        ORDER BY chain, deployment_epoch
-        "#,
-    )
-    .fetch_all(pool)
-    .await
-    .context(
-        "failed to load active manifest/discovery corpus for replay deployment_profile enforcement",
-    )?;
-
-    if rows.is_empty() {
-        bail!("deployment_profile cannot be enforced because no active manifests are loaded");
-    }
-
-    let all_mainnet = rows.iter().all(|(chain, _)| chain.ends_with("-mainnet"));
-    if all_mainnet {
-        return Ok("mainnet".to_owned());
-    }
-
-    let all_sepolia_dev = rows.iter().all(|(chain, deployment_epoch)| {
-        chain.ends_with("-sepolia") && deployment_epoch.ends_with("_sepolia_dev")
-    });
-    if all_sepolia_dev {
-        return Ok("sepolia-dev".to_owned());
-    }
-
-    bail!(
-        "deployment_profile cannot be enforced because the active manifest/discovery corpus does not match a supported deployment profile"
-    );
-}