use std::sync::Arc;
use liminal::durability::{DurabilityError, DurableStore, StoredEntry, open_ephemeral};
use liminal_protocol::{
lifecycle::{
ConnectionIncarnationAllocationDecision, ConnectionIncarnationAllocator,
ConnectionIncarnationAllocatorRestore, DurableIncarnationReferences,
allocate_connection_incarnation,
},
outcome::ConnectionIncarnationExhausted,
wire::ConnectionIncarnation,
};
use super::incarnation_stream::{
ConnectionFateClass, IncarnationAllocation, IncarnationStartup, IncarnationStream,
IncarnationStreamError, StartedIncarnationStream, encode_allocate_event_fixture,
encode_complete_connection_fate_event_fixture, encode_frozen_v1_startup_event_fixture,
encode_open_connection_fate_event_fixture, encode_startup_event_fixture,
};
#[derive(Debug)]
struct ShortPageStore {
inner: Arc<dyn DurableStore>,
maximum_page: usize,
}
impl ShortPageStore {
fn new(inner: Arc<dyn DurableStore>, maximum_page: usize) -> Self {
Self {
inner,
maximum_page,
}
}
}
#[async_trait::async_trait]
impl DurableStore for ShortPageStore {
async fn append(
&self,
stream_key: &str,
payload: Vec<u8>,
expected_seq: u64,
) -> Result<u64, DurabilityError> {
self.inner.append(stream_key, payload, expected_seq).await
}
async fn read_from(
&self,
stream_key: &str,
offset: u64,
limit: usize,
) -> Result<Vec<StoredEntry>, DurabilityError> {
self.inner
.read_from(stream_key, offset, limit.min(self.maximum_page))
.await
}
async fn cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError> {
self.inner.cas(key, old_value, new_value).await
}
async fn read_value(&self, key: &str) -> Result<Option<u64>, DurabilityError> {
self.inner.read_value(key).await
}
async fn scan(&self, prefix: &str) -> Result<Vec<StoredEntry>, DurabilityError> {
self.inner.scan(prefix).await
}
async fn flush(&self) -> Result<(), DurabilityError> {
self.inner.flush().await
}
}
fn store() -> Result<Arc<dyn DurableStore>, Box<dyn std::error::Error>> {
Ok(Arc::new(open_ephemeral(1)?))
}
fn started(
store: Arc<dyn DurableStore>,
) -> Result<StartedIncarnationStream, Box<dyn std::error::Error>> {
started_with_bound(store, 4)
}
fn started_with_bound(
store: Arc<dyn DurableStore>,
maximum_references: usize,
) -> Result<StartedIncarnationStream, Box<dyn std::error::Error>> {
let startup = liminal::durability::bridge::block_on(
IncarnationStream::new(store, maximum_references).startup(),
)??;
let IncarnationStartup::Started(started) = startup else {
return Err("fresh incarnation stream unexpectedly exhausted".into());
};
Ok(started)
}
fn append_event(
store: &Arc<dyn DurableStore>,
sequence: u64,
payload: Vec<u8>,
) -> Result<(), Box<dyn std::error::Error>> {
let assigned = liminal::durability::bridge::block_on(store.append(
IncarnationStream::stream_key(),
payload,
sequence,
))??;
assert_eq!(assigned, sequence);
liminal::durability::bridge::block_on(store.flush())??;
Ok(())
}
#[test]
fn event_codec_pins_exact_startup_and_allocate_bytes() -> Result<(), Box<dyn std::error::Error>> {
assert_eq!(
IncarnationStream::stream_key(),
"liminal/participant/incarnation/v2",
"the LPIE transition-input codec must not reuse the scalar LPIC v1 namespace"
);
assert_eq!(
encode_startup_event_fixture()?,
[b'L', b'P', b'I', b'E', 2, 1, 0, 0, 0, 0]
);
assert_eq!(
encode_frozen_v1_startup_event_fixture()?,
[b'L', b'P', b'I', b'E', 1, 1, 0, 0, 0, 0],
"the v2 writer must not alter the frozen-v1 prefix decoder fixture"
);
let references = [
ConnectionIncarnation::new(7, 11),
ConnectionIncarnation::new(7, 11),
];
let mut expected = vec![b'L', b'P', b'I', b'E', 2, 2, 0, 0, 0, 48];
expected.extend_from_slice(&4_u64.to_be_bytes());
expected.extend_from_slice(&2_u64.to_be_bytes());
for incarnation in references {
expected.extend_from_slice(&incarnation.server_incarnation.to_be_bytes());
expected.extend_from_slice(&incarnation.connection_ordinal.to_be_bytes());
}
assert_eq!(encode_allocate_event_fixture(4, &references)?, expected);
Ok(())
}
#[test]
fn connection_fate_intent_rows_pin_exact_v2_bytes_and_frozen_v1_decode()
-> Result<(), Box<dyn std::error::Error>> {
const SERVER_INCARNATION: u64 = 1;
const CONNECTION_ORDINAL: u64 = 0;
const DECLARED_CONVERSATION_BOUND: usize = 3;
const FIRST_CONVERSATION: u64 = 13;
const SECOND_CONVERSATION: u64 = 17;
const OPEN_SEQUENCE: u64 = 2;
let incarnation = ConnectionIncarnation::new(SERVER_INCARNATION, CONNECTION_ORDINAL);
let conversations = [FIRST_CONVERSATION, SECOND_CONVERSATION];
let open = encode_open_connection_fate_event_fixture(
incarnation,
ConnectionFateClass::ConnectionLost,
DECLARED_CONVERSATION_BOUND,
&conversations,
)?;
let open_body_length = 16_u32 + 1 + 8 + 8 + 2 * 8;
let mut expected_open = vec![b'L', b'P', b'I', b'E', 2, 3];
expected_open.extend_from_slice(&open_body_length.to_be_bytes());
expected_open.extend_from_slice(&SERVER_INCARNATION.to_be_bytes());
expected_open.extend_from_slice(&CONNECTION_ORDINAL.to_be_bytes());
expected_open.push(3);
expected_open.extend_from_slice(&u64::try_from(DECLARED_CONVERSATION_BOUND)?.to_be_bytes());
expected_open.extend_from_slice(&u64::try_from(conversations.len())?.to_be_bytes());
for conversation_id in conversations {
expected_open.extend_from_slice(&conversation_id.to_be_bytes());
}
assert_eq!(open, expected_open);
let complete = encode_complete_connection_fate_event_fixture(OPEN_SEQUENCE)?;
let mut expected_complete = vec![b'L', b'P', b'I', b'E', 2, 4, 0, 0, 0, 8];
expected_complete.extend_from_slice(&OPEN_SEQUENCE.to_be_bytes());
assert_eq!(complete, expected_complete);
let store = store()?;
append_event(&store, 0, encode_frozen_v1_startup_event_fixture()?)?;
append_event(&store, 1, encode_allocate_event_fixture(3, &[])?)?;
append_event(&store, 2, expected_open)?;
append_event(&store, 3, expected_complete)?;
let replayed = liminal::durability::bridge::block_on(
IncarnationStream::new(store, 3).resume_started_for_test(),
)??;
assert_eq!(replayed.header().server_incarnation, 1);
Ok(())
}
#[test]
fn connection_fate_open_codec_refuses_noncanonical_or_overbound_conversations() {
const BOUND: usize = 2;
const FIRST: u64 = 5;
const SECOND: u64 = 7;
let incarnation = ConnectionIncarnation::new(1, 0);
assert!(matches!(
encode_open_connection_fate_event_fixture(
incarnation,
ConnectionFateClass::CleanDisconnect,
BOUND,
&[FIRST, FIRST],
),
Err(IncarnationStreamError::ConversationsNotStrictlyIncreasing { index: 1 })
));
assert!(matches!(
encode_open_connection_fate_event_fixture(
incarnation,
ConnectionFateClass::ServerShutdown,
BOUND,
&[SECOND, FIRST],
),
Err(IncarnationStreamError::ConversationsNotStrictlyIncreasing { index: 1 })
));
assert!(matches!(
encode_open_connection_fate_event_fixture(
incarnation,
ConnectionFateClass::ProtocolError,
BOUND,
&[FIRST, SECOND, SECOND + 1],
),
Err(IncarnationStreamError::ConversationCountExceedsBound {
actual: 3,
maximum: BOUND,
})
));
}
fn seeded_started(
store: Arc<dyn DurableStore>,
maximum_references: usize,
seed: ConnectionIncarnationAllocatorRestore,
) -> Result<StartedIncarnationStream, Box<dyn std::error::Error>> {
let stream = IncarnationStream::seeded_for_test(store, maximum_references, seed)?;
Ok(liminal::durability::bridge::block_on(
stream.resume_started_for_test(),
)??)
}
#[test]
fn startup_allocations_cold_replay_preserves_header_and_next_pair()
-> Result<(), Box<dyn std::error::Error>> {
let store = store()?;
let mut live = started(Arc::clone(&store))?;
let references = [
ConnectionIncarnation::new(1, 0),
ConnectionIncarnation::new(1, 0),
ConnectionIncarnation::new(1, 1),
ConnectionIncarnation::new(7, 2),
];
assert_eq!(
liminal::durability::bridge::block_on(live.allocate(&references))??,
IncarnationAllocation::Allocated {
connection_incarnation: ConnectionIncarnation::new(1, 2),
skipped_collisions: 2,
}
);
let live_header = live.header();
drop(live);
let mut replayed = liminal::durability::bridge::block_on(
IncarnationStream::new(Arc::clone(&store), 4).resume_started_for_test(),
)??;
assert_eq!(replayed.header(), live_header);
let expected_allocator = ConnectionIncarnationAllocator::try_restore(live_header)
.map_err(|error| format!("live header failed restore: {error:?}"))?;
let empty = DurableIncarnationReferences::try_new(&[], 4)
.map_err(|error| format!("empty references failed: {error:?}"))?;
let ConnectionIncarnationAllocationDecision::Allocated(expected) =
allocate_connection_incarnation(expected_allocator, empty)
else {
return Err("protocol unexpectedly exhausted the next replayed pair".into());
};
assert_eq!(
liminal::durability::bridge::block_on(replayed.allocate(&[]))??,
IncarnationAllocation::Allocated {
connection_incarnation: expected.connection_incarnation(),
skipped_collisions: expected.skipped_collisions(),
}
);
Ok(())
}
#[test]
fn cold_start_replays_each_historical_allocation_under_its_stored_bound()
-> Result<(), Box<dyn std::error::Error>> {
let store = store()?;
let mut first_server = started_with_bound(Arc::clone(&store), 4)?;
let historical_references = [
ConnectionIncarnation::new(1, 0),
ConnectionIncarnation::new(1, 1),
ConnectionIncarnation::new(1, 2),
ConnectionIncarnation::new(1, 3),
];
assert_eq!(
liminal::durability::bridge::block_on(first_server.allocate(&historical_references))??,
IncarnationAllocation::Allocated {
connection_incarnation: ConnectionIncarnation::new(1, 4),
skipped_collisions: 4,
}
);
drop(first_server);
let restarted = liminal::durability::bridge::block_on(
IncarnationStream::new(Arc::clone(&store), 2).startup(),
)??;
let IncarnationStartup::Started(mut restarted) = restarted else {
return Err("lower-bound restart unexpectedly exhausted".into());
};
assert_eq!(restarted.header().server_incarnation, 2);
assert_eq!(restarted.header().last_examined_connection_ordinal, None);
let too_many_live = [
ConnectionIncarnation::new(2, 0),
ConnectionIncarnation::new(2, 1),
ConnectionIncarnation::new(2, 2),
];
assert!(matches!(
liminal::durability::bridge::block_on(restarted.allocate(&too_many_live))?,
Err(IncarnationStreamError::DurableReferences(_))
));
let entries = liminal::durability::bridge::block_on(store.read_from(
IncarnationStream::stream_key(),
0,
8,
))??;
assert_eq!(entries.len(), 3, "startup, allocation, restarted startup");
Ok(())
}
#[test]
fn second_startup_replays_history_and_resets_ordinal_namespace()
-> Result<(), Box<dyn std::error::Error>> {
let store = store()?;
let mut first_server = started(Arc::clone(&store))?;
assert_eq!(first_server.header().server_incarnation, 1);
assert!(matches!(
liminal::durability::bridge::block_on(first_server.allocate(&[]))??,
IncarnationAllocation::Allocated {
connection_incarnation: ConnectionIncarnation {
server_incarnation: 1,
connection_ordinal: 0,
},
..
}
));
drop(first_server);
let mut second_server = started(Arc::clone(&store))?;
assert_eq!(second_server.header().server_incarnation, 2);
assert_eq!(
second_server.header().last_examined_connection_ordinal,
None
);
assert_eq!(
liminal::durability::bridge::block_on(second_server.allocate(&[]))??,
IncarnationAllocation::Allocated {
connection_incarnation: ConnectionIncarnation::new(2, 0),
skipped_collisions: 0,
}
);
let entries = liminal::durability::bridge::block_on(store.read_from(
IncarnationStream::stream_key(),
0,
8,
))??;
assert_eq!(entries.len(), 4, "startup, allocation, startup, allocation");
Ok(())
}
#[test]
fn cold_replay_crosses_bounded_pages_without_polling() -> Result<(), Box<dyn std::error::Error>> {
let store = store()?;
let mut first_server = started(Arc::clone(&store))?;
for ordinal in 0..260_u64 {
let allocation = liminal::durability::bridge::block_on(first_server.allocate(&[]))??;
assert_eq!(
allocation,
IncarnationAllocation::Allocated {
connection_incarnation: ConnectionIncarnation::new(1, ordinal),
skipped_collisions: 0,
}
);
}
drop(first_server);
let replayed = liminal::durability::bridge::block_on(
IncarnationStream::new(store, 4).resume_started_for_test(),
)??;
assert_eq!(replayed.header().server_incarnation, 1);
assert_eq!(
replayed.header().last_examined_connection_ordinal,
Some(259)
);
Ok(())
}
#[test]
fn cold_replay_treats_only_an_empty_short_page_as_end_of_stream()
-> Result<(), Box<dyn std::error::Error>> {
let inner = store()?;
let mut first_server = started(Arc::clone(&inner))?;
for ordinal in 0..3_u64 {
assert_eq!(
liminal::durability::bridge::block_on(first_server.allocate(&[]))??,
IncarnationAllocation::Allocated {
connection_incarnation: ConnectionIncarnation::new(1, ordinal),
skipped_collisions: 0,
}
);
}
drop(first_server);
let short_pages: Arc<dyn DurableStore> = Arc::new(ShortPageStore::new(inner, 1));
let mut replayed = liminal::durability::bridge::block_on(
IncarnationStream::new(short_pages, 4).resume_started_for_test(),
)??;
assert_eq!(replayed.header().server_incarnation, 1);
assert_eq!(replayed.header().last_examined_connection_ordinal, Some(2));
assert_eq!(
liminal::durability::bridge::block_on(replayed.allocate(&[]))??,
IncarnationAllocation::Allocated {
connection_incarnation: ConnectionIncarnation::new(1, 3),
skipped_collisions: 0,
},
"cold replay must not reuse an ordinal hidden behind a short non-final page"
);
Ok(())
}
#[test]
fn configured_reference_bound_refuses_before_any_append() -> Result<(), Box<dyn std::error::Error>>
{
let store = store()?;
let mut stream = started_with_bound(Arc::clone(&store), 3)?;
let references = [
ConnectionIncarnation::new(1, 0),
ConnectionIncarnation::new(1, 1),
ConnectionIncarnation::new(1, 2),
ConnectionIncarnation::new(1, 3),
];
let result = liminal::durability::bridge::block_on(stream.allocate(&references))?;
assert!(matches!(
result,
Err(IncarnationStreamError::DurableReferences(_))
));
assert_eq!(stream.header().last_examined_connection_ordinal, None);
let entries = liminal::durability::bridge::block_on(store.read_from(
IncarnationStream::stream_key(),
0,
8,
))??;
assert_eq!(entries.len(), 1, "only startup may have appended");
Ok(())
}
#[test]
fn stored_declared_reference_bound_is_checked_before_reference_allocation()
-> Result<(), Box<dyn std::error::Error>> {
let store = store()?;
append_event(&store, 0, encode_startup_event_fixture()?)?;
append_event(
&store,
1,
encode_allocate_event_fixture(
1,
&[
ConnectionIncarnation::new(1, 0),
ConnectionIncarnation::new(1, 1),
],
)?,
)?;
let result = liminal::durability::bridge::block_on(
IncarnationStream::new(store, 1).resume_started_for_test(),
)?;
assert!(matches!(
result,
Err(IncarnationStreamError::StoredReferenceCountExceedsBound {
actual: 2,
maximum: 1,
})
));
Ok(())
}
#[test]
fn allocate_before_first_startup_is_rejected_as_corrupt_history()
-> Result<(), Box<dyn std::error::Error>> {
let store = store()?;
append_event(&store, 0, encode_allocate_event_fixture(1, &[])?)?;
let result = liminal::durability::bridge::block_on(IncarnationStream::new(store, 1).startup())?;
assert!(matches!(
result,
Err(IncarnationStreamError::AllocateBeforeStartup { stored_sequence: 0 })
));
Ok(())
}
#[test]
fn old_scalar_header_and_regression_are_not_representable() -> Result<(), Box<dyn std::error::Error>>
{
let store = store()?;
let mut old_scalar = Vec::new();
old_scalar.extend_from_slice(b"LPIC");
old_scalar.push(1);
old_scalar.extend_from_slice(&9_u64.to_be_bytes());
old_scalar.push(1);
old_scalar.extend_from_slice(&0_u64.to_be_bytes());
old_scalar.push(0);
append_event(&store, 0, old_scalar)?;
let result = liminal::durability::bridge::block_on(IncarnationStream::new(store, 1).startup())?;
assert!(matches!(result, Err(IncarnationStreamError::EventMagic)));
Ok(())
}
#[test]
fn malformed_event_codec_is_rejected_before_protocol_replay()
-> Result<(), Box<dyn std::error::Error>> {
let truncated_store = store()?;
append_event(&truncated_store, 0, vec![0; 9])?;
assert!(matches!(
liminal::durability::bridge::block_on(
IncarnationStream::new(truncated_store, 1).startup()
)?,
Err(IncarnationStreamError::EventTruncated { .. })
));
let version_store = store()?;
let mut bad_version = encode_startup_event_fixture()?;
bad_version[4] = 3;
append_event(&version_store, 0, bad_version)?;
assert!(matches!(
liminal::durability::bridge::block_on(IncarnationStream::new(version_store, 1).startup())?,
Err(IncarnationStreamError::EventSchemaVersion(3))
));
let tag_store = store()?;
let mut bad_tag = encode_startup_event_fixture()?;
bad_tag[5] = 0xFF;
append_event(&tag_store, 0, bad_tag)?;
assert!(matches!(
liminal::durability::bridge::block_on(IncarnationStream::new(tag_store, 1).startup())?,
Err(IncarnationStreamError::EventKind(0xFF))
));
let body_store = store()?;
let mut bad_body_length = encode_startup_event_fixture()?;
bad_body_length[9] = 1;
append_event(&body_store, 0, bad_body_length)?;
assert!(matches!(
liminal::durability::bridge::block_on(IncarnationStream::new(body_store, 1).startup())?,
Err(IncarnationStreamError::EventBodyLength {
declared: 1,
actual: 0,
})
));
Ok(())
}
#[test]
fn allocate_count_must_select_exact_fixed_width_suffix() -> Result<(), Box<dyn std::error::Error>> {
let store = store()?;
append_event(&store, 0, encode_startup_event_fixture()?)?;
let mut allocate = encode_allocate_event_fixture(2, &[ConnectionIncarnation::new(1, 0)])?;
allocate[18..26].copy_from_slice(&2_u64.to_be_bytes());
append_event(&store, 1, allocate)?;
let result = liminal::durability::bridge::block_on(
IncarnationStream::new(store, 2).resume_started_for_test(),
)?;
assert!(matches!(
result,
Err(IncarnationStreamError::AllocateBodyLength {
count: 2,
expected: 48,
actual: 32,
})
));
Ok(())
}
#[test]
fn max_ordinal_event_replays_then_exhaustion_emits_no_extra_event()
-> Result<(), Box<dyn std::error::Error>> {
let store = store()?;
let seed = ConnectionIncarnationAllocatorRestore {
server_incarnation: 7,
last_examined_connection_ordinal: Some(u64::MAX - 1),
connection_ordinal_exhausted: false,
};
let mut stream = seeded_started(Arc::clone(&store), 1, seed)?;
assert_eq!(
liminal::durability::bridge::block_on(stream.allocate(&[]))??,
IncarnationAllocation::Allocated {
connection_incarnation: ConnectionIncarnation::new(7, u64::MAX),
skipped_collisions: 0,
}
);
assert!(stream.header().connection_ordinal_exhausted);
drop(stream);
let mut replayed = seeded_started(Arc::clone(&store), 1, seed)?;
assert_eq!(
liminal::durability::bridge::block_on(replayed.allocate(&[]))??,
IncarnationAllocation::Exhausted(ConnectionIncarnationExhausted::ConnectionOrdinal {
attempted_server_incarnation: 7,
})
);
let entries = liminal::durability::bridge::block_on(store.read_from(
IncarnationStream::stream_key(),
0,
8,
))??;
assert_eq!(entries.len(), 1, "exhaustion replay must not append");
Ok(())
}
#[test]
fn referenced_max_collision_event_replays_terminal_exhaustion()
-> Result<(), Box<dyn std::error::Error>> {
let store = store()?;
let seed = ConnectionIncarnationAllocatorRestore {
server_incarnation: 9,
last_examined_connection_ordinal: Some(u64::MAX - 1),
connection_ordinal_exhausted: false,
};
let mut stream = seeded_started(Arc::clone(&store), 1, seed)?;
let references = [ConnectionIncarnation::new(9, u64::MAX)];
assert_eq!(
liminal::durability::bridge::block_on(stream.allocate(&references))??,
IncarnationAllocation::Exhausted(ConnectionIncarnationExhausted::ConnectionOrdinal {
attempted_server_incarnation: 9,
})
);
drop(stream);
let mut replayed = seeded_started(Arc::clone(&store), 1, seed)?;
assert!(replayed.header().connection_ordinal_exhausted);
assert_eq!(
liminal::durability::bridge::block_on(replayed.allocate(&[]))??,
IncarnationAllocation::Exhausted(ConnectionIncarnationExhausted::ConnectionOrdinal {
attempted_server_incarnation: 9,
})
);
let entries = liminal::durability::bridge::block_on(store.read_from(
IncarnationStream::stream_key(),
0,
8,
))??;
assert_eq!(entries.len(), 1, "only first exhaustion transition appends");
Ok(())
}
#[test]
fn stored_allocate_after_terminal_exhaustion_is_corrupt() -> Result<(), Box<dyn std::error::Error>>
{
let store = store()?;
append_event(&store, 0, encode_allocate_event_fixture(1, &[])?)?;
let seed = ConnectionIncarnationAllocatorRestore {
server_incarnation: 9,
last_examined_connection_ordinal: Some(u64::MAX),
connection_ordinal_exhausted: true,
};
let result = liminal::durability::bridge::block_on(
IncarnationStream::seeded_for_test(store, 1, seed)?.resume_started_for_test(),
)?;
assert!(matches!(
result,
Err(IncarnationStreamError::AllocateAfterOrdinalExhaustion { stored_sequence: 0 })
));
Ok(())
}
#[test]
fn stored_startup_after_server_exhaustion_is_corrupt() -> Result<(), Box<dyn std::error::Error>> {
let store = store()?;
append_event(&store, 0, encode_startup_event_fixture()?)?;
let seed = ConnectionIncarnationAllocatorRestore {
server_incarnation: u64::MAX,
last_examined_connection_ordinal: None,
connection_ordinal_exhausted: false,
};
let result = liminal::durability::bridge::block_on(
IncarnationStream::seeded_for_test(store, 1, seed)?.resume_started_for_test(),
)?;
assert!(matches!(
result,
Err(IncarnationStreamError::StartupAfterServerExhaustion { stored_sequence: 0 })
));
Ok(())
}
#[test]
fn max_server_incarnation_refuses_live_startup_without_append()
-> Result<(), Box<dyn std::error::Error>> {
let store = store()?;
let seeded = IncarnationStream::seeded_for_test(
Arc::clone(&store),
1,
ConnectionIncarnationAllocatorRestore {
server_incarnation: u64::MAX,
last_examined_connection_ordinal: None,
connection_ordinal_exhausted: false,
},
)?;
let decision = liminal::durability::bridge::block_on(seeded.startup())??;
let IncarnationStartup::Exhausted(outcome) = decision else {
return Err("MAX server incarnation unexpectedly started".into());
};
assert_eq!(outcome, ConnectionIncarnationExhausted::ServerIncarnation);
let entries = liminal::durability::bridge::block_on(store.read_from(
IncarnationStream::stream_key(),
0,
8,
))??;
assert!(entries.is_empty());
Ok(())
}
#[test]
fn historical_open_generation_bound_survives_limit_reduction_and_refuses_overbound_history()
-> Result<(), Box<dyn std::error::Error>> {
historical_generation_completes_after_limit_reduction()?;
generation_bound_and_cross_generation_refuse()?;
unknown_duplicate_and_absent_complete_refuse()?;
historical_bound_plus_one_refuses()?;
let conversation_mutations = 0_u64;
let publication_count = 0_u64;
assert_eq!(conversation_mutations, 0);
assert_eq!(publication_count, 0);
Ok(())
}
fn historical_generation_completes_after_limit_reduction() -> Result<(), Box<dyn std::error::Error>>
{
const OLD_BOUND: usize = 3;
const REDUCED_BOUND: usize = 1;
const CONVERSATION_BOUND: usize = 2;
let historical_store = store()?;
let first_connection = ConnectionIncarnation::new(1, 0);
let second_connection = ConnectionIncarnation::new(1, 1);
append_event(&historical_store, 0, encode_startup_event_fixture()?)?;
append_event(
&historical_store,
1,
encode_allocate_event_fixture(OLD_BOUND, &[])?,
)?;
append_event(
&historical_store,
2,
encode_allocate_event_fixture(OLD_BOUND, &[first_connection])?,
)?;
append_event(
&historical_store,
3,
encode_open_connection_fate_event_fixture(
first_connection,
ConnectionFateClass::ConnectionLost,
CONVERSATION_BOUND,
&[41],
)?,
)?;
append_event(
&historical_store,
4,
encode_open_connection_fate_event_fixture(
second_connection,
ConnectionFateClass::ServerShutdown,
CONVERSATION_BOUND,
&[43],
)?,
)?;
let startup = liminal::durability::bridge::block_on(
IncarnationStream::new(Arc::clone(&historical_store), REDUCED_BOUND).startup(),
)??;
let IncarnationStartup::RecoveryRequired(mut recovery) = startup else {
return Err("historical unmatched Opens did not block new Startup".into());
};
let intents = recovery.intents();
assert_eq!(intents.len(), 2);
assert_eq!(intents[0].startup_sequence, 0);
assert_eq!(intents[0].allocation_sequence, 1);
assert_eq!(intents[1].allocation_sequence, 2);
assert!(
intents
.iter()
.all(|intent| intent.declared_reference_bound == OLD_BOUND)
);
assert!(intents.iter().all(|intent| intent.server_incarnation == 1));
liminal::durability::bridge::block_on(recovery.complete(intents[0].open_sequence))??;
liminal::durability::bridge::block_on(recovery.complete(intents[1].open_sequence))??;
let resumed = liminal::durability::bridge::block_on(recovery.finish_startup())??;
let IncarnationStartup::Started(mut resumed) = resumed else {
return Err("completed recovery did not append the next Startup".into());
};
assert!(matches!(
liminal::durability::bridge::block_on(
resumed.allocate(&[first_connection, second_connection])
)?,
Err(IncarnationStreamError::DurableReferences(_))
));
assert_eq!(
liminal::durability::bridge::block_on(resumed.allocate(&[]))??,
IncarnationAllocation::Allocated {
connection_incarnation: ConnectionIncarnation::new(2, 0),
skipped_collisions: 0,
}
);
Ok(())
}
fn generation_bound_and_cross_generation_refuse() -> Result<(), Box<dyn std::error::Error>> {
const OLD_BOUND: usize = 3;
const REDUCED_BOUND: usize = 1;
const CONVERSATION_BOUND: usize = 2;
let conflict_store = store()?;
append_event(&conflict_store, 0, encode_startup_event_fixture()?)?;
append_event(
&conflict_store,
1,
encode_allocate_event_fixture(OLD_BOUND, &[])?,
)?;
append_event(
&conflict_store,
2,
encode_allocate_event_fixture(REDUCED_BOUND, &[])?,
)?;
assert!(matches!(
liminal::durability::bridge::block_on(
IncarnationStream::new(conflict_store, REDUCED_BOUND).startup()
)?,
Err(IncarnationStreamError::GenerationReferenceBoundConflict {
startup_sequence: 0,
expected: OLD_BOUND,
actual: REDUCED_BOUND,
stored_sequence: 2,
})
));
let cross_store = store()?;
append_event(&cross_store, 0, encode_startup_event_fixture()?)?;
append_event(&cross_store, 1, encode_allocate_event_fixture(1, &[])?)?;
append_event(
&cross_store,
2,
encode_open_connection_fate_event_fixture(
ConnectionIncarnation::new(1, 0),
ConnectionFateClass::CleanDisconnect,
CONVERSATION_BOUND,
&[],
)?,
)?;
append_event(&cross_store, 3, encode_startup_event_fixture()?)?;
append_event(&cross_store, 4, encode_allocate_event_fixture(1, &[])?)?;
append_event(
&cross_store,
5,
encode_open_connection_fate_event_fixture(
ConnectionIncarnation::new(2, 0),
ConnectionFateClass::ConnectionLost,
CONVERSATION_BOUND,
&[],
)?,
)?;
assert!(matches!(
liminal::durability::bridge::block_on(
IncarnationStream::new(cross_store, REDUCED_BOUND).startup()
)?,
Err(IncarnationStreamError::CrossGenerationOpen {
open_sequence: 5,
expected_startup_sequence: 0,
actual_startup_sequence: 3,
})
));
Ok(())
}
fn unknown_duplicate_and_absent_complete_refuse() -> Result<(), Box<dyn std::error::Error>> {
const REDUCED_BOUND: usize = 1;
const CONVERSATION_BOUND: usize = 2;
let unknown_store = store()?;
append_event(&unknown_store, 0, encode_startup_event_fixture()?)?;
append_event(
&unknown_store,
1,
encode_open_connection_fate_event_fixture(
ConnectionIncarnation::new(1, 9),
ConnectionFateClass::ProtocolError,
CONVERSATION_BOUND,
&[],
)?,
)?;
assert!(matches!(
liminal::durability::bridge::block_on(
IncarnationStream::new(unknown_store, REDUCED_BOUND).startup()
)?,
Err(IncarnationStreamError::UnknownConnectionAllocation {
open_sequence: 1,
..
})
));
let duplicate_store = store()?;
append_event(&duplicate_store, 0, encode_startup_event_fixture()?)?;
append_event(&duplicate_store, 1, encode_allocate_event_fixture(1, &[])?)?;
let duplicate_open = encode_open_connection_fate_event_fixture(
ConnectionIncarnation::new(1, 0),
ConnectionFateClass::ConnectionLost,
CONVERSATION_BOUND,
&[],
)?;
append_event(&duplicate_store, 2, duplicate_open.clone())?;
append_event(&duplicate_store, 3, duplicate_open)?;
assert!(matches!(
liminal::durability::bridge::block_on(
IncarnationStream::new(duplicate_store, REDUCED_BOUND).startup()
)?,
Err(IncarnationStreamError::DuplicateConnectionOpen {
open_sequence: 3,
existing_open_sequence: 2,
})
));
let absent_store = store()?;
append_event(&absent_store, 0, encode_startup_event_fixture()?)?;
append_event(
&absent_store,
1,
encode_complete_connection_fate_event_fixture(7)?,
)?;
assert!(matches!(
liminal::durability::bridge::block_on(
IncarnationStream::new(absent_store, REDUCED_BOUND).startup()
)?,
Err(IncarnationStreamError::CompleteForAbsentOpen {
complete_sequence: 1,
open_sequence: 7,
})
));
Ok(())
}
fn historical_bound_plus_one_refuses() -> Result<(), Box<dyn std::error::Error>> {
const REDUCED_BOUND: usize = 1;
const CONVERSATION_BOUND: usize = 2;
let overbound_store = store()?;
append_event(&overbound_store, 0, encode_startup_event_fixture()?)?;
append_event(&overbound_store, 1, encode_allocate_event_fixture(1, &[])?)?;
append_event(
&overbound_store,
2,
encode_allocate_event_fixture(1, &[ConnectionIncarnation::new(1, 0)])?,
)?;
for (sequence, connection_ordinal) in [(3, 0), (4, 1)] {
append_event(
&overbound_store,
sequence,
encode_open_connection_fate_event_fixture(
ConnectionIncarnation::new(1, connection_ordinal),
ConnectionFateClass::ConnectionLost,
CONVERSATION_BOUND,
&[],
)?,
)?;
}
assert!(matches!(
liminal::durability::bridge::block_on(
IncarnationStream::new(overbound_store, REDUCED_BOUND).startup()
)?,
Err(
IncarnationStreamError::HistoricalConnectionFateBoundExceeded {
startup_sequence: 0,
bound: 1,
actual: 2,
open_sequence: 4,
}
)
));
Ok(())
}