Make Domain Store mutations row-local

crates/lazaret-domain-state/src/domain_store.rs

@@ -18,6 +18,9 @@ use redb::{
 };
 use serde::{Deserialize, Serialize};
 
+mod transactions;
+use transactions::{apply_row_mutation, require_transaction_domain};
+
 const METADATA: TableDefinition<u8, &[u8]> = TableDefinition::new("metadata");
 const DOMAINS: TableDefinition<u8, &[u8]> = TableDefinition::new("domains");
 const GENERATIONS: TableDefinition<u64, &[u8]> = TableDefinition::new("generations");
@@ -263,12 +266,13 @@ impl HostCoordinatorDomainStore {
         if mutations.is_empty() {
             return Ok(());
         }
-        let mut snapshot = self.snapshot()?;
+        self.validate_metadata()?;
+        let write = self.database.begin_write()?;
         for mutation in mutations {
-            apply_mutation(&mut snapshot, mutation)?;
+            apply_row_mutation(&write, mutation)?;
         }
-        validate_snapshot(&snapshot)?;
-        self.write_snapshot(&snapshot)
+        write.commit()?;
+        Ok(())
     }
 
     /// Reads the typed reset recovery point from the versioned Domain Store.
@@ -288,7 +292,8 @@ impl HostCoordinatorDomainStore {
         };
         let intent = decode::<StateResetIntentV1>(value.value())?;
         intent.validate()?;
-        if intent.domain_id != self.snapshot()?.domain.domain_id {
+        let domain = read_domain(&read.open_table(DOMAINS)?)?;
+        if intent.domain_id != domain.domain_id {
             return Err(DomainStoreError::DomainMismatch);
         }
         Ok(Some(intent))
@@ -300,10 +305,9 @@ impl HostCoordinatorDomainStore {
     /// Refuses an invalid, foreign, or already-present intent.
     pub fn begin_state_reset(&self, intent: &StateResetIntentV1) -> Result<(), DomainStoreError> {
         intent.validate()?;
-        if intent.domain_id != self.snapshot()?.domain.domain_id {
-            return Err(DomainStoreError::DomainMismatch);
-        }
+        self.validate_metadata()?;
         let write = self.database.begin_write()?;
+        require_transaction_domain(&write, intent.domain_id)?;
         {
             let mut table = write.open_table(STATE_RESET)?;
             if table.get(SINGLETON_KEY)?.is_some() {
@@ -324,12 +328,19 @@ impl HostCoordinatorDomainStore {
         &self,
         expected: &StateResetIntentV1,
     ) -> Result<(), DomainStoreError> {
-        if self.state_reset_intent()?.as_ref() != Some(expected) {
-            return Err(DomainStoreError::InvalidRow("State Reset"));
-        }
+        expected.validate()?;
+        self.validate_metadata()?;
         let write = self.database.begin_write()?;
+        require_transaction_domain(&write, expected.domain_id)?;
         {
             let mut table = write.open_table(STATE_RESET)?;
+            let actual = table
+                .get(SINGLETON_KEY)?
+                .map(|value| decode::<StateResetIntentV1>(value.value()))
+                .transpose()?;
+            if actual.as_ref() != Some(expected) {
+                return Err(DomainStoreError::InvalidRow("State Reset"));
+            }
             table.remove(SINGLETON_KEY)?;
         }
         write.commit()?;
@@ -415,6 +426,19 @@ impl HostCoordinatorDomainStore {
         write.commit()?;
         Ok(())
     }
+
+    #[cfg(test)]
+    fn apply_reference(&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 read_generations(
@@ -524,6 +548,7 @@ fn read_mailboxes(
     Ok(records)
 }
 
+#[cfg(test)]
 fn apply_mutation(
     snapshot: &mut DomainStore,
     mutation: &DomainStoreMutation,
@@ -548,6 +573,7 @@ fn apply_mutation(
     }
 }
 
+#[cfg(test)]
 fn put_generation(snapshot: &mut DomainStore, next: Generation) -> Result<(), DomainStoreError> {
     if next.domain_id != snapshot.domain.domain_id {
         return Err(DomainStoreError::DomainMismatch);
@@ -568,6 +594,7 @@ fn put_generation(snapshot: &mut DomainStore, next: Generation) -> Result<(), Do
     Ok(())
 }
 
+#[cfg(test)]
 fn put_participant(
     snapshot: &mut DomainStore,
     next: Participant,
@@ -593,6 +620,7 @@ fn put_participant(
     Ok(())
 }
 
+#[cfg(test)]
 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| {
@@ -612,6 +640,7 @@ fn put_session(snapshot: &mut DomainStore, next: Session) -> Result<(), DomainSt
     Ok(())
 }
 
+#[cfg(test)]
 fn put_mailbox(
     snapshot: &mut DomainStore,
     next: Mailbox,
@@ -634,6 +663,7 @@ fn put_mailbox(
     Ok(())
 }
 
+#[cfg(test)]
 fn put_session_terminal(
     snapshot: &mut DomainStore,
     next: SessionTerminal,

crates/lazaret-domain-state/src/domain_store/tests.rs

@@ -25,11 +25,10 @@ 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();
+        let path = Path::new(env!("CARGO_MANIFEST_DIR"))
+            .join("../../target/test-artifacts/domain-store")
+            .join(format!("{label}-{}-{sequence}", std::process::id()));
+        std::fs::create_dir_all(&path).unwrap();
         Self(path)
     }
 
@@ -123,6 +122,23 @@ fn publication_partial() -> Participant {
     .unwrap()
 }
 
+fn publication_clean() -> Participant {
+    Participant::snapshot(
+        GenerationId::new(7),
+        ParticipantRole::Publication,
+        ParticipantState::Clean,
+        Some(ParticipantOwnership::Publication(PublicationOwnership {
+            staged_socket_path: None,
+            staged_identity_path: None,
+            socket_path: None,
+            identity_path: None,
+            socket_identity: None,
+            identity_file_identity: None,
+        })),
+    )
+    .unwrap()
+}
+
 fn lock_file(path: &Path) -> File {
     OpenOptions::new()
         .read(true)
@@ -138,6 +154,132 @@ fn open_store(directory: &TestDirectory) -> HostCoordinatorDomainStore {
     HostCoordinatorDomainStore::open(lock, &directory.path("domain.redb"), &domain()).unwrap()
 }
 
+fn assert_differential_step(
+    row_store: &HostCoordinatorDomainStore,
+    reference_store: &HostCoordinatorDomainStore,
+    mutations: &[DomainStoreMutation],
+) {
+    let row_result = row_store.apply(mutations);
+    let reference_result = reference_store.apply_reference(mutations);
+    match (&row_result, &reference_result) {
+        (Ok(()), Ok(())) => {}
+        (Err(row_error), Err(reference_error)) => {
+            assert_eq!(format!("{row_error:?}"), format!("{reference_error:?}"));
+        }
+        _ => {
+            panic!("row-local and reference results differ: {row_result:?} != {reference_result:?}")
+        }
+    }
+    assert_eq!(
+        row_store.snapshot().unwrap(),
+        reference_store.snapshot().unwrap()
+    );
+}
+
+#[test]
+fn row_local_mutations_match_the_snapshot_reference_engine() {
+    let row_directory = TestDirectory::new("differential-row");
+    let reference_directory = TestDirectory::new("differential-reference");
+    let row_store = open_store(&row_directory);
+    let reference_store = open_store(&reference_directory);
+
+    let mut foreign_generation = generation(DurableDecision::None);
+    foreign_generation.domain_id = DomainId::new(bytes(99));
+    let steps = vec![
+        vec![DomainStoreMutation::PutParticipant(publication_intent())],
+        vec![DomainStoreMutation::PutGeneration(foreign_generation)],
+        vec![
+            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)),
+        ],
+        vec![DomainStoreMutation::PutGeneration(generation(
+            DurableDecision::Preparing,
+        ))],
+        vec![DomainStoreMutation::PutGeneration(generation(
+            DurableDecision::None,
+        ))],
+        vec![DomainStoreMutation::PutSession(session(
+            SessionPhase::Waiting,
+        ))],
+        vec![DomainStoreMutation::PutSession(session(
+            SessionPhase::Attached,
+        ))],
+        vec![DomainStoreMutation::PutParticipant(publication_partial())],
+        vec![DomainStoreMutation::PutParticipant(publication_clean())],
+        vec![DomainStoreMutation::CleanParticipantAfterAbsence(
+            publication_clean(),
+        )],
+        vec![DomainStoreMutation::PutMailbox(mailbox(
+            ParticipantState::Clean,
+        ))],
+        vec![DomainStoreMutation::CleanMailboxAfterAbsence(mailbox(
+            ParticipantState::Clean,
+        ))],
+        vec![DomainStoreMutation::PutSessionTerminal(bare_terminal(
+            SessionTerminalState::Clean,
+        ))],
+        vec![DomainStoreMutation::CleanSessionTerminalAfterAbsence(
+            bare_terminal(SessionTerminalState::Clean),
+        )],
+    ];
+
+    for mutations in steps {
+        assert_differential_step(&row_store, &reference_store, &mutations);
+    }
+
+    let before = row_store.snapshot().unwrap();
+    assert_differential_step(
+        &row_store,
+        &reference_store,
+        &[
+            DomainStoreMutation::PutGeneration(generation(DurableDecision::Published)),
+            DomainStoreMutation::PutSession(session(SessionPhase::Attached)),
+        ],
+    );
+    assert_eq!(row_store.snapshot().unwrap(), before);
+
+    drop(row_store);
+    drop(reference_store);
+    assert_eq!(
+        open_store(&row_directory).snapshot().unwrap(),
+        open_store(&reference_directory).snapshot().unwrap()
+    );
+}
+
+#[test]
+fn row_local_mutation_does_not_rewrite_an_unrelated_invalid_row() {
+    let directory = TestDirectory::new("unrelated-row");
+    let store = open_store(&directory);
+    store
+        .apply(&[DomainStoreMutation::PutGeneration(generation(
+            DurableDecision::None,
+        ))])
+        .unwrap();
+    let unrelated = b"unrelated-invalid-mailbox";
+    let write = store.database.begin_write().unwrap();
+    {
+        let mut table = write.open_table(MAILBOXES).unwrap();
+        table.insert(99, unrelated.as_slice()).unwrap();
+    }
+    write.commit().unwrap();
+
+    store
+        .apply(&[DomainStoreMutation::PutGeneration(generation(
+            DurableDecision::Preparing,
+        ))])
+        .unwrap();
+
+    let read = store.database.begin_read().unwrap();
+    let table = read.open_table(MAILBOXES).unwrap();
+    assert_eq!(table.get(99).unwrap().unwrap().value(), unrelated);
+    drop(table);
+    drop(read);
+    assert!(matches!(store.snapshot(), Err(DomainStoreError::Codec(_))));
+}
+
 #[test]
 fn reset_intent_is_typed_transactional_and_domain_bound() {
     let directory = TestDirectory::new("state-reset");

crates/lazaret-domain-state/src/domain_store/transactions.rs

@@ -0,0 +1,207 @@
+use super::{
+    DOMAINS, Domain, DomainStoreError, DomainStoreMutation, DurableDecision, GENERATIONS,
+    Generation, GenerationId, MAILBOXES, Mailbox, PARTICIPANTS, Participant, ParticipantState,
+    ReadableTable, SESSION_TERMINALS, SESSIONS, SINGLETON_KEY, Session, SessionId, SessionPhase,
+    SessionTerminal, SessionTerminalState, decode, encode, participant_key, session_key,
+    validate_generation_transition, validate_mailbox_transition, validate_session_transition,
+};
+
+pub(super) fn apply_row_mutation(
+    write: &redb::WriteTransaction,
+    mutation: &DomainStoreMutation,
+) -> Result<(), DomainStoreError> {
+    match mutation {
+        DomainStoreMutation::PutGeneration(next) => put_generation_row(write, next),
+        DomainStoreMutation::PutParticipant(next) => put_participant_row(write, next, false),
+        DomainStoreMutation::CleanParticipantAfterAbsence(next) => {
+            put_participant_row(write, next, true)
+        }
+        DomainStoreMutation::PutSession(next) => put_session_row(write, next),
+        DomainStoreMutation::PutMailbox(next) => put_mailbox_row(write, next, false),
+        DomainStoreMutation::CleanMailboxAfterAbsence(next) => put_mailbox_row(write, next, true),
+        DomainStoreMutation::PutSessionTerminal(next) => {
+            put_session_terminal_row(write, next, false)
+        }
+        DomainStoreMutation::CleanSessionTerminalAfterAbsence(next) => {
+            put_session_terminal_row(write, next, true)
+        }
+    }
+}
+
+fn put_generation_row(
+    write: &redb::WriteTransaction,
+    next: &Generation,
+) -> Result<(), DomainStoreError> {
+    require_transaction_domain(write, next.domain_id)?;
+    let mut table = write.open_table(GENERATIONS)?;
+    let current = table
+        .get(next.generation_id.get())?
+        .map(|value| decode::<Generation>(value.value()))
+        .transpose()?;
+    if let Some(current) = current {
+        validate_generation_transition(&current, next)?;
+    } else if next.durable_decision != DurableDecision::None {
+        return Err(DomainStoreError::InvalidRow("initial Generation"));
+    }
+    let encoded = encode(next)?;
+    table.insert(next.generation_id.get(), encoded.as_slice())?;
+    Ok(())
+}
+
+fn put_participant_row(
+    write: &redb::WriteTransaction,
+    next: &Participant,
+    absence_proven: bool,
+) -> Result<(), DomainStoreError> {
+    require_transaction_generation(write, next.generation_id())?;
+    let mut table = write.open_table(PARTICIPANTS)?;
+    let key = participant_key(next);
+    let current = table
+        .get(&key)?
+        .map(|value| decode::<Participant>(value.value()))
+        .transpose()?;
+    if let Some(current) = current {
+        if absence_proven {
+            current.validate_transition_to_clean_after_absence(next)?;
+        } else {
+            current.validate_transition_to(next)?;
+        }
+    } else if absence_proven
+        || next.state() != ParticipantState::Intent
+        || next.ownership().is_some()
+    {
+        return Err(DomainStoreError::InvalidRow("initial Participant"));
+    }
+    let encoded = encode(next)?;
+    table.insert(&key, encoded.as_slice())?;
+    Ok(())
+}
+
+fn put_session_row(write: &redb::WriteTransaction, next: &Session) -> Result<(), DomainStoreError> {
+    require_transaction_generation(write, next.generation_id)?;
+    let mut table = write.open_table(SESSIONS)?;
+    let key = session_key(next.generation_id, next.session_id);
+    let current = table
+        .get(&key)?
+        .map(|value| decode::<Session>(value.value()))
+        .transpose()?;
+    if let Some(current) = current {
+        validate_session_transition(&current, next)?;
+    } else if !matches!(
+        next.session_phase,
+        SessionPhase::New | SessionPhase::Waiting | SessionPhase::Registering
+    ) {
+        return Err(DomainStoreError::InvalidRow("initial Session"));
+    }
+    let encoded = encode(next)?;
+    table.insert(&key, encoded.as_slice())?;
+    Ok(())
+}
+
+fn put_mailbox_row(
+    write: &redb::WriteTransaction,
+    next: &Mailbox,
+    absence_proven: bool,
+) -> Result<(), DomainStoreError> {
+    require_transaction_generation(write, next.generation_id)?;
+    let mut table = write.open_table(MAILBOXES)?;
+    let current = table
+        .get(next.generation_id.get())?
+        .map(|value| decode::<Mailbox>(value.value()))
+        .transpose()?;
+    if let Some(current) = current {
+        validate_mailbox_transition(&current, next, absence_proven)?;
+    } else if absence_proven || next.state != ParticipantState::Intent {
+        return Err(DomainStoreError::InvalidRow("initial Mailbox"));
+    }
+    let encoded = encode(next)?;
+    table.insert(next.generation_id.get(), encoded.as_slice())?;
+    Ok(())
+}
+
+fn put_session_terminal_row(
+    write: &redb::WriteTransaction,
+    next: &SessionTerminal,
+    absence_proven: bool,
+) -> Result<(), DomainStoreError> {
+    require_transaction_session(write, next.generation_id(), next.session_id())?;
+    let mut table = write.open_table(SESSION_TERMINALS)?;
+    let key = session_key(next.generation_id(), next.session_id());
+    let current = table
+        .get(&key)?
+        .map(|value| decode::<SessionTerminal>(value.value()))
+        .transpose()?;
+    if let Some(current) = current {
+        if absence_proven {
+            current.validate_transition_to_clean_after_absence(next)?;
+        } else {
+            current.validate_transition_to(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"));
+    }
+    let encoded = encode(next)?;
+    table.insert(&key, encoded.as_slice())?;
+    Ok(())
+}
+
+pub(super) fn require_transaction_domain(
+    write: &redb::WriteTransaction,
+    expected: lazaret_core::identity::DomainId,
+) -> Result<(), DomainStoreError> {
+    let table = write.open_table(DOMAINS)?;
+    let domain = table
+        .get(SINGLETON_KEY)?
+        .map(|value| decode::<Domain>(value.value()))
+        .transpose()?
+        .ok_or(DomainStoreError::MissingRow("Domain"))?;
+    if domain.domain_id == expected {
+        Ok(())
+    } else {
+        Err(DomainStoreError::DomainMismatch)
+    }
+}
+
+fn require_transaction_generation(
+    write: &redb::WriteTransaction,
+    generation_id: GenerationId,
+) -> Result<(), DomainStoreError> {
+    let table = write.open_table(GENERATIONS)?;
+    let generation = table
+        .get(generation_id.get())?
+        .map(|value| decode::<Generation>(value.value()))
+        .transpose()?
+        .ok_or(DomainStoreError::GenerationMismatch)?;
+    if generation.generation_id == generation_id {
+        Ok(())
+    } else {
+        Err(DomainStoreError::GenerationMismatch)
+    }
+}
+
+fn require_transaction_session(
+    write: &redb::WriteTransaction,
+    generation_id: GenerationId,
+    session_id: SessionId,
+) -> Result<(), DomainStoreError> {
+    require_transaction_generation(write, generation_id)?;
+    let table = write.open_table(SESSIONS)?;
+    let key = session_key(generation_id, session_id);
+    let session = table
+        .get(&key)?
+        .map(|value| decode::<Session>(value.value()))
+        .transpose()?
+        .ok_or(DomainStoreError::MissingRow("Session"))?;
+    if session.generation_id == generation_id && session.session_id == session_id {
+        Ok(())
+    } else {
+        Err(DomainStoreError::InvalidRow("Session key"))
+    }
+}