use std::io;
use std::sync::Arc;
use std::thread;
use liminal_protocol::client::ClientOperationRecordRefusalReason;
use liminal_protocol::wire::{
AttachAttemptToken, AttachSecret, AuthorityStateTag, BindingStateTag, ClientDiscriminant,
ClientRequest, CredentialAttachRequest, DetachAttemptToken, DetachCommitted, DetachInProgress,
DetachRequest, DetachStaleAuthority, LeaveAttemptToken, LeaveCommitted, ProtocolVersion,
ServerDiscriminant, ServerValue, StaleAuthority, decode_server_value_body,
};
use super::support::{Action, Loopback, MemoryStore, PausedReconnectLoopback};
use super::{
CONVERSATION, PARTICIPANT, TestResult, enroll, enroll_bound, epoch, generation, recorded, sent,
};
use crate::connection::ConnectionPoolConfig;
use crate::remote::{RemoteConfig, RemoteParticipantHandle};
use crate::{ParticipantResumeStore, SdkError};
#[test]
fn all_detach_terminal_arms_apply_and_survive_lpcr_restore() -> TestResult {
let token = DetachAttemptToken::new([4; 16]);
let values = [
ServerValue::DetachCommitted(DetachCommitted::new(
CONVERSATION,
PARTICIPANT,
token,
epoch(1)?,
13,
)),
ServerValue::DetachInProgress(DetachInProgress {
conversation_id: CONVERSATION,
participant_id: PARTICIPANT,
presented_token: token,
presented_generation: generation(1)?,
committed_binding_epoch: epoch(1)?,
}),
terminalized_detach(token)?,
];
for value in values {
run_detach_terminal(value, token)?;
}
Ok(())
}
#[test]
fn durable_leave_from_real_receive_supersedes_inflight_detach() -> TestResult {
let leave = LeaveCommitted::new(
CONVERSATION,
LeaveAttemptToken::new([6; 16]),
PARTICIPANT,
generation(1)?,
Some(epoch(1)?),
None,
14,
)
.ok_or_else(|| io::Error::other("leave fixture must be internally consistent"))?;
let loopback = Loopback::spawn(vec![vec![
Action::Respond(vec![enroll_bound(CONVERSATION, [1; 16])?]),
Action::Respond(vec![ServerValue::LeaveCommitted(leave)]),
]])?;
let config = loopback.connected_config()?;
let store = MemoryStore::default();
let observed = store.clone();
let handle = RemoteParticipantHandle::new(&config, store)?;
enroll_and_receive(&handle)?;
send_detach(&handle, DetachAttemptToken::new([4; 16]))?;
let super::RemoteParticipantInbound::Refused {
value: ServerValue::LeaveCommitted(leave),
..
} = handle.receive()?
else {
return Err(io::Error::other("Leave must first preserve detach correlation").into());
};
assert_eq!(
handle.apply_leave_durable(leave)?,
super::RemoteReplayApplyOutcome::Applied
);
assert_replay_inactive(&handle)?;
let canonical = observed.bytes()?;
loopback.finish()?;
let restored = RemoteParticipantHandle::restore(&config, MemoryStore::default(), &canonical)?;
assert_replay_inactive(&restored)?;
Ok(())
}
#[test]
fn issued_tokenless_restore_reports_durable_abandonment() -> TestResult {
let loopback = Loopback::spawn(vec![vec![
Action::Respond(vec![enroll_bound(CONVERSATION, [1; 16])?]),
Action::DropAfterRequest,
]])?;
let config = loopback.connected_config()?;
let store = MemoryStore::default();
let observed = store.clone();
let handle = RemoteParticipantHandle::new(&config, store)?;
enroll_and_receive(&handle)?;
let request =
ClientRequest::ObserverRecovery(liminal_protocol::wire::ObserverRecoveryHandshake {
observer_refusals: vec![],
});
let operation = recorded(handle.record_operation(request.clone())?)?;
sent(&handle.send_operation(operation)?)?;
let canonical = observed.bytes()?;
drop(handle);
let restored = RemoteParticipantHandle::restore(&config, MemoryStore::default(), &canonical)?;
let abandonment = restored
.take_restored_operation_abandonment()?
.ok_or_else(|| {
io::Error::other("issued tokenless operation must be abandoned on restore")
})?;
assert_eq!(abandonment.request(), &request);
assert!(abandonment.was_issued());
assert!(restored.take_restored_operation_abandonment()?.is_none());
loopback.finish()?;
Ok(())
}
#[test]
fn in_progress_real_reconnect_restores_take_once_testimony() -> TestResult {
let loopback = PausedReconnectLoopback::spawn()?;
let config = loopback.connected_config()?;
let store = MemoryStore::default();
let observed = store.clone();
let handle = Arc::new(RemoteParticipantHandle::new(&config, store)?);
let super::RemoteReconnectPermitOutcome::Permitted { permit, .. } =
handle.record_explicit_reconnect()?
else {
return Err(io::Error::other("explicit event must mint reconnect authority").into());
};
let reconnect_handle = Arc::clone(&handle);
let reconnect = thread::spawn(move || reconnect_handle.reconnect(permit));
loopback.wait_until_attempt_started()?;
let canonical_attempt = observed.bytes()?;
let restored =
RemoteParticipantHandle::restore(&config, MemoryStore::default(), &canonical_attempt)?;
assert_eq!(
restored.resolve_lost_reconnect_authority()?,
super::RemoteLostReconnectResolution::Recorded {
testimony: liminal_protocol::client::LostAuthorityKind::ReconnectAttempt,
}
);
assert!(matches!(
restored.resolve_lost_reconnect_authority()?,
super::RemoteLostReconnectResolution::Refused { .. }
));
loopback.finish()?;
let reconnect = reconnect
.join()
.map_err(|_| io::Error::other("real reconnect thread panicked"))??;
assert!(matches!(
reconnect,
super::RemoteReconnectAttemptOutcome::Connected { .. }
));
Ok(())
}
#[test]
fn failed_checkpoint_withholds_successor_authority() -> TestResult {
let config = RemoteConfig::new(
"participant-checkpoint.invalid:1",
"participant-tests",
"participant-tests",
ConnectionPoolConfig::new(1, 1, 1),
)?;
let handle = RemoteParticipantHandle::new(&config, FailSecondWrite::default())?;
assert!(matches!(
handle.record_explicit_reconnect(),
Err(super::RemoteParticipantError::Storage(_))
));
assert!(matches!(
handle.record_explicit_reconnect(),
Err(super::RemoteParticipantError::StateUnavailable { .. })
));
Ok(())
}
#[test]
fn a_store_originated_unavailability_names_its_typed_cause() -> TestResult {
let config = RemoteConfig::new(
"participant-checkpoint.invalid:1",
"participant-tests",
"participant-tests",
ConnectionPoolConfig::new(1, 1, 1),
)?;
let handle = RemoteParticipantHandle::new(&config, FailSecondWrite::default())?;
assert_eq!(
handle.unavailability_cause(),
None,
"a live handle must report no unavailability cause, or a Some proves nothing later"
);
assert!(matches!(
handle.record_explicit_reconnect(),
Err(super::RemoteParticipantError::Storage(_))
));
assert!(matches!(
handle.record_explicit_reconnect(),
Err(super::RemoteParticipantError::StateUnavailable { .. })
));
let Some(cause) = handle.unavailability_cause() else {
return Err(
"#62 REPRODUCED: the handle is bricked by a store failure and cannot say so -- the \
cause was returned once and lost, leaving StateUnavailable indistinguishable from \
every other hold"
.into(),
);
};
assert_eq!(
cause,
SdkError::Store {
description: "injected checkpoint failure".to_string(),
},
"the retained cause must be the caller's own store error, by value"
);
Ok(())
}
#[test]
fn a_bricked_handle_carries_its_typed_cause_on_the_state_unavailable_error() -> TestResult {
let config = RemoteConfig::new(
"participant-checkpoint.invalid:1",
"participant-tests",
"participant-tests",
ConnectionPoolConfig::new(1, 1, 1),
)?;
let handle = RemoteParticipantHandle::new(&config, FailSecondWrite::default())?;
assert!(matches!(
handle.record_explicit_reconnect(),
Err(super::RemoteParticipantError::Storage(_))
));
let expected = SdkError::Store {
description: "injected checkpoint failure".to_string(),
};
let Err(error) = handle.record_explicit_reconnect() else {
return Err("a handle bricked by a store failure must refuse every later call".into());
};
let super::RemoteParticipantError::StateUnavailable {
source: Some(carried),
} = &error
else {
return Err(format!(
"#62 leg B REPRODUCED: the propagated error names the condition and not the cause, so \
a caller holding only this value cannot attribute the hold: {error:?}"
)
.into());
};
assert_eq!(
*carried, expected,
"the carried cause must be the caller's own store error, by value"
);
let walked = std::error::Error::source(&error).ok_or(
"a StateUnavailable carrying a cause must expose it as Error::source, or the chain \
walker still falls off at the rendered text",
)?;
assert_eq!(
walked.downcast_ref::<SdkError>(),
Some(&expected),
"the source must be the retained SdkError itself, not a re-wrapped copy"
);
Ok(())
}
#[test]
fn a_state_unavailable_reports_a_source_exactly_when_a_store_cause_was_retained() {
let cause = SdkError::Store {
description: "injected checkpoint failure".to_string(),
};
let attributed = super::RemoteParticipantError::StateUnavailable {
source: Some(cause.clone()),
};
assert_eq!(
std::error::Error::source(&attributed).and_then(|source| source.downcast_ref::<SdkError>()),
Some(&cause),
"a retained store cause must be reachable as Error::source"
);
let unattributed = super::RemoteParticipantError::StateUnavailable { source: None };
assert!(
std::error::Error::source(&unattributed).is_none(),
"an unavailability that did not originate in the store must report no source rather \
than a fabricated one"
);
}
#[test]
fn a_storage_error_exposes_its_store_failure_as_a_typed_source() -> TestResult {
let inner = SdkError::Store {
description: "injected checkpoint failure".to_string(),
};
let error = super::RemoteParticipantError::Storage(inner.clone());
let source = std::error::Error::source(&error).ok_or(
"#62 REPRODUCED: Storage declares no source, so the typed store error is \
reachable only by parsing rendered text",
)?;
let downcast = source
.downcast_ref::<SdkError>()
.ok_or("the source must be the SdkError itself, not a re-wrapped copy")?;
assert_eq!(*downcast, inner);
let transport = super::RemoteParticipantError::Transport(inner.clone());
let source = std::error::Error::source(&transport)
.ok_or("#62 REPRODUCED: Transport declares no source either")?;
assert_eq!(source.downcast_ref::<SdkError>(), Some(&inner));
Ok(())
}
#[test]
fn refused_detach_authority_never_destroys_live_participant_state() -> TestResult {
let token = DetachAttemptToken::new([9; 16]);
let (loopback, torn, config) = torn_rotation_state(token)?;
let restored = RemoteParticipantHandle::restore(&config, MemoryStore::default(), &torn)?;
replay_into_rotated_broker(&restored)?;
alive(&restored.receive(), "the refusing inbound")?;
alive(
&restored.recover_expected_operation(),
"recover_expected_operation",
)?;
alive(&restored.record_transport_fate(), "record_transport_fate")?;
alive(
&restored.recover_expected_operation(),
"recover_expected_operation after a failing checkpoint",
)?;
loopback.finish()?;
Ok(())
}
#[test]
fn a_refused_detach_authority_checkpoints_and_restores_steerable() -> TestResult {
let token = DetachAttemptToken::new([10; 16]);
let (loopback, torn, config) = torn_rotation_state(token)?;
let store = MemoryStore::default();
let observed = store.clone();
let restored = RemoteParticipantHandle::restore(&config, store, &torn)?;
replay_into_rotated_broker(&restored)?;
let inbound = restored.receive().map_err(|error| {
io::Error::other(format!(
"#59 REPRODUCED: the refusal could not be checkpointed: {error:?}"
))
})?;
assert!(
matches!(
inbound,
super::RemoteParticipantInbound::Applied {
value: ServerValue::StaleAuthority(_),
..
}
),
"the rotated broker's refusal must be applied, not retained"
);
let canonical = observed.bytes()?;
loopback.finish()?;
let reborn = RemoteParticipantHandle::restore(&config, MemoryStore::default(), &canonical)
.map_err(|error| {
io::Error::other(format!(
"#59 REPRODUCED: the checkpointed refusal does not restore: {error:?}"
))
})?;
assert!(
matches!(
reborn.record_operation(stale_reattach()?)?,
super::RemoteOperationRecordOutcome::Refused {
reason: ClientOperationRecordRefusalReason::BindingMismatch,
..
}
),
"the restored participant must answer the re-attach probe with a catchable refusal"
);
Ok(())
}
fn stale_reattach() -> Result<ClientRequest, io::Error> {
Ok(ClientRequest::CredentialAttach(CredentialAttachRequest {
conversation_id: CONVERSATION,
participant_id: PARTICIPANT,
capability_generation: generation(2)?,
attach_secret: AttachSecret::new([2; 32]),
attach_attempt_token: AttachAttemptToken::new([11; 16]),
accept_marker_delivery_seq: None,
}))
}
fn torn_rotation_state(token: DetachAttemptToken) -> TestResult<(Loopback, Vec<u8>, RemoteConfig)> {
let loopback = Loopback::spawn(vec![
vec![
Action::Respond(vec![enroll_bound(CONVERSATION, [1; 16])?]),
Action::DropAfterRequest,
],
vec![Action::Respond(vec![rotated_broker_refusal(token)?])],
])?;
let first = loopback.connected_config()?;
let store = MemoryStore::default();
let observed = store.clone();
let handle = RemoteParticipantHandle::new(&first, store)?;
enroll_and_receive(&handle)?;
send_detach(&handle, token)?;
let torn = observed.bytes()?;
drop(handle);
let second = loopback.connected_config()?;
Ok((loopback, torn, second))
}
fn replay_into_rotated_broker(handle: &RemoteParticipantHandle<MemoryStore>) -> TestResult {
assert!(matches!(
handle.resolve_lost_operation_authority()?,
super::RemoteLostOperationResolution::DetachParked { .. }
));
assert!(matches!(
handle.replay_detach()?,
super::RemoteDetachReplayOutcome::Send(super::RemoteParticipantSendOutcome::Sent { .. })
));
Ok(())
}
fn rotated_broker_refusal(token: DetachAttemptToken) -> Result<ServerValue, io::Error> {
Ok(ServerValue::StaleAuthority(StaleAuthority::Detach(
DetachStaleAuthority::Live {
conversation_id: CONVERSATION,
participant_id: PARTICIPANT,
capability_generation: generation(1)?,
detach_attempt_token: token,
current_generation: generation(2)?,
},
)))
}
fn alive<T>(result: &Result<T, super::RemoteParticipantError>, seam: &str) -> TestResult {
if matches!(
result,
Err(super::RemoteParticipantError::StateUnavailable { .. })
) {
return Err(io::Error::other(format!(
"#59 REPRODUCED: {seam} reported destroyed state after a pure encode refusal"
))
.into());
}
Ok(())
}
fn run_detach_terminal(value: ServerValue, token: DetachAttemptToken) -> TestResult {
let loopback = Loopback::spawn(vec![vec![
Action::Respond(vec![enroll_bound(CONVERSATION, [1; 16])?]),
Action::Respond(vec![value]),
]])?;
let config = loopback.connected_config()?;
let store = MemoryStore::default();
let observed = store.clone();
let handle = RemoteParticipantHandle::new(&config, store)?;
enroll_and_receive(&handle)?;
send_detach(&handle, token)?;
assert!(matches!(
handle.receive()?,
super::RemoteParticipantInbound::Applied { .. }
));
assert_replay_inactive(&handle)?;
let canonical = observed.bytes()?;
loopback.finish()?;
let restored = RemoteParticipantHandle::restore(&config, MemoryStore::default(), &canonical)?;
assert_replay_inactive(&restored)?;
Ok(())
}
fn enroll_and_receive(handle: &RemoteParticipantHandle<MemoryStore>) -> TestResult {
enroll(handle)?;
assert!(matches!(
handle.receive()?,
super::RemoteParticipantInbound::Applied { .. }
));
Ok(())
}
fn send_detach(
handle: &RemoteParticipantHandle<MemoryStore>,
token: DetachAttemptToken,
) -> TestResult {
let operation = recorded(
handle.record_operation(ClientRequest::Detach(DetachRequest {
conversation_id: CONVERSATION,
participant_id: PARTICIPANT,
capability_generation: generation(1)?,
detach_attempt_token: token,
}))?,
)?;
sent(&handle.send_operation(operation)?)?;
Ok(())
}
fn assert_replay_inactive(handle: &RemoteParticipantHandle<MemoryStore>) -> TestResult {
assert!(matches!(
handle.replay_detach()?,
super::RemoteDetachReplayOutcome::Refused {
reason: liminal_protocol::client::DetachReplayRefusalReason::InvalidStatus,
}
));
Ok(())
}
fn terminalized_detach(token: DetachAttemptToken) -> Result<ServerValue, io::Error> {
let mut body = Vec::new();
put_u16(&mut body, ClientDiscriminant::DetachRequest.wire_value());
put_u16(
&mut body,
AuthorityStateTag::TerminalizedDetachCell.wire_value(),
);
put_u64(&mut body, CONVERSATION);
put_u64(&mut body, PARTICIPANT);
put_u64(&mut body, generation(1)?.get());
body.extend_from_slice(token.as_bytes());
put_u64(&mut body, generation(2)?.get());
put_u64(&mut body, 7);
put_u64(&mut body, 8);
put_u64(&mut body, generation(1)?.get());
put_u16(&mut body, BindingStateTag::Detached.wire_value());
decode_server_value_body(
ServerDiscriminant::StaleAuthority,
ProtocolVersion::V1,
&body,
)
.map(|(value, _)| value)
.map_err(|error| io::Error::other(format!("terminalized fixture decode failed: {error:?}")))
}
fn put_u16(output: &mut Vec<u8>, value: u16) {
output.extend_from_slice(&value.to_be_bytes());
}
fn put_u64(output: &mut Vec<u8>, value: u64) {
output.extend_from_slice(&value.to_be_bytes());
}
#[derive(Debug, Default)]
struct FailSecondWrite {
writes: usize,
}
impl ParticipantResumeStore for FailSecondWrite {
fn persist(&mut self, canonical_lpcr: &[u8]) -> Result<(), SdkError> {
self.writes += 1;
if self.writes == 1 {
assert!(!canonical_lpcr.is_empty());
Ok(())
} else {
Err(SdkError::Store {
description: "injected checkpoint failure".to_string(),
})
}
}
}