tako-rs-core 2.0.0

//! In-process signal arbiter and dispatch system.
//!
//! This module defines a small abstraction for named signals that can be emitted
//! and handled within a Tako application. It is intended for cross-cutting
//! concerns such as metrics, logging hooks, or custom application events.

use std::any::Any;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
use std::time::Duration;

use arc_swap::ArcSwap;
use futures_util::future::BoxFuture;
use futures_util::future::join_all;
use once_cell::sync::Lazy;
use scc::HashMap as SccHashMap;
use tokio::sync::broadcast;
use tokio::sync::mpsc;
#[cfg(not(feature = "compio"))]
use tokio::time::timeout;

use crate::types::BuildHasher;

const DEFAULT_BROADCAST_CAPACITY: usize = 64;
static GLOBAL_BROADCAST_CAPACITY: AtomicUsize = AtomicUsize::new(DEFAULT_BROADCAST_CAPACITY);
static EXPORTER_KEY_COUNTER: AtomicU64 = AtomicU64::new(0);

type HandlerList = Arc<ArcSwap<Vec<SignalHandler>>>;

/// Well-known signal identifiers for common lifecycle and request events.
///
/// **Naming conventions (v2):**
///
/// | prefix         | scope                                                                     |
/// |----------------|---------------------------------------------------------------------------|
/// | `server.*`     | process-level events (server start / stop)                                |
/// | `connection.*` | per-connection events (open / close, transport snapshot)                  |
/// | `request.*`    | per-request events on the **global** application arbiter                  |
/// | `route.*`      | per-route events on the **route-local** arbiter (one arbiter per route)   |
/// | `queue.*`      | background-job lifecycle (queue.job.queued / started / completed / …)     |
/// | `rpc.*`        | typed-RPC errors raised through the arbiter                               |
/// | `router.*`     | router-level events (hot reloads, future config swaps)                    |
///
/// `route.request.*` is intentionally a separate id (not an alias of
/// `request.*`) because the two are emitted on different arbiters: the route
/// arbiter sees only its own route's traffic, while the global arbiter sees
/// every request. Subscribers that want both should listen on the global
/// arbiter and join through the matched-path label.
///
/// Cluster-scope signals (cross-pod fan-out via Redis pub/sub or NATS) are
/// out of scope for this module — see the [`bus`] sub-module for the
/// `SignalBus` trait that companion crates implement.
pub mod ids {
  pub const SERVER_STARTED: &str = "server.started";
  pub const SERVER_STOPPED: &str = "server.stopped";
  pub const CONNECTION_OPENED: &str = "connection.opened";
  pub const CONNECTION_CLOSED: &str = "connection.closed";
  pub const REQUEST_STARTED: &str = "request.started";
  pub const REQUEST_COMPLETED: &str = "request.completed";
  pub const ROUTER_HOT_RELOAD: &str = "router.hot_reload";
  pub const RPC_ERROR: &str = "rpc.error";
  pub const ROUTE_REQUEST_STARTED: &str = "route.request.started";
  pub const ROUTE_REQUEST_COMPLETED: &str = "route.request.completed";
}

/// Cluster-scope signal bridge.
///
/// A `SignalBus` lifts the in-process `SignalArbiter` to a multi-node fan-out
/// (Redis pub/sub, NATS, Kafka, …). Companion crates provide concrete impls;
/// this trait is the contract.
pub mod bus {
  use async_trait::async_trait;

  use super::Signal;

  /// Inbound + outbound bridge between an in-process arbiter and a remote
  /// pub/sub topic. Implementations should be cheap to clone (`Arc`-based).
  #[async_trait]
  pub trait SignalBus: Send + Sync + 'static {
    /// Publish a signal to the remote topic.
    async fn publish(&self, signal: &Signal);
  }

  /// No-op bus — every published signal is dropped. Useful as a default and
  /// in tests.
  #[derive(Clone, Default)]
  pub struct LocalBus;

  #[async_trait]
  impl SignalBus for LocalBus {
    async fn publish(&self, _signal: &Signal) {}
  }
}

/// A signal emitted through the arbiter.
///
/// Signals are identified by an arbitrary string and can carry a map of
/// metadata. Callers are free to define their own conventions for ids and
/// fields.
#[derive(Clone, Debug, Default)]
pub struct Signal {
  /// Identifier of the signal, for example "request.started" or "metrics.tick".
  pub id: String,
  /// Optional metadata payload carried with the signal.
  pub metadata: HashMap<String, String, BuildHasher>,
}

impl Signal {
  /// Creates a new signal with the given id and empty metadata.
  #[inline]
  #[must_use]
  pub fn new(id: impl Into<String>) -> Self {
    Self {
      id: id.into(),
      metadata: HashMap::with_hasher(BuildHasher::default()),
    }
  }

  /// Creates a signal with pre-allocated capacity for the given number of metadata entries.
  #[inline]
  #[must_use]
  pub fn with_capacity(id: impl Into<String>, capacity: usize) -> Self {
    Self {
      id: id.into(),
      metadata: HashMap::with_capacity_and_hasher(capacity, BuildHasher::default()),
    }
  }

  /// Adds a metadata entry, returning self for chaining.
  #[inline]
  pub fn meta(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
    self.metadata.insert(key.into(), value.into());
    self
  }

  /// Creates a new signal with initial metadata.
  #[inline]
  #[must_use]
  pub fn with_metadata(
    id: impl Into<String>,
    metadata: HashMap<String, String, BuildHasher>,
  ) -> Self {
    Self {
      id: id.into(),
      metadata,
    }
  }

  /// Creates a signal from a typed payload implementing `SignalPayload`.
  #[inline]
  #[must_use]
  pub fn from_payload<P: SignalPayload>(payload: &P) -> Self {
    Self {
      id: payload.id().to_string(),
      metadata: payload.to_metadata(),
    }
  }
}

/// Trait for types that can be converted into a `Signal`.
pub trait SignalPayload {
  /// The canonical id for this kind of signal, e.g. "request.completed".
  fn id(&self) -> &'static str;

  /// Serializes the payload into the metadata map.
  fn to_metadata(&self) -> HashMap<String, String, BuildHasher>;
}

/// Boxed async signal handler.
pub type SignalHandler = Arc<dyn Fn(Signal) -> BoxFuture<'static, ()> + Send + Sync>;

/// Boxed typed RPC handler used by the signal arbiter.
pub type RpcHandler = Arc<
  dyn Fn(Arc<dyn Any + Send + Sync>) -> BoxFuture<'static, Arc<dyn Any + Send + Sync>>
    + Send
    + Sync,
>;

/// Exporter callback invoked for every emitted signal.
pub type SignalExporter = Arc<dyn Fn(&Signal) + Send + Sync>;

/// Simple stream type returned by filtered subscriptions.
///
/// Bounded with [`FILTERED_SUBSCRIPTION_BUFFER`] capacity: a slow consumer
/// no longer accumulates an unbounded backlog (OOM risk). When the queue is
/// full, the producer-side forwarder drops the signal rather than blocking.
pub type SignalStream = mpsc::Receiver<Signal>;

/// Bounded buffer size for [`SignalArbiter::subscribe_filtered`]. Picked to
/// absorb short bursts while keeping per-subscriber memory bounded (~1024
/// `Signal`s ≈ 64 KiB plus per-signal metadata). Slow consumers experience
/// overflow as silent drops, not OOM.
pub const FILTERED_SUBSCRIPTION_BUFFER: usize = 1024;

/// Upper bound for [`SignalArbiter::set_global_broadcast_capacity`]. Caps each
/// new topic channel at 1 Mi elements so a misconfigured caller cannot ask
/// for a `usize::MAX`-element channel that would OOM the process on the next
/// topic creation.
pub const MAX_BROADCAST_CAPACITY: usize = 1 << 20;

#[derive(Default)]
struct Inner {
  handlers: SccHashMap<String, HandlerList>,
  topics: SccHashMap<String, broadcast::Sender<Signal>>,
  rpc: SccHashMap<String, RpcHandler>,
  exporters: SccHashMap<u64, SignalExporter>,
}

fn new_handler_list() -> HandlerList {
  Arc::new(ArcSwap::new(Arc::new(Vec::new())))
}

/// Shared arbiter used to register and dispatch named signals.
#[derive(Clone, Default)]
pub struct SignalArbiter {
  inner: Arc<Inner>,
}

/// Global application-level signal arbiter.
static APP_SIGNAL_ARBITER: Lazy<SignalArbiter> = Lazy::new(SignalArbiter::new);

/// Returns a reference to the global application-level signal arbiter.
pub fn app_signals() -> &'static SignalArbiter {
  &APP_SIGNAL_ARBITER
}

/// Returns the global application-level signal arbiter.
pub fn app_events() -> &'static SignalArbiter {
  app_signals()
}

/// Error type for typed RPC calls.
///
/// This error type implements `std::error::Error` for integration with
/// error handling libraries.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RpcError {
  /// No handler registered for the requested RPC method.
  NoHandler,
  /// The response type did not match the expected type.
  TypeMismatch,
}

impl std::fmt::Display for RpcError {
  fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
    match self {
      Self::NoHandler => write!(f, "no handler registered for RPC method"),
      Self::TypeMismatch => write!(f, "RPC response type mismatch"),
    }
  }
}

impl std::error::Error for RpcError {}

/// Result type for RPC calls with explicit error reporting.
pub type RpcResult<T> = Result<T, RpcError>;

/// Error type for RPC calls with timeout support.
///
/// This error type implements `std::error::Error` for integration with
/// error handling libraries.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RpcTimeoutError {
  /// The RPC call timed out before completing.
  Timeout,
  /// An RPC error occurred.
  Rpc(RpcError),
}

impl std::fmt::Display for RpcTimeoutError {
  fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
    match self {
      Self::Timeout => write!(f, "RPC call timed out"),
      Self::Rpc(err) => write!(f, "{err}"),
    }
  }
}

impl std::error::Error for RpcTimeoutError {
  fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
    match self {
      Self::Rpc(err) => Some(err),
      Self::Timeout => None,
    }
  }
}

impl From<RpcError> for RpcTimeoutError {
  #[inline]
  fn from(err: RpcError) -> Self {
    Self::Rpc(err)
  }
}

impl SignalArbiter {
  /// Creates a new, empty signal arbiter.
  pub fn new() -> Self {
    Self::default()
  }

  /// Sets the global broadcast capacity used for topic channels.
  ///
  /// This affects all newly created topics across all arbiters. The capacity
  /// is clamped to `[1, MAX_BROADCAST_CAPACITY]` (1 MiB-element ceiling) so a
  /// caller can never request a `usize::MAX`-element channel that would OOM
  /// on the next `topic_sender` allocation.
  pub fn set_global_broadcast_capacity(capacity: usize) {
    let cap = capacity.clamp(1, MAX_BROADCAST_CAPACITY);
    GLOBAL_BROADCAST_CAPACITY.store(cap, Ordering::SeqCst);
  }

  /// Returns the current global broadcast capacity.
  pub fn global_broadcast_capacity() -> usize {
    GLOBAL_BROADCAST_CAPACITY.load(Ordering::SeqCst)
  }

  /// Returns (and lazily initializes) the broadcast sender for a signal id.
  pub(crate) fn topic_sender(&self, id: &str) -> broadcast::Sender<Signal> {
    if let Some(existing) = self.inner.topics.get_sync(id) {
      existing.clone()
    } else {
      let cap = GLOBAL_BROADCAST_CAPACITY.load(Ordering::SeqCst);
      let (tx, _rx) = broadcast::channel(cap);
      let entry = self.inner.topics.entry_sync(id.to_string()).or_insert(tx);
      entry.clone()
    }
  }

  /// Registers a handler for the given signal id.
  ///
  /// Handlers are invoked in registration order whenever a matching signal is emitted.
  pub fn on<F, Fut>(&self, id: impl Into<String>, handler: F)
  where
    F: Fn(Signal) -> Fut + Send + Sync + 'static,
    Fut: std::future::Future<Output = ()> + Send + 'static,
  {
    let id = id.into();
    let handler: SignalHandler = Arc::new(move |signal: Signal| {
      let fut = handler(signal);
      Box::pin(fut)
    });

    let list = self.handler_list_for(id);
    list.rcu(|current| {
      let mut next = Vec::with_capacity(current.len() + 1);
      next.extend(current.iter().cloned());
      next.push(handler.clone());
      Arc::new(next)
    });
  }

  /// Returns (creating if necessary) the per-id `ArcSwap` holding the handler list.
  /// The SCC hashmap protects only the slot lookup; handler-list updates are
  /// wait-free via `ArcSwap` RCU, so concurrent `on()` / `emit()` calls cannot
  /// race on a `Vec::push` reallocation.
  fn handler_list_for(&self, id: String) -> HandlerList {
    let entry = self
      .inner
      .handlers
      .entry_sync(id)
      .or_insert_with(new_handler_list);
    entry.clone()
  }

  /// Subscribes to a broadcast channel for the given signal id.
  ///
  /// This is useful for long-lived listeners such as metrics collectors,
  /// background workers, plugins, or middleware driven tasks.
  ///
  /// ⚠️ **Topic retention:** the topic map is keyed by `id` and entries
  /// are created lazily on the first `subscribe`/`emit`. The `Sender` is
  /// retained for the lifetime of the arbiter — there is no TTL or LRU
  /// eviction. With **high-cardinality dynamic ids** (e.g. one signal id
  /// per session or per request) this is an unbounded slow-leak.
  ///
  /// Use a low-cardinality id set ("`request.started`", "`order.placed`")
  /// and put the per-request discriminator inside the [`Signal`] payload
  /// instead of the id string.
  pub fn subscribe(&self, id: impl AsRef<str>) -> broadcast::Receiver<Signal> {
    let id_str = id.as_ref();
    let sender = self.topic_sender(id_str);
    sender.subscribe()
  }

  /// Subscribes to all signals whose id starts with the given prefix.
  ///
  /// For example, `subscribe_prefix("request.")` will receive
  /// `request.started`, `request.completed`, etc.
  pub fn subscribe_prefix(&self, prefix: impl AsRef<str>) -> broadcast::Receiver<Signal> {
    let mut key = prefix.as_ref().to_string();
    if !key.ends_with('*') {
      key.push('*');
    }
    let sender = self.topic_sender(&key);
    sender.subscribe()
  }

  /// Subscribes to all signals regardless of their id.
  ///
  /// This is a special variant that receives every emitted signal.
  /// Internally uses a wildcard prefix matching (empty prefix = all signals).
  pub fn subscribe_all(&self) -> broadcast::Receiver<Signal> {
    self.subscribe_prefix("")
  }

  /// Broadcasts a signal to all subscribers without awaiting handler completion.
  pub(crate) fn broadcast(&self, signal: Signal) {
    // Exact id subscribers
    if let Some(sender) = self.inner.topics.get_sync(&signal.id) {
      let _ = sender.send(signal.clone());
    }

    // Prefix subscribers: keys ending with '*'.
    // Snapshot matching senders before sending so the SCC entry locks are
    // released by the time we deliver — even though `broadcast::Sender::send`
    // is non-blocking, this also bounds inconsistency to the moment of the
    // snapshot rather than spreading it across every per-entry send.
    let mut targets: Vec<broadcast::Sender<Signal>> = Vec::new();
    self.inner.topics.iter_sync(|key, v| {
      if let Some(prefix) = key.strip_suffix('*')
        && signal.id.starts_with(prefix)
      {
        targets.push(v.clone());
      }
      true
    });
    for sender in targets {
      let _ = sender.send(signal.clone());
    }
  }

  /// Subscribes using a filter function on top of an id-based subscription.
  ///
  /// Spawns a background task that forwards only matching signals into a
  /// bounded mpsc of capacity [`FILTERED_SUBSCRIPTION_BUFFER`]. When the
  /// consumer cannot keep up, new signals are dropped via `try_send` rather
  /// than queued unboundedly (which was the previous behavior and an OOM
  /// vector for long-running observers).
  pub fn subscribe_filtered<F>(&self, id: impl AsRef<str>, filter: F) -> SignalStream
  where
    F: Fn(&Signal) -> bool + Send + Sync + 'static,
  {
    let mut rx = self.subscribe(id);
    let (tx, out_rx) = mpsc::channel(FILTERED_SUBSCRIPTION_BUFFER);
    let filter = Arc::new(filter);

    #[cfg(not(feature = "compio"))]
    tokio::spawn(async move {
      while let Ok(signal) = rx.recv().await {
        if filter(&signal) {
          match tx.try_send(signal) {
            Ok(()) => {}
            Err(mpsc::error::TrySendError::Full(_)) => {
              // Slow consumer: drop the signal silently to preserve back-pressure.
            }
            Err(mpsc::error::TrySendError::Closed(_)) => break,
          }
        }
      }
    });

    #[cfg(feature = "compio")]
    compio::runtime::spawn(async move {
      while let Ok(signal) = rx.recv().await {
        if filter(&signal) {
          match tx.try_send(signal) {
            Ok(()) => {}
            Err(mpsc::error::TrySendError::Full(_)) => {}
            Err(mpsc::error::TrySendError::Closed(_)) => break,
          }
        }
      }
    })
    .detach();

    out_rx
  }

  /// Waits for the next occurrence of a signal id (oneshot-style).
  ///
  /// This uses the broadcast channel under the hood but resolves on the
  /// first successfully received signal.
  pub async fn once(&self, id: impl AsRef<str>) -> Option<Signal> {
    let mut rx = self.subscribe(id);
    loop {
      match rx.recv().await {
        Ok(sig) => return Some(sig),
        Err(broadcast::error::RecvError::Lagged(_)) => {}
        Err(_) => return None,
      }
    }
  }

  /// Registers a typed RPC handler under the given id.
  ///
  /// This allows request/response style interactions over the same arbiter,
  /// using type-erased storage internally for flexibility.
  ///
  /// # Panics
  ///
  /// The returned handler panics *at call time* (not at registration) if a
  /// caller invokes [`SignalArbiter::call_rpc`] under this id with a request
  /// type that does not match `Req`. Type erasure happens during dispatch, so
  /// the mismatch surfaces as a fail-fast panic instead of returning `None`.
  /// Keep the `Req` type stable across registrations.
  pub fn register_rpc<Req, Res, F, Fut>(&self, id: impl Into<String>, f: F)
  where
    Req: Send + Sync + 'static,
    Res: Send + Sync + 'static,
    F: Fn(Arc<Req>) -> Fut + Send + Sync + 'static,
    Fut: std::future::Future<Output = Res> + Send + 'static,
  {
    let id_str = id.into();
    let id_for_panic = id_str.clone();
    let func = Arc::new(f);

    let handler: RpcHandler = Arc::new(move |raw: Arc<dyn Any + Send + Sync>| {
      let func = func.clone();
      let id_for_panic = id_for_panic.clone();
      Box::pin(async move {
        let req = raw
          .downcast::<Req>()
          .unwrap_or_else(|_| panic!("Signal RPC type mismatch for id: {id_for_panic}"));
        let res = func(req).await;
        Arc::new(res) as Arc<dyn Any + Send + Sync>
      })
    });

    // `upsert_sync`: re-registering the same id replaces the prior handler.
    // `insert_sync` would keep the old one and silently drop the new closure
    // — a re-`register_rpc` after hot-reload or test reset would be a no-op.
    self.inner.rpc.upsert_sync(id_str, handler);
  }

  /// Calls a typed RPC handler and returns a shared pointer to the response.
  pub async fn call_rpc_arc<Req, Res>(&self, id: impl AsRef<str>, req: Req) -> Option<Arc<Res>>
  where
    Req: Send + Sync + 'static,
    Res: Send + Sync + 'static,
  {
    let id_str = id.as_ref();
    let entry = self.inner.rpc.get_async(id_str).await?;
    let handler = entry.clone();
    drop(entry);

    let raw_req: Arc<dyn Any + Send + Sync> = Arc::new(req);
    let raw_res = handler(raw_req).await;

    raw_res.downcast::<Res>().ok()
  }

  /// Calls a typed RPC handler and returns an owned response with an error type.
  pub async fn call_rpc_result<Req, Res>(&self, id: impl AsRef<str>, req: Req) -> RpcResult<Res>
  where
    Req: Send + Sync + 'static,
    Res: Send + Sync + Clone + 'static,
  {
    let id_str = id.as_ref();
    let Some(entry) = self.inner.rpc.get_async(id_str).await else {
      return Err(RpcError::NoHandler);
    };
    let handler = entry.clone();
    drop(entry);

    let raw_req: Arc<dyn Any + Send + Sync> = Arc::new(req);
    let raw_res = handler(raw_req).await;

    match raw_res.downcast::<Res>() {
      Ok(res) => Ok((*res).clone()),
      Err(_) => Err(RpcError::TypeMismatch),
    }
  }

  /// Calls a typed RPC handler and returns an owned response.
  pub async fn call_rpc<Req, Res>(&self, id: impl AsRef<str>, req: Req) -> Option<Res>
  where
    Req: Send + Sync + 'static,
    Res: Send + Sync + Clone + 'static,
  {
    self.call_rpc_result::<Req, Res>(id, req).await.ok()
  }

  /// Calls a typed RPC handler with a timeout.
  #[cfg(not(feature = "compio"))]
  pub async fn call_rpc_timeout<Req, Res>(
    &self,
    id: impl AsRef<str>,
    req: Req,
    dur: Duration,
  ) -> Result<Res, RpcTimeoutError>
  where
    Req: Send + Sync + 'static,
    Res: Send + Sync + Clone + 'static,
  {
    match timeout(dur, self.call_rpc_result::<Req, Res>(id, req)).await {
      Ok(Ok(res)) => Ok(res),
      Ok(Err(e)) => Err(RpcTimeoutError::Rpc(e)),
      Err(_) => Err(RpcTimeoutError::Timeout),
    }
  }

  /// Calls a typed RPC handler with a timeout (compio variant).
  #[cfg(feature = "compio")]
  pub async fn call_rpc_timeout<Req, Res>(
    &self,
    id: impl AsRef<str>,
    req: Req,
    dur: Duration,
  ) -> Result<Res, RpcTimeoutError>
  where
    Req: Send + Sync + 'static,
    Res: Send + Sync + Clone + 'static,
  {
    let sleep = std::pin::pin!(compio::time::sleep(dur));
    let work = std::pin::pin!(self.call_rpc_result::<Req, Res>(id, req));
    match futures_util::future::select(work, sleep).await {
      futures_util::future::Either::Left((Ok(res), _)) => Ok(res),
      futures_util::future::Either::Left((Err(e), _)) => Err(RpcTimeoutError::Rpc(e)),
      futures_util::future::Either::Right(((), _)) => Err(RpcTimeoutError::Timeout),
    }
  }

  /// Emits a signal and awaits all registered handlers.
  ///
  /// Handlers run concurrently and this method resolves once all handlers have completed.
  pub async fn emit(&self, signal: Signal) {
    // First, broadcast to any subscribers.
    self.broadcast(signal.clone());

    // Call exporters asynchronously.
    self
      .inner
      .exporters
      .iter_async(|_, v| {
        v(&signal);
        true
      })
      .await;

    if let Some(entry) = self.inner.handlers.get_async(&signal.id).await {
      let list = entry.clone();
      drop(entry);
      let handlers = list.load_full();

      let futures = handlers.iter().map(|handler| {
        let s = signal.clone();
        handler(s)
      });

      let _ = join_all(futures).await;
    }
  }

  /// Emits a signal using the global application-level arbiter.
  pub async fn emit_app(signal: Signal) {
    app_signals().emit(signal).await;
  }

  /// Registers a global exporter that is invoked for every emitted signal.
  ///
  /// Exporters are merged when routers are merged, similar to handlers.
  pub fn register_exporter<F>(&self, exporter: F)
  where
    F: Fn(&Signal) + Send + Sync + 'static,
  {
    let key = EXPORTER_KEY_COUNTER.fetch_add(1, Ordering::Relaxed);
    let exporter: SignalExporter = Arc::new(exporter);
    // `upsert_sync` mirrors `register_rpc`: makes re-registration on the
    // same (very-rare) duplicate key a replace instead of a silent drop.
    self.inner.exporters.upsert_sync(key, exporter);
  }

  /// Merges all handlers from `other` into `self`.
  ///
  /// This is used by router merging so that signal handlers attached to
  /// a merged router continue to be active.
  pub(crate) fn merge_from(&self, other: &SignalArbiter) {
    other.inner.handlers.iter_sync(|k, other_list| {
      let other_handlers = other_list.load_full();
      if other_handlers.is_empty() {
        return true;
      }
      let target_list = self.handler_list_for(k.clone());
      target_list.rcu(|current| {
        let mut next = Vec::with_capacity(current.len() + other_handlers.len());
        next.extend(current.iter().cloned());
        next.extend(other_handlers.iter().cloned());
        Arc::new(next)
      });

      true
    });

    other.inner.topics.iter_sync(|k, v| {
      self.inner.topics.entry_sync(k.clone()).or_insert(v.clone());
      true
    });

    // On merge, the merged-in arbiter wins for conflicting ids/keys.
    // `insert_sync` would silently keep `self`'s entry, dropping the
    // intentional value the caller wanted to pull in.
    other.inner.rpc.iter_sync(|k, v| {
      self.inner.rpc.upsert_sync(k.clone(), v.clone());
      true
    });

    other.inner.exporters.iter_sync(|k, v| {
      self.inner.exporters.upsert_sync(*k, v.clone());
      true
    });
  }

  /// Returns a list of known signal ids (exact topics) currently registered.
  pub fn signal_ids(&self) -> Vec<String> {
    let mut ids = Vec::new();
    self.inner.topics.iter_sync(|k, _| {
      if !k.ends_with('*') {
        ids.push(k.clone());
      }
      true
    });
    ids
  }

  /// Returns a list of known signal prefixes (topics ending with '*').
  pub fn signal_prefixes(&self) -> Vec<String> {
    let mut prefixes = Vec::new();
    self.inner.topics.iter_sync(|k, _| {
      if k.ends_with('*') {
        prefixes.push(k.clone());
      }
      true
    });
    prefixes
  }

  /// Returns a list of registered RPC ids.
  pub fn rpc_ids(&self) -> Vec<String> {
    let mut ids = Vec::new();
    self.inner.rpc.iter_sync(|k, _| {
      ids.push(k.clone());
      true
    });
    ids
  }
}

/// Connection-lifecycle signal helpers used by every transport.
///
/// `Router::dispatch` already emits the per-request `REQUEST_STARTED` /
/// `REQUEST_COMPLETED` signals automatically; these helpers cover the
/// connection-level events (`SERVER_STARTED`, `CONNECTION_OPENED`,
/// `CONNECTION_CLOSED`) that have no natural per-request hook. They keep the
/// emit boilerplate out of every transport file.
pub mod transport {
  use super::Signal;
  use super::SignalArbiter;
  use super::ids;

  /// Emits the `server.started` signal with `addr` / `transport` / `tls` meta.
  pub async fn emit_server_started(addr: &str, transport: &str, tls: bool) {
    SignalArbiter::emit_app(
      Signal::with_capacity(ids::SERVER_STARTED, 3)
        .meta("addr", addr)
        .meta("transport", transport)
        .meta("tls", if tls { "true" } else { "false" }),
    )
    .await;
  }

  /// Emits the `connection.opened` signal with `remote_addr` / `tls` / optional `protocol`.
  pub async fn emit_connection_opened(remote_addr: &str, tls: bool, protocol: Option<&str>) {
    let mut sig = Signal::with_capacity(ids::CONNECTION_OPENED, 3)
      .meta("remote_addr", remote_addr)
      .meta("tls", if tls { "true" } else { "false" });
    if let Some(p) = protocol {
      sig = sig.meta("protocol", p);
    }
    SignalArbiter::emit_app(sig).await;
  }

  /// Emits the `connection.closed` signal with `remote_addr` / `tls` / optional `protocol`.
  pub async fn emit_connection_closed(remote_addr: &str, tls: bool, protocol: Option<&str>) {
    let mut sig = Signal::with_capacity(ids::CONNECTION_CLOSED, 3)
      .meta("remote_addr", remote_addr)
      .meta("tls", if tls { "true" } else { "false" });
    if let Some(p) = protocol {
      sig = sig.meta("protocol", p);
    }
    SignalArbiter::emit_app(sig).await;
  }
}