crates/monetize-cloud/src/grow.rs
//! **Growing a data set — the vendor-neutral half of the OUTSIDE actor.**
//!
//! `private-gunnar-ops/DATA-SET-GROWTH-FLOW.md`, torture test T6: the one
//! crash that matters — volumes detached, the appliance halted at boot on a
//! missing member — kills the only thing on the appliance that could repair
//! it. So the cloud side of a growth (detach → resize → re-attach, per volume)
//! is owned by an actor OUTSIDE the box, on the front, with a journal that
//! survives its own crash and a `resume` that can always put the volumes back
//! so the box boots. That actor is monetize's cloud provider; this module is
//! the seam it is written against, so a bench on KVM (QMP) could implement it
//! tomorrow without a single UpCloud word in the contract.
//!
//! # Not a purchase
//!
//! A growth is an ENLARGEMENT of resources that already exist and were never
//! bought through monetize (the appliance's four volumes carry no
//! `monetize_ref`). It is priced as the DELTA — `price(target) − price(current)`
//! per member — reserved against the spend ceiling BEFORE volume 1 is touched
//! (T7) and committed to the ledger for what actually grew.
//!
//! # The contract every implementation keeps
//!
//! 1. **Preflight refuses by name and touches nothing**: every member present
//! and healthy, sizes consistent (all at the old size, or some already at
//! the target from an earlier, interrupted run — never a third size), the
//! account able to pay, the ceiling admitting the delta.
//! 2. **Per volume, per step, journaled BEFORE the call and confirmed AFTER.**
//! The [`Journal`] is durable (append, fsync) and vendor-neutral; the
//! provider is the truth on resume, never the journal alone.
//! 3. **Grow only, never shrink, never delete.** A resize that would shrink is
//! refused; a volume that is already at the target is left alone.
//! 4. **Stop at the first refusal that leaves a volume attached** (a hot-plug
//! the server will not do, a size the provider will not take) and say so;
//! leave what is attached attached.
//! 5. **`Change::Settle` from ANY journal state brings every
//! member to an attached state** — grown or not — and reports: `Done` when
//! all are at the target, `Partial` when sizes are unequal (the appliance's
//! own G1 then refuses to grow and serves the old size), `Untouched` when
//! nothing changed. It is run at monetize-server start and on every poll
//! tick while a journal is open.
//! 6. **Idempotent on the target**: a second `grow_set` to a size every member
//! already has is `Done` with a cost of zero and no call that writes.
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use crate::{CloudError, Price, ServerId};
/// A provider-side block volume that is a MEMBER of a product's data set. The
/// provider's own id (an UpCloud storage UUID, a QMP drive id).
#[derive(Clone, PartialEq, Eq, Hash, Debug, serde::Serialize, serde::Deserialize)]
pub struct VolumeId(pub String);
/// **One data set to grow**: the server it is attached to, its members, and
/// the per-member size wanted. Members are enlarged to ONE equal size — the
/// appliance's engine assembles the largest equal group and refuses a set
/// whose members disagree.
#[derive(Clone, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
pub struct DataSet {
/// The product whose set this is (`gunnar`); the ledger row and the
/// journal name it.
pub product: String,
pub server: ServerId,
pub members: Vec<VolumeId>,
/// Per member, GiB. The same number for every member.
pub target_gib: u64,
}
impl DataSet {
/// `grow/<product>/<server>` — the id a ledger row carries for what was
/// grown, in the `ResourceId::id` slot; stable across runs so the spend
/// fold attributes every growth of one set to one line.
pub fn resource_key(&self) -> String {
format!("grow/{}/{}", self.product, self.server.0)
}
}
/// One member as the provider reports it right now.
#[derive(Clone, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
pub struct VolumeFact {
pub id: VolumeId,
pub gib: u64,
/// The provider's own word (`online`, `maintenance`, …).
pub state: String,
/// Attached to the set's server.
pub attached: bool,
/// The device address it is (or was last) attached at, when the provider
/// says (`virtio:2`); what a re-attach must reuse.
pub address: Option<String>,
}
/// The set as surveyed from the provider — what the console draws and what
/// preflight and resume decide from.
#[derive(Clone, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
pub struct SetSurvey {
pub volumes: Vec<VolumeFact>,
/// The server's own state word (`started`, `stopped`).
pub server_state: String,
/// The smallest member, GiB — the size the set is at.
pub current_gib: u64,
/// Every member is the same size.
pub equal: bool,
/// Every member is attached.
pub all_attached: bool,
}
impl SetSurvey {
pub fn at_or_above(&self, gib: u64) -> usize {
self.volumes.iter().filter(|v| v.gib >= gib).count()
}
pub fn detached(&self) -> Vec<VolumeId> {
self.volumes.iter().filter(|v| !v.attached).map(|v| v.id.clone()).collect()
}
}
/// **What a growth, or a resume of one, left behind.** Every arm has every
/// member ATTACHED; an implementation that cannot get there answers `Err`
/// with the journal still open, and the next resume tries again.
#[derive(Clone, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
pub enum GrowOutcome {
/// Every member is at the target. `grown` is what THIS call enlarged
/// (empty on the idempotent repeat) and `cost` is the monthly delta for
/// exactly those — zero when nothing was touched.
Done { grown: Vec<VolumeId>, from_gib: u64, to_gib: u64, cost: Price },
/// Stopped with UNEQUAL members — some at the target, some not — and every
/// one attached. The appliance refuses to grow into this and serves the old
/// size; an operator finishes it with another `grow_set` to the same
/// target once `why` is dealt with. `cost` is the delta for what did grow.
Partial { grown: Vec<VolumeId>, not_grown: Vec<VolumeId>, cost: Price, why: String },
/// Nothing was resized; every member is attached at the size it had. The
/// provider refused something before the first resize (a hot-plug the
/// server will not do, a preflight fact that changed under us).
Untouched { why: String },
}
impl GrowOutcome {
pub fn cost(&self) -> Option<Price> {
match self {
GrowOutcome::Done { cost, .. } | GrowOutcome::Partial { cost, .. } => Some(*cost),
GrowOutcome::Untouched { .. } => None,
}
}
/// `done` | `partial` | `untouched` — one writer for the wire and the journal.
pub fn word(&self) -> &'static str {
match self {
GrowOutcome::Done { .. } => "done",
GrowOutcome::Partial { .. } => "partial",
GrowOutcome::Untouched { .. } => "untouched",
}
}
pub fn grown(&self) -> &[VolumeId] {
match self {
GrowOutcome::Done { grown, .. } | GrowOutcome::Partial { grown, .. } => grown,
GrowOutcome::Untouched { .. } => &[],
}
}
}
// ── the journal ──────────────────────────────────────────────────────────────
/// The step words. Constants rather than an enum so a provider can add a
/// vendor-specific step without changing the seam, and so the file stays
/// readable by a person: `at=… epoch=… volume=… step=detach mark=intent`.
pub mod step {
/// The set-wide check before anything is touched.
pub const PREFLIGHT: &str = "preflight";
pub const DETACH: &str = "detach";
pub const RESIZE: &str = "resize";
/// Waiting for the provider to report the volume usable again.
pub const WAIT_ONLINE: &str = "wait-online";
pub const ATTACH: &str = "attach";
/// Size read back from the provider after the attach.
pub const VERIFY: &str = "verify";
/// The server was stopped (`make_fresh` only — a re-image; no growth stops a server).
pub const STOP: &str = "stop";
/// The server was started again (`make_fresh` only).
pub const START: &str = "start";
/// **The installer medium was attached to the box** (`make_fresh`). Its own
/// word and not [`ATTACH`], because an attach record is read as a data-set
/// MEMBER going back where a detach took it from — `Journal::address_of`
/// folds exactly those — and the medium is not a member and was never
/// detached from anything.
pub const MEDIUM: &str = "medium";
/// **A freshening began** (`make_fresh`), and is confirmed when the box has
/// been stopped and started again. Its own word and not [`PREFLIGHT`],
/// because `Journal::target_of` reads a preflight intent's `gib` as the
/// open epoch's growth target, and a freshening has no target: a
/// `grow_set`/`resume_grow` over this journal would read that 0 and refuse
/// "a target of 0 GiB".
pub const FRESH: &str = "fresh";
/// A resume pass began (after a crash, or a poll tick with the epoch open).
pub const RESUME: &str = "resume";
/// The epoch is closed; `detail` carries the outcome word.
pub const CLOSED: &str = "closed";
}
/// Whether a record is the INTENT written before a call, the CONFIRMATION
/// written after the provider answered, or the provider's refusal.
#[derive(Clone, Copy, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum Mark {
Intent,
Confirmed,
Failed,
}
impl Mark {
pub fn word(self) -> &'static str {
match self {
Mark::Intent => "intent",
Mark::Confirmed => "confirmed",
Mark::Failed => "failed",
}
}
}
/// One journal line.
#[derive(Clone, PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
pub struct GrowRecord {
pub at_unix_ms: u64,
/// The growth this record belongs to: the unix ms the epoch opened at.
pub epoch: u64,
/// The member, or empty for a set-wide record.
#[serde(default)]
pub volume: String,
pub step: String,
pub mark: Mark,
/// The size asked for or read back, GiB, when the step is about one.
#[serde(default)]
pub gib: u64,
/// The device address (`virtio:2`) a detach recorded and an attach reuses.
#[serde(default)]
pub address: String,
#[serde(default)]
pub detail: String,
/// **The measured wall time of the step this CONFIRMS**, ms, taken by the
/// provider around its own calls: on `detach` confirmed the detach, on
/// `wait-online` confirmed resize-until-online (from the resize intent),
/// on `attach` confirmed the attach. 0 = not measured (an older journal,
/// an intent); [`step_timings`] then falls back to the timestamps.
#[serde(default, skip_serializing_if = "is_zero")]
pub elapsed_ms: u64,
}
fn is_zero(v: &u64) -> bool {
*v == 0
}
/// **The durable, append-only journal of one data set's growths**, one JSON
/// object per line, `fsync` after each. Vendor-neutral: the provider writes
/// step words from [`step`], the console reads them, and a resume reads the
/// last detach's `address` to put a volume back where it was.
///
/// An EPOCH is open from its first record until a [`step::CLOSED`] record with
/// the same epoch. `monetize-server` runs `resume_grow` at start and on every
/// tick while one is open — that is the crash story, and the reason the file
/// is written before the provider is asked and not after.
/// **Where an append goes before it counts.** A file (the host's own journal), or a
/// SINK that makes the record durable elsewhere and answers only once it is — the
/// cloud guest's journal, whose sink is the host's journal door
/// (`monetize-cloud-guest` `JOURNAL_KEY`): the guest has no file and no fsync, the
/// host has both, and the step is not taken until the host said the line is down.
pub type JournalSink = Box<dyn Fn(&GrowRecord) -> Result<(), CloudError> + Send + Sync>;
pub struct Journal {
path: PathBuf,
records: Mutex<Vec<GrowRecord>>,
sink: Option<JournalSink>,
}
impl std::fmt::Debug for Journal {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Journal").field("path", &self.path).finish_non_exhaustive()
}
}
impl Journal {
/// Open (or create) the journal at `path`, reading every line it holds. A
/// line that will not parse is refused rather than skipped: a journal a
/// resume cannot read completely is one it cannot act on safely.
///
/// **Except a torn TAIL** — bytes after the last newline — which is cut
/// off the file, loudly, and the journal opens (a-journal-svans,
/// 2026-09-28). [`Journal::append`] writes a record and its newline in one
/// write and returns only after the fsync; every provider calls only after
/// the append returned. So a record without its newline is one whose
/// append never returned: the intent of a call that was never made, or
/// the confirmation of one the provider answers for again on resume. It
/// is cut, whether or not it happens to parse, so the next append starts
/// on a clean line. A bad line a newline FOLLOWS was written whole and is
/// refused as before, and nothing is cut off a refused journal.
pub fn open(path: impl Into<PathBuf>) -> Result<Journal, CloudError> {
let path = path.into();
let mut records = Vec::new();
match std::fs::read(&path) {
Ok(bytes) => {
let whole = bytes.iter().rposition(|b| *b == b'\n').map_or(0, |i| i + 1);
let torn = &bytes[whole..];
let text = std::str::from_utf8(&bytes[..whole]).map_err(|e| CloudError::Refused(format!("grow journal {}: {e}; refusing to act on a journal that cannot be read whole", path.display())))?;
for (n, line) in text.lines().enumerate() {
if line.trim().is_empty() {
continue;
}
let rec: GrowRecord = serde_json::from_str(line).map_err(|e| {
CloudError::Refused(format!("grow journal {} line {}: unreadable ({e}); refusing to act on a journal that cannot be read whole", path.display(), n + 1))
})?;
records.push(rec);
}
if !torn.is_empty() {
let cut = || -> std::io::Result<()> {
let f = std::fs::OpenOptions::new().write(true).open(&path)?;
f.set_len(whole as u64)?;
f.sync_all()
};
cut().map_err(|e| CloudError::Refused(format!("grow journal {}: a torn last line ({} bytes) could not be cut off: {e}", path.display(), torn.len())))?;
eprintln!(
"monetize-cloud: grow journal {}: CUT a torn last line ({} bytes, no newline: an append that never returned, so no call was made on it): {:?}",
path.display(),
torn.len(),
String::from_utf8_lossy(torn)
);
}
}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => return Err(CloudError::Refused(format!("grow journal {}: {e}", path.display()))),
}
Ok(Journal { path, records: Mutex::new(records), sink: None })
}
/// **A journal whose durability is another's**: `records` is what the durable
/// copy holds right now, and every [`Journal::append`] is handed to `sink`
/// FIRST — the in-memory copy grows only after the sink said the record is
/// down, exactly the file journal's rule. [`Journal::path`] is empty.
pub fn with_sink(records: Vec<GrowRecord>, sink: JournalSink) -> Journal {
Journal { path: PathBuf::new(), records: Mutex::new(records), sink: Some(sink) }
}
/// **Read the file again**: what another writer (the cloud guest, through the
/// host's journal door) appended to the same file is now in this copy too. A
/// sink journal has no file and keeps its records.
pub fn reload(&self) -> Result<(), CloudError> {
if self.sink.is_some() {
return Ok(());
}
let fresh = Journal::open(self.path.clone())?;
let got = fresh.records.into_inner().unwrap_or_else(|p| p.into_inner());
*self.records.lock().unwrap_or_else(|p| p.into_inner()) = got;
Ok(())
}
pub fn path(&self) -> &Path {
&self.path
}
/// Append one record and `fsync` before returning. The in-memory copy is
/// updated only after the bytes are on disk, so a reader never sees a
/// record the disk does not hold.
pub fn append(&self, record: GrowRecord) -> Result<(), CloudError> {
let mut records = self.records.lock().unwrap_or_else(|e| e.into_inner());
if let Some(sink) = &self.sink {
sink(&record)?;
records.push(record);
return Ok(());
}
if let Some(parent) = self.path.parent().filter(|p| !p.as_os_str().is_empty()) {
std::fs::create_dir_all(parent).map_err(|e| CloudError::Refused(format!("grow journal dir: {e}")))?;
}
let mut line = serde_json::to_string(&record).map_err(|e| CloudError::Refused(format!("grow journal encode: {e}")))?;
line.push('\n');
let mut file = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&self.path)
.map_err(|e| CloudError::Refused(format!("grow journal {}: {e}", self.path.display())))?;
file.write_all(line.as_bytes()).map_err(|e| CloudError::Refused(format!("grow journal write: {e}")))?;
file.sync_all().map_err(|e| CloudError::Refused(format!("grow journal fsync: {e}")))?;
records.push(record);
Ok(())
}
/// Every record, oldest first.
pub fn records(&self) -> Vec<GrowRecord> {
self.records.lock().unwrap_or_else(|e| e.into_inner()).clone()
}
/// The epoch that is open — has records and no `closed` record — if any.
pub fn open_epoch(&self) -> Option<u64> {
let records = self.records.lock().unwrap_or_else(|e| e.into_inner());
let last = records.last()?;
let epoch = last.epoch;
let closed = records.iter().any(|r| r.epoch == epoch && r.step == step::CLOSED);
(!closed).then_some(epoch)
}
/// The records of `epoch`, oldest first.
pub fn epoch_records(&self, epoch: u64) -> Vec<GrowRecord> {
self.records.lock().unwrap_or_else(|e| e.into_inner()).iter().filter(|r| r.epoch == epoch).cloned().collect()
}
/// The newest epoch in the file, open or closed.
pub fn last_epoch(&self) -> Option<u64> {
self.records.lock().unwrap_or_else(|e| e.into_inner()).last().map(|r| r.epoch)
}
/// **Where each member of `epoch` last was, by the journal**: the address
/// its detach recorded, if any — what a re-attach must reuse.
pub fn address_of(&self, epoch: u64, volume: &VolumeId) -> Option<String> {
self.records
.lock()
.unwrap_or_else(|e| e.into_inner())
.iter().rfind(|r| r.epoch == epoch && r.volume == volume.0 && !r.address.is_empty())
.map(|r| r.address.clone())
}
/// The target the open (or last) epoch was opened for: the `gib` on its
/// `preflight` intent.
pub fn target_of(&self, epoch: u64) -> Option<u64> {
self.records.lock().unwrap_or_else(|e| e.into_inner()).iter().find(|r| r.epoch == epoch && r.step == step::PREFLIGHT && r.mark == Mark::Intent).map(|r| r.gib)
}
}
/// A little helper for the records every provider writes the same way.
impl GrowRecord {
pub fn set_wide(at_unix_ms: u64, epoch: u64, step: &str, mark: Mark, detail: impl Into<String>) -> GrowRecord {
GrowRecord { at_unix_ms, epoch, volume: String::new(), step: step.into(), mark, gib: 0, address: String::new(), detail: detail.into(), elapsed_ms: 0 }
}
pub fn volume(at_unix_ms: u64, epoch: u64, volume: &VolumeId, step: &str, mark: Mark) -> GrowRecord {
GrowRecord { at_unix_ms, epoch, volume: volume.0.clone(), step: step.into(), mark, gib: 0, address: String::new(), detail: String::new(), elapsed_ms: 0 }
}
pub fn with_gib(mut self, gib: u64) -> GrowRecord {
self.gib = gib;
self
}
pub fn with_address(mut self, address: impl Into<String>) -> GrowRecord {
self.address = address.into();
self
}
/// The measured wall time of the step this record confirms.
pub fn with_elapsed_ms(mut self, ms: u64) -> GrowRecord {
self.elapsed_ms = ms;
self
}
pub fn with_detail(mut self, detail: impl Into<String>) -> GrowRecord {
self.detail = detail.into();
self
}
}
/// **The measured wall time of one member's cloud steps**, folded from the
/// journal's own timestamps (intent written before the call, confirmation
/// after): `detach` = detach intent → confirmed, `resize` = resize intent →
/// wait-online confirmed (the provider's resize is not over until the volume
/// is `online` again — 37 s measured on UpCloud 2026-09-14), `attach` = the
/// last attach intent → confirmed. A step that has not completed is 0.
#[derive(Clone, PartialEq, Eq, Debug, Default, serde::Serialize, serde::Deserialize)]
#[serde(default)]
pub struct StepTiming {
pub volume: String,
pub detach_ms: u64,
pub resize_ms: u64,
pub attach_ms: u64,
}
impl StepTiming {
pub fn total_ms(&self) -> u64 {
self.detach_ms + self.resize_ms + self.attach_ms
}
}
/// [`StepTiming`] per member of one epoch's records, in the order the members
/// first appear. Vendor-neutral: only the step words and `at_unix_ms`.
pub fn step_timings(records: &[GrowRecord]) -> Vec<StepTiming> {
let mut out: Vec<StepTiming> = Vec::new();
let at = |volume: &str, step_word: &str, mark: Mark, last: bool| {
let mut it = records.iter().filter(|r| r.volume == volume && r.step == step_word && r.mark == mark).map(|r| r.at_unix_ms);
if last {
it.next_back()
} else {
it.next()
}
};
for r in records {
if r.volume.is_empty() || out.iter().any(|t| t.volume == r.volume) {
continue;
}
let v = r.volume.as_str();
let span = |from: Option<u64>, to: Option<u64>| match (from, to) {
(Some(a), Some(b)) if b >= a => b - a,
_ => 0,
};
// The provider's own measurement wins; the timestamps are the fallback.
let measured = |step_word: &str| records.iter().filter(|r| r.volume == v && r.step == step_word && r.mark == Mark::Confirmed && r.elapsed_ms > 0).map(|r| r.elapsed_ms).next_back();
let resize_end = at(v, step::WAIT_ONLINE, Mark::Confirmed, true).or_else(|| at(v, step::RESIZE, Mark::Confirmed, true));
out.push(StepTiming {
volume: v.to_owned(),
detach_ms: measured(step::DETACH).unwrap_or_else(|| span(at(v, step::DETACH, Mark::Intent, false), at(v, step::DETACH, Mark::Confirmed, true))),
resize_ms: measured(step::WAIT_ONLINE).unwrap_or_else(|| span(at(v, step::RESIZE, Mark::Intent, false), resize_end)),
attach_ms: measured(step::ATTACH).unwrap_or_else(|| span(at(v, step::ATTACH, Mark::Intent, true), at(v, step::ATTACH, Mark::Confirmed, true))),
});
}
out
}
/// The refusal every default answers with: this provider has no growth path.
pub fn unsupported(provider: &str) -> CloudError {
CloudError::Refused(format!(
"the {provider} cloud cannot grow a data set: it has no `grow_set` implementation, so nothing here can enlarge a member volume. Nothing was touched"
))
}
#[cfg(test)]
mod tests {
use super::*;
fn tmp(name: &str) -> PathBuf {
let dir = std::env::temp_dir().join(format!("monetize-grow-journal-{}-{name}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
dir.join("grow.journal")
}
/// **A sink journal takes a step only once the sink holds it**, and a file
/// journal RELOADS what another writer appended to its file — the two halves
/// of the cloud guest's journal door.
#[test]
fn a_sink_journal_appends_through_its_sink_first_and_a_file_journal_reloads() {
let path = tmp("sink");
let host = std::sync::Arc::new(Journal::open(&path).unwrap());
let h2 = std::sync::Arc::clone(&host);
let first = GrowRecord::set_wide(1, 7, step::PREFLIGHT, Mark::Intent, "2 member(s)").with_gib(20);
let guest = Journal::with_sink(vec![], Box::new(move |r: &GrowRecord| h2.append(r.clone())));
guest.append(first.clone()).unwrap();
assert_eq!(guest.records(), vec![first.clone()]);
assert_eq!(guest.open_epoch(), Some(7));
assert_eq!(guest.path(), Path::new(""));
// The caller's own copy of the same file sees it after a reload.
let caller = Journal::open(&path).unwrap();
let second = GrowRecord::set_wide(2, 7, step::CLOSED, Mark::Confirmed, "done");
guest.append(second.clone()).unwrap();
assert_eq!(caller.records(), vec![first.clone()]);
caller.reload().unwrap();
assert_eq!(caller.records(), vec![first.clone(), second]);
assert_eq!(caller.open_epoch(), None);
// A sink that refuses: the step is not taken.
let refusing = Journal::with_sink(vec![first.clone()], Box::new(|_: &GrowRecord| Err(CloudError::Unavailable("the host is gone".into()))));
assert!(refusing.append(GrowRecord::set_wide(3, 8, step::PREFLIGHT, Mark::Intent, "x")).is_err());
assert_eq!(refusing.records(), vec![first]);
}
#[test]
fn a_journal_is_durable_and_knows_which_epoch_is_open() {
let path = tmp("open");
let j = Journal::open(&path).unwrap();
assert_eq!(j.open_epoch(), None);
let v = VolumeId("vol-1".into());
j.append(GrowRecord::set_wide(1, 100, step::PREFLIGHT, Mark::Intent, "4 × 32 → 64").with_gib(64)).unwrap();
j.append(GrowRecord::volume(2, 100, &v, step::DETACH, Mark::Intent).with_address("virtio:2")).unwrap();
assert_eq!(j.open_epoch(), Some(100));
assert_eq!(j.target_of(100), Some(64));
// A fresh process reads the same file and sees the same open epoch and address.
let again = Journal::open(&path).unwrap();
assert_eq!(again.open_epoch(), Some(100));
assert_eq!(again.address_of(100, &v).as_deref(), Some("virtio:2"));
again.append(GrowRecord::set_wide(3, 100, step::CLOSED, Mark::Confirmed, "done")).unwrap();
assert_eq!(again.open_epoch(), None);
assert_eq!(Journal::open(&path).unwrap().last_epoch(), Some(100));
assert_eq!(again.epoch_records(100).len(), 3);
}
#[test]
fn step_timings_fold_each_members_detach_resize_until_online_and_attach() {
let v1 = VolumeId("vol-1".into());
let v2 = VolumeId("vol-2".into());
let rec = |at, v: &VolumeId, s: &str, m| GrowRecord::volume(at, 9, v, s, m);
let records = vec![
GrowRecord::set_wide(0, 9, step::PREFLIGHT, Mark::Intent, "x"),
rec(1_000, &v1, step::DETACH, Mark::Intent),
rec(1_400, &v1, step::DETACH, Mark::Confirmed),
rec(1_500, &v1, step::RESIZE, Mark::Intent),
rec(1_700, &v1, step::RESIZE, Mark::Confirmed),
rec(1_700, &v1, step::WAIT_ONLINE, Mark::Intent),
rec(38_500, &v1, step::WAIT_ONLINE, Mark::Confirmed),
rec(38_600, &v1, step::ATTACH, Mark::Intent),
rec(40_600, &v1, step::ATTACH, Mark::Confirmed),
rec(41_000, &v2, step::DETACH, Mark::Intent),
];
let t = step_timings(&records);
assert_eq!(t.len(), 2);
assert_eq!((t[0].volume.as_str(), t[0].detach_ms, t[0].resize_ms, t[0].attach_ms, t[0].total_ms()), ("vol-1", 400, 37_000, 2_000, 39_400));
assert_eq!((t[1].detach_ms, t[1].resize_ms, t[1].attach_ms), (0, 0, 0), "an unfinished step is 0, never a guess");
// The provider's own measurement, when the journal carries it, wins over the timestamps.
let mut measured = records.clone();
measured.push(rec(42_000, &v2, step::DETACH, Mark::Confirmed).with_elapsed_ms(812));
assert_eq!(step_timings(&measured)[1].detach_ms, 812);
let line = serde_json::to_string(&rec(1, &v1, step::DETACH, Mark::Intent)).unwrap();
assert!(!line.contains("elapsed_ms"), "an unmeasured record keeps the old line shape: {line}");
}
/// A bad line that a newline FOLLOWS is not a torn append — `append`
/// writes the line and its newline in one write, so a line that ended was
/// written whole — and records after it may be what a resume must act on.
/// That journal is refused whole, as it always was. (This test used a torn
/// LAST line until a-journal-svans, 2026-09-28: that shape is now cut off
/// and opened, see the test below, so the refusal is pinned where it still
/// holds — in the middle, and on a finished last line.)
#[test]
fn refuse_twin_a_torn_line_refuses_the_whole_journal() {
let path = tmp("torn");
std::fs::create_dir_all(path.parent().unwrap()).unwrap();
let good = "{\"at_unix_ms\":1,\"epoch\":5,\"step\":\"preflight\",\"mark\":\"intent\"}\n";
std::fs::write(&path, format!("{good}{{\"at_unix_ms\":2,\"epo\n{good}")).unwrap();
let err = Journal::open(&path).unwrap_err().to_string();
assert!(err.contains("line 2"), "{err}");
std::fs::write(&path, format!("{good}{{\"at_unix_ms\":2,\"epo\n")).unwrap();
let err = Journal::open(&path).unwrap_err().to_string();
assert!(err.contains("line 2"), "a finished line that will not parse is refused, not cut: {err}");
assert_eq!(std::fs::read_to_string(&path).unwrap(), format!("{good}{{\"at_unix_ms\":2,\"epo\n"), "and nothing is cut off a refused journal");
}
/// **a-journal-svans: a torn LAST line is an append that never returned,
/// and it is cut off — the journal opens and resumes.** `append` writes the
/// line and its newline in one write and returns only after the fsync, and
/// every provider makes its call only after `append` returned. So bytes
/// after the last newline are a record nobody acted on: the intent of a
/// call that was never made, or the confirmation of one the provider
/// answers for again on resume (the provider is the truth, never the
/// journal alone). Refusing the whole journal for them left the product's
/// growth, and with it the server, unable to start at all.
#[test]
fn a_torn_final_line_is_cut_off_and_the_journal_opens_and_resumes() {
let path = tmp("torn-tail");
std::fs::create_dir_all(path.parent().unwrap()).unwrap();
let first = "{\"at_unix_ms\":1,\"epoch\":5,\"step\":\"preflight\",\"mark\":\"intent\",\"gib\":64}\n";
// Torn inside a multi-byte character, as a detail with `→` can be.
let mut bytes = first.as_bytes().to_vec();
bytes.extend_from_slice("{\"at_unix_ms\":2,\"epoch\":5,\"step\":\"detach\",\"detail\":\"32 →".as_bytes());
bytes.truncate(bytes.len() - 1);
std::fs::write(&path, &bytes).unwrap();
let j = Journal::open(&path).expect("a torn tail is not a reason to refuse the journal");
assert_eq!(j.records().len(), 1);
assert_eq!((j.open_epoch(), j.target_of(5)), (Some(5), Some(64)), "the epoch is still open, so the resume runs");
assert_eq!(std::fs::read(&path).unwrap(), first.as_bytes(), "the torn bytes are cut off the file");
// The resume appends on a clean line, and the next open reads it whole.
j.append(GrowRecord::set_wide(3, 5, step::RESUME, Mark::Intent, "after the crash")).unwrap();
j.append(GrowRecord::set_wide(4, 5, step::CLOSED, Mark::Confirmed, "done")).unwrap();
let again = Journal::open(&path).unwrap();
assert_eq!(again.records().len(), 3);
assert_eq!(again.open_epoch(), None);
}
#[test]
fn outcomes_name_themselves_and_carry_a_cost_only_when_something_grew() {
let p = Price { minor_per_month: 5, currency: monetize_pay::Currency::Eur };
assert_eq!(GrowOutcome::Done { grown: vec![], from_gib: 32, to_gib: 64, cost: p }.word(), "done");
assert_eq!(GrowOutcome::Partial { grown: vec![], not_grown: vec![], cost: p, why: "x".into() }.word(), "partial");
let u = GrowOutcome::Untouched { why: "hotplug".into() };
assert_eq!(u.word(), "untouched");
assert_eq!(u.cost(), None);
let set = DataSet { product: "gunnar".into(), server: ServerId("srv".into()), members: vec![], target_gib: 64 };
assert_eq!(set.resource_key(), "grow/gunnar/srv");
}
#[test]
fn the_freshening_step_words_are_not_the_growth_ones() {
// `medium` is not `attach`: `address_of` folds attach records into
// "where a detached MEMBER goes back", and the medium is neither.
let j_dir = tmp("fresh-steps");
let j = Journal::open(&j_dir).unwrap();
let medium = VolumeId("installer".into());
j.append(GrowRecord::set_wide(1, 7, step::FRESH, Mark::Intent, "attach installer, stop, start")).unwrap();
j.append(GrowRecord::volume(2, 7, &medium, step::MEDIUM, Mark::Confirmed).with_address("virtio:9")).unwrap();
// `fresh` is not `preflight`, so a `resume_grow` over this
// journal reads no target of 0 GiB and refuses nothing on its account.
assert_eq!(j.target_of(7), None);
assert_eq!(j.open_epoch(), Some(7));
j.append(GrowRecord::set_wide(3, 7, step::CLOSED, Mark::Confirmed, "fresh")).unwrap();
assert_eq!(j.open_epoch(), None);
}
/// R5a: a drive that panicked while it held the records must not make every
/// later read of the journal panic. A record joins the copy only after the
/// disk holds it, so the copy a panic leaves is the disk's.
#[test]
fn a_poisoned_journal_still_reads_and_appends() {
let path = tmp("poisoned");
let j = Journal::open(&path).unwrap();
j.append(GrowRecord::set_wide(1, 7, step::PREFLIGHT, Mark::Intent, "1 member(s)").with_gib(20)).unwrap();
let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let _held = j.records.lock();
panic!("a panic while the records are held");
}));
assert!(j.records.is_poisoned());
assert_eq!(j.records().len(), 1);
assert_eq!(j.open_epoch(), Some(7));
assert_eq!(j.epoch_records(7).len(), 1);
assert_eq!(j.last_epoch(), Some(7));
assert_eq!(j.address_of(7, &VolumeId("v".into())), None);
assert_eq!(j.target_of(7), Some(20));
j.append(GrowRecord::set_wide(2, 7, step::CLOSED, Mark::Confirmed, "done")).expect("a record after the panic");
assert_eq!(j.open_epoch(), None);
}
}