mumble_server_runtime_gateway/
runtime.rs

1//! Several shards, one runtime, and the only operation that touches two shards
2//! at once.
3//!
4//! Shards are independent by design: each owns a task, a view, a journal and a
5//! routing table, and nothing here coordinates their turns. What the runtime
6//! adds is the small amount of shared state that genuinely cannot be per shard -
7//! the identifier allocator, the connection registry, the shard directory - plus
8//! the one interaction that spans two of them, [`RuntimeHandle::move_connection`].
9//!
10//! REF: docs/design/guide-implementation.md 9.6, 10
11
12use std::collections::HashMap;
13use std::sync::atomic::{AtomicU64, Ordering};
14use std::sync::{Arc, PoisonError, RwLock, RwLockReadGuard, RwLockWriteGuard};
15
16use mumble_server_runtime_shard::{
17    ConnectionId, Handover, Occupant, SessionId, Shard, ShardCommand, ShardHandle, ShardId,
18    ShardLogic, SharedIds,
19};
20use tokio::sync::{mpsc, oneshot};
21use tokio::task::{JoinHandle, JoinSet};
22
23use crate::peer::{Peer, Peers, ShardPlane};
24
25/// How many runtime commands queue before a caller is refused.
26const MAILBOX_DEPTH: usize = 256;
27
28/// What an operator can see about one shard without touching its task.
29#[derive(Debug, Clone, PartialEq, Eq)]
30pub struct ShardStatus {
31    pub shard: ShardId,
32    pub connections: usize,
33    /// The furthest any attached connection is behind the others, in journal
34    /// versions. The guide calls this the best single health indicator, and it
35    /// is: a connection that stops advancing is one that has stopped keeping up.
36    pub worst_lag: u64,
37}
38
39/// A command only the runtime supervisor may carry out.
40#[derive(Debug)]
41enum RuntimeCommand {
42    Move {
43        connection: ConnectionId,
44        to: ShardId,
45    },
46    Destroy {
47        shard: ShardId,
48        reason: String,
49    },
50}
51
52/// One shard, as the runtime tracks it.
53struct ShardEntry {
54    handle: ShardHandle,
55    plane: ShardPlane,
56    /// Owning the task is what makes destruction real: dropping the entry
57    /// aborts it. A detached task would keep rendering a shard nobody can reach.
58    task: JoinHandle<()>,
59}
60
61struct RuntimeInner {
62    ids: SharedIds,
63    peers: Arc<Peers>,
64    shards: RwLock<HashMap<ShardId, ShardEntry>>,
65    next_shard: AtomicU64,
66    next_connection: AtomicU64,
67    commands: mpsc::Sender<RuntimeCommand>,
68}
69
70/// The runtime, owned by whoever started it.
71///
72/// Holding it keeps the supervisor task alive; dropping it ends the supervisor
73/// and, with it, every shard.
74pub struct Runtime {
75    handle: RuntimeHandle,
76    supervisor: JoinHandle<()>,
77}
78
79impl Runtime {
80    /// Build a runtime and start its supervisor.
81    #[must_use]
82    pub fn start() -> Runtime {
83        let (commands, mailbox) = mpsc::channel(MAILBOX_DEPTH);
84        let inner = Arc::new(RuntimeInner {
85            ids: SharedIds::new(),
86            peers: Arc::new(Peers::new()),
87            shards: RwLock::new(HashMap::new()),
88            next_shard: AtomicU64::new(1),
89            next_connection: AtomicU64::new(1),
90            commands,
91        });
92        let supervisor = tokio::spawn(supervise(Arc::clone(&inner), mailbox));
93        Runtime {
94            handle: RuntimeHandle { inner },
95            supervisor,
96        }
97    }
98
99    #[must_use]
100    pub fn handle(&self) -> RuntimeHandle {
101        self.handle.clone()
102    }
103
104    /// Stop the supervisor and every shard.
105    pub fn shutdown(self) {
106        self.supervisor.abort();
107        write(&self.handle.inner.shards).clear();
108    }
109}
110
111/// A cheap clone of the runtime, safe to hand to flavors and connection tasks.
112#[derive(Clone)]
113pub struct RuntimeHandle {
114    inner: Arc<RuntimeInner>,
115}
116
117impl std::fmt::Debug for RuntimeHandle {
118    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
119        formatter
120            .debug_struct("RuntimeHandle")
121            .field("shards", &read(&self.inner.shards).len())
122            .field("connections", &self.inner.peers.len())
123            .finish()
124    }
125}
126
127impl RuntimeHandle {
128    /// The runtime-wide identifier allocator.
129    ///
130    /// Public because the gateway needs a connection's session before any shard
131    /// has rendered it: `ServerSync` carries it, and the handshake happens
132    /// before the first turn.
133    #[must_use]
134    pub fn ids(&self) -> &SharedIds {
135        &self.inner.ids
136    }
137
138    #[must_use]
139    pub fn peers(&self) -> &Arc<Peers> {
140        &self.inner.peers
141    }
142
143    /// Reserve the next connection identifier. Never reused.
144    #[must_use]
145    pub fn next_connection(&self) -> ConnectionId {
146        ConnectionId(self.inner.next_connection.fetch_add(1, Ordering::Relaxed))
147    }
148
149    /// The session a connection will be rendered under, allocated now so the
150    /// handshake can announce it.
151    ///
152    /// # Errors
153    ///
154    /// Once the session space is spent.
155    pub fn session_for(
156        &self,
157        connection: ConnectionId,
158    ) -> Result<SessionId, mumble_server_runtime_shard::Exhausted> {
159        self.inner.ids.session(Occupant::Connection(connection))
160    }
161
162    /// Create a shard and start its task.
163    ///
164    /// The chicken and the egg - a flavor usually wants to wake the shard that
165    /// owns it - are resolved by a closure: the handle exists before the logic
166    /// that will hold it.
167    pub fn create_shard<L: ShardLogic>(&self, build: impl FnOnce(ShardHandle) -> L) -> ShardHandle {
168        let id = ShardId(self.inner.next_shard.fetch_add(1, Ordering::Relaxed));
169        let (handle, wake, mailbox) = mumble_server_runtime_shard::spawn_parts(id);
170        let logic = build(handle.clone());
171        let mut shard = Shard::with_ids(id, logic, self.inner.ids.clone());
172        shard.route_effects(self.effects());
173        let routing = shard.routing();
174        let task = tokio::spawn(async move {
175            let _shard = mumble_server_runtime_shard::run(shard, wake, mailbox).await;
176        });
177
178        write(&self.inner.shards).insert(
179            id,
180            ShardEntry {
181                handle: handle.clone(),
182                plane: ShardPlane { shard: id, routing },
183                task,
184            },
185        );
186        handle
187    }
188
189    /// Destroy a shard.
190    ///
191    /// The guide leaves the policy for its connections open (17.4). The safe
192    /// default, and what this does, is to close them: a connection left pointing
193    /// at a shard that no longer renders would hold a view nothing can ever
194    /// update. A flavor that wants a fallback shard migrates them first and
195    /// destroys afterwards.
196    pub fn destroy_shard(&self, shard: ShardId, reason: &str) {
197        let refused = self.inner.commands.try_send(RuntimeCommand::Destroy {
198            shard,
199            reason: reason.to_owned(),
200        });
201        if let Err(error) = refused {
202            eprintln!("mumble-server-runtime-gateway: cannot destroy shard {shard:?}: {error}");
203        }
204    }
205
206    /// What a shard hands back when its flavor asks for something only the
207    /// runtime can do.
208    ///
209    /// Weak on purpose. The closure lives inside the shard, the shard inside its
210    /// task, and the task inside this runtime's directory: holding a strong
211    /// reference here would close that ring, and the runtime would outlive every
212    /// handle to it forever. An effect arriving after the runtime is gone has
213    /// nothing left to act on, which is exactly what a failed upgrade says.
214    fn effects(&self) -> mumble_server_runtime_shard::Effects {
215        let runtime = Arc::downgrade(&self.inner);
216        Arc::new(move |effect| {
217            let Some(inner) = runtime.upgrade() else {
218                return;
219            };
220            match effect {
221                mumble_server_runtime_shard::Effect::Move { connection, to } => {
222                    RuntimeHandle { inner }.move_connection(connection, to);
223                }
224                // The vocabulary is non-exhaustive: a shard that starts asking
225                // for something new must not have it silently ignored.
226                other => {
227                    eprintln!("mumble-server-runtime-gateway: no runtime support for {other:?}")
228                }
229            }
230        })
231    }
232
233    /// Move a connection to another shard.
234    ///
235    /// Not a disconnect followed by a connect. The source shard hands over the
236    /// view the client still holds, and the destination plans **one** transition
237    /// from it onto its own tree, which is the same slow path a scope change
238    /// takes: whatever the two trees have in common does not flicker.
239    ///
240    /// Doing it the other way round is not merely wasteful, it disconnects the
241    /// official client. It keeps itself in its own model after a `UserRemove`
242    /// naming its own session, so the `ChannelRemove` that follows looks to it
243    /// like the server removing an occupied channel, which it treats as a
244    /// protocol violation.
245    ///
246    /// REF: references/mumble/src/mumble/Messages.cpp : `msgUserRemove` calls
247    ///   `removeUser` only `if (pDst != pSelf)`; `msgChannelRemove` logs
248    ///   "Protocol violation. Server sent remove for occupied channel." and
249    ///   disconnects when `UserModel::removeChannel(c, true)` refuses.
250    pub fn move_connection(&self, connection: ConnectionId, to: ShardId) {
251        let refused = self
252            .inner
253            .commands
254            .try_send(RuntimeCommand::Move { connection, to });
255        if let Err(error) = refused {
256            eprintln!(
257                "mumble-server-runtime-gateway: cannot move {connection:?} to {to:?}: {error}"
258            );
259        }
260    }
261
262    /// Attach a connection that has just been routed.
263    ///
264    /// Returns the receiver that fires once the shard has accepted the
265    /// connection's first transition, which is what the handshake waits on
266    /// before sending `ServerSync`.
267    ///
268    /// # Errors
269    ///
270    /// When the shard does not exist, or its mailbox is full.
271    pub fn attach(
272        &self,
273        peer: &Arc<Peer>,
274        shard: ShardId,
275    ) -> Result<oneshot::Receiver<()>, AttachError> {
276        let plane = self.plane(shard).ok_or(AttachError::NoSuchShard(shard))?;
277        let handle = self
278            .shard_handle(shard)
279            .ok_or(AttachError::NoSuchShard(shard))?;
280        peer.move_to(plane);
281
282        let (ready, awaited) = oneshot::channel();
283        handle
284            .send(ShardCommand::Attach {
285                connection: peer.connection(),
286                queue: peer.queue(),
287                cursor: peer.cursor_cell(),
288                held: mumble_server_runtime_shard::ShardView::empty(),
289                ready: Some(ready),
290            })
291            .map_err(|_full| AttachError::Unreachable(shard))?;
292        Ok(awaited)
293    }
294
295    /// Tell a connection's shard that it is gone.
296    pub fn detach(&self, connection: ConnectionId, shard: ShardId, reason: &str) {
297        let Some(handle) = self.shard_handle(shard) else {
298            return;
299        };
300        // A full mailbox here would leave the shard rendering a connection that
301        // no longer exists, so it is reported rather than swallowed. The next
302        // render still cannot reach it: its queue is closed.
303        if let Err(error) = handle.send(ShardCommand::detach(connection, reason)) {
304            eprintln!(
305                "mumble-server-runtime-gateway: shard {shard:?} did not take a detach: {error}"
306            );
307        }
308    }
309
310    /// Forward a command to a shard.
311    ///
312    /// # Errors
313    ///
314    /// When the shard does not exist or its mailbox is full.
315    pub fn send(&self, shard: ShardId, command: ShardCommand) -> Result<(), AttachError> {
316        let handle = self
317            .shard_handle(shard)
318            .ok_or(AttachError::NoSuchShard(shard))?;
319        handle
320            .send(command)
321            .map_err(|_full| AttachError::Unreachable(shard))?;
322        Ok(())
323    }
324
325    /// What an operator would want on a status page.
326    #[must_use]
327    pub fn status(&self) -> Vec<ShardStatus> {
328        let shards: Vec<ShardId> = read(&self.inner.shards).keys().copied().collect();
329        shards
330            .into_iter()
331            .map(|shard| {
332                let peers = self.inner.peers.on_shard(shard);
333                let cursors: Vec<u64> = peers.iter().map(|peer| peer.cursor()).collect();
334                let worst_lag = match (cursors.iter().min(), cursors.iter().max()) {
335                    (Some(behind), Some(ahead)) => ahead.saturating_sub(*behind),
336                    _ => 0,
337                };
338                ShardStatus {
339                    shard,
340                    connections: peers.len(),
341                    worst_lag,
342                }
343            })
344            .collect()
345    }
346
347    #[must_use]
348    pub fn shard_handle(&self, shard: ShardId) -> Option<ShardHandle> {
349        read(&self.inner.shards)
350            .get(&shard)
351            .map(|entry| entry.handle.clone())
352    }
353
354    #[must_use]
355    fn plane(&self, shard: ShardId) -> Option<ShardPlane> {
356        read(&self.inner.shards)
357            .get(&shard)
358            .map(|entry| entry.plane.clone())
359    }
360}
361
362/// Why a connection could not reach a shard.
363#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
364pub enum AttachError {
365    #[error("shard {0:?} does not exist")]
366    NoSuchShard(ShardId),
367    #[error("shard {0:?} is not taking commands")]
368    Unreachable(ShardId),
369}
370
371/// The runtime's own task: the only place two shards are touched at once.
372async fn supervise(inner: Arc<RuntimeInner>, mut mailbox: mpsc::Receiver<RuntimeCommand>) {
373    // Migrations run concurrently and are owned: a migration that never
374    // completes because a shard stopped answering is reaped when the set is
375    // dropped, rather than left running unowned.
376    let mut migrations: JoinSet<()> = JoinSet::new();
377
378    loop {
379        tokio::select! {
380            // Cancellation-safe: `recv` takes a command only when it completes.
381            command = mailbox.recv() => match command {
382                Some(RuntimeCommand::Move { connection, to }) => {
383                    let inner = Arc::clone(&inner);
384                    migrations.spawn(migrate(inner, connection, to));
385                }
386                Some(RuntimeCommand::Destroy { shard, reason }) => destroy(&inner, shard, &reason),
387                // Every handle is gone.
388                None => break,
389            },
390            // Cancellation-safe: joining is idempotent, and a task not reaped on
391            // this pass is reaped on the next. Disabled when empty so the branch
392            // cannot resolve to `None` in a busy loop.
393            Some(_finished) = migrations.join_next(), if !migrations.is_empty() => {}
394        }
395    }
396}
397
398/// Move one connection between two shards.
399async fn migrate(inner: Arc<RuntimeInner>, connection: ConnectionId, to: ShardId) {
400    let Some(peer) = inner.peers.by_connection(connection) else {
401        // It disconnected while the command was in flight. Nothing to move.
402        return;
403    };
404    let from = peer.shard();
405    if from == to {
406        return;
407    }
408
409    let (source, destination) = {
410        let shards = read(&inner.shards);
411        let source = shards.get(&from).map(|entry| entry.handle.clone());
412        let destination = shards
413            .get(&to)
414            .map(|entry| (entry.handle.clone(), entry.plane.clone()));
415        (source, destination)
416    };
417    let Some((destination, plane)) = destination else {
418        eprintln!(
419            "mumble-server-runtime-gateway: {connection:?} cannot move to {to:?}: no such shard"
420        );
421        return;
422    };
423
424    // The held view has to come from the source before the destination can plan
425    // onto it. If the source is gone the client's view is unknown, so the
426    // connection is closed rather than handed a transition planned from a guess.
427    let held = match source {
428        Some(source) => {
429            let (view, awaited) = oneshot::channel();
430            let sent = source.send(ShardCommand::Detach {
431                connection,
432                reason: format!("moving to {to:?}"),
433                handover: Some(Handover { to, view }),
434            });
435            if sent.is_err() {
436                peer.close();
437                return;
438            }
439            match awaited.await {
440                Ok(held) => held,
441                Err(_dropped) => {
442                    peer.close();
443                    return;
444                }
445            }
446        }
447        None => {
448            peer.close();
449            return;
450        }
451    };
452
453    // The routing table follows the connection before its view does: a route it
454    // is no longer entitled to disappears immediately, and a route it gains is
455    // inert until its cursor catches up (guide 9.5). Cutting early is always
456    // safe; the reverse never is.
457    peer.move_to(plane);
458
459    let handed = destination.send(ShardCommand::Attach {
460        connection,
461        queue: peer.queue(),
462        cursor: peer.cursor_cell(),
463        held,
464        ready: None,
465    });
466    if handed.is_err() {
467        // Attached nowhere and holding a view no shard will ever update.
468        eprintln!("mumble-server-runtime-gateway: {connection:?} could not be handed to {to:?}");
469        peer.close();
470    }
471}
472
473fn destroy(inner: &Arc<RuntimeInner>, shard: ShardId, reason: &str) {
474    // Dropping the entry aborts the shard's task, so the connections have to be
475    // dealt with first: after this, nothing can render them.
476    for peer in inner.peers.on_shard(shard) {
477        eprintln!(
478            "mumble-server-runtime-gateway: closing {:?}: its shard was destroyed ({reason})",
479            peer.connection()
480        );
481        peer.close();
482    }
483    write(&inner.shards).remove(&shard);
484}
485
486/// A poisoned directory means a panic unwound mid-update. The map is still
487/// structurally sound, and refusing every later lookup would take the runtime
488/// down over one thread.
489fn read<T>(lock: &RwLock<T>) -> RwLockReadGuard<'_, T> {
490    lock.read().unwrap_or_else(PoisonError::into_inner)
491}
492
493fn write<T>(lock: &RwLock<T>) -> RwLockWriteGuard<'_, T> {
494    lock.write().unwrap_or_else(PoisonError::into_inner)
495}
496
497impl Drop for ShardEntry {
498    fn drop(&mut self) {
499        self.task.abort();
500    }
501}
mumble_server_runtime_gateway/runtime.rs

mumble_server_runtime_gateway/
runtime.rs
