liminal-server 0.25.4

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

//! The connection process's inbound buffer: complete frames decoded and
//! applied in order.

use crate::ServerError;
use crate::server::connection::outbound::OutboundWriter;
use crate::server::connection::services::server_error_from_protocol;
use crate::server::connection::state::{ConnectionProcessState, FrameAction, ProcessStatus};
use crate::server::connection::supervisor::relay_pump::apply_frame;
use crate::server::connection::supervisor::runtime::ConnectionRuntime;
use crate::server::participant::ConnectionFateClass;
use liminal::protocol::{ProtocolError, decode};

pub(super) fn process_buffer(
    pid: u64,
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    buffer: &mut Vec<u8>,
    outbound: &mut OutboundWriter,
) -> Result<ProcessStatus, ServerError> {
    loop {
        if runtime.force_close_pending(pid) {
            return Ok(ProcessStatus::CloseWithFate(
                ConnectionFateClass::ServerShutdown,
            ));
        }
        if let Some(rejection) = crate::server::participant::preflight_generic_bytes(
            buffer,
            state.authenticated,
            state.participant_session,
        ) {
            let response = crate::server::participant::encode_server_value(
                liminal_protocol::wire::ServerValue::ParticipantTransportRejected(rejection),
            )
            .map_err(|error| ServerError::ListenerAccept {
                message: format!("failed to encode participant frame-limit rejection: {error:?}"),
            })?;
            outbound
                .enqueue_frame(&response)
                .map_err(|error| ServerError::ListenerAccept {
                    message: format!(
                        "failed to enqueue participant frame-limit rejection: {error}"
                    ),
                })?;
            buffer.clear();
            return Ok(ProcessStatus::Close);
        }
        if let Err(refusal) =
            super::super::pre_admission::check_declared_frame(buffer, state.authenticated)
        {
            tracing::warn!(connection_pid = pid, %refusal, "unadmitted inbound frame refused");
            buffer.clear();
            if let Err(enqueue_error) =
                outbound.enqueue_frame(&super::super::pre_admission::connect_error_frame(refusal))
            {
                tracing::warn!(
                    connection_pid = pid,
                    error = %enqueue_error,
                    "pre-admission refusal frame could not be enqueued; closing anyway"
                );
            }
            return Ok(ProcessStatus::Close);
        }
        let (frame, consumed) = match decode(buffer) {
            Ok(decoded) => decoded,
            Err(
                ProtocolError::IncompleteHeader { .. } | ProtocolError::TruncatedPayload { .. },
            ) => {
                return Ok(ProcessStatus::Continue);
            }
            Err(error) => {
                let conversations = state.participant_conversations.tracked_conversations();
                if runtime.connection_has_bound_participant(
                    state.connection_incarnation,
                    &conversations,
                )? {
                    tracing::warn!(connection_pid = pid, %error, "bound connection protocol refusal");
                    return Ok(ProcessStatus::CloseWithFate(
                        ConnectionFateClass::ProtocolError,
                    ));
                }
                return Err(server_error_from_protocol(&error));
            }
        };
        buffer.drain(..consumed);
        match apply_frame(pid, runtime, state, frame) {
            FrameAction::Respond(response) => {
                outbound
                    .enqueue_frame(&response)
                    .map_err(|error| ServerError::ListenerAccept {
                        message: format!("failed to enqueue connection response: {error}"),
                    })?;
            }
            FrameAction::NoResponse => {}
            FrameAction::RespondThenClose(response) => {
                // Best-effort: enqueue the rejection frame (a `ConnectError` from the
                // auth gate) so the slice's Close-path drain can flush it, then close
                // regardless. Unlike `Respond`, a failed enqueue here is logged and
                // swallowed, never propagated: a rejected connection must be torn down
                // even when its rejection notice cannot be queued. The single
                // best-effort drain happens on the `ProcessStatus::Close` path in
                // `service_socket`, so no write is forced (or awaited) here.
                if let Err(error) = outbound.enqueue_frame(&response) {
                    tracing::warn!(
                        connection_pid = pid,
                        %error,
                        "auth-rejection frame could not be enqueued; closing anyway"
                    );
                }
                return Ok(ProcessStatus::Close);
            }
            FrameAction::Close => return Ok(ProcessStatus::Close),
            FrameAction::CloseWithFate(class) => {
                return Ok(ProcessStatus::CloseWithFate(class));
            }
        }
    }
}