Enable pruning storage by size

bee-ledger/Cargo.toml

@@ -23,6 +23,7 @@ digest = { version = "0.9.0", default-features = false, optional = true }
 futures = { version = "0.3.17", default-features = false, optional = true }
 hashbrown = { version = "0.11.2", default-features = false, optional = true }
 hex = { version = "0.4.3", default-features = false, optional = true }
+humanize-rs = { version = "0.1.5", default-features = false, optional = true }
 iota-crypto = { version = "0.10.0", default-features = false, features = [ "blake2b" ], optional = true }
 log = { version = "0.4.14", default-features = false, optional = true }
 ref-cast = { version = "1.0.6", default-features = false, optional = true }
@@ -36,6 +37,10 @@ trace-tools = { version = "0.3.0", default-features = false, optional = true }
 tracing = { version = "0.1.29", default-features = false, optional = true }
 url = { version = "2.2.2", default-features = false, optional = true }
 
+[dev-dependencies]
+serde_json = { version = "1.0.68", default-features = false, features = [ "std" ] }
+toml = { version = "0.5.8", default-features = false }
+
 [features]
 workers = [
   "bee-runtime",
@@ -47,6 +52,7 @@ workers = [
   "futures",
   "hashbrown",
   "hex",
+  "humanize-rs",
   "iota-crypto",
   "log",
   "ref-cast",

bee-ledger/src/workers/consensus/worker.rs

@@ -24,7 +24,11 @@ use crate::{
         consensus::{metadata::WhiteFlagMetadata, state::validate_ledger_state, white_flag},
         error::Error,
         event::{MessageReferenced, MilestoneConfirmed, OutputConsumed, OutputCreated},
-        pruning::{condition::should_prune, config::PruningConfig, prune},
+        pruning::{
+            condition::{should_prune, PruningTask},
+            config::PruningConfig,
+            prune,
+        },
         snapshot::{condition::should_snapshot, config::SnapshotConfig, worker::SnapshotWorker},
         storage::{self, StorageBackend},
     },
@@ -279,28 +283,28 @@ where
         let bmd = tangle.config().below_max_depth();
 
         let snapshot_depth_min = bmd + EXTRA_SNAPSHOT_DEPTH;
-        let snapshot_depth = if snapshot_config.depth() < snapshot_depth_min {
+        let snapshot_depth = if snapshot_config.snapshotting().depth() < snapshot_depth_min {
             warn!(
                 "Configuration value for \"snapshot.depth\" is too low ({}), value changed to {}.",
-                snapshot_config.depth(),
+                snapshot_config.snapshotting().depth(),
                 snapshot_depth_min
             );
             snapshot_depth_min
         } else {
-            snapshot_config.depth()
+            snapshot_config.snapshotting().depth()
         };
 
         let snapshot_pruning_delta = bmd + EXTRA_PRUNING_DEPTH;
-        let pruning_delay_min = snapshot_depth + snapshot_pruning_delta;
-        let pruning_delay = if pruning_config.delay() < pruning_delay_min {
+        let milestones_to_keep_min = snapshot_depth + snapshot_pruning_delta;
+        let milestones_to_keep = if pruning_config.milestones().max_milestones_to_keep() < milestones_to_keep_min {
             warn!(
-                "Configuration value for \"pruning.delay\" is too low ({}), value changed to {}.",
-                pruning_config.delay(),
-                pruning_delay_min
+                "Configuration value for \"max_milestones_to_keep\" is too low ({}), value changed to {}.",
+                pruning_config.milestones().max_milestones_to_keep(),
+                milestones_to_keep_min
             );
-            pruning_delay_min
+            milestones_to_keep_min
         } else {
-            pruning_config.delay()
+            pruning_config.milestones().max_milestones_to_keep()
         };
 
         // Unwrap is fine because ledger index was already in storage or just added by the snapshot worker.
@@ -341,24 +345,46 @@ where
                                 // }
                             }
                             Err(reason) => {
-                                debug!("Snapshotting skipped: {:?}", reason);
+                                debug!("Snapshotting skipped: {reason}");
                             }
                         }
 
-                        match should_prune(&tangle, ledger_index, pruning_delay, &pruning_config) {
-                            Ok((start_index, target_index)) => {
-                                if let Err(e) =
-                                    prune::prune(&tangle, &storage, &bus, start_index, target_index, &pruning_config)
-                                        .await
-                                {
-                                    error!("Pruning failed: {:?}.", e);
-                                    // TODO: consider initiating an emergency (but graceful) shutdown instead of just
-                                    // panicking.
-                                    panic!("Aborting due to an unexpected pruning error.");
+                        match should_prune(&tangle, &storage, ledger_index, milestones_to_keep, &pruning_config) {
+                            Ok(pruning_task) => match pruning_task {
+                                PruningTask::ByIndexRange {
+                                    start_index,
+                                    target_index,
+                                } => {
+                                    if let Err(e) = prune::prune_by_range(
+                                        &tangle,
+                                        &storage,
+                                        &bus,
+                                        start_index,
+                                        target_index,
+                                        &pruning_config,
+                                    )
+                                    .await
+                                    {
+                                        error!("Pruning milestone range failed: {e}.");
+                                    }
                                 }
-                            }
+                                PruningTask::ByDbSize { num_bytes_to_prune } => {
+                                    if let Err(e) = prune::prune_by_size(
+                                        &tangle,
+                                        &storage,
+                                        &bus,
+                                        ledger_index,
+                                        num_bytes_to_prune,
+                                        &pruning_config,
+                                    )
+                                    .await
+                                    {
+                                        error!("Pruning by storage size failed: {e}.");
+                                    }
+                                }
+                            },
                             Err(reason) => {
-                                debug!("Pruning skipped: {:?}", reason);
+                                debug!("Pruning skipped: {reason}");
                             }
                         }
                     }

bee-ledger/src/workers/error.rs

@@ -7,76 +7,57 @@ use bee_message::{address::Address, milestone::MilestoneIndex, Error as MessageE
 
 use crate::{
     types::{Balance, Error as TypesError, Unspent},
-    workers::snapshot::error::Error as SnapshotError,
+    workers::{pruning::error::PruningError, snapshot::error::Error as SnapshotError},
 };
 
 /// Errors occurring during ledger workers operations.
 #[derive(Debug, thiserror::Error)]
+#[allow(missing_docs)]
 pub enum Error {
-    /// Snapshot error.
     #[error("Snapshot error: {0}")]
     Snapshot(#[from] SnapshotError),
-    /// Types error.
+    #[error("Pruning error: {0}")]
+    Pruning(#[from] PruningError),
     #[error("Types error: {0}")]
     Types(#[from] TypesError),
-    /// Message error.
     #[error("Message error: {0}")]
     Message(#[from] MessageError),
-    /// Missing message in the past cone of the milestone
     #[error("Message {0} is missing in the past cone of the milestone")]
     MissingMessage(MessageId),
-    /// Unsupported input kind.
     #[error("Unsupported input kind: {0}")]
     UnsupportedInputKind(u8),
-    /// Unsupported output kind.
     #[error("Unsupported output kind: {0}")]
     UnsupportedOutputKind(u8),
-    /// Unsupported payload kind.
     #[error("Unsupported payload kind: {0}")]
     UnsupportedPayloadKind(u32),
-    /// Milestone message not found.
     #[error("Milestone message not found: {0}")]
     MilestoneMessageNotFound(MessageId),
-    /// Message payload is not a milestone
     #[error("Message payload is not a milestone")]
     NoMilestonePayload,
-    /// Non contiguous milestones.
     #[error("Non contiguous milestones: tried to confirm {0} on top of {1}")]
     NonContiguousMilestones(u32, u32),
-    /// Merkle proof mismatch.
     #[error("Merkle proof mismatch on milestone {0}: computed {1} != provided {2}")]
     MerkleProofMismatch(MilestoneIndex, String, String),
-    /// Invalid messages count.
     #[error("Invalid messages count: referenced ({0}) != no transaction ({1}) + conflicting ({2}) + included ({3})")]
     InvalidMessagesCount(usize, usize, usize, usize),
-    /// Invalid ledger unspent state.
     #[error("Invalid ledger unspent state: {0}")]
     InvalidLedgerUnspentState(u64),
-    /// Invalid ledger balance state.
     #[error("Invalid ledger balance state: {0}")]
     InvalidLedgerBalanceState(u64),
-    /// Invalid ledger dust state.
     #[error("Invalid ledger dust state: {0:?} {1:?}")]
     InvalidLedgerDustState(Address, Balance),
-    /// Consumed amount overflow.
     #[error("Consumed amount overflow: {0}.")]
     ConsumedAmountOverflow(u128),
-    /// Created amount overflow.
     #[error("Created amount overflow: {0}.")]
     CreatedAmountOverflow(u128),
-    /// Ledger state overflow.
     #[error("Ledger state overflow: {0}")]
     LedgerStateOverflow(u128),
-    /// Non zero balance diff sum.
     #[error("Non zero balance diff sum: {0}.")]
     NonZeroBalanceDiffSum(i64),
-    /// Decreasing receipt migrated at index.
     #[error("Decreasing receipt migrated at index: {0} < {1}")]
     DecreasingReceiptMigratedAtIndex(MilestoneIndex, MilestoneIndex),
-    /// Missing unspent output.
     #[error("Missing unspent output {0}")]
     MissingUnspentOutput(Unspent),
-    /// Storage backend error.
     #[error("Storage backend error: {0}")]
     Storage(Box<dyn std::error::Error + Send>),
 }