liminal-server 0.24.9

//! The terminal tap belongs to the same operation that owns the correlated reply.

use super::state::{ConnectionProcessState, FrameAction};
use super::supervisor::runtime::ConnectionRuntime;
use super::worker_route::Request;
use crate::console::ConsoleHub;
use crate::tap::message::CallFact;
use crate::tap::{TapOutcome, TapSource};
use liminal::protocol::{CONVERSATION_REPLY_REQUESTED_FLAG, Frame, MessageEnvelope};
use std::sync::Arc;
use std::time::Instant;

#[derive(Debug, Default)]
pub(super) enum CallDestination {
    #[default]
    NotRouted,
    Queued(Option<Arc<str>>),
}

#[derive(Debug)]
pub(super) struct CallTrace {
    console: Arc<ConsoleHub>,
    subject: String,
    size: u64,
    connection: u64,
    station: Option<Arc<str>>,
    served_by: Option<Arc<str>>,
    received_at: Instant,
    forwarded_at: Instant,
    dispatched: bool,
    operation: Option<u64>,
}

impl CallTrace {
    pub(super) const fn dispatched_at(&mut self, at: Instant) {
        self.forwarded_at = at;
        self.dispatched = true;
    }

    pub(super) fn finish(self, outcome: TapOutcome, refusal: Option<&str>, finished_at: Instant) {
        let fact = CallFact {
            subject: &self.subject,
            size: self.size,
            connection: self.connection,
            operation: self.operation,
            station: self.station.as_ref(),
            served_by: if outcome == TapOutcome::Refused && !self.dispatched {
                None
            } else {
                self.served_by.as_ref()
            },
            received_at: self.received_at,
            forwarded_at: self.forwarded_at,
            finished_at,
            outcome,
            refusal,
        };
        if let Err(error) = self.console.tap().observe_call(&fact) {
            self.console.observe_failed();
            tracing::warn!(%error, subject = self.subject, "SERVICE_TAP_OBSERVE_FAILED");
        }
    }

    pub(super) fn frame(self, frame: &Frame) {
        let now = Instant::now();
        match frame {
            Frame::ConversationMessage { .. } => self.finish(TapOutcome::Answered, None, now),
            Frame::ConversationError { message, .. } => {
                self.finish(TapOutcome::Refused, message.as_deref(), now);
            }
            _ => self.finish(
                TapOutcome::Refused,
                Some("service returned no conversation outcome"),
                now,
            ),
        }
    }
}

pub(super) fn conversation_message(
    pid: u64,
    runtime: &ConnectionRuntime,
    state: &mut ConnectionProcessState,
    request: Request,
    envelope: &MessageEnvelope,
) -> FrameAction {
    state.call_station = CallDestination::NotRouted;
    let received_at = Instant::now();
    let trace = state
        .conversations
        .get(&request.conversation_id)
        .filter(|conversation| crate::console::tappable(conversation.subject()))
        .and_then(|conversation| {
            let console = runtime.signals.get()?.console()?;
            match trace(
                console,
                pid,
                conversation.subject(),
                envelope.payload.len(),
                received_at,
            ) {
                Ok(trace) => Some(trace),
                Err(error) => {
                    console.observe_failed();
                    tracing::warn!(%error, "SERVICE_TAP_BEGIN_FAILED");
                    None
                }
            }
        });
    let operation = state.pending_replies.next_operation();
    let action = super::apply::conversation_message(pid, runtime, state, request, envelope);
    let station = std::mem::take(&mut state.call_station);
    if let Some(mut trace) = trace {
        if let CallDestination::Queued(station) = station {
            trace.served_by = station;
        }
        match &action {
            FrameAction::Respond(frame) => trace.frame(frame),
            FrameAction::NoResponse if request.flags & CONVERSATION_REPLY_REQUESTED_FLAG != 0 => {
                trace.operation = Some(operation);
                state.pending_replies.bind_tap(operation, trace);
            }
            FrameAction::NoResponse => {
                trace.dispatched_at(Instant::now());
                trace.finish(
                    TapOutcome::Unanswered,
                    Some("reply_not_requested"),
                    Instant::now(),
                );
            }
            _ => trace.finish(
                TapOutcome::Refused,
                Some("connection_closed"),
                Instant::now(),
            ),
        }
    }
    action
}

fn trace(
    console: &Arc<ConsoleHub>,
    connection: u64,
    subject: &str,
    size: usize,
    received_at: Instant,
) -> Result<CallTrace, crate::tap::TapError> {
    let station = console
        .tap()
        .source_station(TapSource::Connection(connection))?;
    let served_by = console.serving_station(subject)?.map(Arc::from);
    Ok(CallTrace {
        console: Arc::clone(console),
        subject: subject.to_owned(),
        size: u64::try_from(size).unwrap_or(u64::MAX),
        connection,
        station,
        served_by,
        received_at,
        forwarded_at: Instant::now(),
        dispatched: false,
        operation: None,
    })
}