use std::collections::BTreeSet;
use std::error::Error;
use std::sync::Arc;
use liminal::durability::{CountingStore, DurableStore, bridge::block_on};
use liminal_protocol::wire::{
ClientRequest, ConnectionIncarnation, EnrollmentRequest, EnrollmentToken, Generation,
RecordAdmission, RecordAdmissionAttemptToken, ServerValue,
};
use super::ProductionParticipantHandler;
use super::log::STREAM_PREFIX;
use super::restore_checkpoint_locator::{CheckpointLocator, checkpoint_conversation_route};
use super::routed_stream::read_head;
use super::tests::{dispatch, test_participant_config};
use super::tests_memory_store::MemoryStore;
#[test]
fn a_serving_conversation_crossing_the_checkpoint_boundary_uses_atomic_batches()
-> Result<(), Box<dyn Error>> {
let ServingFixture {
store,
handler,
connection,
participant_id,
} = serving_fixture()?;
let route = checkpoint_conversation_route(10);
let stream = format!("{STREAM_PREFIX}10");
let mut boundaries = BTreeSet::new();
let before = store.snapshot();
for operation in 0_u64..257 {
let mut token = [0; 16];
token[..8].copy_from_slice(&operation.to_be_bytes());
let result = dispatch(
&handler,
connection,
ClientRequest::RecordAdmission(RecordAdmission {
conversation_id: 10,
participant_id,
capability_generation: Generation::ONE,
record_admission_attempt_token: RecordAdmissionAttemptToken::new(token),
payload: operation.to_be_bytes().to_vec(),
}),
)?;
assert!(
matches!(result, ServerValue::RecordCommitted(_)),
"{result:?}"
);
let head = block_on(read_head(store.as_ref(), &route, &stream))??
.ok_or("serving operation omitted its routed head")?;
if [255, 256, 257].contains(&head.next_sequence) {
assert_checkpoint_boundary(store.as_ref(), head.next_sequence)?;
boundaries.insert(head.next_sequence);
}
}
assert_eq!(boundaries, BTreeSet::from([255, 256, 257]));
let counts = store.snapshot().since(&before);
assert_eq!(
counts.apply_routed_batch_calls, 257,
"every logical operation must commit once, including the checkpoint crossing"
);
assert_eq!(counts.append_calls, 0, "serving writes must be staged");
assert_eq!(counts.append_all_calls, 0, "serving writes must be staged");
assert_eq!(counts.flushes, 0, "the atomic batch owns durability");
Ok(())
}
struct ServingFixture {
store: Arc<CountingStore<MemoryStore>>,
handler: ProductionParticipantHandler,
connection: ConnectionIncarnation,
participant_id: u64,
}
fn serving_fixture() -> Result<ServingFixture, Box<dyn Error>> {
let store = Arc::new(CountingStore::new(MemoryStore::default()));
let handler = ProductionParticipantHandler::new(
Arc::clone(&store) as Arc<dyn DurableStore>,
test_participant_config(),
)?;
let connection = ConnectionIncarnation::new(10, 1);
let enrolled = dispatch(
&handler,
connection,
ClientRequest::Enrollment(EnrollmentRequest {
conversation_id: 10,
enrollment_token: EnrollmentToken::new([1; 16]),
}),
)?;
let ServerValue::EnrollBound(bound) = enrolled else {
return Err(format!("enrollment did not bind: {enrolled:?}").into());
};
let recipient = dispatch(
&handler,
ConnectionIncarnation::new(10, 2),
ClientRequest::Enrollment(EnrollmentRequest {
conversation_id: 10,
enrollment_token: EnrollmentToken::new([2; 16]),
}),
)?;
assert!(matches!(recipient, ServerValue::EnrollBound(_)));
Ok(ServingFixture {
store,
handler,
connection,
participant_id: bound.participant_id(),
})
}
fn assert_checkpoint_boundary(store: &dyn DurableStore, next: u64) -> Result<(), Box<dyn Error>> {
let route = checkpoint_conversation_route(10);
let bytes = block_on(store.get_routed(&route, CheckpointLocator::routed_key()))??;
if next == 255 {
assert!(bytes.is_none(), "a checkpoint preceded the first crossing");
} else {
let bytes = bytes.ok_or("checkpoint locator is absent at the boundary")?;
let locator = CheckpointLocator::decode_canonical(&bytes, 10)?;
assert_eq!(locator.covered_sequence(), 255);
let body = block_on(store.get_routed(&route, &locator.body_key().routed_key()))??
.ok_or("checkpoint body is absent at the boundary")?;
locator.body_key().verify_body(&body)?;
}
Ok(())
}
#[test]
fn a_real_last_identity_seal_is_covered_by_its_checkpoint_before_reopen()
-> Result<(), Box<dyn Error>> {
let fixture = super::tests_f8b_seal::last_identity_store()?;
let store = Arc::clone(&fixture.store);
let conversation_id = fixture.conversation_id;
drop(fixture);
let handler = ProductionParticipantHandler::new(Arc::clone(&store), seal_fixture_config())?;
assert_sealed_owner(&handler, conversation_id)?;
let route = checkpoint_conversation_route(conversation_id);
let stream = format!("{STREAM_PREFIX}{conversation_id}");
let head = block_on(read_head(store.as_ref(), &route, &stream))??
.ok_or("sealed conversation has no routed head")?;
assert!(head.next_sequence > 0);
assert!(
head.next_sequence < 256,
"fixture reached the periodic crossing"
);
let saved_locator = block_on(store.get_routed(&route, CheckpointLocator::routed_key()))??
.ok_or("seal did not publish a checkpoint locator")?;
let locator = CheckpointLocator::decode_canonical(&saved_locator, conversation_id)?;
assert_eq!(
locator.covered_sequence().checked_add(1),
Some(head.next_sequence)
);
let saved_body = block_on(store.get_routed(&route, &locator.body_key().routed_key()))??
.ok_or("seal did not publish its checkpoint body")?;
locator.body_key().verify_body(&saved_body)?;
drop(handler);
let reopened = ProductionParticipantHandler::new(Arc::clone(&store), seal_fixture_config())?;
assert_sealed_owner(&reopened, conversation_id)?;
assert_eq!(
block_on(store.get_routed(&route, CheckpointLocator::routed_key()))??,
Some(saved_locator)
);
assert_eq!(
block_on(store.get_routed(&route, &locator.body_key().routed_key()))??,
Some(saved_body)
);
Ok(())
}
fn seal_fixture_config() -> crate::config::types::ParticipantConfig {
let mut config = test_participant_config();
config.max_retained_record_rows = 4;
config
}
fn assert_sealed_owner(
handler: &ProductionParticipantHandler,
conversation_id: u64,
) -> Result<(), Box<dyn Error>> {
let cell = handler.cell(conversation_id)?;
let owner = cell
.lock()
.map_err(|_| "checkpoint seal owner lock poisoned")?;
let authority = owner.as_ref().ok_or("checkpoint seal owner is absent")?;
assert!(authority.is_closed());
assert!(authority.tokens.is_empty());
assert!(authority.frontier().is_none());
assert!(authority.slots.is_empty());
drop(owner);
Ok(())
}