Prover Client Checkpoint Runner

bin/prover-client/src/args.rs

@@ -125,6 +125,18 @@ pub struct Args {
     /// Defaults to `true`.
     #[argh(option, description = "enable prover client dev rpc", default = "true")]
     pub enable_dev_rpcs: bool,
+
+    /// Controls the checkpoint proof runner service.
+    ///
+    /// When enabled, this prover will automatically generate and submit proofs for checkpoints.
+    /// This should only be enabled for provers designated as checkpoint provers.
+    /// Defaults to `false`.
+    #[argh(
+        option,
+        description = "enable prover client checkpoint runner",
+        default = "false"
+    )]
+    pub enable_checkpoint_runner: bool,
 }
 
 impl Args {

bin/prover-client/src/checkpoint_runner/error.rs

@@ -0,0 +1,25 @@
+use thiserror::Error;
+
+/// Represents errors that can occur during checkpoint-related operations.
+///
+/// This error type encompasses various failure scenarios that may occur when
+/// interacting with checkpoints, including RPC communication issues,
+/// data validation problems, and serialization errors. Each variant provides
+/// detailed information about the specific error condition.
+#[derive(Error, Debug)]
+pub enum CheckpointError {
+    /// Occurs when the RPC request to fetch checkpoint data fails.
+    #[error("Failed to fetch checkpoint data: {0}")]
+    FetchError(String),
+
+    /// Occurs when no checkpoint data is returned from the sequencer.
+    #[error("No checkpoint data returned from sequencer for index {0}")]
+    CheckpointNotFound(u64),
+
+    /// Occurs when failed to submit checkpoint proof to the sequencer.
+    #[error("Failed to submit checkpoint proof for index {index}: {error}")]
+    SubmitProofError { index: u64, error: String },
+}
+
+/// A type alias for Results involving checkpoint operations.
+pub type CheckpointResult<T> = Result<T, CheckpointError>;

bin/prover-client/src/checkpoint_runner/fetch.rs

@@ -0,0 +1,21 @@
+use jsonrpsee::http_client::HttpClient;
+use strata_rpc_api::StrataApiClient;
+use tracing::error;
+
+use super::error::CheckpointResult;
+use crate::checkpoint_runner::error::CheckpointError;
+
+/// Fetches the latest checkpoint index from the sequencer client.
+pub async fn fetch_latest_checkpoint_index(cl_client: &HttpClient) -> CheckpointResult<u64> {
+    cl_client
+        .get_latest_checkpoint_index()
+        .await
+        .map_err(|e| {
+            error!("Failed to fetch current checkpoint index: {e}");
+            CheckpointError::FetchError(e.to_string())
+        })?
+        .ok_or_else(|| {
+            error!("No checkpoint index returned from sequencer");
+            CheckpointError::CheckpointNotFound(0)
+        })
+}

bin/prover-client/src/checkpoint_runner/mod.rs

@@ -0,0 +1,6 @@
+//! A module defining operations for the checkpoint proof generation
+
+pub mod error;
+pub mod fetch;
+pub mod runner;
+pub mod submit;

bin/prover-client/src/checkpoint_runner/runner.rs

@@ -0,0 +1,65 @@
+use std::sync::Arc;
+
+use strata_rocksdb::prover::db::ProofDb;
+use tokio::{
+    sync::Mutex,
+    time::{interval, Duration},
+};
+use tracing::{error, info};
+
+use crate::{
+    checkpoint_runner::fetch::fetch_latest_checkpoint_index,
+    operators::{checkpoint::CheckpointOperator, ProvingOp},
+    task_tracker::TaskTracker,
+};
+
+const CHECKPOINT_POLL_INTERVAL: u64 = 10;
+
+/// Periodically polls for the latest checkpoint index and updates the current index.
+/// Dispatches tasks when a new checkpoint is detected.
+pub async fn checkpoint_proof_runner(
+    operator: CheckpointOperator,
+    task_tracker: Arc<Mutex<TaskTracker>>,
+    db: Arc<ProofDb>,
+) {
+    info!("Checkpoint runner started");
+    let mut ticker = interval(Duration::from_secs(CHECKPOINT_POLL_INTERVAL));
+    let mut current_checkpoint_idx: Option<u64> = None;
+
+    loop {
+        ticker.tick().await;
+
+        if let Err(e) =
+            process_checkpoint(&operator, &task_tracker, &db, &mut current_checkpoint_idx).await
+        {
+            error!("Error processing checkpoint: {e:?}");
+        }
+    }
+}
+
+async fn process_checkpoint(
+    operator: &CheckpointOperator,
+    task_tracker: &Arc<Mutex<TaskTracker>>,
+    db: &Arc<ProofDb>,
+    current_checkpoint_idx: &mut Option<u64>,
+) -> anyhow::Result<()> {
+    let new_checkpoint = fetch_latest_checkpoint_index(operator.cl_client()).await?;
+
+    if !should_update_checkpoint(*current_checkpoint_idx, new_checkpoint) {
+        info!(
+           "Fetched checkpoint {new_checkpoint} is not newer than current {current_checkpoint_idx:?}"
+       );
+        return Ok(());
+    }
+
+    operator
+        .create_task(new_checkpoint, task_tracker.clone(), db)
+        .await?;
+    *current_checkpoint_idx = Some(new_checkpoint);
+
+    Ok(())
+}
+
+fn should_update_checkpoint(current: Option<u64>, new: u64) -> bool {
+    current.is_none_or(|current| new > current)
+}

bin/prover-client/src/checkpoint_runner/submit.rs

@@ -0,0 +1,34 @@
+use std::sync::Arc;
+
+use jsonrpsee::http_client::HttpClient;
+use strata_db::traits::ProofDatabase;
+use strata_rocksdb::prover::db::ProofDb;
+use strata_rpc_api::StrataSequencerApiClient;
+use strata_rpc_types::{HexBytes, ProofKey};
+use tracing::info;
+
+use super::error::{CheckpointError, CheckpointResult};
+
+/// Submits checkpoint proof to the sequencer.
+pub async fn submit_checkpoint_proof(
+    checkpoint_index: u64,
+    sequencer_client: &HttpClient,
+    proof_key: ProofKey,
+    proof_db: Arc<ProofDb>,
+) -> CheckpointResult<()> {
+    let proof = proof_db.get_proof(proof_key).unwrap().unwrap();
+    let proof_bytes = HexBytes::from(proof.proof().as_bytes());
+
+    info!(
+        "Sending checkpoint proof: {:?} ckp id: {:?} to the sequencer",
+        proof_key, checkpoint_index
+    );
+
+    sequencer_client
+        .submit_checkpoint_proof(checkpoint_index, proof_bytes)
+        .await
+        .map_err(|e| CheckpointError::SubmitProofError {
+            index: checkpoint_index,
+            error: e.to_string(),
+        })
+}

bin/prover-client/src/main.rs

@@ -4,6 +4,7 @@
 
 use anyhow::Context;
 use args::Args;
+use checkpoint_runner::runner::checkpoint_proof_runner;
 use db::open_rocksdb_database;
 use jsonrpsee::http_client::HttpClientBuilder;
 use operators::ProofOperator;
@@ -17,6 +18,7 @@
 use tracing::debug;
 
 mod args;
+mod checkpoint_runner;
 mod db;
 mod errors;
 mod hosts;
@@ -90,6 +92,22 @@
     spawn(async move { manager.process_pending_tasks().await });
     debug!("Spawn process pending tasks");
 
+    // run the checkpoint runner
+    if args.enable_checkpoint_runner {
+        let checkpoint_operator = operator.checkpoint_operator().clone();
+        let task_tracker_for_checkpoint = task_tracker.clone();
+        let db_for_checkpoint = db.clone();
+
+        spawn(async move {
+            checkpoint_proof_runner(
+                checkpoint_operator,
+                task_tracker_for_checkpoint,
+                db_for_checkpoint,
+            )
+            .await
+        });
+    }
+
     // Run prover manager in dev mode or runner mode
     if args.enable_dev_rpcs {
         // Run the RPC server on dev mode only

bin/prover-client/src/operators/checkpoint.rs

@@ -68,6 +68,11 @@
             .map_err(|e| ProvingTaskError::RpcError(e.to_string()))?
             .ok_or(ProvingTaskError::WitnessNotFound)
     }
+
+    /// Returns a reference to the internal CL (Consensus Layer) `HttpClient`.
+    pub fn cl_client(&self) -> &HttpClient {
+        &self.cl_client
+    }
 }
 
 impl ProvingOp for CheckpointOperator {