mosaik 0.4.2

//! # Handlers
//!
//! A [`Handler`] serves a registered function to remote callers. Remote
//! [`Caller`](super::caller)s discover handlers automatically through the
//! [`discovery`](crate::discovery) subsystem and invoke the function over
//! QUIC — one connection per call.
//!
//! Register a function either with a plain closure via
//! [`Builder::serve`], or with a [`Function`](super::Function)
//! implementation via [`Builder::serve_fn`] when the function carries
//! state or startup logic.
//!
//! A [`Handler`] exclusively owns its registration: dropping the handle
//! deregisters the function, removes it from the local discovery
//! advertisement, and aborts in-flight calls.

use {
	super::FunctionId,
	crate::primitives::Datum,
	core::{
		marker::PhantomData,
		sync::atomic::{AtomicUsize, Ordering},
	},
	std::sync::Arc,
	tokio_util::sync::CancellationToken,
};

mod builder;
pub(in crate::functions) mod registry;

pub use builder::{Builder, BuilderError, HandlerConfig};
pub(in crate::functions) use registry::Registry;

/// A registered function handler serving calls from remote peers.
///
/// Created through [`Functions::handler`](super::Functions::handler) and
/// the builder's [`serve`](Builder::serve) / [`serve_fn`](Builder::serve_fn)
/// terminal methods.
///
/// The handler exclusively owns its registration — dropping it
/// deregisters the function, un-advertises it from discovery, and aborts
/// any in-flight calls.
pub struct Handler<Req: Datum, Res: Datum, E: Datum = ()> {
	function_id: FunctionId,
	stats: Arc<Stats>,
	registry: Arc<Registry>,
	cancel: CancellationToken,
	_marker: PhantomData<fn(&Req, &Res, &E)>,
}

impl<Req: Datum, Res: Datum, E: Datum> Handler<Req, Res, E> {
	pub(in crate::functions) fn new(
		function_id: FunctionId,
		stats: Arc<Stats>,
		registry: Arc<Registry>,
		cancel: CancellationToken,
	) -> Self {
		Self {
			function_id,
			stats,
			registry,
			cancel,
			_marker: PhantomData,
		}
	}

	/// The effective wire id this handler is registered and advertised
	/// under (base id with all ticket validator signatures folded in).
	pub const fn function_id(&self) -> &FunctionId {
		&self.function_id
	}

	/// Invocation statistics for this function.
	pub fn stats(&self) -> &Stats {
		&self.stats
	}
}

impl<Req: Datum, Res: Datum, E: Datum> Drop for Handler<Req, Res, E> {
	fn drop(&mut self) {
		self.cancel.cancel();
		self.registry.remove(self.function_id);
	}
}

impl<Req: Datum, Res: Datum, E: Datum> core::fmt::Debug
	for Handler<Req, Res, E>
{
	fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
		f.debug_struct("Handler")
			.field("function_id", &self.function_id)
			.finish_non_exhaustive()
	}
}

/// Invocation statistics for a registered function handler.
#[derive(Debug, Default)]
pub struct Stats {
	served: AtomicUsize,
	failed: AtomicUsize,
	inflight: AtomicUsize,
}

impl Stats {
	/// Number of calls that completed with a reply — including
	/// application-level errors, which are successful protocol exchanges.
	pub fn served(&self) -> usize {
		self.served.load(Ordering::Relaxed)
	}

	/// Number of calls that failed with a handler-side fault (undecodable
	/// request, reply encoding failure, or handler shutdown mid-call).
	pub fn failed(&self) -> usize {
		self.failed.load(Ordering::Relaxed)
	}

	/// Number of calls currently executing.
	pub fn inflight(&self) -> usize {
		self.inflight.load(Ordering::Relaxed)
	}

	pub(in crate::functions) fn record_started(&self) {
		self.inflight.fetch_add(1, Ordering::Relaxed);
	}

	pub(in crate::functions) fn record_served(&self) {
		self.inflight.fetch_sub(1, Ordering::Relaxed);
		self.served.fetch_add(1, Ordering::Relaxed);
	}

	pub(in crate::functions) fn record_failed(&self) {
		self.inflight.fetch_sub(1, Ordering::Relaxed);
		self.failed.fetch_add(1, Ordering::Relaxed);
	}
}