feat: Add flag to quit when reaching tip (#369)

src/bin/dolos/daemon.rs

@@ -22,6 +22,7 @@ pub async fn run(config: super::Config, _args: &Args) -> miette::Result<()> {
         shelley,
         mempool.clone(),
         &config.retries,
+        false,
     )
     .into_diagnostic()
     .context("bootstrapping sync pipeline")?;

src/bin/dolos/sync.rs

@@ -1,9 +1,13 @@
 use miette::{Context, IntoDiagnostic};
 
 #[derive(Debug, clap::Args)]
-pub struct Args {}
+pub struct Args {
+    /// Skip the bootstrap if there's already data in the stores
+    #[arg(long, action)]
+    quit_on_tip: bool,
+}
 
-pub fn run(config: &super::Config, _args: &Args) -> miette::Result<()> {
+pub fn run(config: &super::Config, args: &Args) -> miette::Result<()> {
     crate::common::setup_tracing(&config.logging)?;
 
     let (wal, ledger) = crate::common::open_data_stores(config)?;
@@ -21,6 +25,7 @@ pub fn run(config: &super::Config, _args: &Args) -> miette::Result<()> {
         shelley,
         mempool,
         &config.retries,
+        args.quit_on_tip,
     )
     .into_diagnostic()
     .context("bootstrapping sync pipeline")?;

src/sync/mod.rs

@@ -53,12 +53,14 @@ pub fn pipeline(
     shelley: shelley::GenesisFile,
     mempool: Mempool,
     retries: &Option<gasket::retries::Policy>,
+    quit_on_tip: bool,
 ) -> Result<Vec<gasket::runtime::Tether>, Error> {
     let mut pull = pull::Stage::new(
         upstream.peer_address.clone(),
         upstream.network_magic,
         config.pull_batch_size.unwrap_or(50),
         wal.clone(),
+        quit_on_tip,
     );
 
     let mut roll = roll::Stage::new(wal.clone());

src/sync/pull.rs

@@ -36,6 +36,7 @@ pub enum WorkUnit {
 
 pub struct Worker {
     peer_session: PeerClient,
+    quit_on_tip: bool,
 }
 
 impl Worker {
@@ -110,7 +111,10 @@ impl gasket::framework::Worker<Stage> for Worker {
 
         info!(?intersection, "found intersection");
 
-        let worker = Self { peer_session };
+        let worker = Self {
+            peer_session,
+            quit_on_tip: stage.quit_on_tip,
+        };
 
         Ok(worker)
     }
@@ -125,8 +129,13 @@ impl gasket::framework::Worker<Stage> for Worker {
             debug!("should request next batch of blocks");
             Ok(WorkSchedule::Unit(WorkUnit::Pull))
         } else {
-            debug!("should await next block");
-            Ok(WorkSchedule::Unit(WorkUnit::Await))
+            if self.quit_on_tip {
+                debug!("reached tip, exiting");
+                Ok(WorkSchedule::Done)
+            } else {
+                debug!("should await next block");
+                Ok(WorkSchedule::Unit(WorkUnit::Await))
+            }
         }
     }
 
@@ -204,6 +213,7 @@ pub struct Stage {
     network_magic: u64,
     block_fetch_batch_size: usize,
     wal: WalStore,
+    quit_on_tip: bool,
 
     pub downstream: DownstreamPort,
 
@@ -220,11 +230,13 @@ impl Stage {
         network_magic: u64,
         block_fetch_batch_size: usize,
         wal: WalStore,
+        quit_on_tip: bool,
     ) -> Self {
         Self {
             peer_address,
             network_magic,
             wal,
+            quit_on_tip,
             block_fetch_batch_size,
             downstream: Default::default(),
             block_count: Default::default(),