vorma 0.86.0-pre.4

//! Phase execution. Task middleware completes before the endpoint or view
//! phase starts. Each phase runs every handler concurrently and commits
//! contributions in declaration order. A redirect or a failure eagerly stops
//! only later positions; earlier positions keep contributing. A handler that
//! panics fails at its own position with status 500.

use std::any::Any;
use std::collections::HashMap;

use futures_util::future::BoxFuture;
use http::StatusCode;
use serde_json::value::RawValue;
use tokio::task::{AbortHandle, JoinSet};
use tracing::Instrument;

use crate::outcome::{Contribution, HandlerKind, HandlerLabel, Outcome, PublicError, Redirection};

/// One handler's work in a phase.
pub(crate) struct PhaseInvocation {
	pub(crate) future: BoxFuture<'static, Outcome>,
	pub(crate) label: HandlerLabel,
}

/// A committed position.
pub(crate) enum Committed<PositionKey> {
	/// The handler succeeded; `output` is absent without output.
	Completed {
		position_key: PositionKey,
		output: Option<Box<RawValue>>,
	},
	/// The endpoint answered 304 without a body.
	NotModified { position_key: PositionKey },
	/// The position has no handler.
	NoHandler { position_key: PositionKey },
}

/// The first failed position of a phase.
pub(crate) struct FailedPosition {
	pub(crate) index: usize,
	pub(crate) status: StatusCode,
	pub(crate) public_error: Option<PublicError>,
}

pub(crate) enum PhaseResult<PositionKey> {
	Completed {
		contribution: Contribution,
		positions: Vec<Committed<PositionKey>>,
	},
	Redirect {
		contribution: Contribution,
		redirection: Redirection,
	},
	/// The failure's status, headers, and cookies are merged into the
	/// contribution. Later positions are stopped.
	Failed {
		contribution: Contribution,
		before: Vec<Committed<PositionKey>>,
		failed: FailedPosition,
	},
}

fn create_span(label: &HandlerLabel) -> tracing::Span {
	let pattern = &*label.pattern;
	match label.kind {
		HandlerKind::TaskMiddleware => {
			tracing::info_span!("vorma.task_middleware", vorma.pattern = pattern)
		}
		HandlerKind::ViewLoader => {
			tracing::info_span!("vorma.view_loader", vorma.pattern = pattern)
		}
		HandlerKind::ApiEndpoint => {
			tracing::info_span!("vorma.api_endpoint", vorma.pattern = pattern)
		}
	}
}

fn describe_panic(payload: &(dyn Any + Send)) -> &str {
	payload
		.downcast_ref::<&str>()
		.copied()
		.or_else(|| payload.downcast_ref::<String>().map(String::as_str))
		.unwrap_or("a non-text panic payload")
}

/// A panic is a defect: the position fails with status 500 and no payload.
fn panic_outcome(label: &HandlerLabel, payload: &(dyn Any + Send)) -> Outcome {
	tracing::error!(
		handler = label.kind.span_name(),
		pattern = &*label.pattern,
		path = &*label.request_path,
		panic = describe_panic(payload),
		"Vorma handler panicked",
	);
	Outcome::Failure {
		contribution: Contribution::default(),
		status: StatusCode::INTERNAL_SERVER_ERROR,
		public_error: None,
	}
}

/// Runs one phase. `initial` holds the contributions committed before it.
pub(crate) async fn run_phase<PositionKey>(
	positions: Vec<(PositionKey, Option<PhaseInvocation>)>,
	initial: Contribution,
) -> PhaseResult<PositionKey> {
	let count = positions.len();
	let mut position_keys: Vec<Option<PositionKey>> = Vec::with_capacity(count);
	let mut has_handler = Vec::with_capacity(count);
	let mut outcomes: Vec<Option<Outcome>> = (0..count).map(|_| None).collect();
	let mut labels: Vec<Option<HandlerLabel>> = Vec::with_capacity(count);
	let mut invocations = Vec::new();
	for (index, (position_key, invocation)) in positions.into_iter().enumerate() {
		position_keys.push(Some(position_key));
		has_handler.push(invocation.is_some());
		labels.push(
			invocation
				.as_ref()
				.map(|invocation| invocation.label.clone()),
		);
		if let Some(invocation) = invocation {
			invocations.push((index, invocation));
		}
	}

	let mut tasks = JoinSet::new();
	let mut task_indexes = HashMap::new();
	let mut abort_handles: Vec<Option<AbortHandle>> = (0..count).map(|_| None).collect();
	// Every handler runs in its own Tokio task, so that no handler sees the
	// host's task-local values, whatever the number of handlers in the phase.
	for (index, invocation) in invocations {
		let span = create_span(&invocation.label);
		let handle = tasks.spawn(invocation.future.instrument(span));
		task_indexes.insert(handle.id(), index);
		abort_handles[index] = Some(handle);
	}

	let mut contribution = initial;
	let mut committed = Vec::with_capacity(count);
	let mut next = 0;
	loop {
		while next < count {
			if !has_handler[next] {
				let position_key = position_keys[next]
					.take()
					.expect("each position_key commits once");
				committed.push(Committed::NoHandler { position_key });
				next += 1;
				continue;
			}
			let Some(outcome) = outcomes[next].take() else {
				break;
			};
			match outcome {
				Outcome::Continue {
					contribution: later,
					output,
				} => {
					contribution.merge(later);
					let position_key = position_keys[next]
						.take()
						.expect("each position_key commits once");
					committed.push(Committed::Completed {
						position_key,
						output,
					});
					next += 1;
				}
				Outcome::NotModified {
					contribution: later,
				} => {
					contribution.merge(later);
					let position_key = position_keys[next]
						.take()
						.expect("each position_key commits once");
					committed.push(Committed::NotModified { position_key });
					next += 1;
				}
				Outcome::Redirect {
					contribution: later,
					redirection,
				} => {
					contribution.merge(later);
					return PhaseResult::Redirect {
						contribution,
						redirection,
					};
				}
				Outcome::Failure {
					contribution: mut later,
					status,
					public_error,
				} => {
					later.status = Some(status);
					contribution.merge(later);
					return PhaseResult::Failed {
						contribution,
						before: committed,
						failed: FailedPosition {
							index: next,
							status,
							public_error,
						},
					};
				}
			}
		}
		if next == count {
			return PhaseResult::Completed {
				contribution,
				positions: committed,
			};
		}
		let Some(joined) = tasks.join_next_with_id().await else {
			unreachable!("an uncommitted position with a handler has a running task");
		};
		let (index, outcome) = match joined {
			Ok((id, outcome)) => (task_indexes[&id], outcome),
			// Only positions after a redirect or failure are aborted, and they
			// never commit.
			Err(error) if error.is_cancelled() => continue,
			Err(error) => {
				let index = task_indexes[&error.id()];
				let label = labels[index].as_ref().expect("a task has a label");
				let payload = error.into_panic();
				(index, panic_outcome(label, &*payload))
			}
		};
		if outcome.is_terminal() {
			// Aborting a handler drops its future, which ends its part in every
			// task run that it waits on.
			for handle in abort_handles[index + 1..]
				.iter_mut()
				.filter_map(Option::take)
			{
				handle.abort();
			}
		}
		outcomes[index] = Some(outcome);
	}
}