smarm/src/gen_server.rs

//! gen_server — synchronous call / asynchronous cast over a single actor.
//!
//! A thin request-reply layer on top of [`channel`](crate::channel()), modelled
//! on Erlang's `gen_server`. A *server* is an actor owning a state value that
//! implements [`GenServer`]; clients hold a clonable [`ServerRef`] and issue
//! [`call`](ServerRef::call) (synchronous, returns a reply) or
//! [`cast`](ServerRef::cast) (fire-and-forget).
//!
//! ## One inbox, many arms
//!
//! Every call and cast travels the *same* inbox channel as an [`Envelope`]
//! (calls and casts are ordered relative to each other, like Erlang); the
//! server loop dispatches by variant. A `call` carries a freshly made
//! one-shot reply channel; the server sends the reply straight back down it.
//!
//! Out-of-band messages ride *separate* channels composed at the wait via
//! [`select`](channel::select): info channels handed over at start
//! ([`ServerBuilder::with_info`]) are dispatched to
//! [`handle_info`](GenServer::handle_info). Arm priority is
//! **infos before inbox**, in declaration order — a hot inbox cannot starve
//! an out-of-band message; conversely a hot info channel CAN starve the
//! inbox, deliberately (system-message semantics). An info channel whose
//! senders are all gone is silently dropped from the arm set (per the
//! closed-arm-is-ready-forever rule on `select`); a closed *inbox* still
//! means graceful shutdown.
//!
//! ## Server death
//!
//! Detection falls out of channel closure, so no monitor is required:
//!   - if the server is already gone, its inbox is closed and the `send` in
//!     `call`/`cast` fails → [`CallError::ServerDown`] / [`CastError::ServerDown`];
//!   - if the server dies *after* a call is enqueued but before it replies
//!     (a handler panic, or a cooperative `request_stop`), the reply sender is
//!     dropped as the server's stack unwinds, closing the reply channel; the
//!     parked caller wakes and its `recv` returns `Err` → `ServerDown`.
//!
//! ## Lifecycle / callbacks
//!
//! [`GenServer::init`] runs once inside the server actor before the first
//! message; [`GenServer::terminate`] runs on the way out. terminate is wired
//! through a drop guard, so it fires on *every* exit path — graceful inbox
//! close, a handler panic, or a cooperative `request_stop` — not only the clean
//! one. Keep it cheap and non-blocking: it may run mid-unwind, and a panic
//! inside it during an unwind aborts the process (a double panic).
//!
//! ## Time (timers and idle)
//!
//! A server arms timers through a [`TimerHandle`] cloned from the [`ServerCtx`]
//! in `init` and stored on the state — the same shape as [`Watcher`] for
//! monitors. [`arm_after`](TimerHandle::arm_after) is a one-shot,
//! [`tick_every`](TimerHandle::tick_every) a periodic; both fire into
//! [`handle_timer`](GenServer::handle_timer) at *system priority* (above infos
//! and the inbox, by arm position), and [`cancel`](TimerHandle::cancel) carries
//! the substrate's race signal. Separately, [`ServerCtx::idle_after`] sets a
//! receive-timeout window: quiet for the whole window fires
//! [`handle_idle`](GenServer::handle_idle).
//!
//! Two unrelated things share the word *timeout*: the server-side **idle /
//! receive timeout** above, and the client-side **call deadline**
//! ([`ServerRef::call_timeout`]) — how long a caller waits for a reply. They sit
//! on different axes and never interact (RFC 015 §7).
//!
//! ## Not here (yet)
//!
//! No dynamic info subscription: the info-channel set is fixed at start.
//! Revisit against a real consumer. (Monitor `Down` forwarding is dynamic —
//! see `handle_down` — because monitors are inherently created at runtime.)
//! The idle window is likewise set once, in `init` (RFC 015 §4.4).

use crate::channel::{channel, select, select_timeout, Receiver, RecvTimeoutError, Selectable, Sender};
use crate::monitor::{demonitor, monitor, Down, Monitor};
use crate::pid::Pid;
use crate::registry::{register_with, resolve_named_sender, RegisterError};
use crate::scheduler::{cancel_timer, request_stop, send_after_to, spawn, spawn_under};
use crate::timer::TimerId;
use std::cell::Cell;
use std::collections::HashMap;
use std::marker::PhantomData;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

/// Behaviour for a gen_server: a state value plus call/cast handlers.
///
/// `handle_call` and `handle_cast` are required; [`init`](Self::init) and
/// [`terminate`](Self::terminate) are optional lifecycle hooks with no-op
/// defaults.
pub trait GenServer: Send + 'static {
    /// Synchronous request type (carried by [`ServerRef::call`]).
    type Call: Send + 'static;
    /// Reply type returned for a `Call`.
    type Reply: Send + 'static;
    /// Asynchronous request type (carried by [`ServerRef::cast`]).
    type Cast: Send + 'static;
    /// Out-of-band message type, delivered to [`handle_info`](Self::handle_info)
    /// from the info channels registered at start
    /// ([`ServerBuilder::with_info`]). Servers with several out-of-band
    /// sources enum them up into one `Info`. Use `()` if unused.
    type Info: Send + 'static;
    /// The server's own scheduled-timer payload, delivered to
    /// [`handle_timer`](Self::handle_timer) when a timer armed via the loop's
    /// [`TimerHandle`] fires (RFC 015). Kept distinct from
    /// [`Info`](Self::Info) — `Info` is *external* out-of-band traffic, `Timer`
    /// is the server's *own* fires — so the system/userspace split stays
    /// legible and a tagged timer round-trips
    /// (`arm_after(d, Tk::Retry(n))` → `handle_timer(Tk::Retry(n))`). Use `()`
    /// if unused, exactly as `Info` does.
    type Timer: Send + 'static;

    /// Runs once inside the server actor before any message is handled. The
    /// [`ServerCtx`] is the loop's one runtime hook: clone its [`Watcher`]
    /// into the state here to be able to [`watch`](Watcher::watch) monitors
    /// from any later handler.
    fn init(&mut self, _ctx: &ServerCtx<Self>)
    where
        Self: Sized,
    {
    }

    /// Handle a synchronous call and produce the reply sent back to the caller.
    fn handle_call(&mut self, request: Self::Call) -> Self::Reply;

    /// Handle a fire-and-forget cast.
    fn handle_cast(&mut self, request: Self::Cast);

    /// Handle an out-of-band message from one of the info channels. Default:
    /// drop it.
    fn handle_info(&mut self, _info: Self::Info) {}

    /// Handle a [`Down`] from a monitor handed to the loop via
    /// [`Watcher::watch`]. Default: drop it.
    fn handle_down(&mut self, _down: Down) {}

    /// Handle a fired timer armed through the loop's [`TimerHandle`]
    /// ([`arm_after`](TimerHandle::arm_after) /
    /// [`tick_every`](TimerHandle::tick_every)). Timer fires re-enter at system
    /// priority — above infos and the inbox — so a heartbeat cannot be starved
    /// by userspace traffic. Default: drop it (RFC 015 §6).
    fn handle_timer(&mut self, _msg: Self::Timer) {}

    /// Handle a receive/idle timeout: fired when the loop has waited a full
    /// idle window (set once via [`ServerCtx::idle_after`]) with no message of
    /// any kind dispatched. The window resets on every dispatched message and
    /// re-arms after this fires (a steady idle detector); a server wanting
    /// one-shot idle-shutdown simply requests its own stop here. Default: no-op
    /// (RFC 015 §4.4, §6).
    fn handle_idle(&mut self) {}

    /// Runs as the server actor exits, on any exit path (see module docs).
    fn terminate(&mut self) {}
}

/// What travels the server's single inbox channel: a synchronous call (with a
/// reply sender) or an asynchronous cast.
enum Envelope<G: GenServer> {
    Call(G::Call, Sender<G::Reply>),
    Cast(G::Cast),
}

/// A clonable handle to a running server. Cloning yields another sender to the
/// same inbox; the server lives until the last `ServerRef` is dropped, at which
/// point its inbox closes and the loop exits normally.
pub struct ServerRef<G: GenServer> {
    tx: Sender<Envelope<G>>,
    pid: Pid,
}

impl<G: GenServer> Clone for ServerRef<G> {
    fn clone(&self) -> Self {
        ServerRef { tx: self.tx.clone(), pid: self.pid }
    }
}

/// Returned by [`ServerRef::call`] when the server is no longer reachable.
#[derive(Debug, PartialEq, Eq, Clone, Copy)]
pub enum CallError {
    /// The server was already gone, or died before replying.
    ServerDown,
}

/// Returned by [`ServerRef::call_timeout`].
#[derive(Debug, PartialEq, Eq, Clone, Copy)]
pub enum CallTimeoutError {
    /// The server was already gone, or died before replying.
    ServerDown,
    /// The deadline passed before a reply arrived. The request stays in the
    /// server's inbox: it will still be *handled*, but the reply is discarded
    /// (the abandoned reply channel's receiver is dropped, so the server's
    /// reply send fails harmlessly). Erlang behaves the same way; design
    /// idempotent calls accordingly.
    Timeout,
}

/// Returned by [`ServerRef::cast`] when the server is no longer reachable.
#[derive(Debug, PartialEq, Eq, Clone, Copy)]
pub enum CastError {
    /// The server inbox was closed (the server is gone).
    ServerDown,
}

impl<G: GenServer> ServerRef<G> {
    /// The server actor's pid — usable with `monitor`, `request_stop`, `link`.
    pub fn pid(&self) -> Pid {
        self.pid
    }

    /// Synchronous request-reply. Blocks (parking the calling actor) until the
    /// server replies, or returns [`CallError::ServerDown`] if the server is or
    /// becomes unreachable before a reply arrives.
    pub fn call(&self, request: G::Call) -> Result<G::Reply, CallError> {
        let (reply_tx, reply_rx) = channel::<G::Reply>();
        self.tx
            .send(Envelope::Call(request, reply_tx))
            .map_err(|_| CallError::ServerDown)?;
        reply_rx.recv().map_err(|_| CallError::ServerDown)
    }

    /// Bounded synchronous request-reply: like [`call`](Self::call), but
    /// gives up after `timeout`, returning [`CallTimeoutError::Timeout`].
    ///
    /// This `timeout` is a **client-side call deadline** — how long *this caller*
    /// waits for a reply — and is wholly separate from the server-side idle /
    /// receive timeout ([`ServerCtx::idle_after`] →
    /// [`GenServer::handle_idle`]), which measures quiet on the *server's* inbox.
    /// Same word ("timeout"), two different axes (RFC 015 §7); neither touches
    /// the other.
    ///
    /// The roadmap sketched this as monitor + wait-reply-or-Down + demonitor;
    /// that machinery is unnecessary here because server death is already
    /// observable on the reply channel itself — the reply sender is dropped
    /// as the server's stack unwinds, closing the channel and waking the
    /// parked caller (see the module docs). So a bounded call is exactly
    /// [`Receiver::recv_timeout`] on the reply channel: `Ok` on a reply,
    /// `Disconnected` → [`CallTimeoutError::ServerDown`], `Timeout` →
    /// [`CallTimeoutError::Timeout`]. Nothing is registered, so nothing can
    /// leak on the timeout path by construction.
    pub fn call_timeout(
        &self,
        request: G::Call,
        timeout: std::time::Duration,
    ) -> Result<G::Reply, CallTimeoutError> {
        let (reply_tx, reply_rx) = channel::<G::Reply>();
        self.tx
            .send(Envelope::Call(request, reply_tx))
            .map_err(|_| CallTimeoutError::ServerDown)?;
        reply_rx.recv_timeout(timeout).map_err(|e| match e {
            crate::channel::RecvTimeoutError::Disconnected => CallTimeoutError::ServerDown,
            crate::channel::RecvTimeoutError::Timeout => CallTimeoutError::Timeout,
        })
    }

    /// Fire-and-forget request. Returns once the message is enqueued; does not
    /// wait for the server to handle it. [`CastError::ServerDown`] if the inbox
    /// is already closed.
    pub fn cast(&self, request: G::Cast) -> Result<(), CastError> {
        self.tx
            .send(Envelope::Cast(request))
            .map_err(|_| CastError::ServerDown)
    }

    /// Ask the server to terminate and block until it has — the `sys`-style
    /// stop (cf. Erlang's `gen_server:stop/1`). Sends a cooperative stop to the
    /// server actor and waits for its `Down`, so [`GenServer::terminate`] has
    /// run by the time this returns (it fires from the loop's drop guard on the
    /// stop unwind). Returns immediately if the server was already gone.
    ///
    /// This is the explicit teardown for a server pinned alive by a registered
    /// [`ServerName`] (whose stored sender means dropping every external
    /// [`ServerRef`] no longer closes the inbox). Best-effort like all
    /// cooperative cancellation: a server wedged in a tight loop with no
    /// observation point cannot be stopped. Panics if called outside
    /// `Runtime::run()`.
    pub fn shutdown(&self) {
        let mon = monitor(self.pid);
        request_stop(self.pid);
        // The Down lands when the server finalizes; an already-dead target makes
        // `monitor` deliver NoProc immediately, so this never blocks forever.
        let _ = mon.rx.recv();
        demonitor(&mon);
    }
}

/// The server loop's internal *system intake* channel (RFC 015 §4.5). Control
/// (monitor handoff) and armed-timer fires are structurally the same — both are
/// loop-internal sources selected above the inbox — so they fold into one
/// channel, dispatched by variant. The arm stays open while any [`Watcher`] or
/// [`TimerHandle`] clone (or an in-flight fire thunk) still holds a sender.
enum Sys<G: GenServer> {
    /// A monitor handed in via [`Watcher::watch`]; pushed onto the loop's
    /// monitor set.
    Watch(Monitor),
    /// A fired one-shot timer carrying its loop-local id (so the loop can retire
    /// it from the live set) and the server's payload, dispatched to
    /// [`GenServer::handle_timer`].
    Timer(crate::timer::TimerId, G::Timer),
    /// A fired periodic tick carrying its stable local id. The loop resolves the
    /// payload from the registry's factory, dispatches it to
    /// [`GenServer::handle_timer`], and re-arms the next tick (RFC 015 §4.3).
    Tick(crate::timer::TimerId),
}

/// The server loop's runtime hook, passed to [`GenServer::init`]. Hands out the
/// loop's two clonable intake handles — the [`Watcher`] (monitors) and the
/// [`TimerHandle`] (timers) — plus the one-shot idle-window setter. Opaque, so
/// fields can grow without breaking.
pub struct ServerCtx<G: GenServer> {
    sys_tx: Sender<Sys<G>>,
    reg: Arc<Mutex<TimerReg<G>>>,
    /// The idle/receive-timeout window, set once via [`idle_after`](Self::idle_after)
    /// during `init` and read by the loop after `init` returns. Interior-mutable
    /// because `init` holds only `&ctx`; not `Send`, but `ServerCtx` is only ever
    /// borrowed on the actor's own stack during `init`, never sent.
    idle: Cell<Option<Duration>>,
}

impl<G: GenServer> ServerCtx<G> {
    /// A clonable handle to the loop's monitor intake. Store it in the state
    /// during `init` to watch monitors from later handlers.
    pub fn watcher(&self) -> Watcher<G> {
        Watcher { tx: self.sys_tx.clone() }
    }

    /// Shorthand for `ctx.watcher().watch(m)` when watching during `init`.
    pub fn watch(&self, m: Monitor) {
        self.watcher().watch(m)
    }

    /// A clonable handle to the loop's timer intake (the [`Watcher`] pattern,
    /// for time). Store it on the state during `init` to
    /// [`arm_after`](TimerHandle::arm_after) /
    /// [`tick_every`](TimerHandle::tick_every) /
    /// [`cancel`](TimerHandle::cancel) from any later handler.
    pub fn timer(&self) -> TimerHandle<G> {
        TimerHandle { sys_tx: self.sys_tx.clone(), reg: self.reg.clone() }
    }

    /// Set the idle / receive-timeout window: if no message of any kind is
    /// dispatched for `after`, the loop fires [`GenServer::handle_idle`]
    /// (RFC 015 §4.4). Set once, in `init`; the window is loop-owned and fixed
    /// (dynamic per-message reconfiguration is a non-goal). The window resets on
    /// every dispatched message and re-arms after `handle_idle` (a steady idle
    /// detector); a server wanting one-shot idle-shutdown requests its own stop
    /// in `handle_idle`.
    ///
    /// Distinct from [`ServerRef::call_timeout`], which is a *client-side* call
    /// deadline — same word, different axis (§7).
    pub fn idle_after(&self, after: Duration) {
        self.idle.set(Some(after));
    }
}

/// Per-server timer bookkeeping, shared between the loop and every
/// [`TimerHandle`] clone. A gen_server actor is single-threaded — handle calls
/// (made from inside handlers) and the loop never run concurrently — so this
/// `Mutex` is always uncontended; it is `Arc<Mutex>` rather than `Rc<RefCell>`
/// only because the handle is stored on `self` and `GenServer: Send` forces the
/// handle (hence its shared state) to be `Send`.
struct TimerReg<G: GenServer> {
    /// Monotonic minter for loop-local [`TimerId`]s — the ids handed to users,
    /// kept distinct from the substrate `seq` (which changes on every periodic
    /// re-arm). A local id is resolved only through this registry, never passed
    /// to [`cancel_timer`], so the two id roles never mix.
    next_local: u64,
    /// Live one-shot timers: local id → current substrate id. Present from
    /// [`arm_after`](TimerHandle::arm_after) until the loop retires it on fire
    /// or [`cancel`](TimerHandle::cancel) removes it.
    oneshots: HashMap<TimerId, crate::timer::TimerId>,
    /// Live periodic timers: stable local id → its bookkeeping. The stable id
    /// is minted once by [`tick_every`](TimerHandle::tick_every) and survives
    /// every re-arm; the entry lives until [`cancel`](TimerHandle::cancel)
    /// removes it (steady re-arm never self-removes).
    periodics: HashMap<TimerId, Periodic<G>>,
    /// A `Sys` sender held *only while ≥1 periodic is live*, used by the loop to
    /// re-arm the next tick. `Some` exactly when `periodics` is non-empty: this
    /// keeps the system arm open for a server whose only system use is a
    /// periodic, while leaving a non-timer server's arm to auto-close (the
    /// `unused_ctx` behaviour) since this stays `None` for it.
    rearm_tx: Option<Sender<Sys<G>>>,
}

/// One periodic timer's loop-side bookkeeping.
struct Periodic<G: GenServer> {
    /// Re-arm interval; each fire schedules the next at `now + every` (§4.3).
    every: Duration,
    /// The currently-armed substrate timer for this periodic (changes on every
    /// re-arm); cancelled by [`cancel`](TimerHandle::cancel).
    live: crate::timer::TimerId,
    /// Produces a fresh payload for each tick. Built in `tick_every` as
    /// `move || msg.clone()` — so the `Clone` bound lives there, on the one
    /// method that needs it, and never leaks onto `type Timer` or the loop
    /// (which is monomorphised per server and only *calls* this box).
    make: Box<dyn FnMut() -> G::Timer + Send>,
}

// Manual Default: deriving would demand `G: Default`, but a fresh registry is
// independent of the server type.
impl<G: GenServer> Default for TimerReg<G> {
    fn default() -> Self {
        TimerReg {
            next_local: 0,
            oneshots: HashMap::new(),
            periodics: HashMap::new(),
            rearm_tx: None,
        }
    }
}

impl<G: GenServer> TimerReg<G> {
    /// Mint the next loop-local id.
    fn mint(&mut self) -> TimerId {
        let id = TimerId::from_raw(self.next_local);
        self.next_local = self.next_local.wrapping_add(1);
        id
    }
}

/// Arms, cancels, and (chunk 4) periodically ticks server timers, the time-side
/// twin of [`Watcher`] (RFC 015 §4.3). Handed out by [`ServerCtx::timer`] in
/// `init` and stored on the state; later handlers arm timers without any change
/// to their signatures. Clonable; the arm stays open while any clone (or an
/// in-flight fire) lives.
pub struct TimerHandle<G: GenServer> {
    sys_tx: Sender<Sys<G>>,
    reg: Arc<Mutex<TimerReg<G>>>,
}

// Manual Clone for the same reason as `Watcher`: no `G: Clone` needed.
impl<G: GenServer> Clone for TimerHandle<G> {
    fn clone(&self) -> Self {
        TimerHandle { sys_tx: self.sys_tx.clone(), reg: self.reg.clone() }
    }
}

impl<G: GenServer> TimerHandle<G> {
    /// Arm a one-shot timer: deliver `msg` to [`GenServer::handle_timer`] after
    /// `after`, unless [`cancel`](Self::cancel)led first. Returns a loop-local
    /// [`TimerId`] for cancellation. A thin wrapper over the `send_after`
    /// substrate that lands the fire on the loop's own system arm (above infos
    /// and the inbox), not in the inbox.
    pub fn arm_after(&self, after: Duration, msg: G::Timer) -> TimerId {
        let mut reg = self.reg.lock().unwrap();
        let local = reg.mint();
        // The fire thunk carries the local id so the loop can retire the entry
        // on dispatch; `msg` is moved in (no `Clone` needed for one-shots).
        let sub = send_after_to(after, self.sys_tx.clone(), Sys::Timer(local, msg));
        reg.oneshots.insert(local, sub);
        local
    }

    /// Arm a periodic timer: deliver a fresh `msg` to
    /// [`GenServer::handle_timer`] every `every`, until
    /// [`cancel`](Self::cancel)led. Loop-managed sugar over the one-shot
    /// substrate (RFC 015 §4.3): the substrate stays one-shot; the loop re-arms
    /// at `now + every` after each fire and exposes one stable [`TimerId`], so a
    /// single `cancel` stops the re-arm *and* the pending instance.
    ///
    /// Requires `Self::Timer: Clone` — a periodic re-delivers the same logical
    /// message each tick, so the loop needs a fresh copy per period. The bound
    /// sits on this method alone; one-shot [`arm_after`](Self::arm_after) and
    /// the `type Timer` declaration stay unconstrained.
    pub fn tick_every(&self, every: Duration, msg: G::Timer) -> TimerId
    where
        G::Timer: Clone,
    {
        let mut reg = self.reg.lock().unwrap();
        let local = reg.mint();
        // A live periodic must keep the system arm open so the next tick can be
        // re-armed; the first periodic installs the loop's re-arm sender.
        if reg.periodics.is_empty() {
            reg.rearm_tx = Some(self.sys_tx.clone());
        }
        let make: Box<dyn FnMut() -> G::Timer + Send> = Box::new(move || msg.clone());
        // First instance fires after `every`; the payload is produced loop-side
        // from `make` on fire, so the tick carries only the stable id.
        let sub = send_after_to(every, self.sys_tx.clone(), Sys::Tick(local));
        reg.periodics.insert(local, Periodic { every, live: sub, make });
        local
    }

    /// Cancel an armed timer (one-shot or periodic). Returns the substrate's
    /// race signal — `true` if the cancel beat the (pending) fire, `false` if
    /// it had already fired / been cancelled / is unknown. For a periodic this
    /// also stops the re-arm: the entry is dropped, so the loop will not
    /// schedule another tick even if the pending one already escaped onto the
    /// channel (that in-flight tick is discarded on dispatch).
    pub fn cancel(&self, id: TimerId) -> bool {
        let mut reg = self.reg.lock().unwrap();
        if let Some(sub) = reg.oneshots.remove(&id) {
            return cancel_timer(sub);
        }
        if let Some(p) = reg.periodics.remove(&id) {
            let beat = cancel_timer(p.live);
            if reg.periodics.is_empty() {
                reg.rearm_tx = None;
            }
            return beat;
        }
        false
    }
}

/// Hands [`Monitor`]s to a server loop; their [`Down`]s are dispatched to
/// [`GenServer::handle_down`]. Down arms outrank every other arm (a death
/// notice cannot be starved), and a delivered `Down` retires its arm —
/// monitors are one-shot.
pub struct Watcher<G: GenServer> {
    tx: Sender<Sys<G>>,
}

// Manual Clone: deriving would demand `G: Clone`, but the handle is clonable
// regardless of the server type (it clones only the inner sender).
impl<G: GenServer> Clone for Watcher<G> {
    fn clone(&self) -> Self {
        Watcher { tx: self.tx.clone() }
    }
}

impl<G: GenServer> Watcher<G> {
    /// Transfer `m` to the server loop. If the server is already gone the
    /// monitor is silently dropped (its queued `Down`, if any, with it) —
    /// there is no loop left to care.
    pub fn watch(&self, m: Monitor) {
        let _ = self.tx.send(Sys::Watch(m));
    }
}

/// Configure-then-start construction for servers that need more than a bare
/// [`start`]: info channels, a supervisor. Consumed by [`start`](Self::start).
///
/// ```ignore
/// let server = ServerBuilder::new(state)
///     .with_info(events_rx)
///     .under(supervisor_pid)
///     .start();
/// ```
pub struct ServerBuilder<G: GenServer> {
    state: G,
    infos: Vec<Receiver<G::Info>>,
    supervisor: Option<Pid>,
}

impl<G: GenServer> ServerBuilder<G> {
    pub fn new(state: G) -> Self {
        ServerBuilder { state, infos: Vec::new(), supervisor: None }
    }

    /// Add an out-of-band channel; messages arriving on it are dispatched to
    /// [`GenServer::handle_info`]. Repeatable; arm priority is call order
    /// (earlier = higher), and every info channel outranks the inbox.
    pub fn with_info(mut self, rx: Receiver<G::Info>) -> Self {
        self.infos.push(rx);
        self
    }

    /// Spawn the server under an explicit supervisor pid (via [`spawn_under`])
    /// so it slots into the supervision tree.
    pub fn under(mut self, supervisor: Pid) -> Self {
        self.supervisor = Some(supervisor);
        self
    }

    /// Spawn the server actor and hand back its [`ServerRef`]. The server's
    /// lifetime is governed by its refs, not by joining, so the backing join
    /// handle is dropped.
    pub fn start(self) -> ServerRef<G> {
        self.spawn_server()
    }

    /// Bind the server to a durable [`ServerName`] as it starts. Switches to the
    /// fallible [`NamedServerBuilder::start`] (the name may already be held by a
    /// live server). Consumes the builder, carrying its `with_info` / `under`
    /// configuration through.
    pub fn named(self, name: ServerName<G>) -> NamedServerBuilder<G> {
        NamedServerBuilder { builder: self, name: name.as_str() }
    }

    /// The shared spawn body behind [`start`](Self::start) and
    /// [`NamedServerBuilder::start`]: make the inbox, spawn the loop, return the
    /// ref. (The named path additionally publishes the inbox sender under the
    /// name before returning.)
    fn spawn_server(self) -> ServerRef<G> {
        let (tx, rx) = channel::<Envelope<G>>();
        let ServerBuilder { state, infos, supervisor } = self;
        let handle = match supervisor {
            Some(sup) => spawn_under(sup, move || server_loop::<G>(rx, state, infos)),
            None => spawn(move || server_loop::<G>(rx, state, infos)),
        };
        ServerRef { tx, pid: handle.pid() }
    }
}

/// A durable name typed by the *server* (RFC 014): a gen_server is multi-message
/// (call / cast over one inbox), so it is addressed by `G` rather than by a
/// single message type. By-name [`call`] / [`cast`] check the request against
/// `G`'s `Call` / `Cast` / `Reply`. Declared as a constant and shared freely:
///
/// ```ignore
/// const COUNTER: ServerName<Counter> = ServerName::new("counter");
/// ```
///
/// Under the hood the server's inbox is published into the registry as a
/// `Sender<Envelope<G>>` keyed by its message `TypeId` — the same channel store
/// every other name uses — so naming needs no separate directory. `Envelope`
/// stays private: only `ServerName<G>` opens that door.
pub struct ServerName<G> {
    name: &'static str,
    _marker: PhantomData<fn() -> G>,
}

impl<G> ServerName<G> {
    /// Bind a static string as a server name. `const`, so names live as
    /// associated constants at call sites.
    #[inline]
    pub const fn new(name: &'static str) -> Self {
        Self { name, _marker: PhantomData }
    }

    /// The underlying registry key.
    #[inline]
    pub const fn as_str(self) -> &'static str {
        self.name
    }
}

impl<G> Copy for ServerName<G> {}
impl<G> Clone for ServerName<G> {
    fn clone(&self) -> Self {
        *self
    }
}

/// A [`ServerBuilder`] that will bind a [`ServerName`] as it starts. Reached via
/// [`ServerBuilder::named`]; its [`start`](Self::start) is fallible because the
/// name may already be held by a live server. `with_info` / `under` stay
/// available so configuration can come before or after `named`.
pub struct NamedServerBuilder<G: GenServer> {
    builder: ServerBuilder<G>,
    name: &'static str,
}

impl<G: GenServer> NamedServerBuilder<G> {
    /// Add an out-of-band info channel (see [`ServerBuilder::with_info`]).
    pub fn with_info(mut self, rx: Receiver<G::Info>) -> Self {
        self.builder = self.builder.with_info(rx);
        self
    }

    /// Spawn under an explicit supervisor (see [`ServerBuilder::under`]).
    pub fn under(mut self, supervisor: Pid) -> Self {
        self.builder = self.builder.under(supervisor);
        self
    }

    /// Spawn the server and bind its name in one step. Fallible: returns
    /// [`RegisterError::NameTaken`] if the name is already held by a different
    /// live server.
    ///
    /// The inbox sender is published under the name **from the parent side,
    /// before this returns**, so a by-name `call` / `cast` resolves the instant
    /// `start()` returns — no race with the server body. On a name clash the
    /// just-spawned server is wound down (its only ref is dropped, closing the
    /// inbox), so a failed bind leaks no actor.
    pub fn start(self) -> Result<ServerRef<G>, RegisterError> {
        let NamedServerBuilder { builder, name } = self;
        let server = builder.spawn_server();
        match register_with::<Envelope<G>>(server.pid, name, server.tx.clone()) {
            Ok(()) => Ok(server),
            Err(e) => {
                drop(server); // inbox closes → loop exits gracefully
                Err(e)
            }
        }
    }
}

/// Resolve a [`ServerName`] to a [`ServerRef`] when you want a handle to hold or
/// pass on rather than resolve per call. Rebuilds the ref from the registry's
/// stored inbox sender; `None` if no live server holds the name.
///
/// Panics if called outside `Runtime::run()`.
pub fn whereis_server<G: GenServer>(name: ServerName<G>) -> Option<ServerRef<G>> {
    resolve_named_sender::<Envelope<G>>(name.as_str()).map(|(pid, tx)| ServerRef { tx, pid })
}

/// Synchronous request-reply to the server currently registered under `name`,
/// resolving through the registry on every call (so a server restarted under
/// the same name is reached transparently). [`CallError::ServerDown`] if no live
/// server holds the name, or if it dies before replying.
///
/// Panics if called outside `Runtime::run()`.
pub fn call<G: GenServer>(name: ServerName<G>, request: G::Call) -> Result<G::Reply, CallError> {
    match whereis_server(name) {
        Some(server) => server.call(request),
        None => Err(CallError::ServerDown),
    }
}

/// Fire-and-forget to the server registered under `name`, resolving per cast.
/// [`CastError::ServerDown`] if no live server holds the name.
///
/// Panics if called outside `Runtime::run()`.
pub fn cast<G: GenServer>(name: ServerName<G>, request: G::Cast) -> Result<(), CastError> {
    match whereis_server(name) {
        Some(server) => server.cast(request),
        None => Err(CastError::ServerDown),
    }
}

/// Terminate the server registered under `name` and block until it is down (see
/// [`ServerRef::shutdown`]). A no-op if no live server holds the name.
///
/// Panics if called outside `Runtime::run()`.
pub fn shutdown<G: GenServer>(name: ServerName<G>) {
    if let Some(server) = whereis_server(name) {
        server.shutdown();
    }
}

/// Spawn `state` as a server under the current actor (via [`spawn`]). Returns a
/// [`ServerRef`]. Shorthand for `ServerBuilder::new(state).start()`.
pub fn start<G: GenServer>(state: G) -> ServerRef<G> {
    ServerBuilder::new(state).start()
}

/// Like [`start`], but spawns the server under an explicit supervisor pid (via
/// [`spawn_under`]) so it slots into the supervision tree.
pub fn start_under<G: GenServer>(supervisor: Pid, state: G) -> ServerRef<G> {
    ServerBuilder::new(state).under(supervisor).start()
}

fn server_loop<G: GenServer>(
    rx: Receiver<Envelope<G>>,
    state: G,
    mut infos: Vec<Receiver<G::Info>>,
) {
    // terminate() must run on every exit path (clean close, panic, stop), so it
    // lives in this guard's Drop rather than after the loop. The guard also owns
    // a registry handle so the *same* drop drains every live timer (RFC 015
    // §4.7): once the loop is gone no periodic can re-arm anyway, but cancelling
    // here keeps no armed substrate entry lingering past the server, and the
    // assert pins the invariant.
    struct Terminate<G: GenServer>(G, Arc<Mutex<TimerReg<G>>>);
    impl<G: GenServer> Drop for Terminate<G> {
        fn drop(&mut self) {
            {
                let mut reg = self.1.lock().unwrap();
                for (_, sub) in reg.oneshots.drain() {
                    cancel_timer(sub);
                }
                for (_, p) in reg.periodics.drain() {
                    cancel_timer(p.live);
                }
                reg.rearm_tx = None;
                // A fired-but-gone send already no-ops on the closed Sys channel;
                // this guards the armed-but-unfired re-arming case specifically.
                debug_assert!(
                    reg.oneshots.is_empty() && reg.periodics.is_empty(),
                    "an armed timer survived gen_server loop exit"
                );
            }
            self.0.terminate();
        }
    }

    fn dispatch<G: GenServer>(state: &mut G, env: Envelope<G>) {
        match env {
            Envelope::Call(request, reply_tx) => {
                let reply = state.handle_call(request);
                // The caller may have gone away (e.g. cancelled while parked);
                // a failed reply send is not the server's problem.
                let _ = reply_tx.send(reply);
            }
            Envelope::Cast(request) => state.handle_cast(request),
        }
    }

    // Shared timer bookkeeping: the loop keeps one clone (to retire one-shots on
    // fire, re-arm periodics, and read the idle window); the guard holds another
    // to drain on exit; every TimerHandle the state stores holds more.
    let reg = Arc::new(Mutex::new(TimerReg::default()));
    let mut guard = Terminate(state, reg.clone());

    // The system intake arm: Watchers feed Monitors and armed timers feed fires
    // to the loop through it. The ctx (and with it the loop's own sender) drops
    // right after init — a state that cloned no Watcher/TimerHandle closes the
    // arm, the first select observes the closure, and the loop falls back to the
    // plain-inbox park.
    let (sys_tx, sys_rx) = channel::<Sys<G>>();
    // Bind the ctx so the idle window set during init can be read back, then
    // drop it — that drops the loop's own Sys sender, so a state that cloned no
    // Watcher/TimerHandle lets the arm auto-close (the unused-ctx behaviour).
    let ctx = ServerCtx { sys_tx, reg: reg.clone(), idle: Cell::new(None) };
    guard.0.init(&ctx);
    let idle = ctx.idle.get();
    drop(ctx);

    let mut monitors: Vec<Monitor> = Vec::new();
    let mut sys_open = true;
    // The receive-timeout deadline, live only when an idle window is set. Reset
    // to `now + idle` after any dispatched message and after handle_idle fires
    // (steady detector); `None` disables the timeout entirely.
    let mut idle_deadline: Option<Instant> = idle.map(|d| Instant::now() + d);
    let reset_idle = |dl: &mut Option<Instant>| {
        if let Some(d) = idle {
            *dl = Some(Instant::now() + d);
        }
    };

    loop {
        if monitors.is_empty() && !sys_open && infos.is_empty() {
            // Nothing to select over: park on the inbox alone (the pre-v0.8
            // loop), with the idle window as the recv timeout when one is set.
            match idle_deadline {
                Some(dl) => {
                    let wait = dl.saturating_duration_since(Instant::now());
                    match rx.recv_timeout(wait) {
                        Ok(env) => {
                            dispatch(&mut guard.0, env);
                            reset_idle(&mut idle_deadline);
                        }
                        // Quiet for the whole window → the idle event.
                        Err(RecvTimeoutError::Timeout) => {
                            guard.0.handle_idle();
                            reset_idle(&mut idle_deadline);
                        }
                        // All ServerRefs dropped → inbox closed → shutdown.
                        Err(RecvTimeoutError::Disconnected) => break,
                    }
                }
                None => match rx.recv() {
                    Ok(env) => dispatch(&mut guard.0, env),
                    Err(_) => break,
                },
            }
        } else {
            // Arm priority: downs, then the system arm, then infos (each in
            // declaration order), then the inbox. The arm slice is rebuilt
            // per iteration because every set but the inbox shrinks or grows.
            // The whole wait carries the idle window as its timeout.
            let nd = monitors.len();
            let nw = sys_open as usize;
            let sel = {
                let mut arms: Vec<&dyn Selectable> =
                    Vec::with_capacity(nd + nw + infos.len() + 1);
                for m in &monitors {
                    arms.push(&m.rx);
                }
                if sys_open {
                    arms.push(&sys_rx);
                }
                for r in &infos {
                    arms.push(r);
                }
                arms.push(&rx);
                match idle_deadline {
                    Some(dl) => {
                        select_timeout(&arms, dl.saturating_duration_since(Instant::now()))
                    }
                    None => Some(select(&arms)),
                }
            };
            let i = match sel {
                Some(i) => i,
                // No arm ready for the whole window → idle, then re-arm steady.
                None => {
                    guard.0.handle_idle();
                    reset_idle(&mut idle_deadline);
                    crate::check!();
                    continue;
                }
            };
            if i < nd {
                // A Down retires its arm either way: delivered (one-shot) or
                // closed without delivering (defensive; shouldn't happen).
                let m = monitors.remove(i);
                if let Ok(Some(down)) = m.rx.try_recv() {
                    guard.0.handle_down(down);
                    reset_idle(&mut idle_deadline);
                }
            } else if i < nd + nw {
                match sys_rx.try_recv() {
                    // Control intake, not a dispatched message: no idle reset.
                    Ok(Some(Sys::Watch(m))) => monitors.push(m),
                    Ok(Some(Sys::Timer(id, msg))) => {
                        // The one-shot fired: retire its registry entry so the
                        // live set tracks only still-pending timers, then
                        // dispatch.
                        reg.lock().unwrap().oneshots.remove(&id);
                        guard.0.handle_timer(msg);
                        reset_idle(&mut idle_deadline);
                    }
                    Ok(Some(Sys::Tick(id))) => {
                        // A periodic fired. Under the lock: if it is still live
                        // (cancel didn't beat us here), produce its payload and
                        // re-arm the next tick at now + every *before* dispatch,
                        // so the period is measured from fire-handling and a
                        // handler that cancels stops the just-armed instance.
                        let msg = {
                            let mut g = reg.lock().unwrap();
                            let r = &mut *g;
                            if let Some(p) = r.periodics.get_mut(&id) {
                                let every = p.every;
                                let msg = (p.make)();
                                let tx = r
                                    .rearm_tx
                                    .as_ref()
                                    .expect("a live periodic keeps rearm_tx Some")
                                    .clone();
                                p.live = send_after_to(every, tx, Sys::Tick(id));
                                Some(msg)
                            } else {
                                // Cancelled between fire and dispatch: discard.
                                None
                            }
                        };
                        if let Some(msg) = msg {
                            guard.0.handle_timer(msg);
                            reset_idle(&mut idle_deadline);
                        }
                    }
                    // Single-receiver: nothing can drain the arm between
                    // select's ready and our try_recv.
                    Ok(None) => debug_assert!(false, "ready system arm was empty"),
                    // Every Watcher/TimerHandle gone: stop selecting on the arm.
                    Err(_) => sys_open = false,
                }
            } else if i < nd + nw + infos.len() {
                let j = i - nd - nw;
                match infos[j].try_recv() {
                    Ok(Some(info)) => {
                        guard.0.handle_info(info);
                        reset_idle(&mut idle_deadline);
                    }
                    Ok(None) => debug_assert!(false, "ready info arm was empty"),
                    // Senders all gone: drop the arm, keep serving. `remove`
                    // (not swap_remove) — order is priority.
                    Err(_) => {
                        infos.remove(j);
                    }
                }
            } else {
                match rx.try_recv() {
                    Ok(Some(env)) => {
                        dispatch(&mut guard.0, env);
                        reset_idle(&mut idle_deadline);
                    }
                    Ok(None) => debug_assert!(false, "ready inbox was empty"),
                    Err(_) => break,
                }
            }
        }
        // Observation point so a server whose arms are never empty stays
        // preemptible and cancellable.
        crate::check!();
    }
}