liminal-server 0.23.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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112

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

impl ConversationAuthority {
    /// Whether the positions a checkpoint of this owner would carry now are
    /// valid, at the owner's own log and extension heads. `validate` never
    /// reads the schema (it is carried, not checked), so any phase answers
    /// the same; the crossing that stages the checkpoint computes the real
    /// positions with its own schema.
    ///
    /// # Errors
    ///
    /// The position the checkpoint could not carry, by name.
    pub(in crate::server::participant::production) fn checkpoint_positions_framable(
        &self,
    ) -> Result<(), CheckpointError> {
        let extension = self
            .outbox
            .as_ref()
            .ok_or(CheckpointError::Position {
                field: "absent outbox",
            })?
            .next_extension_sequence();
        self.checkpoint_positions(
            CheckpointSchema::V3Suffix,
            self.next_log_sequence,
            extension,
        )
        .map(|_| ())
    }

    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()
    }
}