use std::error::Error;
use liminal::durability::bridge::block_on;
use liminal::durability::{CountingStore, DurableStore, RoutedBatchItem, StoredEntry};
use super::OperationBatch;
use crate::server::participant::production::routed_stream::{
RoutedStreamError, RoutedStreamRefusal,
};
use crate::server::participant::production::routed_stream_codec::{
encode_head, encode_row, head_key, row_key,
};
use crate::server::participant::production::tests_memory_store::MemoryStore;
#[test]
fn a_pending_page_combines_committed_and_staged_rows_without_publishing()
-> Result<(), Box<dyn Error>> {
let store = CountingStore::new(MemoryStore::default());
block_on(store.apply_routed_batch(
"route",
vec![
RoutedBatchItem {
key: head_key("source")?,
expected: None,
value: encode_head(5).to_vec(),
ttl: None,
},
RoutedBatchItem {
key: row_key("source", 4)?,
expected: None,
value: encode_row(&StoredEntry {
sequence: 4,
timestamp: 40,
payload: vec![4],
})?,
ttl: None,
},
],
))??;
let mut batch = OperationBatch::new("route");
block_on(batch.stage_stream(&store, "source", vec![vec![5]], 5, 50))??;
block_on(batch.stage_stream(&store, "source", vec![vec![6]], 6, 60))??;
let before = store.snapshot();
let rows = block_on(batch.read_stream_page(&store, "source", 4, 10))??
.ok_or("candidate head missing")?;
assert_eq!(
rows,
vec![
StoredEntry {
sequence: 4,
timestamp: 40,
payload: vec![4]
},
StoredEntry {
sequence: 5,
timestamp: 50,
payload: vec![5]
},
StoredEntry {
sequence: 6,
timestamp: 60,
payload: vec![6]
},
]
);
let reads = store.snapshot().since(&before);
assert_eq!(reads.get_routed_calls, 1);
assert_eq!(reads.apply_routed_batch_calls, 0);
assert_eq!(reads.append_calls, 0);
assert_eq!(reads.append_all_calls, 0);
assert_eq!(reads.flushes, 0);
assert_eq!(
block_on(store.get_routed("route", &row_key("source", 5)?))??,
None
);
assert_eq!(
block_on(store.get_routed("route", &row_key("source", 6)?))??,
None
);
assert_eq!(
block_on(store.get_routed("route", &head_key("source")?))??,
Some(encode_head(5).to_vec())
);
let before = store.snapshot();
let marks = std::cell::Cell::new(0);
block_on(batch.commit(&store, || marks.set(marks.get() + 1)))??;
assert_eq!(marks.get(), 1);
assert_eq!(store.snapshot().since(&before).apply_routed_batch_calls, 1);
assert_eq!(store.snapshot().since(&before).apply_routed_batch_items, 3);
Ok(())
}
#[test]
fn a_pending_only_page_and_candidate_eof_need_no_store_reads() -> Result<(), Box<dyn Error>> {
let store = CountingStore::new(MemoryStore::default());
let mut batch = OperationBatch::new("route");
block_on(batch.stage_stream(&store, "source", vec![vec![0, 255]], 0, 12))??;
let before = store.snapshot();
let rows = block_on(batch.read_stream_page(&store, "source", 0, 1))??
.ok_or("candidate head missing")?;
assert_eq!(
rows,
vec![StoredEntry {
sequence: 0,
timestamp: 12,
payload: vec![0, 255]
}]
);
assert_eq!(
block_on(batch.read_stream_page(&store, "source", 1, 1))??,
Some(vec![])
);
assert_eq!(store.snapshot().since(&before).get_routed_calls, 0);
assert_eq!(store.snapshot().since(&before).apply_routed_batch_calls, 0);
assert_eq!(store.snapshot().since(&before).flushes, 0);
assert_eq!(
block_on(store.get_routed("route", &head_key("source")?))??,
None
);
Ok(())
}
#[test]
fn an_unstaged_absent_head_remains_explicit_absence() -> Result<(), Box<dyn Error>> {
let store = CountingStore::new(MemoryStore::default());
let mut batch = OperationBatch::new("route");
assert_eq!(
block_on(batch.read_stream_page(&store, "source", 0, 1))??,
None
);
assert_eq!(store.snapshot().get_routed_calls, 1);
let marks = std::cell::Cell::new(0);
assert!(matches!(
block_on(batch.commit(&store, || marks.set(marks.get() + 1)))?,
Err(liminal::durability::DurabilityError::EmptyOperationBatch)
));
assert_eq!(marks.get(), 0);
assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
Ok(())
}
#[test]
fn a_missing_committed_row_refuses_the_entire_pending_operation() -> Result<(), Box<dyn Error>> {
let store = CountingStore::new(MemoryStore::default());
block_on(store.apply_routed_batch(
"route",
vec![RoutedBatchItem {
key: head_key("source")?,
expected: None,
value: encode_head(3).to_vec(),
ttl: None,
}],
))??;
let mut batch = OperationBatch::new("route");
block_on(batch.stage_stream(&store, "source", vec![vec![3]], 3, 30))??;
let before = store.snapshot();
assert!(matches!(
block_on(batch.read_stream_page(&store, "source", 2, 1))?,
Err(RoutedStreamError::Refusal(
RoutedStreamRefusal::MissingRow { sequence: 2, .. }
))
));
assert_eq!(store.snapshot().since(&before).get_routed_calls, 1);
let marks = std::cell::Cell::new(0);
assert!(
matches!(block_on(batch.commit(&store, || marks.set(marks.get() + 1)))?,
Err(liminal::durability::DurabilityError::ConfigError(message)) if message == "operation batch staging was refused")
);
assert_eq!(marks.get(), 0);
assert_eq!(store.snapshot().since(&before).apply_routed_batch_calls, 0);
assert_eq!(
block_on(store.get_routed("route", &row_key("source", 3)?))??,
None
);
Ok(())
}
#[test]
fn invalid_page_bounds_poison_the_batch_before_a_head_read() -> Result<(), Box<dyn Error>> {
let store = CountingStore::new(MemoryStore::default());
let mut batch = OperationBatch::new("route");
assert!(matches!(
block_on(batch.read_stream_page(&store, "source", 0, 0))?,
Err(RoutedStreamError::Refusal(
RoutedStreamRefusal::InvalidLimit
))
));
assert!(matches!(
block_on(batch.stage_stream(&store, "source", vec![vec![1]], 0, 0))?,
Err(RoutedStreamError::Refusal(
RoutedStreamRefusal::BatchRefused
))
));
let marks = std::cell::Cell::new(0);
assert!(block_on(batch.commit(&store, || marks.set(marks.get() + 1)))?.is_err());
assert_eq!(marks.get(), 0);
assert_eq!(store.snapshot().get_routed_calls, 0);
assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
Ok(())
}