ProsodyEvents.Prosody 0.6.1

Prosody: C# Bindings for Kafka

Prosody offers C# bindings to the Prosody Kafka client, providing features for message production and consumption, including configurable retry mechanisms, failure handling strategies, and integrated OpenTelemetry support for distributed tracing.

Features

• Kafka Consumer: Per-key ordering with cross-key concurrency, offset management, consumer groups
• Kafka Producer: Idempotent delivery with configurable retries
• Timer System: Persistent scheduled execution backed by Cassandra or in-memory store
• Keyed State: Per-key persistent state (value/map/deque) with transactional, at-least-once semantics
• Quality of Service: Fair scheduling limits concurrency and prevents failures from starving fresh traffic. Pipeline mode adds deferred retry and monopolization detection
• Distributed Tracing: OpenTelemetry integration for tracing message flow across services
• Error Monitoring: Optional Sentry integration for automatic handler exception reporting
• Backpressure: Pauses partitions when handlers fall behind
• Mocking: In-memory Kafka broker for tests (WithMock(true))
• Failure Handling: Pipeline (retry forever), Low-Latency (dead letter), Best-Effort (log and skip)

Installation

Add the NuGet package to your project:

dotnet add package ProsodyEvents.Prosody

Quick Start

using Prosody;

// Initialize the client with the builder pattern
await using var client = await ProsodyClientBuilder.Create()
    // Bootstrap servers should normally be set using the PROSODY_BOOTSTRAP_SERVERS environment variable
    .WithBootstrapServers("localhost:9092")
    // To allow loopbacks, the SourceSystem must be different from the GroupId.
    // Normally, the SourceSystem would be left unspecified, which would default to the GroupId.
    .WithSourceSystem("my-application-source")
    // The GroupId should be set to the name of your application
    .WithGroupId("my-application")
    // Topics the client should subscribe to
    .WithSubscribedTopics("my-topic")
    .BuildAsync();

// Define a message handler
var messageHandler = new MyHandler();

// Subscribe to messages using the message handler
await client.SubscribeAsync(messageHandler);

// Send a message to a topic
await client.SendAsync("my-topic", "message-key", new { Content = "Hello, Kafka!" });
await client.ExciseAsync("my-topic", "obsolete-key");

// Ensure proper shutdown when done
await client.ShutdownAsync();

// Handler implementation
public class MyHandler : IProsodyHandler<MyPayload>
{
    public async Task OnExciseAsync(ProsodyContext prosodyContext, ExciseMessage message, CancellationToken cancellationToken)
    {
        Console.WriteLine($"Excise key: {message.Key}");
        await prosodyContext.ClearScheduledAsync();
    }

    public async Task OnMessageAsync(ProsodyContext prosodyContext, Message<MyPayload> message, CancellationToken cancellationToken)
    {
        // Process the received message
        var payload = message.Payload;
        Console.WriteLine($"Received message: {payload}");

        // Schedule a timer for delayed processing (requires Cassandra unless Mock = true)
        if (payload?.ScheduleFollowup == true)
        {
            var futureTime = DateTimeOffset.UtcNow.AddSeconds(30);
            await prosodyContext.ScheduleAsync(futureTime);
        }
    }

    public Task OnTimerAsync(ProsodyContext prosodyContext, ProsodyTimer timer, CancellationToken cancellationToken)
    {
        // Handle timer firing
        Console.WriteLine($"Timer fired for key: {timer.Key} at {timer.Time}");
        return Task.CompletedTask;
    }
}

Excise records

Applications can copy event data into keyed state and external stores. A regulatory or contractual deletion must remove every copy for one key.

An excise record carries this deletion command. Kafka encodes the command as a key with no payload. During topic compaction, Kafka deletes earlier values for the key. Call ExciseAsync(topic, key) to send the record. Prosody routes the record to OnExciseAsync. The handler must delete all consumer-owned data for the key.

Each handler must implement OnMessageAsync, OnExciseAsync, and OnTimerAsync. OnExciseAsync receives record metadata and no payload.

If an excise record is a request, return a response from OnExciseAsync. Prosody uses this response as the subsystem result.

Architecture

Prosody enables efficient, parallel processing of Kafka messages while maintaining order for messages with the same key:

• Partition-Level Parallelism: Separate management of each Kafka partition
• Key-Based Queuing: Ordered processing for each key within a partition
• Concurrent Processing: Simultaneous processing of different keys
• Backpressure Management: Pause consumption from backed-up partitions

Quality of Service

All modes use fair scheduling to limit concurrency and distribute execution time. Pipeline mode adds deferred retry and monopolization detection.

Fair Scheduling (All Modes)

The scheduler controls which message runs next and how many run concurrently.

Virtual Time (VT): Each key accumulates VT equal to its handler execution time. The scheduler picks the key with the lowest VT. A key that runs for 500ms accumulates 500ms of VT; a key that hasn't run recently has zero VT and gets priority.

Two-Class Split: Normal messages and failure retries have separate VT pools. The scheduler allocates execution time between them (default: 70% normal, 30% failure). During a failure spike, retries get at most 30% of execution time—fresh messages continue processing.

Starvation Prevention: Tasks receive a quadratic priority boost based on wait time. A task waiting 2 minutes (configurable) gets maximum boost, overriding VT disadvantage.

Deferred Retry (Pipeline Mode)

Moves failing keys to timer-based retry so the partition can continue processing other keys.

On transient failure: store the message offset in Cassandra, schedule a timer, return success. The partition advances. When the timer fires, reload the message from Kafka and retry.

// Configure defer behavior
await using var client = await ProsodyClientBuilder.Create()
    .WithGroupId("my-consumer-group")
    .WithSubscribedTopics("my-topic")
    .Configure(options =>
    {
        options.DeferEnabled = true;                              // Enable deferral (default: true)
        options.DeferBase = TimeSpan.FromSeconds(1);              // Wait 1s before first retry
        options.DeferMaxDelay = TimeSpan.FromHours(24);           // Cap at 24 hours
        options.DeferFailureThreshold = 0.9;                      // Disable when >90% failing
    })
    .BuildAsync();

Failure Rate Gating: When >90% of recent messages fail, deferral disables. The retry middleware blocks the partition, applying backpressure upstream.

Monopolization Detection (Pipeline Mode)

Rejects keys that consume too much execution time.

The middleware tracks per-key execution time in 5-minute rolling windows. Keys exceeding 90% of window time are rejected with a transient error, routing them through defer.

// Configure monopolization detection
await using var client = await ProsodyClientBuilder.Create()
    .WithGroupId("my-consumer-group")
    .WithSubscribedTopics("my-topic")
    .Configure(options =>
    {
        options.MonopolizationEnabled = true;                     // Enable detection (default: true)
        options.MonopolizationThreshold = 0.9;                    // Reject keys using >90% of window
        options.MonopolizationWindow = TimeSpan.FromMinutes(5);   // 5-minute window
    })
    .BuildAsync();

Handler Timeout

Handlers are automatically cancelled if they exceed a deadline:

await using var client = await ProsodyClientBuilder.Create()
    .WithGroupId("my-consumer-group")
    .WithSubscribedTopics("my-topic")
    .Configure(options =>
    {
        options.Timeout = TimeSpan.FromSeconds(30);               // Cancel after 30 seconds
        options.StallThreshold = TimeSpan.FromSeconds(60);        // Report unhealthy after 60 seconds
    })
    .BuildAsync();

When a handler times out, prosodyContext.ShouldCancel becomes true and the CancellationToken is cancelled. The handler should exit promptly. If not specified, timeout defaults to 80% of StallThreshold.

Configuration

For the complete configuration reference, see CONFIGURATION.md.

ClientOptions properties take precedence. Unset properties use environment variables, then library defaults.

Client construction is asynchronous. Use ProsodyClient.CreateAsync or ProsodyClientBuilder.BuildAsync. Both validate the options, connect to Kafka, and connect to Cassandra when configured. They return a client that can send, request, and subscribe, or they throw. The client is not subscribed. The readiness probe reports ready only after SubscribeAsync receives partitions.

One client is safe for concurrent use from any thread. It holds one subscription at a time. SubscribeAsync throws when the client is already subscribed, so one owner coordinates subscribe and unsubscribe calls. Concurrent and repeated ShutdownAsync calls await the same shutdown.

Liveness and Readiness Probes

Prosody includes a built-in probe server for consumer-based applications that provides health check endpoints. The probe server is tied to the consumer's lifecycle and offers two main endpoints:

1. /readyz: A readiness probe that checks if any partitions are assigned to the consumer. Returns a success status only when the consumer has at least one partition assigned, indicating it's ready to process messages.

2. /livez: A liveness probe that checks if any partitions have stalled (haven't processed a message within a configured time threshold).

Configure the probe server using the builder:

await using var client = await ProsodyClientBuilder.Create()
    .WithGroupId("my-consumer-group")
    .WithSubscribedTopics("my-topic")
    .WithProbePort(8000)                                          // Set to 0 to disable
    .Configure(options =>
    {
        options.StallThreshold = TimeSpan.FromSeconds(15);        // 15 seconds before considering a partition stalled
    })
    .BuildAsync();

Or via environment variables:

PROSODY_PROBE_PORT=8000  # Set to 0 to disable
PROSODY_STALL_THRESHOLD=15s  # Default stall detection threshold

Important Notes

1. The probe server starts automatically when the consumer is subscribed and stops when unsubscribed.
2. A partition is considered "stalled" if it hasn't processed a message within the StallThreshold duration.
3. The stall threshold should be set based on your application's message processing latency and expected message frequency.
4. Setting the threshold too low might cause false positives, while setting it too high could delay detection of actual issues.
5. The probe server is only active when consuming messages (not for producer-only usage).

You can monitor the stall state programmatically using the client's methods:

// Get the number of partitions currently assigned to this consumer
var partitionCount = await client.AssignedPartitionCountAsync();

// Check if the consumer has stalled partitions
if (await client.IsStalledAsync())
{
    Console.WriteLine("Consumer has stalled partitions");
}

Subsystems

A consumer group ID identifies a set of processes that share records and the keyed state that the group owns. A subsystem can include one or more services and consumer groups. If callers use these IDs, a refactor can require changes to each caller.

A subsystem gives requests and published state one stable public name. Callers use this name instead of consumer group IDs. You can change its services and consumer groups without changing callers. Prosody uses the first response to a subsystem request. For each published-state read, it uses one consumer group that publishes the collection.

Requests

Kafka decouples producers from consumers, so a send does not return consumer results. This asynchronous model lets each service process records independently. Some operations must wait for consumer results before they continue. A request recovers synchrony for the caller while consumers continue asynchronous processing.

Send a request from a handler or other application code. The Prosody client does not need an active subscription. The result dictionary uses canonical subsystem names as keys. Use RequestExciseAsync to send an excise record and collect the same outcome type.

Do not rely on dictionary order. The dictionary contains one entry for each selected subsystem. A missing response becomes a timeout Failure<T>; Prosody does not omit the subsystem. A request-level failure throws: ArgumentException for invalid input, InvalidOperationException after shutdown, and ProsodyException when Kafka does not accept the request. Do not await a request if the current consumer group must process it for the same key. That group cannot process it until the handler returns.

Message and excise handler return values become successful outcomes. Timer handlers do not return request outcomes.

Set Subsystem to inventory on the client that subscribes this handler.

public sealed record Order(string Type);
public sealed record InventoryResponse(string Accepted);

public sealed class InventoryHandler : IProsodyRequestHandler<Order, InventoryResponse>
{
    public Task<InventoryResponse> OnMessageAsync(
        ProsodyContext context,
        Message<Order> message,
        CancellationToken cancellationToken
    ) => Task.FromResult(new InventoryResponse(message.Key));

    public Task<InventoryResponse> OnExciseAsync(
        ProsodyContext context,
        ExciseMessage message,
        CancellationToken cancellationToken
    ) => Task.FromResult(new InventoryResponse(message.Key));

    public Task OnTimerAsync(
        ProsodyContext context,
        ProsodyTimer timer,
        CancellationToken cancellationToken
    ) => Task.CompletedTask;
}

Send the request:

string[] subsystems = ["inventory", "billing"];
IReadOnlyDictionary<string, Outcome<InventoryResponse>> results = await client.RequestAsync<Order, InventoryResponse>(
    "orders",
    "order-1",
    new Order("order.created"),
    subsystems,
    TimeSpan.FromSeconds(2)
);

foreach (var (subsystem, outcome) in results)
{
    if (outcome is Success<InventoryResponse> success)
    {
        Console.WriteLine($"{subsystem}: {success.Value}");
    }
    else if (outcome is Failure<InventoryResponse> failure)
    {
        Console.Error.WriteLine($"{subsystem}: {failure.Error.Message}");
    }
}

The example can print these results:

inventory: InventoryResponse { Accepted = order-1 }
billing: no response arrived before the deadline

Each Failure<T> contains one typed response error.

Each response error uses Prosody's message.

Local JSON errors use the .NET decoder's message.

JSON null remains a successful result. Use a nullable response type when a handler can return null.

Use the JsonTypeInfo overload for trim-safe request and response serialization.

Advanced Usage

Pipeline Mode

Pipeline mode is the default mode. Ensures ordered processing, retrying failed operations indefinitely:

// Initialize client in pipeline mode
await using var client = await ProsodyClientBuilder.Create()
    .WithMode(ClientMode.Pipeline)  // Explicitly set pipeline mode (this is the default)
    .WithGroupId("my-consumer-group")
    .WithSubscribedTopics("my-topic")
    .BuildAsync();

Low-Latency Mode

Prioritizes quick processing, sending persistently failing messages to a failure topic:

// Initialize client in low-latency mode
await using var client = await ProsodyClientBuilder.Create()
    .WithMode(ClientMode.LowLatency)  // Set low-latency mode
    .WithGroupId("my-consumer-group")
    .WithSubscribedTopics("my-topic")
    .WithFailureTopic("failed-messages")  // Specify a topic for failed messages
    .BuildAsync();

Best-Effort Mode

Optimized for development environments or services where message processing failures are acceptable:

// Initialize client in best-effort mode
await using var client = await ProsodyClientBuilder.Create()
    .WithMode(ClientMode.BestEffort)  // Set best-effort mode
    .WithGroupId("my-consumer-group")
    .WithSubscribedTopics("my-topic")
    .BuildAsync();

Event Type Filtering

Prosody supports filtering messages based on event type prefixes, allowing your consumer to process only specific types of events:

// Process only events with types starting with "user." or "account."
await using var client = await ProsodyClientBuilder.Create()
    .WithGroupId("my-consumer-group")
    .WithSubscribedTopics("my-topic")
    .WithAllowedEvents("user.", "account.")
    .BuildAsync();

Or via environment variables:

PROSODY_ALLOWED_EVENTS=user.,account.

Matching Behavior

Prefixes must match exactly from the start of the event type:

✓ Matches:

• {"type": "user.created"} matches prefix user.
• {"type": "account.deleted"} matches prefix account.

✗ No Match:

• {"type": "admin.user.created"} doesn't match user.
• {"type": "my.account.deleted"} doesn't match account.
• {"type": "notification"} doesn't match any prefix

If no prefixes are configured, all messages are processed. Messages without a type field are always processed.

Source System Deduplication

Prosody prevents processing loops in distributed systems by tracking the source of each message:

// Consumer and producer in one application
await using var client = await ProsodyClientBuilder.Create()
    .WithGroupId("my-service")
    .WithSourceSystem("my-service-producer")  // Must differ from GroupId to allow loopbacks; defaults to GroupId
    .WithSubscribedTopics("my-topic")
    .BuildAsync();

Or via environment variable:

PROSODY_SOURCE_SYSTEM=my-service-producer

How It Works

1. Producers add a source-system header to all outgoing messages.
2. Consumers check this header on incoming messages.
3. If a message's source system matches the consumer's group ID, the message is skipped.

This prevents endless loops where a service consumes its own produced messages.

Message Deduplication

Prosody uses the JSON payload's id field to detect duplicate messages. For a consumer group and idempotence version, messages with the same ID, key, topic, and partition share a deduplication record. Other messages can arrive between duplicates. Prosody skips a duplicate while its record remains in the cache or persistent store. Deduplication is best-effort; handlers must remain idempotent.

// Messages with IDs are deduplicated per key
await client.SendAsync("my-topic", "key1", new
{
    Id = "msg-123",      // Message will be processed
    Content = "Hello!"
});

await client.SendAsync("my-topic", "key1", new
{
    Id = "msg-123",      // Message will be skipped (duplicate)
    Content = "Hello again!"
});

await client.SendAsync("my-topic", "key2", new
{
    Id = "msg-123",      // Message will be processed (different key)
    Content = "Hello!"
});

Deduplication uses a global in-memory cache shared across all partitions, which survives partition reassignments within the same process. For cross-restart deduplication, a Cassandra-backed persistent store is used when Cassandra is configured.

Deduplication is always active. IdempotenceCacheSize must be greater than 0; a value of 0 (via either the option or PROSODY_IDEMPOTENCE_CACHE_SIZE=0) is rejected when the client is built. The same value sizes the producer idempotence cache, which skips a send whose event ID the client already sent. The cache capacity can be tuned:

await using var client = await ProsodyClientBuilder.Create()
    .WithGroupId("my-consumer-group")
    .WithSubscribedTopics("my-topic")
    .Configure(options => options.IdempotenceCacheSize = 16384)   // Tune the shared cache capacity
    .BuildAsync();

To invalidate all previously recorded dedup entries and force reprocessing, change the version string:

.Configure(options => options.IdempotenceVersion = "2")          // Invalidate all prior dedup records

The Cassandra TTL for dedup records defaults to 7 days and can be adjusted:

.Configure(options => options.IdempotenceTtl = TimeSpan.FromDays(14))  // Keep records for 14 days

Note that in-memory deduplication is best-effort and not guaranteed. Duplicates can still occur when instances restart if Cassandra is not configured.

Timer Functionality

Prosody supports timer-based delayed execution within message handlers. When a timer fires, your handler's OnTimerAsync method will be called:

public class MyHandler : IProsodyHandler<MyPayload>
{
    public async Task OnExciseAsync(
        ProsodyContext prosodyContext,
        ExciseMessage message,
        CancellationToken cancellationToken)
    {
        await prosodyContext.ClearScheduledAsync();
    }

    public async Task OnMessageAsync(ProsodyContext prosodyContext, Message<MyPayload> message, CancellationToken cancellationToken)
    {
        // Schedule a timer to fire in 30 seconds
        var futureTime = DateTimeOffset.UtcNow.AddSeconds(30);
        await prosodyContext.ScheduleAsync(futureTime);

        // Schedule multiple timers
        var oneMinute = DateTimeOffset.UtcNow.AddMinutes(1);
        var twoMinutes = DateTimeOffset.UtcNow.AddMinutes(2);
        await prosodyContext.ScheduleAsync(oneMinute);
        await prosodyContext.ScheduleAsync(twoMinutes);

        // Check what's scheduled
        var scheduledTimes = await prosodyContext.ScheduledAsync();
        Console.WriteLine($"Scheduled timers: {scheduledTimes.Length}");
    }

    public Task OnTimerAsync(ProsodyContext prosodyContext, ProsodyTimer timer, CancellationToken cancellationToken)
    {
        Console.WriteLine("Timer fired!");
        Console.WriteLine($"Key: {timer.Key}");
        Console.WriteLine($"Scheduled time: {timer.Time}");
        return Task.CompletedTask;
    }
}

Timer Methods

The context provides timer scheduling methods that allow you to delay execution or implement timeout behavior:

• ScheduleAsync(DateTimeOffset time): Schedules a timer to fire at the specified time
• ClearAndScheduleAsync(DateTimeOffset time): Clears all timers and schedules a new one
• UnscheduleAsync(DateTimeOffset time): Removes a timer scheduled for the specified time
• ClearScheduledAsync(): Removes all scheduled timers
• ScheduledAsync(): Returns an array of all scheduled timer times

Timer Object

When a timer fires, the OnTimerAsync method receives a timer object with these properties:

• Key (string): The entity key identifying what this timer belongs to
• Time (DateTimeOffset): The time when this timer was scheduled to fire

Note: Timer precision is limited to seconds due to the underlying storage format. Sub-second precision in scheduled times will be rounded to the nearest second.

Timer Configuration

Timer functionality requires Cassandra for persistence unless running in mock mode. Configure Cassandra connection via environment variable:

PROSODY_CASSANDRA_NODES=localhost:9042  # Required for timer persistence

Or programmatically when creating the client:

await using var client = await ProsodyClientBuilder.Create()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("my-application")
    .WithSubscribedTopics("my-topic")
    .Configure(options => options.CassandraNodes = ["localhost:9042"])  // Required unless Mock = true
    .BuildAsync();

For testing, you can use mock mode to avoid Cassandra dependency:

// Mock mode for testing (timers work but aren't persisted)
await using var client = await ProsodyClientBuilder.Create()
    .WithBootstrapServers("localhost:9092")
    .WithGroupId("my-application")
    .WithSubscribedTopics("my-topic")
    .WithMock(true)  // No Cassandra required in mock mode
    .BuildAsync();

Keyed State

Many stream transformations must reason across multiple events or timer firings. Windows, state machines, aggregates, and complex event processing all require state.

A Kafka key identifies an entity, such as a customer or order. Keyed state gives each key independent working state for these transformations. With Cassandra, the state survives restarts and partition reassignment.

Prosody selects the current message or timer key. It processes one event at a time for that key but can process other keys concurrently. By default, Prosody commits pending keyed-state changes only when the handler succeeds. If the handler returns an error, Prosody discards those changes.

Give most collections a time to live (TTL). Set the TTL beyond the longest timer or workflow that uses the collection. Omit it when state must remain for inactive keys.

A counter for each key

Declare each collection once. Register it on the client. In a handler, get the current key's state from the event context:

var count = StateDefinition.Value<int>("count", ttl: TimeSpan.FromDays(30));

public sealed class CountHandler(ValueStateDefinition<int> count)
    : IProsodyHandler<Event>
{
    public async Task OnExciseAsync(
        ProsodyContext context,
        ExciseMessage message,
        CancellationToken cancellationToken)
    {
        await context.State(count).ClearAsync(cancellationToken);
    }

    public async Task OnMessageAsync(
        ProsodyContext context,
        Message<Event> message,
        CancellationToken cancellationToken)
    {
        var state = context.State(count);
        var current = (await state.GetAsync(cancellationToken)).GetValueOrDefault(0);
        await state.SetAsync(current + 1, cancellationToken);
    }

    public Task OnTimerAsync(
        ProsodyContext context,
        ProsodyTimer timer,
        CancellationToken cancellationToken) =>
        Task.CompletedTask;
}

var client = await ProsodyClientBuilder.Create()
    .WithStateCollections(count)
    .BuildAsync();

Each Kafka key now has an independent counter. A counter expires when that key has no update for 30 days.

Window activity into one notification

This example sends the first event for a user immediately. It collects later events for five minutes and then sends one summary.

The user ID is the Kafka key. Each user has an independent window.

var window = StateDefinition.Value<bool>("window", ttl: TimeSpan.FromDays(1));
var pending = StateDefinition.MessageDeque<Activity>(
    "pending", ttl: TimeSpan.FromDays(1), capacity: 100);

public async Task OnMessageAsync(
    ProsodyContext context,
    Message<Activity> message,
    CancellationToken cancellationToken)
{
    var windowState = context.State(window);
    var pendingState = context.State(pending);

    if ((await windowState.GetAsync(cancellationToken)).GetValueOrDefault(false))
    {
        await pendingState.PushBackAsync(message, cancellationToken);
        return;
    }

    await NotifyAsync(message.Key, [message]);
    await windowState.SetAsync(true, cancellationToken);
    await context.ClearAndScheduleAsync(
        DateTimeOffset.UtcNow + TimeSpan.FromMinutes(5));
}

public async Task OnTimerAsync(
    ProsodyContext context,
    ProsodyTimer timer,
    CancellationToken cancellationToken)
{
    var pendingState = context.State(pending);
    var batch = new List<Message<Activity>>();
    await foreach (var message in pendingState.WithCancellation(cancellationToken))
        batch.Add(message);

    if (batch.Count > 0) await NotifyAsync(timer.Key, batch);
    await pendingState.ClearAsync(cancellationToken);
    await context.State(window).ClearAsync(cancellationToken);
}

public async Task OnExciseAsync(
    ProsodyContext context,
    ExciseMessage message,
    CancellationToken cancellationToken)
{
    await context.State(pending).ClearAsync(cancellationToken);
    await context.State(window).ClearAsync(cancellationToken);
    await context.ClearScheduledAsync();
}

See the complete, compiled example for imports, types, client setup, and NotifyAsync: examples/keyed_state_windowing.cs.

Why this works:

• Register both definitions with WithStateCollections before BuildAsync(). Keyed state uses Cassandra unless Mock = true.
• Use ClearAndScheduleAsync, not ScheduleAsync, so a retried event does not add another timer for the same key.
• capacity: 100 and the one-day TTL bound the saved backlog. Overflow drops the oldest message because this example pushes at the back.
• A MessageDeque requires the original Kafka messages during the window. Use Deque when topic retention or compaction cannot provide them.
• Prosody runs one handler at a time for each key, so a user's message and timer handlers cannot overlap.
• A notification is outside the state transaction. A retry can send it again. Use a stable operation ID to reject duplicate notifications.

Collections and handles

A definition sets a collection's durable name, kind, and options. Register it once. Pass it to context.State() in a handler.

Do not reuse a durable name for a different collection kind or payload type. Create handles inside the handler. Do not retain handles or iterators.

Collection	JSON payload	Kafka message	Main operations
Value	StateDefinition.Value<T>	StateDefinition.MessageValue<TPayload>	GetAsync, SetAsync, ClearAsync
Ordered string map	StateDefinition.Map<TValue>	StateDefinition.MessageMap<TPayload>	GetAsync, GetManyAsync, ContainsKeyAsync, ContainsManyAsync, IsEmptyAsync, SetAsync, RemoveAsync, EnumerateAsync, EnumerateKeysAsync, EnumerateValuesAsync, ClearAsync
Deque	StateDefinition.Deque<T>	StateDefinition.MessageDeque<TPayload>	PushBackAsync, PushFrontAsync, PopBackAsync, PopFrontAsync, GetAsync, CountAsync, EnumerateAsync, ClearAsync
Ordered string set	StateDefinition.Set	-	AddAsync, RemoveAsync, ContainsAsync, ContainsManyAsync, IsEmptyAsync, EnumerateAsync, ClearAsync

Map, set, and deque scans use await foreach. Map keys and set members are strings. A set stores presence only.

Query a collection

Pass a KeyQuery to a map or set enumeration. Pass a PositionQuery to a deque enumeration. A query selects a direction, bounds, and a limit. A KeyQuery can also select a key prefix. Prosody applies each option in storage, so a query reads only the selected entries, members, or values.

• From and To are inclusive. After and Before are exclusive. Set at most one start and one end.
• Bounds are in iteration order. A ScanDirection.Backward query starts at the high end.
• KeyQuery.Range accepts an ascending KeyRange of keys, such as new KeyRange("a", "m"). The start is inclusive and the end is exclusive. A null start or end leaves that side open. It applies in either direction. A descending range selects no keys. Keys sort in the order of their UTF-8 bytes.
• PositionQuery.Range accepts an ascending System.Range of positions, such as 2..5 or 3... It applies in either direction. Use positions that count from the front; only ^0 is allowed, as the end. A descending range such as 5..2 throws, as in .NET slicing.
• Every option narrows the selection. Limit must be positive.
• To read the last N elements of a deque, use a reverse query with Limit = N: new PositionQuery { Direction = ScanDirection.Backward, Limit = N }.

To page through a map, pass the last key of the previous page as After:

var orders = StateDefinition.Map<Order>("orders");

// In a handler: read the order keys 100 at a time.
var map = context.State(orders);
var page = new KeyQuery { Prefix = "order:", Limit = 100 };
string? last = null;
do
{
    var keys = new List<string>();
    await foreach (var key in map.EnumerateKeysAsync(page with { After = last }, cancellationToken))
        keys.Add(key);

    await Process(keys);
    last = keys.Count == page.Limit ? keys[^1] : null;
}
while (last is not null);

The same query works on published readers and in either direction.

Reads return StateValue<T>. This type distinguishes an absent value from a stored default(T). A write of JSON null fails with PermanentStateException. Use ClearAsync (value, deque) or RemoveAsync (map) to delete.

Payload types guide JSON serialization. They do not add runtime validation.

When keyed-state changes become visible

By default, retries do not see pending state from a failed attempt. Reads in a handler see its earlier keyed-state writes. Prosody commits pending changes when the event succeeds and discards them when the handler throws.

This transaction applies only to keyed state. Some workflows need state changes before the handler ends, so each collection also provides explicit controls:

• readUncommitted: true persists keyed-state changes before Prosody records the event as complete. If the process stops between these steps, Prosody can process the same event again. The retry sees state changes from the earlier attempt. You must make these keyed-state changes idempotent. Each retry must produce the same state.
• await state.CommitAsync() commits the collection's pending changes before the handler ends. A later handler failure does not remove them.
• await state.RollbackAsync() discards pending changes since the last CommitAsync(). It cannot undo committed changes.
• Both return a StoreOutcome: Applied when they wrote or discarded buffered changes, or NoOp when nothing was buffered.

Keyed-state payloads use the client's JsonSerializerOptions. For AOT or trimmed builds, include every state payload type in the source-generated JsonSerializerContext; see AOT / Trim-safe Usage.

Published state

Some callers need only the current value for a key. They can accept a stale value or a race with a concurrent update.

Use topics and event sourcing when a consumer must process each state change in order. Use published state for direct, read-only lookup of persisted keyed state. The caller does not need to consume the owner's topics or maintain a separate lookup store.

Configure the subsystem name on each publisher. Enable publication on the collection definition. Register the definition on the Prosody client:

var currentOrder = StateDefinition.Value<Order>("current-order", published: true);
var options = new ClientOptions
{
    GroupId = "order-writer",
    Subsystem = "checkout",
    StateCollections = [currentOrder],
};

// The handler uses the key from its current event.
var ownedOrder = context.State(currentOrder);
await ownedOrder.SetAsync(updatedOrder, cancellationToken);

Read published state from a handler or other application code. The Prosody client does not need an active subscription. A client that only reads published state needs no SubscribedTopics.

Use the subsystem and the same definition to open a reader:

PublishedValue<Order> orderReader = await client.StateAsync("checkout", currentOrder);
StateValue<Order> value = await orderReader.GetAsync("customer-123", cancellationToken);

The reader cannot see pending changes that exist only in a handler. It cannot change the collection. Each read takes an explicit key because no handler supplies one.

Map, set, and deque readers return IAsyncEnumerable<T>. They fetch data in chunks. Pass ScanDirection.Backward to read in reverse order, or pass a KeyQuery or PositionQuery to select a page.

The default cache window is five seconds. Set readCache: StateReadCache.For(ttl) to select a different window. Use StateReadCache.Disabled to bypass the cache.

To stop publication, deploy the definition with published: false. Keep the definition registered during that deployment. Keep the subsystem configured during that deployment.

OpenTelemetry Tracing

Prosody supports OpenTelemetry tracing, allowing you to monitor and analyze the performance of your Kafka-based applications. The library will emit traces using the OTLP protocol if the OTEL_EXPORTER_OTLP_ENDPOINT environment variable is defined.

Note: Prosody emits its own traces separately because it uses its own tracing runtime, as it would be expensive to send all traces to C#.

Required Packages

To use OpenTelemetry tracing with Prosody, you need to install the following packages:

dotnet add package OpenTelemetry
dotnet add package OpenTelemetry.Extensions.Hosting
dotnet add package OpenTelemetry.Exporter.OpenTelemetryProtocol

Initializing Tracing

To initialize tracing in your application:

using OpenTelemetry;
using OpenTelemetry.Resources;
using OpenTelemetry.Trace;

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddOpenTelemetry()
    .ConfigureResource(resource => resource.AddService("my-service-name"))
    .WithTracing(tracing => tracing
        .AddSource("my-service-name")
        .AddOtlpExporter());

var app = builder.Build();

Setting OpenTelemetry Environment Variables

Set the following standard OpenTelemetry environment variables:

OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318
OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf
OTEL_SERVICE_NAME=my-service-name

For more information on these and other OpenTelemetry environment variables, refer to the OpenTelemetry specification.

Using Tracing in Your Application

After initializing tracing, you can define spans in your application, and they will be properly propagated through Kafka:

using System.Diagnostics;

public class MyHandler : IProsodyHandler<MyPayload>
{
    private static readonly ActivitySource ActivitySource = new("my-service-name");

    public Task OnExciseAsync(
        ProsodyContext prosodyContext,
        ExciseMessage message,
        CancellationToken cancellationToken) =>
        Task.CompletedTask;

    public async Task OnMessageAsync(ProsodyContext prosodyContext, Message<MyPayload> message, CancellationToken cancellationToken)
    {
        using var activity = ActivitySource.StartActivity("process-message");

        // Process the received message
        activity?.AddEvent(new ActivityEvent("message.received"));

        Console.WriteLine($"Received message: {message.Payload}");
    }

    public Task OnTimerAsync(ProsodyContext prosodyContext, ProsodyTimer timer, CancellationToken cancellationToken) => Task.CompletedTask;
}

Span Linking

By default, message execution spans use Child (child-of relationship — the execution span is part of the same trace as the producer). Timer execution spans use FollowsFrom (the execution span starts a new trace with a span link back to the scheduling span, since timer execution is causally related but not part of the same operation).

Both strategies are configurable via the MessageSpans / PROSODY_MESSAGE_SPANS and TimerSpans / PROSODY_TIMER_SPANS options. Accepted values: child, follows_from.

Best Practices

Ensuring Thread-Safe Handlers

Your event handler methods will be called concurrently from multiple threads. NEVER use mutable shared state across event handler calls, like setting instance variables. Sharing state can introduce subtle data races and corruption that may only appear in production. If you must use shared state, use appropriate synchronization primitives like lock, SemaphoreSlim, or concurrent collections.

Ensuring Idempotent Message Handlers

Idempotent message handlers are crucial for maintaining data consistency, fault tolerance, and scalability when working with distributed, event-based systems. They ensure that processing a message multiple times has the same effect as processing it once, which is essential for recovering from failures.

Strategies for achieving idempotence:

1. Natural Idempotence: Use inherently idempotent operations (e.g., setting a value in a key-value store).

2. Deduplication with Unique Identifiers:

• Kafka messages can be uniquely identified by their partition and offset.
• Before processing, check if the message has been handled before.
• Store processed message identifiers with an appropriate TTL.

3. Database Upserts: Use upsert operations for database writes (e.g., MERGE in SQL Server or INSERT ... ON CONFLICT DO UPDATE in PostgreSQL via EF Core).

4. Partition Offset Tracking:

• Store the latest processed offset for each partition.
• Only process messages with higher offsets than the last processed one.
• Critically, store these offsets transactionally with other state updates to ensure consistency.

5. Idempotency Keys for External APIs: Utilize idempotency keys when supported by external APIs.

6. Check-then-Act Pattern:

• For non-idempotent external systems, verify if an operation was previously completed before execution.
• Maintain a record of completed operations, keyed by a unique message identifier.

7. Saga Pattern:

• Implement a state machine in your database for multi-step operations.
• Each message advances the state machine, allowing for idempotent processing and easy failure recovery.
• Particularly useful for complex, distributed transactions across multiple services.

Application shutdown

A Prosody client runs a subscription, timers, and other services in the background. Before an application terminates, it must stop all client services.

The owner of the client calls ShutdownAsync() or DisposeAsync() when the application terminates. Both stop all client services and reject new operations. ShutdownAsync() stops the consumer after in-flight handlers finish, then stops the remaining client services. It has no time limit, so a handler that ignores its CancellationToken delays shutdown. DisposeAsync() calls ShutdownAsync(), flushes telemetry, and releases the native handle. It logs a shutdown failure instead of throwing it. Synchronous Dispose() releases the native handle only and does not stop client services. A component that does not own the client calls UnsubscribeAsync() instead. It stops the subscription, and the client can subscribe again later. You do not need to call UnsubscribeAsync() before ShutdownAsync().

await client.ShutdownAsync();

The example in Dependency Injection shows both. The worker unsubscribes when the host stops, and Program.cs disposes the client after the host exits.

Error Handling

Prosody classifies errors as transient (temporary, can be retried) or permanent (won't be resolved by retrying). By default, all errors are considered transient.

The attribute, marker interface, and classifier apply to message, excise, and timer methods.

Using Attributes

Use the [PermanentError] attribute to classify exceptions that should not be retried:

using Prosody;
using System.Text.Json;

public class MyHandler : IProsodyHandler<MyPayload>
{
    public Task OnExciseAsync(ProsodyContext prosodyContext, ExciseMessage message, CancellationToken cancellationToken) => Task.CompletedTask;

    [PermanentError(typeof(JsonException), typeof(ArgumentException))]
    public async Task OnMessageAsync(ProsodyContext prosodyContext, Message<MyPayload> message, CancellationToken cancellationToken)
    {
        // Your message handling logic here
        // JsonException and ArgumentException will be treated as permanent
        // All other exceptions will be treated as transient (default behavior)
    }

    public Task OnTimerAsync(ProsodyContext prosodyContext, ProsodyTimer timer, CancellationToken cancellationToken) => Task.CompletedTask;
}

Using PermanentException

You can also throw a PermanentException directly:

using Prosody;

public class MyHandler : IProsodyHandler<MyPayload>
{
    public Task OnExciseAsync(ProsodyContext prosodyContext, ExciseMessage message, CancellationToken cancellationToken) => Task.CompletedTask;

    public async Task OnMessageAsync(ProsodyContext prosodyContext, Message<MyPayload> message, CancellationToken cancellationToken)
    {
        var payload = message.Payload;

        if (payload?.Version < MinimumSupportedVersion)
        {
            throw new PermanentException("Message version is no longer supported");
        }

        // Process message...
    }

    public Task OnTimerAsync(ProsodyContext prosodyContext, ProsodyTimer timer, CancellationToken cancellationToken) => Task.CompletedTask;
}

Using IPermanentError Interface

For custom exception types, implement the IPermanentError marker interface:

using Prosody;

public class ValidationException : Exception, IPermanentError
{
    public ValidationException(string message) : base(message) { }
}

Best Practices for Error Handling

• Use permanent errors for issues like malformed data or business logic violations.
• Use transient errors for temporary issues like network problems.
• Be cautious with permanent errors as they prevent retries and can result in data loss.
• Consider system reliability and data consistency when classifying errors.

Handling Task Cancellation

Prosody cancels tasks during partition rebalancing or timeout. During shutdown, handlers run freely for most of the shutdown timeout before the cancellation signal fires — giving in-flight work time to complete. How you handle cancellation is critical:

• A handler that returns normally (no exception) is considered successful — Prosody treats the message as processed.
• Any exception — including OperationCanceledException — signals failure. Prosody does not distinguish cancellation from other errors; all exceptions are classified as transient (or permanent if marked).
• Never silently return on cancellation. If the handler returns without an exception, Prosody assumes the message was fully processed. Swallowing cancellation (e.g., if (cancellationToken.IsCancellationRequested) return;) tells Prosody the message succeeded when it didn't, which can cause data loss.

The correct pattern is to let OperationCanceledException propagate. When Prosody initiates the cancellation (rebalance, timeout, shutdown), it already knows the handler didn't complete — the transient error result simply confirms this. Prosody will not naively retry a message it just cancelled during shutdown; the retry behavior depends on the operating mode and the reason for cancellation.

The library provides a CancellationToken to your handler methods. Pass this token to any async operations that support it to ensure prompt cancellation.

Best practices:

1. Throw, don't swallow. Use ThrowIfCancellationRequested() or pass the token to async APIs that throw on cancellation. Never check IsCancellationRequested and silently return — this breaks the cancellation contract and causes Prosody to treat incomplete work as successful.
2. Exit promptly when cancelled to avoid rebalancing delays.
3. Use try/finally blocks for clean resource handling.
4. Pass the CancellationToken to all async operations that support it.

Example of using CancellationToken in message processing:

public class MyHandler : IProsodyHandler<MyPayload>
{
    private readonly HttpClient _httpClient;
    private readonly MyDbContext _dbContext;
    private readonly ProsodyClient _client;

    public Task OnExciseAsync(ProsodyContext prosodyContext, ExciseMessage message, CancellationToken cancellationToken) => Task.CompletedTask;

    public async Task OnMessageAsync(ProsodyContext prosodyContext, Message<MyPayload> message, CancellationToken cancellationToken)
    {
        // Pass the token to HTTP calls — throws OperationCanceledException on cancellation
        var response = await _httpClient.GetAsync("https://api.example.com", cancellationToken);
        var data = await response.Content.ReadAsStringAsync(cancellationToken);

        // Pass the token to database operations
        await _dbContext.Messages.AddAsync(new MessageEntity { Payload = data }, cancellationToken);
        await _dbContext.SaveChangesAsync(cancellationToken);

        // Send a message, passing the cancellation token
        await _client.SendAsync("topic", "key", new { Data = "value" }, cancellationToken);
    }

    public Task OnTimerAsync(ProsodyContext prosodyContext, ProsodyTimer timer, CancellationToken cancellationToken) => Task.CompletedTask;
}

For CPU-bound loops, poll ThrowIfCancellationRequested() periodically. This throws OperationCanceledException when cancellation is requested, correctly signaling to Prosody that the handler did not complete:

public class MyHandler : IProsodyHandler<List<Item>>
{
    public Task OnExciseAsync(ProsodyContext prosodyContext, ExciseMessage message, CancellationToken cancellationToken) => Task.CompletedTask;

    public async Task OnMessageAsync(ProsodyContext prosodyContext, Message<List<Item>> message, CancellationToken cancellationToken)
    {
        foreach (var item in message.Payload ?? [])
        {
            // Correct: throws OperationCanceledException, signaling incomplete work
            cancellationToken.ThrowIfCancellationRequested();

            ProcessItem(item);
        }
    }

    public Task OnTimerAsync(ProsodyContext prosodyContext, ProsodyTimer timer, CancellationToken cancellationToken) => Task.CompletedTask;
}

Anti-pattern — do not silently return on cancellation:

// WRONG: Prosody sees success and commits the offset, losing the unprocessed message
foreach (var item in items)
{
    if (cancellationToken.IsCancellationRequested)
        return; // Silent return = Prosody thinks the message was fully processed

    ProcessItem(item);
}

Failing to follow these practices can lead to:

• Data loss from incomplete work being marked as successful when cancellation is silently swallowed.
• Slower message processing due to delayed rebalancing.
• Resource leaks if long-running operations aren't properly cancelled.

Logging Configuration

Prosody provides flexible logging integration with your application.

Static Configuration

using Microsoft.Extensions.Logging;
using Prosody;
using Prosody.Logging;

// Configure logging globally for all Prosody clients (must be called once, before creating clients)
var loggerFactory = LoggerFactory.Create(builder => builder.AddConsole());
ProsodyLogging.Configure(loggerFactory);

To reset logging in test fixtures (e.g., during teardown so Configure can be called again):

ProsodyLogging.ResetForTesting();

Dependency Injection

Prefer explicit construction and dependency passing. Pass required dependencies through constructors or method parameters, rather than retrieving them from a service provider inside application code. Create the Prosody client at application startup, await its initialization, and keep disposal with the code that creates it. This makes dependencies, initialization failures, and shutdown order visible. A DI container can pass the existing client to consumers while the application retains ownership.

For ASP.NET Core or Generic Host applications, create the client before you build the host, and register that instance:

var builder = Host.CreateApplicationBuilder(args);
builder.Services.AddProsodyLogging();

var options = builder.Configuration.GetSection("Prosody").Get<ClientOptions>() ?? new ClientOptions();
// Configuration binding cannot set StateCollections or ConfigureJsonOptions. Set them here.
await using var client = await ProsodyClient.CreateAsync(options);
builder.Services.AddSingleton(client);
builder.Services.AddHostedService<Worker>();

await builder.Build().RunAsync();

The container does not dispose an instance that it did not create. The application owns the client, and await using disposes it after the host stops. Register the instance, not a factory. The container disposes the object a factory returns, and the application then no longer controls when the client stops.

A worker receives the client through its constructor. It stops its subscription when the host stops, and it does not dispose the client:

public sealed class Worker(ProsodyClient client) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await client.SubscribeAsync(new MyHandler());

        await Task.Delay(Timeout.InfiniteTimeSpan, stoppingToken)
            .ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);

        await client.UnsubscribeAsync();
    }
}

Without UnsubscribeAsync, the subscription stays active until the final DisposeAsync, and handlers can run after the host has disposed the services they use.

AddProsodyLogging configures the log sink when the host starts and clears it when the host stops. Prosody discards native log events from CreateAsync and from the final DisposeAsync. To keep them, call ProsodyLogging.Configure with your own ILoggerFactory before CreateAsync, and omit AddProsodyLogging.

AddProsodyClient is the alternative. It registers ProsodyClientProvider, which creates and owns its own client. Do not call both. Inject the provider into hosted services and call GetAsync to get the shared client. The provider disposes the client when the host stops. A failed GetAsync call does not poison the provider. A later call retries client construction. Use asynchronous host disposal when possible. Synchronous disposal starts client shutdown without blocking.

Log messages are emitted under the Prosody.Native category.

Error Monitoring (Sentry)

Prosody automatically reports handler exceptions to Sentry when the host application has Sentry initialized. Prosody never calls SentrySdk.Init — it only enriches an already-initialized Sentry instance.

Setup

Initialize Sentry in your host application before subscribing to messages:

SentrySdk.Init(o =>
{
    o.Dsn = "https://examplePublicKey@o0.ingest.sentry.io/0";
    o.Environment = "production";
    o.Release = "my-app@1.2.3";
});

Or with ASP.NET Core / Generic Host:

builder.WebHost.UseSentry("https://examplePublicKey@o0.ingest.sentry.io/0");

If Sentry is not initialized, Prosody silently skips error reporting with zero overhead.

How It Works

Prosody checks SentrySdk.IsEnabled on each handler failure. If the host has Sentry initialized, Prosody captures the exception and enriches it with handler context. Prosody never owns the Sentry lifecycle — initialization and disposal remain entirely in the host application.

What Gets Reported

Both transient and permanent handler exceptions are captured with contextual data:

• prosody.event_type tag: "message" or "timer"
• prosody.error_class tag: "permanent" or "transient"
• prosody context:
  • For messages: topic, key, partition, offset
  • For timers: key, fire time

Safety Guarantee

Sentry failures never affect message processing. If Sentry is unreachable or misconfigured, the exception is logged and handler results are unchanged.

Note: The Sentry package is currently a hard dependency of ProsodyEvents.Prosody. A future improvement is to extract Sentry support into a separate ProsodyEvents.Prosody.Sentry package so consumers who don't use Sentry don't pull in the dependency.

Administrative Operations

⚠️ Important Note: Topic management in production environments should typically be handled through GitOps using Strimzi KafkaTopic manifests. The AdminClient is provided for testing scenarios and specific cases where manual topic creation and deletion is required.

AdminClient

The AdminClient provides administrative operations for Kafka topics:

using Prosody;

// Initialize admin client
using var admin = new AdminClient("localhost:9092");

// Create a topic for testing
await admin.CreateTopicAsync(
    name: "test-topic",
    partitionCount: 4,
    replicationFactor: 1
);

// Create a compacted topic that keeps messages for one day
await admin.CreateTopicAsync(
    name: "state-topic",
    partitionCount: 4,
    replicationFactor: 1,
    cleanupPolicy: "compact",
    retention: TimeSpan.FromDays(1)
);

// Delete a topic
await admin.DeleteTopicAsync("test-topic");

cleanupPolicy sets cleanup.policy, such as delete, compact, or delete,compact. retention sets retention.ms. When you omit them, the topic uses the cluster defaults.

Configuration Parameters

The AdminClient constructor accepts:

• bootstrapServers (params string[]): Kafka bootstrap servers (required)

Or via environment variable:

PROSODY_BOOTSTRAP_SERVERS=localhost:9092  # Single server
PROSODY_BOOTSTRAP_SERVERS=localhost:9092,localhost:9093  # Multiple servers

Release Process

Prosody uses an automated release process managed by GitHub Actions. Here's an overview of how releases are handled:

1. Trigger: The release process is triggered automatically on pushes to the main branch.

2. Release Please: The process starts with the "Release Please" action, which:

   • Analyzes commit messages since the last release.
   • Creates or updates a release pull request with changelog updates and version bumps.
   • When the PR is merged, it creates a GitHub release and a git tag.

3. Build Process: If a new release is created, the following native build jobs are triggered:

   • Linux builds for x86_64 and aarch64 architectures.
   • Windows builds for x64 and arm64 architectures.
   • macOS builds for arm64 (Apple Silicon) architecture.

4. Pack: A single NuGet package (ProsodyEvents.Prosody) is assembled from the native artifacts and the generated C# bindings, bundling all supported runtimes under runtimes/<rid>/native/ inside the .nupkg.

5. Test: The packed .nupkg is consumed by the test project (via TestPackage=true) and run against Kafka and Cassandra on each supported RID / target framework combination (.NET 8, 9, 10) before publication.

6. Publication: If all tests pass, the package is published to nuget.org.

Contributing to Releases

To contribute to a release:

1. Make your changes in a feature branch.
2. Use Conventional Commits syntax for your commit messages. This helps Release Please determine the next version number and generate the changelog.
3. Create a pull request to merge your changes into the main branch.
4. Once your PR is approved and merged, Release Please will include your changes in the next release PR.

Manual Releases

While the process is automated, manual intervention may sometimes be necessary:

• You can manually trigger the release workflow from the GitHub Actions tab if needed (including the release_as input to force a specific version, e.g. 2.2.0-beta.1).
• If you need to make changes to the release PR created by Release Please, you can do so before merging it.

All releases are automatically published to nuget.org. Ensure you have thoroughly tested your changes before merging to main.

API Reference

ProsodyClientBuilder

Fluent builder for configuring and creating a ProsodyClient. All With* methods return the builder for chaining.

• static ProsodyClientBuilder Create(): Creates a new builder instance.

Builder Methods:

• WithBootstrapServers(params string[] servers): Set Kafka bootstrap servers
• WithGroupId(string groupId): Set consumer group ID
• WithSubscribedTopics(params string[] topics): Set topics to subscribe to
• WithMode(ClientMode mode): Set client operating mode
• WithAllowedEvents(params string[] prefixes): Set event type prefixes to allow
• WithSourceSystem(string sourceSystem): Set source system identifier
• WithMock(bool mock): Enable/disable in-memory mock client
• ForPipeline(): Select pipeline mode.
• ForLowLatency(string failureTopic): Select low-latency mode and its failure topic.
• ForBestEffort(): Select best-effort mode.
• WithMaxConcurrency(uint maxConcurrency): Set max concurrent messages
• WithMaxRetries(uint maxRetries): Set max retry attempts
• WithFailureTopic(string topic): Set dead letter topic
• WithProbePort(ushort port): Set health check probe port
• WithSendTimeout(TimeSpan timeout): Set max time to wait for message delivery
• Configure(Action<ClientOptions> configure): Set any option on ClientOptions directly
• ConfigureJsonOptions(Action<JsonSerializerOptions> configure): Override JSON serialization options (runs after defaults are applied)
• WithStateCollections(params StateDefinition[] definitions): Register keyed-state collections before subscribe

Build:

• Task<ProsodyClient> BuildAsync(): Validates configuration and creates a client asynchronously.

ProsodyClient

• Task<ProsodyClient> ProsodyClient.CreateAsync(ClientOptions options): Create a client asynchronously.
• string SourceSystem { get; }: Get the source system identifier configured for the client.
• Task<ConsumerState> GetConsumerStateAsync(): Get the current state of the consumer.
• Task<uint> AssignedPartitionCountAsync(): Get the number of partitions currently assigned to this consumer.
• Task<bool> IsStalledAsync(): Check if the consumer has stalled partitions.
• Task<PublishedValue<T>> StateAsync<T>(string subsystem, ValueStateDefinition<T> definition, CancellationToken cancellationToken = default): Open a read-only published value.
• Task<PublishedMap<TValue>> StateAsync<TValue>(string subsystem, MapStateDefinition<TValue> definition, CancellationToken cancellationToken = default): Open a read-only published map.
• Task<PublishedDeque<T>> StateAsync<T>(string subsystem, DequeStateDefinition<T> definition, CancellationToken cancellationToken = default): Open a read-only published deque.
• Task<PublishedSet> StateAsync(string subsystem, SetStateDefinition definition, CancellationToken cancellationToken = default): Open a read-only published set.
• Task SendAsync<T>(string topic, string key, T payload, CancellationToken cancellationToken = default): Send with the configured JsonSerializerOptions.
• Task ExciseAsync(string topic, string key, CancellationToken cancellationToken = default): Send an excise record for a key.
• Task SendAsync<T>(string topic, string key, T payload, JsonTypeInfo<T> typeInfo, CancellationToken cancellationToken = default): Send with supplied JSON metadata. This overload supports trimming.
• Task SendAsync<T>(string topic, string key, T payload, JsonTypeInfo<T> typeInfo, SendOptions options, CancellationToken cancellationToken = default): Override event metadata during a trim-safe send.
• Task<IReadOnlyDictionary<string, Outcome<TResponse>>> RequestAsync<TPayload, TResponse>(...): Return one outcome for each subsystem.
• Task<IReadOnlyDictionary<string, Outcome<TResponse>>> RequestAsync<TPayload, TResponse>(..., JsonTypeInfo<TPayload>, JsonTypeInfo<TResponse>, ...): Return outcomes in trimmed applications.
• Task<IReadOnlyDictionary<string, Outcome<TResponse>>> RequestAsync<TPayload, TResponse>(..., JsonTypeInfo<TPayload>, JsonTypeInfo<TResponse>, IReadOnlyList<string> subsystems, TimeSpan timeout, SendOptions options, ...): Override event metadata during a trim-safe request.
• Task<IReadOnlyDictionary<string, Outcome<TResponse>>> RequestExciseAsync<TResponse>(...): Return one excise outcome for each subsystem.
• Task<IReadOnlyDictionary<string, Outcome<TResponse>>> RequestExciseAsync<TResponse>(..., JsonTypeInfo<TResponse>, ...): Return excise outcomes in trimmed applications.
• Task SubscribeAsync<T>(IProsodyHandler<T> handler): Start event processing with a typed payload handler.
• Task SubscribeAsync<T>(IProsodyHandler<T> handler, IPermanentErrorClassifier classifier): Classify errors without reflection. Use this overload in trimmed applications.
• Task SubscribeAsync<TPayload, TResponse>(IProsodyRequestHandler<TPayload, TResponse> handler): Subscribe with typed request responses.
• Task SubscribeAsync<TPayload, TResponse>(IProsodyRequestHandler<TPayload, TResponse> handler, IPermanentErrorClassifier classifier): Use explicit request-handler error classification.
• Task UnsubscribeAsync(): Stop the consumer. You can subscribe again later.
• Task ShutdownAsync(): Stop all client services. Concurrent and repeated calls await the same operation.
• void Dispose(): Release resources immediately. It does not wait for shutdown. Use ShutdownAsync or DisposeAsync to stop client services.
• ValueTask DisposeAsync(): Shut down and dispose of client resources. Enables await using.

AdminClient

• AdminClient(params string[] bootstrapServers): Initialize a new AdminClient with the given configuration.
• Task CreateTopicAsync(string name, ushort partitionCount, ushort replicationFactor, string? cleanupPolicy = null, TimeSpan? retention = null): Create a Kafka topic.
• Task DeleteTopicAsync(string name): Delete an existing Kafka topic.
• void Dispose(): Dispose of admin client resources.

IProsodyHandler<TPayload>

Interface for handling messages and timers:

Implement all three methods. The compiler rejects an incomplete handler before subscription.

public interface IProsodyHandler<TPayload>
{
    Task OnMessageAsync(ProsodyContext prosodyContext, Message<TPayload> message, CancellationToken cancellationToken);
    Task OnExciseAsync(ProsodyContext prosodyContext, ExciseMessage message, CancellationToken cancellationToken);
    Task OnTimerAsync(ProsodyContext prosodyContext, ProsodyTimer timer, CancellationToken cancellationToken);
}

IProsodyRequestHandler<TPayload, TResponse> has the same methods. Its message and excise methods return Task<TResponse>.

Message<T>

Represents a Kafka message with the following properties:

• Topic (string): The name of the topic.
• Partition (int): The partition number.
• Offset (long): The message offset within the partition.
• Timestamp (DateTimeOffset): The timestamp when the message was created or sent.
• Key (string): The message key.
• T? Payload: The deserialized payload (deserialized once before the handler is invoked).
• SourceSystem (string?): The source system that produced the message, or null when its headers name none.
• IsResponseRequested (bool): true when the sender waits for a response. Skip the work that only a response needs when it is false.

ExciseMessage

An ExciseMessage has Topic, Partition, Offset, Timestamp, Key, SourceSystem, and IsResponseRequested properties. It has no Payload property.

ProsodyContext

Represents the current event context:

• bool ShouldCancel { get; }: Check if cancellation has been requested (includes timeout and shutdown).
• Task OnCancelAsync(): Returns a task that completes when cancellation is signaled.
• Demand Demand { get; }: The demand this call serves. Demand.Kind is DemandKind.Normal for a first attempt or DemandKind.Failure for a retry. Demand.RetryAttempt is the retry count: 0 for normal demand and 1 on the first retry. The count is an estimate. Keep an exact attempt count in keyed state if you need one.

Keyed-state binding:

• State(definition): Bind a registered collection for the current attempt. Message definitions contain Message<TPayload>. An unregistered or mismatched definition throws PermanentStateException. See Keyed State.

Timer scheduling methods:

• Task ScheduleAsync(DateTimeOffset time): Schedules a timer to fire at the specified time
• Task ClearAndScheduleAsync(DateTimeOffset time): Clears all timers and schedules a new one
• Task UnscheduleAsync(DateTimeOffset time): Removes a timer scheduled for the specified time
• Task ClearScheduledAsync(): Removes all scheduled timers
• Task<DateTimeOffset[]> ScheduledAsync(): Returns an array of all scheduled timer times

ProsodyTimer

Represents a timer that has fired, provided to the OnTimerAsync method:

• Key (string): The entity key identifying what this timer belongs to
• Time (DateTimeOffset): The time when this timer was scheduled to fire

ConsumerState

Enum representing the consumer lifecycle state:

• Unconfigured: Consumer has not been configured
• Configured: Consumer is configured but not running
• Running: Consumer is actively processing messages
• Shutdown: Client is shut down.

ClientMode

Enum representing the operating mode:

• Pipeline: Default mode, retry indefinitely with defer and monopolization detection
• LowLatency: Few retries then dead letter (requires FailureTopic)
• BestEffort: Log failures, no retries

SpanRelation selects Child or FollowsFrom for message and timer spans.

Requests

• Outcome<T>: A Success<T> or Failure<T> result for one subsystem.
• Success<T>: Contains the response in Value. A JSON null response is a success with a null value, so use a nullable T, such as RequestAsync<Order, Order?>, when a subsystem can respond with null.
• Failure<T>: Contains a ResponseError in Error.
• ResponseError: Base record with a Message property.
• HandlerError, TimeoutError, FormatMismatchError, and MalformedResponseError: The possible response errors.
• SendOptions: Optionally overrides EventId and EventType for a trim-safe send or request.

Keyed State

Definition factories (each returns an immutable, validated record used both in WithStateCollections(...) and with context.State(...)):

• StateDefinition.Value<T>(string name, TimeSpan? ttl = null, bool readUncommitted = false, bool published = false, StateReadCache? readCache = null) → ValueStateDefinition<T>
• StateDefinition.Map<TValue>(string name, TimeSpan? ttl = null, bool readUncommitted = false, int? keysetLimit = null, bool published = false, StateReadCache? readCache = null) → MapStateDefinition<TValue>
• StateDefinition.Deque<T>(string name, TimeSpan? ttl = null, bool readUncommitted = false, int? capacity = null, bool published = false, StateReadCache? readCache = null) → DequeStateDefinition<T>
• StateDefinition.Set(string name, TimeSpan? ttl = null, bool readUncommitted = false, int? keysetLimit = null, bool published = false, StateReadCache? readCache = null) → SetStateDefinition
• StateDefinition.MessageValue<TPayload>(string name, TimeSpan? ttl = null, bool readUncommitted = false) → MessageValueDefinition<TPayload>
• StateDefinition.MessageMap<TPayload>(string name, TimeSpan? ttl = null, bool readUncommitted = false, int? keysetLimit = null) → MessageMapDefinition<TPayload>
• StateDefinition.MessageDeque<TPayload>(string name, TimeSpan? ttl = null, bool readUncommitted = false, int? capacity = null) → MessageDequeDefinition<TPayload>

Each definition exposes its validated Name.

The item type parameter (T / TValue) uses notnull on JSON collections. Thus, a nullable item type causes a compile-time error. Message collections use Message<TPayload>. Its payload can be null when TPayload permits a JSON null.

Published JSON and set collections use the same definition for owned and read-only access. See Published state for setup and examples. PublishedMap<TValue> provides GetAsync, batched GetManyAsync, ContainsKeyAsync, batched ContainsManyAsync, IsEmptyAsync, EnumerateAsync, key-only EnumerateKeysAsync, and EnumerateValuesAsync. PublishedSet provides ContainsAsync, batched ContainsManyAsync, IsEmptyAsync, and EnumerateAsync. PublishedDeque<T> provides GetAsync (an Index such as ^1 counts from the back), CountAsync, IsEmptyAsync, PeekFrontAsync, PeekBackAsync, and EnumerateAsync. Each enumeration accepts a ScanDirection or a query.

IValueState<T> where T : notnull:

• Task<StateValue<T>> GetAsync(CancellationToken cancellationToken = default)
• Task SetAsync(T value, CancellationToken cancellationToken = default)
• Task ClearAsync(CancellationToken cancellationToken = default)
• Task<StoreOutcome> CommitAsync(CancellationToken cancellationToken = default)
• Task<StoreOutcome> RollbackAsync(CancellationToken cancellationToken = default)

IMapState<TValue> : IAsyncEnumerable<KeyValuePair<string, TValue>> (keys are string, TValue : notnull):

• Task<StateValue<TValue>> GetAsync(string key, CancellationToken cancellationToken = default)
• Task<IReadOnlyList<StateValue<TValue>>> GetManyAsync(IEnumerable<string> keys, CancellationToken cancellationToken = default)
• Task<bool> ContainsKeyAsync(string key, CancellationToken cancellationToken = default)
• Task<IReadOnlyList<bool>> ContainsManyAsync(IEnumerable<string> keys, CancellationToken cancellationToken = default)
• Task<bool> IsEmptyAsync(CancellationToken cancellationToken = default)
• Task SetAsync(string key, TValue value, CancellationToken cancellationToken = default)
• Task RemoveAsync(string key, CancellationToken cancellationToken = default)
• Task ClearAsync(CancellationToken cancellationToken = default)
• IAsyncEnumerable<KeyValuePair<string, TValue>> EnumerateAsync(ScanDirection direction = ScanDirection.Forward, CancellationToken cancellationToken = default)
• IAsyncEnumerable<KeyValuePair<string, TValue>> EnumerateAsync(KeyQuery query, CancellationToken cancellationToken = default)
• IAsyncEnumerable<string> EnumerateKeysAsync(ScanDirection direction = ScanDirection.Forward, CancellationToken cancellationToken = default)
• IAsyncEnumerable<string> EnumerateKeysAsync(KeyQuery query, CancellationToken cancellationToken = default)
• IAsyncEnumerable<TValue> EnumerateValuesAsync(ScanDirection direction = ScanDirection.Forward, CancellationToken cancellationToken = default)
• IAsyncEnumerable<TValue> EnumerateValuesAsync(KeyQuery query, CancellationToken cancellationToken = default)
• Task<StoreOutcome> CommitAsync(CancellationToken cancellationToken = default)
• Task<StoreOutcome> RollbackAsync(CancellationToken cancellationToken = default)

IDequeState<T> : IAsyncEnumerable<T> (T : notnull):

• Task PushBackAsync(T value, CancellationToken cancellationToken = default)
• Task PushFrontAsync(T value, CancellationToken cancellationToken = default)
• Task<StateValue<T>> PopFrontAsync(CancellationToken cancellationToken = default)
• Task<StateValue<T>> PopBackAsync(CancellationToken cancellationToken = default)
• Task<StateValue<T>> PeekFrontAsync(CancellationToken cancellationToken = default)
• Task<StateValue<T>> PeekBackAsync(CancellationToken cancellationToken = default)
• Task<StateValue<T>> GetAsync(int index, CancellationToken cancellationToken = default): A negative index throws ArgumentOutOfRangeException.
• Task<StateValue<T>> GetAsync(Index index, CancellationToken cancellationToken = default): GetAsync(^1) reads the back element.
• Task<int> CountAsync(CancellationToken cancellationToken = default)
• Task<bool> IsEmptyAsync(CancellationToken cancellationToken = default)
• IAsyncEnumerable<T> EnumerateAsync(ScanDirection direction = ScanDirection.Forward, CancellationToken cancellationToken = default)
• IAsyncEnumerable<T> EnumerateAsync(PositionQuery query, CancellationToken cancellationToken = default)
• Task<StoreOutcome> CommitAsync(CancellationToken cancellationToken = default)
• Task<StoreOutcome> RollbackAsync(CancellationToken cancellationToken = default)

ISetState : IAsyncEnumerable<string>:

• Task AddAsync(string member, CancellationToken cancellationToken = default)
• Task RemoveAsync(string member, CancellationToken cancellationToken = default): An absent member is a no-op.
• Task<bool> ContainsAsync(string member, CancellationToken cancellationToken = default)
• Task<IReadOnlyList<bool>> ContainsManyAsync(IEnumerable<string> members, CancellationToken cancellationToken = default)
• Task<bool> IsEmptyAsync(CancellationToken cancellationToken = default)
• Task ClearAsync(CancellationToken cancellationToken = default)
• IAsyncEnumerable<string> EnumerateAsync(ScanDirection direction = ScanDirection.Forward, CancellationToken cancellationToken = default)
• IAsyncEnumerable<string> EnumerateAsync(KeyQuery query, CancellationToken cancellationToken = default)
• Task<StoreOutcome> CommitAsync(CancellationToken cancellationToken = default)
• Task<StoreOutcome> RollbackAsync(CancellationToken cancellationToken = default)

KeyQuery (a sealed record with init properties): ScanDirection Direction, string? Prefix, string? From, string? After, string? To, string? Before, KeyRange? Range, int? Limit. A non-positive Limit throws ArgumentOutOfRangeException. Both From and After, or both To and Before, throw ArgumentException when the query is used.

KeyRange(string? Start, string? End) (a readonly record struct): an ascending key range that includes Start and excludes End. A null side is open.

PositionQuery (a sealed record with init properties): ScanDirection Direction, int? From, int? After, int? To, int? Before, Range? Range, int? Limit. A negative position, a descending range, or a from-end index other than an end of ^0 throws ArgumentOutOfRangeException. The pairs follow the KeyQuery rules.

StoreOutcome: Applied (buffered changes were written or discarded) or NoOp (nothing was buffered).

StateValue<T> (a readonly struct optional read result, T : notnull):

• bool HasValue { get; }: whether a value is present.
• T Value { get; }: the stored value; throws InvalidOperationException when absent.
• T GetValueOrDefault(T defaultValue): the stored value when present, otherwise defaultValue (mirrors Nullable<T>).
• bool TryGetValue(out T value): the familiar Try pattern — true with the value when present, false otherwise.

ScanDirection: Forward (ascending) or Backward (descending).

StateReadCache.Disabled disables published-read caching. StateReadCache.For(ttl) creates a cache setting with the specified TTL.

Errors:

• StateException: abstract base; exposes StateErrorCategory Category { get; }.
• TransientStateException : StateException: Reports a keyed-state error that Prosody can retry.
• PermanentStateException : StateException, IPermanentError: Reports a keyed-state error that another attempt cannot resolve.
• StateErrorCategory: Permanent or Transient.

Client errors:

• ProsodyException: Reports a broker or runtime failure. A handler that rethrows it retries the event.
• A caller mistake throws ArgumentException or InvalidOperationException. A second SubscribeAsync is an example.

Handler error classification:

• IPermanentError: Marks an exception as permanent.
• PermanentException: A permanent handler exception.
• PermanentErrorAttribute: Classifies selected exception types as permanent for one handler method.
• IPermanentErrorClassifier: Classifies handler exceptions without reflection.

IPermanentErrorClassifier provides IsMessageErrorPermanent, IsExciseErrorPermanent, and IsTimerErrorPermanent.

Configuration and dependency injection

• ClientOptions: Contains the client settings in Configuration.
• Prosody.CreateClient(): Create a ProsodyClientBuilder.
• ProsodyServiceCollectionExtensions.AddProsodyClient(...): Register ProsodyClientProvider with dependency injection. To inject ProsodyClient directly, see Dependency Injection.
• ProsodyServiceCollectionExtensions.AddProsodyLogging(): Register Prosody logging.
• ProsodyClientProvider.GetAsync(): Get the shared client.
• ProsodyClientProvider.Dispose() and DisposeAsync(): Dispose the shared client.

Logging and telemetry

• ProsodyLogging.Configure(loggerFactory): Connect Prosody to Microsoft logging.
• ProsodyLogging.FlushTelemetry(): Export pending telemetry.
• ProsodyLogging.ShutdownTelemetry(): Export pending telemetry and stop its providers.

License

MIT