From d7b3fb527e648c3ba9cadd580e7ee34e4fec89bc Mon Sep 17 00:00:00 2001
From: Ytallo Layon <ytallo.layon@gmail.com>
Date: Mon, 13 Jul 2026 21:24:26 -0300
Subject: [PATCH] fix(queue): make the effective store visible and surface
 delivery stats

The builtin adapter silently fell back to the in-memory store when the
stored configuration lacked an adapter entry, even if the --config seed
asked for file_based persistence. Boot now logs the effective store
(file_based + path, or in_memory) and warns when the seed's adapter is
discarded in favor of the authoritative stored value.

engine::queue::topic_stats now returns the delivered/failed counters
the store already tracks, so queue throughput is observable via the
API.
---
 queue/src/boot.rs      | 12 ++++++++++--
 queue/src/functions.rs |  8 +++++++-
 queue/src/main.rs      |  9 +++++++++
 3 files changed, 26 insertions(+), 3 deletions(-)

diff --git a/queue/src/boot.rs b/queue/src/boot.rs
index d88e68368..05b9210f0 100644
--- a/queue/src/boot.rs
+++ b/queue/src/boot.rs
@@ -174,9 +174,17 @@ pub async fn build_store(config: &QueueConfig) -> anyhow::Result<Arc<dyn QueueSt
                 .file_path
                 .unwrap_or_else(|| "queue_store_data".to_string());
             let save_interval_ms = builtin.save_interval_ms.unwrap_or(5000);
-            Ok(Arc::new(FileStore::open(path, save_interval_ms).await?))
+            let store = FileStore::open(&path, save_interval_ms).await?;
+            tracing::info!(store = "file_based", path = %path, "queue store ready");
+            Ok(Arc::new(store))
+        }
+        "builtin" | "in_memory" => {
+            tracing::info!(
+                store = "in_memory",
+                "queue store ready; jobs do not survive restarts"
+            );
+            Ok(Arc::new(InMemoryStore::new()))
         }
-        "builtin" | "in_memory" => Ok(Arc::new(InMemoryStore::new())),
         other => anyhow::bail!("unknown builtin queue store_method '{other}'"),
     }
 }
diff --git a/queue/src/functions.rs b/queue/src/functions.rs
index 25689e8b5..876904990 100644
--- a/queue/src/functions.rs
+++ b/queue/src/functions.rs
@@ -94,6 +94,8 @@ pub struct TopicStatsOutput {
     pub depth: u64,
     pub consumer_count: u64,
     pub dlq_depth: u64,
+    pub delivered: u64,
+    pub failed: u64,
     pub config: Option<Value>,
 }
 
@@ -327,6 +329,8 @@ pub async fn topic_stats(
         depth: stats.depth,
         consumer_count: 0,
         dlq_depth: stats.dlq_depth,
+        delivered: stats.delivered,
+        failed: stats.failed,
         config: None,
     })
 }
@@ -713,7 +717,7 @@ mod tests {
         *_mock.topic_stats_result.lock().unwrap() = Some(Ok(TopicStats {
             depth: 1,
             dlq_depth: 1,
-            delivered: 0,
+            delivered: 2,
             failed: 1,
         }));
         let stats = topic_stats(
@@ -726,6 +730,8 @@ mod tests {
         .unwrap();
         assert_eq!(stats.depth, 1);
         assert_eq!(stats.dlq_depth, 1);
+        assert_eq!(stats.delivered, 2);
+        assert_eq!(stats.failed, 1);
     }
 
     #[tokio::test]
diff --git a/queue/src/main.rs b/queue/src/main.rs
index 591263f8c..103f175ab 100644
--- a/queue/src/main.rs
+++ b/queue/src/main.rs
@@ -89,6 +89,15 @@ async fn main() -> Result<()> {
         .await
         .map_err(anyhow::Error::msg)
         .context("loading queue configuration")?;
+    if let Some(seed) = seed.as_ref() {
+        if seed.adapter != config.adapter {
+            tracing::warn!(
+                seed_adapter = ?seed.adapter,
+                stored_adapter = ?config.adapter,
+                "--config seed adapter ignored; the stored configuration is authoritative"
+            );
+        }
+    }
 
     let boot = iii_queue::boot::start(iii.clone(), config).await?;
     configuration::register_config_trigger(