use std::error::Error;
use std::sync::Arc;
use liminal::durability::{DurableStore, bridge::block_on, open_ephemeral};
use liminal_protocol::wire::{
ClientRequest, ConnectionIncarnation, EnrollmentRequest, EnrollmentToken, Generation,
ParticipantAck, ServerValue,
};
use super::super::ProductionParticipantHandler;
use super::super::log::OperationLog;
use super::super::outbox::ConversationOutboxLimits;
use super::super::outbox_log::OutboxLog;
use super::super::restore_checkpoint::{
CheckpointLiveState, CheckpointPositions, CheckpointSchema,
};
use super::super::restore_checkpoint_history::CheckpointHistoryRoot;
use super::super::restore_checkpoint_history_capture::capture_identity_history;
use super::super::tests::{dispatch_tracked, test_participant_config};
use super::ConversationAuthority;
use super::checkpoint_body::CheckpointOwnerBody;
use super::checkpoint_owner::ConversationAuthorityRestoreContext;
use crate::config::types::ParticipantConfig;
use crate::server::participant::ParticipantConnectionConversations;
type TestResult<T = ()> = Result<T, Box<dyn Error>>;
struct Fixture {
owner: ConversationAuthority,
positions: CheckpointPositions,
root: Vec<u8>,
config: ParticipantConfig,
}
struct FrozenFixture {
held: Result<Vec<u8>, String>,
pruned: Result<Vec<u8>, String>,
positions: CheckpointPositions,
root: Vec<u8>,
config: ParticipantConfig,
}
#[derive(serde::Serialize, serde::Deserialize)]
struct RedFixture {
held: Vec<u8>,
pruned: Vec<u8>,
positions: [u64; 12],
schema: u8,
root: Vec<u8>,
}
fn freeze_fixture() -> TestResult<FrozenFixture> {
let mut fixture = fixture()?;
assert_eq!(fixture.owner.observer_progress_witnesses().len(), 1);
let held = fixture
.owner
.encode_live()
.map_err(|error| format!("{error:?}"));
fixture.owner.prune_durable_observer_witnesses()?;
assert_eq!(fixture.owner.observer_progress_witnesses().len(), 0);
assert_eq!(
fixture.owner.observer_progress_pruned_through(),
fixture.positions.observer_progress
);
let pruned = fixture
.owner
.encode_live()
.map_err(|error| format!("{error:?}"));
let frozen = FrozenFixture {
held,
pruned,
positions: fixture.positions,
root: fixture.root,
config: fixture.config,
};
if let Some(path) = std::env::var_os("LIMINAL_OBSERVER_RED_FIXTURE_EXPORT") {
std::fs::write(path, serde_json::to_vec(&red_fixture(&frozen)?)?)?;
}
Ok(frozen)
}
#[test]
fn checkpoint_observer_held_and_pruned_capture_restore_keep_native_sources() -> TestResult {
let fixture = freeze_fixture()?;
let imported = std::env::var_os("LIMINAL_OBSERVER_RED_FIXTURE_IMPORT")
.map(|path| imported_fixture(&std::fs::read(path)?))
.transpose()?;
let restore = imported.as_ref().unwrap_or(&fixture);
let controls = [
(
"held witness capture",
fixture
.held
.as_ref()
.map_err(|error| error.clone().into())
.and_then(|bytes| capture_control(bytes, "held witness capture", 1)),
),
(
"pruned bookkeeping capture",
fixture
.pruned
.as_ref()
.map_err(|error| error.clone().into())
.and_then(|bytes| capture_control(bytes, "pruned bookkeeping capture", 0)),
),
(
"held witness restore",
restore
.held
.as_ref()
.map_err(|error| error.clone().into())
.and_then(|bytes| restore_control(restore, bytes, "held witness restore", 1)),
),
(
"pruned bookkeeping restore",
restore
.pruned
.as_ref()
.map_err(|error| error.clone().into())
.and_then(|bytes| restore_control(restore, bytes, "pruned bookkeeping restore", 0)),
),
];
let mut failures = Vec::new();
for (name, result) in controls {
if let Err(error) = result {
println!("observer_checkpoint_control {name} FAILED: {error:?}");
failures.push(format!("{name}: {error:?}"));
} else {
println!("observer_checkpoint_control {name} passed");
}
}
println!(
"observer_checkpoint_controls total=4 passed={} failed={}",
4 - failures.len(),
failures.len()
);
if !failures.is_empty() {
return Err(failures.join("; ").into());
}
let held = fixture.held.as_ref().map_err(String::as_str)?;
let pruned = fixture.pruned.as_ref().map_err(String::as_str)?;
assert_ne!(
CheckpointOwnerBody::decode_canonical(held)?.observer_progress_witnesses,
CheckpointOwnerBody::decode_canonical(pruned)?.observer_progress_witnesses
);
refusal_controls(&fixture, pruned)
}
fn fixture() -> TestResult<Fixture> {
let config = test_participant_config();
let conversation_id = 59;
let incarnation = ConnectionIncarnation::new(89, 7);
let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
let handler = ProductionParticipantHandler::new(Arc::clone(&store), config)?;
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("observer fixture enrollment refused".into());
};
let peer = dispatch_tracked(
&handler,
ConnectionIncarnation::new(89, 8),
&mut ParticipantConnectionConversations::default(),
ClientRequest::Enrollment(EnrollmentRequest {
conversation_id,
enrollment_token: EnrollmentToken::new([62; 16]),
}),
)?;
assert!(
matches!(peer, ServerValue::EnrollBound(_)),
"observer fixture peer enrollment refused: {peer:?}"
);
let peer_attached_seq = 2;
let acknowledged = dispatch_tracked(
&handler,
incarnation,
&mut conversations,
ClientRequest::ParticipantAck(ParticipantAck {
conversation_id,
participant_id: receipt.participant_id(),
capability_generation: Generation::ONE,
through_seq: peer_attached_seq,
}),
)?;
assert!(
matches!(acknowledged, ServerValue::AckCommitted(_)),
"observer fixture ack did not commit: {acknowledged:?}"
);
let log = OperationLog::new(Arc::clone(&store), conversation_id);
let outbox_log = OutboxLog::new(Arc::clone(&store), conversation_id);
let limits =
ConversationOutboxLimits::try_new(config.max_retained_record_rows, config.identity_slots)?;
let (mut owner, audit, schema, _) = block_on(ConversationAuthority::replay_for_rebuild(
conversation_id,
&log,
&outbox_log,
&config,
limits,
))??;
let history = block_on(capture_identity_history(
store.as_ref(),
&owner,
CheckpointHistoryRoot::empty(conversation_id),
))??;
owner.retain_checkpoint_metadata(history.root, &audit)?;
let extension = owner
.outbox
.as_ref()
.ok_or("observer fixture outbox absent")?
.next_extension_sequence();
let positions = owner.checkpoint_positions(schema, owner.next_log_sequence, extension)?;
assert_eq!(positions.observer_progress, peer_attached_seq);
Ok(Fixture {
owner,
positions,
root: history.root.encode_canonical().to_vec(),
config,
})
}
fn capture_control(encoded: &[u8], name: &str, held: usize) -> TestResult {
let section = CheckpointOwnerBody::decode_canonical(encoded)?.observer_progress_witnesses;
assert!(!section.is_empty(), "{name} omitted observer bookkeeping");
let state: serde_json::Value = serde_json::from_slice(section)?;
assert_eq!(
state["witnesses"]
.as_array()
.ok_or("witness list absent")?
.len(),
held
);
let pruned = usize::from(held == 0);
assert_eq!(
state["pruned_sources"]
.as_array()
.ok_or("pruned source list absent")?
.len(),
pruned
);
for source in state["witnesses"]
.as_array()
.ok_or("witness list absent")?
.iter()
.chain(
state["pruned_sources"]
.as_array()
.ok_or("pruned source list absent")?,
)
{
assert!(
!source["source"]
.as_array()
.ok_or("native source absent")?
.is_empty()
);
}
println!("{name}: held={held} pruned_native_sources={pruned}");
Ok(())
}
fn restore_control(fixture: &FrozenFixture, encoded: &[u8], name: &str, held: usize) -> TestResult {
let context = ConversationAuthorityRestoreContext::new(
fixture.positions,
&fixture.root,
&fixture.config,
)?;
let restored =
ConversationAuthority::decode_live(encoded, &fixture.positions, &fixture.root, &context)?;
assert_eq!(restored.observer_progress_witnesses().len(), held);
let pruned_through = if held == 0 {
fixture.positions.observer_progress
} else {
0
};
assert_eq!(restored.observer_progress_pruned_through(), pruned_through);
assert_eq!(
restored.encode_live()?,
encoded,
"{name} changed saved owner bytes"
);
println!("{name}: exact_owner_bytes={} held={held}", encoded.len());
Ok(())
}
fn refusal_controls(fixture: &FrozenFixture, encoded: &[u8]) -> TestResult {
let mut body = CheckpointOwnerBody::decode_canonical(encoded)?;
let section: serde_json::Value = serde_json::from_slice(body.observer_progress_witnesses)?;
let actual_source = serde_json::to_string(§ion["pruned_sources"][0]["source"])?;
let field = format!("\"source\":{actual_source}");
let section = std::str::from_utf8(body.observer_progress_witnesses)?;
assert!(section.contains(&field));
let missing = section.replacen(&field, "\"source\":[]", 1).into_bytes();
body.observer_progress_witnesses = &missing;
let missing = body.encode_canonical()?;
let context = ConversationAuthorityRestoreContext::new(
fixture.positions,
&fixture.root,
&fixture.config,
)?;
assert!(
ConversationAuthority::decode_live(&missing, &fixture.positions, &fixture.root, &context)
.is_err()
);
body.observer_progress_witnesses = &[];
let omitted = body.encode_canonical()?;
assert!(
ConversationAuthority::decode_live(&omitted, &fixture.positions, &fixture.root, &context)
.is_err()
);
Ok(())
}
fn red_fixture(fixture: &FrozenFixture) -> TestResult<RedFixture> {
let p = fixture.positions;
Ok(RedFixture {
held: fixture.held.as_ref().map_err(String::as_str)?.clone(),
pruned: fixture.pruned.as_ref().map_err(String::as_str)?.clone(),
positions: [
p.conversation_id,
p.covered_log_sequence,
p.next_event_ordinal,
p.next_log_sequence,
p.next_seq,
p.next_order,
p.next_participant,
p.extension_sequence,
p.highest_delivery_seq,
p.observer_progress,
p.merged_base_sequence,
p.merged_extension_sequence,
],
schema: match p.schema {
CheckpointSchema::V2Prefix => 2,
CheckpointSchema::V3Suffix => 3,
CheckpointSchema::V4Suffix => 4,
},
root: fixture.root.clone(),
})
}
fn imported_fixture(bytes: &[u8]) -> TestResult<FrozenFixture> {
let saved: RedFixture = serde_json::from_slice(bytes)?;
let p = saved.positions;
let schema = match saved.schema {
2 => CheckpointSchema::V2Prefix,
3 => CheckpointSchema::V3Suffix,
4 => CheckpointSchema::V4Suffix,
_ => return Err("observer red fixture schema is invalid".into()),
};
let positions = CheckpointPositions {
conversation_id: p[0],
covered_log_sequence: p[1],
next_event_ordinal: p[2],
next_log_sequence: p[3],
next_seq: p[4],
next_order: p[5],
next_participant: p[6],
extension_sequence: p[7],
highest_delivery_seq: p[8],
observer_progress: p[9],
merged_base_sequence: p[10],
merged_extension_sequence: p[11],
schema,
};
positions.validate()?;
Ok(FrozenFixture {
held: Ok(saved.held),
pruned: Ok(saved.pruned),
positions,
root: saved.root,
config: test_participant_config(),
})
}