feat: add batch-submitter module

Author: stephenctw

AGENTS.md

@@ -75,6 +75,7 @@ Primary objective in this phase: make sequencer behavior, safety checks, and per
 - Storage model is append-oriented; avoid mutable status flags for open/closed entities.
 - Open batch/frame are derived by “latest row” convention.
 - A frame’s leading direct-input prefix is derivable from `sequenced_l2_txs` plus `frames.safe_block`.
+- `direct_inputs` contains only L1 app direct input **bodies**. InputBox payload first byte: **0x00** = direct input (tag stripped, body stored and executed), **0x01** = batch submission (for scheduler, not stored), **others** = discarded (invalid/garbage). The input reader only accepts 0x00-tagged payloads and stores `payload[1..]`.
 - Safe cursor/head values should be derived from persisted facts when possible, not duplicated as mutable fields.
 - Replay/catch-up must use persisted ordering plus persisted frame fee (`frames.fee`) to mirror inclusion semantics.
 - Included user-op identity is constrained by `UNIQUE(sender, nonce)`.

Cargo.lock

@@ -4,6 +4,16 @@
 use crate::user_op::UserOp;
 use ssz_derive::{Decode, Encode};
 
+/// Tag byte for InputBox payloads that are L1 app direct inputs (e.g. deposits).
+/// L1/app must post such inputs as `0x00 || body`. Only these are stored (body only) and executed.
+pub const INPUT_TAG_DIRECT_INPUT: u8 = 0x00;
+
+/// Tag byte for InputBox payloads that are batch submissions (for the off-chain scheduler).
+/// Batch submitter posts `0x01 || ssz(Batch)`. These are not stored in `direct_inputs`.
+pub const INPUT_TAG_BATCH_SUBMISSION: u8 = 0x01;
+
+/// Any other first byte is invalid; the input reader discards such payloads (garbage).
+
 #[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
 pub struct Batch {
     pub frames: Vec<Frame>,
@@ -35,3 +45,28 @@ impl WireUserOp {
         }
     }
 }
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct BatchForSubmission {
+    pub batch_index: u64,
+    pub created_at_ms: u64,
+    pub batch: Batch,
+}
+
+impl BatchForSubmission {
+    /// Encode the batch payload for the on-chain scheduler.
+    ///
+    /// This uses the same SSZ encoding that the canonical scheduler expects when
+    /// decoding inputs, and prefixes the payload with:
+    /// - a single tag byte so that batch submissions can be distinguished from other InputBox inputs; and
+    /// - the batch index as an 8-byte big-endian nonce, so the scheduler can deduplicate.
+    pub fn encode_for_scheduler(&self) -> Vec<u8> {
+        let mut out = Vec::new();
+        // First byte identifies this InputBox payload as a batch submission.
+        out.push(INPUT_TAG_BATCH_SUBMISSION);
+        // Next 8 bytes carry the batch index as a big-endian nonce for the scheduler.
+        out.extend_from_slice(&self.batch_index.to_be_bytes());
+        out.extend(ssz::Encode::as_ssz_bytes(&self.batch));
+        out
+    }
+}

sequencer-core/src/batch.rs

@@ -23,13 +23,14 @@ rusqlite = { version = "0.38.0", features = ["bundled"] }
 rusqlite_migration = "2.3.0"
 alloy-primitives = { version = "1.4.1", features = ["serde", "k256"] }
 alloy-sol-types = "1.4.1"
-alloy = { version = "1.0", features = ["contract", "network", "reqwest", "rpc-types", "sol-types", "node-bindings"] }
+alloy = { version = "1.0", features = ["contract", "network", "reqwest", "rpc-types", "sol-types", "node-bindings", "signer-local", "signers"] }
 thiserror = "1"
 ssz = { package = "ethereum_ssz", version = "0.10" }
 ssz_derive = { package = "ethereum_ssz_derive", version = "0.10" }
 clap = { version = "4", features = ["derive", "env"] }
 async-recursion = "1"
 cartesi-rollups-contracts = "=2.2.0"
+async-trait = "0.1"
 
 [dev-dependencies]
 futures-util = "0.3"

sequencer/Cargo.toml

@@ -0,0 +1,42 @@
+// (c) Cartesi and individual authors (see AUTHORS)
+// SPDX-License-Identifier: Apache-2.0 (see LICENSE)
+
+use std::time::Duration;
+
+use clap::Parser;
+
+/// Batch-submitter-specific options. L1 RPC URL and InputBox address are shared with the
+/// input reader and come from the same discovery at startup (see `L1Config` in `config`).
+#[derive(Debug, Clone, Parser)]
+pub struct BatchSubmitterConfig {
+    /// Number of blocks to look back from the chain head when reading
+    /// the latest submitted batch nonce.
+    #[clap(
+        long = "seq-batch-submitter-confirmations-depth",
+        env = "SEQ_BATCH_SUBMITTER_CONFIRMATIONS_DEPTH",
+        default_value = "5"
+    )]
+    pub confirmations_depth: u64,
+
+    /// How often the submitter polls storage and L1 for new work.
+    #[clap(
+        long = "seq-batch-submitter-idle-poll-interval-ms",
+        env = "SEQ_BATCH_SUBMITTER_IDLE_POLL_INTERVAL_MS",
+        default_value = "5000"
+    )]
+    pub idle_poll_interval_ms: u64,
+
+    /// Maximum number of batches to submit in a single loop iteration.
+    #[clap(
+        long = "seq-batch-submitter-max-batches-per-loop",
+        env = "SEQ_BATCH_SUBMITTER_MAX_BATCHES_PER_LOOP",
+        default_value = "4"
+    )]
+    pub max_batches_per_loop: usize,
+}
+
+impl BatchSubmitterConfig {
+    pub fn idle_poll_interval(&self) -> Duration {
+        Duration::from_millis(self.idle_poll_interval_ms)
+    }
+}

sequencer/src/batch_submitter/config.rs

@@ -0,0 +1,14 @@
+// (c) Cartesi and individual authors (see AUTHORS)
+// SPDX-License-Identifier: Apache-2.0 (see LICENSE)
+
+//! Batch submitter: posts closed batches to L1 with at-least-once semantics.
+//!
+//! The batch index is used as the batch nonce (id). The scheduler checks that nonces are
+//! strictly increasing and invalidates otherwise, so duplicates are deduplicated at the
+//! scheduler level. See `worker` for the wake → read S → compare → submit → sleep loop.
+
+mod config;
+mod worker;
+
+pub use config::BatchSubmitterConfig;
+pub use worker::BatchSubmitter;