diff --git a/engine/src/pacemaker.rs b/engine/src/pacemaker.rs index dbea204c..444389a2 100644 --- a/engine/src/pacemaker.rs +++ b/engine/src/pacemaker.rs @@ -7,6 +7,7 @@ use ledger::schema::Block; use nucleus::{ Slot, config::BlockstoreParams, + runtime, shutdown::{Service, ShutdownHandle, ShutdownManager, ShutdownReason}, unix_time, }; @@ -179,12 +180,16 @@ impl PaceMaker { /// race it. Holding the boundary here is what buys that exclusivity, at the /// cost of stalling block production until the seal completes. async fn handle(&self, block: Block) -> Result<()> { - self.sequencer.send(SequencerMessage::Block(block)).await?; self.sequencer.simulation.send(SimulatorMessage::Block(block)).await?; - if block.slot.is_multiple_of(self.superblock.get()) { - let _guard = self.barrier().await?; - self.finalize_superblock()?; + if !block.slot.is_multiple_of(self.superblock.get()) { + self.sequencer.send(SequencerMessage::Block(block)).await?; + return Ok(()); } + + let (controller, guard) = runtime::barrier(); + self.sequencer.send(SequencerMessage::Checkpoint(block, guard)).await?; + controller.acknowledged.await?; + self.finalize_superblock()?; Ok(()) } } diff --git a/nucleus/README.md b/nucleus/README.md index 94bc691a..219947f7 100644 --- a/nucleus/README.md +++ b/nucleus/README.md @@ -20,8 +20,9 @@ zero when the system clock predates the epoch. Its default feature set is empty. - `metrics`: Prometheus metric construction, `engine_`-namespaced registration, labels, and timers. - `service`: the `metrics` and `shutdown` feature bundle. -- `runtime`: transaction views, execution messages, sequencer handles, and the - quiescence barrier; it also enables `ledger`, `service`, and `tls`. +- `runtime`: transaction views, execution messages, sequencer handles, and + quiescence barriers, including atomic block-checkpoint pauses; it also enables + `ledger`, `service`, and `tls`. - `tls`: thread-local MagicRoot authority and encoded service-message state. - `testkit`: engine-independent fixtures, temporary directories, Legacy/V0/V1 transaction encoding, v42 instructions, transaction views, and tracing setup diff --git a/nucleus/src/runtime.rs b/nucleus/src/runtime.rs index 06735410..0fd4ecbd 100644 --- a/nucleus/src/runtime.rs +++ b/nucleus/src/runtime.rs @@ -44,6 +44,8 @@ pub enum SequencerMessage { Transaction(TransactionView), /// A block boundary to seal before scheduling further transactions. Block(Block), + /// Finalize a block boundary, then pause before accepting subsequent work. + Checkpoint(Block, BarrierGuard), /// Quiesce the sequencer and all its executors until released — used to take /// a consistent snapshot at superblock boundaries (see ledger replay and /// `finalize_superblock`). diff --git a/processor/README.md b/processor/README.md index f250e7c6..a6c0b3b4 100644 --- a/processor/README.md +++ b/processor/README.md @@ -22,7 +22,8 @@ that contains it. The sequencer barrier drains all executor work, acknowledges the caller, and holds new execution until its guard is released. Engine uses the barrier for coherent superblock snapshots, replay seal checks, replication handshakes, and -shutdown. +shutdown. A superblock checkpoint finalizes its block and enters that pause as +one sequencer message, so later transactions cannot enter the sealed snapshot. ## Simulation diff --git a/processor/src/sequencer/mod.rs b/processor/src/sequencer/mod.rs index 2970ed20..f4e2ed6c 100644 --- a/processor/src/sequencer/mod.rs +++ b/processor/src/sequencer/mod.rs @@ -165,6 +165,11 @@ impl Sequencer { match msg { SequencerMessage::Transaction(txn) => self.schedule(txn).await, SequencerMessage::Block(block) => self.finalize(block).await, + SequencerMessage::Checkpoint(block, guard) => { + self.finalize(block).await?; + self.pause(guard).await; + Ok(()) + } SequencerMessage::Barrier(guard) => self.barrier(guard).await, } } @@ -227,12 +232,17 @@ impl Sequencer { /// released before resuming. Used to take a consistent state snapshot at /// superblock boundaries. async fn barrier(&mut self, guard: BarrierGuard) -> Result<()> { - info!("sequencer is halting operation"); self.drain().await?; + self.pause(guard).await; + Ok(()) + } + + /// Acknowledges quiescence and holds the sequencer until released. + async fn pause(&self, guard: BarrierGuard) { + info!("sequencer is halting operation"); let _ = guard.acknowledged.send(()); let _ = guard.released.await; info!("sequencer is resuming operation"); - Ok(()) } /// Awaits executor-ready signals, reclaiming each finished executor