Kanject.Core.Channels 3.14.0

Kanject.Core.Channels

One handler signature, more than one hosting mode.

Kanject.Core.Channels is the runtime for Kanject channels: a bounded, batching drain loop over System.Threading.Channels with time-based flush, per-batch retry, restart-on-fault, deadline-aware cancellation, and projection of un-acknowledged message IDs for partial-batch-failure responses. Write a (ReadOnlyMemory<T> batch, CancellationToken ct) → ValueTask handler once and run it as a long-lived BackgroundService consumer or inside a single AWS Lambda invocation.

Pair it with the Kanject.Core.Channels.Annotations source generator: annotate a method with [Channel] and the typed writer, DI registration, and host-appropriate consumer are generated at compile time. The engine is also usable directly, without the generator.

Installation

dotnet add package Kanject.Core.Channels

To use the [Channel] attribute and the generated wiring, add the generator next to the runtime:

<ItemGroup>
  <PackageReference Include="Kanject.Core.Channels" />
  <PackageReference Include="Kanject.Core.Channels.Annotations" PrivateAssets="all" />
</ItemGroup>

dotnet add package Kanject.Core.Channels.Annotations writes an IncludeAssets line for that package that omits compile. Delete that line (keep PrivateAssets="all") — otherwise the attribute types from Kanject.Core.Channels.Annotations.Attributes can be stripped from compilation.

Kanject.Core.Channels already depends on Kanject.Core.Channels.Annotations.Attributes, so the [Channel] attribute and its enums are available as soon as the runtime is referenced; the generator package is what turns them into code.

Targets .NET 8, .NET 9 and .NET 10. The runtime is Native AOT / trimming compatible (no reflection in the drain loop).

Quick start: in-process consumer

using Kanject.Core.Channels.Annotations.Attributes;

namespace Shop.Orders;

public sealed record OrderPlaced(string OrderId, decimal Total);

public interface IOrderReadModel
{
    Task UpsertAsync(IReadOnlyList<OrderPlaced> orders, CancellationToken ct);
}

// The containing type must be partial (KANCHN002). The handler is an ordinary method with a body.
public sealed partial class OrderProjector(IOrderReadModel readModel)
{
    [Channel(Capacity = 5000, BatchSize = 25, FlushIntervalMs = 200)]
    public async ValueTask ProjectAsync(ReadOnlyMemory<OrderPlaced> batch, CancellationToken ct)
    {
        await readModel.UpsertAsync(batch.ToArray(), ct);
    }
}

Register it from the composition root with the generated extension:

using Shop.Orders;

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddSingleton<IOrderReadModel, SqlOrderReadModel>();
builder.Services.AddOrderProjectorProjectAsyncChannel();   // generated

var app = builder.Build();
app.Run();

Produce into the channel through the generated writer interface:

public sealed class CheckoutService(IOrderProjectorProjectAsyncChannelWriter orders)
{
    public ValueTask OnOrderPlacedAsync(OrderPlaced order, CancellationToken ct) =>
        orders.WriteAsync(order, ct);
}

Items are dispatched to ProjectAsync in batches of up to 25; a partial batch is flushed after 200 ms. With the default Lifetime = ChannelLifetime.Scoped, each batch runs in a fresh DI scope with a freshly resolved OrderProjector.

The generated files fully qualify every type and DI extension call, so they compile without any using in your project.

What the generator emits

For a [Channel] method M on type T:

All generated types go in the handler's namespace. The DI extension:

1. Builds a ChannelOptions from the attribute values, then applies your configure callback. ChannelFullMode maps to the BoundedChannelFullMode member of the same name.
2. Creates the bounded channel with ChannelDrainer.CreateChannel<TItem>(options).
3. Registers ChannelOptions, Channel<TItem>, ChannelReader<TItem> and ChannelWriter<TItem> as keyed singletons under T{M}ChannelServiceCollectionExtensions.ServiceKey (the handler's fully qualified Namespace.Type.Method), and the writer interface as a singleton bound to this channel. Nothing is registered unkeyed, so any number of channels — including several over the same item type — coexist in one container.
4. In InProcess mode only, also registers the hosted consumer, built over this channel's reader and options, and the handler type (with the lifetime from Lifetime).

To use the raw channel types, resolve them by key:

var writer = provider.GetRequiredKeyedService<ChannelWriter<OrderPlaced>>(
    OrderProjectorProjectAsyncChannelServiceCollectionExtensions.ServiceKey);

Each AddT{M}Channel may be called once per IServiceCollection; a second call throws InvalidOperationException.

Handler shape

• Return Task or ValueTask.
• The partition is the first parameter that isn't a CancellationToken: the item (TItem) when BatchSize is 1, or a batch (ReadOnlyMemory<TItem> or IReadOnlyList<TItem>) when BatchSize > 1. Prefer ReadOnlyMemory<TItem>: it is passed through without copying. Both generated consumers copy each batch into an array for IReadOnlyList<TItem> handlers.
• A CancellationToken is optional and may appear in any position. The generated consumers bind arguments by name, so any other parameter must be optional (or params) and keeps its default; ref / out / in parameters are not supported.
• Use a non-generic, public or internal instance method on a partial, non-generic type. Static handlers are accepted (with warning KANCHN015) but are invoked directly, without DI.
• Every shape the generator can't emit is a compile-time error (see Diagnostics); it never silently generates nothing.

Hosting modes

ChannelMode is resolved when the project compiles, not at runtime.

Note that the DI helper registers the in-process writer in every mode. In Distributed mode, register your own implementation of the writer interface after calling the helper so it takes precedence.

Lambda: invocation-local draining

InvocationLocalDrainer arms a linked cancellation token that fires at remainingTime − ChannelOptions.SafetyMargin, drains the channel, and returns a ChannelDrainResult whose UnAckedMessageIds you project into the trigger's partial-batch-failure response. On a soft deadline, the in-flight batch and everything still buffered in the channel are reported as un-acked.

Rules enforced by ChannelOptions.Validate(isInvocationLocal: true), which the drainer calls on entry (it throws ArgumentException):

• FullMode must not be BoundedChannelFullMode.Wait — a producer blocked on a full channel would burn the invocation budget. Size Capacity to the record count and use a drop mode.
• ConsumerConcurrency must be 1.

Because the attribute default is FullMode = Wait, set a drop mode for invocation-local channels, for example through the generated helper's configure callback.

Per-invocation channel (recommended)

The generated DI helper registers one process-wide channel, and completing its writer closes it for the lifetime of the process — a problem for warm Lambda invocations. For Lambda entry points, create a channel per invocation and call the engine directly:

using System.Text.Json;
using System.Threading.Channels;
using Amazon.Lambda.Core;
using Amazon.Lambda.SQSEvents;
using Kanject.Core.Channels.Engine;
using Kanject.Core.Channels.Engine.Models;

public readonly record struct QueuedOrder(string MessageId, OrderPlaced Order);

public sealed class OrderFunction(OrderBatchHandler handler)
{
    public async Task<SQSBatchResponse> HandleAsync(SQSEvent sqsEvent, ILambdaContext context)
    {
        var options = new ChannelOptions
        {
            Capacity = Math.Max(1, sqsEvent.Records.Count),
            BatchSize = 25,
            FullMode = BoundedChannelFullMode.DropWrite, // Wait is rejected for invocation-local drains
            SingleWriter = true,
        };

        var channel = ChannelDrainer.CreateChannel<QueuedOrder>(options);
        foreach (var record in sqsEvent.Records)
        {
            // Deserialize with your own JsonSerializerContext for trimming / Native AOT.
            var order = JsonSerializer.Deserialize(record.Body, AppJsonContext.Default.OrderPlaced)!;
            channel.Writer.TryWrite(new QueuedOrder(record.MessageId, order));
        }
        channel.Writer.Complete();

        ChannelDrainResult result = await InvocationLocalDrainer.DrainBatchedAsync(
            channel.Reader,
            (batch, ct) => handler.HandleAsync(batch, ct),   // ValueTask HandleAsync(ReadOnlyMemory<QueuedOrder>, CancellationToken)
            options,
            context.RemainingTime,
            CancellationToken.None,
            messageIdSelector: item => item.MessageId);

        return new SQSBatchResponse
        {
            BatchItemFailures = result.UnAckedMessageIds
                .Select(id => new SQSBatchResponse.BatchItemFailure { ItemIdentifier = id })
                .ToList(),
        };
    }
}

The commercial Kanject.Core.Channels.Provider.AwsSqs and Kanject.Core.Channels.Provider.AwsKinesis packages package this loop for SQS and Kinesis events (including the response projection) and ship generators that emit the Lambda entry point.

Generated DrainAsync helper

In InvocationLocal mode the generator emits T{M}ChannelDrain.DrainAsync(serviceProvider, remainingTime, cancellationToken). It resolves this channel's keyed ChannelReader<TItem> and ChannelOptions and the handler type from serviceProvider, then delegates to InvocationLocalDrainer. Things to know:

• The DI helper does not register the handler type in this mode — register it yourself.
• No typed message-ID selector is passed; set ChannelOptions.MessageIdSelector in configure if you need UnAckedMessageIds.
• The channel is the process-wide singleton described above.

Lambda timeout behaviour

1. Soft deadline. The drainer checks the token between batches (batched handlers) or between items (per-item handlers). A batch is either executed or reported un-acked as a whole — never torn.
2. Partial-batch failures. Un-acked IDs go into the trigger's failure response so only those records are redelivered. This requires ReportBatchItemFailures on the event source mapping.
3. Hard kill. If the margin was too small and the runtime kills the invocation, un-acked records are redelivered after the visibility timeout: at-least-once delivery, at the cost of duplicates.

Using the engine directly

drainBufferedOnCancel pulls items still buffered in the channel into the un-acked set when cancellation fires. Only enable it when the producer has finished writing (the invocation-local drainers set it); a hosted consumer running alongside a live writer must leave it false.

Partial batch success

A batched handler can report which items failed instead of failing the whole batch:

ChannelDrainResult result = await ChannelDrainer.RunBatchedWithResultAsync(
    channel.Reader,
    async (batch, ct) =>
    {
        var failed = new List<int>();
        for (var i = 0; i < batch.Length; i++)
        {
            if (!await TryShipAsync(batch.Span[i], ct))
                failed.Add(i);
        }
        return failed.Count == 0 ? BatchResult.Success : BatchResult.WithFailures(failed);
    },
    options,
    cancellationToken,
    messageIdSelector: item => item.MessageId);

Indices refer to positions in the batch the handler received. Returned failures are terminal: they are projected into UnAckedMessageIds but never retried. To have a batch retried, throw — RetryCount retries cover unhandled exceptions only, with RetryBackoff × 2^(attempt−1) backoff capped at MaxRestartBackoff. BatchResult.Success (and default) are allocation-free.

Contracts at a glance

Cancellation is all-or-nothing within a batch. If the handler observes the token and exits — by throwing OperationCanceledException or returning — the whole batch is treated as un-acked. There is no partial ack within a cancelled batch; handlers that need per-item ack semantics should use BatchSize = 1 or return a BatchResult.

Cancellation means the drain's own token. An OperationCanceledException ends the drain pass (un-acked batch, SoftDeadlineHit) only when the token the drainer was given is cancelled — the host stopping, or the Lambda soft deadline. An OperationCanceledException your handler throws for its own reasons, such as the TaskCanceledException from an HttpClient timeout, is an ordinary fault: it is retried per RetryCount, then reported through OnBatchFault and un-acked, and the drain carries on. The hosted consumer keeps running.

A HandlerTimeout expiry is a fault, not a soft deadline. The timed-out batch is counted as failed and un-acked, reported with a TimeoutException, and the drain moves on to the next batch. It is never retried: the abandoned invocation is still running, so a retry would process the batch twice at once. The abandoned invocation keeps its batch memory — the drainer never refills or re-pools it — but it does keep running, so handlers should still honour their token.

Scope boundary per mode.

Ordering. Items written by one producer are dispatched in write order within a channel. There is no ordering between concurrent producers.

Faults. When a batch or item fails (after retries, or on a HandlerTimeout expiry), its IDs are added to the un-acked set and the loop continues while RestartOnFault is true. With RestartOnFault = false the exception propagates: the hosted consumer stops, and the invocation-local drainers throw.

Flushing. The first item of a batch is awaited indefinitely; after that the batch fills until BatchSize items, the channel completes, or FlushInterval elapses. With FlushInterval = TimeSpan.Zero (the default), a partial batch waits until it fills or the writer completes — set FlushIntervalMs for slow producers.

ChannelOptions reference

Observability

The drain loop logs soft deadlines, handler timeouts, and faults through Kanject's PrintInConsole pipeline under the tag Kanject.Channels. For metrics, implement IChannelMetrics and assign it to ChannelOptions.Metrics:

Exceptions thrown by the sink are swallowed; observability never breaks the drain loop.

Current limitations

• [ChannelProducer] is declared in Kanject.Core.Channels.Annotations.Attributes, but no generator emits code for it; the analyzer reports every use as error KANCHN017. For a producer-only port, declare a [Channel(Mode = ChannelMode.Distributed)] handler.

Channels or Parallel?

Kanject.Core ships [Parallel] and ParallelLoop.ForEachChunkAsync, which solve an adjacent problem. They compose well; pick the simpler one when only one is needed.

[Channel] and [Parallel] cannot be applied to the same method (KANCHN011).

Channels SQS adapter or Kanject.Core.Queue.Provider.AwsSqs?

If you need to send, or want the framework to own queue lifecycle and routing, use Kanject.Core.Queue.Provider.AwsSqs. If you only need to receive a batch inside a Lambda, or want to move a handler between hosting modes, use the Channels adapter. The two can coexist in one service.

Diagnostics

Kanject.Core.Channels.Annotations reports these at compile time:

Design notes

• Mode resolution happens at compile time. Default is resolved from the project's referenced assemblies, so there is no runtime reflection and no behaviour change when a dependency is added at deploy time.
• Batch boundaries are respected. Cancellation never tears a batch, which is what makes partial-batch-failure acknowledgement safe to rely on.
• Drop modes warn by default. Silent loss is rarely intended; KANCHN013 makes it explicit.
• Transport is chosen at the composition root. The attribute never names a transport; the domain assembly carries no AWS SDK reference.

Related packages

License

Licensed under the Kanject Code Libraries License Agreement (KCLLA); the full text ships in this package as LICENSE.md. Organizations whose trailing-twelve-month gross revenue and total funding raised are each below US$250,000 may use it at no cost under the Free Tier. At or above either threshold a commercial license is required — contact commercial@kanjectbusiness.com.