mosaik 0.4.2

//! # Callers
//!
//! A [`Caller`] invokes a function on the network without knowing which
//! peer will serve the call. Eligible handler peers are discovered
//! automatically through the [`discovery`](crate::discovery) subsystem,
//! ranked by measured round-trip time, and called over a dedicated
//! connection per call — with automatic failover to the next candidate on
//! transport failures and handler-side rejections.
//!
//! If no eligible handler is known when a call is made, the call waits —
//! within its timeout — for one to appear in the catalog, so callers and
//! handlers can start in any order.
//!
//! Application-level errors returned by the function are typed end-to-end
//! and never trigger failover: the function ran, its answer stands.

use {
	super::{
		Functions,
		accept::{CallReply, CallRequest},
		status::{CandidateInfo, CandidatesMap, When},
	},
	crate::{
		discovery::Discovery,
		network::{LocalNode, UnknownPeer, error::Success, link::LinkError},
		primitives::Datum,
	},
	builder::CallerConfig,
	core::{marker::PhantomData, time::Duration},
	iroh::endpoint::ApplicationClose,
	std::sync::Arc,
	tokio::sync::watch,
	tokio_util::sync::DropGuard,
};

mod builder;
pub(super) mod worker;

pub use builder::{Builder, CallerConfig as Config};

/// A type-erased error box for encoding/decoding and transport failures.
type BoxedError = Box<dyn core::error::Error + Send + Sync + 'static>;

/// Errors that can occur when invoking a function through a [`Caller`].
#[derive(Debug, thiserror::Error)]
pub enum CallError<E> {
	/// The function ran and returned its typed application-level error.
	///
	/// Application errors never trigger failover to another handler — the
	/// function executed and its answer stands.
	#[error("function returned an application error")]
	Application(E),

	/// Eligible handlers were available but no call attempt completed
	/// within the timeout.
	#[error("call timed out")]
	Timeout,

	/// No eligible handler appeared in the catalog within the timeout.
	#[error("no eligible handler available")]
	Unavailable,

	/// Every attempted handler rejected the call; carries the last
	/// rejection's close reason (e.g. function not found, not allowed, no
	/// capacity).
	#[error("call rejected by handler: {}", .0.reason.escape_ascii())]
	Rejected(ApplicationClose),

	/// Every attempted handler failed with a transport error; carries the
	/// last failure.
	#[error("transport failure: {0}")]
	Transport(#[source] BoxedError),

	/// The request payload could not be encoded.
	#[error("request encoding failed: {0}")]
	Encode(#[source] BoxedError),

	/// The reply payload could not be decoded.
	#[error("response decoding failed: {0}")]
	Decode(#[source] BoxedError),
}

/// A caller handle for invoking a function on the network.
///
/// Created through [`Functions::caller`](super::Functions::caller) (or
/// [`caller_for`](super::Functions::caller_for)) and the builder's
/// [`build`](Builder::build) method. Dropping the caller stops watching
/// the catalog for eligible handlers.
///
/// Calls take `&self`, so a single handle can serve any number of
/// concurrent calls.
pub struct Caller<Req: Datum, Res: Datum, E: Datum = ()> {
	config: Arc<CallerConfig>,
	local: LocalNode,
	discovery: Discovery,
	candidates: watch::Receiver<CandidatesMap>,
	status: When,
	metrics_labels: [(&'static str, String); 2],
	_abort: DropGuard,
	_marker: PhantomData<fn(&Req, &Res, &E)>,
}

impl<Req: Datum, Res: Datum, E: Datum> Caller<Req, Res, E> {
	/// The effective wire id of the function this caller invokes (base id
	/// with all ticket validator signatures folded in).
	pub fn function_id(&self) -> &super::FunctionId {
		&self.config.function_id
	}

	/// Returns the reactive availability conditions for this caller, e.g.
	/// `caller.when().available().minimum_of(2)`.
	pub const fn when(&self) -> &When {
		&self.status
	}

	/// Invokes the function with the given request using the caller's
	/// default call timeout.
	///
	/// See [`call_with_timeout`](Self::call_with_timeout) for the exact
	/// semantics.
	pub async fn call(&self, req: Req) -> Result<Res, CallError<E>> {
		self.call_with_timeout(req, self.config.call_timeout).await
	}

	/// Invokes the function with the given request, bounding the entire
	/// call by the given timeout.
	///
	/// The call discovers eligible handlers from the catalog, ranks them by
	/// measured RTT, and invokes the best candidate over a dedicated
	/// connection. On transport failures and handler-side rejections it
	/// fails over to the next candidate; if no eligible handler is known it
	/// waits for one to appear. Application-level errors returned by the
	/// function never trigger failover.
	pub async fn call_with_timeout(
		&self,
		req: Req,
		timeout: Duration,
	) -> Result<Res, CallError<E>> {
		let payload = req.encode().map_err(|e| CallError::Encode(Box::new(e)))?;
		let request = CallRequest {
			network_id: *self.local.network_id(),
			function_id: self.config.function_id,
			payload,
		};

		let mut state = CallState::default();
		let attempts = self.run_attempts(&request, &mut state);
		match tokio::time::timeout(timeout, attempts).await {
			Ok(result) => result,
			Err(_elapsed) => Err(state.into_timeout_error()),
		}
	}

	/// Drives call attempts until one succeeds, looping over ranked
	/// candidate snapshots as the catalog evolves. Runs unbounded — the
	/// caller wraps it in the call timeout.
	async fn run_attempts(
		&self,
		request: &CallRequest,
		state: &mut CallState,
	) -> Result<Res, CallError<E>> {
		let mut candidates = self.candidates.clone();
		candidates.mark_changed();

		loop {
			if candidates.changed().await.is_err() {
				// the caller worker terminated (network shutdown)
				return Err(CallError::Unavailable);
			}

			let snapshot = candidates.borrow_and_update().clone();
			let mut ranked: Vec<CandidateInfo> = snapshot.values().cloned().collect();
			ranked.sort_by_key(|c| (c.rtt().is_none(), c.rtt()));

			if ranked.is_empty() {
				// wait-for-handler semantics: no eligible candidate yet; the next
				// catalog movement republishes the snapshot and wakes us up.
				continue;
			}
			state.had_candidates = true;

			for candidate in &ranked {
				// re-validate the handler's tickets against the freshest entry
				// (admission is per-call, so no expiry timers are needed)
				if candidate
					.entry()
					.validate_tickets(&self.config.handler_auth)
					.is_err()
				{
					continue;
				}

				match self.invoke_once(candidate, request).await {
					Ok(CallReply::Ok(bytes)) => {
						metrics::counter!(
							"mosaik.functions.calls.completed",
							&self.metrics_labels
						)
						.increment(1);
						return Res::decode(&bytes)
							.map_err(|e| CallError::Decode(Box::new(e)));
					}
					Ok(CallReply::Err(bytes)) => {
						metrics::counter!(
							"mosaik.functions.calls.completed",
							&self.metrics_labels
						)
						.increment(1);
						return match E::decode(&bytes) {
							Ok(app_err) => Err(CallError::Application(app_err)),
							Err(e) => Err(CallError::Decode(Box::new(e))),
						};
					}
					Err(Attempt::Rejected(reason)) => {
						state.last_rejection = Some(reason);
					}
					Err(Attempt::Transport(error)) => {
						state.last_transport = Some(error);
					}
				}
			}

			// every candidate in this snapshot failed; loop back and wait for
			// catalog movement (bounded by the outer call timeout)
		}
	}

	/// Performs a single call attempt against one candidate, with a
	/// one-shot catalog-sync recovery when the handler does not know us
	/// yet.
	async fn invoke_once(
		&self,
		candidate: &CandidateInfo,
		request: &CallRequest,
	) -> Result<CallReply, Attempt> {
		match self.attempt(candidate, request).await {
			Err(Attempt::Rejected(reason)) if reason == UnknownPeer => {
				// The handler does not have us in its catalog; force a catalog
				// exchange so it learns our entry, then retry this candidate once.
				let addr = candidate.entry().address().clone();
				let _ = self.discovery.sync_with(addr).await;
				self.attempt(candidate, request).await
			}
			other => other,
		}
	}

	/// One connect → request → reply → close exchange with a candidate.
	async fn attempt(
		&self,
		candidate: &CandidateInfo,
		request: &CallRequest,
	) -> Result<CallReply, Attempt> {
		metrics::counter!("mosaik.functions.calls.attempts", &self.metrics_labels)
			.increment(1);

		let mut link = self
			.local
			.connect::<Functions>(candidate.entry().address().clone())
			.await
			.map_err(|e| Attempt::classify(e.into()))?;

		link
			.send(request)
			.await
			.map_err(|e| Attempt::classify(e.into()))?;

		let reply: CallReply =
			link.recv().await.map_err(|e| Attempt::classify(e.into()))?;

		// the reply concludes the exchange; close errors are inconsequential
		let _ = link.close(Success).await;

		Ok(reply)
	}
}

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

/// The outcome of a single failed call attempt against one candidate.
enum Attempt {
	/// The handler explicitly rejected the call with an application-level
	/// close reason.
	Rejected(ApplicationClose),

	/// The attempt failed at the transport level without a handler-side
	/// verdict.
	Transport(LinkError),
}

impl Attempt {
	/// Classifies a link error: an application-level close frame from the
	/// remote is a rejection verdict, anything else is a transport failure.
	fn classify(error: LinkError) -> Self {
		if let Some(reason) = error.close_reason() {
			return Self::Rejected(reason.clone());
		}
		Self::Transport(error)
	}
}

/// Bookkeeping across the attempts of one call, used to produce the most
/// informative error when the call times out.
#[derive(Default)]
struct CallState {
	had_candidates: bool,
	last_rejection: Option<ApplicationClose>,
	last_transport: Option<LinkError>,
}

impl CallState {
	/// Maps the recorded attempt history to the timeout error taxonomy:
	/// never had a candidate → `Unavailable`; the last handler verdict if
	/// any → `Rejected`; the last transport failure if any → `Transport`;
	/// otherwise → `Timeout`.
	fn into_timeout_error<E>(self) -> CallError<E> {
		if !self.had_candidates {
			return CallError::Unavailable;
		}
		if let Some(reason) = self.last_rejection {
			return CallError::Rejected(reason);
		}
		if let Some(error) = self.last_transport {
			return CallError::Transport(Box::new(error));
		}
		CallError::Timeout
	}
}