liminal-server 0.20.0

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

use std::error::Error;

use liminal::durability::bridge::block_on;
use liminal::durability::{DurabilityError, DurableStore, RoutedBatchItem};

use super::{RoutedStreamError, RoutedStreamRefusal, read_head};
use crate::server::participant::production::log::OperationLogError;
use crate::server::participant::production::outbox_log::OutboxLogError;
use crate::server::participant::production::routed_stream_codec::{CodecError, head_key};
use crate::server::participant::production::tests_memory_store::MemoryStore;

#[test]
fn native_failures_keep_the_existing_unwrapped_durability_variant() {
    let log: OperationLogError =
        RoutedStreamError::Store(DurabilityError::OutcomeUnknown { pid: 17 }).into();
    assert!(matches!(
        log,
        OperationLogError::Durability(DurabilityError::OutcomeUnknown { pid: 17 })
    ));
    let outbox: OutboxLogError =
        RoutedStreamError::Store(DurabilityError::NotStarted { pid: 19 }).into();
    assert!(matches!(
        outbox,
        OutboxLogError::Durability(DurabilityError::NotStarted { pid: 19 })
    ));
}

#[test]
fn a_real_head_decode_failure_keeps_the_original_codec_source() -> Result<(), Box<dyn Error>> {
    let store = MemoryStore::default();
    block_on(store.apply_routed_batch(
        "route",
        vec![RoutedBatchItem {
            key: head_key("source")?,
            expected: None,
            value: vec![0],
            ttl: None,
        }],
    ))??;
    let original = block_on(read_head(&store, "route", "source"))?
        .err()
        .ok_or("head unexpectedly decoded")?;
    let log: OperationLogError = original.into();
    assert!(matches!(
        &log,
        OperationLogError::RoutedStream(RoutedStreamRefusal::Codec(CodecError::InvalidHead))
    ));
    let refusal = log.source().ok_or("refusal source missing")?;
    assert!(refusal.is::<RoutedStreamRefusal>());
    let codec = refusal.source().ok_or("codec source missing")?;
    assert!(matches!(
        codec.downcast_ref::<CodecError>(),
        Some(CodecError::InvalidHead)
    ));
    Ok(())
}

#[test]
fn allocation_refusal_keeps_the_original_allocation_source() -> Result<(), Box<dyn Error>> {
    let allocation = Vec::<u8>::new()
        .try_reserve(usize::MAX)
        .err()
        .ok_or("unrepresentable allocation accepted")?;
    let outbox: OutboxLogError =
        RoutedStreamError::Refusal(RoutedStreamRefusal::Allocation(allocation)).into();
    let refusal = outbox.source().ok_or("refusal source missing")?;
    assert!(refusal.is::<RoutedStreamRefusal>());
    assert!(
        refusal
            .source()
            .ok_or("allocation source missing")?
            .is::<std::collections::TryReserveError>()
    );
    Ok(())
}