vorma 0.86.0-pre.5

//! 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,
	}
}

struct PhasePosition<PositionKey> {
	key: Option<PositionKey>,
	handler: Option<RunningHandler>,
}

struct RunningHandler {
	label: HandlerLabel,
	abort: Option<AbortHandle>,
	outcome: Option<Outcome>,
}

/// 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 tasks = JoinSet::new();
	let mut task_indexes = HashMap::new();
	// 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.
	let mut positions: Vec<_> = positions
		.into_iter()
		.enumerate()
		.map(|(index, (key, invocation))| {
			let handler = invocation.map(|invocation| {
				let span = create_span(&invocation.label);
				let abort = tasks.spawn(invocation.future.instrument(span));
				task_indexes.insert(abort.id(), index);
				RunningHandler {
					label: invocation.label,
					abort: Some(abort),
					outcome: None,
				}
			});
			PhasePosition {
				key: Some(key),
				handler,
			}
		})
		.collect();

	let mut contribution = initial;
	let mut committed = Vec::with_capacity(count);
	let mut next = 0;
	loop {
		while next < count {
			let position = &mut positions[next];
			let outcome = if let Some(handler) = &mut position.handler {
				let Some(outcome) = handler.outcome.take() else {
					break;
				};
				Some(outcome)
			} else {
				None
			};
			let position_key = position.key.take().expect("each position key commits once");
			match outcome {
				None => committed.push(Committed::NoHandler { position_key }),
				Some(Outcome::Continue {
					contribution: later,
					output,
				}) => {
					contribution.merge(later);
					committed.push(Committed::Completed {
						position_key,
						output,
					});
				}
				Some(Outcome::NotModified {
					contribution: later,
				}) => {
					contribution.merge(later);
					committed.push(Committed::NotModified { position_key });
				}
				Some(Outcome::Redirect {
					contribution: later,
					redirection,
				}) => {
					contribution.merge(later);
					return PhaseResult::Redirect {
						contribution,
						redirection,
					};
				}
				Some(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,
						},
					};
				}
			}
			next += 1;
		}
		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 = &positions[index]
					.handler
					.as_ref()
					.expect("a task has a handler")
					.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 positions[index + 1..]
				.iter_mut()
				.filter_map(|position| position.handler.as_mut()?.abort.take())
			{
				handle.abort();
			}
		}
		positions[index]
			.handler
			.as_mut()
			.expect("a task has a handler")
			.outcome = Some(outcome);
	}
}