feat: drive the batch builder through a typestate seal-ahead pipeline

[bronxyz]

Summary

• BatchOrdering publishes this authority's accepted batch sequence as a watch, seeded from the persisted last_executed_seq and advanced on in-order accept, overflow-forced accept, and parked-batch drain. It marks acceptance into execution order, not block finalization.
• Replace the poll-based BatchBuilder future with an async run loop over a BatchPipeline typestate, so an illegal build/seal ordering cannot be represented. The builder seals up to MAX_SEAL_AHEAD batches ahead of its own watermark and then stalls until execution advances, keeping the in-flight prefix inside the per-authority parking budget; the budget collapses to one batch inside the boundary quiesce window and the builder ends once the canonical tip reaches the epoch boundary.
• On restart the sequence resumes from the execution watermark rather than from the persisted counter alone.

Stack 5/9 of the txpool in-flight tracker and observer-forwarder series.

Surface areas touched

• Consensus protocol (primary / worker / network / state-sync)
• Execution / EVM
• JSON-RPC (eth_*, rayls_*, faucet)
• Middleware (orchestrator / processor / bridge)
• Infrastructure (types / storage / config / network-cli)
• On-chain contracts (rayls-contracts/)
• Operations (etc/, scripts, Docker, compose)
• CI / build (.github/workflows/, Makefile)
• Documentation only (doc/, in-crate READMEs, root docs)
• Tests only

Breaking / compatibility

None. Node-local pacing only; no wire, storage, or fork changes. A fast proposer now seals at most MAX_SEAL_AHEAD batches past its own executed sequence.

Test plan

• tests/it/pipeline_typestate.rs covers the phase transitions and the seal-ahead / quiesce / closed budget.
• tests/it/build_batches.rs runs the builder through the new run() loop end to end.
• make check on the stack tip; CI on this branch.

[github-actions]

Security Scan - Code

Severity: HIGH, CRITICAL

No vulnerabilities found

View scan results

Report Summary

┌─────────────────────────────────────────────┬────────────┬─────────────────┬───────────────────┐
│                   Target                    │    Type    │ Vulnerabilities │ Misconfigurations │
├─────────────────────────────────────────────┼────────────┼─────────────────┼───────────────────┤
│ Cargo.lock                                  │   cargo    │        0        │         -         │
├─────────────────────────────────────────────┼────────────┼─────────────────┼───────────────────┤
│ crates/testing/fuzz-targets/Cargo.lock      │   cargo    │        0        │         -         │
├─────────────────────────────────────────────┼────────────┼─────────────────┼───────────────────┤
│ crates/testing/fuzz-targets/fuzz/Cargo.lock │   cargo    │        0        │         -         │
├─────────────────────────────────────────────┼────────────┼─────────────────┼───────────────────┤
│ etc/state-sum/Cargo.lock                    │   cargo    │        0        │         -         │
├─────────────────────────────────────────────┼────────────┼─────────────────┼───────────────────┤
│ etc/tps/package-lock.json                   │    npm     │        0        │         -         │
├─────────────────────────────────────────────┼────────────┼─────────────────┼───────────────────┤
│ rayls-contracts/package-lock.json           │    npm     │        0        │         -         │
├─────────────────────────────────────────────┼────────────┼─────────────────┼───────────────────┤
│ etc/chaos-network/Dockerfile                │ dockerfile │        -        │         0         │
├─────────────────────────────────────────────┼────────────┼─────────────────┼───────────────────┤
│ etc/docker-network/Dockerfile               │ dockerfile │        -        │         0         │
├─────────────────────────────────────────────┼────────────┼─────────────────┼───────────────────┤
│ etc/docker-replay/Dockerfile                │ dockerfile │        -        │         0         │
└─────────────────────────────────────────────┴────────────┴─────────────────┴───────────────────┘
Legend:
- '-': Not scanned
- '0': Clean (no security findings detected)

[github-actions]

Claude finished @bronxyz's task in 3m 36s —— View job

---

Code Review

• Read CLAUDE.md for repo-specific instructions
• Read changed files and understand the implementation
• Review pipeline.rs (new typestate machine)
• Review lib.rs (main builder loop)
• Review watermark.rs (new watermark helper)
• Review ordering.rs (BatchOrdering changes)
• Review supporting files (batch.rs, error.rs, node_inner.rs, etc.)
• Check status of prior review findings
• Post final review

---

This is an independent review against the current diff. The structural shape — typestate machine, run() loop, select! dispatch — is sound. What follows covers carry-over findings from the previous round and two new observations from reading the fresh diff.

---

Required fix (carried over, still blocking approval)

Bug: elapsed_ms in quorum logs always ~0

lib.rs:322 captures start_time after pipeline.await_quorum() has already completed and returned res via the select! branch. The quorum wait is already over before handle_quorum_resolution is called, so elapsed_ms measures only internal function-call overhead (nanoseconds), not the real quorum latency.

// lib.rs:307–361
fn handle_quorum_resolution(
    &mut self,
    res: Result<...>,          // ← quorum already resolved here
    pipeline: PipelineState,
    interval: &mut Interval,
) -> BatchBuilderResult<Option<PipelineState>> {
    let awaiting = match pipeline { ... };
    let current_seq = awaiting.current_seq();
    let start_time = std::time::Instant::now(); // ← too late, result is already in `res`
    ...
    TaskOutcome::QuorumFailed => {
        warn!(..., elapsed_ms = start_time.elapsed().as_millis(), ...); // always ~0
    }
    TaskOutcome::QuorumSucceeded { ... } => {
        debug!(..., elapsed_ms = start_time.elapsed().as_millis(), ...); // always ~0
    }
}

Fix: Add a spawn_instant: std::time::Instant field to TaskOutcome::QuorumSucceeded and TaskOutcome::QuorumFailed in pipeline.rs, capture it at the top of spawn_build_task in lib.rs, and use it in handle_quorum_resolution instead of the post-hoc Instant::now(). Fix this →

---

Medium (carried over, unaddressed)

check_boundary body copy-pasted across four impl blocks

pipeline.rs lines 183, 214, 244, and 310 are byte-for-byte identical:

if self.data.last_canonical_timestamp >= epoch_boundary {
    Err(BatchPipeline { state: Closed, data: self.data })
} else {
    Ok(self)
}

The only reason this can't be on impl<S> BatchPipeline<S> is the return type — but a free helper fn is_boundary_reached(data: &PipelineData, epoch_boundary: u64) -> bool eliminates the body duplication while keeping each impl's consuming signature intact. Fix this →

Inner spawn_blocking invisible to task_spawner shutdown tracking

In spawn_build_task (lib.rs:374-440), task_spawner.spawn_task wraps an async block that itself calls tokio::task::spawn_blocking. The spawner tracks the outer async task's join handle; the blocking thread is not tracked. If the outer task's future is dropped mid-await on the blocking join handle, the blocking work (build_batch) continues on the thread pool until natural completion. For build_batch this is short-lived, but it means the blocking thread can outlive the tracked task boundary and is invisible to whatever shutdown contract task_spawner enforces. Worth a comment documenting why this is acceptable, or replacing with a structured cancellation token if the spawner's contract assumes all work stops.

AwaitingQuorum::check_boundary silently drops an in-flight result — needs a doc comment

lib.rs:184 calls pipeline.check_boundary(...) unconditionally, including when in AwaitingQuorum. BatchPipeline<AwaitingQuorum>::check_boundary consumes self, dropping outcome_rx. If the spawned task had already passed the outcome_tx.is_closed() guard at lib.rs:405 and submitted the batch before the receiver was dropped, the worker can still reach quorum — but QuorumSucceeded is never processed and in-flight marks are never set.

This is safe (marks are epoch-scoped and reset at rollover; the transactions re-seal next epoch), but there is no comment anywhere explaining the invariant. Future readers of BatchPipeline<AwaitingQuorum>::check_boundary or the check_boundary callsite in run() will not understand why dropping the receiver mid-flight is intentional. Fix this →

---

Low (carried over, unaddressed)

Unconditional on_event() at startup fires a build on an empty pool

// lib.rs:181
pipeline = self.try_start_build(pipeline.on_event(), &mut interval);

on_event() unconditionally transitions Clean → Accumulating, then try_start_build immediately spawns a blocking build_batch call. On a freshly opened epoch against an idle node, this produces a spawn_blocking round-trip that returns NothingToSeal and discards the thread wakeup. A pool.pool_size().pending > 0 guard before on_event() avoids the wasted work.

---

New finding — Nit: typo in error.rs:38

// error.rs:38
"The transaction was not included becuase it would exceed..."
//                                 ^^^^^^ "becuase" should be "because"

Fix this →

---

Nits (carried over, unaddressed)

• No integration test for Quiescing budget cap in the live BatchBuilder. pipeline_typestate.rs verifies the budget math, but no test in build_batches.rs confirms the live builder limits itself to one outstanding batch when canonical_tip >= epoch_boundary - BOUNDARY_QUIESCE_WINDOW_SECS. builder_stalls_at_the_seal_ahead_budget_until_the_watermark_advances is the right model.
• node_inner.rs:175: .expect("own_executed_sequence must be initialized…") is consistent with programmer-error semantics but diverges from the .ok_or_eyre(…)? style used elsewhere in the file.

---

What's correct and well-designed

The core contribution is sound:

• Typestate machine (pipeline.rs): the Clean → Accumulating → AwaitingQuorum → {Clean | Accumulating | BacklogDraining} graph is correctly represented; an illegal build-without-accumulate or double-seal is unrepresentable at the type level.
• PipelineState::await_quorum parks with std::future::pending() in every non-AwaitingQuorum phase, making the select! branch inert without any flag.
• calculate_in_flight_depth: the start_seq.saturating_sub(1) fallback when execution has not reported yet is correct — it ensures the budget check starts conservative.
• BatchOrdering::drain_consecutive correctly calls update_executed_watermark for drained own batches (line 222), so the builder's pacing watch never stalls on a parked-then-drained batch.
• from_history restart semantics: the epoch + 1 == current_epoch window that preserves an undrained closing epoch's parked set (rather than reseeding and dropping them) is the right guard to avoid the chain-fork-on-restart class.
• OwnWatermarkReceiver::resume_seq: prefers executed_watermark + 1 over persisted_next correctly covers the restart-past-the-persisted-counter case.
• SelectedForSeal #[must_use]: good defensive annotation to prevent silent in-flight mark drops.
• Tests: executed_own_watermark_advances_when_parked_own_batches_drain directly exercises the drain-path wake, and builder_stalls_at_the_seal_ahead_budget_until_the_watermark_advances exercises the pacing invariant end-to-end.

---

Summary

Severity	Finding	Status
Bug	elapsed_ms always ~0 (lib.rs:322)	Unaddressed — required for approval
Medium	check_boundary body duplicated 4× (pipeline.rs:183,214,244,310)	Unaddressed
Medium	Inner spawn_blocking invisible to task_spawner shutdown tracking	Unaddressed
Medium	AwaitingQuorum::check_boundary needs doc comment explaining intentional result-drop	Unaddressed
Low	Unconditional on_event() at startup fires empty build (lib.rs:181)	Unaddressed
Nit	No live-builder test for Quiescing budget cap	Unaddressed
Nit	node_inner.rs:175 expect vs .ok_or_eyre	Unaddressed
Nit	Typo "becuase" in error.rs:38	New

The elapsed_ms logging bug remains the single required fix for approval.