Merge pull request #3834 from dusk-network/blob_expire

node: delete expired blobs

Author: herr-seppia

core/src/transfer.rs

@@ -304,6 +304,15 @@ impl Transaction {
         }
     }
 
+    /// Returns the Blob used with the transaction, if any.
+    #[must_use]
+    pub fn blob_mut(&mut self) -> Option<&mut Vec<BlobData>> {
+        match self {
+            Self::Phoenix(tx) => tx.blob_mut(),
+            Self::Moonlight(tx) => tx.blob_mut(),
+        }
+    }
+
     /// Extracts and removes the blob sidecar from the transaction, if any, and
     /// returns it as a vector of tuples containing the blob hash and the
     /// corresponding blob sidecar.

node/src/chain.rs

@@ -62,6 +62,7 @@ pub struct ChainSrv<N: Network, DB: database::DB, VM: vm::VMExecution> {
     genesis_timestamp: u64,
     dusk_key: BlsPublicKey,
     finality_activation: u64,
+    blob_expire_after: u64,
     #[cfg(feature = "archive")]
     archive: Archive,
 }
@@ -96,6 +97,7 @@ impl<N: Network, DB: database::DB, VM: vm::VMExecution>
             self.event_sender.clone(),
             self.dusk_key,
             self.finality_activation,
+            self.blob_expire_after,
         )
         .await?;
 
@@ -249,13 +251,15 @@ impl<N: Network, DB: database::DB, VM: vm::VMExecution>
 }
 
 impl<N: Network, DB: database::DB, VM: vm::VMExecution> ChainSrv<N, DB, VM> {
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         keys_path: String,
         max_inbound_size: usize,
         event_sender: Sender<Event>,
         genesis_timestamp: u64,
         dusk_key: BlsPublicKey,
         finality_activation: u64,
+        blob_expire_after: u64,
         #[cfg(feature = "archive")] archive: Archive,
     ) -> Self {
         info!(
@@ -272,6 +276,7 @@ impl<N: Network, DB: database::DB, VM: vm::VMExecution> ChainSrv<N, DB, VM> {
             genesis_timestamp,
             dusk_key,
             finality_activation,
+            blob_expire_after,
             #[cfg(feature = "archive")]
             archive,
         }

node/src/chain/acceptor.rs

@@ -103,6 +103,7 @@ pub(crate) struct Acceptor<N: Network, DB: database::DB, VM: vm::VMExecution> {
     dusk_key: bls::PublicKey,
 
     finality_activation: u64,
+    blob_expire_after: u64,
 }
 
 impl<DB: database::DB, VM: vm::VMExecution, N: Network> Drop
@@ -206,6 +207,7 @@ impl<DB: database::DB, VM: vm::VMExecution, N: Network> Acceptor<N, DB, VM> {
         event_sender: Sender<Event>,
         dusk_key: bls::PublicKey,
         finality_activation: u64,
+        blob_expire_after: u64,
     ) -> anyhow::Result<Self> {
         let tip_height = tip.inner().header().height;
         let tip_state_hash = tip.inner().header().state_hash;
@@ -235,6 +237,7 @@ impl<DB: database::DB, VM: vm::VMExecution, N: Network> Acceptor<N, DB, VM> {
             event_sender,
             dusk_key,
             finality_activation,
+            blob_expire_after,
         };
 
         // NB. After restart, state_root returned by VM is always the last
@@ -888,6 +891,9 @@ impl<DB: database::DB, VM: vm::VMExecution, N: Network> Acceptor<N, DB, VM> {
             {
                 let legacy = blk.header().height < self.finality_activation;
 
+                let new_final_heights: Vec<_> =
+                    new_finals.keys().cloned().collect();
+
                 let (_, new_final_state) =
                     new_finals.pop_last().expect("new_finals to be not empty");
                 let new_final_state_root = new_final_state.state_root;
@@ -904,6 +910,24 @@ impl<DB: database::DB, VM: vm::VMExecution, N: Network> Acceptor<N, DB, VM> {
                 };
 
                 vm.finalize_state(new_final_state_root, old_final_state_roots)?;
+
+                if self.blob_expire_after > 0 {
+                    let _ = self.db.read().await.update(|db| {
+                        for height in new_final_heights {
+                            if height < self.blob_expire_after {
+                                // Skip the check for first finalized blocks and prevent underflow error
+                                continue;
+                            }
+                            let expired_block = height - self.blob_expire_after;
+                            let _ = db.delete_blobs_by_height(expired_block).inspect_err(|e| {
+                                warn!("Error while deleting blobs for finalized block {expired_block}: {e}");
+                            });
+                        }
+                        Ok(())
+                    }).inspect_err(|e| {
+                        warn!("Error while deleting blobs while accepting block {}: {e}", blk.header().height);
+                    });
+                }
             }
 
             anyhow::Ok((label, finalized))

node/src/database.rs

@@ -78,6 +78,16 @@ pub trait Ledger {
 
     fn store_blob_data(&self, hash: &[u8; 32], data: Vec<u8>) -> Result<()>;
     fn blob_data_by_hash(&self, hash: &[u8; 32]) -> Result<Option<Vec<u8>>>;
+    fn store_blobs_height(
+        &self,
+        block_height: u64,
+        blob_hashes: &[[u8; 32]],
+    ) -> Result<()>;
+    fn blobs_by_height(
+        &self,
+        block_height: u64,
+    ) -> Result<Option<Vec<[u8; 32]>>>;
+    fn delete_blobs_by_height(&self, block_height: u64) -> Result<()>;
 
     fn block_exists(&self, hash: &[u8]) -> Result<bool>;
 

node/src/database/rocksdb.rs

@@ -12,6 +12,7 @@ use std::sync::Arc;
 use std::{io, vec};
 
 use anyhow::Result;
+use dusk_core::transfer::data::BlobSidecar;
 use node_data::ledger::{
     Block, Fault, Header, Label, SpendingId, SpentTransaction, Transaction,
 };
@@ -34,6 +35,7 @@ use crate::database::Mempool;
 const CF_LEDGER_HEADER: &str = "cf_ledger_header";
 const CF_LEDGER_TXS: &str = "cf_ledger_txs";
 const CF_LEDGER_BLOBS: &str = "cf_ledger_blobs";
+const CF_LEDGER_BLOBS_HEIGHT: &str = "cf_ledger_blobs_height";
 const CF_LEDGER_FAULTS: &str = "cf_ledger_faults";
 const CF_LEDGER_HEIGHT: &str = "cf_ledger_height";
 const CF_CANDIDATES: &str = "cf_candidates";
@@ -128,6 +130,11 @@ impl Backend {
             .cf_handle(CF_LEDGER_BLOBS)
             .expect("CF_LEDGER_BLOBS column family must exist");
 
+        let ledger_blobs_height_cf = self
+            .rocksdb
+            .cf_handle(CF_LEDGER_BLOBS_HEIGHT)
+            .expect("CF_LEDGER_BLOBS_HEIGHT column family must exist");
+
         DBTransaction::<'_, OptimisticTransactionDB> {
             inner,
             candidates_cf,
@@ -141,6 +148,7 @@ impl Backend {
             fees_cf,
             ledger_height_cf,
             ledger_blobs_cf,
+            ledger_blobs_height_cf,
             metadata_cf,
             cumulative_inner_size: RefCell::new(0),
         }
@@ -211,6 +219,10 @@ impl DB for Backend {
                 CF_LEDGER_BLOBS,
                 blocks_cf_opts.clone(),
             ),
+            ColumnFamilyDescriptor::new(
+                CF_LEDGER_BLOBS_HEIGHT,
+                blocks_cf_opts.clone(),
+            ),
             ColumnFamilyDescriptor::new(CF_CANDIDATES, blocks_cf_opts.clone()),
             ColumnFamilyDescriptor::new(
                 CF_CANDIDATES_HEIGHT,
@@ -305,6 +317,7 @@ pub struct DBTransaction<'db, DB: DBAccess> {
     ledger_txs_cf: &'db ColumnFamily,
     ledger_height_cf: &'db ColumnFamily,
     ledger_blobs_cf: &'db ColumnFamily,
+    ledger_blobs_height_cf: &'db ColumnFamily,
 
     // Mempool column families
     mempool_cf: &'db ColumnFamily,
@@ -347,15 +360,19 @@ impl<DB: DBAccess> Ledger for DBTransaction<'_, DB> {
         {
             let cf = self.ledger_txs_cf;
 
+            let mut stored_blobs = Vec::with_capacity(6);
+
             // store all block transactions
             for tx in txs {
                 let mut d = vec![];
+
                 if tx.inner.inner.blob().is_some() {
                     let mut strip_tx = tx.clone();
                     if let Some(blobs) = strip_tx.inner.inner.strip_blobs() {
                         for (hash, sidecar) in blobs.into_iter() {
                             let sidecar_bytes = sidecar.to_var_bytes();
                             self.store_blob_data(&hash, sidecar_bytes)?;
+                            stored_blobs.push(hash);
                         }
                     }
                     strip_tx.write(&mut d)?;
@@ -364,6 +381,11 @@ impl<DB: DBAccess> Ledger for DBTransaction<'_, DB> {
                 }
                 self.put_cf(cf, tx.inner.id(), d)?;
             }
+
+            if !stored_blobs.is_empty() {
+                // Store all blobs hashes in the ledger
+                self.store_blobs_height(header.height, &stored_blobs)?;
+            }
         }
 
         // COLUMN FAMILY: CF_LEDGER_FAULTS
@@ -438,6 +460,7 @@ impl<DB: DBAccess> Ledger for DBTransaction<'_, DB> {
             self.inner.delete_cf(self.ledger_faults_cf, f.id())?;
         }
 
+        self.delete_blobs_by_height(b.header().height)?;
         self.inner.delete_cf(self.ledger_cf, b.header().hash)?;
 
         Ok(())
@@ -477,6 +500,61 @@ impl<DB: DBAccess> Ledger for DBTransaction<'_, DB> {
         self.inner.put_cf(self.ledger_blobs_cf, hash, data)?;
         Ok(())
     }
+    fn store_blobs_height(
+        &self,
+        block_height: u64,
+        blob_hashes: &[[u8; 32]],
+    ) -> Result<()> {
+        if blob_hashes.is_empty() {
+            return Ok(());
+        }
+        let blob_hashes_bytes: Vec<_> =
+            blob_hashes.iter().flat_map(|hash| hash.to_vec()).collect();
+        self.inner.put_cf(
+            self.ledger_blobs_height_cf,
+            block_height.to_be_bytes(),
+            blob_hashes_bytes,
+        )?;
+        Ok(())
+    }
+
+    fn delete_blobs_by_height(&self, block_height: u64) -> Result<()> {
+        let blobs_to_delete = self.blobs_by_height(block_height)?;
+        if let Some(blob_hashes) = blobs_to_delete {
+            for hash in blob_hashes {
+                // What happen if the blobs also exists linked to another
+                // transaction?
+                self.inner.delete_cf(self.ledger_blobs_cf, hash)?;
+            }
+            self.inner.delete_cf(
+                self.ledger_blobs_height_cf,
+                block_height.to_be_bytes(),
+            )?;
+        }
+
+        Ok(())
+    }
+
+    fn blobs_by_height(
+        &self,
+        block_height: u64,
+    ) -> Result<Option<Vec<[u8; 32]>>> {
+        let blob_hashes_bytes = self
+            .inner
+            .get_cf(self.ledger_blobs_height_cf, block_height.to_be_bytes())?;
+
+        if let Some(blob_hashes_bytes) = blob_hashes_bytes {
+            let mut blob_hashes = vec![];
+            for chunk in blob_hashes_bytes.chunks(32) {
+                let mut hash = [0u8; 32];
+                hash.copy_from_slice(chunk);
+                blob_hashes.push(hash);
+            }
+            Ok(Some(blob_hashes))
+        } else {
+            Ok(None)
+        }
+    }
 
     fn block(&self, hash: &[u8]) -> Result<Option<Block>> {
         match self.inner.get_cf(self.ledger_cf, hash)? {
@@ -495,7 +573,24 @@ impl<DB: DBAccess> Ledger for DBTransaction<'_, DB> {
                 let mut txs = vec![];
                 for buf in txs_buffers {
                     let buf = buf?.unwrap();
-                    let tx = SpentTransaction::read(&mut &buf[..])?;
+                    let mut tx = SpentTransaction::read(&mut &buf[..])?;
+                    if let Some(blobs) = tx.inner.inner.blob_mut() {
+                        for blob in blobs {
+                            // Retrieve blob data from the ledger
+                            let sidecar = self
+                                .blob_data_by_hash(&blob.hash)?
+                                .map(|bytes| {
+                                    BlobSidecar::from_buf(&mut &bytes[..])
+                                })
+                                .transpose()
+                                .map_err(|e| {
+                                    anyhow::anyhow!(
+                                        "Failed to parse blob sidecar: {e:?}"
+                                    )
+                                })?;
+                            blob.data = sidecar;
+                        }
+                    }
                     txs.push(tx.inner);
                 }
 

rusk-wallet/src/lib.rs

@@ -28,6 +28,7 @@ pub mod currency;
 pub mod dat;
 pub mod gas;
 
+pub use dusk_core::stake::EPOCH;
 pub use error::Error;
 pub use gql::{BlockData, BlockTransaction, GraphQL};
 pub use rues::HttpClient as RuesHttpClient;
@@ -43,8 +44,6 @@ pub const MAX_FUNCTION_NAME_SIZE: usize = 64;
 pub const MAX_CONVERTIBLE: Dusk = Dusk::MAX;
 /// The smallest amount of Dusk that is possible to convert
 pub const MIN_CONVERTIBLE: Dusk = Dusk::new(1);
-/// The length of an epoch in blocks
-pub const EPOCH: u64 = 2160;
 /// Max addresses the wallet can store
 pub const MAX_PROFILES: usize = get_max_profiles();
 /// Size in bytes of the IV used to encrypt wallet data

rusk/CHANGELOG.md

@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Add `mempool_nonce` field to `/on/account/status` response
 - Add `status` to GQL block fields
 - Add activaction height for host queries
+- Add blob config section
 
 ### Changed
 

rusk/default.config.toml

@@ -23,6 +23,13 @@
 #consensus_keys_path = '/home/user/.dusk/rusk/consensus.keys'
 min_gas_limit = 150000
 
+[blob]
+# If no setting is provided, it falls back to default value according to feature flag:
+# - `archive`: 0 (never expire)
+# - !`archive`: 86400 block (40 epochs -> 10 days)
+
+# expire_after = <#blocks>
+
 # Note: changing the vm settings is equivalent to forking the chain.
 [vm]
 generation_timeout = '3s'

rusk/src/bin/config.rs

@@ -4,6 +4,8 @@
 //
 // Copyright (c) DUSK NETWORK. All rights reserved.
 
+#[cfg(feature = "chain")]
+pub mod blob;
 #[cfg(feature = "chain")]
 pub mod chain;
 #[cfg(feature = "chain")]
@@ -22,8 +24,8 @@ use std::str::FromStr;
 
 #[cfg(feature = "chain")]
 use self::{
-    chain::ChainConfig, databroker::DataBrokerConfig, kadcast::KadcastConfig,
-    mempool::MempoolConfig, telemetry::TelemetryConfig,
+    blob::BlobConfig, chain::ChainConfig, databroker::DataBrokerConfig,
+    kadcast::KadcastConfig, mempool::MempoolConfig, telemetry::TelemetryConfig,
 };
 
 #[cfg(feature = "chain")]
@@ -67,6 +69,10 @@ pub(crate) struct Config {
     #[cfg(feature = "chain")]
     #[serde(default = "MempoolConfig::default")]
     pub(crate) mempool: MempoolConfig,
+
+    #[cfg(feature = "chain")]
+    #[serde(default = "BlobConfig::default")]
+    pub(crate) blob: BlobConfig,
 }
 
 /// Default log_level.

rusk/src/bin/config/blob.rs

@@ -0,0 +1,12 @@
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at http://mozilla.org/MPL/2.0/.
+//
+// Copyright (c) DUSK NETWORK. All rights reserved.
+
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Clone, Default)]
+pub(crate) struct BlobConfig {
+    pub expire_after: Option<u64>,
+}