feat:enhance archiver

Signed-off-by: Chen Kai <[email protected]>

bin/archiver/Cargo.toml

@@ -18,5 +18,6 @@ serde_json.workspace = true
 serde.workspace = true
 tracing.workspace = true
 tokio.workspace = true
+eyre.workspace = true
 blob-archiver-storage = { path = "../../crates/storage" }
 

bin/archiver/src/archiver.rs

@@ -1,54 +1,282 @@
-use blob_archiver_storage::{BlobData, BlobSidecars, Header};
-use eth2::types::{BlockId, MainnetEthSpec};
-use eth2::{BeaconNodeHttpClient, Error};
-use tracing::log::trace;
+#![allow(incomplete_features)]
+
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+use blob_archiver_storage::{BackfillProcess, BackfillProcesses, BlobData, BlobSidecars, Header, LockFile, Storage};
+use eth2::types::{BlockHeaderData, BlockId, Hash256, MainnetEthSpec};
+use eth2::{BeaconNodeHttpClient};
+use eyre::Result;
+use serde::{Deserialize, Serialize};
+use tokio::sync::watch::Receiver;
+use tokio::time::{interval, sleep};
+use tracing::log::{error, info, trace};
+
+#[allow(dead_code)]
+const LIVE_FETCH_BLOB_MAXIMUM_RETRIES: i32 = 10;
+#[allow(dead_code)]
+const STARTUP_FETCH_BLOB_MAXIMUM_RETRIES: i32 = 3;
+#[allow(dead_code)]
+const REARCHIVE_MAXIMUM_RETRIES: i32 = 3;
+#[allow(dead_code)]
+const BACKFILL_ERROR_RETRY_INTERVAL: Duration = Duration::from_secs(5);
+#[allow(dead_code)]
+const LOCK_UPDATE_INTERVAL: Duration = Duration::from_secs(10);
+#[allow(dead_code)]
+const LOCK_TIMEOUT: Duration = Duration::from_secs(20);
+#[allow(dead_code)]
+const OBTAIN_LOCK_RETRY_INTERVAL_SECS: u64 = 10;
+#[allow(dead_code)]
+static OBTAIN_LOCK_RETRY_INTERVAL: AtomicU64 = AtomicU64::new(OBTAIN_LOCK_RETRY_INTERVAL_SECS);
+
+#[derive(Debug, PartialEq, Eq, Clone, Default, Serialize, Deserialize)]
+pub struct Config {
+    pub poll_interval: Duration,
+
+    pub listen_addr: String,
+
+    pub origin_block: Hash256,
+}
 
 pub struct Archiver {
     pub beacon_client: BeaconNodeHttpClient,
+
+    storage: Arc<dyn Storage>,
+    #[allow(dead_code)]
+    id: String,
+    #[allow(dead_code)]
+    pub config: Config,
+
+    shutdown_rx: Receiver<bool>,
 }
 
 impl Archiver {
-    pub fn new(beacon_client: BeaconNodeHttpClient) -> Self {
-        Self { beacon_client }
+    pub fn new(beacon_client: BeaconNodeHttpClient, storage: Arc<dyn Storage>, shutdown_rx: Receiver<bool>) -> Self {
+        Self { beacon_client, storage, id: "".to_string(), config: Default::default(), shutdown_rx }
     }
 
-    pub async fn persist_blobs_for_block(&self, block_id: BlockId) -> Result<(), Error> {
+    pub async fn persist_blobs_for_block(&self, block_id: BlockId, overwrite: bool) -> Result<(BlockHeaderData, bool)> {
         let header_resp_opt = self
             .beacon_client
             .get_beacon_headers_block_id(block_id)
-            .await?;
-        if let Some(header) = header_resp_opt {
-            let beacon_client = self.beacon_client.clone();
-            let blobs_resp_opt = beacon_client
-                .get_blobs::<MainnetEthSpec>(BlockId::Root(header.data.root), None)
-                .await?;
-            if let Some(blob_sidecars) = blobs_resp_opt {
-                let blob_sidecar_list = blob_sidecars.data;
-                let blob_data = BlobData::new(
-                    Header {
-                        beacon_block_hash: header.data.root,
-                    },
-                    BlobSidecars {
-                        data: blob_sidecar_list,
-                    },
-                );
-                trace!("Persisting blobs for block: {:?}", blob_data);
-                return Ok(());
+            .await.map_err(|e| eyre::eyre!(e))?;
+
+        match header_resp_opt {
+            None => { Err(eyre::eyre!("No header response")) }
+            Some(header) => {
+                let exists = self.storage.exists(&header.data.root).await;
+
+                if exists && !overwrite {
+                    return Ok((header.data, true));
+                }
+
+                let blobs_resp_opt = self.beacon_client
+                    .get_blobs::<MainnetEthSpec>(BlockId::Root(header.data.root), None)
+                    .await.map_err(|e| eyre::eyre!(e))?;
+                if let Some(blob_sidecars) = blobs_resp_opt {
+                    let blob_sidecar_list = blob_sidecars.data;
+                    let blob_data = &BlobData::new(
+                        Header {
+                            beacon_block_hash: header.data.root,
+                        },
+                        BlobSidecars {
+                            data: blob_sidecar_list,
+                        },
+                    );
+                    self.storage.write_blob_data(blob_data).await?;
+                    trace!("Persisting blobs for block: {:?}", blob_data);
+                    return Ok((header.data, exists));
+                }
+                Ok((header.data, exists))
+            }
+        }
+    }
+
+    #[allow(dead_code)]
+    async fn wait_obtain_storage_lock(&self) {
+        let mut lock_file_res = self.storage.read_lock_file().await;
+        let mut now = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_secs();
+        let mut shutdown = self.shutdown_rx.clone();
+        match lock_file_res {
+            Ok(mut lock_file) => {
+                trace!("Lock file: {:#?}", lock_file);
+                if lock_file == LockFile::default() {
+                    while lock_file.archiver_id != self.id && lock_file.timestamp + LOCK_TIMEOUT.as_secs() > now {
+                        tokio::select! {
+                        _ = shutdown.changed() => {
+                            info!("Received shutdown signal, exiting wait_obtain_storage_lock");
+                            return;
+                        }
+                        _ = sleep(Duration::from_secs(OBTAIN_LOCK_RETRY_INTERVAL.load(Ordering::Relaxed))) => {
+                            lock_file_res = self.storage.read_lock_file().await;
+                            match lock_file_res {
+                                Ok(new_lock_file) => {
+                                    lock_file = new_lock_file;
+                                    now = std::time::SystemTime::now()
+                                        .duration_since(std::time::UNIX_EPOCH)
+                                        .unwrap()
+                                        .as_secs();
+                                }
+                                Err(e) => {
+                                    error!("Error reading lock file: {:#?}", e);
+                                    panic!("Error reading lock file: {:#?}", e);
+                                }
+                            }
+                        }
+                    }
+                    }
+                }
+
+                let written_res = self.storage.write_lock_file(&LockFile {
+                    archiver_id: lock_file.archiver_id.clone(),
+                    timestamp: now,
+                }).await;
+
+                match written_res {
+                    Ok(_) => {
+                        info!("Obtained storage lock");
+                    }
+                    Err(e) => {
+                        error!("Error writing lock file: {:#?}", e);
+                        panic!("Error writing lock file: {:#?}", e);
+                    }
+                }
+
+                let storage = self.storage.clone();
+                let archiver_id = self.id.clone();
+                let mut shutdown_clone = shutdown.clone();
+
+                tokio::spawn(async move {
+                    let mut ticket = interval(LOCK_UPDATE_INTERVAL);
+                    loop {
+                        tokio::select! {
+                        _ = shutdown_clone.changed() => {
+                            info!("Received shutdown signal, exiting lock update loop");
+                            break;
+                        }
+                        _ = ticket.tick() => {
+                            let now = std::time::SystemTime::now()
+                                .duration_since(std::time::UNIX_EPOCH)
+                                .unwrap()
+                                .as_secs();
+                            let written_res = storage.write_lock_file(&LockFile {
+                                archiver_id: archiver_id.clone(),
+                                timestamp: now,
+                            }).await;
+
+                            if let Err(e) = written_res {
+                                error!("Error update lockfile timestamp: {:#?}", e);
+                            }
+                        }
+                    }
+                    }
+                });
+            }
+            Err(e) => {
+                error!("Error reading lock file: {:#?}", e);
+                panic!("Error reading lock file: {:#?}", e);
             }
-            return Ok(());
         }
+    }
+
+    #[allow(dead_code)]
+    async fn backfill_blobs(&self, latest: &BlockHeaderData) {
+        let backfill_processes_res = self.storage.read_backfill_processes().await;
+
+        match backfill_processes_res {
+            Ok(mut backfill_processes) => {
+                let backfill_process = BackfillProcess {
+                    start_block: latest.clone(),
+                    current_block: latest.clone(),
+                };
+                backfill_processes.insert(latest.root, backfill_process);
+                let _ = self.storage.write_backfill_processes(&backfill_processes).await;
 
-        Ok(())
+                let mut processes = backfill_processes.clone();
+                for (_, process) in backfill_processes.iter() {
+                    self.backfill_loop(&process.start_block, &process.current_block, &mut processes).await;
+                }
+            }
+            Err(e) => {
+                error!("Error reading backfill processes: {:#?}", e);
+                panic!("Error reading backfill processes: {:#?}", e);
+            }
+        }
     }
+
+    #[allow(dead_code)]
+    async fn backfill_loop(&self, start: &BlockHeaderData, current: &BlockHeaderData, backfill_processes: &mut BackfillProcesses) {
+        let mut curr = current.clone();
+        let mut already_exists = false;
+        let mut count = 0;
+        let mut res: Result<(BlockHeaderData, bool)>;
+        info!("backfill process initiated, curr_hash: {:#?}, curr_slot: {:#?}, start_hash: {:#?},start_slot: {:#?}", curr.root, curr.header.message.slot.clone(), start.root, start.header.message.slot.clone());
+
+        while !already_exists {
+            if curr.root == self.config.origin_block {
+                info!("reached origin block, hash: {:#?}", curr.root);
+                self.defer_fn(start, &curr, backfill_processes).await;
+                return;
+            }
+
+            res = self.persist_blobs_for_block(BlockId::Root(curr.header.message.parent_root), false).await;
+            if let Err(e) = res {
+                error!("failed to persist blobs for block, will retry: {:#?}, hash: {:#?}", e, curr.header.message.parent_root);
+                sleep(BACKFILL_ERROR_RETRY_INTERVAL).await;
+                continue;
+            };
+
+            let (parent, parent_exists) = res.unwrap();
+            curr = parent;
+            already_exists = parent_exists;
+
+            if !already_exists {
+                // todo: metrics
+            }
+
+            count += 1;
+            if count % 10 == 0 {
+                let backfill_process = BackfillProcess {
+                    start_block: start.to_owned(),
+                    current_block: curr.clone(),
+                };
+                backfill_processes.insert(start.root, backfill_process);
+                let _ = self.storage.write_backfill_processes(backfill_processes).await;
+            }
+        }
+        self.defer_fn(start, &curr, backfill_processes).await;
+    }
+
+    #[allow(dead_code)]
+    async fn defer_fn(&self, start: &BlockHeaderData, current: &BlockHeaderData, backfill_processes: &mut BackfillProcesses) {
+        info!("backfill process complete, end_hash: {:#?}, end_slot: {:#?}, start_hash: {:#?},start_slot: {:#?}", current.root, current.header.message.slot.clone(), start.root, start.header.message.slot.clone());
+        backfill_processes.remove(&start.root);
+        let _ = self.storage.write_backfill_processes(backfill_processes).await;
+    }
+
+    // async fn process_blocks_until_known_block() {
+    //     debug!("refreshing live data");
+    //     let mut start: &BlockHeaderData;
+    //     let mut current_block_id = "head";
+    //
+    //     loop {
+    //
+    //     }
+    //
+    // }
 }
 
 #[cfg(test)]
 mod tests {
+    use std::path::PathBuf;
     use std::str::FromStr;
     use std::time::Duration;
 
     use eth2::{SensitiveUrl, Timeouts};
-
+    use blob_archiver_storage::fs::FSStorage;
     use super::*;
 
     #[tokio::test]
@@ -57,9 +285,11 @@ mod tests {
             SensitiveUrl::from_str("https://ethereum-beacon-api.publicnode.com").unwrap(),
             Timeouts::set_all(Duration::from_secs(30)),
         );
-        let archiver = Archiver::new(beacon_client);
+        let storage = FSStorage::new(PathBuf::from("test_dir")).await.unwrap();
+        let (_, rx) = tokio::sync::watch::channel(false);
+        let archiver = Archiver::new(beacon_client, Arc::new(storage), rx);
 
         let block_id = BlockId::Head;
-        archiver.persist_blobs_for_block(block_id).await.unwrap();
+        archiver.persist_blobs_for_block(block_id, false).await.unwrap();
     }
 }

bin/archiver/src/main.rs

@@ -1,9 +1,11 @@
+use std::path::PathBuf;
 use std::str::FromStr;
+use std::sync::Arc;
 use std::time::Duration;
 
 use eth2::types::BlockId;
 use eth2::{BeaconNodeHttpClient, SensitiveUrl, Timeouts};
-
+use blob_archiver_storage::fs::FSStorage;
 use crate::archiver::Archiver;
 
 mod archiver;
@@ -14,12 +16,14 @@ async fn main() {
         SensitiveUrl::from_str("https://ethereum-beacon-api.publicnode.com").unwrap(),
         Timeouts::set_all(Duration::from_secs(30)),
     );
-    let archiver = Archiver::new(beacon_client);
+    let storage = FSStorage::new(PathBuf::from("test_dir")).await.unwrap();
+    let (_, shutdown_rx) = tokio::sync::watch::channel(false);
+    let archiver = Archiver::new(beacon_client, Arc::new(storage), shutdown_rx);
 
     let block_id = BlockId::Head;
 
     archiver
-        .persist_blobs_for_block(block_id)
+        .persist_blobs_for_block(block_id, false)
         .await
         .expect("TODO: panic message");
 }