liminal-server 0.25.4

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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

//! 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. SRV-048 R1: the swap is
//! left unrun by the start and settled after it.

use std::sync::Arc;

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

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

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

/// Starts the stream over `store`, its anchor swap left unrun.
fn start_unsettled(store: &Arc<dyn DurableStore>) -> TestResult<StartedIncarnationStream> {
    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)
}

/// Starts the stream over `store` and settles its anchor swap, as the start
/// thread does; answers the server incarnation it named.
fn start(store: &Arc<dyn DurableStore>) -> TestResult<u64> {
    let mut started = start_unsettled(store)?;
    block_on(started.settle_anchor())??;
    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(())
}

/// SRV-048 R1: a restart's start makes no anchor swap; the anchor still names
/// the predecessor when the start returns, and the settle moves it.
#[test]
fn a_restart_start_makes_no_anchor_swap() -> TestResult {
    let counting = Arc::new(CountingStore::new(open_ephemeral(1)?));
    let store: Arc<dyn DurableStore> = counting.clone();
    assert_eq!(start(&store)?, 1);
    let before = counting.snapshot().cas_calls;
    let mut restarted = start_unsettled(&store)?;
    assert_eq!(restarted.header().server_incarnation, 2);
    assert_eq!(
        counting.snapshot().cas_calls - before,
        0,
        "the restart's start swaps no anchor"
    );
    assert_eq!(
        block_on(store.read_value(EPOCH_ANCHOR_KEY))??,
        Some(1),
        "the anchor still names the predecessor when the start returns"
    );
    block_on(restarted.settle_anchor())??;
    assert_eq!(
        counting.snapshot().cas_calls - before,
        1,
        "the settle swaps once"
    );
    assert_eq!(block_on(store.read_value(EPOCH_ANCHOR_KEY))??, Some(2));
    block_on(restarted.settle_anchor())??;
    assert_eq!(
        counting.snapshot().cas_calls - before,
        1,
        "a settled stream swaps nothing more"
    );
    Ok(())
}