fix: comprehensive per-progress heartbeat coverage audit across worker/indexer long operations (#229)

apps/indexer/src/main/backfill.rs

@@ -50,9 +50,10 @@ pub(crate) use reservation_execution::coinbase_sql_backfill_job_source_identity_
 pub(crate) use reservation_execution::effective_hash_pinned_adapter_sync_mode;
 pub(crate) use reservation_execution::{
     DEFAULT_HASH_PINNED_BACKFILL_CHUNK_BLOCKS, backfill_job_source_identity_payload,
-    create_hash_pinned_backfill_job, effective_coinbase_sql_adapter_sync_mode,
-    hash_pinned_backfill_range_specs, run_precreated_hash_pinned_backfill_job,
-    run_resumable_coinbase_sql_backfill_job, run_resumable_hash_pinned_backfill_job,
+    backfill_job_source_identity_payload_with_progress, create_hash_pinned_backfill_job,
+    effective_coinbase_sql_adapter_sync_mode, hash_pinned_backfill_range_specs,
+    run_precreated_hash_pinned_backfill_job, run_resumable_coinbase_sql_backfill_job,
+    run_resumable_hash_pinned_backfill_job, run_resumable_hash_pinned_backfill_job_with_progress,
 };
 #[cfg(test)]
 pub(crate) use selection::SelectedTargetIntervalIndex;

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

@@ -15,7 +15,9 @@ use super::{
     coinbase_sql::load_backfill_topic_plan,
     reservation_execution::{
         backfill_lease_duration_secs, create_coinbase_sql_backfill_job_with_ranges,
-        create_hash_pinned_backfill_job_with_ranges, effective_coinbase_sql_adapter_sync_mode,
+        create_hash_pinned_backfill_job_with_ranges,
+        create_hash_pinned_backfill_job_with_ranges_with_progress,
+        effective_coinbase_sql_adapter_sync_mode,
         ensure_coinbase_sql_registry_range_start_is_replay_safe,
         refreshed_backfill_lease_expires_at, run_reserved_coinbase_sql_backfill_range,
         run_reserved_hash_pinned_backfill_range_with_progress, validate_hash_pinned_chunk_blocks,
@@ -84,8 +86,23 @@ async fn run_resumable_hash_pinned_backfill_job_concurrently_inner(
         );
     validate_hash_pinned_chunk_blocks(config.hash_pinned_chunk_blocks)?;
     let watched_chain = &source_plan.watched_chain_plan;
-    let record =
-        create_hash_pinned_backfill_job_with_ranges(pool, source_plan, &config, ranges).await?;
+    let record = match heartbeat.as_mut() {
+        Some((heartbeat, chain_ids)) => {
+            let mut progress =
+                crate::run::startup_heartbeat::StartupAdapterHeartbeat::new(heartbeat, chain_ids);
+            create_hash_pinned_backfill_job_with_ranges_with_progress(
+                pool,
+                source_plan,
+                &config,
+                ranges,
+                &mut progress,
+            )
+            .await?
+        }
+        None => {
+            create_hash_pinned_backfill_job_with_ranges(pool, source_plan, &config, ranges).await?
+        }
+    };
     config
         .idempotency_key
         .clone_from(&record.job.idempotency_key);