feat: Header pruning

Cargo.toml

@@ -16,6 +16,7 @@ celestia-tendermint-proto = "0.32.1"
 
 [patch.crates-io]
 # Uncomment to apply local changes
+#beetswap = { path = "../beetswap" }
 #blockstore = { path = "../blockstore" }
 #celestia-tendermint = { path = "../celestia-tendermint-rs/tendermint" }
 #celestia-tendermint-proto = { path = "../celestia-tendermint-rs/proto" }

node/src/events.rs

@@ -281,6 +281,18 @@ pub enum NodeEvent {
  error: String,
  },
 
+ /// Pruned headers up to and including specified height.
+ PrunedHeaders {
+ /// Last header height that was pruned
+ to_height: u64,
+ },
+
+ /// Pruning fatal error.
+ FatalPrunerError {
+ /// A human readable error.
+ error: String,
+ },
+
  /// Network was compromised.
  ///
  /// This happens when a valid bad encoding fraud proof is received.
@@ -296,6 +308,7 @@ impl NodeEvent {
  match self {
  NodeEvent::FatalDaserError { .. }
  | NodeEvent::FatalSyncerError { .. }
+ | NodeEvent::FatalPrunerError { .. }
  | NodeEvent::FetchingHeadersFailed { .. }
  | NodeEvent::NetworkCompromised => true,
  NodeEvent::ConnectingToBootnodes
@@ -308,7 +321,8 @@ impl NodeEvent {
  | NodeEvent::FetchingHeadHeaderStarted
  | NodeEvent::FetchingHeadHeaderFinished { .. }
  | NodeEvent::FetchingHeadersStarted { .. }
- | NodeEvent::FetchingHeadersFinished { .. } => false,
+ | NodeEvent::FetchingHeadersFinished { .. }
+ | NodeEvent::PrunedHeaders { .. } => false,
  }
  }
 }
@@ -413,6 +427,12 @@ impl fmt::Display for NodeEvent {
  NodeEvent::FatalSyncerError { error } => {
  write!(f, "Syncer stopped because of a fatal error: {error}")
  }
+ Self::PrunedHeaders { to_height } => {
+ write!(f, "Pruned headers up to and including {to_height}")
+ }
+ NodeEvent::FatalPrunerError { error } => {
+ write!(f, "Pruner stopped because of a fatal error: {error}")
+ }
  NodeEvent::NetworkCompromised => {
  write!(f, "The network is compromised and should not be trusted. ")?;
  write!(f, "Node stopped synchronizing and sampling, but you can still make some queries to the network.")

node/src/lib.rs

@@ -10,6 +10,7 @@ pub mod network;
 pub mod node;
 mod p2p;
 mod peer_tracker;
+mod pruner;
 pub mod store;
 mod syncer;
 #[cfg(any(test, feature = "test-utils"))]

node/src/node.rs

@@ -25,6 +25,7 @@ use crate::daser::{Daser, DaserArgs};
 use crate::events::{EventChannel, EventSubscriber, NodeEvent};
 use crate::executor::spawn;
 use crate::p2p::{P2p, P2pArgs};
+use crate::pruner::{Pruner, PrunerArgs};
 use crate::store::{SamplingMetadata, Store, StoreError};
 use crate::syncer::{Syncer, SyncerArgs};
 
@@ -90,6 +91,7 @@ where
  store: Arc<S>,
  syncer: Arc<Syncer<S>>,
  _daser: Arc<Daser>,
+ _pruner: Arc<Pruner>,
  tasks_cancellation_token: CancellationToken,
 }
 
@@ -125,7 +127,7 @@ where
  local_keypair: config.p2p_local_keypair,
  bootnodes: config.p2p_bootnodes,
  listen_on: config.p2p_listen_on,
- blockstore,
+ blockstore: blockstore.clone(),
  store: store.clone(),
  event_pub: event_channel.publisher(),
  })
@@ -145,13 +147,20 @@ where
  event_pub: event_channel.publisher(),
  })?);
 
+ let pruner = Arc::new(Pruner::start(PrunerArgs {
+ store: store.clone(),
+ blockstore,
+ event_pub: event_channel.publisher(),
+ }));
+
  // spawn the task that will stop the services when the fraud is detected
  let network_compromised_token = p2p.get_network_compromised_token().await?;
  let tasks_cancellation_token = CancellationToken::new();
 
  spawn({
  let syncer = syncer.clone();
  let daser = daser.clone();
+ let pruner = pruner.clone();
  let tasks_cancellation_token = tasks_cancellation_token.child_token();
  let event_pub = event_channel.publisher();
 
@@ -161,6 +170,7 @@ where
  _ = network_compromised_token.cancelled() => {
  syncer.stop();
  daser.stop();
+ pruner.stop();
 
  if event_pub.has_subscribers() {
  event_pub.send(NodeEvent::NetworkCompromised);
@@ -180,6 +190,7 @@ where
  store,
  syncer,
  _daser: daser,
+ _pruner: pruner,
  tasks_cancellation_token,
  };