Add new Delay log record type

The purpose of this record is to log additional context about why
a message might end up in the scheduled queue when it hasn't
logged a TransientFailure.

There are a few situations around handling throttles and limits
where we might put a message back into the scheduled queue, without also
logging a TransientFailure record. It's possible that we should
reconsider some of those, but for the moment, there is an observability
hole that needs to be filled.

What this commit does is introduce an `InsertContext` which can hold one
or more `InsertReason`s about why a message is being inserted into the
scheduled queue.

There are 3 primary reasons for insertion:

* Received - the message was just received/injected
* Enumerated - the message was discovered in spool enumeration
* DueTimeWasReached - the message is now due for delivery and is being
  popped off the scheduled queue

The additional reasons can be added to the context to provide more
color about what happened.

When a message is added to the scheduled queue, the accumulation
in the InsertContext is examined, and if the context doesn't
indicate that the message was Enumerated and it wasn't also
already logged as a TransientFailure, a `Delay` record is
logged.

The `Delay` record includes in its `response.content` the ordered set of
InsertReasons as well as the delay duration and due time.

Logging Delay records might place undesirable pressure on the
logging storage, so you may wish to disable it via:

```lua
kumo.configure_local_logs {
  per_record = {
    Delay = {
      -- Suppress Delay records
      enable = false
    }
  }
}
```

or similar.

Cargo.toml

@@ -161,6 +161,7 @@ serde_yaml = "0.9"
 sha-1 = { version = "0.10", features = ["oid"] }
 sha2 = "0.10"
 slog = "2.7"
+smallvec = "1.13"
 socksv5 = {version="0.3", default-features=false, features=["tokio"]}
 sqlite = "0.36"
 strum = { version = "0.26", features = ["derive"] }

crates/integration-tests/src/test/tsa_tenant_suspension_issue290.rs

@@ -92,6 +92,7 @@ DeliverySummary {
         Reception: 2,
         Delivery: 1,
         TransientFailure: 2,
+        Delayed: 1,
     },
     sink_counts: {
         Rejection: 2,

crates/kumo-log-types/src/lib.rs

@@ -60,6 +60,9 @@ pub enum RecordType {
     /// and into some other queue
     DeferredInjectionRebind,
 
+    /// Explains why a message was put into the scheduled queue
+    Delayed,
+
     /// Special for matching anything in the logging config
     Any,
 }

crates/kumod/Cargo.toml

@@ -54,6 +54,7 @@ rfc5321 = {path="../rfc5321"}
 rustls = {workspace=true}
 serde = {workspace=true}
 serde_json = {workspace=true}
+smallvec.workspace = true
 socksv5 = {workspace=true}
 spool = {path="../spool", features=["rocksdb"]}
 sqlite = {workspace=true}

crates/kumod/src/logging/hooks.rs

@@ -1,6 +1,6 @@
 use crate::logging::files::LogFileParams;
 use crate::logging::{LogCommand, LogRecordParams, LOGGING_RUNTIME};
-use crate::queue::QueueManager;
+use crate::queue::{InsertReason, QueueManager};
 use anyhow::Context;
 use config::{load_config, CallbackSignature};
 use flume::Receiver;
@@ -176,7 +176,7 @@ impl LogHookState {
                 if !deferred_spool {
                     msg.save().await?;
                 }
-                QueueManager::insert(&queue_name, msg).await?;
+                QueueManager::insert(&queue_name, msg, InsertReason::Received.into()).await?;
             }
             drop(permit);
             anyhow::Result::<()>::Ok(())

crates/kumod/src/lua_deliver.rs

@@ -1,6 +1,6 @@
 use crate::delivery_metrics::MetricsWrappedConnection;
 use crate::logging::disposition::{log_disposition, LogDisposition};
-use crate::queue::{IncrementAttempts, QueueManager};
+use crate::queue::{IncrementAttempts, InsertReason, QueueManager};
 use crate::ready_queue::{Dispatcher, QueueDispatcher};
 use crate::smtp_server::RejectError;
 use crate::spool::SpoolManager;
@@ -249,6 +249,7 @@ impl QueueDispatcher for LuaQueueDispatcher {
                                     IncrementAttempts::Yes,
                                     None,
                                     response.clone(),
+                                    InsertReason::LoggedTransientFailure.into(),
                                 ),
                             )?;
                             dispatcher.metrics.inc_transfail();