liminal-server 0.24.5

Standalone server for the liminal messaging bus
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

use super::OperationBatch;
use crate::server::participant::production::routed_stream::{read_head, read_page};
use crate::server::participant::production::tests_memory_store::MemoryStore;
use liminal::durability::bridge::block_on;
use liminal::durability::{CountingStore, RoutedBatchItem};
use std::error::Error;

#[test]
fn validated_bootstrap_starts_only_the_tail_and_never_rewrites_covered_rows()
-> Result<(), Box<dyn Error>> {
    let store = CountingStore::new(MemoryStore::default());
    let mut batch = OperationBatch::new("route");
    block_on(batch.stage_bootstrap_stream(&store, "source", 256))??;
    block_on(batch.stage_stream(&store, "source", vec![vec![7]], 256, 99))??;
    batch.stage(RoutedBatchItem {
        key: b"checkpoint".to_vec(),
        expected: None,
        value: vec![1],
        ttl: None,
    })?;
    assert_eq!(store.snapshot().get_routed_calls, 1);
    assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
    block_on(batch.commit(&store, || {}))??;
    assert_eq!(store.snapshot().apply_routed_batch_calls, 1);
    assert_eq!(store.snapshot().apply_routed_batch_items, 3);
    let head = block_on(read_head(&store, "route", "source"))??.ok_or("head missing")?;
    assert_eq!(head.next_sequence, 257);
    let before = store.snapshot();
    let rows = block_on(read_page(&store, "route", "source", &head, 256, 64))??;
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].payload, [7]);
    assert_eq!(rows[0].sequence, 256);
    assert_eq!(store.snapshot().since(&before).get_routed_calls, 1);
    Ok(())
}

#[test]
fn a_bootstrap_boundary_conflict_poison_refuses_all_other_rows() -> Result<(), Box<dyn Error>> {
    let store = CountingStore::new(MemoryStore::default());
    let mut initial = OperationBatch::new("route");
    block_on(initial.stage_bootstrap_stream(&store, "source", 256))??;
    block_on(initial.commit(&store, || {}))??;
    let before = store.snapshot();
    let mut batch = OperationBatch::new("route");
    batch.stage(RoutedBatchItem {
        key: b"checkpoint".to_vec(),
        expected: None,
        value: vec![1],
        ttl: None,
    })?;
    assert!(block_on(batch.stage_bootstrap_stream(&store, "source", 255))?.is_err());
    assert!(block_on(batch.commit(&store, || {}))?.is_err());
    assert_eq!(store.snapshot().since(&before).apply_routed_batch_calls, 0);
    Ok(())
}