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

@@ -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<Activity, AppError> {
+    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<InjectV1Request>,
 ) -> Result<Json<InjectV1Response>, 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?
 }

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<AppState>,
     body: Bytes,
 ) -> Result<Json<XferResponseV1>, 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)?;
 

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.