server: snapshot

crates/brk_server/examples/server.rs

@@ -1,15 +1,10 @@
-use std::{
-    path::Path,
-    thread::{self, sleep},
-    time::Duration,
-};
+use std::{path::Path, thread};
 
 use brk_computer::Computer;
-
 use brk_error::Result;
 use brk_fetcher::Fetcher;
 use brk_indexer::Indexer;
-use brk_iterator::Blocks;
+use brk_mempool::Mempool;
 use brk_query::AsyncQuery;
 use brk_reader::Reader;
 use brk_rpc::{Auth, Client};
@@ -30,10 +25,7 @@ fn run() -> Result<()> {
     brk_logger::init(Some(Path::new(".log")))?;
 
     let bitcoin_dir = Client::default_bitcoin_path();
-    // let bitcoin_dir = Path::new("/Volumes/WD_BLACK1/bitcoin");
-
     let outputs_dir = Path::new(&std::env::var("HOME").unwrap()).join(".brk");
-    // let outputs_dir = Path::new("../../_outputs");
 
     let client = Client::new(
         Client::default_url(),
@@ -41,51 +33,45 @@ fn run() -> Result<()> {
     )?;
 
     let reader = Reader::new(bitcoin_dir.join("blocks"), &client);
-
-    let blocks = Blocks::new(&client, &reader);
-
-    let mut indexer = Indexer::forced_import(&outputs_dir)?;
-
+    let indexer = Indexer::forced_import(&outputs_dir)?;
     let fetcher = Some(Fetcher::import(true, None)?);
+    let computer = Computer::forced_import(&outputs_dir, &indexer, fetcher)?;
 
-    let mut computer = Computer::forced_import(&outputs_dir, &indexer, fetcher)?;
+    let mempool = Mempool::new(&client);
+    let mempool_clone = mempool.clone();
+    thread::spawn(move || {
+        mempool_clone.start();
+    });
 
     let exit = Exit::new();
     exit.set_ctrlc_handler();
 
-    let query = AsyncQuery::build(&reader, &indexer, &computer, None);
-
-    let future = async move {
-        let server = Server::new(&query, None);
-
-        tokio::spawn(async move {
-            server.serve(true).await.unwrap();
-        });
-
-        Ok(()) as Result<()>
-    };
+    let query = AsyncQuery::build(&reader, &indexer, &computer, Some(mempool));
 
     let runtime = tokio::runtime::Builder::new_multi_thread()
         .enable_all()
         .build()?;
 
-    let _handle = runtime.spawn(future);
+    // Option 1: block_on to run and properly propagate errors
+    runtime.block_on(async move {
+        let server = Server::new(&query, None);
 
-    loop {
-        client.wait_for_synced_node()?;
+        let handle = tokio::spawn(async move { server.serve(true).await });
 
-        let last_height = client.get_last_height()?;
-
-        info!("{} blocks found.", u32::from(last_height) + 1);
-
-        let starting_indexes = indexer.checked_index(&blocks, &client, &exit)?;
-
-        computer.compute(&indexer, starting_indexes, &reader, &exit)?;
-
-        info!("Waiting for new blocks...");
-
-        while last_height == client.get_last_height()? {
-            sleep(Duration::from_secs(1))
+        // Await the handle to catch both panics and errors
+        match handle.await {
+            Ok(Ok(())) => info!("Server shut down cleanly"),
+            Ok(Err(e)) => log::error!("Server error: {e:?}"),
+            Err(e) => {
+                // JoinError - either panic or cancellation
+                if e.is_panic() {
+                    log::error!("Server panicked: {:?}", e.into_panic());
+                } else {
+                    log::error!("Server task cancelled");
+                }
+            }
         }
-    }
+
+        Ok(()) as Result<()>
+    })
 }