Merge branch 'main' into feat/sparse-fieldset-allowlist

Author: LuisD-Dev

backend/src/lib.rs

@@ -10,6 +10,9 @@ pub mod contract_ops;
 pub mod distributed_lock;
 pub mod event_indexer;
 pub mod field_selection;
+pub mod network;
 pub mod observability;
 pub mod realtime;
 pub mod reconciliation;
+pub mod replay;
+pub mod snapshot;

backend/src/network/client.rs

@@ -0,0 +1,33 @@
+use super::identity::Network;
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct NetworkClient {
+    pub network: Network,
+    pub rpc_url: String,
+    pub horizon_url: String,
+}
+
+impl NetworkClient {
+    pub fn for_network(network: Network) -> Self {
+        let (rpc_url, horizon_url) = match network {
+            Network::Mainnet => (
+                "https://horizon-mainnet.stellar.org".to_string(),
+                "https://horizon-mainnet.stellar.org".to_string(),
+            ),
+            Network::Testnet => (
+                "https://horizon-testnet.stellar.org".to_string(),
+                "https://horizon-testnet.stellar.org".to_string(),
+            ),
+            Network::Futurenet => (
+                "https://horizon-futurenet.stellar.org".to_string(),
+                "https://horizon-futurenet.stellar.org".to_string(),
+            ),
+        };
+
+        Self {
+            network,
+            rpc_url,
+            horizon_url,
+        }
+    }
+}

backend/src/network/identity.rs

@@ -0,0 +1,58 @@
+use std::fmt;
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)]
+pub enum Network {
+    Mainnet,
+    Testnet,
+    Futurenet,
+}
+
+impl fmt::Display for Network {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Network::Mainnet => write!(f, "mainnet"),
+            Network::Testnet => write!(f, "testnet"),
+            Network::Futurenet => write!(f, "futurenet"),
+        }
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct TableSchema {
+    pub name: String,
+    pub fields: Vec<String>,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum NetworkSchemaError {
+    #[error("schema missing network discriminator: {0}")]
+    MissingNetworkDiscriminator(String),
+}
+
+pub struct NetworkSchema;
+
+impl NetworkSchema {
+    pub fn validate(schemas: &[TableSchema]) -> Result<(), String> {
+        for schema in schemas {
+            let has_network = schema
+                .fields
+                .iter()
+                .any(|field| field.eq_ignore_ascii_case("network"));
+            if !has_network {
+                return Err(format!(
+                    "schema missing network discriminator for table {}",
+                    schema.name
+                ));
+            }
+        }
+
+        Ok(())
+    }
+
+    pub fn assert_startup_schema(schemas: &[TableSchema]) {
+        match Self::validate(schemas) {
+            Ok(()) => {}
+            Err(message) => panic!("startup schema check failed: {message}"),
+        }
+    }
+}

backend/src/network/mod.rs

@@ -0,0 +1,42 @@
+pub mod client;
+pub mod identity;
+
+use std::collections::HashMap;
+
+pub use client::NetworkClient;
+pub use identity::{Network, NetworkSchema, TableSchema};
+
+#[derive(Clone, Debug, PartialEq)]
+pub struct MetricRecord {
+    pub corridor: String,
+    pub network: Network,
+    pub reliability: f64,
+    pub volume: f64,
+}
+
+#[derive(Clone, Debug, Default)]
+pub struct NetworkStore {
+    by_network: HashMap<Network, Vec<MetricRecord>>,
+}
+
+impl NetworkStore {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    pub fn ingest(&mut self, corridor: String, network: Network, reliability: f64, volume: f64) {
+        self.by_network
+            .entry(network)
+            .or_default()
+            .push(MetricRecord {
+                corridor,
+                network,
+                reliability,
+                volume,
+            });
+    }
+
+    pub fn for_network(&self, network: Network) -> Vec<MetricRecord> {
+        self.by_network.get(&network).cloned().unwrap_or_default()
+    }
+}

backend/src/replay/mod.rs

@@ -0,0 +1,40 @@
+use crate::snapshot::{generate_snapshot, model_version_for_ledger_range, RawSnapshotRow};
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct LedgerRange {
+    pub start_ledger: u64,
+    pub end_ledger: u64,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum ReplayError {
+    #[error("invalid ledger range: start > end")]
+    InvalidRange,
+    #[error("snapshot generation failed: {0}")]
+    Snapshot(#[from] crate::snapshot::SnapshotError),
+}
+
+/// Historical replays are intentionally pinned to a model version and do not read
+/// wall-clock time. Re-running the same range against the same raw inputs must
+/// produce a byte-identical output payload.
+pub fn replay_historical_range(range: &LedgerRange, rows: &[RawSnapshotRow]) -> Result<Vec<u8>, ReplayError> {
+    if range.start_ledger > range.end_ledger {
+        return Err(ReplayError::InvalidRange);
+    }
+
+    let _ = model_version_for_ledger_range(range.start_ledger, range.end_ledger);
+
+    let filtered: Vec<RawSnapshotRow> = rows
+        .iter()
+        .filter(|row| row.ledger_sequence >= range.start_ledger && row.ledger_sequence <= range.end_ledger)
+        .cloned()
+        .collect();
+
+    // Replay of an empty range is intentionally deterministic and yields an empty
+    // JSON payload rather than reading time or state from the live process.
+    if filtered.is_empty() {
+        return Ok(b"{\"model_version\":\"reliability-v2025.01\",\"records\":[]}".to_vec());
+    }
+
+    Ok(generate_snapshot(&filtered)?)
+}

backend/src/snapshot/generator.rs

@@ -0,0 +1,77 @@
+use serde::{Deserialize, Serialize};
+use std::collections::BTreeMap;
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct RawSnapshotRow {
+    pub ledger_sequence: u64,
+    pub corridor: String,
+    pub source: String,
+    pub reliability: f64,
+    pub volume: f64,
+    pub latency_ms: f64,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct SnapshotRecord {
+    pub corridor: String,
+    pub source: String,
+    pub avg_reliability: f64,
+    pub total_volume: f64,
+    pub avg_latency_ms: f64,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct SnapshotPayload {
+    pub model_version: String,
+    pub records: Vec<SnapshotRecord>,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum SnapshotError {
+    #[error("snapshot input is empty")]
+    EmptyInput,
+    #[error("snapshot serialization failed: {0}")]
+    Serialization(#[from] serde_json::Error),
+}
+
+/// Pure, deterministic snapshot generation.
+///
+/// The function aggregates by corridor using a canonical BTreeMap ordering and
+/// sorts ledger rows before computing averages so the serialized output is
+/// stable across repeated replays of the same historical data.
+pub fn generate_snapshot(rows: &[RawSnapshotRow]) -> Result<Vec<u8>, SnapshotError> {
+    if rows.is_empty() {
+        return Err(SnapshotError::EmptyInput);
+    }
+
+    let mut grouped: BTreeMap<String, Vec<&RawSnapshotRow>> = BTreeMap::new();
+    for row in rows {
+        grouped.entry(row.corridor.clone()).or_default().push(row);
+    }
+
+    let mut records = Vec::with_capacity(grouped.len());
+    for (corridor, entries) in grouped {
+        let mut entries = entries;
+        entries.sort_by_key(|entry| entry.ledger_sequence);
+
+        let total_count = entries.len() as f64;
+        let avg_reliability = entries.iter().map(|entry| entry.reliability).sum::<f64>() / total_count;
+        let total_volume = entries.iter().map(|entry| entry.volume).sum::<f64>();
+        let avg_latency_ms = entries.iter().map(|entry| entry.latency_ms).sum::<f64>() / total_count;
+
+        records.push(SnapshotRecord {
+            corridor,
+            source: entries[0].source.clone(),
+            avg_reliability: avg_reliability,
+            total_volume,
+            avg_latency_ms,
+        });
+    }
+
+    let payload = SnapshotPayload {
+        model_version: crate::snapshot::model_version::PINNED_MODEL_VERSION.to_string(),
+        records,
+    };
+
+    Ok(serde_json::to_vec(&payload)?)
+}

backend/src/snapshot/mod.rs

@@ -0,0 +1,8 @@
+pub mod generator;
+pub mod model_version;
+
+pub use generator::{generate_snapshot, RawSnapshotRow, SnapshotRecord, SnapshotPayload, SnapshotError};
+pub use model_version::{
+    model_version_for_ledger_range, model_version_for_range, network_model_version, DETERMINISM_EFFECTIVE_DATE,
+    PINNED_MODEL_VERSION,
+};

backend/src/snapshot/model_version.rs

@@ -0,0 +1,32 @@
+use serde::{Deserialize, Serialize};
+
+pub const DETERMINISM_EFFECTIVE_DATE: &str = "2025-01-01";
+pub const PINNED_MODEL_VERSION: &str = "reliability-v2025.01";
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ModelVersion {
+    pub name: String,
+    pub effective_from: String,
+    pub effective_until: Option<String>,
+}
+
+/// The historical replay path is pinned to a single model version so that
+/// replaying the same ledger range produces identical output regardless of the
+/// model currently shipped in production.
+///
+/// This guarantee applies to data on or after 2025-01-01, which is the
+/// effective date of the snapshot determinism contract. Model changes after that
+/// date are not retroactively applied to historical replays.
+pub fn model_version_for_range(start_ledger: u64, end_ledger: u64) -> &'static str {
+    let _ = (start_ledger, end_ledger);
+    PINNED_MODEL_VERSION
+}
+
+pub fn model_version_for_ledger_range(start_ledger: u64, end_ledger: u64) -> &'static str {
+    model_version_for_range(start_ledger, end_ledger)
+}
+
+pub fn network_model_version(network: &str) -> &'static str {
+    let _ = network;
+    PINNED_MODEL_VERSION
+}