AsyncAPI Code Generation with Corvus: Health and Telemetry
At endjin, we maintain Corvus.JsonSchema, and in the previous post we looked at request/reply patterns over messaging.
Now let's talk about what happens after you deploy. Because once your producers and consumers are running in production, the question shifts from "does it work?" to "is it still working, and how well?"
The observability gap in messaging
HTTP services have it relatively easy when it comes to observability. Every request/response pair has a status code, a duration, and a natural correlation point. Most web frameworks emit OpenTelemetry traces out of the box.
Messaging is harder. A producer publishes a message and moves on. It doesn't know whether the consumer received it, how long processing took, or whether the message ended up in a dead-letter queue three hops downstream. Without intentional instrumentation, you're flying blind between the moment you publish and the moment your handler finishes processing.
The Corvus AsyncAPI runtime addresses this with three complementary mechanisms: distributed tracing through OpenTelemetry, transport-level health checks, and per-subscription liveness monitoring.
Distributed tracing with InstrumentedMessageTransport
The core idea is a decorator. You take whatever transport you're already using (NATS, Kafka, AMQP, or any other) and wrap it with InstrumentedMessageTransport:
using Corvus.Text.Json.AsyncApi;
using Corvus.Text.Json.AsyncApi.Nats;
NatsMessageTransport raw = await NatsMessageTransport.CreateAsync(
new NatsTransportOptions { Url = "nats://localhost:4222" });
InstrumentedMessageTransport transport = new(raw, "nats");
The second argument is a logical name for your transport. It appears in span attributes and metric dimensions, which matters when you have multiple transports in the same process.
From this point on, every operation that flows through the transport is instrumented automatically. Publish operations create a span, subscribe operations create a span, and dead-letter routing creates a span. The producer injects W3C traceparent and tracestate headers into each message, and the consumer extracts them. Your distributed trace therefore links the producer span to the consumer span, even when they run in different processes on different machines.
What gets recorded
The instrumented transport emits both traces and metrics:
Traces use System.Diagnostics.Activity (the .NET embodiment of OpenTelemetry spans). Each publish, receive, and dead-letter operation creates an Activity with attributes for the channel name, message size, transport name, and - for consumers - the handler duration.
Metrics include counters for messages sent and received, histograms for processing duration and message body size, and counters for error-path actions (dead-letters, skips, aborts, retries). These flow through System.Diagnostics.Metrics, so they work with any OpenTelemetry-compatible exporter.
The important design property is that all of this is zero-cost when no listener is attached. The ActivitySource and Meter instances check whether anyone is listening before allocating spans or recording measurements. In a test environment with no exporter configured, the instrumentation adds no measurable overhead.
Wiring up an exporter
If you're using the OpenTelemetry .NET SDK, you subscribe to the instrumented transport's ActivitySource and Meter:
using OpenTelemetry;
using OpenTelemetry.Trace;
using OpenTelemetry.Metrics;
var tracerProvider = Sdk.CreateTracerProviderBuilder()
.AddSource(AsyncApiTelemetry.ActivitySourceName)
.AddOtlpExporter()
.Build();
var meterProvider = Sdk.CreateMeterProviderBuilder()
.AddMeter(AsyncApiTelemetry.MeterName)
.AddOtlpExporter()
.Build();
From here, your traces appear in Jaeger, Zipkin, Azure Monitor, or whichever backend you use. The producer-to-consumer correlation is automatic.
Processing loop heartbeats
Distributed tracing tells you about individual messages. But there's a failure mode it doesn't catch: a consumer loop that has silently stopped processing.
This can happen for several reasons. An unhandled exception escapes the error policy. The cancellation token fires unexpectedly. The transport connection drops and the reconnection logic gives up. In all these cases, the consumer appears healthy from the outside (the process is running, the health endpoint returns 200) but no messages are being consumed.
ProcessingLoopHeartbeat addresses this by tracking liveness at the subscription level:
ProcessingLoopHeartbeat heartbeat = new();
NatsTransportOptions options = new()
{
Url = "nats://localhost:4222",
Heartbeat = heartbeat,
};
NatsMessageTransport transport = await NatsMessageTransport.CreateAsync(options);
Each subscription's processing loop ticks the heartbeat on every iteration. You can then query liveness for a specific channel pattern, or enumerate all subscriptions and their status:
// Check a specific subscription
bool alive = heartbeat.IsAlive(
"smartylighting.streetlights.1.0.action.*.lighting.measured");
// Enumerate all subscriptions
foreach (var status in heartbeat.GetSubscriptionStatuses())
{
Console.WriteLine(
$"{status.Channel}: last tick {status.LastTick}, " +
$"{(status.IsAlive ? "alive" : "STALE")}");
}
The default staleness threshold is 30 seconds. If a loop hasn't ticked in that window, IsAlive returns false. You can surface this in a custom health check, a Kubernetes liveness probe, or a monitoring dashboard - whatever suits your operational model.
ASP.NET Core health checks
For transport-level connectivity monitoring, the Corvus.Text.Json.AsyncApi.HealthChecks package integrates directly with the ASP.NET Core health check infrastructure:
dotnet add package Corvus.Text.Json.AsyncApi.HealthChecks
using Corvus.Text.Json.AsyncApi.HealthChecks;
var builder = WebApplication.CreateBuilder(args);
builder.Services.AddHealthChecks()
.AddAsyncApiTransport("nats-transport", transport);
var app = builder.Build();
app.MapHealthChecks("/health");
app.Run();
The health check works with any transport that implements IHealthCheckableTransport - which includes NatsMessageTransport, KafkaMessageTransport, AmqpMessageTransport, and InMemoryMessageTransport. It checks IsConnected and calls PingAsync to verify that the broker is actually reachable, not just that a socket was opened at some point in the past.
Combining health checks with heartbeats
The transport health check and the processing loop heartbeat answer different questions. The health check tells you "can I reach the broker?" The heartbeat tells you "is my consumer actually processing messages on this channel?"
In production, you typically want both. The health check surfaces in your /health endpoint for load balancers and orchestrators. The heartbeat surfaces in a more granular operational endpoint or alert that tells you when a specific subscription has gone silent.
Putting it together
Here's what a production-ready consumer setup looks like with all three observability mechanisms in place:
// 1. Create the raw transport
NatsMessageTransport raw = await NatsMessageTransport.CreateAsync(
new NatsTransportOptions
{
Url = "nats://localhost:4222",
Heartbeat = heartbeat,
});
// 2. Wrap with instrumentation
InstrumentedMessageTransport transport = new(raw, "nats");
// 3. Create and start the consumer
LightMeasurementHandler handler = new();
ReceiveLightMeasurementConsumer consumer = new(
transport,
handler,
errorPolicy: new DefaultMessageErrorPolicy(
deserializationAction: MessageErrorAction.DeadLetter,
handlerAction: MessageErrorAction.DeadLetter,
transportAction: MessageErrorAction.Abort),
validationMode: ValidationMode.Basic);
await consumer.StartAsync();
The instrumented transport emits traces and metrics for every message. The heartbeat monitors the processing loop for silent failures. And if you add the health check package, your /health endpoint reflects transport connectivity.
If you're not familiar with the consumer patterns and error policies used here, the previous post in this series covers them in detail. For a deeper dive into the full range of AsyncAPI features, the AsyncAPI documentation on corvus-oss.org is the comprehensive reference.
In the [ref slug=asyncapi-code-generation-with-corvus-durability-and-resumption text=next post], we'll look at durability - how each transport handles message resumption after restarts, and what you need to configure for at-least-once delivery guarantees.
