Implements persistence for Events, and snapshot loading

Cargo.toml

@@ -20,6 +20,7 @@ alloy = { version = "1.0", features = [
     "json-rpc",
     "node-bindings",
 ] }
+anyhow = "1"
 async-lock = "3.0"
 bitvec = "1"
 clap = { version = "4.4", features = ["derive", "env"] }
@@ -36,6 +37,12 @@ im = { version = "15", features = ["serde"] }
 log-panics = { version = "2.0", features = ["with-backtrace"] }
 parking_lot = "0.12"
 serde = { version = "1.0", features = ["derive"] }
+serde_json = "1"
+sqlx = { version = "0.8", features = [
+    "sqlite",
+    "runtime-tokio",
+    "migrate",
+] }
 surf-disco = "0.9"
 tagged-base64 = "0.4"
 tide-disco = "0.9"

migrations/sqlite/01_init_schema.sql

@@ -0,0 +1,50 @@
+
+-- L1 Block tracking
+-- only one row, updated for each finalized block
+CREATE TABLE l1_block (
+    hash TEXT PRIMARY KEY,
+    number BIGINT NOT NULL UNIQUE,
+    parent_hash TEXT NOT NULL,
+    `timestamp` BIGINT NOT NULL,
+    exit_escrow_period BIGINT NOT NULL
+);
+
+-- Full node Set
+-- Stores the finalized node set according to L1 state
+CREATE TABLE node (
+    address TEXT PRIMARY KEY,
+    staking_key TEXT NOT NULL UNIQUE,
+    state_key TEXT NOT NULL UNIQUE,
+    stake TEXT NOT NULL,
+    commission REAL NOT NULL
+);
+
+-- Wallet State
+-- Stores the latest finalized state for each wallet
+CREATE TABLE wallet (
+    address TEXT PRIMARY KEY,
+    claimed_rewards TEXT NOT NULL
+);
+
+-- Delegations, pending withdrawals, and exits
+CREATE TABLE delegation (
+    delegator TEXT NOT NULL REFERENCES wallet (address),
+    node TEXT NOT NULL,
+    -- Store as string to preserve precision for U256
+    amount TEXT NOT NULL,
+    unlocks_at BIGINT NOT NULL,
+    -- Store as string to preserve precision for U256
+    withdrawal_amount TEXT NOT NULL,
+    PRIMARY KEY (delegator, node)
+);
+
+CREATE INDEX delegation_by_node ON delegation (node);
+CREATE INDEX delegation_by_status ON delegation (delegator, unlocks_at, withdrawal_amount);
+
+-- Rewards
+-- Stores total accrued rewards for each account
+CREATE TABLE lifetime_rewards (
+    address TEXT PRIMARY KEY,
+    -- Store as string to preserve precision for U256
+    amount TEXT NOT NULL
+);

src/error.rs

@@ -96,6 +96,18 @@ impl From<RequestError> for Error {
     }
 }
 
+impl From<anyhow::Error> for Error {
+    fn from(err: anyhow::Error) -> Self {
+        Self::internal().context(err)
+    }
+}
+
+impl From<sqlx::Error> for Error {
+    fn from(err: sqlx::Error) -> Self {
+        Self::internal().context(err)
+    }
+}
+
 pub type Result<T, E = Error> = core::result::Result<T, E>;
 
 /// Extension functions for converting other result types into [`Result`].

src/input/l1.rs

@@ -14,6 +14,7 @@ use hotshot_contract_adapter::sol_types::{
     RewardClaim::RewardClaimEvents,
     StakeTableV2::{StakeTableV2Events, ValidatorRegistered, ValidatorRegisteredV2},
 };
+use hotshot_types::light_client::StateVerKey;
 use tracing::instrument;
 
 use crate::{
@@ -468,13 +469,13 @@ impl L1BlockSnapshot {
 #[derive(Clone, Debug, PartialEq)]
 pub struct Snapshot {
     /// The L1 block.
-    block: L1BlockSnapshot,
+    pub block: L1BlockSnapshot,
 
     /// The full node set as of this L1 block.
-    node_set: NodeSet,
+    pub node_set: NodeSet,
 
     /// The state of each wallet as of this L1 block.
-    wallets: Wallets,
+    pub wallets: Wallets,
 }
 
 impl Snapshot {
@@ -657,7 +658,7 @@ impl Wallets {
 #[derive(Clone, Debug, Default, PartialEq)]
 pub struct Wallet {
     /// Nodes that this user is delegating to.
-    nodes: im::Vector<Delegation>,
+    pub nodes: im::Vector<Delegation>,
 
     /// Stake that has been undelegated but not yet withdrawn.
     pub pending_undelegations: im::Vector<PendingWithdrawal>,
@@ -780,6 +781,7 @@ impl From<&ValidatorRegistered> for NodeSetEntry {
         NodeSetEntry {
             address: e.account,
             staking_key: PubKey::from(e.blsVk).into(),
+            state_key: StateVerKey::from(e.schnorrVk).into(),
             commission: Ratio::new(e.commission as usize, COMMISSION_BASIS_POINTS as usize),
             // All nodes start with 0 stake, a separate delegation event will be generated when
             // someone delegates non-zero stake to a node.

src/main.rs

@@ -1,4 +1,4 @@
-use std::{path::PathBuf, process::exit, sync::Arc};
+use std::{process::exit, sync::Arc};
 
 use async_lock::RwLock;
 use clap::{Parser, ValueEnum};
@@ -36,9 +36,9 @@ struct Options {
     #[clap(flatten)]
     l1_options: L1ClientOptions,
 
-    /// Location for persistent storage.
-    #[clap(long, env = "ESPRESSO_STAKING_SERVICE_STORAGE")]
-    storage: PathBuf,
+    /// Persistence options.
+    #[clap(flatten)]
+    persistence: sql::PersistenceOptions,
 
     /// Port for the HTTP server.
     #[clap(
@@ -65,7 +65,7 @@ impl Options {
         let genesis = Snapshot::empty(genesis_block);
 
         let l1_input = RpcStream::new(self.l1_options).await?;
-        let storage = sql::Persistence::new(&self.storage).await?;
+        let storage = sql::Persistence::new(&self.persistence).await?;
 
         // Create server state.
         let l1 = Arc::new(RwLock::new(l1::State::new(storage, genesis).await?));
@@ -161,7 +161,10 @@ mod test {
                 ..Default::default()
             },
             port,
-            storage: tmp.path().join("staking-ui-storage"),
+            persistence: sql::PersistenceOptions {
+                path: tmp.path().join("temp.db"),
+                max_connections: 5,
+            },
             log_format: Some(LogFormat::Json),
         };