liminal-server 0.21.1

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
89

//! DUR-021 R4: the reclaim record answers every shard; the quiesce route
//! answers the store's word, refusing a held spine by name and saying when
//! no pass was needed; nothing answers before install or after close.

use std::sync::Arc;

use haematite::db::ShardReclaimAnswer;
use haematite::{Database, DatabaseConfig, EventStore, ReclaimStart, RuntimeOptions};

use super::reclaim_route::SharedReclaimSpine;

fn spine(dir: &std::path::Path) -> Result<Arc<EventStore>, Box<dyn std::error::Error>> {
    let database = Database::create(
        DatabaseConfig {
            data_dir: dir.to_path_buf(),
            shard_count: 3,
            distributed: None,
            executor_threads: None,
            node_cache_budget: Some(haematite::NodeCacheBudget::Unlimited),
            retention: haematite::Retention::reclaim(0),
        },
        RuntimeOptions {
            long_wait_interval: std::time::Duration::from_secs(5),
            reclaim_start: ReclaimStart::Held,
        },
    )?;
    Ok(Arc::new(EventStore::new(database)))
}

#[test]
fn nothing_answers_before_the_spine_is_installed() {
    let shared = SharedReclaimSpine::default();
    assert!(shared.status().is_none());
    assert!(shared.quiesce().is_none());
}

#[test]
fn the_record_answers_every_shard_and_quiesce_answers_the_stores_word()
-> Result<(), Box<dyn std::error::Error>> {
    let dir = tempfile::tempdir()?;
    let shared = SharedReclaimSpine::default();
    // The route holds the spine weakly; the test keeps it open as the
    // services do.
    let spine = spine(dir.path())?;
    assert!(shared.install(&spine));
    let status = shared.status().ok_or("installed")?;
    assert_eq!(status.shards.len(), 3);
    assert!(
        status
            .shards
            .iter()
            .all(|shard| matches!(shard, ShardReclaimAnswer::Answered { .. }))
    );
    let body = serde_json::to_string(&status)?;
    assert!(body.contains("\"shards\""));
    // Held, the store refuses to rest by name rather than wait for ever.
    let held = shared.quiesce().ok_or("installed")?;
    assert!(!held.quiesced, "a held spine was quiesced");
    assert!(
        held.refused
            .as_deref()
            .is_some_and(|refused| refused.contains("reclaim_held")),
        "{:?}",
        held.refused
    );
    spine.database().release_reclaim()?;
    let answer = shared.quiesce().ok_or("installed")?;
    assert!(answer.quiesced, "released: {:?}", answer.refused);
    assert_eq!(answer.status.shards.len(), 3);
    let again = shared.quiesce().ok_or("installed")?;
    assert!(again.quiesced, "{:?}", again.refused);
    assert_eq!(again.passes_run, 0, "nothing written: no pass was needed");
    Ok(())
}

/// The route never keeps the spine open: once the services drop it, the
/// route answers as before install.
#[test]
fn the_route_never_holds_the_spine_open() -> Result<(), Box<dyn std::error::Error>> {
    let dir = tempfile::tempdir()?;
    let shared = SharedReclaimSpine::default();
    let spine = spine(dir.path())?;
    assert!(shared.install(&spine));
    assert!(shared.status().is_some());
    drop(spine);
    assert!(shared.status().is_none(), "the route kept the closed spine");
    assert!(shared.quiesce().is_none());
    Ok(())
}