liminal-server 0.25.2

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

//! SRV-047 R1: a restart's epoch rotation makes one durable flush; the anchor
//! swap is a hint that rides the store's next flush.

use std::sync::Arc;

use liminal::durability::bridge::block_on;
use liminal::durability::{CountingStore, DurableStore, open_ephemeral};

use super::super::{IncarnationStartup, IncarnationStream};
use super::EPOCH_ANCHOR_KEY;

type TestResult<T = ()> = Result<T, Box<dyn std::error::Error>>;

/// Starts the stream over `store` and answers the server incarnation it named.
fn start(store: &Arc<dyn DurableStore>) -> TestResult<u64> {
    let startup = block_on(IncarnationStream::new(Arc::clone(store), 4).startup())??;
    let IncarnationStartup::Started(started) = startup else {
        return Err("the start did not complete".into());
    };
    Ok(started.header().server_incarnation)
}

#[test]
fn a_restart_rotation_flushes_once_and_the_anchor_names_its_epoch() -> TestResult {
    let counting = Arc::new(CountingStore::new(open_ephemeral(1)?));
    let store: Arc<dyn DurableStore> = counting.clone();
    assert_eq!(start(&store)?, 1);
    for restart in [2_u64, 3] {
        let before = counting.snapshot().flushes;
        assert_eq!(start(&store)?, restart);
        assert_eq!(
            counting.snapshot().flushes - before,
            1,
            "restart {restart}: the epoch's Startup is flushed once and the anchor rides it"
        );
        assert_eq!(
            block_on(store.read_value(EPOCH_ANCHOR_KEY))??,
            Some(restart),
            "the anchor names the epoch of record"
        );
    }
    Ok(())
}