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

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

@@ -3,9 +3,10 @@ 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_abi_events_by_chain_and_source_families,
+    load_discovery_admission_epoch, load_log_producing_source_families,
+    load_watched_contracts_by_addresses,
 };
 use bigname_storage::ChainLineageBlock;
 use sqlx::Row;
@@ -15,10 +16,6 @@ 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 +140,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
@@ -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,

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"
-    );
-}

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

@@ -1052,6 +1052,73 @@ async fn replay_normalized_events_rejects_deployment_profile_outside_active_mani
     database.cleanup().await
 }
 
+#[tokio::test]
+async fn replay_normalized_events_accepts_sepolia_manifest_profile() -> Result<()> {
+    let database = TestDatabase::new().await?;
+    let chain = "ethereum-sepolia";
+    let reverse_address = "0x00000000000000000000000000000000000000d5";
+    let claimed_address = "0x5555555555555555555555555555555555555555";
+    let block = provider_block(
+        "0xd5d5d5d5d5d5d5d5d5d5d5d5d5d5d5d5d5d5d5d5d5d5d5d5d5d5d5d5d5d5d5",
+        Some("0x1111111111111111111111111111111111111111111111111111111111111111"),
+        105,
+    );
+
+    insert_active_replay_watched_contract_with_source_family(
+        database.pool(),
+        40,
+        chain,
+        "ens_v1_reverse_l1",
+        Uuid::from_u128(0x940),
+        reverse_address,
+        "reverse_registrar",
+    )
+    .await?;
+    sqlx::query(
+        "UPDATE manifest_versions SET deployment_epoch = 'ens_v1_sepolia_dev' WHERE manifest_id = 40",
+    )
+    .execute(database.pool())
+    .await?;
+    insert_active_replay_manifest(
+        database.pool(),
+        41,
+        "ens",
+        "ens_v1_resolver_l1",
+        chain,
+        "ens_v1_sepolia_dev",
+    )
+    .await?;
+    insert_chain_lineage_for_block(database.pool(), chain, &block, CanonicalityState::Canonical)
+        .await?;
+    insert_raw_reverse_claimed_log(
+        database.pool(),
+        chain,
+        &block,
+        reverse_address,
+        claimed_address,
+        CanonicalityState::Canonical,
+    )
+    .await?;
+
+    let outcome = replay_raw_fact_normalized_events(
+        database.pool(),
+        RawFactNormalizedEventReplayRequest {
+            deployment_profile: "sepolia".to_owned(),
+            chain: chain.to_owned(),
+            selection: RawFactNormalizedEventReplaySelection::BlockRange {
+                from_block: block.block_number,
+                to_block: block.block_number,
+            },
+        },
+    )
+    .await?;
+
+    assert_eq!(outcome.selected_block_count, 1);
+    assert_eq!(outcome.normalized_event_inserted_count, 1);
+
+    database.cleanup().await
+}
+
 #[tokio::test]
 async fn replay_normalized_events_skips_noncanonical_raw_logs_in_selected_block_hashes()
 -> Result<()> {

apps/worker/src/inspect.rs

@@ -1,5 +1,6 @@
 mod backfill;
 mod canonicality;
+mod data_completeness;
 mod execution_trace;
 mod formatting;
 mod manifest_drift;
@@ -32,6 +33,10 @@ pub(crate) enum InspectCommand {
         about = "Inspect canonicality, durable raw fact counts, and retained payload-cache metadata for one block hash"
     )]
     Canonicality(InspectCanonicalityArgs),
+    #[command(
+        about = "Check whether this database is data-complete enough to serve: reconciliation frontier, watch-set code-observation coverage, replay and projection cursors, and projection content"
+    )]
+    DataCompleteness(InspectDataCompletenessArgs),
     #[command(about = "Inspect one persisted execution trace and its ordered steps")]
     ExecutionTrace(InspectExecutionTraceArgs),
     #[command(about = "Inspect stored manifest drift and proxy implementation alert observations")]
@@ -60,6 +65,18 @@ pub(crate) struct InspectCanonicalityArgs {
     pub(crate) block_hash: String,
 }
 
+#[derive(Args, Debug)]
+pub(crate) struct InspectDataCompletenessArgs {
+    #[command(flatten)]
+    pub(crate) database: DatabaseConfig,
+    #[arg(long)]
+    pub(crate) json: bool,
+    #[arg(long)]
+    pub(crate) fail_on_incomplete: bool,
+    #[arg(long)]
+    pub(crate) max_head_lag_blocks: Option<i64>,
+}
+
 #[derive(Args, Debug)]
 pub(crate) struct InspectExecutionTraceArgs {
     #[command(flatten)]
@@ -102,6 +119,9 @@ pub(crate) async fn inspect_command(args: InspectArgs) -> Result<()> {
     match args.command {
         InspectCommand::BackfillJob(args) => backfill::inspect_backfill_job(args).await,
         InspectCommand::Canonicality(args) => canonicality::inspect_canonicality(args).await,
+        InspectCommand::DataCompleteness(args) => {
+            data_completeness::inspect_data_completeness(args).await
+        }
         InspectCommand::ExecutionTrace(args) => {
             execution_trace::inspect_execution_trace(args).await
         }