feat(heartbeats): Keep track of when the last submitter heartbeat was received

Author: tizoc

tools/heartbeats-processor/schema.sql

@@ -45,6 +45,7 @@ CREATE TABLE IF NOT EXISTS submitter_scores (
     score INTEGER NOT NULL DEFAULT 0,
     last_updated INTEGER NOT NULL DEFAULT (strftime('%s', 'now')),
     blocks_produced INTEGER NOT NULL DEFAULT 0,
+    last_heartbeat INTEGER NOT NULL DEFAULT 0,
     FOREIGN KEY (public_key_id) REFERENCES public_keys(id)
 );
 
@@ -83,6 +84,10 @@ ON heartbeat_presence(public_key_id);
 CREATE INDEX IF NOT EXISTS idx_heartbeat_presence_global_slot
 ON heartbeat_presence(best_tip_global_slot);
 
+-- Index for heartbeat time queries
+CREATE INDEX IF NOT EXISTS idx_heartbeat_presence_time
+ON heartbeat_presence(heartbeat_time);
+
 -- Index for submitter counts lookup
 CREATE INDEX IF NOT EXISTS idx_submitter_counts_last_seen 
 ON submitter_counts(last_seen);

tools/heartbeats-processor/src/local_db.rs

@@ -516,23 +516,33 @@ pub async fn update_scores(pool: &SqlitePool, config: &Config) -> Result<()> {
             GROUP BY public_key_id
         ),
         HeartbeatCounts AS (
-            SELECT hp.public_key_id, COUNT(DISTINCT hp.window_id) as heartbeats
+            SELECT
+                hp.public_key_id,
+                COUNT(DISTINCT hp.window_id) as heartbeats,
+                MAX(hp.heartbeat_time) as last_heartbeat
             FROM heartbeat_presence hp
             JOIN ValidWindows vw ON vw.id = hp.window_id
             GROUP BY hp.public_key_id
         )
-        INSERT INTO submitter_scores (public_key_id, score, blocks_produced)
+        INSERT INTO submitter_scores (
+            public_key_id,
+            score,
+            blocks_produced,
+            last_heartbeat
+        )
         SELECT
             pk.id,
             COALESCE(hc.heartbeats, 0) as score,
-            COALESCE(bc.blocks, 0) as blocks_produced
+            COALESCE(bc.blocks, 0) as blocks_produced,
+            COALESCE(hc.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
         WHERE hc.heartbeats > 0 OR bc.blocks > 0
         ON CONFLICT(public_key_id) DO UPDATE SET
             score = excluded.score,
             blocks_produced = excluded.blocks_produced,
+            last_heartbeat = excluded.last_heartbeat,
             last_updated = strftime('%s', 'now')
         "#,
         window_start,
@@ -580,7 +590,8 @@ pub async fn view_scores(pool: &SqlitePool, config: &Config) -> Result<()> {
             pk.public_key,
             ss.score,
             ss.blocks_produced,
-            datetime(ss.last_updated, 'unixepoch') as last_updated
+            datetime(ss.last_updated, 'unixepoch') as last_updated,
+            datetime(ss.last_heartbeat, 'unixepoch') as last_heartbeat
         FROM submitter_scores ss
         JOIN public_keys pk ON pk.id = ss.public_key_id
         ORDER BY ss.score DESC, ss.blocks_produced DESC
@@ -594,19 +605,20 @@ pub async fn view_scores(pool: &SqlitePool, config: &Config) -> Result<()> {
     println!("\nSubmitter Scores:");
     println!("--------------------------------------------------------");
     println!(
-        "Public Key                                              | Score | Blocks | Current Max | Total Max | Last Updated"
+        "Public Key                                              | Score | Blocks | Current Max | Total Max | Last Updated | Last Heartbeat"
     );
     println!("--------------------------------------------------------");
 
     for row in scores {
         println!(
-            "{:<40} | {:>5} | {:>6} | {:>11} | {:>9} | {}",
+            "{:<40} | {:>5} | {:>6} | {:>11} | {:>9} | {} | {}",
             row.public_key,
             row.score,
             row.blocks_produced,
             max_scores.current,
             max_scores.total,
-            row.last_updated.unwrap_or_default()
+            row.last_updated.unwrap_or_default(),
+            row.last_heartbeat.unwrap_or_default()
         );
     }
 

tools/heartbeats-processor/src/main.rs

@@ -75,7 +75,8 @@ async fn post_scores_to_firestore(
             pk.public_key,
             ss.score,
             ss.blocks_produced,
-            ss.last_updated
+            ss.last_updated,
+            ss.last_heartbeat
         FROM submitter_scores ss
         JOIN public_keys pk ON pk.id = ss.public_key_id
         ORDER BY ss.score DESC
@@ -91,6 +92,7 @@ async fn post_scores_to_firestore(
             score: row.score,
             blocks_produced: row.blocks_produced,
             last_updated: row.last_updated,
+            last_heartbeat: row.last_heartbeat,
         })
         .collect();
 

tools/heartbeats-processor/src/remote_db.rs

@@ -135,6 +135,8 @@ pub struct ScoreDocument {
     pub blocks_produced: i64,
     #[serde(rename = "lastUpdated")]
     pub last_updated: i64,
+    #[serde(rename = "lastHeartbeat")]
+    pub last_heartbeat: i64,
 }
 
 pub async fn get_db(config: &Config) -> Result<FirestoreDb> {