AsyncAPI Code Generation with Corvus: Durability and Resumption

By Matthew Adams Co-Founder 29th July 2026 · 5 min read

At endjin, we maintain Corvus.JsonSchema, and in the previous post we added observability to our producers and consumers.

Now let's talk about what happens when things go wrong at the infrastructure level - specifically, what happens to messages when your consumer restarts.

The resumption problem

In the earlier posts, we focused on the happy path: a consumer starts, subscribes to a channel, and processes messages as they arrive. But production systems restart. Deployments roll. Pods get evicted. And the question that matters is: when your consumer comes back, does it pick up where it left off, or does it lose everything that was published while it was down?

The answer depends entirely on how you configure the transport. The Corvus runtime handles acknowledgement automatically. Your handler runs, and if it completes successfully the transport tells the broker "this message is done." But the durability of that arrangement (whether the broker holds messages for you, how long, and what identity it associates with your consumer) is something you configure per-transport.

The general pattern

Across all transports, the Corvus consumer follows the same acknowledgement protocol:

1. Message arrives from the broker
2. The consumer validates the payload against the schema
3. If validation passes, your handler is called
4. If your handler succeeds, the transport acknowledges the message
5. If your handler throws, the error policy decides the outcome (skip, dead-letter, or abort)
6. Acknowledgement is sent for skip and dead-letter; for abort, the consumer stops without acknowledging

The important consequence is that if your process crashes at any point before step 4 completes, the message has not been acknowledged. The broker will redeliver it when your consumer reconnects. This gives you at-least-once delivery semantics by default, which means your handlers should be idempotent. Processing the same message twice should be safe.

Kafka: consumer groups and committed offsets

Kafka tracks progress through consumer group offsets. All consumers that share a GroupId coordinate through Kafka to ensure each partition is consumed by exactly one group member. When your handler completes, the transport commits the offset for that message. On restart, the consumer resumes from the last committed offset.

using Corvus.Text.Json.AsyncApi.Kafka;

KafkaTransportOptions options = new()
{
    BootstrapServers = "localhost:9092",
    GroupId = "order-processor-v1",
    AutoOffsetReset = AutoOffsetReset.Earliest,
};

await using KafkaMessageTransport transport = new(options);

The GroupId is the identity that Kafka associates with your consumer's progress. If you deploy a new version with the same group ID, it resumes from where the previous version left off. If you change the group ID (perhaps to reprocess historical messages), the new group starts from the position specified by AutoOffsetReset.

The transport disables Kafka's auto-commit and commits explicitly after your handler succeeds. This means that if your process crashes mid-handler, the offset is not committed, and the message will be redelivered on restart.

AMQP/RabbitMQ: durable queues and explicit acknowledgement

AMQP takes a different approach. Rather than tracking offsets in a log, RabbitMQ holds messages in a queue and delivers them one at a time (or in configurable batches). When your handler completes, the transport sends BasicAck. If your handler fails and the error policy dead-letters, it sends BasicNack. If your process crashes without acknowledging, RabbitMQ redelivers the message.

using Corvus.Text.Json.AsyncApi.Amqp;

AmqpTransportOptions options = new()
{
    ConnectionUri = "amqp://guest:guest@localhost:5672/",
    QueueDurable = true,
    ExchangeDurable = true,
    PrefetchCount = 10,
    DeadLetterExchange = "orders.dead-letter",
};

await using AmqpMessageTransport transport =
    await AmqpMessageTransport.CreateAsync(options);

The key configuration for durability is QueueDurable = true, which ensures the queue itself survives broker restarts. Combined with the automatic acknowledgement behaviour, this gives you reliable at-least-once delivery: messages persist in the queue until explicitly acknowledged, and unacknowledged messages are redelivered after a consumer disconnects.

The PrefetchCount controls how many unacknowledged messages the broker will deliver to your consumer at once. A lower value gives stronger ordering guarantees (at the cost of throughput). A value of 1 means strict sequential processing.

NATS: Core vs. JetStream

NATS has two distinct modes, with different durability. Core NATS is a pure pub/sub system with no persistence. If your consumer is offline when a message is published, that message is gone. JetStream adds persistence, durable consumers, and acknowledgement semantics.

For ephemeral use cases (live telemetry dashboards, real-time notifications where missing a few messages is acceptable), Core NATS is fine:

NatsTransportOptions options = new()
{
    Url = "nats://localhost:4222",
    Name = "telemetry-dashboard",
};

For durable processing where you cannot afford to lose messages, enable JetStream:

using Corvus.Text.Json.AsyncApi.Nats;

NatsTransportOptions options = new()
{
    Url = "nats://localhost:4222",
    Name = "order-processor",
    UseJetStream = true,
    StreamName = "orders",
    ConsumerName = "order-processor-v1",
    AckWait = TimeSpan.FromSeconds(30),
    MaxDeliver = 5,
    StorageType = StorageType.File,
    DeliverPolicy = DeliverPolicy.All,
};

await using NatsMessageTransport transport =
    await NatsMessageTransport.CreateAsync(options);

The ConsumerName is the durable identity - analogous to Kafka's GroupId. The AckWait defines how long JetStream waits before assuming your consumer crashed and redelivering. The MaxDeliver caps how many times a message can be redelivered before JetStream gives up (at which point it's effectively dead-lettered).

MQTT: persistent sessions

MQTT's durability model revolves around the combination of a stable ClientId and CleanSession = false. When your consumer disconnects with a persistent session, the broker queues messages published to its subscribed topics. On reconnect (with the same ClientId), queued messages are delivered.

using Corvus.Text.Json.AsyncApi.Mqtt;

MqttTransportOptions options = new()
{
    Host = "localhost",
    Port = 1883,
    ClientId = "order-processor-prod-01",
    CleanSession = false,
    QualityOfServiceLevel = MqttQualityOfServiceLevel.AtLeastOnce,
};

await using MqttMessageTransport transport =
    await MqttMessageTransport.CreateAsync(options);

The critical point is that ClientId must be stable across restarts. If your deployment generates a random client ID each time (a common mistake in containerised environments), the broker treats each restart as a new client and the persistent session is never resumed. Use a deterministic ID derived from your service name and instance identifier.

Azure Service Bus: PeekLock settlement

Azure Service Bus uses a lock-based model. When a message is delivered, the broker locks it for your consumer. Your handler processes it, and the transport completes (acknowledges) the lock. If your process crashes before completing, the lock expires after a configurable duration, and the message becomes available for redelivery.

using Corvus.Text.Json.AsyncApi.AzureServiceBus;

AzureServiceBusTransportOptions options = new()
{
    ConnectionString = "<connection-string>",
    QueueName = "orders",
    ReceiveMode = Azure.Messaging.ServiceBus.ServiceBusReceiveMode.PeekLock,
    MaxAutoLockRenewalDuration = TimeSpan.FromMinutes(5),
};

await using AzureServiceBusMessageTransport transport =
    await AzureServiceBusMessageTransport.CreateAsync(options);

The MaxAutoLockRenewalDuration is worth understanding. If your handler takes longer than the lock duration to process a message, the transport automatically renews the lock in the background. Set this to something longer than your longest expected handler execution time.

For topic-based pub/sub (where multiple services independently consume the same messages), use a subscription:

AzureServiceBusTransportOptions options = new()
{
    ConnectionString = "<connection-string>",
    UseTopic = true,
    TopicName = "order-events",
    SubscriptionName = "payment-processor-v1",
};

Choosing the right durability model

The transport you choose (or more accurately, the transport your organisation has already chosen) determines the durability model. But across all of them, the Corvus runtime provides the same contract: your handler runs, and if it succeeds the message is acknowledged. The differences are in what happens during downtime and how identity is managed.

If you're starting fresh and choosing a broker, the key trade-offs are:

Kafka excels when you need message replay (reprocessing from an earlier offset) and high-throughput ordered delivery. The log-based model means messages are retained for a configurable period regardless of whether they've been consumed.

AMQP/RabbitMQ suits work-queue patterns where messages should be processed exactly once (or close to it) and removed from the queue immediately after. It's simpler operationally for moderate throughput.

NATS JetStream offers a middle ground - persistent streams with a lighter operational footprint than Kafka. Good for teams that want durability without the Kafka ecosystem complexity.

Azure Service Bus integrates naturally with Azure-hosted services, provides built-in dead-letter subqueues, and handles lock renewal transparently. The managed service model means less operational burden.

MQTT is the natural fit for IoT and edge scenarios where devices connect intermittently and the broker needs to queue messages during disconnection windows.

In all cases, the generated consumer code remains identical. The only thing that changes is the transport options you pass at startup. That is exactly the kind of infrastructure concern that should live in configuration rather than in your business logic.

For the complete transport configuration reference, see the AsyncAPI documentation on corvus-oss.org.

In the [ref slug=asyncapi-code-generation-with-corvus-filtering text=next post], we'll look at filtering - generating code for just the channels you need, which becomes essential as your event-driven architecture grows.

FAQs