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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131

use super::ParticipantSemanticHandler;
use super::{
    ConnectionFateClass, ConnectionIncarnation, ConnectionRuntime, ParticipantServiceFatal,
    ServerError,
};

impl ConnectionRuntime {
    pub(super) fn participant_service_fatal(
        &self,
    ) -> Result<Option<ParticipantServiceFatal>, ServerError> {
        let Some(service) = self.participant_service() else {
            return Ok(None);
        };
        service
            .service_fatal()
            .map_err(|error| ServerError::ParticipantIncarnation {
                phase: "participant fatal latch inspection",
                message: error.to_string(),
            })
    }

    fn activate_fatal_shutdown(&self) {
        if let Some(shutdown) = self.fatal_shutdown.as_ref() {
            shutdown.initiate();
        }
    }

    pub(super) fn ensure_participant_service_live(&self) -> Result<(), ServerError> {
        let Some(fatal) = self.participant_service_fatal()? else {
            return Ok(());
        };
        self.activate_fatal_shutdown();
        Err(ServerError::ParticipantServiceFatal { fatal })
    }

    fn latch_connection_fate_intent_incomplete(
        &self,
        open_sequence: u64,
        conversation_id: u64,
    ) -> Result<ParticipantServiceFatal, ServerError> {
        let Some(service) = self.participant_service() else {
            return Err(ServerError::ParticipantIncarnation {
                phase: "connection-fate fatal latch",
                message: "a durable Open lacks its installed participant service".to_owned(),
            });
        };
        let fatal = service
            .latch_connection_fate_intent_incomplete(open_sequence, conversation_id)
            .map_err(|error| ServerError::ParticipantIncarnation {
                phase: "connection-fate fatal latch",
                message: error.to_string(),
            })?;
        self.activate_fatal_shutdown();
        Ok(fatal)
    }

    pub(super) fn complete_connection_fate_fatal(
        &self,
        open_sequence: u64,
        conversations: &[u64],
        phase: &'static str,
        error: &impl std::fmt::Display,
    ) -> ServerError {
        tracing::error!(open_sequence, phase, %error, "durable connection-fate intent is incomplete");
        let Some(&conversation_id) = conversations.first() else {
            return ServerError::ParticipantIncarnation {
                phase: "connection-fate fatal target",
                message: "a durable Open has no tracked conversation target".to_owned(),
            };
        };
        match self.latch_connection_fate_intent_incomplete(open_sequence, conversation_id) {
            Ok(fatal) => ServerError::ParticipantServiceFatal { fatal },
            Err(latch_error) => latch_error,
        }
    }

    /// Runs one typed terminal fold after classification and before teardown.
    pub(in crate::server::connection) fn complete_connection_fate(
        &self,
        connection_incarnation: Option<ConnectionIncarnation>,
        class: ConnectionFateClass,
        conversations: &[u64],
    ) -> Result<(), ServerError> {
        if conversations.is_empty() {
            return Ok(());
        }
        self.ensure_participant_service_live()?;
        let (Some(connection_incarnation), Some(_), Some(authority)) = (
            connection_incarnation,
            self.participant_service(),
            self.incarnations.as_ref(),
        ) else {
            return Err(ServerError::ParticipantIncarnation {
                phase: "connection-fate authority composition",
                message: "tracked participant conversations lack a complete service/incarnation authority"
                    .to_owned(),
            });
        };
        let intent =
            authority.open_connection_fate(connection_incarnation, class, conversations)?;
        self.fold_or_park_connection_fate(intent)
    }

    /// Resolves the bound-only protocol-error gate from participant authority.
    pub(in crate::server::connection) fn connection_has_bound_participant(
        &self,
        connection_incarnation: Option<ConnectionIncarnation>,
        conversations: &[u64],
    ) -> Result<bool, ServerError> {
        if conversations.is_empty() {
            return Ok(false);
        }
        self.ensure_participant_service_live()?;
        let (Some(connection_incarnation), Some(service)) =
            (connection_incarnation, self.participant_service())
        else {
            return Err(ServerError::ParticipantIncarnation {
                phase: "bound participant classification",
                message:
                    "tracked participant conversations lack a complete service/incarnation pair"
                        .to_owned(),
            });
        };
        service
            .connection_has_bound_participant(connection_incarnation, conversations)
            .map_err(|error| ServerError::ParticipantIncarnation {
                phase: "bound participant classification",
                message: error.to_string(),
            })
    }
}