use crate::component_logger::{ComponentLogger, LogStrageConfig, log_activities};
use crate::envvar::EnvVar;
use crate::std_output_stream::{LogStream, StdOutput, StdOutputConfig, StdOutputConfigWithSender};
use crate::webhook::webhook_trigger::types_v4_1_0::obelisk::types::join_set::JoinNextError;
use crate::workflow::host_exports::{SUFFIX_FN_SCHEDULE, history_event_schedule_at_from_wast_val};
use crate::{RunnableComponent, WasmFileError};
use assert_matches::assert_matches;
use concepts::prefixed_ulid::{
DeploymentId, ExecutionIdDerived, ExecutionIdTopLevel, JOIN_SET_START_IDX, RunId,
};
use concepts::storage::{
AppendRequest, BacktraceInfo, CreateRequest, DbConnection, DbErrorGeneric,
DbErrorReadWithTimeout, DbErrorWrite, DbPool, ExecutionRequest, HistoryEvent, JoinSetRequest,
LogInfoAppendRow, LogLevel, LogStreamType, TimeoutOutcome, Version,
};
use concepts::time::{ClockFn, Sleep};
use concepts::{
ComponentId, ComponentType, ExecutionFailureKind, ExecutionId, ExecutionMetadata,
FinishedExecutionError, FunctionFqn, FunctionMetadata, FunctionRegistry, IfcFqnName,
JoinSetKind, Params, ReturnType, SUFFIX_PKG_SCHEDULE, SUPPORTED_RETURN_VALUE_OK_EMPTY,
StrVariant, TrapKind,
};
use concepts::{JoinSetId, SupportedFunctionReturnValue};
use http_body_util::combinators::UnsyncBoxBody;
use hyper::body::Bytes;
use hyper::server::conn::http1;
use hyper::{Method, StatusCode, Uri};
use hyper_util::rt::TokioIo;
use route_recognizer::{Match, Router};
use std::ops::Deref;
use std::path::Path;
use std::time::Duration;
use std::{fmt::Debug, sync::Arc};
use tokio::net::TcpListener;
use tokio::select;
use tokio::sync::{OwnedSemaphorePermit, mpsc, watch};
use tracing::{
Instrument, Span, debug, debug_span, error, info, info_span, instrument, trace, warn,
};
use types_v4_1_0::obelisk::types::execution::Host as ExecutionHost;
use types_v4_1_0::obelisk::types::join_set::HostJoinSet;
use val_json::wast_val::WastVal;
use wasmtime::component::ResourceTable;
use wasmtime::component::types::ComponentFunc;
use wasmtime::component::{Linker, Val};
use wasmtime::{Engine, Store, UpdateDeadline};
use wasmtime_wasi::{WasiCtx, WasiCtxBuilder, WasiCtxView, WasiView};
use wasmtime_wasi_http::bindings::ProxyPre;
use wasmtime_wasi_http::bindings::http::types::Scheme;
use wasmtime_wasi_http::body::HyperOutgoingBody;
use wasmtime_wasi_http::{WasiHttpCtx, WasiHttpView};
use wasmtime_wasi_io::IoView;
const HTTP_HANDLER_FFQN: FunctionFqn =
FunctionFqn::new_static("wasi:http/incoming-handler", "handle");
pub(crate) mod types_v4_1_0 {
wasmtime::component::bindgen!({
path: "host-wit-webhook/",
inline: "package any:any;
world bindings {
import obelisk:types/time@4.1.0;
import obelisk:types/execution@4.1.0;
import obelisk:types/join-set@4.1.0;
}",
world: "any:any/bindings",
exports: {
default: trappable | async,
},
with: {
"obelisk:types/join-set.join-set": concepts::JoinSetId,
}
});
}
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
pub struct HttpTriggerConfig {
pub component_id: ComponentId,
}
type StdError = Box<dyn std::error::Error + Send + Sync>;
#[derive(Debug, thiserror::Error)]
pub enum WebhookServerError {
#[error("socket error: {0}")]
SocketError(std::io::Error),
}
pub struct WebhookEndpointCompiled {
pub config: WebhookEndpointConfig,
pub runnable_component: RunnableComponent,
}
impl WebhookEndpointCompiled {
pub fn new(
config: WebhookEndpointConfig,
wasm_path: impl AsRef<Path>,
engine: &Engine,
) -> Result<Self, WasmFileError> {
let runnable_component =
RunnableComponent::new(wasm_path, engine, ComponentType::WebhookEndpoint)?;
Ok(Self {
config,
runnable_component,
})
}
#[must_use]
pub fn imports(&self) -> &[FunctionMetadata] {
&self.runnable_component.wasm_component.exim.imports_flat
}
#[instrument(skip_all, fields(component_id = %self.config.component_id), err)]
pub fn link<S: Sleep>(
self,
engine: &Engine,
fn_registry: &dyn FunctionRegistry,
) -> Result<WebhookEndpointInstanceLinked<S>, WasmFileError> {
let mut linker = Linker::new(engine);
wasmtime_wasi::p2::add_to_linker_async(&mut linker)
.map_err(|err| WasmFileError::linking_error("cannot link `wasmtime_wasi`", err))?;
wasmtime_wasi_http::add_only_http_to_linker_async(&mut linker)
.map_err(|err| WasmFileError::linking_error("cannot link `wasmtime_wasi_http`", err))?;
WebhookEndpointCtx::add_to_linker(&mut linker)?;
for import in fn_registry
.all_exports()
.iter()
.filter(|import| {
!import.ifc_fqn.is_namespace_obelisk() && !import.ifc_fqn.is_namespace_wasi()
})
.filter(|import| {
!import.ifc_fqn.is_extension()
|| import
.ifc_fqn
.package_strip_obelisk_schedule_suffix()
.is_some()
})
{
trace!(
ifc_fqn = %import.ifc_fqn,
"Adding imported interface to the linker",
);
if let Ok(mut linker_instance) = linker.instance(import.ifc_fqn.deref()) {
for function_name in import.fns.keys() {
let ffqn = FunctionFqn {
ifc_fqn: import.ifc_fqn.clone(),
function_name: function_name.clone(),
};
trace!("Adding mock for imported function {ffqn} to the linker");
let res = linker_instance.func_new_async(function_name.deref(), {
let ffqn = ffqn.clone();
move |mut store_ctx: wasmtime::StoreContextMut<
'_,
WebhookEndpointCtx<S>,
>,
_component_func: ComponentFunc,
params: &[Val],
results: &mut [Val]| {
let ffqn = ffqn.clone();
let wasm_backtrace = if self.config.backtrace_persist {
let wasm_backtrace = wasmtime::WasmBacktrace::capture(&store_ctx);
concepts::storage::WasmBacktrace::maybe_from(&wasm_backtrace)
} else {
None
};
Box::new(async move {
Ok(store_ctx
.data_mut()
.call_imported_fn(ffqn, params, results, wasm_backtrace)
.await?)
})
}
});
if let Err(err) = res {
return Err(WasmFileError::linking_error(
format!("cannot add mock for imported function {ffqn}"),
err,
));
}
}
} else {
trace!("Skipping interface {ifc_fqn}", ifc_fqn = import.ifc_fqn);
}
}
let proxy_pre = linker
.instantiate_pre(&self.runnable_component.wasmtime_component)
.map_err(|err: wasmtime::Error| {
WasmFileError::linking_error("linking error while creating instantiate_pre", err)
})?;
let proxy_pre = Arc::new(ProxyPre::new(proxy_pre).map_err(|err: wasmtime::Error| {
WasmFileError::linking_error("linking error while creating ProxyPre instance", err)
})?);
Ok(WebhookEndpointInstanceLinked {
config: self.config,
proxy_pre,
})
}
}
#[derive(Clone, derive_more::Debug)]
pub struct WebhookEndpointInstanceLinked<S: Sleep> {
#[debug(skip)]
proxy_pre: Arc<ProxyPre<WebhookEndpointCtx<S>>>,
config: WebhookEndpointConfig,
}
impl<S: Sleep> WebhookEndpointInstanceLinked<S> {
#[must_use]
pub fn build(
self,
log_forwarder_sender: &mpsc::Sender<LogInfoAppendRow>,
) -> WebhookEndpointInstance<S> {
let stdout = StdOutputConfigWithSender::new(
self.config.forward_stdout,
log_forwarder_sender,
LogStreamType::StdOut,
);
let stderr = StdOutputConfigWithSender::new(
self.config.forward_stderr,
log_forwarder_sender,
LogStreamType::StdErr,
);
WebhookEndpointInstance {
proxy_pre: self.proxy_pre,
stdout,
stderr,
logs_storage_config: self.config.logs_store_min_level.map(|min_level| {
LogStrageConfig {
min_level,
log_sender: log_forwarder_sender.clone(),
}
}),
config: self.config,
}
}
}
#[derive(Clone, derive_more::Debug)]
pub struct WebhookEndpointInstance<S: Sleep> {
#[debug(skip)]
proxy_pre: Arc<ProxyPre<WebhookEndpointCtx<S>>>,
pub config: WebhookEndpointConfig,
#[debug(skip)]
stdout: Option<StdOutputConfigWithSender>,
#[debug(skip)]
stderr: Option<StdOutputConfigWithSender>,
logs_storage_config: Option<LogStrageConfig>,
}
pub struct MethodAwareRouter<T> {
method_map: hashbrown::HashMap<Method, Router<T>>,
fallback: Router<T>, }
impl<T: Clone> MethodAwareRouter<T> {
pub fn add(&mut self, method: Option<Method>, route: &str, dest: T) {
let route = if route.is_empty() { "/*" } else { route };
let mut add = |method, route, dest| {
if let Some(method) = method {
self.method_map.entry(method).or_default().add(route, dest);
} else {
self.fallback.add(route, dest);
}
};
let prefix_with_slash;
if let Some(prefix) = route.strip_suffix("/*") {
prefix_with_slash = format!("{prefix}/");
add(method.clone(), &prefix_with_slash, dest.clone());
}
add(method, route, dest);
}
}
impl<T> MethodAwareRouter<T> {
fn find(&self, method: &Method, path: &Uri) -> Option<Match<&T>> {
let path = path.path();
self.method_map
.get(method)
.and_then(|router| router.recognize(path).ok())
.or_else(|| self.fallback.recognize(path).ok())
}
}
impl<T> Default for MethodAwareRouter<T> {
fn default() -> Self {
Self {
method_map: hashbrown::HashMap::default(),
fallback: Router::default(),
}
}
}
#[expect(clippy::too_many_arguments)]
pub async fn server<S: Sleep>(
deployment_id: DeploymentId,
http_server: StrVariant,
listener: TcpListener,
engine: Arc<Engine>,
router: MethodAwareRouter<WebhookEndpointInstance<S>>,
db_pool: Arc<dyn DbPool>,
clock_fn: Box<dyn ClockFn>,
sleep: S,
fn_registry: Arc<dyn FunctionRegistry>,
max_inflight_requests: Option<Arc<tokio::sync::Semaphore>>,
server_termination_watcher: watch::Receiver<()>,
) -> Result<(), WebhookServerError> {
let router = Arc::new(router);
loop {
let (stream, _) = listener
.accept()
.await
.map_err(WebhookServerError::SocketError)?;
let stream = TokioIo::new(stream);
tokio::task::spawn(
{
let router = router.clone();
let engine = engine.clone();
let clock_fn = clock_fn.clone_box();
let sleep = sleep.clone();
let db_pool = db_pool.clone();
let fn_registry = fn_registry.clone();
let http_server = http_server.clone();
let connection_span = info_span!("connection", %http_server);
let max_inflight_requests = max_inflight_requests.clone();
let server_termination_watcher = server_termination_watcher.clone();
async move {
let (connection_drop_sender, connection_drop_watcher) = watch::channel(());
let res = http1::Builder::new()
.serve_connection(
stream,
hyper::service::service_fn({
move |req| {
let execution_id = ExecutionId::generate().get_top_level();
trace!(%execution_id, method = %req.method(), uri = %req.uri(), "Processing request");
RequestHandler {
deployment_id,
engine: engine.clone(),
clock_fn: clock_fn.clone_box(),
sleep: sleep.clone(),
db_pool: db_pool.clone(),
fn_registry: fn_registry.clone(),
execution_id,
router: router.clone(),
connection_drop_watcher: connection_drop_watcher.clone(),
server_termination_watcher: server_termination_watcher.clone(),
}
.handle_request(req, max_inflight_requests.clone())
}.instrument(info_span!(parent: &connection_span, "request"))
})
)
.await;
if let Err(err) = res {
info!(%http_server, "Error serving connection: {err:?}");
drop(connection_drop_sender);
}
}
}.instrument(debug_span!("tcp stream"))
);
}
}
#[derive(Debug, Clone)]
pub struct WebhookEndpointConfig {
pub component_id: ComponentId,
pub forward_stdout: Option<StdOutputConfig>,
pub forward_stderr: Option<StdOutputConfig>,
pub env_vars: Arc<[EnvVar]>,
pub fuel: Option<u64>,
pub backtrace_persist: bool,
pub subscription_interruption: Option<Duration>,
pub logs_store_min_level: Option<LogLevel>,
}
struct WebhookEndpointCtx<S: Sleep> {
component_id: ComponentId,
deployment_id: DeploymentId,
clock_fn: Box<dyn ClockFn>,
sleep: S,
db_pool: Arc<dyn DbPool>,
fn_registry: Arc<dyn FunctionRegistry>,
table: ResourceTable,
wasi_ctx: WasiCtx,
http_ctx: WasiHttpCtx,
execution_id: ExecutionIdTopLevel,
next_join_set_idx: u64,
version: Option<Version>,
component_logger: ComponentLogger,
subscription_interruption: Option<Duration>,
connection_drop_watcher: watch::Receiver<()>,
server_termination_watcher: watch::Receiver<()>,
}
impl<S: Sleep> HostJoinSet for WebhookEndpointCtx<S> {
fn id(&mut self, _resource: wasmtime::component::Resource<JoinSetId>) -> String {
unreachable!("webhook endpoint instances cannot obtain `join-set-id` resource")
}
fn submit_delay(
&mut self,
_self_: wasmtime::component::Resource<JoinSetId>,
_timeout: types_v4_1_0::obelisk::types::time::ScheduleAt,
) -> types_v4_1_0::obelisk::types::execution::DelayId {
unreachable!("webhook endpoint instances cannot obtain `join-set-id` resource")
}
fn join_next(
&mut self,
_self_: wasmtime::component::Resource<JoinSetId>,
) -> Result<
(
types_v4_1_0::obelisk::types::execution::ResponseId,
Result<(), ()>,
),
JoinNextError,
> {
unreachable!("webhook endpoint instances cannot obtain `join-set-id` resource")
}
fn drop(
&mut self,
_resource: wasmtime::component::Resource<JoinSetId>,
) -> wasmtime::Result<()> {
unreachable!("webhook endpoint instances cannot obtain `join-set-id` resource")
}
}
impl<S: Sleep> ExecutionHost for WebhookEndpointCtx<S> {}
#[derive(thiserror::Error, Debug, Clone)]
enum WebhookEndpointFunctionError {
#[error(transparent)]
DbError(#[from] DbErrorWrite),
#[error(transparent)]
FinishedExecutionError(#[from] FinishedExecutionError),
#[error("uncategorized error: {0}")]
UncategorizedError(&'static str),
#[error("connection closed")]
ConnectionClosed,
}
impl From<DbErrorGeneric> for WebhookEndpointFunctionError {
fn from(value: DbErrorGeneric) -> Self {
WebhookEndpointFunctionError::DbError(DbErrorWrite::Generic(value))
}
}
impl<S: Sleep> wasmtime::component::HasData for WebhookEndpointCtx<S> {
type Data<'a> = &'a mut WebhookEndpointCtx<S>;
}
impl<S: Sleep> WebhookEndpointCtx<S> {
async fn get_version_or_create(&mut self) -> Result<Version, DbErrorWrite> {
if let Some(found) = &self.version {
return Ok(found.clone());
}
let created_at = self.clock_fn.now();
let metadata = concepts::ExecutionMetadata::from_parent_span(&self.component_logger.span);
let create_request = CreateRequest {
created_at,
execution_id: ExecutionId::TopLevel(self.execution_id),
ffqn: HTTP_HANDLER_FFQN,
params: Params::empty(),
parent: None,
metadata,
scheduled_at: created_at,
component_id: self.component_id.clone(),
deployment_id: self.deployment_id,
scheduled_by: None,
};
let conn = self.db_pool.connection().await?;
let version = conn.create(create_request).await?;
self.version = Some(version.clone());
Ok(version)
}
#[instrument(skip_all, fields(%ffqn, version, %execution_id = self.execution_id))]
async fn call_imported_fn(
&mut self,
ffqn: FunctionFqn,
params: &[Val],
results: &mut [Val],
wasm_backtrace: Option<concepts::storage::WasmBacktrace>,
) -> Result<(), WebhookEndpointFunctionError> {
trace!(?params, "call_imported_fn start");
assert_eq!(
1,
results.len(),
"direct call: no-ext export must return `result`, -schedule returns `execuiton-id`"
);
if self.connection_drop_watcher.has_changed().is_err()
|| self.server_termination_watcher.has_changed().is_err()
{
debug!("Cancellation request detected");
return Err(WebhookEndpointFunctionError::ConnectionClosed);
}
if let Some(package_name) = ffqn.ifc_fqn.package_strip_obelisk_schedule_suffix() {
let ifc_fqn = IfcFqnName::from_parts(
ffqn.ifc_fqn.namespace(),
package_name,
ffqn.ifc_fqn.ifc_name(),
ffqn.ifc_fqn.version(),
);
if let Some(function_name) = ffqn.function_name.strip_suffix(SUFFIX_FN_SCHEDULE) {
let ffqn =
FunctionFqn::new_arc(Arc::from(ifc_fqn.to_string()), Arc::from(function_name));
debug!("Got `-schedule` extension for {ffqn}");
let Some((schedule_at, params)) = params.split_first() else {
error!(
"Error running `-schedule` extension function: exepcted at least one parameter of type `schedule-at`, got empty parameter list"
);
return Err(WebhookEndpointFunctionError::UncategorizedError(
"error running `-schedule` extension function: exepcted at least one parameter of type `schedule-at`, got empty parameter list",
));
};
let schedule_at =
WastVal::try_from(schedule_at.clone()).map_err(|err| {
error!("Error running `-schedule` extension function: cannot convert to internal representation - {err:?}");
WebhookEndpointFunctionError::UncategorizedError(
"error running `-schedule` extension function: cannot convert to internal representation",
)
})?;
let schedule_at = match history_event_schedule_at_from_wast_val(&schedule_at) {
Ok(ok) => ok,
Err(err) => {
error!(
"Wrong type for the first `-schedule` extension function parameter, expected `schedule-at`, got `{schedule_at:?}` - {err:?}"
);
return Err(WebhookEndpointFunctionError::UncategorizedError(
"error running `-schedule` extension function: wrong first parameter type",
));
}
};
let version = self.get_version_or_create().await?;
let span = Span::current();
span.record("version", tracing::field::display(&version));
let new_execution_id = ExecutionId::generate();
let (_function_metadata, child_component_id) = self
.fn_registry
.get_by_exported_function(&ffqn)
.expect("target function must be found in fn_registry");
let created_at = self.clock_fn.now();
let event = HistoryEvent::Schedule {
execution_id: new_execution_id.clone(),
schedule_at,
};
let schedule_at = schedule_at.as_date_time(created_at).map_err(|_err| {
WebhookEndpointFunctionError::UncategorizedError("schedule-at conversion error")
})?;
let child_exec_req = AppendRequest {
event: ExecutionRequest::HistoryEvent { event },
created_at,
};
let create_child_req = CreateRequest {
created_at,
execution_id: new_execution_id.clone(),
ffqn,
params: Params::from_wasmtime(Arc::from(params)),
parent: None, metadata: ExecutionMetadata::from_linked_span(&self.component_logger.span),
scheduled_at: schedule_at,
component_id: child_component_id.clone(),
deployment_id: self.deployment_id,
scheduled_by: Some(ExecutionId::TopLevel(self.execution_id)),
};
let db_connection = self.db_pool.connection().await?;
let expected_next_version = version.increment();
let backtrace_info = wasm_backtrace.map(|wasm_backtrace| BacktraceInfo {
execution_id: ExecutionId::TopLevel(self.execution_id),
component_id: self.component_id.clone(),
version_min_including: version.clone(),
version_max_excluding: expected_next_version.clone(),
wasm_backtrace,
});
let version = db_connection
.append_batch_create_new_execution(
created_at,
vec![child_exec_req],
ExecutionId::TopLevel(self.execution_id),
version.clone(),
vec![create_child_req],
backtrace_info.into_iter().collect(),
)
.await?;
assert_eq!(version, expected_next_version); self.version = Some(version.clone());
results[0] = execution_id_into_val(&new_execution_id);
} else {
error!("unrecognized `{SUFFIX_PKG_SCHEDULE}` extension function {ffqn}");
return Err(WebhookEndpointFunctionError::UncategorizedError(
"unrecognized extension function",
));
}
} else {
let version = self.get_version_or_create().await?;
let span = Span::current();
span.record("version", tracing::field::display(&version));
let join_set_id_direct = JoinSetId::new(
JoinSetKind::OneOff,
StrVariant::from(self.next_join_set_idx.to_string()),
)
.expect("numeric names must be allowed");
self.next_join_set_idx += 1;
let child_execution_id =
ExecutionId::TopLevel(self.execution_id).next_level(&join_set_id_direct);
let created_at = self.clock_fn.now();
let (fn_metadata, child_component_id) = self
.fn_registry
.get_by_exported_function(&ffqn)
.expect("import was mocked using fn_registry exports limited to -schedule and no-ext functions");
assert!(
fn_metadata.extension.is_none(),
"direct call: function must be no-ext"
);
let return_type_tl = assert_matches!(fn_metadata.return_type, ReturnType::Extendable(compatible) => compatible.type_wrapper_tl);
let req_join_set_created = AppendRequest {
created_at,
event: ExecutionRequest::HistoryEvent {
event: HistoryEvent::JoinSetCreate {
join_set_id: join_set_id_direct.clone(),
},
},
};
let params = Params::from_wasmtime(Arc::from(params));
let req_child_exec = AppendRequest {
created_at,
event: ExecutionRequest::HistoryEvent {
event: HistoryEvent::JoinSetRequest {
join_set_id: join_set_id_direct.clone(),
request: JoinSetRequest::ChildExecutionRequest {
child_execution_id: child_execution_id.clone(),
target_ffqn: ffqn.clone(),
params: params.clone(),
},
},
},
};
let req_join_next = AppendRequest {
created_at,
event: ExecutionRequest::HistoryEvent {
event: HistoryEvent::JoinNext {
join_set_id: join_set_id_direct.clone(),
run_expires_at: created_at, closing: false,
requested_ffqn: Some(ffqn.clone()), },
},
};
let req_create_child = CreateRequest {
created_at,
execution_id: ExecutionId::Derived(child_execution_id.clone()),
ffqn: ffqn.clone(),
params,
parent: Some((ExecutionId::TopLevel(self.execution_id), join_set_id_direct)),
metadata: ExecutionMetadata::from_parent_span(&self.component_logger.span),
scheduled_at: created_at,
component_id: child_component_id.clone(),
deployment_id: self.deployment_id,
scheduled_by: None,
};
let db_connection = self.db_pool.connection().await?;
let appended = vec![req_join_set_created, req_child_exec, req_join_next];
let expected_next_version = Version(version.0 + 3);
let backtrace_info = wasm_backtrace.map(|wasm_backtrace| BacktraceInfo {
execution_id: ExecutionId::TopLevel(self.execution_id),
component_id: self.component_id.clone(),
version_min_including: version.clone(),
version_max_excluding: expected_next_version.clone(),
wasm_backtrace,
});
let version = db_connection
.append_batch_create_new_execution(
created_at,
appended,
ExecutionId::TopLevel(self.execution_id),
version,
vec![req_create_child],
backtrace_info.into_iter().collect(),
)
.await?;
assert_eq!(version, expected_next_version); self.version = Some(version);
let res = Self::wait_for_finished_result(
self.subscription_interruption,
&self.sleep,
db_connection.as_ref(),
child_execution_id,
&self.connection_drop_watcher,
&self.server_termination_watcher,
)
.await?;
results[0] = res.into_wast_val(move || return_type_tl).as_val();
trace!(?results, "call_imported_fn finish");
}
Ok(())
}
async fn wait_for_finished_result(
subscription_interruption: Option<Duration>,
sleep: &S,
db_connection: &dyn DbConnection,
child_execution_id: ExecutionIdDerived,
connection_drop_watcher: &watch::Receiver<()>,
server_termination_watcher: &watch::Receiver<()>,
) -> Result<SupportedFunctionReturnValue, WebhookEndpointFunctionError> {
let child_execution_id = ExecutionId::Derived(child_execution_id);
let timeout_factory = move || {
let subscription_interruption = subscription_interruption.unwrap_or(Duration::MAX);
let sleep = sleep.clone();
let mut connection_drop_watcher = connection_drop_watcher.clone();
let mut server_termination_watcher = server_termination_watcher.clone();
Box::pin(async move {
select! {
() = sleep.sleep(subscription_interruption) => TimeoutOutcome::Timeout,
_ = connection_drop_watcher.changed() => TimeoutOutcome::Cancel,
_ = server_termination_watcher.changed() => TimeoutOutcome::Cancel,
}
})
};
loop {
let timeout = timeout_factory();
let res = db_connection
.wait_for_finished_result(&child_execution_id, Some(timeout))
.await;
match res {
Ok(ok) => {
trace!("Finished ok");
return Ok(ok);
}
Err(DbErrorReadWithTimeout::Timeout(TimeoutOutcome::Timeout)) => {
trace!("Timeout triggers resubscribing");
}
Err(DbErrorReadWithTimeout::Timeout(TimeoutOutcome::Cancel)) => {
debug!("Connection closed, not waiting for result");
return Err(WebhookEndpointFunctionError::ConnectionClosed);
}
Err(DbErrorReadWithTimeout::DbErrorRead(err)) => {
warn!("Database error: {err:?}");
return Err(WebhookEndpointFunctionError::from(DbErrorWrite::from(err)));
}
}
}
}
fn add_to_linker(linker: &mut Linker<WebhookEndpointCtx<S>>) -> Result<(), WasmFileError> {
log_activities::obelisk::log::log::add_to_linker::<_, WebhookEndpointCtx<S>>(linker, |x| x)
.map_err(|err| WasmFileError::linking_error("cannot link log activities", err))?;
types_v4_1_0::obelisk::types::execution::add_to_linker::<_, WebhookEndpointCtx<S>>(
linker,
|x| x,
)
.map_err(|err| WasmFileError::linking_error("cannot link obelisk:types", err))?;
Ok(())
}
#[must_use]
#[expect(clippy::too_many_arguments)]
fn new<'a>(
deployment_id: DeploymentId,
config: WebhookEndpointConfig,
engine: &Engine,
clock_fn: Box<dyn ClockFn>,
sleep: S,
db_pool: Arc<dyn DbPool>,
fn_registry: Arc<dyn FunctionRegistry>,
params: impl Iterator<Item = (&'a str, &'a str)>,
execution_id: ExecutionIdTopLevel,
request_span: Span,
connection_drop_watcher: watch::Receiver<()>,
server_termination_watcher: watch::Receiver<()>,
stdout: Option<StdOutput>,
stderr: Option<StdOutput>,
run_id: RunId,
logs_storage_config: Option<LogStrageConfig>,
) -> Store<WebhookEndpointCtx<S>> {
let mut wasi_ctx = WasiCtxBuilder::new();
if let Some(stdout) = stdout {
let stdout = LogStream::new(
format!(
"[{component_id} {execution_id} stdout]",
component_id = config.component_id
),
stdout,
);
wasi_ctx.stdout(stdout);
}
if let Some(stderr) = stderr {
let stderr = LogStream::new(
format!(
"[{component_id} {execution_id} stderr]",
component_id = config.component_id
),
stderr,
);
wasi_ctx.stderr(stderr);
}
for env_var in config.env_vars.as_ref() {
wasi_ctx.env(&env_var.key, &env_var.val);
}
for (key, val) in params {
wasi_ctx.env(key, val);
}
let wasi_ctx = wasi_ctx.build();
let ctx = WebhookEndpointCtx {
clock_fn,
sleep,
db_pool,
fn_registry,
table: ResourceTable::new(),
wasi_ctx,
http_ctx: WasiHttpCtx::new(),
version: None,
component_id: config.component_id,
deployment_id,
next_join_set_idx: JOIN_SET_START_IDX,
execution_id,
component_logger: ComponentLogger {
span: request_span,
execution_id: ExecutionId::TopLevel(execution_id),
run_id,
logs_storage_config,
},
subscription_interruption: config.subscription_interruption,
connection_drop_watcher,
server_termination_watcher,
};
let mut store = Store::new(engine, ctx);
if let Some(fuel) = config.fuel {
store
.set_fuel(fuel)
.expect("engine must have `consume_fuel` enabled");
}
store.epoch_deadline_callback(|_store_ctx| Ok(UpdateDeadline::Yield(1)));
store
}
async fn close(
self,
original_result: wasmtime::Result<()>,
assigned_fuel: Option<u64>,
) -> wasmtime::Result<()> {
#[derive(Debug, thiserror::Error)]
#[error("webhook {trap_kind}: {reason}")]
struct WebhookTrap {
reason: String,
trap_kind: TrapKind,
detail: Option<String>,
}
let result = match &original_result {
Ok(()) => SUPPORTED_RETURN_VALUE_OK_EMPTY,
Err(err) => {
let err = if let Some(trap) = err
.source()
.and_then(|source| source.downcast_ref::<wasmtime::Trap>())
{
if *trap == wasmtime::Trap::OutOfFuel {
WebhookTrap {
reason: format!(
"total fuel consumed: {}",
assigned_fuel
.expect("must have been set as it was the reason of trap")
),
detail: None,
trap_kind: TrapKind::OutOfFuel,
}
} else {
WebhookTrap {
reason: trap.to_string(),
detail: Some(format!("{err:?}")),
trap_kind: TrapKind::Trap,
}
}
} else {
WebhookTrap {
reason: err.to_string(),
trap_kind: TrapKind::HostFunctionError,
detail: Some(format!("{err:?}")),
}
};
SupportedFunctionReturnValue::ExecutionError(FinishedExecutionError {
reason: Some(err.to_string()),
kind: ExecutionFailureKind::Uncategorized,
detail: err.detail,
})
}
};
if let Some(version) = self.version {
self.db_pool
.connection()
.await?
.append(
ExecutionId::TopLevel(self.execution_id),
version,
AppendRequest {
created_at: self.clock_fn.now(),
event: ExecutionRequest::Finished {
result,
http_client_traces: None,
},
},
)
.await?;
}
original_result
}
}
impl<S: Sleep> log_activities::obelisk::log::log::Host for WebhookEndpointCtx<S> {
fn trace(&mut self, message: String) {
self.component_logger.log(LogLevel::Trace, message);
}
fn debug(&mut self, message: String) {
self.component_logger.log(LogLevel::Debug, message);
}
fn info(&mut self, message: String) {
self.component_logger.log(LogLevel::Info, message);
}
fn warn(&mut self, message: String) {
self.component_logger.log(LogLevel::Warn, message);
}
fn error(&mut self, message: String) {
self.component_logger.log(LogLevel::Error, message);
}
}
impl<S: Sleep> WasiView for WebhookEndpointCtx<S> {
fn ctx(&mut self) -> WasiCtxView<'_> {
WasiCtxView {
ctx: &mut self.wasi_ctx,
table: &mut self.table,
}
}
}
impl<S: Sleep> IoView for WebhookEndpointCtx<S> {
fn table(&mut self) -> &mut ResourceTable {
&mut self.table
}
}
impl<S: Sleep> WasiHttpView for WebhookEndpointCtx<S> {
fn ctx(&mut self) -> &mut WasiHttpCtx {
&mut self.http_ctx
}
fn table(&mut self) -> &mut ResourceTable {
&mut self.table
}
}
struct RequestHandler<S: Sleep> {
deployment_id: DeploymentId,
engine: Arc<Engine>,
clock_fn: Box<dyn ClockFn>,
sleep: S,
db_pool: Arc<dyn DbPool>,
fn_registry: Arc<dyn FunctionRegistry>,
execution_id: ExecutionIdTopLevel,
router: Arc<MethodAwareRouter<WebhookEndpointInstance<S>>>,
connection_drop_watcher: watch::Receiver<()>,
server_termination_watcher: watch::Receiver<()>,
}
fn respond(body: &str, status_code: StatusCode) -> hyper::Response<HyperOutgoingBody> {
let body = UnsyncBoxBody::new(http_body_util::BodyExt::map_err(
http_body_util::Full::new(Bytes::copy_from_slice(body.as_bytes())),
|_| unreachable!(),
));
hyper::Response::builder()
.status(status_code)
.body(body)
.unwrap()
}
impl<S: Sleep> RequestHandler<S> {
#[instrument(skip_all, name="incoming webhook request", fields(execution_id = %self.execution_id))]
async fn handle_request(
self,
req: hyper::Request<hyper::body::Incoming>,
max_inflight_requests: Option<Arc<tokio::sync::Semaphore>>,
) -> Result<hyper::Response<HyperOutgoingBody>, hyper::Error> {
let http_request_guard = if let Some(http_request_semaphore) = &max_inflight_requests {
http_request_semaphore.clone().try_acquire_owned().map(Some)
} else {
Ok(None)
};
let Ok(http_request_guard) = http_request_guard else {
debug!(method = %req.method(), uri = %req.uri(), "Too many requests");
return Ok::<_, hyper::Error>(respond("Out of permits", StatusCode::TOO_MANY_REQUESTS));
};
let res = self
.handle_request_inner(req, http_request_guard, Span::current())
.await;
match res {
Ok(body) => Ok(body),
Err(err) => {
debug!("{err:?}");
Ok(match err {
HandleRequestError::IncomingRequestError(err) => respond(
&format!("Incoming request error: {err}"),
StatusCode::BAD_REQUEST,
),
HandleRequestError::ResponseCreationError(err) => respond(
&format!("Cannot create response: {err}"),
StatusCode::INTERNAL_SERVER_ERROR,
),
HandleRequestError::InstantiationError(err) => respond(
&format!("Cannot instantiate: {err}"),
StatusCode::SERVICE_UNAVAILABLE,
),
HandleRequestError::ErrorCode(code) => respond(
&format!("Error code: {code}"),
StatusCode::INTERNAL_SERVER_ERROR,
),
HandleRequestError::ExecutionError(_) => {
respond("Component Error", StatusCode::INTERNAL_SERVER_ERROR)
}
HandleRequestError::RouteNotFound => {
respond("Route not found", StatusCode::NOT_FOUND)
}
HandleRequestError::Timeout => respond("Timeout", StatusCode::REQUEST_TIMEOUT),
HandleRequestError::InstanceLimitReached => {
respond("Instance limit reached", StatusCode::SERVICE_UNAVAILABLE)
}
})
}
}
}
async fn handle_request_inner(
self,
req: hyper::Request<hyper::body::Incoming>,
http_request_guard: Option<OwnedSemaphorePermit>,
request_span: Span,
) -> Result<hyper::Response<HyperOutgoingBody>, HandleRequestError> {
#[derive(Debug, thiserror::Error)]
#[error("timeout")]
struct TimeoutError;
if let Some(instance_match) = self.router.find(req.method(), req.uri()) {
let found_instance = instance_match.handler();
let run_id = RunId::generate();
let stdout = found_instance.stdout.as_ref().map(|stdoutput| {
stdoutput.build(&ExecutionId::TopLevel(self.execution_id), run_id)
});
let stderr = found_instance.stderr.as_ref().map(|stdoutput| {
stdoutput.build(&ExecutionId::TopLevel(self.execution_id), run_id)
});
let (sender, receiver) = tokio::sync::oneshot::channel();
let mut store = WebhookEndpointCtx::new(
self.deployment_id,
found_instance.config.clone(),
&self.engine,
self.clock_fn,
self.sleep,
self.db_pool,
self.fn_registry,
instance_match.params().iter(),
self.execution_id,
request_span.clone(),
self.connection_drop_watcher,
self.server_termination_watcher,
stdout,
stderr,
run_id,
found_instance.logs_storage_config.clone(),
);
let req = store
.data_mut()
.new_incoming_request(Scheme::Http, req)
.map_err(|err| HandleRequestError::IncomingRequestError(err.into()))?;
let out = store
.data_mut()
.new_response_outparam(sender)
.map_err(|err| HandleRequestError::ResponseCreationError(err.into()))?;
let proxy = found_instance
.proxy_pre
.instantiate_async(&mut store)
.await
.map_err(|err| HandleRequestError::InstantiationError(err.into()))?;
let task = tokio::task::spawn({
let assigned_fuel = found_instance.config.fuel;
async move {
let _http_request_guard = http_request_guard;
let result = proxy
.wasi_http_incoming_handler()
.call_handle(&mut store, req, out)
.await
.inspect_err(|err| debug!("Webhook instance returned error: {err:?}"));
let ctx = store.into_data();
ctx.close(result, assigned_fuel).await
}
.instrument(request_span)
});
match receiver.await {
Ok(Ok(resp)) => {
trace!("Streaming the response");
Ok(resp)
}
Ok(Err(err)) => {
debug!("Webhook instance sent error code {err:?}");
Err(HandleRequestError::ErrorCode(err))
}
Err(_recv_err) => {
let err = match task.await {
Ok(r) => {
r.expect_err("if the receiver has an error, the task must have failed")
} Err(e) => e.into(), };
if err.downcast_ref::<TimeoutError>().is_some() {
Err(HandleRequestError::Timeout)
} else {
info!("Webhook task ended with ExecutionError - {err:?}");
Err(HandleRequestError::ExecutionError(err.into()))
}
}
}
} else {
Err(HandleRequestError::RouteNotFound)
}
}
}
#[derive(Debug, thiserror::Error)]
pub enum HandleRequestError {
#[error("incoming request error: {0}")]
IncomingRequestError(StdError),
#[error("response creation error: {0}")]
ResponseCreationError(StdError),
#[error("instantiation error: {0}")]
InstantiationError(StdError),
#[error("error code: {0}")]
ErrorCode(wasmtime_wasi_http::bindings::http::types::ErrorCode),
#[error("execution error: {0}")]
ExecutionError(StdError),
#[error("route not found")]
RouteNotFound,
#[error("instance limit reached")]
InstanceLimitReached,
#[error("timeout")]
Timeout,
}
fn execution_id_into_val(execution_id: &ExecutionId) -> Val {
Val::Record(vec![(
"id".to_string(),
Val::String(execution_id.to_string()),
)])
}
#[cfg(test)]
pub(crate) mod tests {
use super::MethodAwareRouter;
use hyper::{Method, Uri};
pub(crate) mod nosim {
use super::*;
use crate::RunnableComponent;
use crate::activity::activity_worker::tests::{compile_activity, new_activity_fibo};
use crate::activity::cancel_registry::CancelRegistry;
use crate::engines::{EngineConfig, Engines};
use crate::testing_fn_registry::TestingFnRegistry;
use crate::webhook::webhook_trigger::{
self, WebhookEndpointCompiled, WebhookEndpointConfig,
};
use crate::workflow::workflow_worker::JoinNextBlockingStrategy;
use crate::workflow::workflow_worker::tests::{
FIBOA_WORKFLOW_FFQN, compile_workflow, new_workflow_fibo,
};
use concepts::component_id::InputContentDigest;
use concepts::prefixed_ulid::{DEPLOYMENT_ID_DUMMY, RunId};
use concepts::storage::DbPoolCloseable;
use concepts::time::ClockFn;
use concepts::time::TokioSleep;
use concepts::{ComponentId, ComponentType, Params, StrVariant};
use concepts::{ExecutionId, storage::DbPool};
use db_tests::{Database, DbGuard, DbPoolCloseableWrapper};
use executor::executor::{ExecTask, LockingStrategy};
use rstest::rstest;
use serde_json::json;
use std::net::SocketAddr;
use std::str::FromStr;
use std::sync::Arc;
use std::time::Duration;
use test_db_macro::expand_enum_database;
use test_utils::sim_clock::SimClock;
use tokio::net::TcpListener;
use tokio::sync::{mpsc, watch};
use tracing::info;
use utils::sha256sum::calculate_sha256_file;
struct AbortOnDrop(tokio::task::AbortHandle);
impl Drop for AbortOnDrop {
fn drop(&mut self) {
self.0.abort();
}
}
pub(crate) fn compile_webhook(wasm_path: &str) -> RunnableComponent {
let engine = Engines::get_webhook_engine(EngineConfig::on_demand_testing()).unwrap();
RunnableComponent::new(wasm_path, &engine, ComponentType::WebhookEndpoint).unwrap()
}
struct SetUpFiboWebhook {
#[expect(dead_code)]
server: AbortOnDrop,
#[expect(dead_code)]
guard: DbGuard,
db_pool: Arc<dyn DbPool>,
server_addr: SocketAddr,
activity_exec: ExecTask,
workflow_exec: ExecTask,
sim_clock: SimClock,
db_close: DbPoolCloseableWrapper,
#[expect(dead_code)]
server_termination_sender: watch::Sender<()>,
}
impl SetUpFiboWebhook {
async fn new(
db: db_tests::Database,
locking_strategy: LockingStrategy,
) -> SetUpFiboWebhook {
let addr = SocketAddr::from(([127, 0, 0, 1], 0));
let sim_clock = SimClock::default();
let (guard, db_pool, db_close) = db.set_up().await;
let activity_exec = new_activity_fibo(
db_pool.clone(),
sim_clock.clone_box(),
TokioSleep,
locking_strategy,
)
.await;
let (workflow_runnable, workflow_component_id) = compile_workflow(
test_programs_fibo_workflow_builder::TEST_PROGRAMS_FIBO_WORKFLOW,
)
.await;
let fn_registry = TestingFnRegistry::new_from_components(vec![
compile_activity(
test_programs_fibo_activity_builder::TEST_PROGRAMS_FIBO_ACTIVITY,
)
.await,
(workflow_runnable, workflow_component_id),
]);
let cancel_registry = CancelRegistry::new();
let engine =
Engines::get_webhook_engine(EngineConfig::on_demand_testing()).unwrap();
let workflow_exec = new_workflow_fibo(
db_pool.clone(),
sim_clock.clone_box(),
JoinNextBlockingStrategy::Interrupt,
&fn_registry,
cancel_registry,
locking_strategy,
)
.await;
let (db_forwarder_sender, _) = mpsc::channel(1);
let wasm_file = test_programs_fibo_webhook_builder::TEST_PROGRAMS_FIBO_WEBHOOK;
let router = {
let instance = WebhookEndpointCompiled::new(
WebhookEndpointConfig {
component_id: ComponentId::new(
ComponentType::WebhookEndpoint,
StrVariant::empty(),
InputContentDigest(calculate_sha256_file(wasm_file).await.unwrap()),
)
.unwrap(),
forward_stdout: None,
forward_stderr: None,
env_vars: Arc::from([]),
fuel: None,
backtrace_persist: false,
subscription_interruption: None,
logs_store_min_level: None,
},
wasm_file,
&engine,
)
.unwrap()
.link(&engine, fn_registry.as_ref())
.unwrap()
.build(&db_forwarder_sender);
let mut router = MethodAwareRouter::default();
router.add(Some(Method::GET), "/fibo/:N/:ITERATIONS", instance);
router
};
let tcp_listener = TcpListener::bind(addr).await.unwrap();
let server_addr = tcp_listener.local_addr().unwrap();
info!("Listening on port {}", server_addr.port());
let (server_termination_sender, server_termination_watcher) = watch::channel(());
let server = AbortOnDrop(
tokio::spawn(webhook_trigger::server(
DEPLOYMENT_ID_DUMMY,
StrVariant::Static("test"),
tcp_listener,
engine,
router,
db_pool.clone(),
sim_clock.clone_box(),
TokioSleep,
fn_registry,
None,
server_termination_watcher,
))
.abort_handle(),
);
SetUpFiboWebhook {
server,
guard,
db_pool,
server_addr,
activity_exec,
workflow_exec,
sim_clock,
db_close,
server_termination_sender,
}
}
async fn fetch(
server_addr: &str,
n: u8,
iterations: u32,
expected_status_code: u16,
) -> String {
let resp = reqwest::get(format!("http://{server_addr}/fibo/{n}/{iterations}",))
.await
.unwrap();
assert_eq!(resp.status().as_u16(), expected_status_code);
resp.text().await.unwrap()
}
async fn close(self) {
self.db_close.close().await;
}
}
#[rstest]
#[tokio::test]
async fn hardcoded_result_should_work(
#[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
locking_strategy: LockingStrategy,
) {
test_utils::set_up();
let fibo_webhook_harness =
SetUpFiboWebhook::new(Database::Memory, locking_strategy).await;
let server_addr = fibo_webhook_harness.server_addr.to_string();
assert_eq!(
"fiboa(1, 0) = hardcoded: 1",
SetUpFiboWebhook::fetch(&server_addr, 1, 0, 200).await
);
}
#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn direct_call_should_work_nondeterministic(
db: Database,
#[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
locking_strategy: LockingStrategy,
) {
test_utils::set_up();
let fibo_webhook_harness = SetUpFiboWebhook::new(db, locking_strategy).await;
let server_addr = fibo_webhook_harness.server_addr.to_string();
let fetch_task =
tokio::spawn(async move { SetUpFiboWebhook::fetch(&server_addr, 2, 1, 200).await });
let now = fibo_webhook_harness.sim_clock.now();
while fibo_webhook_harness
.workflow_exec
.tick_test_await(now, RunId::generate())
.await
.is_empty()
{
tokio::time::sleep(Duration::from_millis(100)).await;
}
assert_eq!(
1,
fibo_webhook_harness
.activity_exec
.tick_test_await(now, RunId::generate())
.await
.len()
);
assert_eq!(
1,
fibo_webhook_harness
.workflow_exec
.tick_test_await(now, RunId::generate())
.await
.len()
);
let res = fetch_task.await.unwrap();
assert_eq!("fiboa(2, 1) = direct call: 1", res);
}
#[expand_enum_database]
#[rstest]
#[tokio::test]
async fn scheduling_should_work(
db: db_tests::Database,
#[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
locking_strategy: LockingStrategy,
) {
test_utils::set_up();
let fibo_webhook_harness = SetUpFiboWebhook::new(db, locking_strategy).await;
let server_addr = fibo_webhook_harness.server_addr.to_string();
let n = 10;
let iterations = 1;
let resp = SetUpFiboWebhook::fetch(&server_addr, n, iterations, 200).await;
let execution_id = resp
.strip_prefix(&format!("fiboa({n}, {iterations}) = scheduled: "))
.unwrap();
let execution_id = ExecutionId::from_str(execution_id).unwrap();
let conn = fibo_webhook_harness.db_pool.connection().await.unwrap();
let create_req = conn.get_create_request(&execution_id).await.unwrap();
assert_eq!(FIBOA_WORKFLOW_FFQN, create_req.ffqn);
let expected_params = Params::from_json_values_test(vec![json!(10), json!(1)]);
assert_eq!(
serde_json::to_string(&expected_params).unwrap(),
serde_json::to_string(&create_req.params).unwrap()
);
}
#[rstest]
#[tokio::test]
async fn test_routing_error_handling(
#[values(LockingStrategy::ByFfqns, LockingStrategy::ByComponentDigest)]
locking_strategy: LockingStrategy,
) {
test_utils::set_up();
let fibo_webhook_harness =
SetUpFiboWebhook::new(Database::Memory, locking_strategy).await;
let resp = reqwest::get(format!(
"http://{}/unknown",
&fibo_webhook_harness.server_addr
))
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 404);
assert_eq!("Route not found", resp.text().await.unwrap());
let resp = reqwest::get(format!(
"http://{}/fibo/0/1",
&fibo_webhook_harness.server_addr
))
.await
.unwrap();
assert_eq!(resp.status().as_u16(), 500);
assert_eq!("Component Error", resp.text().await.unwrap());
fibo_webhook_harness.close().await;
}
}
#[test]
fn routes() {
let mut router = MethodAwareRouter::default();
router.add(Some(Method::GET), "/foo", 1);
router.add(Some(Method::GET), "/foo/*", 2);
router.add(None, "/foo", 3);
router.add(None, "/*", 4);
router.add(None, "/", 5);
router.add(Some(Method::GET), "/path/:param1/:param2", 6);
assert_eq!(
1,
**router
.find(&Method::GET, &Uri::from_static("/foo"))
.unwrap()
.handler()
);
assert_eq!(
2,
**router
.find(&Method::GET, &Uri::from_static("/foo/"))
.unwrap()
.handler()
);
assert_eq!(
2,
**router
.find(&Method::GET, &Uri::from_static("/foo/foo/"))
.unwrap()
.handler()
);
assert_eq!(
2,
**router
.find(&Method::GET, &Uri::from_static("/foo/foo/bar"))
.unwrap()
.handler()
);
assert_eq!(
3,
**router
.find(&Method::POST, &Uri::from_static("/foo"))
.unwrap()
.handler()
);
assert_eq!(
5,
**router
.find(&Method::GET, &Uri::from_static("/"))
.unwrap()
.handler()
);
let found = router
.find(&Method::GET, &Uri::from_static("/path/p1/p2"))
.unwrap();
assert_eq!(6, **found.handler());
assert_eq!(
hashbrown::HashMap::from([("param1", "p1"), ("param2", "p2")]),
found
.params()
.into_iter()
.collect::<hashbrown::HashMap<_, _>>()
);
let found = router
.find(&Method::GET, &Uri::from_static("/path/p1/p2/p3"))
.unwrap();
assert_eq!(4, **found.handler());
}
#[test]
fn routes_empty_fallback() {
let mut router = MethodAwareRouter::default();
router.add(Some(Method::GET), "/foo", 1);
router.add(None, "", 9);
assert_eq!(
1,
**router
.find(&Method::GET, &Uri::from_static("/foo"))
.unwrap()
.handler()
);
assert_eq!(
9,
**router
.find(&Method::GET, &Uri::from_static("/"))
.unwrap()
.handler()
);
assert_eq!(
9,
**router
.find(&Method::GET, &Uri::from_static("/x"))
.unwrap()
.handler()
);
assert_eq!(
9,
**router
.find(&Method::GET, &Uri::from_static("/x/"))
.unwrap()
.handler()
);
}
}