liminal-server 0.24.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
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
90
91
92
93
94
95
96
97
98
99
100
101
102
103

use std::cell::{Cell, RefCell};
use std::error::Error;
use std::sync::Arc;

use liminal::durability::{CountingStore, DurabilityError, DurableStore};

use super::LogAppender;
use crate::server::participant::production::checkpoint_batches::CheckpointPublisher;
use crate::server::participant::production::log::{
    OperationLog, OperationLogError, StoredOperation,
};
use crate::server::participant::production::operation_batch::OperationBatch;
use crate::server::participant::production::outbox_log::OutboxLog;
use crate::server::participant::production::registry::ConversationRegistry;
use crate::server::participant::production::restore_checkpoint_locator::checkpoint_conversation_route;
use crate::server::participant::production::routed_stream::RoutedStreamRefusal;
use crate::server::participant::production::state::DurableAppend;
use crate::server::participant::production::tests_memory_store::MemoryStore;

#[test]
fn appender_commits_once_and_cannot_append_or_commit_after_consumption()
-> Result<(), Box<dyn Error>> {
    let store = Arc::new(CountingStore::new(MemoryStore::default()));
    let shared = Arc::clone(&store) as Arc<dyn DurableStore>;
    let registry = ConversationRegistry::new(Arc::clone(&shared));
    let log = OperationLog::new(Arc::clone(&shared), 55);
    let outbox_log = OutboxLog::new(shared, 55);
    let batch = RefCell::new(Some(OperationBatch::new(&checkpoint_conversation_route(
        55,
    ))));
    let publisher = CheckpointPublisher::default();
    let appender = LogAppender {
        log: &log,
        registry: &registry,
        conversation_id: 55,
        outbox_log: &outbox_log,
        prune_keep: None,
        publisher: &publisher,
        outstanding_extension_rows: Cell::new(0),
        batch: Some(&batch),
        timestamp: 10,
    };
    appender.append(&StoredOperation::Genesis { event: vec![1] }, 0)?;
    let before = store.snapshot();
    // The genesis registered the conversation: the registry row and its
    // routed head are one routed batch of their own (DUR-019 R2), and the
    // operation's own rows are still only staged.
    assert_eq!(before.apply_routed_batch_calls, 1);
    assert_eq!(before.flushes, 0);
    assert!(appender.has_staged()?);
    appender.commit_staged()?;
    assert_eq!(store.snapshot().since(&before).apply_routed_batch_calls, 1);
    assert!(matches!(
        appender.commit_staged(),
        Err(OperationLogError::RoutedStream(
            RoutedStreamRefusal::BatchRefused
        ))
    ));
    assert!(matches!(
        appender.append(&StoredOperation::Genesis { event: vec![2] }, 1),
        Err(OperationLogError::RoutedStream(
            RoutedStreamRefusal::BatchRefused
        ))
    ));
    assert_eq!(store.snapshot().since(&before).apply_routed_batch_calls, 1);
    Ok(())
}

#[test]
fn an_empty_appender_commit_preserves_the_named_refusal_without_issuance()
-> Result<(), Box<dyn Error>> {
    let store = Arc::new(CountingStore::new(MemoryStore::default()));
    let shared = Arc::clone(&store) as Arc<dyn DurableStore>;
    let registry = ConversationRegistry::new(Arc::clone(&shared));
    let log = OperationLog::new(Arc::clone(&shared), 56);
    let outbox_log = OutboxLog::new(shared, 56);
    let batch = RefCell::new(Some(OperationBatch::new(&checkpoint_conversation_route(
        56,
    ))));
    let publisher = CheckpointPublisher::default();
    let appender = LogAppender {
        log: &log,
        registry: &registry,
        conversation_id: 56,
        outbox_log: &outbox_log,
        prune_keep: None,
        publisher: &publisher,
        outstanding_extension_rows: Cell::new(0),
        batch: Some(&batch),
        timestamp: 10,
    };
    assert!(!appender.has_staged()?);
    assert!(matches!(
        appender.commit_staged(),
        Err(OperationLogError::Durability(
            DurabilityError::EmptyOperationBatch
        ))
    ));
    assert!(!log.issuance().was_submitted());
    assert_eq!(store.snapshot().apply_routed_batch_calls, 0);
    assert_eq!(store.snapshot().append_calls, 0);
    Ok(())
}