use std::error::Error;
use std::sync::Arc;
use liminal::durability::bridge::block_on;
use liminal::durability::{CountingStore, DurableStore, RoutedBatchItem};
use super::{CheckpointStageError, OperationBatch};
use crate::server::participant::production::restore_checkpoint::{
CheckpointError, CheckpointLiveState, CheckpointPositions, CheckpointSchema, RestoreCheckpoint,
};
use crate::server::participant::production::restore_checkpoint_locator::{
CheckpointBodyKey, CheckpointLocator, checkpoint_conversation_route,
};
use crate::server::participant::production::tests_memory_store::MemoryStore;
#[derive(Debug)]
struct Live(u8);
impl CheckpointLiveState for Live {
type RestoreContext = u8;
fn encode_live(&self) -> Result<Vec<u8>, CheckpointError> {
Ok(vec![self.0])
}
fn decode_live(
bytes: &[u8],
positions: &CheckpointPositions,
root: &[u8],
context: &u8,
) -> Result<Self, CheckpointError> {
if bytes != [*context] {
return Err(CheckpointError::LiveState {
source: std::io::Error::other("live identity differs").into(),
});
}
let live = Self(*context);
live.validate_live(positions, root)?;
Ok(live)
}
fn validate_live(
&self,
positions: &CheckpointPositions,
root: &[u8],
) -> Result<(), CheckpointError> {
if u64::from(self.0) != positions.conversation_id || root != [9; 32] {
return Err(CheckpointError::LiveState {
source: std::io::Error::other("live association differs").into(),
});
}
Ok(())
}
}
fn checkpoint(covered: u64) -> Result<RestoreCheckpoint<Live, 32>, CheckpointError> {
RestoreCheckpoint::new(
CheckpointPositions {
conversation_id: 7,
covered_log_sequence: covered,
next_event_ordinal: covered + 1,
next_log_sequence: covered + 1,
next_seq: 2,
next_order: 1,
next_participant: 2,
extension_sequence: 1,
highest_delivery_seq: 1,
observer_progress: 1,
merged_base_sequence: covered + 1,
merged_extension_sequence: 1,
schema: CheckpointSchema::V4Suffix,
},
[9; 32],
Live(7),
)
}
#[test]
fn covered_row_body_and_locator_publish_in_one_atomic_submission() -> Result<(), Box<dyn Error>> {
let store = CountingStore::new(MemoryStore::default());
let route = checkpoint_conversation_route(7);
let checkpoint = checkpoint(255)?;
let body = checkpoint.encode_canonical()?;
let body_key = CheckpointBodyKey::for_body(&body);
let mut batch = OperationBatch::new(&route);
batch.stage(RoutedBatchItem {
key: b"covered-row".to_vec(),
expected: None,
value: vec![4],
ttl: None,
})?;
block_on(batch.stage_borrowed_checkpoint(
&store,
checkpoint.positions(),
checkpoint.history_root(),
checkpoint.live(),
[3; 32],
))??;
assert_eq!(store.snapshot().get_routed_calls, 2);
assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
assert!(block_on(store.get_routed(&route, CheckpointLocator::routed_key()))??.is_none());
assert!(block_on(store.get_routed(&route, &body_key.routed_key()))??.is_none());
block_on(batch.commit(&store, || {}))??;
let counts = store.snapshot();
assert_eq!(counts.apply_routed_batch_calls, 1);
assert_eq!(counts.apply_routed_batch_items, 3);
assert_eq!(counts.append_calls, 0);
assert_eq!(counts.flushes, 0);
let saved = block_on(store.get_routed(&route, CheckpointLocator::routed_key()))??
.ok_or("locator missing")?;
let locator = CheckpointLocator::decode_canonical(&saved, 7)?;
assert_eq!(locator.covered_sequence(), 255);
assert_eq!(locator.extension_boundary(), 1);
assert_eq!(locator.prefix_digest(), &[3; 32]);
assert_eq!(
block_on(store.get_routed(&route, &locator.body_key().routed_key()))??,
Some(body)
);
assert_eq!(
block_on(store.get_routed(&route, b"covered-row"))??,
Some(vec![4])
);
Ok(())
}
#[test]
fn a_locator_fence_refuses_the_covered_row_and_new_body_together() -> Result<(), Box<dyn Error>> {
let store = Arc::new(CountingStore::new(MemoryStore::default()));
let route = checkpoint_conversation_route(7);
let checkpoint = checkpoint(255)?;
let body_key = CheckpointBodyKey::for_body(&checkpoint.encode_canonical()?);
let mut batch = OperationBatch::new(&route);
batch.stage(RoutedBatchItem {
key: b"covered-row".to_vec(),
expected: None,
value: vec![4],
ttl: None,
})?;
block_on(batch.stage_borrowed_checkpoint(
store.as_ref(),
checkpoint.positions(),
checkpoint.history_root(),
checkpoint.live(),
[3; 32],
))??;
let competing =
CheckpointLocator::new(7, 255, 1, [8; 32], CheckpointBodyKey::for_body(b"other"))?
.encode_canonical()
.to_vec();
block_on(store.apply_routed_batch(
&route,
vec![RoutedBatchItem {
key: CheckpointLocator::routed_key().to_vec(),
expected: None,
value: competing.clone(),
ttl: None,
}],
))??;
let before = store.snapshot();
assert!(block_on(batch.commit(store.as_ref(), || {}))?.is_err());
let calls = store.snapshot().since(&before);
assert_eq!(calls.apply_routed_batch_calls, 1);
assert_eq!(calls.apply_routed_batch_items, 0);
assert!(block_on(store.get_routed(&route, b"covered-row"))??.is_none());
assert!(block_on(store.get_routed(&route, &body_key.routed_key()))??.is_none());
assert_eq!(
block_on(store.get_routed(&route, CheckpointLocator::routed_key()))??,
Some(competing)
);
Ok(())
}
#[test]
fn a_foreign_route_poison_refuses_all_staged_rows_without_submission() -> Result<(), Box<dyn Error>>
{
let store = CountingStore::new(MemoryStore::default());
let mut batch = OperationBatch::new("other-route");
batch.stage(RoutedBatchItem {
key: b"covered-row".to_vec(),
expected: None,
value: vec![4],
ttl: None,
})?;
let captured = checkpoint(255)?;
assert!(matches!(
block_on(batch.stage_borrowed_checkpoint(
&store,
captured.positions(),
captured.history_root(),
captured.live(),
[3; 32]
))?,
Err(CheckpointStageError::RouteMismatch)
));
assert_eq!(store.snapshot().get_routed_calls, 0);
assert!(block_on(batch.commit(&store, || {}))?.is_err());
assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
Ok(())
}
#[test]
fn a_loudly_rebuilt_checkpoint_replaces_corrupt_locator_with_covered_rows()
-> Result<(), Box<dyn Error>> {
let store = CountingStore::new(MemoryStore::default());
let route = checkpoint_conversation_route(7);
block_on(store.apply_routed_batch(
&route,
vec![RoutedBatchItem {
key: CheckpointLocator::routed_key().to_vec(),
expected: None,
value: b"corrupt locator".to_vec(),
ttl: None,
}],
))??;
let captured = checkpoint(255)?;
let mut normal = OperationBatch::new(&route);
assert!(
block_on(normal.stage_borrowed_checkpoint(
&store,
captured.positions(),
captured.history_root(),
captured.live(),
[3; 32]
))?
.is_err()
);
assert!(block_on(normal.commit(&store, || {}))?.is_err());
let mut rebuilt = OperationBatch::new(&route);
rebuilt.stage(RoutedBatchItem {
key: b"covered-row".to_vec(),
expected: None,
value: vec![4],
ttl: None,
})?;
let before = store.snapshot();
block_on(rebuilt.stage_rebuilt_checkpoint(
&store,
&captured,
[3; 32],
Some(b"corrupt locator"),
))??;
assert_eq!(store.snapshot().since(&before).apply_routed_batch_calls, 0);
block_on(rebuilt.commit(&store, || {}))??;
let calls = store.snapshot().since(&before);
assert_eq!(calls.apply_routed_batch_calls, 1);
assert_eq!(calls.apply_routed_batch_items, 3);
let bytes = block_on(store.get_routed(&route, CheckpointLocator::routed_key()))??
.ok_or("rebuilt locator missing")?;
let locator = CheckpointLocator::decode_canonical(&bytes, 7)?;
assert_eq!(locator.covered_sequence(), 255);
assert_eq!(
block_on(store.get_routed(&route, &locator.body_key().routed_key()))??,
Some(captured.encode_canonical()?)
);
assert_eq!(
block_on(store.get_routed(&route, b"covered-row"))??,
Some(vec![4])
);
Ok(())
}
#[test]
fn a_rebuild_keeps_the_loaders_witness_and_refuses_a_changed_locator() -> Result<(), Box<dyn Error>>
{
let store = CountingStore::new(MemoryStore::default());
let route = checkpoint_conversation_route(7);
let observed = b"old corrupt locator";
block_on(store.apply_routed_batch(
&route,
vec![RoutedBatchItem {
key: CheckpointLocator::routed_key().to_vec(),
expected: None,
value: b"replacement locator".to_vec(),
ttl: None,
}],
))??;
let captured = checkpoint(255)?;
let body_key = CheckpointBodyKey::for_body(&captured.encode_canonical()?);
let mut rebuilt = OperationBatch::new(&route);
rebuilt.stage(RoutedBatchItem {
key: b"covered-row".to_vec(),
expected: None,
value: vec![4],
ttl: None,
})?;
let before = store.snapshot();
block_on(rebuilt.stage_rebuilt_checkpoint(&store, &captured, [3; 32], Some(observed)))??;
assert_eq!(store.snapshot().since(&before).get_routed_calls, 1);
assert!(block_on(rebuilt.commit(&store, || {}))?.is_err());
assert!(block_on(store.get_routed(&route, b"covered-row"))??.is_none());
assert!(block_on(store.get_routed(&route, &body_key.routed_key()))??.is_none());
assert_eq!(
block_on(store.get_routed(&route, CheckpointLocator::routed_key()))??,
Some(b"replacement locator".to_vec())
);
Ok(())
}
#[test]
fn borrowed_capture_keeps_the_live_owner_and_commits_the_same_canonical_body()
-> Result<(), Box<dyn Error>> {
let store = CountingStore::new(MemoryStore::default());
let checkpoint = checkpoint(255)?;
let (positions, root, live) = checkpoint.into_parts();
let expected = RestoreCheckpoint::new(positions, root, Live(live.0))?.encode_canonical()?;
let route = checkpoint_conversation_route(7);
let mut batch = OperationBatch::new(&route);
block_on(batch.stage_borrowed_checkpoint(&store, positions, &root, &live, [4; 32]))??;
assert_eq!(live.0, 7);
assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
block_on(batch.commit(&store, || {}))??;
let locator = block_on(store.get_routed(&route, CheckpointLocator::routed_key()))??
.ok_or("borrowed checkpoint locator is absent")?;
let locator = CheckpointLocator::decode_canonical(&locator, 7)?;
assert_eq!(
block_on(store.get_routed(&route, &locator.body_key().routed_key()))??,
Some(expected)
);
assert_eq!(store.snapshot().apply_routed_batch_calls, 1);
Ok(())
}
#[test]
fn prefix_covers_only_staged_sources_in_operation_order_without_store_reads()
-> Result<(), Box<dyn Error>> {
use crate::server::participant::production::restore_checkpoint_locator::{
CheckpointSource, advance_prefix_digest, checkpoint_prefix_seed,
};
let store = CountingStore::new(MemoryStore::default());
let mut batch = OperationBatch::new(&checkpoint_conversation_route(7));
for (stream, source, sequence, payload) in [
("base", CheckpointSource::Base, 0, vec![1]),
("extension", CheckpointSource::Extension, 0, vec![2]),
("base", CheckpointSource::Base, 1, vec![3]),
] {
block_on(batch.stage_source_stream(
&store,
stream,
source,
liminal::durability::StoredEntry {
payload,
sequence,
timestamp: 0,
},
))??;
}
batch.stage(RoutedBatchItem {
key: b"non-source".to_vec(),
expected: None,
value: vec![9],
ttl: None,
})?;
let before = store.snapshot();
let first = advance_prefix_digest(checkpoint_prefix_seed(), CheckpointSource::Base, 0, &[1])?;
let second = advance_prefix_digest(first, CheckpointSource::Extension, 0, &[2])?;
let expected = advance_prefix_digest(second, CheckpointSource::Base, 1, &[3])?;
assert_eq!(batch.checkpoint_prefix(checkpoint_prefix_seed())?, expected);
assert_eq!(store.snapshot(), before);
assert_eq!(before.apply_routed_batch_calls, 0);
Ok(())
}
struct LocalLive(std::cell::Cell<u8>);
impl CheckpointLiveState for LocalLive {
type RestoreContext = u8;
fn encode_live(&self) -> Result<Vec<u8>, CheckpointError> {
Live(self.0.get()).encode_live()
}
fn decode_live(
bytes: &[u8],
positions: &CheckpointPositions,
root: &[u8],
context: &u8,
) -> Result<Self, CheckpointError> {
let live = Live::decode_live(bytes, positions, root, context)?;
Ok(Self(std::cell::Cell::new(live.0)))
}
fn validate_live(
&self,
positions: &CheckpointPositions,
root: &[u8],
) -> Result<(), CheckpointError> {
Live(self.0.get()).validate_live(positions, root)
}
}
fn require_send<T: Send>(value: T) -> T {
value
}
#[test]
fn checkpoint_store_future_releases_the_non_sync_live_owner_before_poll()
-> Result<(), Box<dyn Error>> {
let store = CountingStore::new(MemoryStore::default());
let checkpoint = checkpoint(255)?;
let positions = checkpoint.positions();
let expected = checkpoint.encode_canonical()?;
let route = checkpoint_conversation_route(7);
let mut batch = OperationBatch::new(&route);
let pending = {
let root = [9; 32];
let live = LocalLive(std::cell::Cell::new(7));
require_send(batch.stage_borrowed_checkpoint(&store, positions, &root, &live, [4; 32]))
};
assert_eq!(store.snapshot().get_routed_calls, 0);
assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
block_on(pending)??;
block_on(batch.commit(&store, || {}))??;
let locator = block_on(store.get_routed(&route, CheckpointLocator::routed_key()))??
.ok_or("borrowed checkpoint locator is absent")?;
let locator = CheckpointLocator::decode_canonical(&locator, 7)?;
assert_eq!(
block_on(store.get_routed(&route, &locator.body_key().routed_key()))??,
Some(expected)
);
assert_eq!(store.snapshot().apply_routed_batch_calls, 1);
Ok(())
}