use std::cell::Cell;
use std::error::Error;
use std::sync::Arc;
use liminal::durability::bridge::block_on;
use liminal::durability::{CountingStore, DurableStore, RoutedBatchItem};
use super::OperationBatch;
use crate::server::participant::production::tests_memory_store::MemoryStore;
use crate::server::participant::production::tests_seed::Shared;
#[test]
fn covered_operations_and_checkpoint_publish_in_one_store_call() -> Result<(), Box<dyn Error>> {
let memory = Arc::new(MemoryStore::default());
let store = CountingStore::new(Shared(Arc::clone(&memory) as Arc<dyn DurableStore>));
let mut batch = OperationBatch::new("conversation");
for operation in 0_u64..257 {
batch.stage(RoutedBatchItem {
key: format!("operation/{operation}").into_bytes(),
expected: None,
value: operation.to_be_bytes().to_vec(),
ttl: None,
})?;
}
assert!(OperationBatch::checkpoint_due(0, 257, false)?);
batch.stage(RoutedBatchItem {
key: b"checkpoint".to_vec(),
expected: None,
value: b"covered:257".to_vec(),
ttl: None,
})?;
let submissions = Cell::new(0_usize);
block_on(batch.commit(&store, || {
assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
submissions.set(submissions.get() + 1);
}))??;
assert_eq!(submissions.get(), 1);
let counts = store.snapshot();
assert_eq!(counts.apply_routed_batch_calls, 1);
assert_eq!(counts.apply_routed_batch_items, 258);
assert_eq!(counts.append_calls, 0);
assert_eq!(counts.append_all_calls, 0);
assert_eq!(counts.flushes, 0);
for operation in 0_u64..257 {
assert_eq!(
block_on(
store.get_routed("conversation", format!("operation/{operation}").as_bytes())
)??,
Some(operation.to_be_bytes().to_vec())
);
}
assert_eq!(
block_on(store.get_routed("conversation", b"checkpoint"))??,
Some(b"covered:257".to_vec())
);
Ok(())
}
#[test]
fn sealing_and_crossings_require_a_checkpoint_without_clock_waits() -> Result<(), Box<dyn Error>> {
assert!(!OperationBatch::checkpoint_due(0, 255, false)?);
assert!(OperationBatch::checkpoint_due(255, 256, false)?);
assert!(!OperationBatch::checkpoint_due(256, 257, false)?);
assert!(OperationBatch::checkpoint_due(257, 257, true)?);
assert!(OperationBatch::checkpoint_due(257, 256, false).is_err());
Ok(())
}
#[test]
fn repeated_keys_are_refused_before_commit() -> Result<(), Box<dyn Error>> {
let mut batch = OperationBatch::new("conversation");
batch.stage(RoutedBatchItem {
key: b"operation".to_vec(),
expected: None,
value: vec![1],
ttl: None,
})?;
assert!(
batch
.stage(RoutedBatchItem {
key: b"operation".to_vec(),
expected: None,
value: vec![2],
ttl: None
})
.is_err()
);
Ok(())
}
#[test]
fn an_empty_logical_operation_is_refused_without_submitting() -> Result<(), Box<dyn Error>> {
let store = CountingStore::new(MemoryStore::default());
let submissions = Cell::new(0_usize);
let refused = block_on(OperationBatch::new("conversation").commit(&store, || {
assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
submissions.set(submissions.get() + 1);
}))?;
assert!(matches!(
refused,
Err(liminal::durability::DurabilityError::EmptyOperationBatch)
));
assert_eq!(submissions.get(), 0);
assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
Ok(())
}
#[test]
fn an_unpolled_commit_does_not_mark_or_submit() -> Result<(), Box<dyn Error>> {
let store = CountingStore::new(MemoryStore::default());
let mut batch = OperationBatch::new("conversation");
batch.stage(RoutedBatchItem {
key: b"operation".to_vec(),
expected: None,
value: vec![1],
ttl: None,
})?;
let submissions = Cell::new(0_usize);
let pending = batch.commit(&store, || {
assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
submissions.set(submissions.get() + 1);
});
assert_eq!(submissions.get(), 0);
assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
drop(pending);
assert_eq!(submissions.get(), 0);
assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
Ok(())
}
#[test]
fn a_refused_batch_marks_once_and_publishes_no_partial_checkpoint() -> Result<(), Box<dyn Error>> {
let memory = Arc::new(MemoryStore::default());
block_on(memory.apply_routed_batch(
"conversation",
vec![RoutedBatchItem {
key: b"operation".to_vec(),
expected: None,
value: vec![1],
ttl: None,
}],
))??;
let store = CountingStore::new(Shared(Arc::clone(&memory) as Arc<dyn DurableStore>));
let mut batch = OperationBatch::new("conversation");
batch.stage(RoutedBatchItem {
key: b"checkpoint".to_vec(),
expected: None,
value: b"covered:1".to_vec(),
ttl: None,
})?;
batch.stage(RoutedBatchItem {
key: b"operation".to_vec(),
expected: None,
value: vec![2],
ttl: None,
})?;
let submissions = Cell::new(0_usize);
let refused = block_on(batch.commit(&store, || {
assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
submissions.set(submissions.get() + 1);
}))?;
assert!(matches!(
refused,
Err(liminal::durability::DurabilityError::ConfigError(message))
if message.contains("content-hash mismatch")
));
assert_eq!(submissions.get(), 1);
let counts = store.snapshot();
assert_eq!(counts.apply_routed_batch_calls, 1);
assert_eq!(counts.apply_routed_batch_items, 0);
assert_eq!(counts.append_calls, 0);
assert_eq!(counts.append_all_calls, 0);
assert_eq!(counts.flushes, 0);
assert_eq!(
block_on(store.get_routed("conversation", b"operation"))??,
Some(vec![1])
);
assert_eq!(
block_on(store.get_routed("conversation", b"checkpoint"))??,
None
);
Ok(())
}