From 1def7a4c978d5aea795a893741388e006d4d39a9 Mon Sep 17 00:00:00 2001 From: joshua-spacetime Date: Mon, 27 Jul 2026 10:35:06 -0700 Subject: [PATCH 1/3] Add internal endpoint for tokio task dumps --- Cargo.lock | 1 + crates/client-api/Cargo.toml | 5 +- crates/client-api/src/routes/internal.rs | 10 +- .../src/routes/internal/task_dump.rs | 255 ++++++++++++++++++ crates/client-api/src/routes/mod.rs | 22 +- crates/standalone/src/subcommands/start.rs | 14 +- 6 files changed, 296 insertions(+), 11 deletions(-) create mode 100644 crates/client-api/src/routes/internal/task_dump.rs diff --git a/Cargo.lock b/Cargo.lock index 2fca17a44f6..d1f8130fd09 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9582,6 +9582,7 @@ version = "1.48.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff360e02eab121e0bc37a2d3b4d4dc622e6eda3a8e5253d5435ecf5bd4c68408" dependencies = [ + "backtrace", "bytes", "libc", "mio 1.1.0", diff --git a/crates/client-api/Cargo.toml b/crates/client-api/Cargo.toml index afbb5a2a340..09fcb563300 100644 --- a/crates/client-api/Cargo.toml +++ b/crates/client-api/Cargo.toml @@ -17,7 +17,7 @@ spacetimedb-schema.workspace = true base64.workspace = true http-body-util.workspace = true -tokio = { version = "1.2", features = ["full"] } +tokio = { workspace = true, features = ["full"] } lazy_static = "1.4.0" log = "0.4.4" serde = "1.0.136" @@ -60,6 +60,9 @@ thiserror.workspace = true [target.'cfg(not(target_env = "msvc"))'.dependencies] jemalloc_pprof.workspace = true +[target.'cfg(all(target_os = "linux", any(target_arch = "aarch64", target_arch = "x86", target_arch = "x86_64")))'.dependencies] +tokio = { workspace = true, features = ["taskdump"] } + [dev-dependencies] tower = "0.5" jsonwebtoken.workspace = true diff --git a/crates/client-api/src/routes/internal.rs b/crates/client-api/src/routes/internal.rs index b9faf10d391..5c6087968fb 100644 --- a/crates/client-api/src/routes/internal.rs +++ b/crates/client-api/src/routes/internal.rs @@ -1,5 +1,9 @@ use crate::NodeDelegate; +mod task_dump; + +pub use task_dump::TaskDumpRegistry; + #[cfg(not(target_env = "msvc"))] mod jemalloc_profiling { use axum::body::Body; @@ -154,9 +158,11 @@ mod jemalloc_profiling { } // The internal router is for things that are not meant to be exposed to the public API. -pub fn router() -> axum::Router +pub fn router(task_dumps: TaskDumpRegistry) -> axum::Router where S: NodeDelegate + Clone + 'static, { - axum::Router::new().nest("/heap", jemalloc_profiling::jemalloc_router()) + axum::Router::new() + .nest("/heap", jemalloc_profiling::jemalloc_router()) + .nest("/task-dump", task_dump::router(task_dumps)) } diff --git a/crates/client-api/src/routes/internal/task_dump.rs b/crates/client-api/src/routes/internal/task_dump.rs new file mode 100644 index 00000000000..f61dd7af94d --- /dev/null +++ b/crates/client-api/src/routes/internal/task_dump.rs @@ -0,0 +1,255 @@ +use axum::{extract::Extension, routing::get}; + +#[cfg(all( + tokio_unstable, + target_os = "linux", + any(target_arch = "aarch64", target_arch = "x86", target_arch = "x86_64") +))] +mod imp { + use std::{collections::BTreeMap, fmt::Write as _, sync::Arc, time::Duration}; + + use axum::{ + extract::{Extension, Query}, + response::Response, + }; + use http::{header::CONTENT_TYPE, StatusCode}; + use serde::Deserialize; + use tokio::{runtime::Handle, sync::Mutex}; + + const MAX_TIMEOUT_MS: u64 = 30_000; + + #[derive(Clone)] + struct Runtime { + handle: Handle, + dump_lock: Arc>, + } + + /// The Tokio runtimes which can be inspected by the internal task dump endpoint. + #[derive(Clone, Default)] + pub struct TaskDumpRegistry { + runtimes: Arc>, + } + + impl TaskDumpRegistry { + pub fn new(runtimes: impl IntoIterator) -> Self { + let runtimes = runtimes + .into_iter() + .map(|(name, handle)| { + ( + name, + Runtime { + handle, + dump_lock: Arc::new(Mutex::new(())), + }, + ) + }) + .collect(); + Self { + runtimes: Arc::new(runtimes), + } + } + + fn get(&self, name: &str) -> Option<&Runtime> { + self.runtimes.get(name) + } + + fn names(&self) -> impl Iterator + '_ { + self.runtimes.keys().copied() + } + } + + #[derive(Deserialize)] + pub(super) struct TaskDumpQuery { + runtime: String, + timeout_ms: u64, + } + + impl TaskDumpQuery { + fn validate(&self) -> Result<(), (StatusCode, String)> { + if !(1..=MAX_TIMEOUT_MS).contains(&self.timeout_ms) { + return Err(( + StatusCode::BAD_REQUEST, + format!("timeout_ms must be between 1 and {MAX_TIMEOUT_MS}"), + )); + } + Ok(()) + } + } + + pub(super) async fn handle_get_task_dump( + Extension(registry): Extension, + Query(query): Query, + ) -> Result { + query.validate()?; + + let Some(runtime) = registry.get(&query.runtime).cloned() else { + let valid = registry.names().collect::>().join(", "); + return Err(( + StatusCode::BAD_REQUEST, + format!("unknown Tokio runtime {:?}; valid runtimes: {valid}", query.runtime), + )); + }; + + let permit = runtime.dump_lock.try_lock_owned().map_err(|_| { + ( + StatusCode::CONFLICT, + format!("a task dump is already in progress for runtime {:?}", query.runtime), + ) + })?; + + // Keep the permit in the spawned task so that a timed-out dump continues + // to exclude new requests until Tokio's dump future actually finishes. + let dump_task = tokio::spawn(async move { + let dump = runtime.handle.dump().await; + (permit, dump) + }); + let (_permit, dump) = tokio::time::timeout(Duration::from_millis(query.timeout_ms), dump_task) + .await + .map_err(|_| { + ( + StatusCode::GATEWAY_TIMEOUT, + format!( + "timed out after {}ms while dumping Tokio runtime {:?}", + query.timeout_ms, query.runtime + ), + ) + })? + .map_err(|err| { + ( + StatusCode::INTERNAL_SERVER_ERROR, + format!("task dump worker failed for runtime {:?}: {err}", query.runtime), + ) + })?; + + let runtime_name = query.runtime; + let body = tokio::task::spawn_blocking(move || format_dump(&runtime_name, dump)) + .await + .map_err(|err| { + ( + StatusCode::INTERNAL_SERVER_ERROR, + format!("task dump formatting failed: {err}"), + ) + })?; + + Response::builder() + .header(CONTENT_TYPE, "text/plain; charset=utf-8") + .body(body.into()) + .map_err(|err| (StatusCode::INTERNAL_SERVER_ERROR, err.to_string())) + } + + fn format_dump(runtime_name: &str, dump: tokio::runtime::Dump) -> String { + let tasks = dump.tasks(); + let mut output = String::new(); + writeln!(output, "runtime: {runtime_name}").unwrap(); + writeln!(output, "tasks: {}", tasks.iter().count()).unwrap(); + + for task in tasks.iter() { + writeln!(output, "\nTASK {}:", task.id()).unwrap(); + writeln!(output, "{}", task.trace()).unwrap(); + } + + output + } + + #[cfg(test)] + mod tests { + use super::*; + use http_body_util::BodyExt as _; + + #[tokio::test] + async fn registry_names_are_sorted() { + let handle = Handle::current(); + let registry = TaskDumpRegistry::new([("replication", handle.clone()), ("main", handle)]); + + assert_eq!(registry.names().collect::>(), ["main", "replication"]); + } + + #[test] + fn timeout_must_be_in_range() { + for timeout_ms in [1, MAX_TIMEOUT_MS] { + assert!(TaskDumpQuery { + runtime: "main".into(), + timeout_ms, + } + .validate() + .is_ok()); + } + + for timeout_ms in [0, MAX_TIMEOUT_MS + 1] { + assert!(TaskDumpQuery { + runtime: "main".into(), + timeout_ms, + } + .validate() + .is_err()); + } + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn dumps_registered_runtime() { + let registry = TaskDumpRegistry::new([("main", Handle::current())]); + let response = handle_get_task_dump( + Extension(registry), + Query(TaskDumpQuery { + runtime: "main".into(), + timeout_ms: 10_000, + }), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + let body = response.into_body().collect().await.unwrap().to_bytes(); + let body = std::str::from_utf8(&body).unwrap(); + assert!(body.starts_with("runtime: main\ntasks: ")); + } + } +} + +#[cfg(not(all( + tokio_unstable, + target_os = "linux", + any(target_arch = "aarch64", target_arch = "x86", target_arch = "x86_64") +)))] +mod imp { + use axum::response::{IntoResponse as _, Response}; + use http::StatusCode; + use tokio::runtime::Handle; + + /// The Tokio runtimes which can be inspected by the internal task dump endpoint. + #[derive(Clone, Default)] + pub struct TaskDumpRegistry; + + impl TaskDumpRegistry { + pub fn new(_: impl IntoIterator) -> Self { + Self + } + } + + pub(super) async fn handle_get_task_dump() -> Response { + ( + StatusCode::NOT_IMPLEMENTED, + "Tokio task dumps require a Linux aarch64, x86, or x86_64 build with tokio_unstable enabled", + ) + .into_response() + } + + #[cfg(test)] + mod tests { + use super::*; + + #[tokio::test] + async fn reports_unsupported_platform() { + assert_eq!(handle_get_task_dump().await.status(), StatusCode::NOT_IMPLEMENTED); + } + } +} + +use imp::handle_get_task_dump; +pub use imp::TaskDumpRegistry; + +pub fn router(registry: TaskDumpRegistry) -> axum::Router { + axum::Router::new() + .route("/", get(handle_get_task_dump)) + .layer(Extension(registry)) +} diff --git a/crates/client-api/src/routes/mod.rs b/crates/client-api/src/routes/mod.rs index 0a649db75b4..14c9e3db6bc 100644 --- a/crates/client-api/src/routes/mod.rs +++ b/crates/client-api/src/routes/mod.rs @@ -13,6 +13,7 @@ pub mod metrics; pub mod prometheus; pub mod subscribe; +pub use self::internal::TaskDumpRegistry; use self::{database::DatabaseRoutes, identity::IdentityRoutes}; /// This API call is just designed to allow clients to determine whether or not they can @@ -26,6 +27,25 @@ pub fn router( identity_routes: IdentityRoutes, extra: axum::Router, ) -> axum::Router +where + S: NodeDelegate + ControlStateDelegate + Authorization + Clone + 'static, +{ + router_with_task_dumps( + ctx, + database_routes, + identity_routes, + extra, + TaskDumpRegistry::default(), + ) +} + +pub fn router_with_task_dumps( + ctx: &S, + database_routes: DatabaseRoutes, + identity_routes: IdentityRoutes, + extra: axum::Router, + task_dumps: TaskDumpRegistry, +) -> axum::Router where S: NodeDelegate + ControlStateDelegate + Authorization + Clone + 'static, { @@ -46,5 +66,5 @@ where axum::Router::new() .nest("/v1", router.layer(cors)) - .nest("/internal", internal::router()) + .nest("/internal", internal::router(task_dumps)) } diff --git a/crates/standalone/src/subcommands/start.rs b/crates/standalone/src/subcommands/start.rs index 2cc1ede03c3..3386de52820 100644 --- a/crates/standalone/src/subcommands/start.rs +++ b/crates/standalone/src/subcommands/start.rs @@ -17,8 +17,9 @@ use spacetimedb::startup::{self, TracingOptions}; use spacetimedb::util::jobs::JobCores; use spacetimedb::worker_metrics; use spacetimedb_client_api::routes::database::DatabaseRoutes; -use spacetimedb_client_api::routes::router; +use spacetimedb_client_api::routes::router_with_task_dumps; use spacetimedb_client_api::routes::subscribe::WebSocketOptions; +use spacetimedb_client_api::routes::TaskDumpRegistry; use spacetimedb_paths::cli::{PrivKeyPath, PubKeyPath}; use spacetimedb_paths::server::{ConfigToml, ServerDataDir}; use tokio::net::TcpListener; @@ -197,11 +198,8 @@ pub async fn exec(args: &ArgMatches, db_cores: JobCores) -> anyhow::Result<()> { ) .await?; worker_metrics::spawn_jemalloc_stats(listen_addr.clone()); - worker_metrics::spawn_tokio_stats( - listen_addr.clone(), - "main".to_string(), - tokio::runtime::Handle::current(), - ); + let main_rt = tokio::runtime::Handle::current(); + worker_metrics::spawn_tokio_stats(listen_addr.clone(), "main".to_string(), main_rt.clone()); worker_metrics::spawn_page_pool_stats(listen_addr.clone(), ctx.page_pool().clone()); worker_metrics::spawn_bsatn_rlb_pool_stats(listen_addr.clone(), ctx.bsatn_rlb_pool().clone()); let mut db_routes = DatabaseRoutes::default(); @@ -209,7 +207,9 @@ pub async fn exec(args: &ArgMatches, db_cores: JobCores) -> anyhow::Result<()> { db_routes.db_put = db_routes.db_put.layer(DefaultBodyLimit::disable()); db_routes.pre_publish = db_routes.pre_publish.layer(DefaultBodyLimit::disable()); let extra = axum::Router::new().nest("/health", spacetimedb_client_api::routes::health::router()); - let service = router(&ctx, db_routes, IdentityRoutes::default(), extra).with_state(ctx.clone()); + let task_dumps = TaskDumpRegistry::new([("main", main_rt)]); + let service = + router_with_task_dumps(&ctx, db_routes, IdentityRoutes::default(), extra, task_dumps).with_state(ctx.clone()); // Check if the requested port is available on both IPv4 and IPv6. // If not, offer to find an available port by incrementing (unless non-interactive). From 416804b153b07944b3956898293dc46fbe50b9c2 Mon Sep 17 00:00:00 2001 From: joshua-spacetime Date: Mon, 27 Jul 2026 10:57:49 -0700 Subject: [PATCH 2/3] simplify --- crates/client-api/src/routes/internal.rs | 4 +- .../src/routes/internal/task_dump.rs | 133 ++++-------------- crates/client-api/src/routes/mod.rs | 21 +-- crates/standalone/src/subcommands/start.rs | 9 +- 4 files changed, 34 insertions(+), 133 deletions(-) diff --git a/crates/client-api/src/routes/internal.rs b/crates/client-api/src/routes/internal.rs index 5c6087968fb..79bae5ce5d4 100644 --- a/crates/client-api/src/routes/internal.rs +++ b/crates/client-api/src/routes/internal.rs @@ -158,11 +158,11 @@ mod jemalloc_profiling { } // The internal router is for things that are not meant to be exposed to the public API. -pub fn router(task_dumps: TaskDumpRegistry) -> axum::Router +pub fn router() -> axum::Router where S: NodeDelegate + Clone + 'static, { axum::Router::new() .nest("/heap", jemalloc_profiling::jemalloc_router()) - .nest("/task-dump", task_dump::router(task_dumps)) + .nest("/task-dump", task_dump::router()) } diff --git a/crates/client-api/src/routes/internal/task_dump.rs b/crates/client-api/src/routes/internal/task_dump.rs index f61dd7af94d..bd616658183 100644 --- a/crates/client-api/src/routes/internal/task_dump.rs +++ b/crates/client-api/src/routes/internal/task_dump.rs @@ -1,4 +1,4 @@ -use axum::{extract::Extension, routing::get}; +use axum::routing::get; #[cfg(all( tokio_unstable, @@ -6,7 +6,7 @@ use axum::{extract::Extension, routing::get}; any(target_arch = "aarch64", target_arch = "x86", target_arch = "x86_64") ))] mod imp { - use std::{collections::BTreeMap, fmt::Write as _, sync::Arc, time::Duration}; + use std::{collections::BTreeMap, fmt::Write as _, num::NonZeroU64, sync::Arc, time::Duration}; use axum::{ extract::{Extension, Query}, @@ -14,111 +14,60 @@ mod imp { }; use http::{header::CONTENT_TYPE, StatusCode}; use serde::Deserialize; - use tokio::{runtime::Handle, sync::Mutex}; - - const MAX_TIMEOUT_MS: u64 = 30_000; + use tokio::runtime::Handle; - #[derive(Clone)] - struct Runtime { - handle: Handle, - dump_lock: Arc>, - } + const DEFAULT_TIMEOUT_MS: u64 = 2_000; /// The Tokio runtimes which can be inspected by the internal task dump endpoint. #[derive(Clone, Default)] pub struct TaskDumpRegistry { - runtimes: Arc>, + runtimes: Arc>, } impl TaskDumpRegistry { pub fn new(runtimes: impl IntoIterator) -> Self { - let runtimes = runtimes - .into_iter() - .map(|(name, handle)| { - ( - name, - Runtime { - handle, - dump_lock: Arc::new(Mutex::new(())), - }, - ) - }) - .collect(); Self { - runtimes: Arc::new(runtimes), + runtimes: Arc::new(runtimes.into_iter().collect()), } } - - fn get(&self, name: &str) -> Option<&Runtime> { - self.runtimes.get(name) - } - - fn names(&self) -> impl Iterator + '_ { - self.runtimes.keys().copied() - } } #[derive(Deserialize)] pub(super) struct TaskDumpQuery { runtime: String, - timeout_ms: u64, - } - - impl TaskDumpQuery { - fn validate(&self) -> Result<(), (StatusCode, String)> { - if !(1..=MAX_TIMEOUT_MS).contains(&self.timeout_ms) { - return Err(( - StatusCode::BAD_REQUEST, - format!("timeout_ms must be between 1 and {MAX_TIMEOUT_MS}"), - )); - } - Ok(()) - } + timeout_ms: Option, } pub(super) async fn handle_get_task_dump( - Extension(registry): Extension, + registry: Option>, Query(query): Query, ) -> Result { - query.validate()?; + let Some(Extension(registry)) = registry else { + return Err(( + StatusCode::NOT_IMPLEMENTED, + "Tokio task dumps are not configured for this server".into(), + )); + }; - let Some(runtime) = registry.get(&query.runtime).cloned() else { - let valid = registry.names().collect::>().join(", "); + let Some(runtime) = registry.runtimes.get(query.runtime.as_str()).cloned() else { + let valid = registry.runtimes.keys().copied().collect::>().join(", "); return Err(( StatusCode::BAD_REQUEST, format!("unknown Tokio runtime {:?}; valid runtimes: {valid}", query.runtime), )); }; - let permit = runtime.dump_lock.try_lock_owned().map_err(|_| { - ( - StatusCode::CONFLICT, - format!("a task dump is already in progress for runtime {:?}", query.runtime), - ) - })?; - - // Keep the permit in the spawned task so that a timed-out dump continues - // to exclude new requests until Tokio's dump future actually finishes. - let dump_task = tokio::spawn(async move { - let dump = runtime.handle.dump().await; - (permit, dump) - }); - let (_permit, dump) = tokio::time::timeout(Duration::from_millis(query.timeout_ms), dump_task) + let timeout_ms = query.timeout_ms.map_or(DEFAULT_TIMEOUT_MS, NonZeroU64::get); + let dump = tokio::time::timeout(Duration::from_millis(timeout_ms), runtime.dump()) .await .map_err(|_| { ( StatusCode::GATEWAY_TIMEOUT, format!( - "timed out after {}ms while dumping Tokio runtime {:?}", - query.timeout_ms, query.runtime + "timed out after {timeout_ms}ms while dumping Tokio runtime {:?}", + query.runtime ), ) - })? - .map_err(|err| { - ( - StatusCode::INTERNAL_SERVER_ERROR, - format!("task dump worker failed for runtime {:?}: {err}", query.runtime), - ) })?; let runtime_name = query.runtime; @@ -155,44 +104,16 @@ mod imp { mod tests { use super::*; use http_body_util::BodyExt as _; - - #[tokio::test] - async fn registry_names_are_sorted() { - let handle = Handle::current(); - let registry = TaskDumpRegistry::new([("replication", handle.clone()), ("main", handle)]); - - assert_eq!(registry.names().collect::>(), ["main", "replication"]); - } - - #[test] - fn timeout_must_be_in_range() { - for timeout_ms in [1, MAX_TIMEOUT_MS] { - assert!(TaskDumpQuery { - runtime: "main".into(), - timeout_ms, - } - .validate() - .is_ok()); - } - - for timeout_ms in [0, MAX_TIMEOUT_MS + 1] { - assert!(TaskDumpQuery { - runtime: "main".into(), - timeout_ms, - } - .validate() - .is_err()); - } - } + use tokio::runtime::Handle; #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn dumps_registered_runtime() { + async fn dumps_registered_runtime_with_default_timeout() { let registry = TaskDumpRegistry::new([("main", Handle::current())]); let response = handle_get_task_dump( - Extension(registry), + Some(Extension(registry)), Query(TaskDumpQuery { runtime: "main".into(), - timeout_ms: 10_000, + timeout_ms: None, }), ) .await @@ -248,8 +169,6 @@ mod imp { use imp::handle_get_task_dump; pub use imp::TaskDumpRegistry; -pub fn router(registry: TaskDumpRegistry) -> axum::Router { - axum::Router::new() - .route("/", get(handle_get_task_dump)) - .layer(Extension(registry)) +pub fn router() -> axum::Router { + axum::Router::new().route("/", get(handle_get_task_dump)) } diff --git a/crates/client-api/src/routes/mod.rs b/crates/client-api/src/routes/mod.rs index 14c9e3db6bc..c7bbdeeaf2e 100644 --- a/crates/client-api/src/routes/mod.rs +++ b/crates/client-api/src/routes/mod.rs @@ -27,25 +27,6 @@ pub fn router( identity_routes: IdentityRoutes, extra: axum::Router, ) -> axum::Router -where - S: NodeDelegate + ControlStateDelegate + Authorization + Clone + 'static, -{ - router_with_task_dumps( - ctx, - database_routes, - identity_routes, - extra, - TaskDumpRegistry::default(), - ) -} - -pub fn router_with_task_dumps( - ctx: &S, - database_routes: DatabaseRoutes, - identity_routes: IdentityRoutes, - extra: axum::Router, - task_dumps: TaskDumpRegistry, -) -> axum::Router where S: NodeDelegate + ControlStateDelegate + Authorization + Clone + 'static, { @@ -66,5 +47,5 @@ where axum::Router::new() .nest("/v1", router.layer(cors)) - .nest("/internal", internal::router(task_dumps)) + .nest("/internal", internal::router()) } diff --git a/crates/standalone/src/subcommands/start.rs b/crates/standalone/src/subcommands/start.rs index 3386de52820..8961d55bc42 100644 --- a/crates/standalone/src/subcommands/start.rs +++ b/crates/standalone/src/subcommands/start.rs @@ -7,7 +7,7 @@ use std::sync::Arc; use crate::{StandaloneEnv, StandaloneOptions}; use anyhow::Context; -use axum::extract::DefaultBodyLimit; +use axum::extract::{DefaultBodyLimit, Extension}; use clap::ArgAction::SetTrue; use clap::{Arg, ArgMatches}; use spacetimedb::config::{parse_config, CertificateAuthority}; @@ -17,7 +17,7 @@ use spacetimedb::startup::{self, TracingOptions}; use spacetimedb::util::jobs::JobCores; use spacetimedb::worker_metrics; use spacetimedb_client_api::routes::database::DatabaseRoutes; -use spacetimedb_client_api::routes::router_with_task_dumps; +use spacetimedb_client_api::routes::router; use spacetimedb_client_api::routes::subscribe::WebSocketOptions; use spacetimedb_client_api::routes::TaskDumpRegistry; use spacetimedb_paths::cli::{PrivKeyPath, PubKeyPath}; @@ -208,8 +208,9 @@ pub async fn exec(args: &ArgMatches, db_cores: JobCores) -> anyhow::Result<()> { db_routes.pre_publish = db_routes.pre_publish.layer(DefaultBodyLimit::disable()); let extra = axum::Router::new().nest("/health", spacetimedb_client_api::routes::health::router()); let task_dumps = TaskDumpRegistry::new([("main", main_rt)]); - let service = - router_with_task_dumps(&ctx, db_routes, IdentityRoutes::default(), extra, task_dumps).with_state(ctx.clone()); + let service = router(&ctx, db_routes, IdentityRoutes::default(), extra) + .layer(Extension(task_dumps)) + .with_state(ctx.clone()); // Check if the requested port is available on both IPv4 and IPv6. // If not, offer to find an available port by incrementing (unless non-interactive). From e65bca1c57e171cb96cfc56c4b5ee8566c27beaf Mon Sep 17 00:00:00 2001 From: joshua-spacetime Date: Mon, 27 Jul 2026 14:30:15 -0700 Subject: [PATCH 3/3] empty