feat: persist and clean upgrade intent msg, ratelimit, and API

.gitignore

@@ -11,3 +11,4 @@ data
 .vscode
 *.json
 .DS_Store
+configs

subgraph-radio/benches/gossips.rs

@@ -53,6 +53,7 @@ fn gossip_poi_bench(c: &mut Criterion) {
             )],
             coverage: CoverageLevel::Comprehensive,
             collect_message_duration: 10,
+            ratelimit_threshold: 60000,
             log_level: String::from("info"),
             slack_token: None,
             slack_channel: None,

subgraph-radio/src/config.rs

@@ -153,7 +153,7 @@ impl Config {
             state
         } else {
             debug!("Created new state");
-            PersistedState::new(None, None, None)
+            PersistedState::new(None, None, None, None)
         }
     }
 
@@ -311,6 +311,15 @@ pub struct RadioInfrastructure {
             none: no automatic upgrade, only notifications.\nDefault: comprehensive"
     )]
     pub auto_upgrade: CoverageLevel,
+    #[clap(
+        long,
+        value_parser = value_parser!(i64).range(1..),
+        default_value = "86400",
+        value_name = "RATELIMIT_THRESHOLD",
+        env = "RATELIMIT_THRESHOLD",
+        help = "Set upgrade intent ratelimit in seconds: only one upgrade per subgraph within the threshold (default: 86400 seconds = 1 day)"
+    )]
+    pub ratelimit_threshold: i64,
     #[clap(
         long,
         value_parser = value_parser!(i64).range(1..),

subgraph-radio/src/messages/upgrade.rs

@@ -2,18 +2,21 @@ use async_graphql::SimpleObject;
 use ethers_contract::EthAbiType;
 use ethers_core::types::transaction::eip712::Eip712;
 use ethers_derive_eip712::*;
+use prost::Message;
+use serde::{Deserialize, Serialize};
+use tracing::{debug, info};
 
 use graphcast_sdk::{
     graphcast_agent::message_typing::{BuildMessageError, GraphcastMessage},
     graphql::client_graph_account::{owned_subgraphs, subgraph_hash_by_id},
 };
-use prost::Message;
-use serde::{Deserialize, Serialize};
-use tracing::debug;
 
-use crate::config::Config;
-use crate::operator::indexer_management::{check_decision_basis, offchain_sync_indexing_rules};
 use crate::operator::notifier::Notifier;
+use crate::{config::Config, state::PersistedState};
+use crate::{
+    operator::indexer_management::{check_decision_basis, offchain_sync_indexing_rules},
+    OperationError,
+};
 
 #[derive(Eip712, EthAbiType, Clone, Message, Serialize, Deserialize, PartialEq, SimpleObject)]
 #[eip712(
@@ -78,20 +81,36 @@ impl UpgradeIntentMessage {
         Ok(self)
     }
 
+    /// Format the notification for an UpgradeIntentMessage
+    fn notification_formatter(&self) -> String {
+        format!(
+            "Subgraph owner for a deployment announced an upgrade intent:\nSubgraph ID: {}\nNew deployment hash: {}",
+            self.subgraph_id,
+            self.new_hash,
+        )
+    }
+
     /// Process the validated upgrade intent messages
     /// If notification is set up, then notify the indexer
     /// If indexer management server endpoint is set up, radio checks `auto_upgrade` for
-    pub async fn process_valid_message(&self, config: &Config, notifier: &Notifier) {
+    pub async fn process_valid_message(
+        &self,
+        config: &Config,
+        notifier: &Notifier,
+        state: &PersistedState,
+    ) -> Result<&Self, OperationError> {
+        // ratelimit upgrades: return early if there was a recent upgrade
+        if state.recent_upgrade(self, config.radio_infrastructure.ratelimit_threshold) {
+            info!(subgraph = &self.subgraph_id, "Received an Upgrade Intent Message for a recently upgraded subgraph, skiping notification and auto deployment");
+            return Ok(self);
+        }
         // send notifications
-        notifier.notify(format!(
-            "Subgraph owner for a deployment announced an upgrade intent:\nSubgraph ID: {}\nNew deployment hash: {}",
-            self.subgraph_id,
-            self.new_hash,
-        )).await;
-        // auto deployment
-        // If the identifier satisfy the config coverage level
-        // and if indexer management server endpoint is provided
+        notifier.notify(self.notification_formatter()).await;
+
+        // auto-deployment
+        // require configured indexer management server endpoint
         if let Some(url) = &config.graph_stack().indexer_management_server_endpoint {
+            // If the identifier satisfy the config coverage level
             let covered_topics = config
                 .generate_topics(
                     &config.radio_infrastructure().auto_upgrade,
@@ -100,17 +119,14 @@ impl UpgradeIntentMessage {
                 .await;
             // Get the current deployment hash by querying network subgraph and take the latest hash of the subgraph id
             // Should be able to assume valid identifier since the message is valid
-            let identifier = if let Ok(hash) = subgraph_hash_by_id(
+            let identifier = subgraph_hash_by_id(
                 config.graph_stack().network_subgraph(),
                 &self.graph_account,
                 &self.subgraph_id,
             )
             .await
-            {
-                hash
-            } else {
-                return;
-            };
+            .map_err(OperationError::Query)?;
+
             if covered_topics
                 .clone()
                 .into_iter()
@@ -122,7 +138,8 @@ impl UpgradeIntentMessage {
                     res = tracing::field::debug(&res),
                     decision_basis, "New deployment setting"
                 );
-            }
+            };
         }
+        Ok(self)
     }
 }

subgraph-radio/src/operator/mod.rs

@@ -473,7 +473,13 @@ pub async fn process_message(
                 VALIDATED_MESSAGES
                     .with_label_values(&[&msg.identifier, "upgrade_intent_message"])
                     .inc();
-                radio_msg.process_valid_message(&config, &notifier).await;
+                if radio_msg
+                    .process_valid_message(&config, &notifier, &state)
+                    .await
+                    .is_ok()
+                {
+                    state.add_upgrade_intent_message(msg.clone());
+                };
             };
         } else {
             trace!("Waku message not decoded or validated, skipped message",);

subgraph-radio/src/server/model/mod.rs

@@ -5,7 +5,7 @@ use thiserror::Error;
 
 use crate::{
     config::Config,
-    messages::poi::PublicPoiMessage,
+    messages::{poi::PublicPoiMessage, upgrade::UpgradeIntentMessage},
     operator::attestation::{
         self, attestations_to_vec, compare_attestation, process_ppoi_message, Attestation,
         AttestationEntry, AttestationError, ComparisonResult, ComparisonResultType,
@@ -23,7 +23,7 @@ pub struct QueryRoot;
 
 #[Object]
 impl QueryRoot {
-    async fn radio_payload_messages(
+    async fn public_poi_messages(
         &self,
         ctx: &Context<'_>,
         identifier: Option<String>,
@@ -35,6 +35,20 @@ impl QueryRoot {
         Ok(msgs)
     }
 
+    async fn upgrade_intent_messages(
+        &self,
+        ctx: &Context<'_>,
+        subgraph_id: Option<String>,
+    ) -> Result<Vec<UpgradeIntentMessage>, HttpServiceError> {
+        let msgs = ctx
+            .data_unchecked::<Arc<SubgraphRadioContext>>()
+            .upgrade_intent_messages_filtered(&subgraph_id)
+            .into_iter()
+            .map(|m| m.payload)
+            .collect();
+        Ok(msgs)
+    }
+
     async fn local_attestations(
         &self,
         ctx: &Context<'_>,
@@ -257,6 +271,22 @@ impl SubgraphRadioContext {
         filtered
     }
 
+    pub fn upgrade_intent_messages(
+        &self,
+    ) -> HashMap<String, GraphcastMessage<UpgradeIntentMessage>> {
+        self.persisted_state.upgrade_intent_messages()
+    }
+
+    pub fn upgrade_intent_messages_filtered(
+        &self,
+        subgraph_id: &Option<String>,
+    ) -> Vec<GraphcastMessage<UpgradeIntentMessage>> {
+        subgraph_id
+            .as_ref()
+            .and_then(|id| self.upgrade_intent_messages().get(id).cloned())
+            .map_or(vec![], |m| vec![m])
+    }
+
     pub fn comparison_result(&self, identifier: String) -> Option<ComparisonResult> {
         let cmp_results = self.persisted_state.comparison_results();
         cmp_results.get(&identifier).cloned()