Add the lock-owned transactional Domain Store

crates/lazaret-host-coordinator/src/domain_store.rs

@@ -0,0 +1,740 @@
+use std::fmt;
+use std::io;
+use std::path::Path;
+
+use lazaret_core::codec::{CodecError, decode, encode};
+use lazaret_core::identity::{GenerationId, SessionId};
+use lazaret_core::records::{
+    Domain, DomainStore, DomainStoreMetadata, DurableDecision, Generation, Mailbox, Participant,
+    ParticipantRole, ParticipantState, RecordTransitionError, Session, SessionPhase,
+    SessionTerminal, SessionTerminalState,
+};
+use lazaret_linux::filesystem::ExclusiveFileLock;
+use redb::{
+    CommitError, Database, DatabaseError, ReadableDatabase, ReadableTable, StorageError,
+    TableDefinition, TableError, TransactionError,
+};
+
+const METADATA: TableDefinition<u8, &[u8]> = TableDefinition::new("metadata");
+const DOMAINS: TableDefinition<u8, &[u8]> = TableDefinition::new("domains");
+const GENERATIONS: TableDefinition<u64, &[u8]> = TableDefinition::new("generations");
+const PARTICIPANTS: TableDefinition<&[u8; 9], &[u8]> = TableDefinition::new("participants");
+const SESSIONS: TableDefinition<&[u8; 40], &[u8]> = TableDefinition::new("sessions");
+const SESSION_TERMINALS: TableDefinition<&[u8; 40], &[u8]> =
+    TableDefinition::new("session_terminals");
+const MAILBOXES: TableDefinition<u64, &[u8]> = TableDefinition::new("mailboxes");
+const SINGLETON_KEY: u8 = 0;
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum DomainStoreMutation {
+    PutGeneration(Generation),
+    PutParticipant(Participant),
+    CleanParticipantAfterAbsence(Participant),
+    PutSession(Session),
+    PutMailbox(Mailbox),
+    CleanMailboxAfterAbsence(Mailbox),
+    PutSessionTerminal(SessionTerminal),
+    CleanSessionTerminalAfterAbsence(SessionTerminal),
+}
+
+#[derive(Debug)]
+pub enum DomainStoreError {
+    Io(io::Error),
+    Database(DatabaseError),
+    Transaction(TransactionError),
+    Table(TableError),
+    Storage(StorageError),
+    Commit(CommitError),
+    Codec(CodecError),
+    Transition(RecordTransitionError),
+    MissingMetadata,
+    MissingRow(&'static str),
+    DuplicateRow(&'static str),
+    InvalidRow(&'static str),
+    DomainMismatch,
+    GenerationMismatch,
+}
+
+impl fmt::Display for DomainStoreError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Io(error) => write!(formatter, "Domain Store I/O failed: {error}"),
+            Self::Database(error) => write!(formatter, "Domain Store open failed: {error}"),
+            Self::Transaction(error) => {
+                write!(formatter, "Domain Store transaction failed: {error}")
+            }
+            Self::Table(error) => write!(formatter, "Domain Store table failed: {error}"),
+            Self::Storage(error) => write!(formatter, "Domain Store storage failed: {error}"),
+            Self::Commit(error) => write!(formatter, "Domain Store commit failed: {error}"),
+            Self::Codec(error) => {
+                write!(formatter, "Domain Store record failed validation: {error}")
+            }
+            Self::Transition(error) => {
+                write!(formatter, "Domain Store transition refused: {error}")
+            }
+            Self::MissingMetadata => formatter.write_str("Domain Store metadata is missing"),
+            Self::MissingRow(kind) => write!(formatter, "referenced {kind} row is missing"),
+            Self::DuplicateRow(kind) => write!(formatter, "duplicate {kind} row"),
+            Self::InvalidRow(kind) => write!(formatter, "invalid {kind} row"),
+            Self::DomainMismatch => formatter.write_str("Domain Store Domain identity mismatch"),
+            Self::GenerationMismatch => formatter.write_str("Domain Store generation mismatch"),
+        }
+    }
+}
+
+impl std::error::Error for DomainStoreError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            Self::Io(error) => Some(error),
+            Self::Database(error) => Some(error),
+            Self::Transaction(error) => Some(error),
+            Self::Table(error) => Some(error),
+            Self::Storage(error) => Some(error),
+            Self::Commit(error) => Some(error),
+            Self::Codec(error) => Some(error),
+            Self::Transition(error) => Some(error),
+            Self::MissingMetadata
+            | Self::MissingRow(_)
+            | Self::DuplicateRow(_)
+            | Self::InvalidRow(_)
+            | Self::DomainMismatch
+            | Self::GenerationMismatch => None,
+        }
+    }
+}
+
+macro_rules! from_error {
+    ($source:ty, $variant:ident) => {
+        impl From<$source> for DomainStoreError {
+            fn from(error: $source) -> Self {
+                Self::$variant(error)
+            }
+        }
+    };
+}
+
+from_error!(io::Error, Io);
+from_error!(DatabaseError, Database);
+from_error!(TransactionError, Transaction);
+from_error!(TableError, Table);
+from_error!(StorageError, Storage);
+from_error!(CommitError, Commit);
+from_error!(CodecError, Codec);
+from_error!(RecordTransitionError, Transition);
+
+pub struct HostCoordinatorDomainStore {
+    database: Database,
+    _domain_lock: ExclusiveFileLock,
+}
+
+impl HostCoordinatorDomainStore {
+    /// Opens the Domain Store while retaining the caller's exclusive Domain lock.
+    ///
+    /// # Errors
+    ///
+    /// Refuses I/O, database, metadata, row, or Domain identity failures.
+    pub fn open(
+        domain_lock: ExclusiveFileLock,
+        database_path: &Path,
+        expected_domain: &Domain,
+    ) -> Result<Self, DomainStoreError> {
+        let is_new = !database_path.try_exists()?;
+        let database = if is_new {
+            Database::create(database_path)?
+        } else {
+            Database::open(database_path)?
+        };
+        let store = Self {
+            database,
+            _domain_lock: domain_lock,
+        };
+        if is_new {
+            store.initialize(expected_domain)?;
+        } else {
+            store.validate_metadata()?;
+            if store.snapshot()?.domain != *expected_domain {
+                return Err(DomainStoreError::DomainMismatch);
+            }
+        }
+        Ok(store)
+    }
+
+    /// Reads and validates every typed row as one consistent redb snapshot.
+    ///
+    /// # Errors
+    ///
+    /// Refuses database, table, codec, key, duplicate, or relationship failures.
+    pub fn snapshot(&self) -> Result<DomainStore, DomainStoreError> {
+        self.validate_metadata()?;
+        let read = self.database.begin_read()?;
+        let domain = {
+            let table = read.open_table(DOMAINS)?;
+            let value = table
+                .get(SINGLETON_KEY)?
+                .ok_or(DomainStoreError::MissingRow("Domain"))?;
+            decode(value.value())?
+        };
+        let generations = read_generations(&read.open_table(GENERATIONS)?)?;
+        let participants = read_participants(&read.open_table(PARTICIPANTS)?)?;
+        let sessions = read_sessions(&read.open_table(SESSIONS)?)?;
+        let session_terminals = read_session_terminals(&read.open_table(SESSION_TERMINALS)?)?;
+        let mailboxes = read_mailboxes(&read.open_table(MAILBOXES)?)?;
+        let snapshot = DomainStore {
+            metadata: DomainStoreMetadata::v1(),
+            domain,
+            generations,
+            participants,
+            sessions,
+            session_terminals,
+            mailboxes,
+        };
+        validate_snapshot(&snapshot)?;
+        Ok(snapshot)
+    }
+
+    /// Applies the complete mutation batch in one immediate-durability redb transaction.
+    ///
+    /// # Errors
+    ///
+    /// Refuses invalid transitions, generation mismatches, malformed relationships, or storage
+    /// failures without committing any member of the batch.
+    pub fn apply(&self, mutations: &[DomainStoreMutation]) -> Result<(), DomainStoreError> {
+        if mutations.is_empty() {
+            return Ok(());
+        }
+        let mut snapshot = self.snapshot()?;
+        for mutation in mutations {
+            apply_mutation(&mut snapshot, mutation)?;
+        }
+        validate_snapshot(&snapshot)?;
+        self.write_snapshot(&snapshot)
+    }
+
+    fn initialize(&self, domain: &Domain) -> Result<(), DomainStoreError> {
+        let snapshot = DomainStore {
+            metadata: DomainStoreMetadata::v1(),
+            domain: domain.clone(),
+            generations: Vec::new(),
+            participants: Vec::new(),
+            sessions: Vec::new(),
+            session_terminals: Vec::new(),
+            mailboxes: Vec::new(),
+        };
+        self.write_snapshot(&snapshot)
+    }
+
+    fn validate_metadata(&self) -> Result<(), DomainStoreError> {
+        let read = self.database.begin_read()?;
+        let table = read.open_table(METADATA).map_err(|error| match error {
+            TableError::TableDoesNotExist(_) => DomainStoreError::MissingMetadata,
+            other => DomainStoreError::Table(other),
+        })?;
+        let value = table
+            .get(SINGLETON_KEY)?
+            .ok_or(DomainStoreError::MissingMetadata)?;
+        let _: DomainStoreMetadata = decode(value.value())?;
+        Ok(())
+    }
+
+    fn write_snapshot(&self, snapshot: &DomainStore) -> Result<(), DomainStoreError> {
+        let write = self.database.begin_write()?;
+        {
+            let mut table = write.open_table(METADATA)?;
+            let encoded = encode(&snapshot.metadata)?;
+            table.insert(SINGLETON_KEY, encoded.as_slice())?;
+        }
+        {
+            let mut table = write.open_table(DOMAINS)?;
+            let encoded = encode(&snapshot.domain)?;
+            table.insert(SINGLETON_KEY, encoded.as_slice())?;
+        }
+        {
+            let mut table = write.open_table(GENERATIONS)?;
+            for record in &snapshot.generations {
+                let encoded = encode(record)?;
+                table.insert(record.generation_id.get(), encoded.as_slice())?;
+            }
+        }
+        {
+            let mut table = write.open_table(PARTICIPANTS)?;
+            for record in &snapshot.participants {
+                let encoded = encode(record)?;
+                table.insert(&participant_key(record), encoded.as_slice())?;
+            }
+        }
+        {
+            let mut table = write.open_table(SESSIONS)?;
+            for record in &snapshot.sessions {
+                let encoded = encode(record)?;
+                table.insert(
+                    &session_key(record.generation_id, record.session_id),
+                    encoded.as_slice(),
+                )?;
+            }
+        }
+        {
+            let mut table = write.open_table(SESSION_TERMINALS)?;
+            for record in &snapshot.session_terminals {
+                let encoded = encode(record)?;
+                table.insert(
+                    &session_key(record.generation_id(), record.session_id()),
+                    encoded.as_slice(),
+                )?;
+            }
+        }
+        {
+            let mut table = write.open_table(MAILBOXES)?;
+            for record in &snapshot.mailboxes {
+                let encoded = encode(record)?;
+                table.insert(record.generation_id.get(), encoded.as_slice())?;
+            }
+        }
+        write.commit()?;
+        Ok(())
+    }
+}
+
+fn read_generations(
+    table: &redb::ReadOnlyTable<u64, &[u8]>,
+) -> Result<Vec<Generation>, DomainStoreError> {
+    let mut records = Vec::new();
+    for entry in table.iter()? {
+        let (key, value) = entry?;
+        let record: Generation = decode(value.value())?;
+        if key.value() != record.generation_id.get() {
+            return Err(DomainStoreError::InvalidRow("Generation key"));
+        }
+        records.push(record);
+    }
+    Ok(records)
+}
+
+fn read_participants(
+    table: &redb::ReadOnlyTable<&[u8; 9], &[u8]>,
+) -> Result<Vec<Participant>, DomainStoreError> {
+    let mut records = Vec::new();
+    for entry in table.iter()? {
+        let (key, value) = entry?;
+        let record: Participant = decode(value.value())?;
+        if key.value() != &participant_key(&record) {
+            return Err(DomainStoreError::InvalidRow("Participant key"));
+        }
+        records.push(record);
+    }
+    Ok(records)
+}
+
+fn read_sessions(
+    table: &redb::ReadOnlyTable<&[u8; 40], &[u8]>,
+) -> Result<Vec<Session>, DomainStoreError> {
+    let mut records = Vec::new();
+    for entry in table.iter()? {
+        let (key, value) = entry?;
+        let record: Session = decode(value.value())?;
+        if key.value() != &session_key(record.generation_id, record.session_id) {
+            return Err(DomainStoreError::InvalidRow("Session key"));
+        }
+        records.push(record);
+    }
+    Ok(records)
+}
+
+fn read_session_terminals(
+    table: &redb::ReadOnlyTable<&[u8; 40], &[u8]>,
+) -> Result<Vec<SessionTerminal>, DomainStoreError> {
+    let mut records = Vec::new();
+    for entry in table.iter()? {
+        let (key, value) = entry?;
+        let record: SessionTerminal = decode(value.value())?;
+        if key.value() != &session_key(record.generation_id(), record.session_id()) {
+            return Err(DomainStoreError::InvalidRow("Session Terminal key"));
+        }
+        records.push(record);
+    }
+    Ok(records)
+}
+
+fn read_mailboxes(
+    table: &redb::ReadOnlyTable<u64, &[u8]>,
+) -> Result<Vec<Mailbox>, DomainStoreError> {
+    let mut records = Vec::new();
+    for entry in table.iter()? {
+        let (key, value) = entry?;
+        let record: Mailbox = decode(value.value())?;
+        if key.value() != record.generation_id.get() {
+            return Err(DomainStoreError::InvalidRow("Mailbox key"));
+        }
+        records.push(record);
+    }
+    Ok(records)
+}
+
+fn apply_mutation(
+    snapshot: &mut DomainStore,
+    mutation: &DomainStoreMutation,
+) -> Result<(), DomainStoreError> {
+    match mutation {
+        DomainStoreMutation::PutGeneration(next) => put_generation(snapshot, next.clone()),
+        DomainStoreMutation::PutParticipant(next) => put_participant(snapshot, next.clone(), false),
+        DomainStoreMutation::CleanParticipantAfterAbsence(next) => {
+            put_participant(snapshot, next.clone(), true)
+        }
+        DomainStoreMutation::PutSession(next) => put_session(snapshot, next.clone()),
+        DomainStoreMutation::PutMailbox(next) => put_mailbox(snapshot, next.clone(), false),
+        DomainStoreMutation::CleanMailboxAfterAbsence(next) => {
+            put_mailbox(snapshot, next.clone(), true)
+        }
+        DomainStoreMutation::PutSessionTerminal(next) => {
+            put_session_terminal(snapshot, next.clone(), false)
+        }
+        DomainStoreMutation::CleanSessionTerminalAfterAbsence(next) => {
+            put_session_terminal(snapshot, next.clone(), true)
+        }
+    }
+}
+
+fn put_generation(snapshot: &mut DomainStore, next: Generation) -> Result<(), DomainStoreError> {
+    if next.domain_id != snapshot.domain.domain_id {
+        return Err(DomainStoreError::DomainMismatch);
+    }
+    if let Some(current) = snapshot
+        .generations
+        .iter_mut()
+        .find(|record| record.generation_id == next.generation_id)
+    {
+        validate_generation_transition(current, &next)?;
+        *current = next;
+    } else {
+        if next.durable_decision != DurableDecision::None {
+            return Err(DomainStoreError::InvalidRow("initial Generation"));
+        }
+        snapshot.generations.push(next);
+    }
+    Ok(())
+}
+
+fn put_participant(
+    snapshot: &mut DomainStore,
+    next: Participant,
+    absence_proven: bool,
+) -> Result<(), DomainStoreError> {
+    require_generation(snapshot, next.generation_id())?;
+    if let Some(current) = snapshot.participants.iter_mut().find(|record| {
+        record.generation_id() == next.generation_id() && record.role() == next.role()
+    }) {
+        if absence_proven {
+            current.validate_transition_to_clean_after_absence(&next)?;
+        } else {
+            current.validate_transition_to(&next)?;
+        }
+        *current = next;
+    } else {
+        if absence_proven || next.state() != ParticipantState::Intent || next.ownership().is_some()
+        {
+            return Err(DomainStoreError::InvalidRow("initial Participant"));
+        }
+        snapshot.participants.push(next);
+    }
+    Ok(())
+}
+
+fn put_session(snapshot: &mut DomainStore, next: Session) -> Result<(), DomainStoreError> {
+    require_generation(snapshot, next.generation_id)?;
+    if let Some(current) = snapshot.sessions.iter_mut().find(|record| {
+        record.generation_id == next.generation_id && record.session_id == next.session_id
+    }) {
+        validate_session_transition(current, &next)?;
+        *current = next;
+    } else {
+        if !matches!(
+            next.session_phase,
+            SessionPhase::New | SessionPhase::Waiting | SessionPhase::Registering
+        ) {
+            return Err(DomainStoreError::InvalidRow("initial Session"));
+        }
+        snapshot.sessions.push(next);
+    }
+    Ok(())
+}
+
+fn put_mailbox(
+    snapshot: &mut DomainStore,
+    next: Mailbox,
+    absence_proven: bool,
+) -> Result<(), DomainStoreError> {
+    require_generation(snapshot, next.generation_id)?;
+    if let Some(current) = snapshot
+        .mailboxes
+        .iter_mut()
+        .find(|record| record.generation_id == next.generation_id)
+    {
+        validate_mailbox_transition(current, &next, absence_proven)?;
+        *current = next;
+    } else {
+        if absence_proven || next.state != ParticipantState::Intent {
+            return Err(DomainStoreError::InvalidRow("initial Mailbox"));
+        }
+        snapshot.mailboxes.push(next);
+    }
+    Ok(())
+}
+
+fn put_session_terminal(
+    snapshot: &mut DomainStore,
+    next: SessionTerminal,
+    absence_proven: bool,
+) -> Result<(), DomainStoreError> {
+    require_session(snapshot, next.generation_id(), next.session_id())?;
+    if let Some(current) = snapshot.session_terminals.iter_mut().find(|record| {
+        record.generation_id() == next.generation_id() && record.session_id() == next.session_id()
+    }) {
+        if absence_proven {
+            current.validate_transition_to_clean_after_absence(&next)?;
+        } else {
+            current.validate_transition_to(&next)?;
+        }
+        *current = next;
+    } else {
+        if absence_proven
+            || next.state() != SessionTerminalState::Intent
+            || next.container_id().is_some()
+            || !next.host_coordinator().is_unassigned()
+            || next.pty_master().is_some()
+            || !next.exec_client().is_unassigned()
+            || next.guest_process().is_some()
+        {
+            return Err(DomainStoreError::InvalidRow("initial Session Terminal"));
+        }
+        snapshot.session_terminals.push(next);
+    }
+    Ok(())
+}
+
+fn validate_generation_transition(
+    current: &Generation,
+    next: &Generation,
+) -> Result<(), DomainStoreError> {
+    if current.domain_id != next.domain_id
+        || current.generation_id != next.generation_id
+        || current.configuration_fingerprint != next.configuration_fingerprint
+        || current.coordinator_id != next.coordinator_id
+        || current.network_sidecar_id != next.network_sidecar_id
+        || current.sandbox_id != next.sandbox_id
+    {
+        return Err(DomainStoreError::InvalidRow("Generation identity"));
+    }
+    let valid = current.durable_decision == next.durable_decision
+        || matches!(
+            (current.durable_decision, next.durable_decision),
+            (
+                DurableDecision::None,
+                DurableDecision::Preparing | DurableDecision::Clean
+            ) | (
+                DurableDecision::Preparing,
+                DurableDecision::Published | DurableDecision::Clean
+            ) | (DurableDecision::Published, DurableDecision::Clean)
+        );
+    if valid {
+        Ok(())
+    } else {
+        Err(DomainStoreError::InvalidRow("Generation transition"))
+    }
+}
+
+fn validate_session_transition(current: &Session, next: &Session) -> Result<(), DomainStoreError> {
+    if current.generation_id != next.generation_id
+        || current.session_id != next.session_id
+        || current.cli_id != next.cli_id
+        || current.session_anchor_id != next.session_anchor_id
+    {
+        return Err(DomainStoreError::InvalidRow("Session identity"));
+    }
+    let valid = current.session_phase == next.session_phase
+        || matches!(
+            (current.session_phase, next.session_phase),
+            (
+                SessionPhase::New,
+                SessionPhase::Waiting | SessionPhase::Registering | SessionPhase::Failed
+            ) | (
+                SessionPhase::Waiting,
+                SessionPhase::Registering | SessionPhase::Closed | SessionPhase::Failed
+            ) | (
+                SessionPhase::Registering | SessionPhase::HostExecuting,
+                SessionPhase::Attached | SessionPhase::Closed | SessionPhase::Failed
+            ) | (
+                SessionPhase::Attached,
+                SessionPhase::Prompting
+                    | SessionPhase::HostExecuting
+                    | SessionPhase::Closed
+                    | SessionPhase::Failed
+            ) | (
+                SessionPhase::Prompting,
+                SessionPhase::Attached
+                    | SessionPhase::HostExecuting
+                    | SessionPhase::Closed
+                    | SessionPhase::Failed
+            )
+        );
+    if valid {
+        Ok(())
+    } else {
+        Err(DomainStoreError::InvalidRow("Session transition"))
+    }
+}
+
+fn validate_mailbox_transition(
+    current: &Mailbox,
+    next: &Mailbox,
+    absence_proven: bool,
+) -> Result<(), DomainStoreError> {
+    if current.generation_id != next.generation_id
+        || current.identity_path != next.identity_path
+        || current.inbox_path != next.inbox_path
+        || current.outbox_path != next.outbox_path
+    {
+        return Err(DomainStoreError::InvalidRow("Mailbox identity"));
+    }
+    if absence_proven {
+        return if matches!(
+            current.state,
+            ParticipantState::Intent | ParticipantState::ReleaseIntent
+        ) && next.state == ParticipantState::Clean
+        {
+            Ok(())
+        } else {
+            Err(DomainStoreError::InvalidRow("Mailbox cleanup"))
+        };
+    }
+    let valid = current.state == next.state
+        || matches!(
+            (current.state, next.state),
+            (
+                ParticipantState::Intent,
+                ParticipantState::LeaseHeld
+                    | ParticipantState::Prepared
+                    | ParticipantState::Committed
+                    | ParticipantState::ReleaseIntent
+            ) | (
+                ParticipantState::LeaseHeld
+                    | ParticipantState::Prepared
+                    | ParticipantState::Committed,
+                ParticipantState::ReleaseIntent
+            )
+        );
+    if valid {
+        Ok(())
+    } else {
+        Err(DomainStoreError::InvalidRow("Mailbox transition"))
+    }
+}
+
+fn validate_snapshot(snapshot: &DomainStore) -> Result<(), DomainStoreError> {
+    for (index, generation) in snapshot.generations.iter().enumerate() {
+        if generation.domain_id != snapshot.domain.domain_id {
+            return Err(DomainStoreError::DomainMismatch);
+        }
+        if snapshot.generations[..index]
+            .iter()
+            .any(|other| other.generation_id == generation.generation_id)
+        {
+            return Err(DomainStoreError::DuplicateRow("Generation"));
+        }
+    }
+    for (index, participant) in snapshot.participants.iter().enumerate() {
+        require_generation(snapshot, participant.generation_id())?;
+        if snapshot.participants[..index].iter().any(|other| {
+            other.generation_id() == participant.generation_id()
+                && other.role() == participant.role()
+        }) {
+            return Err(DomainStoreError::DuplicateRow("Participant"));
+        }
+    }
+    for (index, session) in snapshot.sessions.iter().enumerate() {
+        require_generation(snapshot, session.generation_id)?;
+        if snapshot.sessions[..index].iter().any(|other| {
+            other.generation_id == session.generation_id && other.session_id == session.session_id
+        }) {
+            return Err(DomainStoreError::DuplicateRow("Session"));
+        }
+    }
+    for (index, terminal) in snapshot.session_terminals.iter().enumerate() {
+        require_session(snapshot, terminal.generation_id(), terminal.session_id())?;
+        if snapshot.session_terminals[..index].iter().any(|other| {
+            other.generation_id() == terminal.generation_id()
+                && other.session_id() == terminal.session_id()
+        }) {
+            return Err(DomainStoreError::DuplicateRow("Session Terminal"));
+        }
+    }
+    for (index, mailbox) in snapshot.mailboxes.iter().enumerate() {
+        require_generation(snapshot, mailbox.generation_id)?;
+        if snapshot.mailboxes[..index]
+            .iter()
+            .any(|other| other.generation_id == mailbox.generation_id)
+        {
+            return Err(DomainStoreError::DuplicateRow("Mailbox"));
+        }
+    }
+    Ok(())
+}
+
+fn require_generation(
+    snapshot: &DomainStore,
+    generation_id: GenerationId,
+) -> Result<(), DomainStoreError> {
+    if snapshot
+        .generations
+        .iter()
+        .any(|record| record.generation_id == generation_id)
+    {
+        Ok(())
+    } else {
+        Err(DomainStoreError::GenerationMismatch)
+    }
+}
+
+fn require_session(
+    snapshot: &DomainStore,
+    generation_id: GenerationId,
+    session_id: SessionId,
+) -> Result<(), DomainStoreError> {
+    require_generation(snapshot, generation_id)?;
+    if snapshot
+        .sessions
+        .iter()
+        .any(|record| record.generation_id == generation_id && record.session_id == session_id)
+    {
+        Ok(())
+    } else {
+        Err(DomainStoreError::MissingRow("Session"))
+    }
+}
+
+fn participant_key(record: &Participant) -> [u8; 9] {
+    let mut key = [0; 9];
+    key[..8].copy_from_slice(&record.generation_id().get().to_be_bytes());
+    key[8] = role_key(record.role());
+    key
+}
+
+fn role_key(role: ParticipantRole) -> u8 {
+    match role {
+        ParticipantRole::PreparedLower => 0,
+        ParticipantRole::StateAnchor => 1,
+        ParticipantRole::GitRouting => 2,
+        ParticipantRole::Network => 3,
+        ParticipantRole::Helper => 4,
+        ParticipantRole::Sandbox => 5,
+        ParticipantRole::Publication => 6,
+    }
+}
+
+fn session_key(generation_id: GenerationId, session_id: SessionId) -> [u8; 40] {
+    let mut key = [0; 40];
+    key[..8].copy_from_slice(&generation_id.get().to_be_bytes());
+    key[8..].copy_from_slice(session_id.as_bytes());
+    key
+}
+
+#[cfg(test)]
+mod tests;

crates/lazaret-host-coordinator/src/domain_store/tests.rs

@@ -0,0 +1,374 @@
+use std::fs::{File, OpenOptions};
+use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicU64, Ordering};
+
+use lazaret_core::codec::encode;
+use lazaret_core::identity::{
+    ClientId, ConfigurationFingerprint, CoordinatorId, DomainId, GenerationId, HostPath,
+    NetworkSidecarId, SandboxId, SessionAnchorId, SessionId,
+};
+use lazaret_core::records::{
+    DurableDecision, ParticipantOwnership, PersistedHostProcessOwnership, PublicationOwnership,
+    SessionTerminalFields,
+};
+use lazaret_linux::filesystem::try_exclusive_flock;
+use redb::{ReadableDatabase, TableDefinition, TableHandle};
+use serde::Serialize;
+
+use super::*;
+
+static NEXT_DIRECTORY: AtomicU64 = AtomicU64::new(0);
+const FOREIGN: TableDefinition<u8, u8> = TableDefinition::new("foreign");
+
+struct TestDirectory(PathBuf);
+
+impl TestDirectory {
+    fn new(label: &str) -> Self {
+        let sequence = NEXT_DIRECTORY.fetch_add(1, Ordering::Relaxed);
+        let path = std::env::temp_dir().join(format!(
+            "lazaret-domain-store-{label}-{}-{sequence}",
+            std::process::id()
+        ));
+        std::fs::create_dir(&path).unwrap();
+        Self(path)
+    }
+
+    fn path(&self, name: &str) -> PathBuf {
+        self.0.join(name)
+    }
+}
+
+impl Drop for TestDirectory {
+    fn drop(&mut self) {
+        std::fs::remove_dir_all(&self.0).unwrap();
+    }
+}
+
+fn bytes(value: u8) -> [u8; 32] {
+    [value; 32]
+}
+
+fn domain() -> Domain {
+    Domain {
+        domain_id: DomainId::new(bytes(1)),
+        profile_name: "trusted".into(),
+        canonical_workspace: HostPath::new(b"/workspace/project".to_vec()),
+    }
+}
+
+fn generation(decision: DurableDecision) -> Generation {
+    Generation {
+        domain_id: domain().domain_id,
+        generation_id: GenerationId::new(7),
+        configuration_fingerprint: ConfigurationFingerprint::new(bytes(2)),
+        coordinator_id: CoordinatorId::new(bytes(3)),
+        network_sidecar_id: NetworkSidecarId::new(bytes(4)),
+        sandbox_id: SandboxId::new(bytes(5)),
+        durable_decision: decision,
+    }
+}
+
+fn session(phase: SessionPhase) -> Session {
+    Session {
+        generation_id: GenerationId::new(7),
+        session_id: SessionId::new(bytes(6)),
+        cli_id: ClientId::new(bytes(7)),
+        session_anchor_id: SessionAnchorId::new(bytes(8)),
+        session_phase: phase,
+    }
+}
+
+fn mailbox(state: ParticipantState) -> Mailbox {
+    Mailbox {
+        generation_id: GenerationId::new(7),
+        identity_path: HostPath::new(b"identity".to_vec()),
+        inbox_path: HostPath::new(b"inbox".to_vec()),
+        outbox_path: HostPath::new(b"outbox".to_vec()),
+        state,
+    }
+}
+
+fn bare_terminal(state: SessionTerminalState) -> SessionTerminal {
+    SessionTerminal::snapshot(SessionTerminalFields {
+        generation_id: GenerationId::new(7),
+        session_id: SessionId::new(bytes(6)),
+        state,
+        container_id: None,
+        host_coordinator: PersistedHostProcessOwnership::unassigned(),
+        pty_master: None,
+        exec_client: PersistedHostProcessOwnership::unassigned(),
+        guest_process: None,
+    })
+    .unwrap()
+}
+
+fn publication_intent() -> Participant {
+    Participant::bare_intent(GenerationId::new(7), ParticipantRole::Publication)
+}
+
+fn publication_partial() -> Participant {
+    Participant::snapshot(
+        GenerationId::new(7),
+        ParticipantRole::Publication,
+        ParticipantState::Intent,
+        Some(ParticipantOwnership::Publication(PublicationOwnership {
+            socket_path: Some(HostPath::new(b"control.sock".to_vec())),
+            identity_path: None,
+        })),
+    )
+    .unwrap()
+}
+
+fn lock_file(path: &Path) -> File {
+    OpenOptions::new()
+        .read(true)
+        .write(true)
+        .create(true)
+        .truncate(false)
+        .open(path)
+        .unwrap()
+}
+
+fn open_store(directory: &TestDirectory) -> HostCoordinatorDomainStore {
+    let lock = try_exclusive_flock(lock_file(&directory.path("domain.lock"))).unwrap();
+    HostCoordinatorDomainStore::open(lock, &directory.path("domain.redb"), &domain()).unwrap()
+}
+
+#[test]
+fn transaction_reopen_and_interrupted_write_preserve_complete_documents() {
+    let directory = TestDirectory::new("transaction");
+    let store = open_store(&directory);
+    store
+        .apply(&[
+            DomainStoreMutation::PutGeneration(generation(DurableDecision::None)),
+            DomainStoreMutation::PutParticipant(publication_intent()),
+            DomainStoreMutation::PutSession(session(SessionPhase::New)),
+            DomainStoreMutation::PutMailbox(mailbox(ParticipantState::Intent)),
+            DomainStoreMutation::PutSessionTerminal(bare_terminal(SessionTerminalState::Intent)),
+        ])
+        .unwrap();
+
+    let refused = store.apply(&[
+        DomainStoreMutation::PutGeneration(generation(DurableDecision::Preparing)),
+        DomainStoreMutation::PutSession(session(SessionPhase::Attached)),
+    ]);
+    assert!(matches!(
+        refused,
+        Err(DomainStoreError::InvalidRow("Session transition"))
+    ));
+    assert_eq!(
+        store.snapshot().unwrap().generations[0].durable_decision,
+        DurableDecision::None
+    );
+    drop(store);
+
+    let database = Database::open(directory.path("domain.redb")).unwrap();
+    let write = database.begin_write().unwrap();
+    {
+        let mut table = write.open_table(GENERATIONS).unwrap();
+        let encoded = encode(&generation(DurableDecision::Preparing)).unwrap();
+        table.insert(7, encoded.as_slice()).unwrap();
+    }
+    drop(write);
+    drop(database);
+
+    let reopened = open_store(&directory);
+    let snapshot = reopened.snapshot().unwrap();
+    assert_eq!(
+        snapshot.generations[0].durable_decision,
+        DurableDecision::None
+    );
+    assert_eq!(snapshot.participants.len(), 1);
+    assert_eq!(snapshot.sessions.len(), 1);
+    assert_eq!(snapshot.mailboxes.len(), 1);
+    assert_eq!(snapshot.session_terminals.len(), 1);
+}
+
+#[test]
+fn opener_retains_the_domain_lock_for_its_complete_lifetime() {
+    let directory = TestDirectory::new("opener");
+    let store = open_store(&directory);
+    assert!(try_exclusive_flock(lock_file(&directory.path("domain.lock"))).is_err());
+    assert!(matches!(
+        Database::open(directory.path("domain.redb")),
+        Err(DatabaseError::DatabaseAlreadyOpen)
+    ));
+    drop(store);
+    assert!(try_exclusive_flock(lock_file(&directory.path("domain.lock"))).is_ok());
+}
+
+#[test]
+fn metadata_and_typed_row_validation_fail_closed() {
+    #[derive(Serialize)]
+    struct RawMetadata<'a> {
+        schema_id: &'a str,
+        schema_version: u32,
+    }
+
+    let directory = TestDirectory::new("metadata");
+    drop(open_store(&directory));
+    let database = Database::open(directory.path("domain.redb")).unwrap();
+    let write = database.begin_write().unwrap();
+    {
+        let mut table = write.open_table(METADATA).unwrap();
+        let encoded = encode(&RawMetadata {
+            schema_id: "lazaret-domain-store-v1",
+            schema_version: 2,
+        })
+        .unwrap();
+        table.insert(SINGLETON_KEY, encoded.as_slice()).unwrap();
+    }
+    write.commit().unwrap();
+    drop(database);
+    let lock = try_exclusive_flock(lock_file(&directory.path("domain.lock"))).unwrap();
+    assert!(matches!(
+        HostCoordinatorDomainStore::open(lock, &directory.path("domain.redb"), &domain()),
+        Err(DomainStoreError::Codec(_))
+    ));
+
+    let row_directory = TestDirectory::new("row-key");
+    let store = open_store(&row_directory);
+    store
+        .apply(&[DomainStoreMutation::PutGeneration(generation(
+            DurableDecision::None,
+        ))])
+        .unwrap();
+    drop(store);
+    let database = Database::open(row_directory.path("domain.redb")).unwrap();
+    let write = database.begin_write().unwrap();
+    {
+        let mut table = write.open_table(GENERATIONS).unwrap();
+        let encoded = encode(&generation(DurableDecision::None)).unwrap();
+        table.insert(8, encoded.as_slice()).unwrap();
+    }
+    write.commit().unwrap();
+    drop(database);
+    let lock = try_exclusive_flock(lock_file(&row_directory.path("domain.lock"))).unwrap();
+    assert!(matches!(
+        HostCoordinatorDomainStore::open(lock, &row_directory.path("domain.redb"), &domain()),
+        Err(DomainStoreError::InvalidRow("Generation key"))
+    ));
+}
+
+#[test]
+fn generation_relationships_and_domain_identity_cannot_be_forged() {
+    let directory = TestDirectory::new("relationships");
+    let store = open_store(&directory);
+    assert!(matches!(
+        store.apply(&[DomainStoreMutation::PutParticipant(publication_intent())]),
+        Err(DomainStoreError::GenerationMismatch)
+    ));
+
+    let mut wrong_domain = generation(DurableDecision::None);
+    wrong_domain.domain_id = DomainId::new(bytes(99));
+    assert!(matches!(
+        store.apply(&[DomainStoreMutation::PutGeneration(wrong_domain)]),
+        Err(DomainStoreError::DomainMismatch)
+    ));
+    store
+        .apply(&[DomainStoreMutation::PutGeneration(generation(
+            DurableDecision::None,
+        ))])
+        .unwrap();
+    drop(store);
+
+    let lock = try_exclusive_flock(lock_file(&directory.path("domain.lock"))).unwrap();
+    let mut other_domain = domain();
+    other_domain.profile_name = "other".into();
+    assert!(matches!(
+        HostCoordinatorDomainStore::open(lock, &directory.path("domain.redb"), &other_domain),
+        Err(DomainStoreError::DomainMismatch)
+    ));
+}
+
+#[test]
+fn lifecycle_updates_enforce_transition_and_absence_boundaries() {
+    let directory = TestDirectory::new("transitions");
+    let store = open_store(&directory);
+    store
+        .apply(&[
+            DomainStoreMutation::PutGeneration(generation(DurableDecision::None)),
+            DomainStoreMutation::PutParticipant(publication_intent()),
+            DomainStoreMutation::PutSession(session(SessionPhase::New)),
+            DomainStoreMutation::PutMailbox(mailbox(ParticipantState::Intent)),
+            DomainStoreMutation::PutSessionTerminal(bare_terminal(SessionTerminalState::Intent)),
+        ])
+        .unwrap();
+    store
+        .apply(&[DomainStoreMutation::PutParticipant(publication_partial())])
+        .unwrap();
+
+    assert!(matches!(
+        store.apply(&[DomainStoreMutation::PutSession(session(
+            SessionPhase::Attached
+        ))]),
+        Err(DomainStoreError::InvalidRow("Session transition"))
+    ));
+    assert!(matches!(
+        store.apply(&[DomainStoreMutation::PutParticipant(
+            Participant::snapshot(
+                GenerationId::new(7),
+                ParticipantRole::Publication,
+                ParticipantState::Clean,
+                Some(ParticipantOwnership::Publication(PublicationOwnership {
+                    socket_path: None,
+                    identity_path: None,
+                })),
+            )
+            .unwrap(),
+        )]),
+        Err(DomainStoreError::Transition(
+            RecordTransitionError::AbsenceProofRequired
+        ))
+    ));
+    assert!(matches!(
+        store.apply(&[DomainStoreMutation::PutMailbox(mailbox(
+            ParticipantState::Clean
+        ))]),
+        Err(DomainStoreError::InvalidRow("Mailbox transition"))
+    ));
+    assert!(matches!(
+        store.apply(&[DomainStoreMutation::PutSessionTerminal(bare_terminal(
+            SessionTerminalState::Clean,
+        ))]),
+        Err(DomainStoreError::Transition(
+            RecordTransitionError::AbsenceProofRequired
+        ))
+    ));
+
+    store
+        .apply(&[DomainStoreMutation::CleanSessionTerminalAfterAbsence(
+            bare_terminal(SessionTerminalState::Clean),
+        )])
+        .unwrap();
+    let snapshot = store.snapshot().unwrap();
+    assert_eq!(
+        snapshot.session_terminals[0].state(),
+        SessionTerminalState::Clean
+    );
+}
+
+#[test]
+fn every_table_is_created_in_the_first_version_document() {
+    let directory = TestDirectory::new("tables");
+    drop(open_store(&directory));
+    let database = Database::open(directory.path("domain.redb")).unwrap();
+    let read = database.begin_read().unwrap();
+    for name in [
+        METADATA.name(),
+        DOMAINS.name(),
+        GENERATIONS.name(),
+        PARTICIPANTS.name(),
+        SESSIONS.name(),
+        SESSION_TERMINALS.name(),
+        MAILBOXES.name(),
+    ] {
+        assert!(
+            read.list_tables()
+                .unwrap()
+                .any(|handle| handle.name() == name)
+        );
+    }
+
+    assert!(read.open_table(FOREIGN).is_err());
+}

crates/lazaret-host-coordinator/src/lib.rs

@@ -0,0 +1,3 @@
+#![doc = "Host-side coordination authority for one Lazaret Domain."]
+
+pub mod domain_store;