use std::error::Error;
use std::io;
use std::sync::OnceLock;
use std::sync::atomic::{AtomicU64, Ordering};
use super::ProductionParticipantHandler;
use super::tests::{dispatch, test_participant_config};
use crate::server::mount::MountKind;
use crate::server::participant::transport::{
ParticipantSession, encode_server_value, normalize_configured_frame_limit,
};
use crate::server::participant::{
ParticipantConnectionContext, ParticipantConnectionConversations, ParticipantDispatch,
dispatch_generic_frame,
};
use liminal::durability::{CountingStore, StoreCounts, open_ephemeral};
use liminal::protocol::Frame;
use liminal_protocol::wire::{
ClientDiscriminant, ClientRequest, CodecError, ConnectionIncarnation, DecodeClass,
EnrollmentRequest, EnrollmentToken, Generation, PARTICIPANT_FRAME_TYPE, ParticipantFrame,
PushDiscriminant, ReceiverDirection, RecordAdmission, RecordAdmissionAttemptToken,
ServerDiscriminant, ServerValue, SettlementAnswer, SettlementOutcome, SettlementQuery, decode,
encode, encoded_len,
};
type TestResult = Result<(), Box<dyn Error>>;
type Snapshot = (u64, u64, u64, usize);
static HANDLER: OnceLock<Result<ProductionParticipantHandler, String>> = OnceLock::new();
static NEXT_CONVERSATION: AtomicU64 = AtomicU64::new(50_000);
fn handler() -> Result<&'static ProductionParticipantHandler, Box<dyn Error>> {
HANDLER
.get_or_init(|| {
let store = CountingStore::new(open_ephemeral(1).map_err(|error| error.to_string())?);
ProductionParticipantHandler::new(std::sync::Arc::new(store), test_participant_config())
.map_err(|error| error.to_string())
})
.as_ref()
.map_err(|error| io::Error::other(error.clone()).into())
}
fn fixture() -> Result<(u64, u64, ConnectionIncarnation), Box<dyn Error>> {
let conversation = NEXT_CONVERSATION.fetch_add(1, Ordering::Relaxed);
let connection = ConnectionIncarnation::new(conversation, 1);
let value = dispatch(
handler()?,
connection,
ClientRequest::Enrollment(EnrollmentRequest {
conversation_id: conversation,
enrollment_token: EnrollmentToken::new([0x51; 16]),
}),
)?;
let liminal_protocol::wire::ServerValue::EnrollBound(receipt) = value else {
return Err("fixture enrollment must bind".into());
};
let counts = store_counts()?;
println!("enrollment positive-control counts: {counts:?}");
assert!(
counts.append_calls + counts.append_all_calls > 0,
"counting fixture must observe enrollment writes"
);
assert!(
counts.flushes > 0,
"counting fixture must observe enrollment flushes"
);
Ok((conversation, receipt.participant_id(), connection))
}
fn raw_query(conversation: u64, participant: u64, generation: Generation) -> Frame {
let mut payload = vec![0, 1, 0, 0, 0, 9];
payload.extend_from_slice(&conversation.to_be_bytes());
payload.extend_from_slice(&participant.to_be_bytes());
payload.extend_from_slice(&generation.get().to_be_bytes());
payload.extend_from_slice(&[0x52; 16]);
payload.extend_from_slice(&[0x53; 32]);
Frame::Unknown {
type_id: PARTICIPANT_FRAME_TYPE,
flags: 0,
stream_id: 0,
payload,
}
}
fn complete(frame: &Frame) -> Result<Vec<u8>, Box<dyn Error>> {
let mut bytes = vec![0; liminal::protocol::encoded_len(frame)?];
liminal::protocol::encode(frame, &mut bytes)?;
Ok(bytes)
}
fn query(
conversation: u64,
participant: u64,
generation: Generation,
connection: ConnectionIncarnation,
) -> Result<ParticipantDispatch, Box<dyn Error>> {
let mut session = ParticipantSession::default();
session.negotiate_v1(
normalize_configured_frame_limit(test_participant_config().wire_frame_limit)
.map_err(|error| format!("{error:?}"))?,
);
let before = store_counts()?;
let value = dispatch_generic_frame(
&raw_query(conversation, participant, generation),
true,
session,
ParticipantConnectionContext::new(connection, MountKind::Tcp),
None,
&mut ParticipantConnectionConversations::default(),
handler()?,
);
let delta = store_counts()?.since(&before);
println!("query counts conversation={conversation} participant={participant}: {delta:?}");
assert_eq!(delta, StoreCounts::default());
Ok(value)
}
fn store_counts() -> Result<StoreCounts, Box<dyn Error>> {
Ok(handler()?
.store
.store_counts()
.ok_or("counted fixture is required")?
.snapshot())
}
fn read_only_dispatch(
connection: ConnectionIncarnation,
request: SettlementQuery,
) -> Result<ServerValue, Box<dyn Error>> {
let conversation = request.conversation_id;
let head = snapshot(conversation)?;
let before = store_counts()?;
let value = dispatch(
handler()?,
connection,
ClientRequest::SettlementQuery(request),
)?;
let delta = store_counts()?.since(&before);
println!("query counts conversation={conversation}: {delta:?}");
assert_eq!(delta, StoreCounts::default());
assert_eq!(snapshot(conversation)?, head);
Ok(value)
}
fn snapshot(conversation: u64) -> Result<Option<Snapshot>, Box<dyn Error>> {
let Some(cell) = handler()?.existing_cell(conversation)? else {
return Ok(None);
};
let owner = cell
.lock()
.map_err(|_| io::Error::other("fixture owner poisoned"))?;
let snapshot = owner.as_ref().map(|value| {
(
value.next_log_sequence,
value.next_order,
value.next_seq,
value.committed_admissions.len(),
)
});
drop(owner);
if let Some((head, ..)) = snapshot {
let log =
super::log::OperationLog::new(std::sync::Arc::clone(&handler()?.store), conversation);
assert!(liminal::durability::bridge::block_on(log.read_at(head))??.is_none());
}
Ok(snapshot)
}
#[test]
fn settlement_query_canonical_request_round_trip() -> TestResult {
let bytes = complete(&raw_query(7, 8, Generation::ONE))?;
let value = decode(&bytes, ReceiverDirection::Server).map_err(|error| format!("{error:?}"))?;
assert_eq!(
value,
ParticipantFrame::ClientRequest(ClientRequest::SettlementQuery(SettlementQuery {
conversation_id: 7,
participant_id: 8,
capability_generation: Generation::ONE,
record_admission_attempt_token: RecordAdmissionAttemptToken::new([0x52; 16]),
payload_fingerprint: [0x53; 32],
}))
);
let mut encoded = vec![0; encoded_len(&value).map_err(|error| format!("{error:?}"))?];
encode(&value, &mut encoded).map_err(|error| format!("{error:?}"))?;
assert_eq!(encoded, bytes);
for (tag, sequence) in [(1, Some(9_u64)), (2, None), (3, None), (4, None), (5, None)] {
let mut payload = vec![0, 1, 0, 0, 1, 41, 0, 9];
payload.extend_from_slice(&7_u64.to_be_bytes());
payload.extend_from_slice(&8_u64.to_be_bytes());
payload.extend_from_slice(&[0x52; 16]);
payload.push(tag);
if let Some(sequence) = sequence {
payload.extend_from_slice(&sequence.to_be_bytes());
}
let frame = Frame::Unknown {
type_id: PARTICIPANT_FRAME_TYPE,
flags: 0,
stream_id: 0,
payload,
};
let bytes = complete(&frame)?;
let value =
decode(&bytes, ReceiverDirection::Client).map_err(|error| format!("{error:?}"))?;
let outcome = match tag {
1 => SettlementOutcome::Committed { delivery_seq: 9 },
2 => SettlementOutcome::Withdrawn,
3 => SettlementOutcome::Expired,
4 => SettlementOutcome::NotCommitted,
5 => SettlementOutcome::Unknown,
_ => return Err("fixture outcome is unassigned".into()),
};
assert_eq!(
value,
ParticipantFrame::ServerValue(ServerValue::SettlementAnswer(SettlementAnswer {
conversation_id: 7,
participant_id: 8,
record_admission_attempt_token: RecordAdmissionAttemptToken::new([0x52; 16]),
outcome,
}))
);
let mut encoded = vec![0; encoded_len(&value).map_err(|error| format!("{error:?}"))?];
encode(&value, &mut encoded).map_err(|error| format!("{error:?}"))?;
assert_eq!(encoded, bytes);
}
Ok(())
}
#[test]
fn settlement_query_registry_allocations_are_unique() {
assert_eq!(
ClientDiscriminant::try_from(9).map(ClientDiscriminant::wire_value),
Ok(9)
);
assert_eq!(
ServerDiscriminant::try_from(0x0129).map(ServerDiscriminant::wire_value),
Ok(0x0129)
);
let mut counts = [0_usize; 3];
for tag in 0..=u16::MAX {
let assigned = [
ClientDiscriminant::try_from(tag).is_ok(),
ServerDiscriminant::try_from(tag).is_ok(),
PushDiscriminant::try_from(tag).is_ok(),
];
assert!(
assigned.iter().filter(|assigned| **assigned).count() <= 1,
"direction registries collide at {tag:#06x}"
);
for (count, assigned) in counts.iter_mut().zip(assigned) {
*count += usize::from(assigned);
}
}
assert_eq!(counts, [9, 42, 3]);
}
#[test]
fn settlement_query_semantic_matrix_is_read_only() -> TestResult {
let (conversation, participant, connection) = fixture()?;
let before = snapshot(conversation)?;
let value = query(conversation, participant, Generation::ONE, connection)?;
assert!(
matches!(value, ParticipantDispatch::Respond(_)),
"an authorized settlement query must answer without a decode close"
);
let ParticipantDispatch::Respond(frame) = value else {
return Err("query must answer".into());
};
let bytes = complete(&frame)?;
assert_eq!(&bytes[14..16], &[1, 41]);
assert_eq!(bytes.last(), Some(&4));
assert_eq!(snapshot(conversation)?, before);
authority_refusals(conversation, participant, connection)?;
let payload = b"settlement fixture payload".to_vec();
let fingerprint = super::facts::ordinary_payload_fingerprint(&payload);
let token = RecordAdmissionAttemptToken::new([0x52; 16]);
let committed = dispatch(
handler()?,
connection,
ClientRequest::RecordAdmission(RecordAdmission {
conversation_id: conversation,
participant_id: participant,
capability_generation: Generation::ONE,
record_admission_attempt_token: token,
payload,
}),
)?;
let expected = encode_server_value(committed).map_err(|error| format!("{error:?}"))?;
let Frame::Unknown { payload, .. } = expected else {
return Err("commit frame must be participant".into());
};
let sequence = payload
.get(payload.len().saturating_sub(8)..)
.ok_or("commit sequence missing")?
.to_vec();
let before = snapshot(conversation)?;
let request = liminal_protocol::wire::SettlementQuery {
conversation_id: conversation,
participant_id: participant,
capability_generation: Generation::ONE,
record_admission_attempt_token: token,
payload_fingerprint: fingerprint,
};
let value = read_only_dispatch(connection, request.clone())?;
let liminal_protocol::wire::ServerValue::SettlementAnswer(answer) = value else {
return Err("exact identity must yield settlement answer".into());
};
let liminal_protocol::wire::SettlementOutcome::Committed { delivery_seq } = answer.outcome
else {
return Err("exact committed identity must be committed".into());
};
assert_eq!(delivery_seq.to_be_bytes().as_slice(), sequence);
assert_eq!(snapshot(conversation)?, before);
let mut conflict = request.clone();
conflict.payload_fingerprint = [0x53; 32];
let value = read_only_dispatch(connection, conflict)?;
assert!(matches!(
value,
liminal_protocol::wire::ServerValue::AttemptTokenBodyConflict(_)
));
assert_eq!(snapshot(conversation)?, before);
foreign_presenter(conversation, &request)?;
unknown_owners(conversation, participant, connection, request)
}
fn foreign_presenter(conversation: u64, original: &SettlementQuery) -> TestResult {
let connection = ConnectionIncarnation::new(conversation, 3);
let enrolled = dispatch(
handler()?,
connection,
ClientRequest::Enrollment(EnrollmentRequest {
conversation_id: conversation,
enrollment_token: EnrollmentToken::new([0x61; 16]),
}),
)?;
let ServerValue::EnrollBound(enrolled) = enrolled else {
return Err("second presenter must bind".into());
};
assert_ne!(enrolled.participant_id(), original.participant_id);
let before = snapshot(conversation)?;
for fingerprint in [original.payload_fingerprint, [0x53; 32]] {
let mut request = original.clone();
request.participant_id = enrolled.participant_id();
request.payload_fingerprint = fingerprint;
let value = read_only_dispatch(connection, request)?;
assert_eq!(
value,
ServerValue::SettlementAnswer(SettlementAnswer {
conversation_id: conversation,
participant_id: enrolled.participant_id(),
record_admission_attempt_token: original.record_admission_attempt_token,
outcome: SettlementOutcome::NotCommitted,
})
);
assert_eq!(snapshot(conversation)?, before);
}
Ok(())
}
fn unknown_owners(
conversation: u64,
participant: u64,
connection: ConnectionIncarnation,
request: liminal_protocol::wire::SettlementQuery,
) -> TestResult {
let missing = NEXT_CONVERSATION.fetch_add(1, Ordering::Relaxed);
assert!(handler()?.existing_cell(missing)?.is_none());
let mut absent = request;
absent.conversation_id = missing;
let value = read_only_dispatch(connection, absent)?;
assert!(matches!(
value,
liminal_protocol::wire::ServerValue::SettlementAnswer(
liminal_protocol::wire::SettlementAnswer {
outcome: liminal_protocol::wire::SettlementOutcome::Unknown,
..
}
)
));
assert!(handler()?.existing_cell(missing)?.is_none());
let cell = handler()?
.existing_cell(conversation)?
.ok_or("existing owner must remain")?;
*cell
.lock()
.map_err(|_| io::Error::other("fixture owner poisoned"))? = None;
let value = query(conversation, participant, Generation::ONE, connection)?;
let ParticipantDispatch::Respond(frame) = value else {
return Err("discarded owner must answer".into());
};
assert_eq!(complete(&frame)?.last(), Some(&5));
assert!(snapshot(conversation)?.is_none());
Ok(())
}
fn authority_refusals(
conversation: u64,
participant: u64,
bound: ConnectionIncarnation,
) -> TestResult {
let stale = Generation::new(2).ok_or("stale generation must be nonzero")?;
for (generation, connection) in [
(stale, bound),
(Generation::ONE, ConnectionIncarnation::new(conversation, 2)),
] {
let before = snapshot(conversation)?;
let admission = dispatch(
handler()?,
connection,
ClientRequest::RecordAdmission(RecordAdmission {
conversation_id: conversation,
participant_id: participant,
capability_generation: generation,
record_admission_attempt_token: RecordAdmissionAttemptToken::new([0x52; 16]),
payload: Vec::new(),
}),
)?;
let expected = encode_server_value(admission).map_err(|error| format!("{error:?}"))?;
let actual = query(conversation, participant, generation, connection)?;
assert!(
matches!(actual, ParticipantDispatch::Respond(_)),
"query authority must be classified rather than closed as an unknown frame"
);
let ParticipantDispatch::Respond(actual) = actual else {
return Err("query must answer".into());
};
assert_eq!(actual, expected);
assert_eq!(snapshot(conversation)?, before);
}
Ok(())
}
fn assert_decode_class(
frame: &Frame,
receiver: ReceiverDirection,
class: DecodeClass,
) -> TestResult {
assert_eq!(
decode(&complete(frame)?, receiver),
Err(CodecError::Decode { class })
);
Ok(())
}
fn answer_frame(outcome: u8, sequence: Option<u64>) -> Frame {
let mut payload = vec![0, 1, 0, 0, 1, 41, 0, 9];
payload.extend_from_slice(&7_u64.to_be_bytes());
payload.extend_from_slice(&8_u64.to_be_bytes());
payload.extend_from_slice(&[0x52; 16]);
payload.push(outcome);
if let Some(sequence) = sequence {
payload.extend_from_slice(&sequence.to_be_bytes());
}
Frame::Unknown {
type_id: PARTICIPANT_FRAME_TYPE,
flags: 0,
stream_id: 0,
payload,
}
}
#[test]
fn settlement_query_rejects_noncanonical_and_incomplete_boundaries() -> TestResult {
let frame = raw_query(7, 8, Generation::ONE);
assert_decode_class(
&frame,
ReceiverDirection::Client,
DecodeClass::UnknownDiscriminant,
)?;
let Frame::Unknown { payload, .. } = &frame else {
return Err("fixture frame must be participant".into());
};
for length in 6..payload.len() {
let mut short = frame.clone();
let Frame::Unknown { payload, .. } = &mut short else {
return Err("fixture frame must be participant".into());
};
payload.truncate(length);
assert_decode_class(
&short,
ReceiverDirection::Server,
DecodeClass::MissingRequiredField,
)?;
}
let mut trailing = frame.clone();
let Frame::Unknown { payload, .. } = &mut trailing else {
return Err("fixture frame must be participant".into());
};
payload.push(0);
assert_decode_class(
&trailing,
ReceiverDirection::Server,
DecodeClass::CanonicalEncoding,
)?;
let mut zero_generation = frame;
let Frame::Unknown { payload, .. } = &mut zero_generation else {
return Err("fixture frame must be participant".into());
};
payload[22..30].fill(0);
assert_decode_class(
&zero_generation,
ReceiverDirection::Server,
DecodeClass::InvalidField,
)?;
answer_boundaries()
}
fn answer_boundaries() -> TestResult {
for outcome in [0, 6, u8::MAX] {
assert_decode_class(
&answer_frame(outcome, None),
ReceiverDirection::Client,
DecodeClass::InvalidField,
)?;
}
let frame = answer_frame(1, Some(9));
assert_decode_class(
&frame,
ReceiverDirection::Server,
DecodeClass::UnknownDiscriminant,
)?;
let Frame::Unknown { payload, .. } = &frame else {
return Err("fixture frame must be participant".into());
};
for length in 6..payload.len() {
let mut short = frame.clone();
let Frame::Unknown { payload, .. } = &mut short else {
return Err("fixture frame must be participant".into());
};
payload.truncate(length);
assert_decode_class(
&short,
ReceiverDirection::Client,
DecodeClass::MissingRequiredField,
)?;
}
let mut wrong_origin = frame;
let Frame::Unknown { payload, .. } = &mut wrong_origin else {
return Err("fixture frame must be participant".into());
};
payload[6..8].copy_from_slice(&7_u16.to_be_bytes());
assert_decode_class(
&wrong_origin,
ReceiverDirection::Client,
DecodeClass::InvalidField,
)?;
for outcome in 1..=5 {
let mut trailing = answer_frame(outcome, (outcome == 1).then_some(9));
let Frame::Unknown { payload, .. } = &mut trailing else {
return Err("fixture frame must be participant".into());
};
payload.push(0);
assert_decode_class(
&trailing,
ReceiverDirection::Client,
DecodeClass::CanonicalEncoding,
)?;
}
Ok(())
}