Merge remote-tracking branch 'origin/main' into chore/merge-to-release

# Conflicts:
#	CHANGELOG.md
#	Cargo.toml
#	VERSION

Author: github-actions[bot]

TODO.md

@@ -8,7 +8,17 @@
 
 ## Current Tasks
 
-### Gap Analysis P2 — HTTP Client, Database URLs, Cache `[NEXT]`
+### Kafka Transport Metrics Parity with gRPC `[NEXT]`
+
+gRPC transport (v1.19.7) auto-emits `dfe_transport_*` metrics. Kafka transport does not — two gaps:
+
+1. **Add metrics instrumentation to `KafkaTransport::send()`** — emit `dfe_transport_sent_total{transport="kafka"}`, `dfe_transport_send_errors_total{transport="kafka"}`, `dfe_transport_backpressured_total{transport="kafka"}`, and send duration histogram, matching gRPC parity.
+
+2. **Wire `StatsContext` into `KafkaTransport`** — currently `new_with_context()` is a stub that ignores the context. The struct uses `DefaultConsumerContext` / plain `FutureProducer`, so `statistics.interval.ms` callbacks (set to 1000ms by all profiles) go to a no-op. Need to either make the struct generic over context (complicates `Transport` trait impl) or use `StatsContext` by default when the `metrics` feature is enabled. This enables `rdkafka_broker_rtt_avg_seconds`, `rdkafka_global_msg_cnt`, `rdkafka_topic_partition_consumer_lag`, etc.
+
+Downstream impact: dfe-fetcher, dfe-receiver, dfe-loader all use `KafkaTransport` and would get these for free once wired.
+
+### Gap Analysis P2 — HTTP Client, Database URLs, Cache
 
 - [ ] HTTP client module with retry middleware (reqwest + reqwest-middleware + reqwest-retry)
   - Wrap reqwest with exponential backoff, configurable timeouts

src/cache/mod.rs

@@ -91,7 +91,17 @@ impl Cache {
     /// Returns `None` if not found or expired.
     pub async fn get<T: serde::de::DeserializeOwned>(&self, source: &str, key: &str) -> Option<T> {
         let full_key = format!("{source}:{key}");
-        let bytes = self.inner.get(&full_key).await?;
+        let bytes = self.inner.get(&full_key).await;
+
+        #[cfg(feature = "metrics")]
+        if bytes.is_some() {
+            metrics::counter!("dfe_cache_hits_total", "source" => source.to_string()).increment(1);
+        } else {
+            metrics::counter!("dfe_cache_misses_total", "source" => source.to_string())
+                .increment(1);
+        }
+
+        let bytes = bytes?;
         serde_json::from_slice(&bytes).ok()
     }
 
@@ -112,6 +122,9 @@ impl Cache {
 
         self.inner.insert(full_key.clone(), bytes).await;
 
+        #[cfg(feature = "metrics")]
+        metrics::gauge!("dfe_cache_entries").set(self.inner.entry_count() as f64);
+
         // Track key for source-level invalidation
         if let Ok(mut keys) = self.source_keys.lock() {
             keys.entry(source.to_string()).or_default().push(full_key);
@@ -130,6 +143,9 @@ impl Cache {
         for key in keys {
             self.inner.invalidate(&key).await;
         }
+
+        #[cfg(feature = "metrics")]
+        metrics::gauge!("dfe_cache_entries").set(self.inner.entry_count() as f64);
     }
 
     /// Invalidate a single entry.

src/dlq/file.rs

@@ -94,7 +94,17 @@ impl DlqBackend for FileDlq {
             .map_err(|e| DlqError::Serialization(format!("failed to serialise DLQ entry: {e}")))?;
         line.push(b'\n');
 
-        self.writer.write_line(&line)?;
+        if let Err(e) = self.writer.write_line(&line) {
+            #[cfg(feature = "metrics")]
+            metrics::counter!("dfe_dlq_write_errors_total").increment(1);
+            return Err(e.into());
+        }
+
+        #[cfg(feature = "metrics")]
+        {
+            metrics::counter!("dfe_dlq_entries_total").increment(1);
+            metrics::gauge!("dfe_dlq_entries_written").set(self.writer.lines_written() as f64);
+        }
 
         debug!(
             service = %self.service_name,

src/http_client/mod.rs

@@ -81,7 +81,20 @@ impl HttpClient {
 
     /// Send a GET request.
     pub async fn get(&self, url: &str) -> Result<Response, reqwest_middleware::Error> {
-        self.inner.get(url).send().await
+        #[cfg(feature = "metrics")]
+        let start = std::time::Instant::now();
+
+        let result = self.inner.get(url).send().await;
+
+        #[cfg(feature = "metrics")]
+        {
+            let status = if result.is_ok() { "success" } else { "error" };
+            metrics::counter!("dfe_http_client_requests_total", "method" => "GET", "status" => status).increment(1);
+            metrics::histogram!("dfe_http_client_duration_seconds", "method" => "GET")
+                .record(start.elapsed().as_secs_f64());
+        }
+
+        result
     }
 
     /// Send a POST request with a JSON body.
@@ -90,12 +103,26 @@ impl HttpClient {
         url: &str,
         body: &T,
     ) -> Result<Response, reqwest_middleware::Error> {
-        self.inner
+        #[cfg(feature = "metrics")]
+        let start = std::time::Instant::now();
+
+        let result = self
+            .inner
             .post(url)
             .header("content-type", "application/json")
             .body(serde_json::to_vec(body).unwrap_or_default())
             .send()
-            .await
+            .await;
+
+        #[cfg(feature = "metrics")]
+        {
+            let status = if result.is_ok() { "success" } else { "error" };
+            metrics::counter!("dfe_http_client_requests_total", "method" => "POST", "status" => status).increment(1);
+            metrics::histogram!("dfe_http_client_duration_seconds", "method" => "POST")
+                .record(start.elapsed().as_secs_f64());
+        }
+
+        result
     }
 
     /// Send a PUT request with a JSON body.
@@ -104,17 +131,44 @@ impl HttpClient {
         url: &str,
         body: &T,
     ) -> Result<Response, reqwest_middleware::Error> {
-        self.inner
+        #[cfg(feature = "metrics")]
+        let start = std::time::Instant::now();
+
+        let result = self
+            .inner
             .put(url)
             .header("content-type", "application/json")
             .body(serde_json::to_vec(body).unwrap_or_default())
             .send()
-            .await
+            .await;
+
+        #[cfg(feature = "metrics")]
+        {
+            let status = if result.is_ok() { "success" } else { "error" };
+            metrics::counter!("dfe_http_client_requests_total", "method" => "PUT", "status" => status).increment(1);
+            metrics::histogram!("dfe_http_client_duration_seconds", "method" => "PUT")
+                .record(start.elapsed().as_secs_f64());
+        }
+
+        result
     }
 
     /// Send a DELETE request.
     pub async fn delete(&self, url: &str) -> Result<Response, reqwest_middleware::Error> {
-        self.inner.delete(url).send().await
+        #[cfg(feature = "metrics")]
+        let start = std::time::Instant::now();
+
+        let result = self.inner.delete(url).send().await;
+
+        #[cfg(feature = "metrics")]
+        {
+            let status = if result.is_ok() { "success" } else { "error" };
+            metrics::counter!("dfe_http_client_requests_total", "method" => "DELETE", "status" => status).increment(1);
+            metrics::histogram!("dfe_http_client_duration_seconds", "method" => "DELETE")
+                .record(start.elapsed().as_secs_f64());
+        }
+
+        result
     }
 
     /// Access the underlying middleware client for custom requests.

src/secrets/mod.rs

@@ -158,12 +158,20 @@ impl SecretsManager {
         // Check cache first
         if let Some(cached) = self.cache.read().get(&cache_key) {
             debug!(path = %path, "Secret loaded from cache");
+            #[cfg(feature = "metrics")]
+            metrics::counter!("dfe_secrets_cache_hits_total").increment(1);
             return Ok(cached);
         }
 
+        #[cfg(feature = "metrics")]
+        metrics::counter!("dfe_secrets_cache_misses_total").increment(1);
+
         // Fetch from file
         let value = self.file_provider.get(path).await?;
 
+        #[cfg(feature = "metrics")]
+        metrics::counter!("dfe_secrets_fetch_total").increment(1);
+
         // Update cache
         if let Err(e) = self.cache.write().set(&cache_key, &value) {
             warn!(error = %e, "Failed to cache secret");
@@ -181,9 +189,14 @@ impl SecretsManager {
         // Check cache first
         if let Some(cached) = self.cache.read().get(cache_key) {
             debug!(key = %cache_key, "Secret loaded from cache");
+            #[cfg(feature = "metrics")]
+            metrics::counter!("dfe_secrets_cache_hits_total").increment(1);
             return Ok(cached);
         }
 
+        #[cfg(feature = "metrics")]
+        metrics::counter!("dfe_secrets_cache_misses_total").increment(1);
+
         // Fetch from provider
         let result = match source {
             SecretSource::File { path } => self.file_provider.get(path).await,
@@ -221,6 +234,9 @@ impl SecretsManager {
             }
         };
 
+        #[cfg(feature = "metrics")]
+        metrics::counter!("dfe_secrets_fetch_total").increment(1);
+
         match result {
             Ok(value) => {
                 // Update cache

src/spool/queue.rs

@@ -109,6 +109,8 @@ impl Spool {
             .map_err(|e| SpoolError::Queue(e.to_string()))?;
 
         self.len += 1;
+        #[cfg(feature = "metrics")]
+        ::metrics::gauge!("dfe_spool_queue_depth").set(self.len as f64);
         Ok(())
     }
 
@@ -181,6 +183,8 @@ impl Spool {
                     .commit()
                     .map_err(|e| SpoolError::Queue(e.to_string()))?;
                 self.len = self.len.saturating_sub(1);
+                #[cfg(feature = "metrics")]
+                ::metrics::gauge!("dfe_spool_queue_depth").set(self.len as f64);
                 Ok(Some(data))
             }
             Err(yaque::TryRecvError::Io(e)) => Err(SpoolError::Io(e)),
@@ -215,6 +219,8 @@ impl Spool {
             .commit()
             .map_err(|e| SpoolError::Queue(e.to_string()))?;
         self.len = self.len.saturating_sub(1);
+        #[cfg(feature = "metrics")]
+        ::metrics::gauge!("dfe_spool_queue_depth").set(self.len as f64);
         Ok(data)
     }
 
@@ -255,6 +261,8 @@ impl Spool {
             }
         }
         self.len = 0;
+        #[cfg(feature = "metrics")]
+        ::metrics::gauge!("dfe_spool_queue_depth").set(0.0);
         Ok(())
     }
 

src/tiered_sink/tiered.rs

@@ -146,6 +146,8 @@ impl<S: Sink> TieredSink<S> {
             match self.try_hot_path(data).await {
                 Ok(()) => {
                     self.hot_path_count.fetch_add(1, AtomicOrdering::Relaxed);
+                    #[cfg(feature = "metrics")]
+                    ::metrics::counter!("dfe_spool_hot_path_total").increment(1);
                     return Ok(());
                 }
                 Err(TieredSinkError::Sink(_)) => {
@@ -161,13 +163,22 @@ impl<S: Sink> TieredSink<S> {
         // Cold path: spool to disk
         self.spool_message(data).await?;
         self.cold_path_count.fetch_add(1, AtomicOrdering::Relaxed);
+        #[cfg(feature = "metrics")]
+        ::metrics::counter!("dfe_spool_cold_path_total").increment(1);
         Ok(())
     }
 
     /// Determine if we should attempt the hot path.
     async fn should_use_hot_path(&self) -> bool {
         let circuit_state = self.circuit.state().await;
 
+        #[cfg(feature = "metrics")]
+        ::metrics::gauge!("dfe_spool_circuit_state").set(match circuit_state {
+            CircuitState::Closed => 0.0,
+            CircuitState::HalfOpen => 1.0,
+            CircuitState::Open => 2.0,
+        });
+
         match circuit_state {
             CircuitState::Open => false,
             CircuitState::Closed | CircuitState::HalfOpen => match self.config.ordering {
@@ -195,6 +206,8 @@ impl<S: Sink> TieredSink<S> {
             }
             Ok(Err(SinkError::Unavailable)) => {
                 self.circuit.record_failure().await;
+                #[cfg(feature = "metrics")]
+                ::metrics::counter!("dfe_spool_circuit_trips_total").increment(1);
                 Err(TieredSinkError::Spool("sink unavailable".into()))
             }
             Ok(Err(SinkError::Fatal(e))) => {
@@ -203,6 +216,8 @@ impl<S: Sink> TieredSink<S> {
             }
             Err(_timeout) => {
                 self.circuit.record_failure().await;
+                #[cfg(feature = "metrics")]
+                ::metrics::counter!("dfe_spool_circuit_trips_total").increment(1);
                 Err(TieredSinkError::Spool("send timeout".into()))
             }
         }
@@ -250,6 +265,14 @@ impl<S: Sink> TieredSink<S> {
         self.spool_bytes
             .fetch_add(compressed_len, AtomicOrdering::Relaxed);
 
+        #[cfg(feature = "metrics")]
+        {
+            ::metrics::gauge!("dfe_spool_messages")
+                .set(self.spool_count.load(AtomicOrdering::Relaxed) as f64);
+            ::metrics::gauge!("dfe_spool_bytes")
+                .set(self.spool_bytes.load(AtomicOrdering::Relaxed) as f64);
+        }
+
         #[cfg(feature = "logger")]
         tracing::debug!(
             spool_items = self.spool_count.load(AtomicOrdering::Relaxed),
@@ -453,7 +476,15 @@ async fn disk_capacity_poller(
             () = tokio::time::sleep(poll_interval) => {}
         }
 
-        let available = check_disk_space(&spool_path).is_none_or(|(total, avail)| {
+        let disk_space = check_disk_space(&spool_path);
+
+        #[cfg(feature = "metrics")]
+        if let Some((total, avail)) = disk_space {
+            ::metrics::gauge!("dfe_spool_disk_available_bytes").set(avail as f64);
+            ::metrics::gauge!("dfe_spool_disk_total_bytes").set(total as f64);
+        }
+
+        let available = disk_space.is_none_or(|(total, avail)| {
             if total == 0 {
                 return true;
             }

src/transport/kafka/metrics.rs

@@ -360,9 +360,20 @@ impl StatsContext {
     }
 }
 
-// StatsContext can be used as a ConsumerContext
+// StatsContext can be used as a ConsumerContext and ProducerContext
 impl rdkafka::consumer::ConsumerContext for StatsContext {}
 
+impl rdkafka::producer::ProducerContext for StatsContext {
+    type DeliveryOpaque = ();
+
+    fn delivery(
+        &self,
+        _result: &rdkafka::producer::DeliveryResult<'_>,
+        _opaque: Self::DeliveryOpaque,
+    ) {
+    }
+}
+
 /// Calculate total consumer lag across all partitions.
 ///
 /// Helper function to sum lag from a `KafkaMetrics` snapshot.