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    },
32    /// A server-initiated frame (`id == PUSH_ID`, WIRE-005).
33    Push(Response),
34    /// Drain what is queued, flush, exit.
35    Shutdown,
36}
37
38/// Session state shared between the read loop and every dispatch task
39/// (SRV-010): the auth flag is a lock-free atomic flipped by `HELLO`/`AUTH`
40/// and read by the dispatch path without locks.
41#[derive(Debug)]
42pub struct Session {
43    connection_id: u64,
44    authenticated: AtomicBool,
45    principal: Mutex<Option<Principal>>,
46    push: Option<PushSender>,
47}
48
49impl Session {
50    /// New session. `pre_authenticated` is true for `Handshake::None`
51    /// profiles (no RPC-layer auth, SRV-011); `push` is present only under
52    /// `PushPolicy::Enabled` (SRV-013).
53    pub(crate) fn new(
54        connection_id: u64,
55        pre_authenticated: bool,
56        push: Option<PushSender>,
57    ) -> Self {
58        Self {
59            connection_id,
60            authenticated: AtomicBool::new(pre_authenticated),
61            principal: Mutex::new(None),
62            push,
63        }
64    }
65
66    /// Listener-scoped connection id, surfaced in the metadata-shape HELLO
67    /// reply (SRV-014).
68    pub fn connection_id(&self) -> u64 {
69        self.connection_id
70    }
71
72    /// Lock-free read of the auth flag (SRV-010).
73    pub fn is_authenticated(&self) -> bool {
74        self.authenticated.load(Ordering::Acquire)
75    }
76
77    /// The principal resolved by the last successful `HELLO`/`AUTH`.
78    pub fn principal(&self) -> Option<Principal> {
79        self.principal
80            .lock()
81            .unwrap_or_else(PoisonError::into_inner)
82            .clone()
83    }
84
85    /// Typed push channel for this connection (SRV-013). `Some` only under
86    /// `push = Enabled` profiles; under `Reserved` no push emission is
87    /// possible.
88    pub fn push_sender(&self) -> Option<&PushSender> {
89        self.push.as_ref()
90    }
91
92    /// Store the authenticated principal, then flip the flag — the
93    /// `Release`/`Acquire` pair makes the principal visible to any task
94    /// that observes `is_authenticated() == true` (SRV-010).
95    pub(crate) fn set_principal(&self, principal: Principal) {
96        *self
97            .principal
98            .lock()
99            .unwrap_or_else(PoisonError::into_inner) = Some(principal);
100        self.authenticated.store(true, Ordering::Release);
101    }
102}
103
104/// Typed, clonable handle for server-initiated push frames (SRV-013).
105///
106/// Handed to product code via [`Session::push_sender`] under
107/// `push = Enabled` profiles. It wraps the connection's writer channel and
108/// forces `id = PUSH_ID`, so product code can never collide with request
109/// ids. Clones stay valid for the connection's lifetime — subscription
110/// flows (a subscribe-style command) may emit long after the registering request
111/// completed.
112#[derive(Debug, Clone)]
113pub struct PushSender {
114    tx: mpsc::Sender<WriteJob>,
115}
116
117impl PushSender {
118    pub(crate) fn new(tx: mpsc::Sender<WriteJob>) -> Self {
119        Self { tx }
120    }
121
122    /// Emit one push frame carrying `value`. Fails once the connection has
123    /// closed and its writer drained (SRV-004).
124    pub async fn push(&self, value: Value) -> Result<(), PushClosed> {
125        self.tx
126            .send(WriteJob::Push(Response::ok(PUSH_ID, value)))
127            .await
128            .map_err(|_| PushClosed)
129    }
130}
131
132/// The connection behind a [`PushSender`] is gone; the frame was dropped.
133#[derive(Debug, thiserror::Error)]
134#[error("connection closed; push frame dropped")]
135pub struct PushClosed;
thunder/server/session.rs

thunder/server/
session.rs
