liminal-server 0.3.3

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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141

use liminal_protocol::lifecycle::{CommittedDiedTerminal, LiveFrontierOwner};
use liminal_protocol::wire::ParticipantId;

use super::binding_fate_completion::{StoredSpecificFateIntentExt, intent_matches_token};
use super::log::{StoredOrdinaryTerminalSource, StoredSpecificFateIntent};
use super::state::{
    ConversationAuthority, PendingBindingFate, PendingSpecificFate, PreparedOrdinaryFinalizer,
    StateError,
};

#[derive(Clone, Copy)]
enum FinalizerMeasurement {
    BeforeEnclosing,
    AfterFencedProof,
}

impl ConversationAuthority {
    /// Measures Ordinary before an enclosing finalizer removes its participant frontier.
    pub(super) fn prepare_pending_died_finalizer(
        &mut self,
        participant_id: ParticipantId,
        died_source_sequence: u64,
        terminal: CommittedDiedTerminal,
        source: StoredOrdinaryTerminalSource,
        owner: LiveFrontierOwner,
    ) -> Result<(LiveFrontierOwner, bool), StateError> {
        self.prepare_pending_died_finalizer_at(
            participant_id,
            died_source_sequence,
            terminal,
            source,
            FinalizerMeasurement::BeforeEnclosing,
            owner,
        )
    }

    /// Measures Ordinary after fenced proof minting consumes marker authority.
    pub(super) fn prepare_pending_died_fenced_finalizer(
        &mut self,
        participant_id: ParticipantId,
        died_source_sequence: u64,
        terminal: CommittedDiedTerminal,
        source: StoredOrdinaryTerminalSource,
        owner: LiveFrontierOwner,
    ) -> Result<(LiveFrontierOwner, bool), StateError> {
        self.prepare_pending_died_finalizer_at(
            participant_id,
            died_source_sequence,
            terminal,
            source,
            FinalizerMeasurement::AfterFencedProof,
            owner,
        )
    }

    fn prepare_pending_died_finalizer_at(
        &mut self,
        participant_id: ParticipantId,
        died_source_sequence: u64,
        terminal: CommittedDiedTerminal,
        source: StoredOrdinaryTerminalSource,
        measurement: FinalizerMeasurement,
        owner: LiveFrontierOwner,
    ) -> Result<(LiveFrontierOwner, bool), StateError> {
        let Some(pending) = self.pending_specific_fates.remove(&participant_id) else {
            return Ok((owner, false));
        };
        if pending.died_source_sequence != died_source_sequence
            || !matches!(pending.intent, StoredSpecificFateIntent::Ordinary { .. })
            || pending.terminal.is_some()
        {
            return Err(StateError::invariant(
                "pending Died finalizer disagrees with its open Ordinary intent",
            ));
        }
        let attached_source_sequence = pending.binding_fate.attached_source_sequence;
        if attached_source_sequence != pending.intent.attached_source_sequence()
            || !intent_matches_token(pending.intent, &pending.binding_fate)
        {
            return Err(StateError::invariant(
                "pending Died finalizer lost its sealed Ordinary authority",
            ));
        }
        let prepared = match measurement {
            FinalizerMeasurement::BeforeEnclosing => owner.prepare_pending_died_ordinary_finalizer(
                pending.binding_fate.token,
                terminal,
                self.observer_progress,
            ),
            FinalizerMeasurement::AfterFencedProof => owner
                .prepare_pending_died_ordinary_after_fenced_proof(
                    pending.binding_fate.token,
                    terminal,
                    self.observer_progress,
                ),
        };
        let prepared = match prepared {
            Ok(prepared) => prepared,
            Err(refused) => {
                let error = refused.error();
                let (owner, token, _) = refused.into_parts();
                self.pending_specific_fates.insert(
                    participant_id,
                    PendingSpecificFate {
                        died_source_sequence,
                        intent: pending.intent,
                        terminal: None,
                        binding_fate: PendingBindingFate {
                            attached_source_sequence,
                            token,
                        },
                    },
                );
                self.install_frontier(owner)?;
                return Err(StateError::invariant(format!(
                    "pending Died finalizer measurement refused: {error:?}"
                )));
            }
        };
        let (owner, fate, finalizer) = prepared.into_parts();
        if self
            .prepared_ordinary_finalizers
            .insert(
                participant_id,
                PreparedOrdinaryFinalizer {
                    attached_source_sequence,
                    terminal,
                    terminal_source: source,
                    fate,
                    finalizer,
                },
            )
            .is_some()
        {
            return Err(StateError::invariant(
                "pending Died finalizer replaced prepared Ordinary authority",
            ));
        }
        Ok((owner, true))
    }
}