use std::error::Error;
use std::sync::Arc;
use haematite::tree::Hash;
use liminal::durability::bridge::block_on;
use liminal::durability::{CountingStore, DurableStore, RoutedBatchItem};
use super::OperationBatch;
use crate::server::participant::production::routed_stream::{
RoutedStreamError, RoutedStreamRefusal, read_head, read_page,
};
use crate::server::participant::production::routed_stream_codec::{
decode_head, encode_head, head_key, row_key,
};
use crate::server::participant::production::tests_memory_store::MemoryStore;
use crate::server::participant::production::tests_seed::Shared;
#[test]
fn repeated_stream_appends_keep_the_original_head_witness() -> Result<(), Box<dyn Error>> {
let memory = Arc::new(MemoryStore::default());
block_on(memory.apply_routed_batch(
"route",
vec![RoutedBatchItem {
key: head_key("source")?,
expected: None,
value: encode_head(5).to_vec(),
ttl: None,
}],
))??;
let store = CountingStore::new(Shared(Arc::clone(&memory) as Arc<dyn DurableStore>));
let mut batch = OperationBatch::new("route");
assert_eq!(
block_on(batch.stage_stream(&store, "source", vec![vec![10]], 5, 40))??,
6
);
assert_eq!(
block_on(batch.stage_stream(&store, "source", vec![vec![11]], 6, 41))??,
7
);
let staged = store.snapshot();
assert_eq!(staged.get_routed_calls, 1);
assert_eq!(staged.apply_routed_batch_calls, 0);
assert_eq!(
block_on(memory.get_routed("route", &head_key("source")?))??,
Some(encode_head(5).to_vec())
);
assert_eq!(
block_on(memory.get_routed("route", &row_key("source", 5)?))??,
None
);
block_on(batch.commit(&store, || {}))??;
let committed = store.snapshot();
assert_eq!(committed.apply_routed_batch_calls, 1);
assert_eq!(committed.apply_routed_batch_items, 3);
assert_eq!(committed.append_calls, 0);
assert_eq!(committed.append_all_calls, 0);
assert_eq!(committed.flushes, 0);
let head = block_on(read_head(&store, "route", "source"))??.ok_or("head missing")?;
assert_eq!(head.next_sequence, 7);
let rows = block_on(read_page(&store, "route", "source", &head, 5, 2))??;
assert_eq!(rows.len(), 2);
assert_eq!(
(
rows[0].sequence,
rows[0].timestamp,
rows[0].payload.as_slice()
),
(5, 40, [10].as_slice())
);
assert_eq!(
(
rows[1].sequence,
rows[1].timestamp,
rows[1].payload.as_slice()
),
(6, 41, [11].as_slice())
);
Ok(())
}
#[test]
fn streams_sharing_a_route_commit_disjoint_rows_and_heads() -> Result<(), Box<dyn Error>> {
let store = CountingStore::new(MemoryStore::default());
let mut batch = OperationBatch::new("route");
assert_eq!(
block_on(batch.stage_stream(&store, "source", vec![vec![1]], 0, 10))??,
1
);
assert_eq!(
block_on(batch.stage_stream(&store, "outbox", vec![vec![2]], 0, 11))??,
1
);
assert_eq!(
block_on(batch.stage_stream(&store, "source", vec![vec![3]], 1, 12))??,
2
);
assert_eq!(store.snapshot().get_routed_calls, 2);
block_on(batch.commit(&store, || {}))??;
assert_eq!(store.snapshot().apply_routed_batch_calls, 1);
assert_eq!(store.snapshot().apply_routed_batch_items, 5);
for (stream, expected) in [("source", 2), ("outbox", 1)] {
let bytes =
block_on(store.get_routed("route", &head_key(stream)?))??.ok_or("head missing")?;
assert_eq!(decode_head(&bytes)?, expected);
}
Ok(())
}
#[test]
fn a_changed_original_head_refuses_every_staged_row_and_checkpoint() -> Result<(), Box<dyn Error>> {
let memory = Arc::new(MemoryStore::default());
let original = encode_head(5);
block_on(memory.apply_routed_batch(
"route",
vec![RoutedBatchItem {
key: head_key("source")?,
expected: None,
value: original.to_vec(),
ttl: None,
}],
))??;
let store = CountingStore::new(Shared(Arc::clone(&memory) as Arc<dyn DurableStore>));
let mut batch = OperationBatch::new("route");
block_on(batch.stage_stream(&store, "source", vec![vec![1], vec![2]], 5, 20))??;
batch.stage(RoutedBatchItem {
key: b"checkpoint".to_vec(),
expected: None,
value: vec![3],
ttl: None,
})?;
block_on(memory.apply_routed_batch(
"route",
vec![RoutedBatchItem {
key: head_key("source")?,
expected: Some(Hash::of(&original)),
value: encode_head(6).to_vec(),
ttl: None,
}],
))??;
let refused = block_on(batch.commit(&store, || {}))?;
assert!(refused.is_err());
assert_eq!(store.snapshot().apply_routed_batch_calls, 1);
assert_eq!(store.snapshot().apply_routed_batch_items, 0);
assert_eq!(
block_on(memory.get_routed("route", &row_key("source", 5)?))??,
None
);
assert_eq!(
block_on(memory.get_routed("route", &row_key("source", 6)?))??,
None
);
assert_eq!(block_on(memory.get_routed("route", b"checkpoint"))??, None);
assert_eq!(
block_on(memory.get_routed("route", &head_key("source")?))??,
Some(encode_head(6).to_vec())
);
Ok(())
}
#[test]
fn stream_preflight_refuses_empty_stale_and_overflow_without_submission()
-> Result<(), Box<dyn Error>> {
let store = CountingStore::new(MemoryStore::default());
let mut batch = OperationBatch::new("route");
assert!(matches!(
block_on(batch.stage_stream(&store, "source", vec![], 0, 0))?,
Err(RoutedStreamError::Refusal(RoutedStreamRefusal::EmptyAppend))
));
assert_eq!(store.snapshot().get_routed_calls, 0);
let mut batch = OperationBatch::new("route");
assert!(matches!(
block_on(batch.stage_stream(&store, "source", vec![vec![1]], 1, 0))?,
Err(RoutedStreamError::Refusal(
RoutedStreamRefusal::HeadMismatch {
expected: 1,
actual: 0
}
))
));
block_on(store.apply_routed_batch(
"route",
vec![RoutedBatchItem {
key: head_key("overflow")?,
expected: None,
value: encode_head(u64::MAX).to_vec(),
ttl: None,
}],
))??;
let before = store.snapshot();
let mut batch = OperationBatch::new("route");
assert!(matches!(
block_on(batch.stage_stream(&store, "overflow", vec![vec![1]], u64::MAX, 0))?,
Err(RoutedStreamError::Refusal(
RoutedStreamRefusal::SequenceOverflow
))
));
assert!(block_on(batch.commit(&store, || {}))?.is_err());
assert_eq!(store.snapshot().since(&before).apply_routed_batch_calls, 0);
Ok(())
}
#[test]
fn raw_keys_cannot_collide_with_a_staged_stream_head() -> 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![1]], 0, 0))??;
assert!(
batch
.stage(RoutedBatchItem {
key: head_key("source")?,
expected: None,
value: vec![0],
ttl: None
})
.is_err()
);
assert!(block_on(batch.commit(&store, || {}))?.is_err());
assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
assert_eq!(
block_on(store.get_routed("route", &head_key("source")?))??,
None
);
Ok(())
}
#[test]
fn a_later_staging_refusal_discards_prior_rows_and_checkpoint() -> 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![1]], 0, 0))??;
batch.stage(RoutedBatchItem {
key: b"checkpoint".to_vec(),
expected: None,
value: vec![2],
ttl: None,
})?;
assert!(matches!(
block_on(batch.stage_stream(&store, "source", vec![vec![3]], 0, 0))?,
Err(RoutedStreamError::Refusal(
RoutedStreamRefusal::HeadMismatch {
expected: 0,
actual: 1
}
))
));
let marks = std::cell::Cell::new(0);
let result = block_on(batch.commit(&store, || marks.set(marks.get() + 1)))?;
assert!(
matches!(result, Err(liminal::durability::DurabilityError::ConfigError(message))
if message == "operation batch staging was refused")
);
assert_eq!(marks.get(), 0);
assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
assert_eq!(
block_on(store.get_routed("route", &row_key("source", 0)?))??,
None
);
assert_eq!(block_on(store.get_routed("route", b"checkpoint"))??, None);
assert_eq!(
block_on(store.get_routed("route", &head_key("source")?))??,
None
);
Ok(())
}