frame-state 0.2.0

Content-addressed state layer — haematite integration, entities, branching, cross-component references
Documentation
#![allow(clippy::expect_used, clippy::unwrap_used)]

mod cross_component;
mod cross_component_reinstall;
mod lifecycle;
mod policy;

use std::error::Error;
use std::io::{BufRead, BufReader, Write};
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::atomic::{AtomicU64, Ordering};

use frame_core::component::ComponentId;

use crate::EntityId;
use crate::store::ComponentStateStore;
use crate::types::{ComponentSchema, MetaRecord, MetaState, ReconcileAction};

pub(super) type TestResult = Result<(), Box<dyn Error>>;

static TEST_DIRECTORY_SEQUENCE: AtomicU64 = AtomicU64::new(0);

pub(super) struct TestDirectory(PathBuf);

impl TestDirectory {
    pub(super) fn new(name: &str) -> Result<Self, std::io::Error> {
        let sequence = TEST_DIRECTORY_SEQUENCE.fetch_add(1, Ordering::Relaxed);
        let path = std::env::temp_dir().join(format!(
            "frame-state-{name}-{}-{sequence}",
            std::process::id()
        ));
        match std::fs::remove_dir_all(&path) {
            Ok(()) => {}
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => return Err(error),
        }
        std::fs::create_dir_all(&path)?;
        Ok(Self(path))
    }

    pub(super) fn path(&self) -> &Path {
        &self.0
    }
}

impl Drop for TestDirectory {
    fn drop(&mut self) {
        if let Err(error) = std::fs::remove_dir_all(&self.0) {
            eprintln!(
                "failed to remove test directory {}: {error}",
                self.0.display()
            );
        }
    }
}

pub(super) fn component(name: &str) -> ComponentId {
    ComponentId::derive("frame-state-tests", name)
}

pub(super) fn install(
    store: &ComponentStateStore,
    component: ComponentId,
) -> Result<crate::ComponentStoreHandle, crate::StateError> {
    store.declare_schema(component, ComponentSchema::lww())?;
    store.install_storage(component)
}

#[test]
fn kill_9_reconciles_every_transition_window() -> TestResult {
    if let Ok(case) = std::env::var("FRAME_STATE_CRASH_CASE") {
        let path = PathBuf::from(std::env::var("FRAME_STATE_CRASH_PATH")?);
        stage_crash_case(&case, &path)?;
        println!("READY {case}");
        std::io::stdout().flush()?;
        loop {
            std::thread::park();
        }
    }

    for case in [
        "after-meta-installing",
        "after-branch-create",
        "after-meta-archiving",
        "after-pin-removal",
    ] {
        let directory = TestDirectory::new(case)?;
        run_and_kill_child(case, directory.path())?;
        let store = ComponentStateStore::open(directory.path())?;
        let report = store.reconcile()?;
        let id = component("crash-target");
        match case {
            "after-meta-installing" => assert_eq!(
                report.actions,
                vec![ReconcileAction::InstallBranchCreated { component: id }]
            ),
            "after-branch-create" => {
                assert!(report.actions.is_empty());
                let resumed = store.install_storage(id)?;
                assert_eq!(resumed.incarnation(), 0);
            }
            "after-meta-archiving" => assert_eq!(
                report.actions,
                vec![ReconcileAction::OrphanedByCrash {
                    component: id,
                    generation: 0,
                }]
            ),
            "after-pin-removal" => assert_eq!(
                report.actions,
                vec![ReconcileAction::ArchiveFinalized {
                    component: id,
                    generation: 0,
                }]
            ),
            _ => unreachable!(),
        }
        if case.contains("archiv") || case.contains("pin") {
            assert_eq!(
                store.get_archived(id, 0, EntityId::of(b"survives-kill-nine"))?,
                Some(b"survives-kill-nine".to_vec())
            );
        } else {
            let _active_root = store.current_root(id)?;
            assert!(store.list_archives(id)?.is_empty());
        }
    }
    Ok(())
}

fn stage_crash_case(case: &str, path: &Path) -> TestResult {
    let store = ComponentStateStore::open(path)?;
    let id = component("crash-target");
    let schema = ComponentSchema::lww();
    match case {
        "after-meta-installing" | "after-branch-create" => {
            let record = MetaRecord {
                component: id,
                incarnation: 0,
                schema,
                state: MetaState::Installing,
            };
            let mut storage = store.lock_storage()?;
            storage.write_meta(&record)?;
            if case == "after-branch-create" {
                storage.create_namespace(id)?;
            }
        }
        "after-meta-archiving" | "after-pin-removal" => {
            let handle = install(&store, id)?;
            handle.put(b"survives-kill-nine")?;
            handle.commit()?;
            let mut storage = store.lock_storage()?;
            let mut record = storage.meta_record(id)?.expect("installed meta record");
            let abandoned_work = storage.outstanding_work(id)?;
            record.state = MetaState::Archiving {
                generation: 0,
                abandoned_work: abandoned_work.clone(),
            };
            storage.write_meta(&record)?;
            if case == "after-pin-removal" {
                storage.pin_remove_archive(id, 0, &abandoned_work)?;
            }
        }
        _ => return Err(format!("unknown crash case {case}").into()),
    }
    Ok(())
}

fn run_and_kill_child(case: &str, path: &Path) -> TestResult {
    let executable = std::env::current_exe()?;
    let mut child = Command::new(executable)
        .args([
            "--exact",
            "tests::kill_9_reconciles_every_transition_window",
            "--nocapture",
        ])
        .env("FRAME_STATE_CRASH_CASE", case)
        .env("FRAME_STATE_CRASH_PATH", path)
        .stdout(Stdio::piped())
        .spawn()?;
    let stdout = child.stdout.take().expect("child stdout was piped");
    let mut ready = false;
    for line in BufReader::new(stdout).lines() {
        if line?.contains(&format!("READY {case}")) {
            ready = true;
            break;
        }
    }
    assert!(ready, "crash child exited before durable READY marker");
    child.kill()?;
    let status = child.wait()?;
    assert!(!status.success(), "kill -9 child unexpectedly succeeded");
    Ok(())
}