feat: Add SPDD and DRDD modules with optional historical storage

common/Cargo.toml

@@ -18,7 +18,7 @@ blake2 = "0.10"
 bs58 = "0.5"
 caryatid_sdk = "0.12"
 caryatid_module_clock = "0.12"
-caryatid_module_rest_server = "0.13"
+caryatid_module_rest_server = "0.14"
 chrono = "0.4"
 gcd = "2.3"
 fraction = "0.15"

common/src/rest_helper.rs

@@ -5,7 +5,7 @@ use anyhow::{anyhow, Result};
 use caryatid_sdk::Context;
 use futures::future::Future;
 use num_traits::ToPrimitive;
-use std::sync::Arc;
+use std::{collections::HashMap, sync::Arc};
 use tokio::task::JoinHandle;
 use tracing::{error, info};
 
@@ -44,7 +44,7 @@ where
 }
 
 /// Handle a REST request with path parameters
-pub fn handle_rest_with_parameter<F, Fut>(
+pub fn handle_rest_with_path_parameter<F, Fut>(
     context: Arc<Context<Message>>,
     topic: &str,
     handler: F,
@@ -83,6 +83,35 @@ where
     })
 }
 
+// Handle a REST request with query parameters
+pub fn handle_rest_with_query_parameters<F, Fut>(
+    context: Arc<Context<Message>>,
+    topic: &str,
+    handler: F,
+) -> JoinHandle<()>
+where
+    F: Fn(HashMap<String, String>) -> Fut + Send + Sync + Clone + 'static,
+    Fut: Future<Output = Result<RESTResponse>> + Send + 'static,
+{
+    context.handle(topic, move |message: Arc<Message>| {
+        let handler = handler.clone();
+        async move {
+            let response = match message.as_ref() {
+                Message::RESTRequest(request) => {
+                    let params = request.query_parameters.clone();
+                    match handler(params).await {
+                        Ok(response) => response,
+                        Err(error) => RESTResponse::with_text(500, &format!("{error:?}")),
+                    }
+                }
+                _ => RESTResponse::with_text(500, "Unexpected message in REST request"),
+            };
+
+            Arc::new(Message::RESTResponse(response))
+        }
+    })
+}
+
 /// Extract parameters from the request path based on the topic pattern.
 /// Skips the first 3 parts of the topic as these are never parameters
 fn extract_params_from_topic_and_path(topic: &str, path_elements: &[String]) -> Vec<String> {
@@ -113,3 +142,37 @@ impl<T: ToPrimitive> ToCheckedF64 for T {
         self.to_f64().ok_or_else(|| anyhow!("Failed to convert {name} to f64"))
     }
 }
+
+// Macros for extracting and validating REST query parameters
+#[macro_export]
+macro_rules! extract_strict_query_params {
+    ($params:expr, { $($key:literal => $var:ident : Option<$type:ty>,)* }) => {
+        $(
+            let mut $var: Option<$type> = None;
+        )*
+
+        for (k, v) in &$params {
+            match k.as_str() {
+                $(
+                    $key => {
+                        $var = match v.parse::<$type>() {
+                            Ok(val) => Some(val),
+                            Err(_) => {
+                                return Ok($crate::messages::RESTResponse::with_text(
+                                    400,
+                                    concat!("Invalid ", $key, " query parameter: must be a valid type"),
+                                ));
+                            }
+                        };
+                    }
+                )*
+                _ => {
+                    return Ok($crate::messages::RESTResponse::with_text(
+                        400,
+                        concat!("Unexpected query parameter: only allowed keys are: ", $( $key, " ", )*)
+                    ));
+                }
+            }
+        }
+    };
+}

common/src/state_history.rs

@@ -1,92 +1,112 @@
 //! Generic state history
-//! Keeps per-block state to allow for rollbacks
+//! Keeps per-block state for rollbacks or per-epoch state for historical lookups
 //! Use imbl collections in the state to avoid memory explosion!
 
 use crate::params::SECURITY_PARAMETER_K;
-use crate::types::BlockInfo;
 use std::collections::VecDeque;
 
 use tracing::info;
 
-/// Entry in the history - S is the state to be stored
-struct HistoryEntry<S> {
-    /// Block number this state is for
-    block: u64,
+pub enum HistoryKind {
+    BlockState, // Used for rollbacks, bounded at k
+    EpochState, // Used for historical lookups, unbounded
+}
 
-    /// State to store
+struct HistoryEntry<S> {
+    index: u64,
     state: S,
 }
 
 /// Generic state history - S is the state to be stored
 pub struct StateHistory<S> {
-    /// History, one per block
+    /// History, one per block or epoch
     history: VecDeque<HistoryEntry<S>>,
 
     /// Module name
     module: String,
+
+    // Block or Epoch based history
+    kind: HistoryKind,
 }
 
 impl<S: Clone + Default> StateHistory<S> {
     /// Construct
-    pub fn new(module: &str) -> Self {
+    pub fn new(module: &str, kind: HistoryKind) -> Self {
         Self {
             history: VecDeque::new(),
             module: module.to_string(),
+            kind: kind,
         }
     }
 
     /// Get the current state (if any), direct ref
     pub fn current(&self) -> Option<&S> {
-        match self.history.back() {
-            Some(entry) => Some(&entry.state),
-            None => None,
-        }
+        self.history.back().map(|entry| &entry.state)
+    }
+
+    /// Get the current state assuming any rollback has been done
+    /// Cloned for modification - call commit() when done
+    pub fn get_current_state(&self) -> S {
+        self.history.back().map(|entry| entry.state.clone()).unwrap_or_default()
     }
 
     /// Get the previous state for the given block, handling rollbacks if required
     /// State returned is cloned ready for modification - call commit() when done
-    pub fn get_rolled_back_state(&mut self, block: &BlockInfo) -> S {
-        loop {
-            match self.history.back() {
-                Some(state) => {
-                    if state.block >= block.number {
+    pub fn get_rolled_back_state(&mut self, index: u64) -> S {
+        match self.kind {
+            HistoryKind::BlockState => {
+                while let Some(entry) = self.history.back() {
+                    if entry.index >= index {
                         info!(
                             "{} rolling back state to {} removing block {}",
-                            self.module, block.number, state.block
+                            self.module, index, entry.index
+                        );
+                        self.history.pop_back();
+                    } else {
+                        break;
+                    }
+                }
+            }
+            HistoryKind::EpochState => {
+                while let Some(entry) = self.history.back() {
+                    if entry.index >= index {
+                        info!(
+                            "{} rolling back epoch state to {} removing epoch {}",
+                            self.module, index, entry.index
                         );
                         self.history.pop_back();
                     } else {
                         break;
                     }
                 }
-                _ => break,
             }
         }
 
         self.get_current_state()
     }
 
-    /// Get the current state assuming any rollback has been done
-    /// Cloned for modification - call commit() when done
-    pub fn get_current_state(&mut self) -> S {
-        if let Some(current) = self.history.back() {
-            current.state.clone()
-        } else {
-            S::default()
-        }
+    /// Get the state for a given index (if any), direct ref
+    pub fn get_by_index(&self, index: u64) -> Option<&S> {
+        self.history.iter().find(|entry| entry.index == index).map(|entry| &entry.state)
+    }
+
+    pub fn len(&self) -> usize {
+        self.history.len()
     }
 
     /// Commit the new state
-    pub fn commit(&mut self, block: &BlockInfo, state: S) {
-        // Prune beyond 'k'
-        while self.history.len() >= SECURITY_PARAMETER_K as usize {
-            self.history.pop_front();
+    pub fn commit(&mut self, index: u64, state: S) {
+        match self.kind {
+            HistoryKind::BlockState => {
+                while self.history.len() >= SECURITY_PARAMETER_K as usize {
+                    self.history.pop_front();
+                }
+                self.history.push_back(HistoryEntry { index, state });
+            }
+            HistoryKind::EpochState => {
+                self.history.push_back(HistoryEntry { index, state });
+            }
         }
-
-        self.history.push_back(HistoryEntry {
-            block: block.number,
-            state,
-        });
     }
 }
 

common/src/types.rs

@@ -482,7 +482,7 @@ pub struct StakeDelegation {
 }
 
 /// SPO total delegation data (for SPDD)
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize, PartialEq)]
 pub struct DelegatedStake {
     /// Active stake - UTXO values only (used for reward calcs)
     pub active: Lovelace,