use std::collections::BTreeMap;
use std::error::Error;
use std::fmt;
use std::sync::Arc;
use liminal::durability::{DurableStore, bridge::block_on, open_ephemeral};
use liminal_protocol::lifecycle::{
ConversationLifecycleView, LifecycleCapsule, LifecycleView, ParticipantLifecycleView,
RestoredParticipantLifecycle,
};
use liminal_protocol::wire::{
AttachAttemptToken, ClientRequest, ConnectionIncarnation, CredentialAttachRequest,
EnrollmentRequest, EnrollmentToken, Generation, ServerValue,
};
use super::super::ProductionParticipantHandler;
use super::super::barrier::ReceiptCapacityLimits;
use super::super::log::{DecodedStoredOperation, OperationLog, StoredOperation};
use super::super::outbox::{ConversationOutbox, ConversationOutboxLimits};
use super::super::outbox_projection::{ReplayedProjectionFacts, project_committed_source};
use super::super::restore_checkpoint::{CheckpointPositions, CheckpointSchema};
use super::super::tests::{dispatch_tracked, test_participant_config};
use super::checkpoint_slots::SlotRestoreParts;
use super::{ConversationAuthority, PreparedOrdinaryFinalizer, Slot};
use crate::server::participant::{
ConnectionFateClass, ConnectionFateWorkItem, ParticipantConnectionConversations,
ParticipantSemanticHandler,
};
type TestResult<T = ()> = Result<T, Box<dyn Error>>;
type RestoredFixture = (
BTreeMap<u64, Slot>,
BTreeMap<u64, (liminal_protocol::lifecycle::BindingOrigin, u64)>,
);
struct AttachedSnapshot {
capsule: Vec<u8>,
sources: BTreeMap<u64, Vec<u8>>,
slots: BTreeMap<u64, Vec<u8>>,
positions: CheckpointPositions,
outbox: Vec<u8>,
}
struct OrdinaryFixture {
attached: AttachedSnapshot,
pending: ConversationAuthority,
prepared: ConversationAuthority,
pending_positions: CheckpointPositions,
}
#[derive(Debug)]
struct RoundtripErrors(Vec<(&'static str, Box<dyn Error>)>);
impl fmt::Display for RoundtripErrors {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
for (name, error) in &self.0 {
writeln!(formatter, "{name}: {error}")?;
}
Ok(())
}
}
impl Error for RoundtripErrors {
fn source(&self) -> Option<&(dyn Error + 'static)> {
self.0.first().map(|(_, error)| error.as_ref())
}
}
#[test]
fn ordinary_slot_pending_and_prepared_checkpoint_paths_round_trip() -> TestResult {
let fixture = build_fixture()?;
let results = [
("ordinary_slot", ordinary_slot_round_trip(&fixture)),
("ordinary_pending", ordinary_pending_round_trip(&fixture)),
(
"ordinary_prepared_finalizer",
ordinary_prepared_round_trip(&fixture),
),
];
let mut failures = Vec::new();
for (name, result) in results {
match result {
Ok(()) => println!("ordinary_checkpoint_control {name} passed"),
Err(error) => {
println!("ordinary_checkpoint_control {name} failed: {error}");
failures.push((name, error));
}
}
}
if failures.is_empty() {
Ok(())
} else {
Err(Box::new(RoundtripErrors(failures)))
}
}
fn build_fixture() -> TestResult<OrdinaryFixture> {
let config = test_participant_config();
let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
let handler = ProductionParticipantHandler::new(Arc::clone(&store), config)?;
let incarnation = ConnectionIncarnation::new(89, 7);
let conversation_id = 59;
let mut conversations = ParticipantConnectionConversations::default();
let enrolled = dispatch_tracked(
&handler,
incarnation,
&mut conversations,
ClientRequest::Enrollment(EnrollmentRequest {
conversation_id,
enrollment_token: EnrollmentToken::new([61; 16]),
}),
)?;
let ServerValue::EnrollBound(receipt) = enrolled else {
return Err("fixture enrollment refused".into());
};
let attached = dispatch_tracked(
&handler,
incarnation,
&mut conversations,
ClientRequest::CredentialAttach(CredentialAttachRequest {
conversation_id,
participant_id: receipt.participant_id(),
capability_generation: Generation::ONE,
attach_secret: receipt.attach_secret(),
attach_attempt_token: AttachAttemptToken::new([67; 16]),
accept_marker_delivery_seq: None,
}),
)?;
let ServerValue::AttachBound(_) = attached else {
return Err("fixture ordinary attach refused".into());
};
let cell = handler.cell(conversation_id)?;
let snapshot = {
let owner = cell.lock().map_err(|_| "fixture owner lock poisoned")?;
capture_component_state(owner.as_ref().ok_or("fixture owner absent")?)?
};
handler.handle_connection_fate(ConnectionFateWorkItem {
open_sequence: 23,
connection_incarnation: incarnation,
class: ConnectionFateClass::ConnectionLost,
tracked_conversations: conversations.tracked_conversations(),
})?;
let log = OperationLog::new(Arc::clone(&store), conversation_id);
let mut pending = replay_attached_prefix(&log, &snapshot, &config, &store)?;
let decoded = block_on(log.read_at(snapshot.positions.next_log_sequence))??
.ok_or("fixture Died source absent")?;
let DecodedStoredOperation::V3(operation) = decoded.operation else {
return Err("fixture source was not a native row".into());
};
let StoredOperation::Died { row } = &operation else {
return Err("fixture source was not a native Died row".into());
};
pending.replay_died_source(row, decoded.sequence)?;
project_replayed_died(&mut pending, decoded.sequence, &operation)?;
let pending_positions = pending.checkpoint_positions(
CheckpointSchema::V4Suffix,
pending.next_log_sequence,
snapshot.positions.extension_sequence,
)?;
let mut prepared = replay_attached_prefix(&log, &snapshot, &config, &store)?;
prepared.replay_died_source(row, decoded.sequence)?;
project_replayed_died(&mut prepared, decoded.sequence, &operation)?;
prepare_ordinary_finalizer(&mut prepared)?;
Ok(OrdinaryFixture {
attached: snapshot,
pending,
prepared,
pending_positions,
})
}
fn project_replayed_died(
authority: &mut ConversationAuthority,
source_log_sequence: u64,
operation: &StoredOperation,
) -> TestResult {
let projected = project_committed_source(
authority,
source_log_sequence,
operation,
ReplayedProjectionFacts::none(),
)?;
if let Some(row) = projected {
let outbox = authority.outbox.as_mut().ok_or("fixture outbox absent")?;
let (next_extension_sequence, _) = outbox.checkpoint_boundaries();
outbox
.apply_row(next_extension_sequence, row)
.map_err(|error| format!("fixture outbox projection refused: {error:?}"))?;
}
Ok(())
}
fn prepare_ordinary_finalizer(prepared: &mut ConversationAuthority) -> TestResult {
let (id, pending_fate) = prepared
.pending_specific_fates
.pop_first()
.ok_or("fixture ordinary pending fate absent")?;
let terminal = pending_fate
.terminal
.ok_or("fixture Died terminal was not committed")?;
let frontier = prepared.take_frontier()?;
let measured = frontier
.prepare_pending_died_ordinary_finalizer(
pending_fate.binding_fate.token,
terminal.terminal,
prepared.observer_progress,
)
.map_err(|refusal| {
format!(
"fixture finalizer measurement refused: {:?}",
refusal.error()
)
})?;
let (frontier, fate, finalizer) = measured.into_parts();
prepared.install_frontier(frontier)?;
prepared.prepared_ordinary_finalizers.insert(
id,
PreparedOrdinaryFinalizer {
attached_source_sequence: pending_fate.binding_fate.attached_source_sequence,
terminal: terminal.terminal,
terminal_source: terminal.source,
fate,
finalizer,
},
);
Ok(())
}
fn capture_component_state(owner: &ConversationAuthority) -> TestResult<AttachedSnapshot> {
let shell = owner.shell.as_ref().ok_or("fixture shell absent")?;
let frontier = owner.frontier().ok_or("fixture frontier absent")?;
let closure = frontier.closure_accounting().state();
let sources: BTreeMap<_, _> = owner
.slots
.iter()
.map(|(&id, slot)| (id, slot.retained_source.binding_origin.clone()))
.collect();
let borrowed: Vec<_> = sources
.iter()
.map(|(&id, bytes)| (id, bytes.as_slice()))
.collect();
let participants = owner
.slots
.values()
.map(|slot| ParticipantLifecycleView::Live {
member: &slot.member,
binding: &slot.binding,
binding_origin: None,
detach_cell: &slot.cell,
})
.collect();
let outbox = owner.outbox.as_ref().ok_or("fixture outbox absent")?;
let positions = owner.checkpoint_positions(
CheckpointSchema::V4Suffix,
owner.next_log_sequence,
outbox.checkpoint_boundaries().0,
)?;
let capsule = LifecycleCapsule::capture_with_binding_sources(
&LifecycleView {
conversation: shell,
covered_log_sequence: positions.covered_log_sequence,
sealed: None,
lifecycle: Some(ConversationLifecycleView {
participants,
frontiers: frontier.frontiers(),
retained_record_limit: frontier.retained_record_limit(),
closure: &closure,
}),
},
&borrowed,
)
.map_err(|error| format!("fixture capsule capture refused: {error:?}"))?
.encode_canonical();
let slots = owner
.slots
.iter()
.map(|(&id, slot)| Ok((id, slot.encode_checkpoint()?)))
.collect::<Result<_, super::super::restore_checkpoint::CheckpointError>>()?;
Ok(AttachedSnapshot {
capsule,
sources,
slots,
positions,
outbox: outbox.encode_checkpoint()?,
})
}
fn replay_attached_prefix(
log: &OperationLog,
snapshot: &AttachedSnapshot,
config: &crate::config::types::ParticipantConfig,
store: &Arc<dyn DurableStore>,
) -> TestResult<ConversationAuthority> {
let mut owner = ConversationAuthority::empty(
snapshot.positions.conversation_id,
ReceiptCapacityLimits::from_config(config),
);
owner.outbox = Some(ConversationOutbox::decode_checkpoint(
&snapshot.outbox,
snapshot.positions.conversation_id,
snapshot.positions.extension_sequence,
ConversationOutboxLimits::try_new(config.max_retained_record_rows, config.identity_slots)?,
)?);
for sequence in 0..snapshot.positions.next_log_sequence {
let row = block_on(log.read_at(sequence))??.ok_or("fixture prefix row absent")?;
let DecodedStoredOperation::V3(operation) = row.operation else {
return Err("fixture prefix used a different operation schema".into());
};
match operation {
StoredOperation::Genesis { event } => owner.replay_genesis(&event)?,
StoredOperation::Enrolled {
request,
allocation,
event,
} => owner.replay_enrolled(request, &allocation, &event, sequence, config)?,
StoredOperation::Attached {
request,
secret_verified,
allocation,
mode,
event,
} => {
assert!(secret_verified);
owner.replay_attached(
request,
&allocation,
&mode,
&event,
sequence,
Arc::clone(store),
)?;
}
_ => return Err("fixture prefix contained an unexpected operation".into()),
}
}
Ok(owner)
}
fn restore_component_slots(
snapshot: &AttachedSnapshot,
slot_bytes: &BTreeMap<u64, Vec<u8>>,
) -> TestResult<RestoredFixture> {
let capsule = LifecycleCapsule::decode_canonical(&snapshot.capsule)
.map_err(|error| format!("fixture capsule decode refused: {error:?}"))?;
let sources: Vec<_> = snapshot
.sources
.iter()
.map(|(&id, bytes)| (id, bytes.as_slice()))
.collect();
let parts = capsule
.restore_with_binding_sources(&sources)
.map_err(|error| format!("fixture native source refused: {error:?}"))?
.into_parts();
let lifecycle = parts.lifecycle.ok_or("fixture lifecycle absent")?;
let (participants, _, _) = lifecycle.into_parts();
let mut slots = BTreeMap::new();
let mut origins = BTreeMap::new();
for participant in participants {
let RestoredParticipantLifecycle::Live {
member,
binding,
binding_origin,
detach_cell,
} = participant
else {
return Err("fixture unexpectedly retired its identity".into());
};
let id = member.participant_id();
let origin = binding_origin.ok_or("fixture native origin absent")?;
let slot = SlotRestoreParts::decode_checkpoint(
slot_bytes.get(&id).ok_or("fixture slot bytes absent")?,
id,
&snapshot.positions,
)?;
origins.insert(id, (origin, slot.retained_source.log_sequence));
slots.insert(
id,
slot.install(
member,
binding,
detach_cell,
Some(&origin),
&snapshot.positions,
None,
)?,
);
}
Ok((slots, origins))
}
fn ordinary_slot_round_trip(fixture: &OrdinaryFixture) -> TestResult {
let (slots, _) = restore_component_slots(&fixture.attached, &fixture.attached.slots)?;
for (&id, bytes) in &fixture.attached.slots {
let restored = slots.get(&id).ok_or("ordinary restored slot absent")?;
assert!(restored.binding_fate.is_some());
assert_eq!(&restored.encode_checkpoint()?, bytes);
}
Ok(())
}
fn ordinary_pending_round_trip(fixture: &OrdinaryFixture) -> TestResult {
for (&id, fate) in &fixture.pending.pending_specific_fates {
if let Some(terminal) = fate.terminal {
assert_eq!(
fixture
.pending
.slots
.get(&id)
.ok_or("ordinary pending member absent")?
.member
.latest_terminal(),
Some(liminal_protocol::lifecycle::CommittedBindingTerminal::Died(
terminal.terminal
))
);
}
}
let bytes = super::checkpoint_specific_fates::encode(
&fixture.pending.pending_specific_fates,
&fixture.pending.slots,
)?;
let slot_bytes = fixture
.pending
.slots
.iter()
.map(|(&id, slot)| Ok((id, slot.encode_checkpoint()?)))
.collect::<Result<_, super::super::restore_checkpoint::CheckpointError>>()?;
let pending_snapshot = capture_component_state(&fixture.pending)?;
let (slots, origins) = restore_component_slots(&pending_snapshot, &slot_bytes)?;
let restored = super::checkpoint_specific_fates::decode(
&bytes,
&slots,
&origins,
&fixture.pending_positions,
&mut super::checkpoint_owner::CheckpointFencedSources::new(),
)?;
assert_eq!(
super::checkpoint_specific_fates::encode(&restored, &slots)?,
bytes
);
assert_eq!(restored.len(), fixture.pending.pending_specific_fates.len());
for (&id, original) in &fixture.pending.pending_specific_fates {
let restored = restored
.get(&id)
.ok_or("ordinary restored pending fate absent")?;
assert_eq!(restored.died_source_sequence, original.died_source_sequence);
assert_eq!(restored.intent, original.intent);
assert_eq!(restored.binding_fate, original.binding_fate);
assert_eq!(
restored.terminal.map(|value| value.terminal),
original.terminal.map(|value| value.terminal)
);
}
Ok(())
}
fn ordinary_prepared_round_trip(fixture: &OrdinaryFixture) -> TestResult {
let bytes =
super::checkpoint_finalizers::encode(&fixture.prepared.prepared_ordinary_finalizers)?;
let (_, origins) = restore_component_slots(&fixture.attached, &fixture.attached.slots)?;
let restored =
super::checkpoint_finalizers::decode(&bytes, &origins, &fixture.pending_positions)?;
assert_eq!(super::checkpoint_finalizers::encode(&restored)?, bytes);
assert_eq!(
restored.len(),
fixture.prepared.prepared_ordinary_finalizers.len()
);
for (&id, original) in &fixture.prepared.prepared_ordinary_finalizers {
let restored = restored
.get(&id)
.ok_or("ordinary restored prepared finalizer absent")?;
assert_eq!(
restored.attached_source_sequence,
original.attached_source_sequence
);
assert_eq!(restored.terminal, original.terminal);
assert_eq!(restored.fate, original.fate);
assert_eq!(restored.finalizer, original.finalizer);
assert_eq!(
serde_json::to_vec(&restored.terminal_source)?,
serde_json::to_vec(&original.terminal_source)?
);
}
Ok(())
}