crates/gunnar-repair/src/pump.rs
//! **The replay pump and the recharge verb** — the ONE copy (5.4 resilience: it
//! was five, in the CLI, the admin console, the wasm guest, the R1 drill and
//! monetize's product). Transport-neutral: the host hands in a [`Twin`] and a
//! [`Main`] over whatever it speaks (tonic, the guest's host calls, a test fake).
//!
//! The logic is the guest's `replay`/`replay_base` (restore.rs, the most tested
//! one: budget pauses, resumed bases, large entries, a twin older than the lane),
//! moved here unchanged in substance.
#![allow(async_fn_in_trait)]
use serde::{Deserialize, Serialize};
use crate::plan::{
finished_on_the_box, plan_recharge, refuse, BaseAsk, RechargeOpts, RechargePlan, Refusal,
RefusalCode, TwinBase, KIND_SNAPSHOT_AT,
};
use crate::view::{Chunk, MainStatus, Restored, TwinHead};
/// A transport failure, as the host words it.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Fault {
/// The gRPC code word (`unavailable`, `unimplemented`, …), or the host's own.
pub code: String,
pub message: String,
}
impl Fault {
pub fn new(code: impl Into<String>, message: impl Into<String>) -> Fault {
Fault {
code: code.into(),
message: message.into(),
}
}
}
impl std::fmt::Display for Fault {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}: {}", self.code, self.message)
}
}
/// One server stream from the twin.
pub trait ChunkStream {
type Chunk: Chunk;
/// `None` = the stream ended.
async fn next(&mut self) -> Option<Result<Self::Chunk, Fault>>;
}
/// **The twin**, as the pump reads it: `Mirror.LogHead`, `Mirror.SnapshotFetch`,
/// `Mirror.LogFetch`.
pub trait Twin {
type Chunk: Chunk;
type Stream: ChunkStream<Chunk = Self::Chunk>;
/// Where it answers, for the words.
fn addr(&self) -> String;
async fn head(&self) -> Result<TwinHead, Fault>;
async fn snapshot(
&self,
from_item: u64,
from_offset: u64,
expect_id: &str,
) -> Result<Self::Stream, Fault>;
async fn log(&self, from: u64, to: u64, from_offset: u64) -> Result<Self::Stream, Fault>;
}
/// The main's `Bootstrap.Restore` client stream, opened.
pub trait RestoreSink<C> {
async fn send(&mut self, chunk: C) -> Result<(), Fault>;
/// Close the request half: the box answers.
async fn close(self) -> Result<Restored, Fault>;
/// **Abort**: the box must NOT take this as a clean end (it would finish below the
/// head); the host breaks the stream (a poison frame, a reset).
async fn abort(self, why: &str);
}
/// **The main** (a fresh or restoring box): `Bootstrap.Status`, `Bootstrap.Restore`.
pub trait Main<C> {
type Sink: RestoreSink<C>;
async fn status(&self) -> Result<MainStatus, Fault>;
async fn open_restore(&self) -> Result<Self::Sink, Fault>;
}
/// The phase words, the same as monetize-product's `RestoreStatus.phase`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Phase {
/// The box restarts itself and opens as the primary it was.
Done,
/// This call's budget ended it at a piece boundary; the next call resumes.
Replaying,
Failed,
}
/// **What one `recharge_from_twin` did.** Its JSON is the order's `receipt_json` and
/// the admin console's answer.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Receipt {
pub phase: Phase,
pub twin: String,
pub head_seq: u64,
pub applied: u64,
pub lost: u64,
pub needs_restart: bool,
/// What the plan said would happen.
pub plan: String,
pub freshness: String,
pub detail: String,
/// Why it failed; empty unless `failed`.
pub why: String,
}
/// How a pump ended without a fault.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Pumped {
/// The box answered the close.
Answered {
restored: Restored,
applied: u64,
highest: u64,
},
/// The budget ended it; the box committed what it got and resumes at `seq` (byte `byte`).
Paused { detail: String, applied: u64 },
}
/// **What the twin holds below its `first_seq`**: [`TwinBase::from_head`], or the first
/// `SnapshotFetch` piece when the head does not say (no snapshot = `None`; a piece with
/// no position = `Old`).
pub async fn twin_base<T: Twin>(twin: &T, head: &TwinHead) -> Result<TwinBase, Fault> {
if let Some(b) = TwinBase::from_head(head) {
return Ok(b);
}
let no_base = |f: &Fault| {
f.code.eq_ignore_ascii_case("unimplemented")
|| f.message.to_ascii_lowercase().contains("unimplemented")
|| f.message.contains("keeps no snapshot")
};
let mut out = match twin.snapshot(0, 0, "").await {
Ok(s) => s,
Err(f) if no_base(&f) => return Ok(TwinBase::None),
Err(f) => return Err(f),
};
match out.next().await {
Some(Ok(c)) if c.snapshot_id().trim().is_empty() => Ok(TwinBase::Old),
Some(Ok(_)) => Ok(TwinBase::Resumable),
None => Ok(TwinBase::None),
Some(Err(f)) if no_base(&f) => Ok(TwinBase::None),
Some(Err(f)) => Err(f),
}
}
/// **The twin's base into the restore stream**, from where the box holds it. A base
/// that changed since (`snapshot-changed`) is asked for again from its start, once.
/// `Ok(None)`: the base went whole (terminator included). `Ok(Some((item, byte)))`: the
/// budget ended it there.
async fn pump_base<T: Twin, S: RestoreSink<T::Chunk>, B: Fn() -> bool + ?Sized>(
twin: &T,
ask: &BaseAsk,
plan: &RechargePlan,
into: &mut S,
spent: &B,
) -> Result<Option<(u64, u64)>, String> {
let addr = twin.addr();
let changed = |f: &Fault| f.message.contains("snapshot-changed");
let mut out = match twin
.snapshot(ask.from_item, ask.from_offset, &ask.expect_id)
.await
{
Ok(s) => s,
Err(f) if changed(&f) => twin
.snapshot(0, 0, "")
.await
.map_err(|f| format!("Mirror.SnapshotFetch on the twin at {addr}: {f}"))?,
Err(f) => return Err(format!("Mirror.SnapshotFetch on the twin at {addr}: {f}")),
};
let mut first = true;
let mut retried = false;
loop {
match out.next().await {
Some(Ok(mut chunk)) => {
first = false;
if chunk.snapshot_id().trim().is_empty() {
return Err(format!("the twin at {addr} streams its snapshot without positions — a twin older than this pump: upgrade the twin"));
}
let item = chunk.snapshot_item().unwrap_or(0);
let reached = chunk.offset().unwrap_or(0) + chunk.piece_len();
let last = chunk.last();
let terminator = chunk.kind() == KIND_SNAPSHOT_AT && last;
chunk.stamp(plan.head, plan.announced_bytes, &plan.log_id);
into.send(chunk).await.map_err(|f| {
format!(
"the main stopped taking the twin's snapshot at item {item}: {}",
f.message
)
})?;
if terminator {
return Ok(None);
}
if spent() {
return Ok(Some(if last { (item + 1, 0) } else { (item, reached) }));
}
}
None => {
return Err(format!(
"the twin's snapshot stream ended before its SnapshotAt terminator (from item {})",
ask.from_item
))
}
Some(Err(f)) if first && !retried && changed(&f) => {
retried = true;
out = twin
.snapshot(0, 0, "")
.await
.map_err(|f| format!("Mirror.SnapshotFetch on the twin at {addr}: {f}"))?;
}
Some(Err(f)) => return Err(format!("the twin's snapshot stream broke: {}", f.message)),
}
}
}
/// **The replay itself**: the twin's base (when the plan says so), then its log
/// `from..=head` walked fetch by fetch (one `LogFetch` stops at a byte ceiling), into an
/// opened `Bootstrap.Restore`. `spent()` = this call's budget is over: both streams end
/// at a piece boundary and the box resumes there next call (never true = no budget).
/// `Err` = why it failed; the sink has been aborted.
pub async fn pump<T: Twin, S: RestoreSink<T::Chunk>, B: Fn() -> bool + ?Sized>(
twin: &T,
mut into: S,
plan: &RechargePlan,
spent: &B,
) -> Result<Pumped, String> {
let addr = twin.addr();
let head = plan.head;
let mut applied = 0u64;
let mut highest = plan.from.saturating_sub(1);
let mut cursor = plan.from;
let mut stopped: Option<(u64, u64)> = None;
if let Some(ask) = plan.base.as_ref().filter(|a| !a.whole) {
match pump_base(twin, ask, plan, &mut into, spent).await {
Ok(None) => {}
Ok(Some((item, byte))) => {
let said = into
.close()
.await
.err()
.map(|f| format!(": {}", f.message))
.unwrap_or_default();
return Ok(Pumped::Paused {
detail: format!("this call's budget ended the stream inside the twin's snapshot at item {item} byte {byte}; the next call resumes there{said}"),
applied: 0,
});
}
Err(why) => {
into.abort(&why).await;
return Err(why);
}
}
}
let mut twin_whole = false;
'fetches: while cursor <= head {
let offset = if cursor == plan.from {
plan.from_offset
} else {
0
};
let mut out = match twin.log(cursor, head, offset).await {
Ok(s) => s,
Err(f) => {
let why = format!("Mirror.LogFetch from {cursor} on the twin at {addr}: {f}");
into.abort(&why).await;
return Err(why);
}
};
let before = highest;
let mut first = true;
// The last piece taken in this fetch: (entry, whether it ended the entry).
let mut at: Option<(u64, bool)> = None;
loop {
match out.next().await {
Some(Ok(mut chunk)) => {
if first && offset > 0 && chunk.seq() == cursor && chunk.offset().is_none() {
twin_whole = true;
}
first = false;
let (seq, last) = (chunk.seq(), chunk.last());
// stinky #17: entry after entry, no hole, nothing past the head — checked
// BEFORE the piece reaches the main.
let next = match at {
None => cursor,
Some((open, false)) => open,
Some((done, true)) => done + 1,
};
let hole = match at {
Some((open, false)) if seq != open => Some(format!("the twin's log stream moved to entry {seq} with entry {open} cut short")),
_ if seq != next => Some(format!("the twin's log stream sent entry {seq} where {next} was next: a hole, refused before it reaches the main")),
_ if seq > head => Some(format!("the twin's log stream sent entry {seq}, past the head {head} it reported")),
_ => None,
};
if let Some(why) = hole {
into.abort(&why).await;
return Err(why);
}
at = Some((seq, last));
let reached = chunk.offset().map(|o| o + chunk.piece_len());
// The head on every piece: the box refuses to FINISH below it, so a stream
// that broke on our side leaves it restoring instead of falsely done.
chunk.stamp(head, plan.announced_bytes, &plan.log_id);
if let Err(f) = into.send(chunk).await {
return Err(format!(
"the main stopped taking the restore at sequence {seq}: {}",
f.message
));
}
highest = highest.max(seq);
if last {
applied += 1;
}
if spent() && !(last && seq >= head) {
stopped = Some((
if last { seq + 1 } else { seq },
if last { 0 } else { reached.unwrap_or(0) },
));
break 'fetches;
}
}
None => {
if let Some((open, false)) = at {
let why = format!("the twin's log stream ended inside entry {open}: its tail would never arrive");
into.abort(&why).await;
return Err(why);
}
break;
}
Some(Err(f)) => {
let why = format!(
"the twin's log stream broke at sequence {highest} of {head}: {}",
f.message
);
into.abort(&why).await;
return Err(why);
}
}
}
// No progress and the head not reached: fetching again would spin for ever.
if highest < cursor || highest == before && highest < head {
let why = format!(
"the twin answered no entry at or above {cursor} although it reports head {head}"
);
into.abort(&why).await;
return Err(why);
}
cursor = highest + 1;
}
let closed = into.close().await;
if let Some((seq, byte)) = stopped {
let said = closed
.err()
.map(|f| format!(": {}", f.message))
.unwrap_or_default();
return Ok(Pumped::Paused {
detail: format!(
"this call's budget ended the stream at seq {seq}{}; the next call resumes there{}{said}",
if byte > 0 { format!(" byte {byte}") } else { String::new() },
if twin_whole { " (the twin sent the entry from its first byte: a twin older than resumed restores — upgrade the twin)" } else { "" },
),
applied,
});
}
match closed {
Ok(restored) if highest < head => Err(format!(
"the twin's log ran out at sequence {highest} although it reports head {head} (the box said: {})",
restored.detail
)),
Ok(restored) => Ok(Pumped::Answered { restored, applied, highest }),
Err(f) => Err(format!("Bootstrap.Restore: {}", f.message)),
}
}
/// **`recharge_from_twin`**: refuse by name or recharge — read the main and the twin,
/// [`plan_recharge`], [`pump`], and settle a close whose answer was cut against the
/// box's own word ([`finished_on_the_box`]). Nothing streams before every refusal is
/// passed. The ONE path behind the console's button, the boot check's auto-fix and a
/// pulled monetize order. A live main must be fenced FIRST (`override_live_main`
/// alone is refused here too: a recharge is only ever into a fresh or restoring box).
///
/// `spent` is generic so a `Sync` budget keeps the future `Send` (axum, tokio::spawn)
/// while a single-threaded host's need not be.
pub async fn recharge_from_twin<T: Twin, M: Main<T::Chunk>, B: Fn() -> bool + ?Sized>(
twin: &T,
main: &M,
opts: &RechargeOpts,
spent: &B,
) -> Result<Receipt, Refusal> {
let addr = twin.addr();
let status = main.status().await.map_err(|f| {
refuse(
RefusalCode::Unreachable,
format!("the main's Bootstrap.Status: {f}"),
)
})?;
let head = twin.head().await.map_err(|f| {
refuse(
RefusalCode::Unreachable,
format!("the twin at {addr} (Mirror.LogHead): {f}"),
)
})?;
crate::plan::source_is_a_twin(&addr, &head, opts)?;
let base = twin_base(twin, &head).await.map_err(|f| {
refuse(
RefusalCode::Unreachable,
format!("the twin at {addr} (Mirror.SnapshotFetch): {f}"),
)
})?;
let plan = plan_recharge(&addr, &status, &head, base, opts)?;
let mut receipt = Receipt {
phase: Phase::Replaying,
twin: addr.clone(),
head_seq: plan.head,
applied: 0,
lost: plan.lost,
needs_restart: false,
plan: plan.detail.clone(),
freshness: plan.freshness.clone(),
detail: String::new(),
why: String::new(),
};
let sink = match main.open_restore().await {
Ok(s) => s,
Err(f) => {
receipt.phase = Phase::Failed;
receipt.why = format!("Bootstrap.Restore would not open: {f}");
return Ok(receipt);
}
};
match pump(twin, sink, &plan, spent).await {
Ok(Pumped::Answered {
restored, applied, ..
}) => {
receipt.phase = Phase::Done;
receipt.head_seq = restored.head_seq;
receipt.applied = restored.applied.max(applied);
receipt.lost += restored.lost;
receipt.needs_restart = restored.restart_required;
receipt.detail = restored.detail;
}
Ok(Pumped::Paused { detail, applied }) => {
receipt.applied = applied;
receipt.detail = detail;
}
Err(why) => match main.status().await {
Ok(b) if finished_on_the_box(&b, plan.head) => {
receipt.phase = Phase::Done;
receipt.needs_restart = true;
receipt.detail = format!(
"the box finished the restore: its log is at seq {} (the twin's head {}), not restoring, its window shut — the close was cut ({why}); it restarts itself and opens as the primary it was",
b.log_head_seq, plan.head
);
}
_ => {
receipt.phase = Phase::Failed;
receipt.why = why;
}
},
}
Ok(receipt)
}
/// **Drive a future to completion on this thread**, for a host whose calls are
/// synchronous (the wasm guest): every [`Twin`]/[`Main`] future it hands the pump is
/// ready at its first poll. A future that is not is a host bug, named.
pub fn block_on_ready<F: std::future::Future>(f: F) -> F::Output {
let mut f = std::pin::pin!(f);
let mut cx = std::task::Context::from_waker(std::task::Waker::noop());
for _ in 0..1_000_000 {
if let std::task::Poll::Ready(v) = f.as_mut().poll(&mut cx) {
return v;
}
}
panic!("block_on_ready: a host future was not ready (a synchronous host must answer at once)")
}
#[cfg(test)]
pub(crate) mod fake {
//! An in-memory twin and main, the pump's contract in miniature.
use super::*;
use std::sync::{Arc, Mutex};
#[derive(Debug, Clone, Default)]
pub struct C {
pub seq: u64,
pub kind: u32,
pub last: bool,
pub offset: Option<u64>,
pub blob: Vec<u8>,
pub snapshot_id: String,
pub snapshot_item: Option<u64>,
pub restore_head: u64,
pub log_id: String,
}
impl Chunk for C {
fn seq(&self) -> u64 {
self.seq
}
fn kind(&self) -> u32 {
self.kind
}
fn last(&self) -> bool {
self.last
}
fn offset(&self) -> Option<u64> {
self.offset
}
fn piece_len(&self) -> u64 {
self.blob.len() as u64
}
fn snapshot_id(&self) -> &str {
&self.snapshot_id
}
fn snapshot_item(&self) -> Option<u64> {
self.snapshot_item
}
fn stamp(&mut self, restore_head: u64, _: Option<u64>, log_id: &str) {
self.restore_head = restore_head;
if self.log_id.is_empty() {
self.log_id = log_id.to_owned();
}
}
}
pub struct S(pub std::vec::IntoIter<Result<C, Fault>>);
impl ChunkStream for S {
type Chunk = C;
async fn next(&mut self) -> Option<Result<C, Fault>> {
self.0.next()
}
}
/// A twin holding entries `first..=head`, at most `page` entries per fetch.
pub struct FakeTwin {
pub head: TwinHead,
pub page: usize,
pub snapshot: Vec<C>,
/// Serves its log only up to here (a twin that claims more than it has).
pub serves_to: Option<u64>,
/// Leaves this entry out of its log (a hole inside one fetch).
pub skips: Option<u64>,
/// Sends this entry's only piece without `last` (a fetch that ends inside an entry).
pub cuts: Option<u64>,
}
impl FakeTwin {
pub fn new(first: u64, head: u64) -> FakeTwin {
FakeTwin {
// The twin of a main that died: level at its last answer, a second ago.
head: TwinHead {
first_seq: first,
head_seq: head,
twin: true,
state: "unreachable".into(),
log_id: "L".into(),
applied_seq: head,
primary_head_seq: head,
last_answer_unix_ms: 1,
last_level_unix_ms: 1,
answer_ago_ms: 1_000,
..Default::default()
},
page: 2,
snapshot: Vec::new(),
serves_to: None,
skips: None,
cuts: None,
}
}
}
impl Twin for FakeTwin {
type Chunk = C;
type Stream = S;
fn addr(&self) -> String {
"twin:1".into()
}
async fn head(&self) -> Result<TwinHead, Fault> {
Ok(self.head.clone())
}
async fn snapshot(&self, from_item: u64, _: u64, _: &str) -> Result<S, Fault> {
if self.snapshot.is_empty() {
return Err(Fault::new("not_found", "this twin keeps no snapshot"));
}
Ok(S(self
.snapshot
.iter()
.filter(|c| c.snapshot_item.unwrap_or(0) >= from_item)
.cloned()
.map(Ok)
.collect::<Vec<_>>()
.into_iter()))
}
async fn log(&self, from: u64, to: u64, _: u64) -> Result<S, Fault> {
let v: Vec<_> = (from.max(self.head.first_seq)
..=to.min(self.serves_to.unwrap_or(self.head.head_seq)))
.filter(|seq| Some(*seq) != self.skips)
.take(self.page)
.map(|seq| {
Ok(C {
seq,
kind: 1,
last: Some(seq) != self.cuts,
offset: Some(0),
blob: vec![seq as u8],
..Default::default()
})
})
.collect();
Ok(S(v.into_iter()))
}
}
#[derive(Default)]
pub struct Box_ {
pub status: Mutex<MainStatus>,
pub got: Arc<Mutex<Vec<C>>>,
pub aborted: Arc<Mutex<Option<String>>>,
}
pub struct Sink {
got: Arc<Mutex<Vec<C>>>,
aborted: Arc<Mutex<Option<String>>>,
}
impl RestoreSink<C> for Sink {
async fn send(&mut self, c: C) -> Result<(), Fault> {
self.got.lock().unwrap().push(c);
Ok(())
}
async fn close(self) -> Result<Restored, Fault> {
let got = self.got.lock().unwrap();
let head = got
.iter()
.filter(|c| c.kind != KIND_SNAPSHOT_AT)
.map(|c| c.seq)
.max()
.unwrap_or(0);
Ok(Restored {
head_seq: head,
applied: got
.iter()
.filter(|c| c.last && c.snapshot_id.is_empty())
.count() as u64,
lost: 0,
restart_required: true,
detail: "restored".into(),
})
}
async fn abort(self, why: &str) {
*self.aborted.lock().unwrap() = Some(why.to_owned());
}
}
impl Main<C> for Box_ {
type Sink = Sink;
async fn status(&self) -> Result<MainStatus, Fault> {
Ok(self.status.lock().unwrap().clone())
}
async fn open_restore(&self) -> Result<Sink, Fault> {
Ok(Sink {
got: self.got.clone(),
aborted: self.aborted.clone(),
})
}
}
pub fn fresh() -> MainStatus {
MainStatus {
window_open: true,
restore_snapshot_resumable: true,
..Default::default()
}
}
}
#[cfg(test)]
mod tests {
use super::fake::*;
use super::*;
/// fence-before-promote: these tests are about the STREAM, so the fence is declared.
fn fenced() -> RechargeOpts {
RechargeOpts {
primary_fenced: true,
..Default::default()
}
}
#[test]
fn a_fresh_main_is_recharged_from_the_whole_log_across_fetches() {
let twin = FakeTwin::new(1, 7);
let main = Box_ {
status: std::sync::Mutex::new(fresh()),
..Default::default()
};
let r = block_on_ready(recharge_from_twin(&twin, &main, &fenced(), &|| false)).unwrap();
assert_eq!(r.phase, Phase::Done, "{r:?}");
assert_eq!(r.head_seq, 7);
let got = main.got.lock().unwrap();
assert_eq!(
got.iter().map(|c| c.seq).collect::<Vec<_>>(),
(1..=7).collect::<Vec<_>>(),
"four fetches of two, in order"
);
assert!(
got.iter().all(|c| c.restore_head == 7 && c.log_id == "L"),
"every piece carries the head and the log id"
);
}
#[test]
fn a_live_main_is_refused_and_nothing_streams() {
let twin = FakeTwin::new(1, 7);
let main = Box_ {
status: std::sync::Mutex::new(MainStatus {
principals: 2,
keys: 3,
..Default::default()
}),
..Default::default()
};
let e = block_on_ready(recharge_from_twin(&twin, &main, &fenced(), &|| false)).unwrap_err();
assert_eq!(e.code, RefusalCode::MainLive);
assert!(main.got.lock().unwrap().is_empty());
}
#[test]
fn a_trimmed_twin_without_a_base_is_refused_until_accept_lost() {
let twin = FakeTwin::new(5, 7);
let main = Box_ {
status: std::sync::Mutex::new(fresh()),
..Default::default()
};
let e = block_on_ready(recharge_from_twin(&twin, &main, &fenced(), &|| false)).unwrap_err();
assert_eq!(e.code, RefusalCode::WouldLose, "{e}");
let r = block_on_ready(recharge_from_twin(
&twin,
&main,
&RechargeOpts {
accept_lost: true,
..fenced()
},
&|| false,
))
.unwrap();
assert_eq!((r.phase, r.lost), (Phase::Done, 4));
}
#[test]
fn the_twins_base_streams_first_then_the_log() {
let mut twin = FakeTwin::new(5, 6);
twin.head.snapshot_at = 4;
twin.snapshot = vec![
C {
seq: 4,
kind: 3,
last: true,
offset: Some(0),
blob: vec![1],
snapshot_id: "s1".into(),
snapshot_item: Some(0),
..Default::default()
},
C {
seq: 4,
kind: KIND_SNAPSHOT_AT,
last: true,
offset: Some(0),
snapshot_id: "s1".into(),
snapshot_item: Some(1),
..Default::default()
},
];
let main = Box_ {
status: std::sync::Mutex::new(fresh()),
..Default::default()
};
let r = block_on_ready(recharge_from_twin(&twin, &main, &fenced(), &|| false)).unwrap();
assert_eq!((r.phase, r.lost), (Phase::Done, 0), "{r:?}");
let kinds: Vec<_> = main
.got
.lock()
.unwrap()
.iter()
.map(|c| (c.seq, c.kind))
.collect();
assert_eq!(kinds, vec![(4, 3), (4, KIND_SNAPSHOT_AT), (5, 1), (6, 1)]);
}
#[test]
fn a_spent_budget_pauses_at_a_piece_boundary() {
let twin = FakeTwin::new(1, 7);
let main = Box_ {
status: std::sync::Mutex::new(fresh()),
..Default::default()
};
let n = std::sync::atomic::AtomicU32::new(0);
let spent = || n.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + 1 >= 3;
let r = block_on_ready(recharge_from_twin(&twin, &main, &fenced(), &spent)).unwrap();
assert_eq!(r.phase, Phase::Replaying, "{r:?}");
assert!(
r.detail.contains("resumes there") && r.detail.contains("seq 4"),
"{}",
r.detail
);
}
#[test]
fn a_twin_that_stops_short_fails_and_aborts_the_stream() {
let mut twin = FakeTwin::new(1, 9);
twin.serves_to = Some(7);
twin.head.head_seq = 9;
twin.head.applied_seq = 9;
twin.head.primary_head_seq = 9;
let main = Box_ {
status: std::sync::Mutex::new(fresh()),
..Default::default()
};
let r = block_on_ready(recharge_from_twin(&twin, &main, &fenced(), &|| false)).unwrap();
assert_eq!(r.phase, Phase::Failed);
assert!(r.why.contains("no entry at or above 8"), "{}", r.why);
assert!(
main.aborted.lock().unwrap().is_some(),
"the box is told it was NOT a clean end"
);
}
/// ★ stinky #17, carried into the ONE pump (BROKEN 11:58): a twin whose fetch jumps
/// a sequence number is refused BEFORE the entry past the hole reaches the main.
#[test]
fn a_hole_inside_one_fetch_is_refused_and_nothing_past_it_reaches_the_main() {
let mut twin = FakeTwin::new(1, 7);
twin.page = 3;
twin.skips = Some(3);
let main = Box_ {
status: std::sync::Mutex::new(fresh()),
..Default::default()
};
let r = block_on_ready(recharge_from_twin(&twin, &main, &fenced(), &|| false)).unwrap();
assert_eq!(r.phase, Phase::Failed, "{r:?}");
assert!(r.why.contains("entry 4 where 3 was next"), "{}", r.why);
assert_eq!(
main.got
.lock()
.unwrap()
.iter()
.map(|c| c.seq)
.collect::<Vec<_>>(),
vec![1, 2],
"nothing past the hole"
);
assert!(
main.aborted.lock().unwrap().is_some(),
"the box is told it was NOT a clean end"
);
}
/// A fetch that ends inside an entry would have the next fetch start one past it,
/// and the entry's tail would never arrive: refused by name.
#[test]
fn a_fetch_that_ends_inside_an_entry_is_refused() {
let mut twin = FakeTwin::new(1, 7);
twin.cuts = Some(2);
let main = Box_ {
status: std::sync::Mutex::new(fresh()),
..Default::default()
};
let r = block_on_ready(recharge_from_twin(&twin, &main, &fenced(), &|| false)).unwrap();
assert_eq!(r.phase, Phase::Failed, "{r:?}");
assert!(r.why.contains("ended inside entry 2"), "{}", r.why);
assert!(main.aborted.lock().unwrap().is_some());
}
#[test]
fn a_primary_is_not_a_twin() {
let mut twin = FakeTwin::new(1, 7);
twin.head.twin = false;
let main = Box_ {
status: std::sync::Mutex::new(fresh()),
..Default::default()
};
assert_eq!(
block_on_ready(recharge_from_twin(&twin, &main, &fenced(), &|| false))
.unwrap_err()
.code,
RefusalCode::NotATwin
);
}
}