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

//! SRV-043 R2: a connection the server's stop closes before its readiness
//! registers is logged below ERROR and torn down exactly as a failed one.

use std::error::Error;
use std::net::{TcpListener, TcpStream};
use std::sync::{Arc, Mutex};

use beamr::native::native_process::NativeOutcome;
use beamr::process::ExitReason;
use tracing_subscriber::prelude::*;

use super::ConnectionProcess;
use crate::ServerError;
use crate::server::connection::supervisor::runtime::ConnectionRuntime;
use crate::server::connection::worker_front_door::WorkerFrontDoorServices;

/// Every event `run` logs on this thread, as its level and message.
pub(in crate::server::connection) fn logged_while<T>(
    run: impl FnOnce() -> T,
) -> (T, Vec<(tracing::Level, String)>) {
    let lines = Arc::new(Mutex::new(Vec::new()));
    let subscriber = tracing_subscriber::registry().with(Lines(Arc::clone(&lines)));
    let value = tracing::subscriber::with_default(subscriber, run);
    let logged = lines.lock().map(|lines| lines.clone()).unwrap_or_default();
    (value, logged)
}

struct Lines(Arc<Mutex<Vec<(tracing::Level, String)>>>);

struct Message(String);

impl tracing::field::Visit for Message {
    fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
        if field.name() == "message" {
            self.0 = format!("{value:?}");
        }
    }
}

impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for Lines {
    fn on_event(&self, event: &tracing::Event<'_>, _: tracing_subscriber::layer::Context<'_, S>) {
        let mut message = Message(String::new());
        event.record(&mut message);
        if let Ok(mut lines) = self.0.lock() {
            lines.push((*event.metadata().level(), message.0));
        }
    }
}

/// The SRV-043 R2 check over one transport: the stop's refusal logs no
/// ERROR and one DEBUG naming the stop, a real failure still logs ERROR, and
/// both end the slice the same way.
pub(in crate::server::connection) fn check_stop_and_failure(
    stopped: (NativeOutcome, Vec<(tracing::Level, String)>),
    failed: (NativeOutcome, Vec<(tracing::Level, String)>),
) {
    let (stopped, stopped_lines) = stopped;
    let (failed, failed_lines) = failed;
    assert!(
        stopped_lines
            .iter()
            .all(|(level, _)| *level != tracing::Level::ERROR),
        "a connection the stop closed logs no ERROR: {stopped_lines:?}"
    );
    assert!(
        stopped_lines
            .iter()
            .any(|(level, message)| *level == tracing::Level::DEBUG
                && message.contains("closed by the server's stop")),
        "the stop's close is logged at DEBUG: {stopped_lines:?}"
    );
    assert!(
        failed_lines
            .iter()
            .any(|(level, message)| *level == tracing::Level::ERROR
                && message.contains("connection readiness contract failed")),
        "a real readiness failure still logs ERROR: {failed_lines:?}"
    );
    assert!(matches!(stopped, NativeOutcome::Stop(ExitReason::Error)));
    assert!(matches!(failed, NativeOutcome::Stop(ExitReason::Error)));
}

fn tcp_process() -> Result<(ConnectionProcess, TcpStream), Box<dyn Error>> {
    let listener = TcpListener::bind("127.0.0.1:0")?;
    let client = TcpStream::connect(listener.local_addr()?)?;
    let (server, _) = listener.accept()?;
    server.set_nonblocking(true)?;
    let runtime = Arc::new(ConnectionRuntime::for_tests(Arc::new(
        WorkerFrontDoorServices::new(),
    )));
    let holder = Arc::new(Mutex::new(Some(server)));
    Ok((
        ConnectionProcess::from_holder(runtime, None, &holder, None),
        client,
    ))
}

#[test]
fn a_tcp_connection_the_stop_closes_logs_no_error_and_is_torn_down_as_a_failed_one()
-> Result<(), Box<dyn Error>> {
    let (mut stopped, _stopped_client) = tcp_process()?;
    let (mut failed, _failed_client) = tcp_process()?;
    check_stop_and_failure(
        logged_while(|| stopped.fail_slice(7, &ServerError::ReadinessClosedByStop { pid: 7 })),
        logged_while(|| {
            failed.fail_slice(
                8,
                &ServerError::ListenerAccept {
                    message:
                        "failed to register connection readiness: readiness poll thread has failed"
                            .to_owned(),
                },
            )
        }),
    );
    Ok(())
}