diff --git a/replicator/README.md b/replicator/README.md index d5a5790..8a95879 100644 --- a/replicator/README.md +++ b/replicator/README.md @@ -50,14 +50,19 @@ at startup before producing their first new block, so followers clear chain-mirrored volatile state at the same stream position while retaining internal system accounts. +On normal follower shutdown, Control keeps consuming ordered transaction batches +until it validates the next replicated block, then barriers execution and +flushes the cursor at that boundary before stopping Ingest. The operational +block heartbeat supplies that boundary, reconnecting first when necessary. +Replication failure and snapshot restart paths do not claim this guarantee. + Ingest decodes transaction batches of at most 128 transactions and typically 128 KiB, fencing them before every block, superblock, reset, or reconnect. A rendezvous channel assigns verification to an idle control thread; otherwise -`Ingest::flush` verifies while Control schedules earlier work. Control alone -schedules verified transactions and advances the block pacer. During the -Control-held handshake, `Ingest::stage_snapshot` may write the snapshot archive -and bootstrap durable superblock state. These roles preserve stream order -without reordering state. +Ingest verifies while Control schedules earlier work. Control alone owns +handshakes, reconnect cursors, execution barriers, snapshot staging, transaction +scheduling, and block pacing. These roles preserve stream order without a +reverse control channel. A shared-key follower may also serve downstream followers. It derives and validates superblock seals from replicated block boundaries and archives its own diff --git a/replicator/src/client.rs b/replicator/src/client.rs index 575912a..765c9b5 100644 --- a/replicator/src/client.rs +++ b/replicator/src/client.rs @@ -1,17 +1,17 @@ use std::{ fs::{self, File}, io::{self, BufReader, Read}, + mem, net::{SocketAddr, TcpStream}, - sync::mpsc, - thread, + thread::{self, JoinHandle}, }; use derive_more::Deref; use engine::{ - Engine, EngineError, ReplayError, TransactionAccessor, VerifiedTransaction, - pacemaker::ExternalBlock, + Engine, EngineError, ReplayError, TransactionAccessor, TransactionVerifier, + VerifiedTransaction, pacemaker::ExternalBlock, }; -use flume::{Sender, TrySendError}; +use flume::{Receiver, Sender, TrySendError}; use ledger::{ Superblock, schema::{Block, OwnedBlockstoreEntry, blockstore}, @@ -20,7 +20,7 @@ use nucleus::{ KB, ledger::{ACCOUNTSDB_SNAPSHOT_FILE, BlockstorePosition}, runtime::BarrierHandle, - shutdown::{CancellationToken, Service, ShutdownHandle, ShutdownManager, ShutdownReason}, + shutdown::{Service, ShutdownHandle, ShutdownManager, ShutdownReason}, }; use tokio::{ runtime, @@ -38,7 +38,6 @@ use crate::{ }; type ReplicationStream = BufReader; -type ReconnectReply = mpsc::SyncSender; /// Maximum transactions retained before offering a batch to Control. const MAX_BATCH_TRANSACTIONS: usize = 128; @@ -46,27 +45,13 @@ const MAX_BATCH_TRANSACTIONS: usize = 128; const MAX_BATCH_BYTES: usize = 128 * KB; /// Consecutive transaction payloads accumulated between ordered stream fences. +#[derive(Default)] struct TransactionsBatch { - /// Raw transaction payloads in stream order. transactions: Vec>, - /// Cumulative payload bytes used to enforce the batch bound. bytes: usize, } -impl Default for TransactionsBatch { - fn default() -> Self { - Self { - transactions: Vec::with_capacity(MAX_BATCH_TRANSACTIONS), - bytes: 0, - } - } -} - impl TransactionsBatch { - fn is_empty(&self) -> bool { - self.transactions.is_empty() - } - fn push(&mut self, transaction: Vec) { self.bytes = self.bytes.saturating_add(transaction.len()); self.transactions.push(transaction); @@ -75,55 +60,57 @@ impl TransactionsBatch { fn is_full(&self) -> bool { self.transactions.len() >= MAX_BATCH_TRANSACTIONS || self.bytes >= MAX_BATCH_BYTES } - - fn take(&mut self) -> Vec> { - self.bytes = 0; - std::mem::replace( - &mut self.transactions, - Vec::with_capacity(MAX_BATCH_TRANSACTIONS), - ) - } } -/// Ordered handoff from blocking stream ingest to asynchronous Engine control. +/// Ordered handoff from blocking Ingest to asynchronous Control. enum ReplicationMessage { - /// Raw batch offered to Control for signature verification. + /// Raw transactions awaiting authority and signature verification. Unverified(Vec>), - /// Verification result completed by Ingest while Control was occupied. - Verified(engine::Result>), - /// Control entry fenced behind every preceding transaction batch. + /// Transactions verified by Ingest while Control was occupied. + Verified(Vec), + /// Non-transaction entry fenced behind every preceding batch. Entry(OwnedBlockstoreEntry), - /// Successful handshake allowing Control to release its sequencing barrier. - Connected, - /// Lost stream requesting Control's next durable resume position. - Disconnected(ReconnectReply), } -/// Owns stream decoding, bounded transaction accumulation, and opportunistic verification. +/// Why connection-scoped Ingest stopped without a terminal replication error. +enum IngestExit { + /// Control dropped its receiver after reaching a boundary or terminal error. + Stopped, + /// The transport failed after all preceding entries were handed to Control. + Disconnected(wincode::io::ReadError), +} + +/// Why Control stopped consuming one connection. +enum ControlExit { + /// Normal shutdown reached and flushed a validated block boundary. + Boundary(BlockstorePosition), + /// Ingest ended; its join result determines whether to reconnect or fail. + HandoffClosed, +} + +/// Decodes one connection and opportunistically verifies bounded transaction batches. struct Ingest { - /// Engine used for batch verification and authenticated stream recovery. - engine: Engine, - /// Upstream replication endpoint reused across reconnects. - addr: SocketAddr, - /// Consecutive transactions awaiting an ordered handoff. + /// Blocking stream for one authenticated connection. + stream: ReplicationStream, + /// Transactions accumulated until a size or entry fence. batch: TransactionsBatch, - /// Rendezvous sender preserving ingest-to-Control message order. + /// Rendezvous handoff preserving decoded stream order. tx: Sender, - /// Cancellation scoped to the ingest worker lifecycle. - shutdown: CancellationToken, + /// Authority-bound verifier used when Control is occupied. + verifier: TransactionVerifier, } /// Pulls a leader blockstore stream into an externally paced follower engine. #[derive(Deref)] pub struct ReplicationClient { - /// Engine receiving replicated transactions, boundaries, seals, and resets. + /// Engine receiving replicated state. #[deref] engine: Engine, - /// Leader endpoint reused after transport loss. + /// Leader endpoint reused for reconnects. addr: SocketAddr, - /// External pacemaker channel used to preserve block-boundary ordering. + /// External block source for the follower pacemaker. pacer: PacerSender, - /// Locally committed block boundaries used to verify replicated output. + /// Locally committed blocks used to validate replicated boundaries. blocks: BlockReceiver, } @@ -138,7 +125,6 @@ impl ReplicationClient { metrics::init(); let shutdown = shutdown.handle(Service::ReplicationClient); let mut blocks = engine.blocks().subscribe(); - // drain the channel from potential leftovers while blocks.try_recv().is_ok() {} let client = Self { engine, addr, pacer, blocks }; let rt = runtime::Builder::new_current_thread().enable_time().build()?; @@ -148,102 +134,88 @@ impl ReplicationClient { Ok(()) } - /// Consumes the leader stream and reports why the client stopped. + /// Reports the terminal client outcome to shutdown management. async fn serve(self, mut shutdown: ShutdownHandle) { - let result = self.run(&shutdown).await; - if shutdown.requested() || result.is_ok() { - shutdown.terminate(ShutdownReason::Signalled); - return; - } - match result { + let reason = match self.run(&shutdown).await { + Ok(position) => { + info!(?position, "replication stopped at a durable boundary"); + ShutdownReason::Signalled + } Err(ReplicationError::RestartRequired(slot)) => { info!(%slot, "replication client has requested node restart"); - shutdown.terminate(ShutdownReason::RestartRequired); + ShutdownReason::RestartRequired } - Err(error) => { - shutdown.terminate(ShutdownReason::Error(Box::new(error))); - } - Ok(()) => (), - } + Err(error) => ShutdownReason::Error(Box::new(error)), + }; + shutdown.terminate(reason); } - /// Starts Ingest and joins it after Control stops consuming its ordered messages. - async fn run(self, shutdown: &ShutdownHandle) -> Result<()> { - let (guard, position) = self.resume().await?; - let (tx, rx) = flume::bounded(0); - let cancellation = shutdown.child(); - let mut ingest = Ingest { - engine: self.engine.clone(), - addr: self.addr, - batch: Default::default(), - tx, - shutdown: cancellation.clone(), - }; - let rt = runtime::Builder::new_current_thread().enable_time().build()?; - let ingest = thread::Builder::new() - .name("replication-ingest".into()) - .spawn(move || rt.block_on(ingest.run(position)))?; - let mut result = self.consume(shutdown, rx, guard).await; - cancellation.cancel(); - match ingest.join() { - Ok(Ok(())) => info!("replication ingest has gracefully shutdown"), - Ok(Err(error)) => result = result.and(Err(error)), - Err(error) => error!(?error, "replication ingest panicked"), + /// Owns connection recovery and returns success only at a validated block boundary. + async fn run(mut self, shutdown: &ShutdownHandle) -> Result { + let (mut barrier, mut position) = self.resume().await?; + loop { + let stream = self.reconnect(position).await?; + let connected = metrics::client_connection(); + drop(barrier); + + let (rx, ingest) = Ingest::spawn(stream, self.verifier())?; + let control = self.consume(shutdown, &rx).await; + drop(rx); + let ingest = ingest.join().map_err(|_| ReplicationError::IngestPanicked)?; + drop(connected); + + match control { + Ok(ControlExit::Boundary(position)) => return Ok(position), + Err(error) => return Err(error), + Ok(ControlExit::HandoffClosed) => match ingest? { + IngestExit::Disconnected(error) => { + warn!(%error, "replication stream disconnected"); + (barrier, position) = self.resume().await?; + } + IngestExit::Stopped => return Err(ReplicationError::IngestStopped), + }, + } } - result } - /// Consumes ordered Ingest messages until shutdown or a terminal failure. + /// Consumes one Ingest stream, draining normal shutdown to the next block. async fn consume( - mut self, + &mut self, shutdown: &ShutdownHandle, - rx: flume::Receiver, - guard: BarrierHandle, - ) -> Result<()> { + rx: &Receiver, + ) -> Result { let verifier = self.verifier(); - let mut connected = None; - let mut barrier = Some(guard); - + let mut draining = shutdown.requested(); loop { - if shutdown.requested() { - return Ok(()); - } - // Complete a ready handoff before observing concurrent cancellation. let message = tokio::select! { biased; message = rx.recv_async() => match message { - Ok(m) => m, - // Ingest has shutdown, the potential error will be captured by caller - Err(_) => return Ok(()), + Ok(message) => message, + Err(_) => return Ok(ControlExit::HandoffClosed), + }, + _ = shutdown.signalled(), if !draining => { + draining = true; + continue; }, - _ = shutdown.signalled() => break, - }; match message { ReplicationMessage::Unverified(batch) => { - let verified = verifier.verify(batch)?; - self.schedule(verified).await?; - } - ReplicationMessage::Verified(result) => self.schedule(result?).await?, - ReplicationMessage::Entry(entry) => self.process(entry).await?, - ReplicationMessage::Connected => { - connected = Some(metrics::client_connection()); - barrier.take(); + self.schedule(verifier.verify(batch)?).await?; } - ReplicationMessage::Disconnected(reply) => { - connected.take(); - let (guard, position) = self.resume().await?; - barrier = Some(guard); - if reply.send(position).is_err() { - return Err(ReplicationError::StreamClosed); + ReplicationMessage::Verified(batch) => self.schedule(batch).await?, + ReplicationMessage::Entry(entry) => { + let boundary = matches!(entry, OwnedBlockstoreEntry::Block(_)); + self.process(entry).await?; + if boundary && draining { + let (_guard, position) = self.resume().await?; + return Ok(ControlExit::Boundary(position)); } } } } - Ok(()) } - /// Schedules a verified batch in stream order without repeating admission checks. + /// Schedules a verified batch in stream order. async fn schedule(&self, transactions: Vec) -> Result<()> { for transaction in transactions { TransactionAccessor::verified(&self.engine, transaction).schedule().await?; @@ -251,14 +223,14 @@ impl ReplicationClient { Ok(()) } - /// Applies one control entry after all preceding transactions are scheduled. + /// Applies and validates one non-transaction stream entry. async fn process(&mut self, entry: OwnedBlockstoreEntry) -> Result<()> { match entry { OwnedBlockstoreEntry::Block(block) => { let (external, guard) = ExternalBlock::new(block); self.pacer.send(external).await.map_err(EngineError::from)?; let pending = time::timeout(IO_TIMEOUT, self.blocks.recv()); - let observed = pending.await?.ok_or(ReplicationError::StreamClosed)?; + let observed = pending.await?.ok_or(ReplicationError::BlockStreamClosed)?; if block != observed { let error = ReplayError::BlockhashMismatch(block.slot); Err(EngineError::from(error))?; @@ -266,7 +238,6 @@ impl ReplicationClient { guard.await.map_err(EngineError::from)?; } OwnedBlockstoreEntry::Superblock(expected) => { - // The preceding boundary finalized local state; this seal only validates it. let observed = self.superblocks().sealed(); if observed != expected { error!(?expected, ?observed, "replication state mismatch detected"); @@ -277,101 +248,20 @@ impl ReplicationClient { OwnedBlockstoreEntry::Reset(slot) => { self.engine.replay(OwnedBlockstoreEntry::Reset(slot)).await?; } - OwnedBlockstoreEntry::Transaction(_) => (), + OwnedBlockstoreEntry::Transaction(_) => unreachable!("Ingest batches transactions"), } Ok(()) } - /// Flushes prior work and returns its durable cursor under a sequencing barrier. async fn resume(&self) -> Result<(BarrierHandle, BlockstorePosition)> { let guard = self.barrier().await?; - self.sync(false)?; + self.superblocks().sync(false)?; Ok((guard, self.superblocks().position())) } -} - -impl Ingest { - /// Decodes the stream while preserving the order of transactions and control entries. - async fn run(&mut self, position: BlockstorePosition) -> Result<()> { - let mut stream = self.open(position).await?; - while !self.shutdown.is_cancelled() { - match blockstore::decode(&mut stream) { - Ok(OwnedBlockstoreEntry::Transaction(transaction)) => { - if !self.push(transaction) { - return Ok(()); - } - } - Ok(entry) => { - if !self.flush() { - return Ok(()); - } - self.tx - .send(ReplicationMessage::Entry(entry)) - .map_err(|_| ReplicationError::StreamClosed)?; - } - Err(wincode::error::ReadError::Io(error)) => { - warn!(%error, "replication stream disconnected"); - drop(stream); - if !self.flush() { - return Ok(()); - } - let position = self.request_position()?; - stream = self.open(position).await?; - } - Err(error) => { - if !self.flush() { - return Ok(()); - } - return Err(wincode::Error::from(error).into()); - } - } - } - Ok(()) - } - - /// Adds a transaction and flushes once either batch bound is reached. - fn push(&mut self, transaction: Vec) -> bool { - self.batch.push(transaction); - !self.batch.is_full() || self.flush() - } - - /// Offers the batch to Control, verifying it locally when Control is occupied. - fn flush(&mut self) -> bool { - if self.batch.is_empty() { - return true; - } - let batch = self.batch.take(); - match self.tx.try_send(ReplicationMessage::Unverified(batch)) { - Ok(()) => true, - Err(TrySendError::Full(ReplicationMessage::Unverified(batch))) => { - let result = self.engine.verifier().verify(batch); - let valid = result.is_ok(); - self.tx.send(ReplicationMessage::Verified(result)).is_ok() && valid - } - Err(_) => false, - } - } - /// Requests a durable resume cursor after Control finishes all preceding work. - fn request_position(&self) -> Result { - let (reply, response) = mpsc::sync_channel(0); - let _ = self.tx.send(ReplicationMessage::Disconnected(reply)); - response.recv().map_err(|_| ReplicationError::StreamClosed) - } - - /// Reconnects from `position` and tells Control it may release the barrier. - async fn open(&mut self, position: BlockstorePosition) -> Result { - let stream = self.reconnect(position).await?; - let _ = self.tx.send(ReplicationMessage::Connected); - Ok(stream) - } - - /// Retries transport establishment while the ordered resume cursor remains quiesced. + /// Retries transport establishment from one quiesced durable cursor. async fn reconnect(&self, position: BlockstorePosition) -> Result { for attempt in 1..=MAX_RECONNECT_ATTEMPTS { - if self.shutdown.is_cancelled() { - return Err(ReplicationError::StreamClosed); - } metrics::client_connection_attempt(); match self.connect(position) { Ok(stream) => { @@ -386,15 +276,13 @@ impl Ingest { } Err(error) => return Err(error), } - let timeout = RETRY_DELAY * attempt as u32; - if time::timeout(timeout, self.shutdown.cancelled()).await.is_ok() { - Err(ReplicationError::StreamClosed)?; - } + let delay = RETRY_DELAY * attempt as u32; + time::interval_at(time::Instant::now() + delay, delay).tick().await; } Err(ReplicationError::ReconnectExhausted) } - /// Handshakes at `position`, staging a snapshot when streaming cannot resume. + /// Authenticates one resume request and returns its blockstore stream. fn connect(&self, position: BlockstorePosition) -> Result { let _timer = metrics::time(Operation::ClientConnect); let mut connection = TcpStream::connect_timeout(&self.addr, IO_TIMEOUT)?; @@ -428,10 +316,9 @@ impl Ingest { } } - /// Stages a complete snapshot and installs its seal for the requested restart. + /// Durably stages a snapshot and records its bootstrap seal. fn stage_snapshot(&self, connection: &mut TcpStream, meta: SnapshotMetadata) -> Result<()> { let _timer = metrics::time(Operation::ClientStageSnapshot); - // Stage in the successor before seal rotation so restart can find it. let superblocks = self.engine.superblocks(); let dir = Superblock::init_dir(superblocks.directory(), meta.id + 1)?; let archive = dir.join(ACCOUNTSDB_SNAPSHOT_FILE); @@ -449,3 +336,70 @@ impl Ingest { Ok(()) } } + +impl Ingest { + /// Starts one blocking decoder for an authenticated connection. + fn spawn( + stream: ReplicationStream, + verifier: TransactionVerifier, + ) -> Result<(Receiver, JoinHandle>)> { + let (tx, rx) = flume::bounded(0); + let ingest = Self { + stream, + batch: Default::default(), + tx, + verifier, + }; + let worker = thread::Builder::new() + .name("replication-ingest".into()) + .spawn(move || ingest.run())?; + Ok((rx, worker)) + } + + /// Decodes until transport loss, terminal failure, or Control exit. + fn run(mut self) -> Result { + loop { + match blockstore::decode(&mut self.stream) { + Ok(OwnedBlockstoreEntry::Transaction(transaction)) => { + self.batch.push(transaction); + if self.batch.is_full() && !self.flush()? { + return Ok(IngestExit::Stopped); + } + } + Ok(entry) => { + if !self.flush()? || self.tx.send(ReplicationMessage::Entry(entry)).is_err() { + return Ok(IngestExit::Stopped); + } + } + Err(wincode::error::ReadError::Io(error)) => { + if !self.flush()? { + return Ok(IngestExit::Stopped); + } + return Ok(IngestExit::Disconnected(error)); + } + Err(error) => { + if !self.flush()? { + return Ok(IngestExit::Stopped); + } + return Err(wincode::Error::from(error).into()); + } + } + } + } + + /// Offers raw work to idle Control, otherwise verifies without losing stream order. + fn flush(&mut self) -> Result { + if self.batch.transactions.is_empty() { + return Ok(true); + } + let batch = mem::take(&mut self.batch).transactions; + match self.tx.try_send(ReplicationMessage::Unverified(batch)) { + Ok(()) => Ok(true), + Err(TrySendError::Full(ReplicationMessage::Unverified(batch))) => { + let verified = self.verifier.verify(batch)?; + Ok(self.tx.send(ReplicationMessage::Verified(verified)).is_ok()) + } + Err(_) => Ok(false), + } + } +} diff --git a/replicator/src/error.rs b/replicator/src/error.rs index 5c8a56d..ead4ea1 100644 --- a/replicator/src/error.rs +++ b/replicator/src/error.rs @@ -48,9 +48,18 @@ pub enum ReplicationError { /// A staged snapshot must be installed by restarting the engine. #[error("replication snapshot for superblock {0} is staged; restart required")] RestartRequired(u64), - /// A replication event stream closed before the transfer completed. - #[error("replication event stream closed")] - StreamClosed, + /// The local committed-block subscription closed while validating a boundary. + #[error("replication block stream closed")] + BlockStreamClosed, + /// The leader's durable-cursor subscription closed unexpectedly. + #[error("replication cursor stream closed")] + CursorStreamClosed, + /// Ingest stopped before Control ended the connection. + #[error("replication ingest stopped unexpectedly")] + IngestStopped, + /// Ingest thread panicked. + #[error("replication ingest panicked")] + IngestPanicked, /// Waiting for a locally committed block boundary timed out. #[error("timed out waiting for a replicated block boundary: {0}")] Timeout(#[from] Elapsed), diff --git a/replicator/src/server.rs b/replicator/src/server.rs index 39dc8d1..31d6220 100644 --- a/replicator/src/server.rs +++ b/replicator/src/server.rs @@ -224,7 +224,7 @@ impl ReplicationServer { } Err(broadcast::error::RecvError::Closed) => { error!("ledger position stream closed unexpectedly"); - return Err(ReplicationError::StreamClosed); + return Err(ReplicationError::CursorStreamClosed); } } } diff --git a/replicator/tests/integration.rs b/replicator/tests/integration.rs index 353fc67..c5a9e91 100644 --- a/replicator/tests/integration.rs +++ b/replicator/tests/integration.rs @@ -102,6 +102,20 @@ async fn restart_from_snapshot(addr: SocketAddr, mut follower: TestEngine) -> Te TestEngine::with(dirs, authority).await } +/// Closes a follower after its producer publishes the boundary and Ingest-stop heartbeat. +async fn close_follower(follower: TestEngine, producer: &mut TestEngine) -> (Dirs, Authority) { + let mut close = Box::pin(follower.close()); + assert!( + time::timeout(Duration::from_millis(100), close.as_mut()).await.is_err(), + "follower waits for a replicated block boundary" + ); + producer.advance(2).await; + producer.sync().await; + time::timeout(TIMEOUT, close) + .await + .expect("follower drains through the producer boundary in time") +} + /// Applies a non-idempotent mutation so duplicate replication changes the result. /// Calls must be separated by a block advance to produce distinct signatures. async fn increment(engine: &TestEngine, state: Pubkey) { @@ -123,6 +137,17 @@ async fn await_replication( expected: BlockstorePosition, state: Pubkey, value: i64, +) { + await_position(positions, expected).await; + follower.sync().await; + assert_eq!(load_v42_data(follower, state), Some(value)); + assert_eq!(follower.superblocks().position(), expected); +} + +/// Waits until a follower publishes exactly the expected durable cursor. +async fn await_position( + positions: &mut broadcast::Receiver, + expected: BlockstorePosition, ) { time::timeout(TIMEOUT, async { loop { @@ -136,9 +161,6 @@ async fn await_replication( }) .await .expect("replication reaches the synced cursor in time"); - follower.sync().await; - assert_eq!(load_v42_data(follower, state), Some(value)); - assert_eq!(follower.superblocks().position(), expected); } /// Loads the serialized transactions committed in `slot`, preserving ledger order. @@ -258,8 +280,8 @@ async fn replays_large_transactions_during_catch_up_and_live_streaming() { .await; assert_eq!(follower.get_account(account), leader.get_account(account)); + close_follower(follower, &mut leader).await; dispatcher.terminate().await; - follower.close().await; leader.close().await; } @@ -323,8 +345,8 @@ async fn batches_transactions_without_crossing_block_boundaries() { leader.ledger().transactions() ); + close_follower(follower, &mut leader).await; dispatcher.terminate().await; - follower.close().await; leader.close().await; } @@ -470,8 +492,8 @@ async fn streams_and_resumes_without_duplicate_application() { "reset replenishes the follower sponsor" ); + close_follower(follower, &mut leader).await; second_dispatcher.terminate().await; - follower.close().await; leader.close().await; } @@ -520,8 +542,54 @@ async fn resumes_after_leader_restart() { let expected = commit_increment(&mut leader, state).await; await_replication(&mut positions, &follower, expected, state, 3).await; + close_follower(follower, &mut leader).await; + second_dispatcher.terminate().await; + leader.close().await; +} + +/// Proves shutdown reconnects, drains to a durable boundary, and reopens from it. +#[tokio::test(flavor = "multi_thread")] +async fn graceful_shutdown_drains_to_next_block_and_reopens() { + let (mut leader, mut follower) = engines(&[], &[]).await; + let addr = loopback_addr(); + let follower_identity = follower.signer().pubkey(); + let mut first_dispatcher = dispatcher(addr, &leader, &[follower_identity]).await; + let mut positions = stream(addr, &mut follower); + leader.advance(1).await; + let expected = leader.sync().await; + await_position(&mut positions, expected).await; + + let mut shutdown = Box::pin(follower.shutdown().terminate()); + assert!( + time::timeout(Duration::from_millis(100), shutdown.as_mut()).await.is_err(), + "shutdown waits while no replicated block boundary exists" + ); + + first_dispatcher.terminate().await; + let mut second_dispatcher = dispatcher(addr, &leader, &[follower_identity]).await; + leader.advance(1).await; + let expected = leader.sync().await; + // The next operational heartbeat lets Ingest observe the closed handoff + // without out-of-band socket interruption. + leader.advance(1).await; + leader.sync().await; + time::timeout(TIMEOUT, shutdown.as_mut()) + .await + .expect("shutdown completes after the next leader block"); + drop(shutdown); + assert_eq!(follower.superblocks().position(), expected); + + let (dirs, authority) = follower.close().await; + let mut follower = TestEngine::with(dirs, authority).await; + assert_eq!(follower.superblocks().position(), expected); + + let mut positions = stream(addr, &mut follower); + leader.advance(1).await; + let expected = leader.sync().await; + await_position(&mut positions, expected).await; + + close_follower(follower, &mut leader).await; second_dispatcher.terminate().await; - follower.close().await; leader.close().await; } @@ -568,8 +636,8 @@ async fn restores_the_newest_snapshot_then_streams_its_tail() { let mut positions = stream(addr, &mut follower); await_replication(&mut positions, &follower, expected, state, 30).await; + close_follower(follower, &mut leader).await; dispatcher.terminate().await; - follower.close().await; leader.close().await; } @@ -651,9 +719,9 @@ async fn cascades_replication_through_a_follower() { assert_eq!(middle.superblocks().sealed(), leader.superblocks().sealed()); assert_eq!(tail.superblocks().sealed(), leader.superblocks().sealed()); - leader_dispatcher.terminate().await; + close_follower(tail, &mut leader).await; + close_follower(middle, &mut leader).await; middle_dispatcher.terminate().await; - tail.close().await; - middle.close().await; + leader_dispatcher.terminate().await; leader.close().await; }