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, ReconcileReport, SessionId, Shard, ShardCommand, ShardHandle,
18    ShardId, 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        self.create_shard_with_reports(build, |_report| {})
169    }
170
171    /// Create a shard and observe every reconciliation outcome.
172    ///
173    /// The observer runs on the shard task immediately after a turn, including
174    /// valid turns with no delta and refused renders. It must return immediately
175    /// and publish any application data through a non-blocking primitive.
176    pub fn create_shard_with_reports<L, O>(
177        &self,
178        build: impl FnOnce(ShardHandle) -> L,
179        observer: O,
180    ) -> ShardHandle
181    where
182        L: ShardLogic,
183        O: FnMut(&ReconcileReport) + Send + 'static,
184    {
185        let id = ShardId(self.inner.next_shard.fetch_add(1, Ordering::Relaxed));
186        let (handle, wake, mailbox) = mumble_server_runtime_shard::spawn_parts(id);
187        let logic = build(handle.clone());
188        let mut shard = Shard::with_ids(id, logic, self.inner.ids.clone());
189        shard.route_effects(self.effects());
190        let routing = shard.routing();
191        let task = tokio::spawn(async move {
192            let _shard =
193                mumble_server_runtime_shard::run_with_reports(shard, wake, mailbox, observer).await;
194        });
195
196        write(&self.inner.shards).insert(
197            id,
198            ShardEntry {
199                handle: handle.clone(),
200                plane: ShardPlane { shard: id, routing },
201                task,
202            },
203        );
204        handle
205    }
206
207    /// Destroy a shard.
208    ///
209    /// The guide leaves the policy for its connections open (17.4). The safe
210    /// default, and what this does, is to close them: a connection left pointing
211    /// at a shard that no longer renders would hold a view nothing can ever
212    /// update. A flavor that wants a fallback shard migrates them first and
213    /// destroys afterwards.
214    pub fn destroy_shard(&self, shard: ShardId, reason: &str) {
215        let refused = self.inner.commands.try_send(RuntimeCommand::Destroy {
216            shard,
217            reason: reason.to_owned(),
218        });
219        if let Err(error) = refused {
220            eprintln!("mumble-server-runtime-gateway: cannot destroy shard {shard:?}: {error}");
221        }
222    }
223
224    /// What a shard hands back when its flavor asks for something only the
225    /// runtime can do.
226    ///
227    /// Weak on purpose. The closure lives inside the shard, the shard inside its
228    /// task, and the task inside this runtime's directory: holding a strong
229    /// reference here would close that ring, and the runtime would outlive every
230    /// handle to it forever. An effect arriving after the runtime is gone has
231    /// nothing left to act on, which is exactly what a failed upgrade says.
232    fn effects(&self) -> mumble_server_runtime_shard::Effects {
233        let runtime = Arc::downgrade(&self.inner);
234        Arc::new(move |effect| {
235            let Some(inner) = runtime.upgrade() else {
236                return;
237            };
238            match effect {
239                mumble_server_runtime_shard::Effect::Move { connection, to } => {
240                    RuntimeHandle { inner }.move_connection(connection, to);
241                }
242                // The vocabulary is non-exhaustive: a shard that starts asking
243                // for something new must not have it silently ignored.
244                other => {
245                    eprintln!("mumble-server-runtime-gateway: no runtime support for {other:?}")
246                }
247            }
248        })
249    }
250
251    /// Move a connection to another shard.
252    ///
253    /// Not a disconnect followed by a connect. The source shard hands over the
254    /// view the client still holds, and the destination plans **one** transition
255    /// from it onto its own tree, which is the same slow path a scope change
256    /// takes: whatever the two trees have in common does not flicker.
257    ///
258    /// Doing it the other way round is not merely wasteful, it disconnects the
259    /// official client. It keeps itself in its own model after a `UserRemove`
260    /// naming its own session, so the `ChannelRemove` that follows looks to it
261    /// like the server removing an occupied channel, which it treats as a
262    /// protocol violation.
263    ///
264    /// REF: runtime/references/mumble/src/mumble/Messages.cpp : `msgUserRemove` calls
265    ///   `removeUser` only `if (pDst != pSelf)`; `msgChannelRemove` logs
266    ///   "Protocol violation. Server sent remove for occupied channel." and
267    ///   disconnects when `UserModel::removeChannel(c, true)` refuses.
268    pub fn move_connection(&self, connection: ConnectionId, to: ShardId) {
269        let refused = self
270            .inner
271            .commands
272            .try_send(RuntimeCommand::Move { connection, to });
273        if let Err(error) = refused {
274            eprintln!(
275                "mumble-server-runtime-gateway: cannot move {connection:?} to {to:?}: {error}"
276            );
277        }
278    }
279
280    /// Attach a connection that has just been routed.
281    ///
282    /// Returns the receiver that fires once the shard has accepted the
283    /// connection's first transition, which is what the handshake waits on
284    /// before sending `ServerSync`.
285    ///
286    /// # Errors
287    ///
288    /// When the shard does not exist, or its mailbox is full.
289    pub fn attach(
290        &self,
291        peer: &Arc<Peer>,
292        shard: ShardId,
293    ) -> Result<oneshot::Receiver<()>, AttachError> {
294        let plane = self.plane(shard).ok_or(AttachError::NoSuchShard(shard))?;
295        let handle = self
296            .shard_handle(shard)
297            .ok_or(AttachError::NoSuchShard(shard))?;
298        peer.move_to(plane);
299
300        let (ready, awaited) = oneshot::channel();
301        handle
302            .send(ShardCommand::Attach {
303                connection: peer.connection(),
304                queue: peer.queue(),
305                cursor: peer.cursor_cell(),
306                held: mumble_server_runtime_shard::ShardView::empty(),
307                ready: Some(ready),
308            })
309            .map_err(|_full| AttachError::Unreachable(shard))?;
310        Ok(awaited)
311    }
312
313    /// Tell a connection's shard that it is gone.
314    pub fn detach(&self, connection: ConnectionId, shard: ShardId, reason: &str) {
315        let Some(handle) = self.shard_handle(shard) else {
316            return;
317        };
318        // A full mailbox here would leave the shard rendering a connection that
319        // no longer exists, so it is reported rather than swallowed. The next
320        // render still cannot reach it: its queue is closed.
321        if let Err(error) = handle.send(ShardCommand::detach(connection, reason)) {
322            eprintln!(
323                "mumble-server-runtime-gateway: shard {shard:?} did not take a detach: {error}"
324            );
325        }
326    }
327
328    /// Forward a command to a shard.
329    ///
330    /// # Errors
331    ///
332    /// When the shard does not exist or its mailbox is full.
333    pub fn send(&self, shard: ShardId, command: ShardCommand) -> Result<(), AttachError> {
334        let handle = self
335            .shard_handle(shard)
336            .ok_or(AttachError::NoSuchShard(shard))?;
337        handle
338            .send(command)
339            .map_err(|_full| AttachError::Unreachable(shard))?;
340        Ok(())
341    }
342
343    /// What an operator would want on a status page.
344    #[must_use]
345    pub fn status(&self) -> Vec<ShardStatus> {
346        let shards: Vec<ShardId> = read(&self.inner.shards).keys().copied().collect();
347        shards
348            .into_iter()
349            .map(|shard| {
350                let peers = self.inner.peers.on_shard(shard);
351                let cursors: Vec<u64> = peers.iter().map(|peer| peer.cursor()).collect();
352                let worst_lag = match (cursors.iter().min(), cursors.iter().max()) {
353                    (Some(behind), Some(ahead)) => ahead.saturating_sub(*behind),
354                    _ => 0,
355                };
356                ShardStatus {
357                    shard,
358                    connections: peers.len(),
359                    worst_lag,
360                }
361            })
362            .collect()
363    }
364
365    #[must_use]
366    pub fn shard_handle(&self, shard: ShardId) -> Option<ShardHandle> {
367        read(&self.inner.shards)
368            .get(&shard)
369            .map(|entry| entry.handle.clone())
370    }
371
372    #[must_use]
373    fn plane(&self, shard: ShardId) -> Option<ShardPlane> {
374        read(&self.inner.shards)
375            .get(&shard)
376            .map(|entry| entry.plane.clone())
377    }
378}
379
380/// Why a connection could not reach a shard.
381#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
382pub enum AttachError {
383    #[error("shard {0:?} does not exist")]
384    NoSuchShard(ShardId),
385    #[error("shard {0:?} is not taking commands")]
386    Unreachable(ShardId),
387}
388
389/// The runtime's own task: the only place two shards are touched at once.
390async fn supervise(inner: Arc<RuntimeInner>, mut mailbox: mpsc::Receiver<RuntimeCommand>) {
391    // Migrations run concurrently and are owned: a migration that never
392    // completes because a shard stopped answering is reaped when the set is
393    // dropped, rather than left running unowned.
394    let mut migrations: JoinSet<()> = JoinSet::new();
395
396    loop {
397        tokio::select! {
398            // Cancellation-safe: `recv` takes a command only when it completes.
399            command = mailbox.recv() => match command {
400                Some(RuntimeCommand::Move { connection, to }) => {
401                    let inner = Arc::clone(&inner);
402                    migrations.spawn(migrate(inner, connection, to));
403                }
404                Some(RuntimeCommand::Destroy { shard, reason }) => destroy(&inner, shard, &reason),
405                // Every handle is gone.
406                None => break,
407            },
408            // Cancellation-safe: joining is idempotent, and a task not reaped on
409            // this pass is reaped on the next. Disabled when empty so the branch
410            // cannot resolve to `None` in a busy loop.
411            Some(_finished) = migrations.join_next(), if !migrations.is_empty() => {}
412        }
413    }
414}
415
416/// Move one connection between two shards.
417async fn migrate(inner: Arc<RuntimeInner>, connection: ConnectionId, to: ShardId) {
418    let Some(peer) = inner.peers.by_connection(connection) else {
419        // It disconnected while the command was in flight. Nothing to move.
420        return;
421    };
422    let from = peer.shard();
423    if from == to {
424        return;
425    }
426
427    let (source, destination) = {
428        let shards = read(&inner.shards);
429        let source = shards.get(&from).map(|entry| entry.handle.clone());
430        let destination = shards
431            .get(&to)
432            .map(|entry| (entry.handle.clone(), entry.plane.clone()));
433        (source, destination)
434    };
435    let Some((destination, plane)) = destination else {
436        eprintln!(
437            "mumble-server-runtime-gateway: {connection:?} cannot move to {to:?}: no such shard"
438        );
439        return;
440    };
441
442    // The held view has to come from the source before the destination can plan
443    // onto it. If the source is gone the client's view is unknown, so the
444    // connection is closed rather than handed a transition planned from a guess.
445    let held = match source {
446        Some(source) => {
447            let (view, awaited) = oneshot::channel();
448            let sent = source.send(ShardCommand::Detach {
449                connection,
450                reason: format!("moving to {to:?}"),
451                handover: Some(Handover { to, view }),
452            });
453            if sent.is_err() {
454                peer.close();
455                return;
456            }
457            match awaited.await {
458                Ok(held) => held,
459                Err(_dropped) => {
460                    peer.close();
461                    return;
462                }
463            }
464        }
465        None => {
466            peer.close();
467            return;
468        }
469    };
470
471    // The routing table follows the connection before its view does: a route it
472    // is no longer entitled to disappears immediately, and a route it gains is
473    // inert until its cursor catches up (guide 9.5). Cutting early is always
474    // safe; the reverse never is.
475    peer.move_to(plane);
476
477    let handed = destination.send(ShardCommand::Attach {
478        connection,
479        queue: peer.queue(),
480        cursor: peer.cursor_cell(),
481        held,
482        ready: None,
483    });
484    if handed.is_err() {
485        // Attached nowhere and holding a view no shard will ever update.
486        eprintln!("mumble-server-runtime-gateway: {connection:?} could not be handed to {to:?}");
487        peer.close();
488    }
489}
490
491fn destroy(inner: &Arc<RuntimeInner>, shard: ShardId, reason: &str) {
492    // Dropping the entry aborts the shard's task, so the connections have to be
493    // dealt with first: after this, nothing can render them.
494    for peer in inner.peers.on_shard(shard) {
495        eprintln!(
496            "mumble-server-runtime-gateway: closing {:?}: its shard was destroyed ({reason})",
497            peer.connection()
498        );
499        peer.close();
500    }
501    write(&inner.shards).remove(&shard);
502}
503
504/// A poisoned directory means a panic unwound mid-update. The map is still
505/// structurally sound, and refusing every later lookup would take the runtime
506/// down over one thread.
507fn read<T>(lock: &RwLock<T>) -> RwLockReadGuard<'_, T> {
508    lock.read().unwrap_or_else(PoisonError::into_inner)
509}
510
511fn write<T>(lock: &RwLock<T>) -> RwLockWriteGuard<'_, T> {
512    lock.write().unwrap_or_else(PoisonError::into_inner)
513}
514
515impl Drop for ShardEntry {
516    fn drop(&mut self) {
517        self.task.abort();
518    }
519}
mumble_server_runtime_gateway/runtime.rs

mumble_server_runtime_gateway/
runtime.rs
