thunder/server/session.rs
1//! Per-connection session state (SRV-010) and the typed push channel
2//! (SRV-013).
3
4use std::sync::atomic::{AtomicBool, Ordering};
5use std::sync::{Mutex, PoisonError};
6use std::time::Duration;
7
8use crate::wire::{Response, Value, PUSH_ID};
9use tokio::sync::mpsc;
10
11use crate::server::dispatch::Principal;
12
13/// One job for the connection's writer task (SRV-002).
14///
15/// Dispatch tasks and read-loop built-ins enqueue [`WriteJob::Response`];
16/// product code enqueues [`WriteJob::Push`] through a [`PushSender`]; the
17/// read side enqueues [`WriteJob::Shutdown`] once every in-flight dispatch
18/// task has finished, so the writer exits even while product-held
19/// `PushSender` clones keep the channel open.
20#[derive(Debug)]
21pub(crate) enum WriteJob {
22 /// A response to a client request plus the metadata the writer records
23 /// after the successful write (SRV-030).
24 Response {
25 response: Response,
26 /// Request frame size straight from the decoder (SRV-007) — never
27 /// re-encoded.
28 in_bytes: usize,
29 /// Dispatch duration for the duration / slow counters.
30 duration: Duration,
31 /// Command name, carried **only** when a [`MetricsObserver`] is
32 /// installed — the observer needs the label, and the built-in metrics
33 /// do not. `None` on the default path, so a listener without an
34 /// observer allocates nothing extra per request.
35 ///
36 /// [`MetricsObserver`]: crate::server::MetricsObserver
37 command: Option<Box<str>>,
38 },
39 /// A server-initiated frame (`id == PUSH_ID`, WIRE-005).
40 Push(Response),
41 /// Drain what is queued, flush, exit.
42 Shutdown,
43}
44
45/// Session state shared between the read loop and every dispatch task
46/// (SRV-010): the auth flag is a lock-free atomic flipped by `HELLO`/`AUTH`
47/// and read by the dispatch path without locks.
48#[derive(Debug)]
49pub struct Session<I = ()> {
50 connection_id: u64,
51 authenticated: AtomicBool,
52 principal: Mutex<Option<Principal<I>>>,
53 push: Option<PushSender>,
54}
55
56impl<I> Session<I> {
57 /// New session. `pre_authenticated` is true for `Handshake::None`
58 /// profiles (no RPC-layer auth, SRV-011); `push` is present only under
59 /// `PushPolicy::Enabled` (SRV-013).
60 pub(crate) fn new(
61 connection_id: u64,
62 pre_authenticated: bool,
63 push: Option<PushSender>,
64 ) -> Self {
65 Self {
66 connection_id,
67 authenticated: AtomicBool::new(pre_authenticated),
68 principal: Mutex::new(None),
69 push,
70 }
71 }
72
73 /// Listener-scoped connection id, surfaced in the metadata-shape HELLO
74 /// reply (SRV-014).
75 pub fn connection_id(&self) -> u64 {
76 self.connection_id
77 }
78
79 /// Lock-free read of the auth flag (SRV-010).
80 pub fn is_authenticated(&self) -> bool {
81 self.authenticated.load(Ordering::Acquire)
82 }
83
84 /// Read the resolved principal **without cloning it** — the
85 /// authorization path (SRV-012).
86 ///
87 /// Authorization runs on every privileged command, so it must not pay a
88 /// `String` clone (or a credential-store lookup) to ask one question:
89 ///
90 /// ```ignore
91 /// let is_admin = session.with_principal(|p| {
92 /// p.map(|p| p.identity.is_admin).unwrap_or(false)
93 /// });
94 /// ```
95 pub fn with_principal<R>(&self, read: impl FnOnce(Option<&Principal<I>>) -> R) -> R {
96 let guard = self
97 .principal
98 .lock()
99 .unwrap_or_else(PoisonError::into_inner);
100 read(guard.as_ref())
101 }
102
103 /// The principal's name, if one was resolved.
104 ///
105 /// Convenience for the common case; prefer
106 /// [`with_principal`](Self::with_principal) on a hot authorization path,
107 /// since this clones the name.
108 pub fn principal_name(&self) -> Option<String> {
109 self.with_principal(|p| p.map(|p| p.name.clone()))
110 }
111
112 /// Typed push channel for this connection (SRV-013). `Some` only under
113 /// `push = Enabled` profiles; under `Reserved` no push emission is
114 /// possible.
115 pub fn push_sender(&self) -> Option<&PushSender> {
116 self.push.as_ref()
117 }
118
119 /// Store the authenticated principal, then flip the flag — the
120 /// `Release`/`Acquire` pair makes the principal visible to any task
121 /// that observes `is_authenticated() == true` (SRV-010).
122 pub(crate) fn set_principal(&self, principal: Principal<I>) {
123 *self
124 .principal
125 .lock()
126 .unwrap_or_else(PoisonError::into_inner) = Some(principal);
127 self.authenticated.store(true, Ordering::Release);
128 }
129}
130
131impl<I: Clone> Session<I> {
132 /// A clone of the principal resolved by the last successful
133 /// `HELLO`/`AUTH`.
134 ///
135 /// Available when the identity is `Clone`; on an authorization hot path
136 /// prefer [`with_principal`](Session::with_principal), which clones
137 /// nothing.
138 pub fn principal(&self) -> Option<Principal<I>> {
139 self.with_principal(|p| p.cloned())
140 }
141}
142
143/// Typed, clonable handle for server-initiated push frames (SRV-013).
144///
145/// Handed to product code via [`Session::push_sender`] under
146/// `push = Enabled` profiles. It wraps the connection's writer channel and
147/// forces `id = PUSH_ID`, so product code can never collide with request
148/// ids. Clones stay valid for the connection's lifetime — subscription
149/// flows (a subscribe-style command) may emit long after the registering request
150/// completed.
151#[derive(Debug, Clone)]
152pub struct PushSender {
153 tx: mpsc::Sender<WriteJob>,
154}
155
156impl PushSender {
157 pub(crate) fn new(tx: mpsc::Sender<WriteJob>) -> Self {
158 Self { tx }
159 }
160
161 /// Emit one push frame carrying `value`. Fails once the connection has
162 /// closed and its writer drained (SRV-004).
163 pub async fn push(&self, value: Value) -> Result<(), PushClosed> {
164 self.tx
165 .send(WriteJob::Push(Response::ok(PUSH_ID, value)))
166 .await
167 .map_err(|_| PushClosed)
168 }
169}
170
171/// The connection behind a [`PushSender`] is gone; the frame was dropped.
172#[derive(Debug, thiserror::Error)]
173#[error("connection closed; push frame dropped")]
174pub struct PushClosed;