Merge branch 'main' into feat/wasm-types-coverage

.github/workflows/release-plz.yml

@@ -61,7 +61,7 @@ jobs:
             exit
           fi
           # Update the types definition for lumina-node
-          npm i
+          npm clean-install
           npm run tsc
           # Update the readme for lumina-node
           wasm-pack build ..

Cargo.toml

@@ -3,7 +3,7 @@ resolver = "2"
 members = ["cli", "grpc", "node", "node-wasm", "proto", "rpc", "types"]
 
 [workspace.dependencies]
-blockstore = "0.7.0"
+blockstore = "0.7.1"
 lumina-node = { version = "0.8.0", path = "node" }
 lumina-node-wasm = { version = "0.7.0", path = "node-wasm" }
 celestia-proto = { version = "0.6.0", path = "proto" }

README.md

@@ -5,7 +5,7 @@ Rust implementation of Celestia's [data availability node](https://github.com/ce
 Run Lumina now at [lumina.rs](https://lumina.rs/) and directly verify Celestia.
 
 Supported features:
-- Backward and forward synchronization of block headers within syncing window
+- Backward and forward synchronization of block headers within sampling window
 - Header exchange (`header-ex`) client and server
 - Listening for, verifying and redistributing extended headers on gossip protocol (`header-sub`)
 - Listening for, verifying and redistributing fraud proofs on gossip protocol (`fraud-sub`)

cli/Cargo.toml

@@ -24,6 +24,7 @@ name = "lumina"
 path = "src/main.rs"
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
+blockstore.workspace = true
 celestia-rpc = { workspace = true, features = ["p2p"] }
 celestia-types.workspace = true
 libp2p.workspace = true

cli/src/native.rs

@@ -4,22 +4,26 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use anyhow::{bail, Context, Result};
+use blockstore::EitherBlockstore;
 use celestia_rpc::prelude::*;
 use celestia_rpc::Client;
 use clap::{value_parser, Parser};
 use directories::ProjectDirs;
 use libp2p::multiaddr::{Multiaddr, Protocol};
-use lumina_node::blockstore::RedbBlockstore;
+use lumina_node::blockstore::{InMemoryBlockstore, RedbBlockstore};
 use lumina_node::events::NodeEvent;
 use lumina_node::network::Network;
-use lumina_node::node::Node;
-use lumina_node::store::{RedbStore, Store};
+use lumina_node::node::{Node, MIN_PRUNING_DELAY, MIN_SAMPLING_WINDOW};
+use lumina_node::store::{EitherStore, InMemoryStore, RedbStore, Store as _};
 use tokio::task::spawn_blocking;
 use tracing::info;
 use tracing::warn;
 
 const CELESTIA_LOCAL_BRIDGE_RPC_ADDR: &str = "ws://localhost:36658";
 
+type Blockstore = EitherBlockstore<InMemoryBlockstore, RedbBlockstore>;
+type Store = EitherStore<InMemoryStore, RedbStore>;
+
 #[derive(Debug, Parser)]
 pub(crate) struct Params {
     /// Network to connect.
@@ -36,34 +40,51 @@ pub(crate) struct Params {
     pub(crate) bootnodes: Vec<Multiaddr>,
 
     /// Persistent header store path.
-    #[arg(short, long = "store")]
+    #[arg(short, long)]
     pub(crate) store: Option<PathBuf>,
 
-    /// Syncing window size, defines maximum age of headers considered for syncing and sampling.
-    /// Headers older than syncing window by more than an hour are eligible for pruning.
-    #[arg(long = "syncing-window", verbatim_doc_comment)]
+    /// Use in-memory store.
+    #[arg(long)]
+    pub(crate) in_memory_store: bool,
+
+    /// Sampling window defines maximum age of a block considered for syncing and sampling.
+    #[arg(long)]
+    #[clap(value_parser = parse_duration::parse)]
+    pub(crate) sampling_window: Option<Duration>,
+
+    /// Pruning delay defines how much time the pruner should wait after sampling window in
+    /// order to prune the block.
+    #[arg(long)]
     #[clap(value_parser = parse_duration::parse)]
-    pub(crate) syncing_window: Option<Duration>,
+    pub(crate) pruning_delay: Option<Duration>,
 }
 
 pub(crate) async fn run(args: Params) -> Result<()> {
-    info!("Initializing store");
-    let db = open_db(args.store, args.network.id()).await?;
-    let store = RedbStore::new(db.clone()).await?;
-    let blockstore = RedbBlockstore::new(db);
-
-    let stored_ranges = store.get_stored_header_ranges().await?;
-    if stored_ranges.is_empty() {
-        info!("Initialised new store");
+    let (blockstore, store) = if args.in_memory_store {
+        open_in_memory_stores()
     } else {
-        info!("Initialised store, present headers: {stored_ranges}");
-    }
+        open_db_stores(args.store, args.network.id()).await?
+    };
 
     let mut node_builder = Node::builder()
         .store(store)
         .blockstore(blockstore)
         .network(args.network.clone());
 
+    if let Some(sampling_window) = args.sampling_window {
+        node_builder = node_builder.sampling_window(sampling_window);
+    } else if args.in_memory_store {
+        // In-memory stores are memory hungry, so we lower sampling window.
+        node_builder = node_builder.sampling_window(MIN_SAMPLING_WINDOW);
+    }
+
+    if let Some(pruning_delay) = args.pruning_delay {
+        node_builder = node_builder.pruning_delay(pruning_delay);
+    } else if args.in_memory_store {
+        // In-memory stores are memory hungry, so we lower pruning window.
+        node_builder = node_builder.pruning_delay(MIN_PRUNING_DELAY);
+    }
+
     if args.bootnodes.is_empty() {
         if args.network.is_custom() {
             let bootnodes = fetch_bridge_multiaddrs(CELESTIA_LOCAL_BRIDGE_RPC_ADDR).await?;
@@ -77,10 +98,6 @@ pub(crate) async fn run(args: Params) -> Result<()> {
         node_builder = node_builder.listen(args.listen_addrs);
     }
 
-    if let Some(syncing_window) = args.syncing_window {
-        node_builder = node_builder.syncing_window(syncing_window);
-    }
-
     let (_node, mut events) = node_builder
         .start_subscribed()
         .await
@@ -98,6 +115,32 @@ pub(crate) async fn run(args: Params) -> Result<()> {
     Ok(())
 }
 
+fn open_in_memory_stores() -> (Blockstore, Store) {
+    info!("Initializing in-memory store");
+    let store = InMemoryStore::new();
+    let blockstore = InMemoryBlockstore::new();
+    (EitherBlockstore::Left(blockstore), EitherStore::Left(store))
+}
+
+async fn open_db_stores(path: Option<PathBuf>, network_id: &str) -> Result<(Blockstore, Store)> {
+    info!("Initializing store");
+    let db = open_db(path, network_id).await?;
+    let store = RedbStore::new(db.clone()).await?;
+    let blockstore = RedbBlockstore::new(db);
+
+    let stored_ranges = store.get_stored_header_ranges().await?;
+    if stored_ranges.is_empty() {
+        info!("Initialised new store",);
+    } else {
+        info!("Initialised store, present headers: {stored_ranges}");
+    }
+
+    Ok((
+        EitherBlockstore::Right(blockstore),
+        EitherStore::Right(store),
+    ))
+}
+
 async fn open_db(path: Option<PathBuf>, network_id: &str) -> Result<Arc<redb::Database>> {
     let network_id = network_id.to_owned();
 

grpc/tests/tonic.rs

@@ -101,7 +101,7 @@ async fn submit_blob() {
         .await
         .unwrap();
 
-    tokio::time::sleep(std::time::Duration::from_secs(3)).await;
+    tokio::time::sleep(std::time::Duration::from_secs(8)).await;
 
     let _submitted_tx = client
         .get_tx(response.txhash)

node-wasm/js/.gitignore

@@ -1,4 +1,3 @@
 /node_modules
-/package-lock.json
 /*.tgz
 /docs