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
76
77
78
79
80
81
82
83
84
85

use super::super::restore_checkpoint::{CheckpointError, CheckpointPositions, CheckpointSchema};
use super::ConversationAuthority;

impl ConversationAuthority {
    pub(in crate::server::participant::production) fn checkpoint_positions(
        &self,
        schema: CheckpointSchema,
        merged_base_sequence: u64,
        merged_extension_sequence: u64,
    ) -> Result<CheckpointPositions, CheckpointError> {
        if self.pending_debt_dispatch_transition.is_some() {
            return Err(CheckpointError::Position {
                field: "pending debt transition",
            });
        }
        let shell = self.shell.as_ref().ok_or(CheckpointError::Position {
            field: "absent conversation shell",
        })?;
        if shell.conversation_id() != self.conversation_id {
            return Err(CheckpointError::ForeignConversation {
                expected: self.conversation_id,
                actual: shell.conversation_id(),
            });
        }
        let covered_log_sequence =
            self.next_log_sequence
                .checked_sub(1)
                .ok_or(CheckpointError::Position {
                    field: "uncommitted conversation",
                })?;
        let outbox = self.outbox.as_ref().ok_or(CheckpointError::Position {
            field: "absent outbox",
        })?;
        let (extension_sequence, highest_delivery_seq) = outbox.checkpoint_boundaries();
        let positions = CheckpointPositions {
            conversation_id: self.conversation_id,
            covered_log_sequence,
            next_event_ordinal: shell.next_event_ordinal(),
            next_log_sequence: self.next_log_sequence,
            next_seq: self.next_seq,
            next_order: self.next_order,
            next_participant: self.next_participant,
            extension_sequence,
            highest_delivery_seq,
            observer_progress: self.observer_progress,
            merged_base_sequence,
            merged_extension_sequence,
            schema,
        };
        positions.validate()?;
        Ok(positions)
    }
}

impl ConversationAuthority {
    pub(in crate::server::participant::production) const fn checkpoint_debt_owner(
        &self,
    ) -> Option<&liminal_protocol::lifecycle::ObligationDebtDispatchState> {
        self.obligation_debt_dispatch.as_ref()
    }

    pub(in crate::server::participant::production) const fn checkpoint_specific_fates(
        &self,
    ) -> &std::collections::BTreeMap<u64, super::PendingSpecificFate> {
        &self.pending_specific_fates
    }

    pub(in crate::server::participant::production) const fn checkpoint_finalizers(
        &self,
    ) -> &std::collections::BTreeMap<u64, super::PreparedOrdinaryFinalizer> {
        &self.prepared_ordinary_finalizers
    }

    pub(in crate::server::participant::production) const fn checkpoint_observer_witnesses(
        &self,
    ) -> &super::super::observer_progress::ObserverProgressWitnessState {
        &self.observer_progress_witnesses
    }

    pub(in crate::server::participant::production) const fn checkpoint_transition_pending(
        &self,
    ) -> bool {
        self.pending_debt_dispatch_transition.is_some()
    }
}