Distributed Direct Publish Telemetry

docs/metrics.md

@@ -56,7 +56,7 @@ Metric labels are intentionally bounded:
 - `status`: bounded dispatch status such as `success`, `unknown_command`,
   `decode_failed`, `guard_rejected`, `repository_error`, or `other_error`.
 - `transport`: built-in source label such as `in_memory`, `sqlite`,
-  `postgres`, `rabbitmq`, `kafka`, `nats`, or `outbox`.
+  `postgres`, `rabbitmq`, `kafka`, `nats`, or `knative`.
 - `outcome`: bounded settle/publish outcome such as `ack`, `nack`,
   `dead_letter`, `park`, `ignored`, `log_and_ack`, `published`, `released`, or
   `failed`.
@@ -83,6 +83,12 @@ metrics should use that vocabulary rather than introducing ad hoc labels.
   counter.
 - `distributed_transport_failures_total{service,transport,failure_class,action}`
   counter.
+- `distributed_transport_publish_total{service,transport,message_kind,outcome}`
+  counter for direct bus producer calls. `outcome` is `published` or `failed`.
+- `distributed_transport_publish_duration_seconds{service,transport,message_kind,outcome}`
+  histogram for direct bus producer call duration.
+- `distributed_transport_publish_failures_total{service,transport,message_kind,failure_class}`
+  counter for failed direct bus producer calls.
 - `distributed_outbox_messages_total{service,outcome}` counter.
 - `distributed_outbox_pending_messages{service}` gauge.
 - `distributed_outbox_oldest_pending_age_seconds{service}` gauge.
@@ -100,10 +106,22 @@ does not install an OpenTelemetry metrics SDK, emit request-level HTTP metrics,
 or record user payload data.
 
 Direct transport receive paths emit receive/settle counters and failure
-counters. Outbox dispatch emits publish outcomes and backlog gauges. Direct
-`MessagePublisher` calls outside the outbox path do not emit publish metrics in
-this release; instrumenting those calls should use the same bounded vocabulary
-and should stay separate from outbox publish outcomes.
+counters. Direct `Bus::send`, `Bus::publish`, `send_message`, and
+`publish_message` calls on built-in buses emit `distributed_transport_publish_*`
+metrics. A direct publish success means the adapter's durable publish threshold
+resolved `Ok`: in-memory accepted the message, SQL committed the insert,
+RabbitMQ confirmed the publish, Kafka acknowledged per `acks`, NATS JetStream
+returned a publish ack, or Knative/HTTP returned a successful broker response.
+If that threshold returns `Err`, the publish outcome is `failed` and the
+failure counter uses the error's `TransportErrorKind` as `retryable` or
+`permanent`.
+
+Outbox-derived publishes are intentionally separate. `BusOutboxPublishHook`,
+`BusPublisher`, `DynBusPublisher`, and `OutboxDispatcher` record outbox
+`published`, `released`, `failed`, and backlog metrics, but they use the buses'
+raw outbox publish path so those rows do not increment direct producer metrics.
+This avoids double-counting; outbox metrics and the `distributed.outbox.publish`
+span remain the authoritative producer signal for outbox rows.
 
 ## Scaffolded GitOps
 

docs/observability.md

@@ -60,9 +60,10 @@ metadata already stored with events.
 ## Optional Span Feature
 
 The `otel` feature adds framework-owned `tracing` spans around dispatch,
-handler execution, transport receive, and outbox publish boundaries. When the
-incoming message carries W3C `traceparent` / `tracestate`, Distributed extracts
-that context and sets it as the OpenTelemetry parent for the framework span:
+handler execution, direct transport publish, transport receive, and outbox
+publish boundaries. When the incoming message carries W3C `traceparent` /
+`tracestate`, Distributed extracts that context and sets it as the
+OpenTelemetry parent for the framework span:
 
 ```toml
 [dependencies]
@@ -79,6 +80,7 @@ Framework span names are intentionally bounded:
 
 - `distributed.microsvc.dispatch`
 - `distributed.handler`
+- `distributed.transport.publish`
 - `distributed.transport.receive`
 - `distributed.outbox.publish`
 
@@ -88,6 +90,17 @@ Each span uses the same framework-owned message attributes:
 - `distributed.message.kind`
 - `messaging.message.id`
 
+Direct producer spans also include:
+
+- `distributed.transport.name`
+- `distributed.bus.operation`: `send` or `publish`
+- `distributed.producer.source`: `direct`
+
+Outbox rows keep using `distributed.outbox.publish` as their producer span and
+do not emit `distributed.transport.publish`, so traces match the metrics
+boundary: direct bus calls are direct producer telemetry; outbox dispatch is
+outbox producer telemetry.
+
 Future spans should use the same helper path so new attributes are reviewed in
 one place. Do not add payload fields, user ids, aggregate ids, or raw metadata
 values as framework span attributes.
@@ -151,10 +164,13 @@ CI verifies the observability surface at the docker level (see
   `GET /metrics`, and lints the exposition with `promtool check metrics`
   (skips when `PROMTOOL` is unset).
 - `tests/otel_export` builds a real OTLP pipeline the way a service binary
-  would, dispatches a message carrying a W3C `traceparent`, and asserts a real
-  OpenTelemetry Collector received `distributed.microsvc.dispatch` parented to
-  the incoming span (skips when `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` /
-  `OTEL_COLLECTOR_TRACES_FILE` are unset).
+  would, dispatches and directly publishes messages carrying W3C `traceparent`,
+  and asserts a real OpenTelemetry Collector received
+  `distributed.microsvc.dispatch` and `distributed.transport.publish` parented
+  to the incoming span. It also verifies active local spans remain the direct
+  parent for nested framework spans (skips when
+  `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` / `OTEL_COLLECTOR_TRACES_FILE` are
+  unset).
 - The scaffolded `ServiceMonitor` / `PrometheusRule` / OTLP env output is
   rendered with `helm template` and validated against published CRD schemas
   with `kubeconform`.

docs/transports.md

@@ -75,9 +75,18 @@ metadata (codec, destination, source aggregate) is namespaced under the reserved
 Telemetry follows the same boundary. Consumer-side direct transports report
 receive/settle outcomes and classified failures. Producer-side outbox dispatch
 reports `published`, `released`, `failed`, and backlog gauges. A direct
-`MessagePublisher` call outside the outbox path is not counted as an outbox
-publish and currently has no direct publish metric; adding that should be a
-separate producer telemetry path with its own bounded labels.
+`Bus::send`, `Bus::publish`, `send_message`, or `publish_message` call on a
+built-in bus reports `distributed_transport_publish_*` metrics and, with
+`otel`, a `distributed.transport.publish` span. Direct publish success means the
+same adapter durable publish threshold resolved `Ok`; a failed threshold records
+`outcome="failed"` and a `retryable` or `permanent` failure class from
+`TransportErrorKind`.
+
+Outbox-derived publishes use the buses' raw outbox path through
+`BusPublisher`/`DynBusPublisher`, so they do not increment direct publish
+metrics. That avoids double-counting and keeps `distributed_outbox_*` metrics
+plus `distributed.outbox.publish` as the authoritative producer signal for
+outbox rows.
 
 Trace context is normal metadata. Distributed preserves W3C `traceparent` and
 `tracestate` across `Message`, `EventRecord`, and `OutboxMessage` carriers

src/bus/bus.rs

@@ -59,6 +59,30 @@ pub trait Bus: Send + Sync {
         &self,
         message: Message,
     ) -> impl Future<Output = Result<(), TransportError>> + Send;
+
+    /// Send an outbox-derived command without direct producer telemetry.
+    ///
+    /// This default preserves downstream custom bus compatibility. Built-in
+    /// buses override it to bypass their direct producer instrumentation so the
+    /// outbox remains the authoritative producer signal for outbox rows.
+    fn send_outbox_message(
+        &self,
+        message: Message,
+    ) -> impl Future<Output = Result<(), TransportError>> + Send {
+        self.send_message(message)
+    }
+
+    /// Publish an outbox-derived event without direct producer telemetry.
+    ///
+    /// See [`send_outbox_message`](Self::send_outbox_message) for why this
+    /// exists as a provided compatibility method instead of a required trait
+    /// method.
+    fn publish_outbox_message(
+        &self,
+        message: Message,
+    ) -> impl Future<Output = Result<(), TransportError>> + Send {
+        self.publish_message(message)
+    }
 }
 
 /// Consume side of the bus — pull transports that run a [`run_source`] loop.

src/bus/in_memory_bus.rs

@@ -13,6 +13,7 @@
 use std::collections::{HashMap, VecDeque};
 use std::sync::{Arc, Mutex};
 
+use super::producer_telemetry::{record_direct_publish, BusOperation};
 use super::source::{MessageSource, ReceivedMessage};
 use super::Message;
 use super::{run_source, Bus, BusConsumer, MessageRouter, RunOptions, TransportError};
@@ -62,10 +63,32 @@ impl InMemoryBus {
 
 impl Bus for InMemoryBus {
     async fn send_message(&self, message: Message) -> Result<(), TransportError> {
-        self.enqueue(message)
+        record_direct_publish(
+            None,
+            "in_memory",
+            BusOperation::Send,
+            message,
+            |message| async { self.enqueue(message) },
+        )
+        .await
     }
 
     async fn publish_message(&self, message: Message) -> Result<(), TransportError> {
+        record_direct_publish(
+            None,
+            "in_memory",
+            BusOperation::Publish,
+            message,
+            |message| async { self.append(message) },
+        )
+        .await
+    }
+
+    async fn send_outbox_message(&self, message: Message) -> Result<(), TransportError> {
+        self.enqueue(message)
+    }
+
+    async fn publish_outbox_message(&self, message: Message) -> Result<(), TransportError> {
         self.append(message)
     }
 }

src/bus/kafka_bus.rs

@@ -24,6 +24,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use super::kafka::{KafkaPublisher, KafkaSource};
+use super::producer_telemetry::{record_direct_publish, BusOperation};
 use super::{
     run_source, Bus, BusConsumer, BusTopologyConfig, MessagePublisher, MessageRouter, RunOptions,
     TransportError,
@@ -185,13 +186,45 @@ impl KafkaBus {
 }
 
 impl Bus for KafkaBus {
-    async fn send_message(&self, mut message: Message) -> Result<(), TransportError> {
+    async fn send_message(&self, message: Message) -> Result<(), TransportError> {
+        record_direct_publish(
+            None,
+            "kafka",
+            BusOperation::Send,
+            message,
+            |message| async { self.send_message_raw(message).await },
+        )
+        .await
+    }
+
+    async fn publish_message(&self, message: Message) -> Result<(), TransportError> {
+        record_direct_publish(
+            None,
+            "kafka",
+            BusOperation::Publish,
+            message,
+            |message| async { self.publish_message_raw(message).await },
+        )
+        .await
+    }
+
+    async fn send_outbox_message(&self, message: Message) -> Result<(), TransportError> {
+        self.send_message_raw(message).await
+    }
+
+    async fn publish_outbox_message(&self, message: Message) -> Result<(), TransportError> {
+        self.publish_message_raw(message).await
+    }
+}
+
+impl KafkaBus {
+    async fn send_message_raw(&self, mut message: Message) -> Result<(), TransportError> {
         // The publisher uses the message name as the topic; namespace it.
         message.name = format!("{}{}", self.command_prefix()?, message.name);
         self.publisher.publish(message).await
     }
 
-    async fn publish_message(&self, mut message: Message) -> Result<(), TransportError> {
+    async fn publish_message_raw(&self, mut message: Message) -> Result<(), TransportError> {
         message.name = format!("{}{}", self.event_prefix()?, message.name);
         self.publisher.publish(message).await
     }

src/bus/knative_bus.rs

@@ -23,6 +23,7 @@ use std::time::Duration;
 use std::collections::HashSet;
 
 use super::knative::unique_k8s_name;
+use super::producer_telemetry::{record_direct_publish, BusOperation};
 use super::{Bus, TransportError};
 use super::{Message, SubscriptionPlan};
 
@@ -223,10 +224,42 @@ impl KnativeBus {
 
 impl Bus for KnativeBus {
     async fn send_message(&self, message: Message) -> Result<(), TransportError> {
-        self.post_cloud_event(&self.commands_broker, &message).await
+        record_direct_publish(
+            None,
+            "knative",
+            BusOperation::Send,
+            message,
+            |message| async { self.send_message_raw(message).await },
+        )
+        .await
     }
 
     async fn publish_message(&self, message: Message) -> Result<(), TransportError> {
+        record_direct_publish(
+            None,
+            "knative",
+            BusOperation::Publish,
+            message,
+            |message| async { self.publish_message_raw(message).await },
+        )
+        .await
+    }
+
+    async fn send_outbox_message(&self, message: Message) -> Result<(), TransportError> {
+        self.send_message_raw(message).await
+    }
+
+    async fn publish_outbox_message(&self, message: Message) -> Result<(), TransportError> {
+        self.publish_message_raw(message).await
+    }
+}
+
+impl KnativeBus {
+    async fn send_message_raw(&self, message: Message) -> Result<(), TransportError> {
+        self.post_cloud_event(&self.commands_broker, &message).await
+    }
+
+    async fn publish_message_raw(&self, message: Message) -> Result<(), TransportError> {
         self.post_cloud_event(&self.events_broker, &message).await
     }
 }

src/bus/mod.rs

@@ -103,6 +103,7 @@ mod nats;
 mod nats_bus;
 #[cfg(feature = "postgres")]
 mod postgres_bus;
+mod producer_telemetry;
 mod publisher;
 #[cfg(feature = "rabbitmq")]
 mod rabbitmq;

src/bus/nats_bus.rs

@@ -28,6 +28,7 @@ use async_nats::jetstream::consumer::pull::Config as PullConfig;
 use async_nats::jetstream::stream::{Config as StreamConfig, Stream};
 
 use super::nats::{NatsJetStreamSource, NatsPublisher};
+use super::producer_telemetry::{record_direct_publish, BusOperation};
 use super::{
     retryable, run_source, Bus, BusConsumer, BusTopologyConfig, MessagePublisher, MessageRouter,
     RunOptions, TransportError,
@@ -270,11 +271,39 @@ impl NatsBus {
 
 impl Bus for NatsBus {
     async fn send_message(&self, message: Message) -> Result<(), TransportError> {
+        record_direct_publish(None, "nats", BusOperation::Send, message, |message| async {
+            self.send_message_raw(message).await
+        })
+        .await
+    }
+
+    async fn publish_message(&self, message: Message) -> Result<(), TransportError> {
+        record_direct_publish(
+            None,
+            "nats",
+            BusOperation::Publish,
+            message,
+            |message| async { self.publish_message_raw(message).await },
+        )
+        .await
+    }
+
+    async fn send_outbox_message(&self, message: Message) -> Result<(), TransportError> {
+        self.send_message_raw(message).await
+    }
+
+    async fn publish_outbox_message(&self, message: Message) -> Result<(), TransportError> {
+        self.publish_message_raw(message).await
+    }
+}
+
+impl NatsBus {
+    async fn send_message_raw(&self, message: Message) -> Result<(), TransportError> {
         self.validated_namespace()?;
         self.cmd_publisher.publish(message).await
     }
 
-    async fn publish_message(&self, message: Message) -> Result<(), TransportError> {
+    async fn publish_message_raw(&self, message: Message) -> Result<(), TransportError> {
         self.validated_namespace()?;
         self.evt_publisher.publish(message).await
     }