refactor: separate db from home and consolidate home_db and syncing

rust/agents/processor/src/processor.rs

@@ -24,6 +24,7 @@ use optics_core::{
     db::HomeDB,
     traits::{CommittedMessage, Common, Home, MessageStatus},
 };
+use std::convert::TryInto;
 
 use crate::{
     prover_sync::ProverSync,
@@ -162,9 +163,9 @@ impl Replica {
     async fn try_msg_by_domain_and_nonce(&self, domain: u32, nonce: u32) -> Result<Flow> {
         use optics_core::traits::Replica;
 
-        let message = match self.home.message_by_nonce(domain, nonce).await {
-            Ok(Some(m)) => m,
-            Ok(None) => {
+        let message: CommittedMessage = match self.home_db.message_by_nonce(domain, nonce)? {
+            Some(m) => m.try_into()?,
+            None => {
                 info!(
                     domain = domain,
                     sequence = nonce,
@@ -174,7 +175,6 @@ impl Replica {
                 );
                 return Ok(Flow::Repeat);
             }
-            Err(e) => bail!(e),
         };
 
         info!(target: "seen_committed_messages", leaf_index = message.leaf_index);
@@ -403,21 +403,7 @@ impl OpticsAgent for Processor {
 
             info!("Starting indexer");
             // indexer setup
-            let block_height = self
-                .as_ref()
-                .metrics
-                .new_int_gauge(
-                    "block_height",
-                    "Height of a recently observed block",
-                    &["network", "agent"],
-                )
-                .expect("failed to register block_height metric")
-                .with_label_values(&[self.home().name(), Self::AGENT_NAME]);
-            let indexer = &self.as_ref().indexer;
-            let index_task = self
-                .home()
-                .index(indexer.from(), indexer.chunk_size(), block_height);
-
+            let index_task = self.syncing_home_db().index();
             info!("started indexer and sync");
 
             // instantiate task array here so we can optionally push run_task

rust/agents/relayer/src/relayer.rs

@@ -9,14 +9,15 @@ use optics_base::{
     home::Homes,
     replica::Replicas,
 };
-use optics_core::traits::Common;
+use optics_core::{db::HomeDB, traits::Common};
 
 use crate::settings::RelayerSettings as Settings;
 
 #[derive(Debug)]
 struct UpdatePoller {
     duration: Duration,
     home: Arc<Homes>,
+    home_db: HomeDB,
     replica: Arc<Replicas>,
     semaphore: Mutex<()>,
 }
@@ -32,9 +33,10 @@ impl std::fmt::Display for UpdatePoller {
 }
 
 impl UpdatePoller {
-    fn new(home: Arc<Homes>, replica: Arc<Replicas>, duration: u64) -> Self {
+    fn new(home: Arc<Homes>, home_db: HomeDB, replica: Arc<Replicas>, duration: u64) -> Self {
         Self {
             home,
+            home_db,
             replica,
             duration: Duration::from_secs(duration),
             semaphore: Mutex::new(()),
@@ -53,7 +55,7 @@ impl UpdatePoller {
         );
 
         // Check for first signed update building off of the replica's current root
-        let signed_update_opt = self.home.signed_update_by_old_root(old_root).await?;
+        let signed_update_opt = self.home_db.update_by_previous_root(old_root)?;
 
         // If signed update exists, update replica's current root
         if let Some(signed_update) = signed_update_opt {
@@ -134,6 +136,7 @@ impl OpticsAgent for Relayer {
     fn run(&self, name: &str) -> Instrumented<JoinHandle<Result<()>>> {
         let replica_opt = self.replica_by_name(name);
         let home = self.home();
+        let home_db = self.home_db();
         let name = name.to_owned();
 
         let duration = self.duration;
@@ -144,7 +147,7 @@ impl OpticsAgent for Relayer {
             }
             let replica = replica_opt.unwrap();
 
-            let update_poller = UpdatePoller::new(home, replica.clone(), duration);
+            let update_poller = UpdatePoller::new(home, home_db, replica.clone(), duration);
             update_poller.spawn().await?
         })
         .in_current_span()

rust/agents/updater/src/main.rs

@@ -14,10 +14,8 @@ use color_eyre::Result;
 
 use futures_util::future::select_all;
 
-use optics_base::{agent::OpticsAgent, cancel_task};
-use optics_core::traits::{Common, Home};
-
 use crate::{settings::UpdaterSettings as Settings, updater::Updater};
+use optics_base::{agent::OpticsAgent, cancel_task};
 
 #[allow(unused_must_use)]
 async fn _main() -> Result<()> {
@@ -34,25 +32,7 @@ async fn _main() -> Result<()> {
 
     let _ = agent.metrics().run_http_server();
 
-    // this is deliberately different from other agents because the updater
-    // does not run replicas. As a result, most of the contents of run_all are
-    // broken out here
-    let indexer = &agent.as_ref().indexer;
-
-    let block_height = agent
-        .as_ref()
-        .metrics
-        .new_int_gauge(
-            "block_height",
-            "Height of a recently observed block",
-            &["network", "agent"],
-        )
-        .expect("failed to register block_height metric")
-        .with_label_values(&[agent.home().name(), Updater::AGENT_NAME]);
-
-    let index_task = agent
-        .home()
-        .index(indexer.from(), indexer.chunk_size(), block_height);
+    let index_task = agent.syncing_home_db().index();
     let run_task = agent.run("");
 
     let futs = vec![index_task, run_task];

rust/agents/updater/src/updater.rs

@@ -32,7 +32,6 @@ use optics_core::{
 #[derive(Debug)]
 struct UpdateHandler {
     home: Arc<Homes>,
-
     rx: Receiver<Update>,
     update_pause: u64,
     signer: Arc<Signers>,
@@ -273,7 +272,7 @@ impl OpticsAgent for Updater {
             rx,
             self.update_pause,
             self.signer.clone(),
-            HomeDB::new(self.db(), self.home().name().to_owned()),
+            self.home_db(),
             Default::default(),
             self.signed_attestation_count.clone(),
         );