use std::collections::BTreeMap;
use std::fs::{File, OpenOptions};
use std::os::unix::io::AsRawFd;
use std::path::{Path, PathBuf};
use serde::{Deserialize, Serialize};
use super::config::expand_path;
use super::error::{PodError, Result};
pub const JOURNAL_DIR_ENV: &str = "FORGE_POD_JOURNAL_DIR";
pub const SCHEMA_VERSION: u32 = 1;
const JOURNAL_FILE: &str = "pod-adoption.json";
const LOCK_FILE: &str = "pod-adoption.lock";
pub fn journal_dir() -> PathBuf {
match std::env::var_os(JOURNAL_DIR_ENV) {
Some(dir) if !dir.is_empty() => PathBuf::from(dir),
_ => expand_path("~/.local/state/forge"),
}
}
pub fn journal_path() -> PathBuf {
journal_dir().join(JOURNAL_FILE)
}
pub fn lock_path() -> PathBuf {
journal_dir().join(LOCK_FILE)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum TransitionState {
Adopting,
Adopted,
Unadopting,
}
impl TransitionState {
pub fn as_str(&self) -> &'static str {
match self {
Self::Adopting => "adopting",
Self::Adopted => "adopted",
Self::Unadopting => "unadopting",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum StepState {
Pending,
Complete,
}
impl StepState {
pub fn is_complete(&self) -> bool {
matches!(self, Self::Complete)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Steps {
pub shim: StepState,
#[serde(default)]
pub hooks: BTreeMap<String, StepState>,
#[serde(default)]
pub recoveries: BTreeMap<String, StepState>,
}
impl Default for Steps {
fn default() -> Self {
Self {
shim: StepState::Pending,
hooks: BTreeMap::new(),
recoveries: BTreeMap::new(),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Journal {
pub schema: u32,
pub state: TransitionState,
pub steps: Steps,
pub ts: String,
pub by: String,
}
impl Journal {
pub fn new(state: TransitionState) -> Self {
Self {
schema: SCHEMA_VERSION,
state,
steps: Steps::default(),
ts: now_iso8601(),
by: format!("forge@{}", env!("CARGO_PKG_VERSION")),
}
}
pub fn touch(&mut self) {
self.ts = now_iso8601();
self.by = format!("forge@{}", env!("CARGO_PKG_VERSION"));
}
pub fn pending_recoveries(&self) -> Vec<&String> {
self.steps
.recoveries
.iter()
.filter(|(_, s)| !s.is_complete())
.map(|(id, _)| id)
.collect()
}
pub fn all_steps_complete(&self) -> bool {
self.steps.shim.is_complete()
&& self.steps.hooks.values().all(StepState::is_complete)
&& self.steps.recoveries.values().all(StepState::is_complete)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Authority {
Unadopted,
InTransition(TransitionState),
Authorized,
Recovery(String),
}
impl Authority {
pub fn permits_production_write(&self) -> bool {
matches!(self, Self::Authorized)
}
pub fn refusal(&self) -> String {
match self {
Self::Authorized => String::new(),
Self::Unadopted => "not adopted — run `forge pod adopt` (cosmux remains the sole \
production writer until then)"
.into(),
Self::InTransition(state) => format!(
"adoption transition in progress (state: {}) — production writes are frozen until \
it completes; run `forge pod adopt --resume` or `--abort`",
state.as_str()
),
Self::Recovery(why) => format!(
"adoption journal is in recovery: {why}\nAll forge production writes are frozen. \
Run `forge pod adopt --repair` (deterministic rollback to unadopted), then re-run \
`forge pod adopt` if adoption was the intent."
),
}
}
}
pub fn verdict(raw: Option<std::result::Result<&str, String>>) -> Authority {
let text = match raw {
None => return Authority::Unadopted,
Some(Err(error)) => return Authority::Recovery(format!("unreadable: {error}")),
Some(Ok(text)) => text,
};
let journal: Journal = match serde_json::from_str(text) {
Ok(journal) => journal,
Err(error) => {
return Authority::Recovery(format!("malformed journal: {error}"));
}
};
if journal.schema != SCHEMA_VERSION {
return Authority::Recovery(format!(
"schema {} is not the supported schema {SCHEMA_VERSION}",
journal.schema
));
}
match journal.state {
TransitionState::Adopting | TransitionState::Unadopting => {
Authority::InTransition(journal.state)
}
TransitionState::Adopted if journal.all_steps_complete() => Authority::Authorized,
TransitionState::Adopted => Authority::Recovery(
"state is `adopted` but the step ledger still has pending entries".into(),
),
}
}
pub fn authority() -> Authority {
let path = journal_path();
match std::fs::read_to_string(&path) {
Ok(text) => verdict(Some(Ok(&text))),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Authority::Unadopted,
Err(error) => verdict(Some(Err(error.to_string()))),
}
}
pub fn require_production_write() -> Result<()> {
let verdict = authority();
if verdict.permits_production_write() {
return Ok(());
}
Err(PodError::Other(anyhow::anyhow!(
"refusing pod-store write: {}",
verdict.refusal()
)))
}
#[cfg(test)]
fn assert_write_allowed(path: &Path) {
let temp = std::env::temp_dir();
assert!(
path.starts_with(&temp),
"REFUSING to write the adoption journal outside the temp dir during tests: {}\n\
Set {JOURNAL_DIR_ENV} to a temp path in this test.",
path.display()
);
}
#[cfg(not(test))]
fn assert_write_allowed(_path: &Path) {}
pub struct JournalLock {
file: File,
dir: PathBuf,
path: PathBuf,
}
impl JournalLock {
pub fn path(&self) -> &Path {
&self.path
}
pub fn read_raw(&self) -> Result<Option<String>> {
match std::fs::read_to_string(&self.path) {
Ok(text) => Ok(Some(text)),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(error) => Err(error.into()),
}
}
pub fn read(&self) -> Result<Option<Journal>> {
match self.read_raw()? {
None => Ok(None),
Some(text) => serde_json::from_str(&text).map(Some).map_err(|e| {
PodError::Other(anyhow::anyhow!("adoption journal is malformed: {e}"))
}),
}
}
pub fn write(&self, journal: &Journal) -> Result<()> {
assert_write_allowed(&self.path);
let raw = serde_json::to_string_pretty(journal)
.map_err(|e| PodError::Other(anyhow::anyhow!("journal serialize: {e}")))?;
let temp = self.dir.join(format!(
".pod-adoption.json.tmp.{}.{}",
std::process::id(),
now_nanos()
));
{
use std::io::Write as _;
let mut handle = File::create(&temp)?;
handle.write_all(raw.as_bytes())?;
handle.write_all(b"\n")?;
handle.sync_all()?;
}
std::fs::rename(&temp, &self.path)?;
Ok(())
}
pub fn remove(&self) -> Result<()> {
assert_write_allowed(&self.path);
match std::fs::remove_file(&self.path) {
Ok(()) => Ok(()),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(error) => Err(error.into()),
}
}
}
impl Drop for JournalLock {
fn drop(&mut self) {
unsafe { libc::flock(self.file.as_raw_fd(), libc::LOCK_UN) };
}
}
fn open_lock_file() -> Result<(File, PathBuf, PathBuf)> {
let dir = journal_dir();
let lock = dir.join(LOCK_FILE);
assert_write_allowed(&lock);
std::fs::create_dir_all(&dir)?;
let file = OpenOptions::new()
.create(true)
.read(true)
.write(true)
.truncate(false)
.open(&lock)?;
Ok((file, dir, journal_path()))
}
pub fn with_lock<T>(f: impl FnOnce(&JournalLock) -> Result<T>) -> Result<T> {
let (file, dir, path) = open_lock_file()?;
if unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX) } != 0 {
return Err(std::io::Error::last_os_error().into());
}
let guard = JournalLock { file, dir, path };
f(&guard)
}
pub fn try_with_lock<T>(f: impl FnOnce(&JournalLock) -> Result<T>) -> Result<Option<T>> {
let (file, dir, path) = open_lock_file()?;
if unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) } != 0 {
let error = std::io::Error::last_os_error();
return match error.raw_os_error() {
Some(code) if code == libc::EWOULDBLOCK || code == libc::EINTR => Ok(None),
_ => Err(error.into()),
};
}
let guard = JournalLock { file, dir, path };
f(&guard).map(Some)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AppendOutcome {
Appended,
AlreadyPresent(StepState),
NotAdopting,
}
pub fn append_recovery(pane_id: &str) -> Result<AppendOutcome> {
with_lock(|lock| {
let Some(mut journal) = lock.read()? else {
return Ok(AppendOutcome::NotAdopting);
};
if journal.state != TransitionState::Adopting {
return Ok(AppendOutcome::NotAdopting);
}
match journal.steps.recoveries.get(pane_id) {
Some(StepState::Pending) => {
return Ok(AppendOutcome::AlreadyPresent(StepState::Pending));
}
Some(StepState::Complete) => {
journal
.steps
.recoveries
.insert(pane_id.to_string(), StepState::Pending);
journal.touch();
lock.write(&journal)?;
return Ok(AppendOutcome::Appended);
}
None => {}
}
journal
.steps
.recoveries
.insert(pane_id.to_string(), StepState::Pending);
journal.touch();
lock.write(&journal)?;
Ok(AppendOutcome::Appended)
})
}
pub fn complete_recovery(pane_id: &str) -> Result<()> {
with_lock(|lock| {
let Some(mut journal) = lock.read()? else {
return Ok(());
};
if journal.steps.recoveries.contains_key(pane_id) {
journal
.steps
.recoveries
.insert(pane_id.to_string(), StepState::Complete);
journal.touch();
lock.write(&journal)?;
}
Ok(())
})
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CommitOutcome {
Committed,
PendingRecoveries(Vec<String>),
IncompleteSteps,
NotAdopting,
}
pub fn commit_terminal() -> Result<CommitOutcome> {
with_lock(|lock| {
let Some(mut journal) = lock.read()? else {
return Ok(CommitOutcome::NotAdopting);
};
if journal.state != TransitionState::Adopting {
return Ok(CommitOutcome::NotAdopting);
}
let pending: Vec<String> = journal
.pending_recoveries()
.into_iter()
.cloned()
.collect::<Vec<_>>();
if !pending.is_empty() {
return Ok(CommitOutcome::PendingRecoveries(pending));
}
if !journal.all_steps_complete() {
return Ok(CommitOutcome::IncompleteSteps);
}
journal.state = TransitionState::Adopted;
journal.touch();
lock.write(&journal)?;
Ok(CommitOutcome::Committed)
})
}
fn now_iso8601() -> String {
chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string()
}
fn now_nanos() -> u128 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::pod::ENV_LOCK;
struct JournalGuard {
dir: PathBuf,
previous: Option<std::ffi::OsString>,
_lock: std::sync::MutexGuard<'static, ()>,
}
impl JournalGuard {
fn new(tag: &str) -> Self {
let lock = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
let dir = std::env::temp_dir().join(format!(
"forge-journal-{tag}-{}-{:?}",
std::process::id(),
std::thread::current().id()
));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).expect("temp journal dir");
let previous = std::env::var_os(JOURNAL_DIR_ENV);
unsafe { std::env::set_var(JOURNAL_DIR_ENV, &dir) };
Self {
dir,
previous,
_lock: lock,
}
}
fn seed(&self, journal: &Journal) {
let raw = serde_json::to_string_pretty(journal).unwrap();
std::fs::write(self.dir.join(JOURNAL_FILE), raw).unwrap();
}
fn seed_raw(&self, raw: &str) {
std::fs::write(self.dir.join(JOURNAL_FILE), raw).unwrap();
}
}
impl Drop for JournalGuard {
fn drop(&mut self) {
match &self.previous {
Some(value) => unsafe { std::env::set_var(JOURNAL_DIR_ENV, value) },
None => unsafe { std::env::remove_var(JOURNAL_DIR_ENV) },
}
let _ = std::fs::remove_dir_all(&self.dir);
}
}
fn adopted() -> Journal {
let mut journal = Journal::new(TransitionState::Adopted);
journal.steps.shim = StepState::Complete;
journal
}
fn adopting_past_shim() -> Journal {
let mut journal = Journal::new(TransitionState::Adopting);
journal.steps.shim = StepState::Complete;
journal
}
#[test]
fn absent_journal_is_unadopted_not_authorized() {
assert_eq!(verdict(None), Authority::Unadopted);
assert!(!verdict(None).permits_production_write());
}
#[test]
fn unreadable_journal_freezes_rather_than_falling_back_to_unadopted() {
let v = verdict(Some(Err("permission denied".into())));
assert!(matches!(v, Authority::Recovery(_)), "{v:?}");
assert!(!v.permits_production_write());
}
#[test]
fn truncated_and_corrupt_journals_freeze() {
for raw in ["", "{", "{\"schema\":1,\"state\":\"ado", "null", "[]"] {
let v = verdict(Some(Ok(raw)));
assert!(matches!(v, Authority::Recovery(_)), "{raw:?} → {v:?}");
}
}
#[test]
fn unknown_state_string_freezes() {
let raw = r#"{"schema":1,"state":"halfway","steps":{"shim":"complete"},"ts":"t","by":"x"}"#;
assert!(matches!(verdict(Some(Ok(raw))), Authority::Recovery(_)));
}
#[test]
fn wrong_schema_freezes() {
let raw = r#"{"schema":2,"state":"adopted","steps":{"shim":"complete"},"ts":"t","by":"x"}"#;
let v = verdict(Some(Ok(raw)));
assert!(matches!(v, Authority::Recovery(_)), "{v:?}");
}
#[test]
fn transitional_states_do_not_authorize_writes() {
for state in [TransitionState::Adopting, TransitionState::Unadopting] {
let raw = serde_json::to_string(&Journal::new(state)).unwrap();
let v = verdict(Some(Ok(&raw)));
assert_eq!(v, Authority::InTransition(state));
assert!(!v.permits_production_write());
}
}
#[test]
fn adopted_with_a_pending_step_is_inconsistent_and_freezes() {
let mut journal = adopted();
journal
.steps
.hooks
.insert("sess".into(), StepState::Pending);
let raw = serde_json::to_string(&journal).unwrap();
assert!(matches!(verdict(Some(Ok(&raw))), Authority::Recovery(_)));
let mut journal = adopted();
journal
.steps
.recoveries
.insert("%1".into(), StepState::Pending);
let raw = serde_json::to_string(&journal).unwrap();
assert!(matches!(verdict(Some(Ok(&raw))), Authority::Recovery(_)));
}
#[test]
fn only_terminal_adopted_with_every_step_complete_authorizes() {
let mut journal = adopted();
journal
.steps
.hooks
.insert("sess".into(), StepState::Complete);
journal
.steps
.recoveries
.insert("%1".into(), StepState::Complete);
let raw = serde_json::to_string(&journal).unwrap();
assert_eq!(verdict(Some(Ok(&raw))), Authority::Authorized);
assert!(verdict(Some(Ok(&raw))).permits_production_write());
}
#[test]
fn every_refusal_names_a_remedy() {
for v in [
Authority::Unadopted,
Authority::InTransition(TransitionState::Adopting),
Authority::Recovery("truncated".into()),
] {
let message = v.refusal();
assert!(!message.is_empty(), "{v:?}");
assert!(message.contains("forge pod"), "{v:?} → {message}");
}
}
#[test]
fn authority_reads_the_journal_at_the_configured_path() {
let guard = JournalGuard::new("authority");
assert_eq!(authority(), Authority::Unadopted);
assert!(require_production_write().is_err());
guard.seed(&adopted());
assert_eq!(authority(), Authority::Authorized);
assert!(require_production_write().is_ok());
guard.seed_raw("{ truncated");
assert!(matches!(authority(), Authority::Recovery(_)));
assert!(
require_production_write().is_err(),
"a corrupt journal must not authorize a pod-store write"
);
}
#[test]
fn the_lock_is_a_sibling_file_never_the_journal() {
let _guard = JournalGuard::new("lockpath");
assert_ne!(lock_path(), journal_path());
assert_eq!(lock_path().file_name().unwrap(), LOCK_FILE);
}
#[test]
fn a_second_acquisition_is_excluded_while_the_first_is_held() {
let _guard = JournalGuard::new("exclude");
with_lock(|_held| {
let attempted = try_with_lock(|_| Ok(()))?;
assert_eq!(attempted, None, "the lock must exclude a second holder");
Ok(())
})
.unwrap();
assert_eq!(try_with_lock(|_| Ok(())).unwrap(), Some(()));
}
#[test]
fn writes_are_atomic_and_leave_no_temp_files_behind() {
let guard = JournalGuard::new("atomic");
with_lock(|lock| lock.write(&Journal::new(TransitionState::Adopting))).unwrap();
let strays: Vec<_> = std::fs::read_dir(&guard.dir)
.unwrap()
.filter_map(|e| e.ok())
.map(|e| e.file_name().to_string_lossy().to_string())
.filter(|name| name.contains(".tmp."))
.collect();
assert!(strays.is_empty(), "temp files left behind: {strays:?}");
assert_eq!(
authority(),
Authority::InTransition(TransitionState::Adopting)
);
}
#[test]
fn read_distinguishes_absent_from_malformed() {
let guard = JournalGuard::new("absent");
with_lock(|lock| {
assert!(lock.read()?.is_none(), "absent journal reads as None");
Ok(())
})
.unwrap();
guard.seed_raw("{ nope");
let result = with_lock(|lock| lock.read());
assert!(
result.is_err(),
"a malformed journal must be an error, never None — None means unadopted"
);
}
#[test]
fn remove_leaves_the_lock_file_intact() {
let guard = JournalGuard::new("remove");
guard.seed(&adopted());
with_lock(|lock| lock.remove()).unwrap();
assert!(!journal_path().exists());
assert!(lock_path().exists(), "lock file must survive a journal rm");
assert_eq!(authority(), Authority::Unadopted);
}
fn flock_binary_available() -> bool {
std::process::Command::new("flock")
.arg("--help")
.output()
.map(|out| out.status.success())
.unwrap_or(false)
}
#[test]
fn the_shell_flock_and_the_binary_exclude_each_other() {
if !flock_binary_available() {
eprintln!("SKIP: flock(1) not on PATH — cross-primitive interop unverified");
return;
}
let guard = JournalGuard::new("interop");
let lock = guard.dir.join(LOCK_FILE);
std::fs::write(&lock, b"").unwrap();
let ready = guard.dir.join("holding");
let mut child = std::process::Command::new("flock")
.arg("-x")
.arg(&lock)
.arg("-c")
.arg(format!("touch '{}' && sleep 3", ready.display()))
.spawn()
.expect("spawn flock(1)");
let mut waited = 0;
while !ready.exists() && waited < 3000 {
std::thread::sleep(std::time::Duration::from_millis(10));
waited += 10;
}
assert!(ready.exists(), "flock(1) never took the lock");
assert_eq!(
try_with_lock(|_| Ok(())).unwrap(),
None,
"the binary must block on a lock held by flock(1)"
);
let _ = child.kill();
let _ = child.wait();
let blocked = with_lock(|_held| {
let status = std::process::Command::new("flock")
.args(["-w", "0", "-x"])
.arg(&lock)
.args(["-c", "true"])
.status()
.expect("run flock(1)");
Ok(status.success())
})
.unwrap();
assert!(
!blocked,
"flock(1) must block on a lock held by the binary — the standalone rollback depends on it"
);
let after = std::process::Command::new("flock")
.args(["-w", "0", "-x"])
.arg(&lock)
.args(["-c", "true"])
.status()
.expect("run flock(1)");
assert!(
after.success(),
"flock(1) must acquire once the binary releases"
);
}
#[test]
fn concurrent_recovery_appends_both_survive() {
let guard = JournalGuard::new("append-race");
guard.seed(&Journal::new(TransitionState::Adopting));
std::thread::scope(|scope| {
let handles: Vec<_> = ["%1", "%2", "%3", "%4"]
.into_iter()
.map(|pane| scope.spawn(move || append_recovery(pane).unwrap()))
.collect();
for handle in handles {
assert_eq!(handle.join().unwrap(), AppendOutcome::Appended);
}
});
let journal = with_lock(|lock| lock.read()).unwrap().unwrap();
assert_eq!(
journal.steps.recoveries.len(),
4,
"every pane identifier must remain present: {:?}",
journal.steps.recoveries
);
}
#[test]
fn appending_the_same_pane_twice_is_idempotent() {
let guard = JournalGuard::new("append-idem");
guard.seed(&Journal::new(TransitionState::Adopting));
assert_eq!(append_recovery("%1").unwrap(), AppendOutcome::Appended);
assert_eq!(
append_recovery("%1").unwrap(),
AppendOutcome::AlreadyPresent(StepState::Pending),
"a re-fired hook must not duplicate an entry"
);
let journal = with_lock(|lock| lock.read()).unwrap().unwrap();
assert_eq!(journal.steps.recoveries.len(), 1);
}
#[test]
fn a_second_death_after_recovery_completes_re_pends_and_blocks_the_commit() {
let guard = JournalGuard::new("repend");
guard.seed(&adopting_past_shim());
assert_eq!(append_recovery("%1").unwrap(), AppendOutcome::Appended);
complete_recovery("%1").unwrap();
assert_eq!(commit_terminal().unwrap(), CommitOutcome::Committed);
guard.seed(&{
let mut j = adopting_past_shim();
j.steps.recoveries.insert("%1".into(), StepState::Complete);
j
});
assert_eq!(
append_recovery("%1").unwrap(),
AppendOutcome::Appended,
"a re-death after completion must be a fresh pending occurrence"
);
let journal = with_lock(|lock| lock.read()).unwrap().unwrap();
assert_eq!(
journal.steps.recoveries.get("%1"),
Some(&StepState::Pending)
);
assert_eq!(journal.pending_recoveries(), vec![&"%1".to_string()]);
match commit_terminal().unwrap() {
CommitOutcome::PendingRecoveries(p) => assert_eq!(p, vec!["%1"]),
other => panic!("commit must wait for the re-pended pane, got {other:?}"),
}
}
#[test]
fn appends_outside_adopting_are_refused_not_silently_dropped() {
let guard = JournalGuard::new("append-terminal");
guard.seed(&adopted());
assert_eq!(append_recovery("%1").unwrap(), AppendOutcome::NotAdopting);
let journal = with_lock(|lock| lock.read()).unwrap().unwrap();
assert!(journal.steps.recoveries.is_empty());
}
#[test]
fn terminal_commit_refuses_while_a_recovery_is_pending() {
let guard = JournalGuard::new("commit-pending");
guard.seed(&adopting_past_shim());
append_recovery("%late").unwrap();
match commit_terminal().unwrap() {
CommitOutcome::PendingRecoveries(pending) => assert_eq!(pending, vec!["%late"]),
other => panic!("terminal commit must loop back to drain, got {other:?}"),
}
assert_eq!(
authority(),
Authority::InTransition(TransitionState::Adopting),
"state must remain adopting — no authority is granted over a pending recovery"
);
complete_recovery("%late").unwrap();
assert_eq!(commit_terminal().unwrap(), CommitOutcome::Committed);
assert_eq!(authority(), Authority::Authorized);
}
#[test]
fn an_append_racing_the_terminal_commit_is_never_lost() {
for attempt in 0..25 {
let guard = JournalGuard::new(&format!("commit-race-{attempt}"));
guard.seed(&adopting_past_shim());
let (append, commit) = std::thread::scope(|scope| {
let appender = scope.spawn(|| append_recovery("%racer").unwrap());
let committer = scope.spawn(|| commit_terminal().unwrap());
(appender.join().unwrap(), committer.join().unwrap())
});
let journal = with_lock(|lock| lock.read()).unwrap().unwrap();
let recorded = journal.steps.recoveries.get("%racer");
match (&append, &commit) {
(AppendOutcome::Appended, CommitOutcome::PendingRecoveries(pending)) => {
assert_eq!(pending, &vec!["%racer".to_string()], "attempt {attempt}");
assert_eq!(recorded, Some(&StepState::Pending), "attempt {attempt}");
assert_eq!(
journal.state,
TransitionState::Adopting,
"attempt {attempt}"
);
}
(AppendOutcome::NotAdopting, CommitOutcome::Committed) => {
assert_eq!(journal.state, TransitionState::Adopted, "attempt {attempt}");
assert_eq!(
recorded, None,
"attempt {attempt}: a terminal journal must not accrue deferred intents"
);
assert!(authority().permits_production_write(), "attempt {attempt}");
}
other => panic!("attempt {attempt}: illegal interleaving {other:?}"),
}
}
}
#[test]
fn commit_refuses_an_incomplete_non_recovery_step() {
let guard = JournalGuard::new("commit-incomplete");
let mut journal = Journal::new(TransitionState::Adopting);
journal.steps.shim = StepState::Pending;
guard.seed(&journal);
assert_eq!(commit_terminal().unwrap(), CommitOutcome::IncompleteSteps);
assert!(!authority().permits_production_write());
}
#[test]
#[should_panic(expected = "REFUSING to write the adoption journal outside the temp dir")]
fn writing_the_live_journal_panics_under_test() {
assert_write_allowed(Path::new("/home/axw/.local/state/forge/pod-adoption.json"));
}
}