Merge pull request #1075 from openmina/feat/heartbeats-last-hb2

feat(heartbeats): Tweaks

Author: tizoc

tools/heartbeats-processor/.sqlx/query-94669ea962673f9d266e0cbeeaf854c789c92aa5ccc2673293022c7b505243e8.json

@@ -245,9 +245,11 @@ pub async fn process_heartbeats(
     db: &FirestoreDb,
     pool: &SqlitePool,
     config: &Config,
-) -> Result<()> {
+) -> Result<usize> {
     let last_processed_time = get_last_processed_time(pool, Some(config)).await?;
     let now = Utc::now();
+    // Don't fetch heartbeats beyond window range end
+    let end_time = config.window_range_end.min(now);
 
     let mut total_heartbeats = 0;
     let mut latest_time = last_processed_time;
@@ -266,7 +268,8 @@ pub async fn process_heartbeats(
     };
 
     loop {
-        let heartbeats = crate::remote_db::fetch_heartbeat_chunk(db, &mut chunk_state, now).await?;
+        let heartbeats =
+            crate::remote_db::fetch_heartbeat_chunk(db, &mut chunk_state, end_time).await?;
         if heartbeats.is_empty() {
             break;
         }
@@ -430,7 +433,7 @@ pub async fn process_heartbeats(
         update_last_processed_time(pool, latest_time).await?;
     }
 
-    Ok(())
+    Ok(total_heartbeats)
 }
 
 pub async fn create_tables_from_file(pool: &SqlitePool) -> Result<()> {
@@ -516,13 +519,21 @@ pub async fn update_scores(pool: &SqlitePool, config: &Config) -> Result<()> {
             GROUP BY public_key_id
         ),
         HeartbeatCounts AS (
+            -- Count heartbeats only within valid windows
             SELECT
                 hp.public_key_id,
-                COUNT(DISTINCT hp.window_id) as heartbeats,
-                MAX(hp.heartbeat_time) as last_heartbeat
+                COUNT(DISTINCT hp.window_id) as heartbeats
             FROM heartbeat_presence hp
             JOIN ValidWindows vw ON vw.id = hp.window_id
             GROUP BY hp.public_key_id
+        ),
+        LastHeartbeats AS (
+            -- Get last heartbeat time across all windows
+            SELECT
+                public_key_id,
+                MAX(heartbeat_time) as last_heartbeat
+            FROM heartbeat_presence
+            GROUP BY public_key_id
         )
         INSERT INTO submitter_scores (
             public_key_id,
@@ -534,10 +545,11 @@ pub async fn update_scores(pool: &SqlitePool, config: &Config) -> Result<()> {
             pk.id,
             COALESCE(hc.heartbeats, 0) as score,
             COALESCE(bc.blocks, 0) as blocks_produced,
-            COALESCE(hc.last_heartbeat, 0) as last_heartbeat
+            COALESCE(lh.last_heartbeat, 0) as last_heartbeat
         FROM public_keys pk
         LEFT JOIN HeartbeatCounts hc ON hc.public_key_id = pk.id
         LEFT JOIN BlockCounts bc ON bc.public_key_id = pk.id
+        LEFT JOIN LastHeartbeats lh ON lh.public_key_id = pk.id
         WHERE hc.heartbeats > 0 OR bc.blocks > 0
         ON CONFLICT(public_key_id) DO UPDATE SET
             score = excluded.score,

tools/heartbeats-processor/.sqlx/query-c69ce193a25e2b2b5c1208df04d22d54114d61c2a2fc0f026b58183e8a8232ca.json

@@ -112,10 +112,12 @@ async fn run_process_loop(
 
     loop {
         println!("Processing heartbeats...");
-        local_db::process_heartbeats(db, pool, config).await?;
+        let count = local_db::process_heartbeats(db, pool, config).await?;
 
-        println!("Posting scores...");
-        post_scores_to_firestore(pool, db, config).await?;
+        if count > 0 {
+            println!("Posting scores...");
+            post_scores_to_firestore(pool, db, config).await?;
+        }
 
         println!("Sleeping for {} seconds...", interval_seconds);
         tokio::time::sleep(interval).await;

tools/heartbeats-processor/src/local_db.rs

@@ -171,10 +171,14 @@ pub async fn fetch_heartbeat_chunk(
     let chunk_end = (state.chunk_start + chunk_duration).min(end_time);
 
     if state.chunk_start >= end_time {
+        println!("Reached end of testing window: {}", end_time);
         return Ok(Vec::new());
     }
 
-    println!("Fetching heartbeat chunk... {}", state.chunk_start);
+    println!(
+        "Fetching heartbeat chunk... {} to {}",
+        state.chunk_start, chunk_end
+    );
 
     let query = db
         .fluent()