feat: db account storage cleanup task

CHANGELOG.md

@@ -4,6 +4,30 @@
 
 ### Enhancements
 
+- Added periodic cleanup of old account data from the database ([#1304](https://github.com/0xMiden/miden-node/issues/1304)).
+- Added support for timeouts in the WASM remote prover clients ([#1383](https://github.com/0xMiden/miden-node/pull/1383)).
+- Added block validation endpoint to validator and integrated with block producer ([#1382](https://github.com/0xMiden/miden-node/pull/1381)).
+- Added support for caching mempool statistics in the block producer server ([#1388](https://github.com/0xMiden/miden-node/pull/1388)).
+- Added mempool statistics to the block producer status in the `miden-network-monitor` binary ([#1392](https://github.com/0xMiden/miden-node/pull/1392)).
+- Added success rate to the `miden-network-monitor` binary ([#1420](https://github.com/0xMiden/miden-node/pull/1420)).
+- Added chain tip to the block producer status ([#1419](https://github.com/0xMiden/miden-node/pull/1419)).
+- The mempool's transaction capacity is now configurable ([#1433](https://github.com/0xMiden/miden-node/pull/1433)).
+- Renamed card's names in the `miden-network-monitor` binary ([#1441](https://github.com/0xMiden/miden-node/pull/1441)).
+- Integrated NTX Builder with validator via `SubmitProvenTransaction` RPC ([#1453](https://github.com/0xMiden/miden-node/pull/1453)).
+- Added pagination to `GetNetworkAccountIds` endpoint ([#1452](https://github.com/0xMiden/miden-node/pull/1452)).
+- Improved tracing in `miden-network-monitor` binary ([#1366](https://github.com/0xMiden/miden-node/pull/1366)).
+- Integrated RPC stack with Validator component for transaction validation ([#1457](https://github.com/0xMiden/miden-node/pull/1457)).
+- Add partial storage map queries to RPC ([#1428](https://github.com/0xMiden/miden-node/pull/1428)).
+- Added validated transactions check to block validation logc in Validator ([#1460](https://github.com/0xMiden/miden-node/pull/1460)).
+- Added explorer status to the `miden-network-monitor` binary ([#1450](https://github.com/0xMiden/miden-node/pull/1450)).
+- Added `GetLimits` endpoint to the RPC server ([#1410](https://github.com/0xMiden/miden-node/pull/1410)).
+- Added gRPC-Web probe support to the `miden-network-monitor` binary ([#1484](https://github.com/0xMiden/miden-node/pull/1484)).
+- Add DB schema change check ([#1268](https://github.com/0xMiden/miden-node/pull/1485)).
+- Improve DB query performance for account queries ([#1496](https://github.com/0xMiden/miden-node/pull/1496).
+- Limit number of storage map keys in `GetAccount` requests ([#1517](https://github.com/0xMiden/miden-node/pull/1517)).
+- The network monitor now marks the chain as unhealthy if it fails to create new blocks ([#1512](https://github.com/0xMiden/miden-node/pull/1512)).
+- Block producer now detects if it is desync'd from the store's chain tip and aborts ([#1520](https://github.com/0xMiden/miden-node/pull/1520)).
+- Pin tool versions in CI ([#1523](https://github.com/0xMiden/miden-node/pull/1523)).
 - [BREAKING] Updated miden-base dependencies to use `next` branch; renamed `NoteInputs` to `NoteStorage`, `.inputs()` to `.storage()`, and database `inputs` column to `storage` ([#1595](https://github.com/0xMiden/miden-node/pull/1595)).
 - [BREAKING] Move block proving from Blocker Producer to the Store ([#1579](https://github.com/0xMiden/miden-node/pull/1579)).
 

crates/store/src/db/mod.rs

@@ -23,7 +23,7 @@ use miden_protocol::note::{
 use miden_protocol::transaction::TransactionId;
 use miden_protocol::utils::{Deserializable, Serializable};
 use tokio::sync::oneshot;
-use tracing::{Instrument, info, instrument};
+use tracing::{Instrument, info, info_span, instrument};
 
 use crate::COMPONENT;
 use crate::db::manager::{ConnectionManager, configure_connection_on_creation};
@@ -54,8 +54,10 @@ pub(crate) mod schema;
 
 pub type Result<T, E = DatabaseError> = std::result::Result<T, E>;
 
+#[derive(Clone)]
 pub struct Db {
     pool: deadpool_diesel::Pool<ConnectionManager, deadpool::managed::Object<ConnectionManager>>,
+    notify_cleanup_task: tokio::sync::mpsc::Sender<BlockNumber>,
 }
 
 /// Describes the value of an asset for an account ID at `block_num` specifically.
@@ -314,6 +316,8 @@ impl Db {
     }
 
     /// Open a connection to the DB and apply any pending migrations.
+    ///
+    /// This also spawns a background task that handles periodic cleanup of old account data.
     #[instrument(target = COMPONENT, skip_all)]
     pub async fn load(database_filepath: PathBuf) -> Result<Self, DatabaseSetupError> {
         let manager = ConnectionManager::new(database_filepath.to_str().unwrap());
@@ -325,8 +329,20 @@ impl Db {
             "Connected to the database"
         );
 
-        let me = Db { pool };
+        // Create channel for cleanup notifications
+        // Buffer size of 2 is sufficient since cleanup coalesces multiple notifications
+        let (notify_cleanup_task, rx) = tokio::sync::mpsc::channel(2);
+
+        let me = Db { pool, notify_cleanup_task };
+        let me2 = me.clone();
+
+        // Spawn background cleanup task
+        // TODO: retain the join handle to coordinate shutdown or surface task failures.
+        let _cleanup_task_handle =
+            tokio::spawn(async move { Self::periodic_cleanup_task(me2, rx).await });
+
         me.query("migrations", apply_migrations).await?;
+
         Ok(me)
     }
 
@@ -586,28 +602,171 @@ impl Db {
         signed_block: SignedBlock,
         notes: Vec<(NoteRecord, Option<Nullifier>)>,
     ) -> Result<()> {
-        self.transact("apply block", move |conn| -> Result<()> {
-            models::queries::apply_block(
-                conn,
-                signed_block.header(),
-                signed_block.signature(),
-                &notes,
-                signed_block.body().created_nullifiers(),
-                signed_block.body().updated_accounts(),
-                signed_block.body().transactions(),
-            )?;
-
-            // XXX FIXME TODO free floating mutex MUST NOT exist
-            // it doesn't bind it properly to the data locked!
-            if allow_acquire.send(()).is_err() {
-                tracing::warn!(target: COMPONENT, "failed to send notification for successful block application, potential deadlock");
+        let block_num = signed_block.header().block_num();
+
+        let result = self
+            .transact("apply block", move |conn| -> Result<()> {
+                // TODO: This span is logged in a root span, we should connect it to the parent one.
+                let _span = info_span!(target: COMPONENT, "write_block_to_db").entered();
+
+                models::queries::apply_block(
+                    conn,
+                    signed_block.header(),
+                    signed_block.signature(),
+                    &notes,
+                    signed_block.body().created_nullifiers(),
+                    signed_block.body().updated_accounts(),
+                    signed_block.body().transactions(),
+                )?;
+
+                // XXX FIXME TODO free floating mutex MUST NOT exist
+                // it doesn't bind it properly to the data locked!
+                if allow_acquire.send(()).is_err() {
+                    tracing::warn!(target: COMPONENT, "failed to send notification for successful block application, potential deadlock");
+                }
+
+                acquire_done.blocking_recv()?;
+
+                Ok(())
+            })
+            .await;
+
+        // Notify the cleanup task of the latest applied block
+        // Ignore errors since cleanup is non-critical and shouldn't block block application
+        // TODO: track dropped cleanup notifications to surface backpressure.
+        let _res = self.notify_cleanup_task.try_send(block_num);
+
+        result
+    }
+
+    /// Background task that handles periodic cleanup of old account data.
+    ///
+    /// This task runs indefinitely, receiving block numbers from the `apply_block` method
+    /// and triggering cleanup whenever new blocks are available. The cleanup process:
+    ///
+    /// 1. Batches incoming notifications using `recv_many` to avoid excessive cleanup operations
+    /// 2. Only processes the most recent block number from the batch (coalescing multiple updates)
+    /// 3. Runs cleanup with a 30-second timeout to prevent blocking
+    /// 4. Logs success or failure but continues running regardless of cleanup outcome
+    ///
+    /// # Batching Strategy
+    ///
+    /// The batching approach ensures that if multiple blocks are applied quickly (e.g., during
+    /// initial sync), only the latest block number triggers cleanup. This prevents redundant
+    /// cleanup operations while ensuring cleanup runs on the most recent state.
+    ///
+    /// # Error Handling
+    ///
+    /// This task never exits on cleanup errors. Cleanup failures are logged but the task
+    /// continues to process future blocks. This ensures that temporary issues (like database
+    /// locks or high load) don't permanently disable the cleanup mechanism.
+    ///
+    /// The task only exits if the channel is closed (i.e., all `Db` instances are dropped),
+    /// which typically happens during application shutdown.
+    async fn periodic_cleanup_task(db: Self, mut notify: tokio::sync::mpsc::Receiver<BlockNumber>) {
+        let mut buf = Vec::with_capacity(128);
+
+        loop {
+            // Receive many notifications at once to batch them
+            // If the channel is closed (returns 0), exit the task
+            let received = notify.recv_many(&mut buf, 128).await;
+            if received == 0 {
+                tracing::info!(target: COMPONENT, "Cleanup task shutting down: channel closed");
+                break;
             }
 
-            acquire_done.blocking_recv()?;
+            // Only process the most recent block number from the batch
+            // This coalesces multiple cleanup requests during fast block processing
+            if let Some(block_num) = buf.pop() {
+                match db.run_periodic_cleanup(block_num).await {
+                    Ok((vault_deleted, storage_deleted)) => {
+                        tracing::info!(
+                            target: COMPONENT,
+                            block_num = block_num.as_u32(),
+                            vault_assets_deleted = vault_deleted,
+                            storage_map_values_deleted = storage_deleted,
+                            "Periodic cleanup completed successfully"
+                        );
+                    },
+                    Err(e) => {
+                        tracing::warn!(
+                            target: COMPONENT,
+                            block_num = block_num.as_u32(),
+                            error = %e,
+                            "Periodic cleanup failed, will retry on next block"
+                        );
+                    },
+                }
+            }
 
-            Ok(())
-        })
-        .await
+            // Clear the buffer for the next batch
+            buf.clear();
+        }
+    }
+
+    /// Runs periodic cleanup of old account data with a timeout.
+    ///
+    /// This function cleans up old vault asset and storage map value entries for all accounts,
+    /// keeping only the latest entry and up to MAX_HISTORICAL_ENTRIES_PER_ACCOUNT historical
+    /// entries per key.
+    ///
+    /// The cleanup operation has a 30-second timeout to prevent it from blocking for too long.
+    /// If the timeout is reached, the cleanup is aborted and returns an error.
+    ///
+    /// # Parameters
+    /// * `block_num` - The block number at which cleanup was triggered (used for logging)
+    ///
+    /// # Returns
+    /// A tuple of (vault_assets_deleted, storage_map_values_deleted) on success, or an error
+    /// if the operation fails or times out.
+    #[instrument(level = "debug", target = COMPONENT, skip(self), fields(block_num = %block_num.as_u32()))]
+    async fn run_periodic_cleanup(&self, block_num: BlockNumber) -> Result<(usize, usize)> {
+        use std::time::Duration;
+
+        let cleanup_timeout = Duration::from_secs(30);
+        let start = std::time::Instant::now();
+
+        let cleanup_task = self.transact("periodic cleanup", move |conn| {
+            models::queries::cleanup_all_accounts(conn, block_num)
+        });
+
+        // Run cleanup with timeout
+        let result = tokio::time::timeout(cleanup_timeout, cleanup_task).await;
+
+        let duration = start.elapsed();
+
+        match result {
+            Ok(Ok((vault_deleted, storage_deleted))) => {
+                tracing::info!(
+                    target: COMPONENT,
+                    block_num = block_num.as_u32(),
+                    vault_assets_deleted = vault_deleted,
+                    storage_map_values_deleted = storage_deleted,
+                    duration_ms = duration.as_millis(),
+                    "Cleanup completed within timeout"
+                );
+                Ok((vault_deleted, storage_deleted))
+            },
+            Ok(Err(e)) => {
+                tracing::error!(
+                    target: COMPONENT,
+                    block_num = block_num.as_u32(),
+                    duration_ms = duration.as_millis(),
+                    error = %e,
+                    "Cleanup failed"
+                );
+                Err(e)
+            },
+            Err(_timeout_err) => {
+                tracing::warn!(
+                    target: COMPONENT,
+                    block_num = block_num.as_u32(),
+                    timeout_ms = cleanup_timeout.as_millis(),
+                    "Cleanup timed out - operation was aborted"
+                );
+                Err(DatabaseError::QueryTimeout("periodic cleanup".to_string()))
+            },
+        }
     }
 
     /// Selects storage map values for syncing storage maps for a specific account ID.

crates/store/src/db/models/queries/accounts.rs

@@ -1252,3 +1252,97 @@ pub(crate) struct AccountStorageMapRowInsert {
     pub(crate) value: Vec<u8>,
     pub(crate) is_latest: bool,
 }
+
+// CLEANUP FUNCTIONS
+// ================================================================================================
+
+/// Number of historical blocks to retain for vault assets and storage map values.
+/// Entries older than `chain_tip - HISTORICAL_BLOCK_RETENTION` will be deleted,
+/// except for entries marked with `is_latest=true` which are always retained.
+pub const HISTORICAL_BLOCK_RETENTION: u32 = 50;
+
+/// Clean up old vault asset entries for a specific account, deleting entries older than
+/// the retention window.
+///
+/// Keeps entries where `block_num >= cutoff_block` OR `is_latest = true`.
+#[cfg(test)]
+pub(crate) fn cleanup_old_account_vault_assets(
+    conn: &mut SqliteConnection,
+    account_id: AccountId,
+    chain_tip: BlockNumber,
+) -> Result<usize, DatabaseError> {
+    let account_id_bytes = account_id.to_bytes();
+    let cutoff_block = i64::from(chain_tip.as_u32().saturating_sub(HISTORICAL_BLOCK_RETENTION));
+
+    diesel::sql_query(
+        r#"
+        DELETE FROM account_vault_assets
+        WHERE account_id = ?1 AND block_num < ?2 AND is_latest = 0
+        "#,
+    )
+    .bind::<diesel::sql_types::Binary, _>(&account_id_bytes)
+    .bind::<diesel::sql_types::BigInt, _>(cutoff_block)
+    .execute(conn)
+    .map_err(DatabaseError::Diesel)
+}
+
+/// Clean up old storage map value entries for a specific account, deleting entries older than
+/// the retention window.
+///
+/// Keeps entries where `block_num >= cutoff_block` OR `is_latest = true`.
+#[cfg(test)]
+pub(crate) fn cleanup_old_account_storage_map_values(
+    conn: &mut SqliteConnection,
+    account_id: AccountId,
+    chain_tip: BlockNumber,
+) -> Result<usize, DatabaseError> {
+    let account_id_bytes = account_id.to_bytes();
+    let cutoff_block = i64::from(chain_tip.as_u32().saturating_sub(HISTORICAL_BLOCK_RETENTION));
+
+    diesel::sql_query(
+        r#"
+        DELETE FROM account_storage_map_values
+        WHERE account_id = ?1 AND block_num < ?2 AND is_latest = 0
+        "#,
+    )
+    .bind::<diesel::sql_types::Binary, _>(&account_id_bytes)
+    .bind::<diesel::sql_types::BigInt, _>(cutoff_block)
+    .execute(conn)
+    .map_err(DatabaseError::Diesel)
+}
+
+/// Clean up old entries for all accounts, deleting entries older than the retention window.
+///
+/// Deletes rows where `block_num < chain_tip - HISTORICAL_BLOCK_RETENTION` and `is_latest = false`.
+/// This is a simple and efficient approach that doesn't require window functions.
+///
+/// # Returns
+/// A tuple of (vault_assets_deleted, storage_map_values_deleted)
+pub fn cleanup_all_accounts(
+    conn: &mut SqliteConnection,
+    chain_tip: BlockNumber,
+) -> Result<(usize, usize), DatabaseError> {
+    let cutoff_block = i64::from(chain_tip.as_u32().saturating_sub(HISTORICAL_BLOCK_RETENTION));
+
+    let vault_deleted = diesel::sql_query(
+        r#"
+        DELETE FROM account_vault_assets
+        WHERE block_num < ?1 AND is_latest = 0
+        "#,
+    )
+    .bind::<diesel::sql_types::BigInt, _>(cutoff_block)
+    .execute(conn)
+    .map_err(DatabaseError::Diesel)?;
+
+    let storage_deleted = diesel::sql_query(
+        r#"
+        DELETE FROM account_storage_map_values
+        WHERE block_num < ?1 AND is_latest = 0
+        "#,
+    )
+    .bind::<diesel::sql_types::BigInt, _>(cutoff_block)
+    .execute(conn)
+    .map_err(DatabaseError::Diesel)?;
+
+    Ok((vault_deleted, storage_deleted))
+}