From 6193331a4ac8e4a1cf35303e7c472147da71febd Mon Sep 17 00:00:00 2001 From: Wez Furlong Date: Wed, 11 Feb 2026 06:17:17 +0000 Subject: [PATCH] http/xfer injection: grab Activity handle for duration of request I noticed that we weren't grabbing the Activity handle for injection requests. Doing so allows us to reject a request that comes in while we are shutting down the service, rather than accept it to have it potentially dropped as we shut down. I also realized that we need to make the xfer injection handler match the same set of rules for the http injection handler, so this commit refactors that logic to reuse it in both places. --- crates/kumod/src/http_server/inject_v1.rs | 63 ++++++++++++++++------- crates/kumod/src/xfer/mod.rs | 15 +----- docs/changelog/main.md | 4 ++ 3 files changed, 49 insertions(+), 33 deletions(-) diff --git a/crates/kumod/src/http_server/inject_v1.rs b/crates/kumod/src/http_server/inject_v1.rs index 952ab566..405931fb 100644 --- a/crates/kumod/src/http_server/inject_v1.rs +++ b/crates/kumod/src/http_server/inject_v1.rs @@ -15,6 +15,7 @@ use kumo_log_types::ResolvedAddress; use kumo_prometheus::AtomicCounter; use kumo_server_common::authn_authz::AuthInfo; use kumo_server_common::http_server::{AppError, AppState}; +use kumo_server_lifecycle::Activity; use kumo_server_runtime::{Runtime, RUNTIME}; use kumo_template::{CompiledTemplates, TemplateDialect, TemplateEngine, TemplateList}; use mailparsing::{AddrSpec, Address, EncodeHeaderValue, Mailbox, MessageBuilder, MimePart}; @@ -1075,6 +1076,43 @@ pub fn register(lua: &Lua) -> anyhow::Result<()> { Ok(()) } +/// Grab an Activity handle for an HTTP injection task. +/// It will generate 503 errors if the service hasn't fully started, +/// is shutting down, memory is low, or the disk is too full. +pub fn activity_for_peer( + label: &str, + peer_address: impl std::fmt::Debug, +) -> Result { + let Some(activity) = Activity::get_opt(format!("{label} for {peer_address:?}")) else { + return Err(AppError::new( + StatusCode::SERVICE_UNAVAILABLE, + "shutting down", + )); + }; + + if kumo_server_memory::get_headroom() == 0 { + // Using too much memory + return Err(AppError::new( + StatusCode::SERVICE_UNAVAILABLE, + "load shedding", + )); + } + if kumo_server_common::disk_space::is_over_limit() { + return Err(AppError::new( + StatusCode::SERVICE_UNAVAILABLE, + "disk is too full", + )); + } + if !SpoolManager::get().spool_started() { + return Err(AppError::new( + StatusCode::SERVICE_UNAVAILABLE, + "waiting for spool startup", + )); + } + + Ok(activity) +} + /// Inject a message using a given message body, with template expansion, /// to a list of recipients. /// Both message assembly and templating are supported, and multiple recipients @@ -1171,25 +1209,7 @@ pub async fn inject_v1( // Note: Json<> must be last in the param list Json(request): Json, ) -> Result, AppError> { - if kumo_server_memory::get_headroom() == 0 { - // Using too much memory - return Err(AppError::new( - StatusCode::SERVICE_UNAVAILABLE, - "load shedding", - )); - } - if kumo_server_common::disk_space::is_over_limit() { - return Err(AppError::new( - StatusCode::SERVICE_UNAVAILABLE, - "disk is too full", - )); - } - if !SpoolManager::get().spool_started() { - return Err(AppError::new( - StatusCode::SERVICE_UNAVAILABLE, - "waiting for spool startup", - )); - } + let activity = activity_for_peer("inject_v1", peer_address)?; let limit = LIMIT.load(); if let Some(limit) = limit.as_ref() { @@ -1222,7 +1242,10 @@ pub async fn inject_v1( let hostname = Some(app_state.params().hostname.to_string()); pool.spawn(format!("http inject_v1 for {peer_address:?}"), async move { - inject_v1_impl(auth, sender, peer_address, request, via_address, hostname).await + let result = + inject_v1_impl(auth, sender, peer_address, request, via_address, hostname).await; + drop(activity); + result })? .await? } diff --git a/crates/kumod/src/xfer/mod.rs b/crates/kumod/src/xfer/mod.rs index ce0730c7..f1be663a 100644 --- a/crates/kumod/src/xfer/mod.rs +++ b/crates/kumod/src/xfer/mod.rs @@ -1,3 +1,4 @@ +use crate::http_server::inject_v1::activity_for_peer; use crate::logging::disposition::{log_disposition, LogDisposition}; use crate::queue::{DeliveryProto, QueueConfig, QueueManager}; use crate::ready_queue::{Dispatcher, QueueDispatcher}; @@ -284,19 +285,7 @@ pub async fn inject_xfer_v1( State(app_state): State, body: Bytes, ) -> Result, AppError> { - if kumo_server_memory::get_headroom() == 0 { - // Using too much memory - return Err(AppError::new( - StatusCode::SERVICE_UNAVAILABLE, - "load shedding", - )); - } - if kumo_server_common::disk_space::is_over_limit() { - return Err(AppError::new( - StatusCode::SERVICE_UNAVAILABLE, - "disk is too full", - )); - } + let _activity = activity_for_peer("inject_xfer_v1", peer_address)?; let msg = Message::deserialize_from_xfer(&body)?; diff --git a/docs/changelog/main.md b/docs/changelog/main.md index 2e94e426..4053748d 100644 --- a/docs/changelog/main.md +++ b/docs/changelog/main.md @@ -82,3 +82,7 @@ there was a race condition on startup where an injection request could begin processing prior to starting spool enumeration, which could then cause a `set_meta_spool has not been called` panic. + * HTTP Injection and XFER Injections didn't grab an Activity handle which + meant that there was a potential race condition when shutting down the + system which could result in loss of accountability of the message(s) + that were part of that request.