liminal-server 0.4.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
113
114
115
116
117
118
119

use liminal_protocol::lifecycle::{
    BindingState, CommittedDiedTerminal, LiveFrontierOwner, PendingFinalization,
};
use liminal_protocol::wire::{CloseCause, CredentialAttachRequest, ParticipantId};

use super::fate_occurrence::{FateOccurrenceKey, PendingFinalizerRoute};
use super::log::{
    StoredAttachModeV3, StoredComposedTerminalCause, StoredComposedTerminalKind,
    StoredFinalizerPresentation, StoredOrdinaryTerminalSource, StoredPendingDiedFinalizer,
};
use super::state::{ConversationAuthority, StateError};

#[derive(Clone, Copy)]
pub(super) struct SelectedFencedFinalizer {
    route: PendingFinalizerRoute,
    committed_died_terminal: Option<CommittedDiedTerminal>,
}

impl SelectedFencedFinalizer {
    pub(super) const fn is_non_presenting(self) -> bool {
        matches!(
            self.route.presentation,
            StoredFinalizerPresentation::ConsumeRecoveredReservation { .. }
        )
    }
}

impl ConversationAuthority {
    pub(super) fn prepare_selected_fenced_finalizer(
        &mut self,
        participant_id: ParticipantId,
        source_sequence: u64,
        finalizer: Option<SelectedFencedFinalizer>,
        owner: LiveFrontierOwner,
    ) -> Result<(LiveFrontierOwner, bool), StateError> {
        match finalizer {
            Some(SelectedFencedFinalizer {
                route,
                committed_died_terminal: Some(terminal),
            }) => self.prepare_pending_died_fenced_finalizer(
                participant_id,
                route.pending_source_sequence,
                terminal,
                StoredOrdinaryTerminalSource::PendingDiedFinalized {
                    died_source_sequence: route.pending_source_sequence,
                    finalizer: StoredPendingDiedFinalizer::FencedAttached { source_sequence },
                },
                owner,
            ),
            Some(_) | None => Ok((owner, false)),
        }
    }

    pub(super) fn select_fenced_finalizer(
        &mut self,
        binding: BindingState,
        mode: &StoredAttachModeV3,
        request: &CredentialAttachRequest,
    ) -> Result<Option<SelectedFencedFinalizer>, StateError> {
        let BindingState::PendingFinalization(pending) = binding else {
            return Ok(None);
        };
        let StoredAttachModeV3::Fenced {
            composed_terminal: Some(terminal),
            ..
        } = mode
        else {
            return Err(StateError::invariant(
                "pending attach finalizer requires a composed fenced terminal",
            ));
        };
        let route = self.fate_occurrences.select_finalizer(FateOccurrenceKey {
            conversation_id: request.conversation_id,
            participant_id: request.participant_id,
            binding_epoch: pending.binding_epoch(),
        })?;
        let expected_kind = match pending {
            PendingFinalization::Died(_) => StoredComposedTerminalKind::Died,
            PendingFinalization::Detached(_) => StoredComposedTerminalKind::Detached,
        };
        let expected_cause = stored_composed_cause(pending.close_cause())?;
        if terminal.kind != expected_kind
            || terminal.cause != expected_cause
            || terminal.transaction_order != pending.admission_order().transaction_order()
            || terminal.pending_source_sequence != route.pending_source_sequence
            || terminal.presentation != route.presentation
        {
            return Err(StateError::invariant(
                "fenced attach composed terminal differs from selected pending finalizer",
            ));
        }
        let committed_died_terminal = match pending {
            PendingFinalization::Died(died) => Some(died.commit(terminal.delivery_seq)),
            PendingFinalization::Detached(_) => None,
        };
        Ok(Some(SelectedFencedFinalizer {
            route,
            committed_died_terminal,
        }))
    }
}

fn stored_composed_cause(cause: CloseCause) -> Result<StoredComposedTerminalCause, StateError> {
    match cause {
        CloseCause::CleanDeregister => Ok(StoredComposedTerminalCause::CleanDeregister),
        CloseCause::ConnectionLost => Ok(StoredComposedTerminalCause::ConnectionLost),
        CloseCause::ProcessKilled => Ok(StoredComposedTerminalCause::ProcessKilled),
        CloseCause::ProtocolError => Ok(StoredComposedTerminalCause::ProtocolError),
        CloseCause::ServerShutdown => Ok(StoredComposedTerminalCause::ServerShutdown),
        CloseCause::UncleanServerRestart {
            prior_server_incarnation,
        } => Ok(StoredComposedTerminalCause::UncleanServerRestart {
            prior_server_incarnation,
        }),
        CloseCause::Superseded => Err(StateError::invariant(
            "superseded terminal cannot be a pending fenced finalizer",
        )),
    }
}