feat: Validator database

bin/node/src/commands/bundled.rs

@@ -257,6 +257,7 @@ impl BundledCommand {
                         address: validator_address,
                         grpc_timeout,
                         signer,
+                        data_directory,
                     }
                     .serve()
                     .await

bin/node/src/commands/validator.rs

@@ -1,3 +1,4 @@
+use std::path::PathBuf;
 use std::time::Duration;
 
 use anyhow::Context;
@@ -9,6 +10,7 @@ use url::Url;
 
 use crate::commands::{
     DEFAULT_TIMEOUT,
+    ENV_DATA_DIRECTORY,
     ENV_ENABLE_OTEL,
     ENV_VALIDATOR_INSECURE_SECRET_KEY,
     ENV_VALIDATOR_URL,
@@ -40,6 +42,10 @@ pub enum ValidatorCommand {
         )]
         grpc_timeout: Duration,
 
+        /// Directory in which to store the database and raw block data.
+        #[arg(long, env = ENV_DATA_DIRECTORY, value_name = "DIR")]
+        data_directory: PathBuf,
+
         /// Insecure, hex-encoded validator secret key for development and testing purposes.
         ///
         /// If not provided, a predefined key is used.
@@ -51,18 +57,27 @@ pub enum ValidatorCommand {
 impl ValidatorCommand {
     pub async fn handle(self) -> anyhow::Result<()> {
         let Self::Start {
-            url, grpc_timeout, insecure_secret_key, ..
+            url,
+            grpc_timeout,
+            insecure_secret_key,
+            data_directory,
+            ..
         } = self;
 
         let address =
             url.to_socket().context("Failed to extract socket address from validator URL")?;
 
         let signer = SecretKey::read_from_bytes(hex::decode(insecure_secret_key)?.as_ref())?;
 
-        Validator { address, grpc_timeout, signer }
-            .serve()
-            .await
-            .context("failed while serving validator component")
+        Validator {
+            address,
+            grpc_timeout,
+            signer,
+            data_directory,
+        }
+        .serve()
+        .await
+        .context("failed while serving validator component")
     }
 
     pub fn is_open_telemetry_enabled(&self) -> bool {

crates/store/src/db/manager.rs

@@ -36,12 +36,12 @@ impl ConnectionManagerError {
 /// Create a connection manager with per-connection setup
 ///
 /// Particularly, `foreign_key` checks are enabled and using a write-append-log for journaling.
-pub(crate) struct ConnectionManager {
+pub struct ConnectionManager {
     pub(crate) manager: deadpool_diesel::sqlite::Manager,
 }
 
 impl ConnectionManager {
-    pub(crate) fn new(database_path: &str) -> Self {
+    pub fn new(database_path: &str) -> Self {
         let manager = deadpool_diesel::sqlite::Manager::new(
             database_path.to_owned(),
             deadpool_diesel::sqlite::Runtime::Tokio1,

crates/store/src/db/mod.rs

@@ -219,6 +219,11 @@ impl From<NoteRecord> for NoteSyncRecord {
 }
 
 impl Db {
+    /// Creates a new database instance with the provided connection pool.
+    pub fn new(pool: deadpool_diesel::Pool<ConnectionManager>) -> Self {
+        Self { pool }
+    }
+
     /// Creates a new database and inserts the genesis block.
     #[instrument(
         target = COMPONENT,
@@ -260,7 +265,7 @@ impl Db {
     }
 
     /// Create and commit a transaction with the queries added in the provided closure
-    pub(crate) async fn transact<R, E, Q, M>(&self, msg: M, query: Q) -> std::result::Result<R, E>
+    pub async fn transact<R, E, Q, M>(&self, msg: M, query: Q) -> std::result::Result<R, E>
     where
         Q: Send
             + for<'a, 't> FnOnce(&'a mut SqliteConnection) -> std::result::Result<R, E>
@@ -285,7 +290,7 @@ impl Db {
     }
 
     /// Run the query _without_ a transaction
-    pub(crate) async fn query<R, E, Q, M>(&self, msg: M, query: Q) -> std::result::Result<R, E>
+    pub async fn query<R, E, Q, M>(&self, msg: M, query: Q) -> std::result::Result<R, E>
     where
         Q: Send + FnOnce(&mut SqliteConnection) -> std::result::Result<R, E> + 'static,
         R: Send + 'static,

crates/store/src/db/models/conv.rs

@@ -50,7 +50,7 @@ pub struct DatabaseTypeConversionError {
 /// Convert from and to it's database representation and back
 ///
 /// We do not assume sanity of DB types.
-pub(crate) trait SqlTypeConvert: Sized {
+pub trait SqlTypeConvert: Sized {
     type Raw: Sized;
 
     fn to_raw_sql(self) -> Self::Raw;

crates/store/src/lib.rs

@@ -10,6 +10,10 @@ pub mod state;
 #[cfg(feature = "rocksdb")]
 pub use accounts::PersistentAccountTree;
 pub use accounts::{AccountTreeWithHistory, HistoricalError, InMemoryAccountTree};
+pub use db::Db;
+pub use db::manager::ConnectionManager;
+pub use db::models::conv::SqlTypeConvert;
+pub use errors::{DatabaseError, DatabaseSetupError};
 pub use genesis::GenesisState;
 pub use server::{DataDirectory, Store};
 

crates/validator/Cargo.toml

@@ -18,7 +18,11 @@ workspace = true
 
 [dependencies]
 anyhow                 = { workspace = true }
+deadpool-diesel        = { features = ["sqlite"], version = "0.6" }
+diesel                 = { features = ["numeric", "sqlite"], version = "2.2" }
+diesel_migrations      = { features = ["sqlite"], version = "2.2" }
 miden-node-proto       = { workspace = true }
+miden-node-store       = { workspace = true }
 miden-node-proto-build = { features = ["internal"], workspace = true }
 miden-node-utils       = { features = ["testing"], workspace = true }
 miden-protocol         = { workspace = true }

crates/validator/build.rs

@@ -0,0 +1,9 @@
+// This build.rs is required to trigger the `diesel_migrations::embed_migrations!` proc-macro in
+// `validator/src/db/migrations.rs` to include the latest version of the migrations into the binary, see <https://docs.rs/diesel_migrations/latest/diesel_migrations/macro.embed_migrations.html#automatic-rebuilds>.
+fn main() {
+    println!("cargo:rerun-if-changed=./src/db/migrations");
+    // If we do one re-write, the default rules are disabled,
+    // hence we need to trigger explicitly on `Cargo.toml`.
+    // <https://doc.rust-lang.org/cargo/reference/build-scripts.html#rerun-if-changed>
+    println!("cargo:rerun-if-changed=Cargo.toml");
+}

crates/validator/diesel.toml

@@ -0,0 +1,5 @@
+# For documentation on how to configure this file,
+# see https://diesel.rs/guides/configuring-diesel-cli
+
+[print_schema]
+file = "src/db/schema.rs"

crates/validator/src/block_validation/mod.rs

@@ -1,12 +1,13 @@
-use std::sync::Arc;
+use std::collections::HashMap;
 
+use miden_node_store::{DatabaseError, Db};
 use miden_protocol::block::{BlockNumber, BlockSigner, ProposedBlock};
 use miden_protocol::crypto::dsa::ecdsa_k256_keccak::Signature;
 use miden_protocol::errors::ProposedBlockError;
-use miden_protocol::transaction::TransactionId;
-use tracing::{Instrument, info_span};
+use miden_protocol::transaction::{TransactionHeader, TransactionId};
+use tracing::info_span;
 
-use crate::server::ValidatedTransactions;
+use crate::db::select_transactions;
 
 // BLOCK VALIDATION ERROR
 // ================================================================================================
@@ -15,8 +16,19 @@ use crate::server::ValidatedTransactions;
 pub enum BlockValidationError {
     #[error("transaction {0} in block {1} has not been validated")]
     TransactionNotValidated(TransactionId, BlockNumber),
+    #[error(
+        "the proposed transaction {proposed_tx} does not match the validated transaction {validated_tx}"
+    )]
+    TransactionMismatch {
+        proposed_tx: TransactionId,
+        validated_tx: TransactionId,
+    },
     #[error("failed to build block")]
     BlockBuildingFailed(#[from] ProposedBlockError),
+    #[error("failed to select transactions")]
+    DatabaseError(#[from] DatabaseError),
+    #[error("internal error: {0}")]
+    Other(String),
 }
 
 // BLOCK VALIDATION
@@ -29,23 +41,39 @@ pub enum BlockValidationError {
 pub async fn validate_block<S: BlockSigner>(
     proposed_block: ProposedBlock,
     signer: &S,
-    validated_transactions: Arc<ValidatedTransactions>,
+    db: &Db,
 ) -> Result<Signature, BlockValidationError> {
-    // Check that all transactions in the proposed block have been validated
     let verify_span = info_span!("verify_transactions");
-    for tx_header in proposed_block.transactions() {
-        let tx_id = tx_header.id();
-        // TODO: LruCache is a poor abstraction since it locks many times.
-        if validated_transactions
-            .get(&tx_id)
-            .instrument(verify_span.clone())
-            .await
-            .is_none()
-        {
+
+    // Create a map of transactions from the proposed block.
+    let proposed_transactions = proposed_block
+        .transactions()
+        .map(|header| (header.id(), header.clone()))
+        .collect::<HashMap<TransactionId, TransactionHeader>>();
+    // Retrieve all validated transactions pertaining to the proposed block.
+    let tx_ids = proposed_block.transactions().map(TransactionHeader::id).collect::<Vec<_>>();
+    let query_tx_ids = tx_ids.clone();
+    let validated_transactions = db
+        .transact("select_transactions", move |conn| select_transactions(conn, &query_tx_ids))
+        .await?;
+
+    // Check that every transaction from the proposed block has been validated.
+    for tx_id in tx_ids {
+        let Some(validated_tx) = validated_transactions.get(&tx_id) else {
             return Err(BlockValidationError::TransactionNotValidated(
                 tx_id,
                 proposed_block.block_num(),
             ));
+        };
+        // Check that the proposed and validated transactions are equal.
+        let proposed_tx = proposed_transactions.get(&tx_id).ok_or(BlockValidationError::Other(
+            "proposed transactions mapped incorrectly".into(),
+        ))?;
+        if validated_tx != proposed_tx {
+            return Err(BlockValidationError::TransactionMismatch {
+                proposed_tx: proposed_tx.id(),
+                validated_tx: validated_tx.id(),
+            });
         }
     }
 

crates/validator/src/db/migrations.rs

@@ -0,0 +1,25 @@
+use diesel::SqliteConnection;
+use diesel_migrations::{EmbeddedMigrations, MigrationHarness, embed_migrations};
+use miden_node_store::DatabaseError;
+use tracing::instrument;
+
+use crate::COMPONENT;
+
+// The rebuild is automatically triggered by `build.rs` as described in
+// <https://docs.rs/diesel_migrations/latest/diesel_migrations/macro.embed_migrations.html#automatic-rebuilds>.
+pub const MIGRATIONS: EmbeddedMigrations = embed_migrations!("src/db/migrations");
+
+#[instrument(level = "debug", target = COMPONENT, skip_all, err)]
+pub fn apply_migrations(conn: &mut SqliteConnection) -> std::result::Result<(), DatabaseError> {
+    let migrations = conn.pending_migrations(MIGRATIONS).expect("In memory migrations never fail");
+    tracing::info!(target = COMPONENT, "Applying {} migration(s)", migrations.len());
+
+    let Err(e) = conn.run_pending_migrations(MIGRATIONS) else {
+        return Ok(());
+    };
+    tracing::warn!(target = COMPONENT, "Failed to apply migration: {e:?}");
+    conn.revert_last_migration(MIGRATIONS)
+        .expect("Duality is maintained by the developer");
+
+    Ok(())
+}

crates/validator/src/db/migrations/2025062000000_setup/up.sql

@@ -0,0 +1,15 @@
+CREATE TABLE transactions (
+    transaction_id               BLOB    NOT NULL,
+    account_id                   BLOB    NOT NULL,
+    block_num                    INTEGER NOT NULL, -- Block number in which the transaction was included.
+    initial_state_commitment     BLOB    NOT NULL, -- State of the account before applying the transaction.
+    final_state_commitment       BLOB    NOT NULL, -- State of the account after applying the transaction.
+    input_notes                  BLOB    NOT NULL, -- Serialized vector with the Nullifier of the input notes.
+    output_notes                 BLOB    NOT NULL, -- Serialized vector with the NoteId of the output notes.
+    size_in_bytes                INTEGER NOT NULL, -- Estimated size of the row in bytes, considering the size of the input and output notes.
+
+    PRIMARY KEY (transaction_id)
+) WITHOUT ROWID;
+
+CREATE INDEX idx_transactions_account_id ON transactions(account_id);
+CREATE INDEX idx_transactions_block_num ON transactions(block_num);