ms2k/lazaret
SHA256
Watch
0
Code Releases 3 Activity

Test Domain Store mutations without a duplicate policy engine

Browse source
This commit is contained in:
ms2k 2026-07-19 05:25:41 +09:00
commit 56b54163ae
2 changed files with 47 additions and 261 deletions
Download patch file Download diff file Expand all files Collapse all files

160
crates/lazaret-domain-state/src/domain_store.rs
View file

@ -434,19 +434,6 @@ 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(
@ -556,153 +543,6 @@ fn read_mailboxes(
Ok(records)
}
#[cfg(test)]
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)
}
}
}
#[cfg(test)]
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(())
}
#[cfg(test)]
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(())
}
#[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| {
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(())
}
#[cfg(test)]
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(())
}
#[cfg(test)]
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,

142
crates/lazaret-domain-state/src/domain_store/tests.rs
View file

@ -133,23 +133,6 @@ 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)
@ -165,99 +148,62 @@ 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![
fn row_local_mutations_produce_explicit_snapshots_and_atomic_rollback() {
let directory = TestDirectory::new("row-local-snapshots");
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)),
],
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);
])
.unwrap();
assert_eq!(
store.snapshot().unwrap(),
DomainStore {
metadata: DomainStoreMetadata::v1(),
domain: domain(),
generations: vec![generation(DurableDecision::None)],
participants: vec![publication_intent()],
sessions: vec![session(SessionPhase::New)],
session_terminals: vec![bare_terminal(SessionTerminalState::Intent)],
mailboxes: vec![mailbox(ParticipantState::Intent)],
}
);
let before = row_store.snapshot().unwrap();
assert_differential_step(
&row_store,
&reference_store,
&[
store
.apply(&[
DomainStoreMutation::PutGeneration(generation(DurableDecision::Preparing)),
DomainStoreMutation::PutSession(session(SessionPhase::Waiting)),
DomainStoreMutation::PutParticipant(publication_partial()),
])
.unwrap();
let before_refusal = DomainStore {
metadata: DomainStoreMetadata::v1(),
domain: domain(),
generations: vec![generation(DurableDecision::Preparing)],
participants: vec![publication_partial()],
sessions: vec![session(SessionPhase::Waiting)],
session_terminals: vec![bare_terminal(SessionTerminalState::Intent)],
mailboxes: vec![mailbox(ParticipantState::Intent)],
};
assert_eq!(store.snapshot().unwrap(), before_refusal);
assert!(
store
.apply(&[
DomainStoreMutation::PutGeneration(generation(DurableDecision::Published)),
DomainStoreMutation::PutSession(session(SessionPhase::Attached)),
],
],)
.is_err()
);
assert_eq!(row_store.snapshot().unwrap(), before);
assert_eq!(store.snapshot().unwrap(), before_refusal);
drop(row_store);
drop(reference_store);
assert_eq!(
open_store(&row_directory).snapshot().unwrap(),
open_store(&reference_directory).snapshot().unwrap()
);
drop(store);
assert_eq!(open_store(&directory).snapshot().unwrap(), before_refusal);
}
#[test]