feat: differentiate between offline and allocated subgraphs

grafana.json

@@ -494,7 +494,7 @@
           "datasource": "${DS_SUBGRAPHRADIOGRAPHQL}",
           "endTimePath": "endTime",
           "groupBy": "null",
-          "queryText": "query {\n  data:comparisonRatio{\n    deployment\n    blockNumber\n    senderRatio\n    stakeRatio\n  }\n}",
+          "queryText": "query {\n  data:comparisonRatio{\n    deployment\n    blockNumber\n    senderRatio\n    stakeRatio\n   allocated\n}\n}",
           "refId": "A",
           "timePath": "Time"
         }
@@ -512,7 +512,8 @@
               "_blockNumber": "Block Number",
               "_deployment": "Deployment",
               "_senderRatio": "Count Ratio",
-              "_stakeRatio": "Stake Ratio"
+              "_stakeRatio": "Stake Ratio",
+              "_allocated": "Allocated"
             }
           }
         }

subgraph-radio/benches/attestations.rs

@@ -7,7 +7,7 @@ mod attestation {
     use criterion::{black_box, criterion_group, Criterion};
     use graphcast_sdk::graphcast_agent::message_typing::GraphcastMessage;
     use sqlx::SqlitePool;
-    use std::collections::HashMap;
+    use std::collections::{HashMap, HashSet};
     use subgraph_radio::{
         database::{insert_local_attestation, insert_remote_ppoi_message},
         messages::poi::PublicPoiMessage,
@@ -75,6 +75,7 @@ mod attestation {
                     42,
                     black_box(&remote_attestations.clone()),
                     "my-awesome-hash",
+                    HashSet::new(),
                 )
             })
         });
@@ -87,6 +88,7 @@ mod attestation {
                 .await
                 .expect("Failed to connect to the in-memory database"),
         );
+
         black_box(
             sqlx::migrate!("../migrations")
                 .run(&pool)

subgraph-radio/src/database/mod.rs

@@ -74,9 +74,15 @@ pub async fn get_local_attestation(
 
         let attestation = Attestation {
             identifier: data.identifier,
-            block_number: data.block_number as u64,
+            block_number: data
+                .block_number
+                .parse::<u64>()
+                .expect("Failed to parse as u64"),
             ppoi: data.ppoi,
-            stake_weight: data.stake_weight as u64,
+            stake_weight: data
+                .stake_weight
+                .parse::<u64>()
+                .expect("Failed to parse as u64"),
             senders,
             sender_group_hash: data.sender_group_hash,
             timestamp: timestamps,
@@ -119,9 +125,15 @@ pub async fn get_local_attestations_by_identifier(
 
         full_attestations.push(Attestation {
             identifier: att.identifier,
-            block_number: att.block_number as u64,
+            block_number: att
+                .block_number
+                .parse::<u64>()
+                .expect("Failed to parse as u64"),
             ppoi: att.ppoi,
-            stake_weight: att.stake_weight as u64,
+            stake_weight: att
+                .stake_weight
+                .parse::<u64>()
+                .expect("Failed to parse as u64"),
             senders,
             sender_group_hash: att.sender_group_hash,
             timestamp: timestamps,
@@ -158,9 +170,15 @@ pub async fn get_local_attestations(pool: &SqlitePool) -> Result<Vec<Attestation
 
         full_attestations.push(Attestation {
             identifier: record.identifier,
-            block_number: record.block_number as u64,
+            block_number: record
+                .block_number
+                .parse::<u64>()
+                .expect("Failed to parse as u64"),
             ppoi: record.ppoi,
-            stake_weight: record.stake_weight as u64,
+            stake_weight: record
+                .stake_weight
+                .parse::<u64>()
+                .expect("Failed to parse as u64"),
             sender_group_hash: record.sender_group_hash,
             senders,
             timestamp: timestamps,
@@ -251,15 +269,18 @@ pub async fn get_remote_ppoi_messages(
         .into_iter()
         .map(|row| GraphcastMessage {
             identifier: row.identifier.clone(),
-            nonce: row.nonce as u64,
+            nonce: row.nonce.parse::<u64>().expect("Failed to parse as u64"),
             graph_account: row.graph_account.clone(),
             signature: row.signature,
             payload: PublicPoiMessage {
                 identifier: row.identifier,
                 content: row.content,
-                nonce: row.nonce as u64,
+                nonce: row.nonce.parse::<u64>().expect("Failed to parse as u64"),
                 network: row.network,
-                block_number: row.block_number as u64,
+                block_number: row
+                    .block_number
+                    .parse::<u64>()
+                    .expect("Failed to parse as u64"),
                 block_hash: row.block_hash,
                 graph_account: row.graph_account,
             },
@@ -288,15 +309,18 @@ pub async fn get_remote_ppoi_messages_by_identifier(
         .into_iter()
         .map(|row| GraphcastMessage {
             identifier: row.identifier.clone(),
-            nonce: row.nonce as u64,
+            nonce: row.nonce.parse::<u64>().expect("Failed to parse as u64"),
             graph_account: row.graph_account.clone(),
             signature: row.signature,
             payload: PublicPoiMessage {
                 identifier: row.identifier,
                 content: row.content,
-                nonce: row.nonce as u64,
+                nonce: row.nonce.parse::<u64>().expect("Failed to parse as u64"),
                 network: row.network,
-                block_number: row.block_number as u64,
+                block_number: row
+                    .block_number
+                    .parse::<u64>()
+                    .expect("Failed to parse as u64"),
                 block_hash: row.block_hash,
                 graph_account: row.graph_account,
             },
@@ -381,13 +405,13 @@ pub async fn get_upgrade_intent_message(
                 deployment: msg.deployment.clone(),
                 subgraph_id: msg.subgraph_id,
                 new_hash: msg.new_hash,
-                nonce: msg.nonce as u64,
+                nonce: msg.nonce.parse::<u64>().expect("Failed to parse as u64"),
                 graph_account: msg.graph_account.clone(),
             };
 
             let graphcast_message = GraphcastMessage::new(
                 msg.deployment,
-                msg.nonce as u64,
+                msg.nonce.parse::<u64>().expect("Failed to parse as u64"),
                 msg.graph_account,
                 payload,
                 msg.signature,
@@ -420,13 +444,13 @@ pub async fn get_upgrade_intent_message_by_id(
                 deployment: msg.deployment.clone(),
                 subgraph_id: msg.subgraph_id,
                 new_hash: msg.new_hash,
-                nonce: msg.nonce as u64,
+                nonce: msg.nonce.parse::<u64>().expect("Failed to parse as u64"),
                 graph_account: msg.graph_account.clone(),
             };
 
             let graphcast_message = GraphcastMessage::new(
                 msg.deployment,
-                msg.nonce as u64,
+                msg.nonce.parse::<u64>().expect("Failed to parse as u64"),
                 msg.graph_account,
                 payload,
                 msg.signature,
@@ -456,13 +480,13 @@ pub async fn get_upgrade_intent_messages(
             deployment: msg.deployment.clone(),
             subgraph_id: msg.subgraph_id,
             new_hash: msg.new_hash,
-            nonce: msg.nonce as u64,
+            nonce: msg.nonce.parse::<u64>().expect("Failed to parse as u64"),
             graph_account: msg.graph_account.clone(),
         };
 
         let graphcast_message = GraphcastMessage::new(
             msg.deployment,
-            msg.nonce as u64,
+            msg.nonce.parse::<u64>().expect("Failed to parse as u64"),
             msg.graph_account,
             payload,
             msg.signature,
@@ -622,7 +646,10 @@ pub async fn get_comparison_results(
 
         results.push(ComparisonResult {
             deployment: row.deployment,
-            block_number: row.block_number as u64,
+            block_number: row
+                .block_number
+                .parse::<u64>()
+                .expect("Failed to parse as u64"),
             result_type: ComparisonResultType::from_str(&row.result_type)?,
             local_attestation,
             attestations,
@@ -659,7 +686,10 @@ pub async fn get_comparison_results_by_type(
                 .map_err(DatabaseError::ParseError)
                 .map(|result_type| ComparisonResult {
                     deployment: row.deployment,
-                    block_number: row.block_number as u64,
+                    block_number: row
+                        .block_number
+                        .parse::<u64>()
+                        .expect("Failed to parse as u64"),
                     result_type,
                     local_attestation,
                     attestations,
@@ -692,7 +722,10 @@ pub async fn get_comparison_results_by_deployment(
 
         Ok(ComparisonResult {
             deployment: row.deployment,
-            block_number: row.block_number as u64,
+            block_number: row
+                .block_number
+                .parse::<u64>()
+                .expect("Failed to parse as u64"),
             result_type: ComparisonResultType::from_str(&row.result_type)?,
             local_attestation,
             attestations,

subgraph-radio/src/database/test.rs

@@ -1,5 +1,7 @@
 #[cfg(test)]
 mod tests {
+    use std::collections::HashSet;
+
     use graphcast_sdk::{graphcast_agent::message_typing::GraphcastMessage, networks::NetworkName};
     use sqlx::SqlitePool;
 
@@ -189,7 +191,7 @@ mod tests {
             attestations: Vec::new(),
         };
 
-        let _ = handle_comparison_result(&pool, &new_result).await;
+        let _ = handle_comparison_result(&pool, &new_result, HashSet::new()).await;
         let comparison_results = get_comparison_results(&pool).await.unwrap();
         let result = comparison_results.first().unwrap();
 
@@ -223,7 +225,7 @@ mod tests {
         };
 
         let _ = save_comparison_result(&pool, &old_result).await;
-        let _ = handle_comparison_result(&pool, &new_result).await;
+        let _ = handle_comparison_result(&pool, &new_result, HashSet::new()).await;
 
         let comparison_results = get_comparison_results(&pool).await.unwrap();
         assert_eq!(