sipp-rs 0.1.0

Unified Rust library for extensible Sipp inference
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

use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};

use crate::client::{SippCancellationHandle, SippCancellationReason, SippRequestContext};

/// Stable reason attached to gateway cancellation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GatewayCancellationReason {
    /// The downstream client disconnected.
    ClientDisconnected,
    /// The hosting application is shutting down.
    ServerShutdown,
    /// The application explicitly cancelled the request.
    CallerCancelled,
    /// The application deadline expired.
    DeadlineExceeded,
}

impl GatewayCancellationReason {
    const fn into_client(self) -> SippCancellationReason {
        match self {
            Self::ClientDisconnected => SippCancellationReason::ClientDisconnected,
            Self::ServerShutdown => SippCancellationReason::ServerShutdown,
            Self::CallerCancelled => SippCancellationReason::CallerCancelled,
            Self::DeadlineExceeded => SippCancellationReason::DeadlineExceeded,
        }
    }
}

#[derive(Default)]
struct CancellationState {
    reason: Option<GatewayCancellationReason>,
    runs: Vec<SippCancellationHandle>,
}

/// Cancellation source shared by application adapters and execution.
#[derive(Clone, Default)]
pub struct GatewayCancellation {
    state: Arc<Mutex<CancellationState>>,
}

impl GatewayCancellation {
    /// Cancel every client run registered with this request.
    pub fn cancel(&self, reason: GatewayCancellationReason) {
        let handles = {
            let Ok(mut state) = self.state.lock() else {
                return;
            };
            if state.reason.is_some() {
                return;
            }
            state.reason = Some(reason);
            std::mem::take(&mut state.runs)
        };
        for handle in handles {
            handle.cancel(reason.into_client());
        }
    }

    /// Return the first cancellation reason.
    pub fn reason(&self) -> Option<GatewayCancellationReason> {
        self.state.lock().ok().and_then(|state| state.reason)
    }

    pub(crate) fn register(&self, handle: SippCancellationHandle) {
        let reason = {
            let Ok(mut state) = self.state.lock() else {
                return;
            };
            match state.reason {
                Some(reason) => Some(reason),
                None => {
                    state.runs.push(handle.clone());
                    None
                }
            }
        };
        if let Some(reason) = reason {
            handle.cancel(reason.into_client());
        }
    }
}

/// Request metadata propagated through application policy and execution.
#[derive(Clone, Default)]
pub struct GatewayRequestContext {
    /// Application-assigned request identifier.
    pub request_id: Option<String>,
    /// Application-defined metadata available to policy implementations.
    pub metadata: BTreeMap<String, serde_json::Value>,
    /// Request cancellation source.
    pub cancellation: GatewayCancellation,
}

impl GatewayRequestContext {
    /// Create a context for an optional application request identifier.
    pub fn new(request_id: Option<String>) -> Self {
        Self {
            request_id,
            ..Self::default()
        }
    }

    /// Attach application-defined metadata.
    pub fn with_metadata(mut self, metadata: BTreeMap<String, serde_json::Value>) -> Self {
        self.metadata = metadata;
        self
    }

    pub(crate) fn client_context(&self) -> SippRequestContext {
        SippRequestContext {
            request_id: self.request_id.clone(),
        }
    }
}