Kanject.Core.Channels 3.9.6

Kanject Channels

One handler signature. Every hosting mode.

Write (ReadOnlyMemory<T> batch, CancellationToken ct) → ValueTask once, and run it unchanged under a long-lived BackgroundService, a Lambda invocation-local drainer, an SQS-triggered Lambda, a Kinesis stream, or an in-process smoke test. The transport is a composition-root concern — your domain code never references an adapter.

// Domain — transport-agnostic, hosting-agnostic
[Channel(Capacity = 5000, BatchSize = 25)]
public partial ValueTask HandleAsync(
    ReadOnlyMemory<OrderEvent> batch, CancellationToken ct);

// Composition root — pick the transport
services
    .AddOrderHandlerHandleChannel()            // domain wiring
    .AddSqsChannelTransport("orders-queue");   // adapter wiring (separate package)

The same handler moves between hosts as your deployment model evolves — no rewrites, no framework lock-in, no Lambda-SDK dependency in your domain assembly.

Status: Phase 1.3 shipped. Cancellation, ordering, scope-boundary and FullMode contracts are locked (see ChannelOptions). SQS adapter + generator shipped (Phase 2.1 / 2.2). Remaining phases: partial-batch atomicity, Kinesis adapter, samples, integration tests.

What you get

🏛️ Hexagonal positioning

• Port: The generated IXChannelWriter interface is the driven port. It lives in the domain assembly, with no dependency on any adapter or SDK.
• Adapters: Each transport (SQS, EventBridge, Kinesis, in-memory, etc.) is a separate package, providing a provider-specific DI extension and generator. No domain code references any adapter directly.
• Composition root: The adapter is selected at startup via DI wiring, not via attribute arguments.

Conformance contract: Every adapter must pass the universal ChannelConformanceTests suite (see below) to guarantee observable behavior is uniform across transports.

🧭 Parallel vs Channels — which do I reach for?

Parallel and Channels solve adjacent problems and are routinely composed. If you only need one, pick the simplest that fits.

Rule of thumb: if the production side is synchronous and bounded, Parallel wins on ceremony; if the production side is asynchronous or unbounded, or if you need the handler to be host-portable, Channels earns its keep.

🧭 SQS Channels adapter vs native SQS QueueConsumer — which do I reach for?

Two SQS packages exist; they solve overlapping-but-distinct problems. Pick by the shape of the thing you're building.

Rule of thumb: if you need to send, or you want the framework to own queue lifecycle and routing, use the native provider. If you only need to receive a batch inside a Lambda (or move an existing handler between hosting modes), use the Channels adapter. The two can coexist in one service — e.g. native IQueueManager on the send side, Channels [SqsChannel] on the Lambda ingress.

📜 Contracts at a glance

These are load-bearing semantics. Authoritative versions live in the ChannelOptions XML docs; the summary here is for orientation.

Cancellation is all-or-nothing within a batch. The handler receives a CancellationToken for cooperative shutdown. If the handler observes the token and returns early — whether by throwing OperationCanceledException or returning normally — the entire batch is treated as un-acked and its message IDs flow into BatchItemFailures. The drainer does not support partial-ack within a batch. Handlers that need per-item ack semantics must set BatchSize = 1. Cancellation is checked between reads (per-item) or between batches (batched) — never torn across a dispatch.

Scope boundary per mode.

Ordering guarantees.

FullMode × mode validity. BoundedChannelFullMode.Wait is rejected at startup for invocation-local drains (would block past the Lambda deadline). Use DropOldest, DropNewest, or DropWrite for invocation-local; the SQS adapter forces DropWrite since capacity matches record count and the channel cannot overflow.

⚠️ The Problem

You have a stream of items and a batched, async per-batch handler. The naïve options:

1. Hand-rolled Channel<T> + BackgroundService — every team rewrites the same drainer, batching, restart-on-fault, graceful-shutdown, and DI-scope code, slightly differently, slightly buggily. ~80 lines of plumbing per channel.
2. SQS poll loop in a hosted service — couples your handler to AWS SDK; no in-proc fast path; no batching layer between the SDK's per-poll batch and your business batch size; no graceful-shutdown story.
3. Generic IBackgroundJobQueue abstractions — usually Task-typed, lose item-type fidelity, can't batch, can't surface backpressure.
4. Lambda + SQS event source — works, but you get whatever batch SQS hands you (max 10 by default; up to 10000 with MaximumBatchingWindowInSeconds), no internal re-batching, no in-invocation fan-out, no DI-scope-per-batch.

What's missing is the same trick Parallel pulled off: declare the channel intent on the method, get a typed producer + a host-appropriate consumer for free.

🛠️ The Solution

Five pieces, mirroring the Parallel quintet:

The engine is fully usable on its own — call ChannelDispatcher.RunAsync(...) directly with your handler. The generator is what removes the call-site boilerplate.

📖 Attribute Surface

[AttributeUsage(AttributeTargets.Method)]
public sealed class ChannelAttribute : Attribute
{
    // ── Channel sizing ────────────────────────────────────────────
    public int Capacity { get; init; } = 1024;
    public BoundedChannelFullMode FullMode { get; init; } = BoundedChannelFullMode.Wait;

    // ── Batching ──────────────────────────────────────────────────
    public int BatchSize       { get; init; } = 1;     // 1 = no batching
    public int FlushIntervalMs { get; init; } = 0;     // 0 = no time-flush

    // ── Hosting ───────────────────────────────────────────────────
    public ChannelMode Mode { get; init; } = ChannelMode.Default;
    public ServiceLifetime Lifetime { get; init; } = ServiceLifetime.Scoped;

    // ── Resilience ────────────────────────────────────────────────
    public bool RestartOnFault     { get; init; } = true;
    public int  MaxRestartAttempts { get; init; } = -1;        // -1 = infinite
    public int  SafetyMarginMs     { get; init; } = 500;       // InvocationLocal only

    // ── Naming ────────────────────────────────────────────────────
    public string? ChannelName { get; init; }
}

public enum ChannelMode
{
    Default,           // generator picks: InvocationLocal if Lambda referenced, else InProcess
    InProcess,         // System.Threading.Channels + BackgroundService consumer
    InvocationLocal,   // System.Threading.Channels + inline drainer (Lambda)
    Distributed        // Adapter-provided; consumer is a separate process/Lambda
}

Producer-only declaration (when the producer doesn't itself host a [Channel] method):

[ChannelProducer(typeof(OrderEvent))]
public partial class OrderProducer { }
// → generates IOrderEventChannelWriter + DI helper

🧱 Hosting Modes & Adapter Selection

Adapter selection is at the composition root:

// Domain (annotated method) — transport-agnostic
[Channel(Capacity = 5000, BatchSize = 25)]
public partial ValueTask HandleBatchAsync(ReadOnlyMemory<OrderEvent> batch, CancellationToken ct);

// Composition root — adapter chosen here
services
    .AddOrderHandlerHandleBatchChannel()                  // domain wiring
    .AddSqsChannelTransport("orders-queue");              // adapter wiring (separate package)

🚀 Generator Topology

Two generators, two packages:

For:

public sealed partial class OrderHandler
{
    [Channel(Capacity = 5000, BatchSize = 25, FlushIntervalMs = 200)]
    public partial ValueTask HandleBatchAsync(
        ReadOnlyMemory<OrderEvent> batch, CancellationToken ct);
}

…the generator emits five outputs (mirrors the Parallel generator's split):

// 1. Producer interface — what callers inject
public interface IOrderHandlerHandleBatchChannelWriter
{
    ValueTask WriteAsync(OrderEvent item, CancellationToken ct = default);
    ValueTask WriteAsync(ReadOnlyMemory<OrderEvent> items, CancellationToken ct = default);
    bool TryWrite(OrderEvent item);
    ValueTask CompleteAsync();      // graceful drain signal
}

// 2. Auto-attached interface fragment on the consumer type
//    (the same partial-class trick KANPAR011 enforces)
partial class OrderHandler : IOrderHandlerHandleBatchChannelHandler { }

// 3. DI helper
public static IServiceCollection AddOrderHandlerHandleBatchChannel(
    this IServiceCollection services,
    Action<ChannelOptions>? configure = null);

// 4. Hosted consumer (Mode = InProcess)
internal sealed class OrderHandlerHandleBatchChannelService : BackgroundService { ... }

// 5. Invocation-local drainer (Mode = InvocationLocal)
public static class OrderHandlerHandleBatchChannelDrain
{
    public static Task<ChannelDrainResult> DrainAsync(
        IServiceProvider sp, CancellationToken ct);
}

ChannelDrainResult exposes:

• int ProcessedCount
• int FailedCount
• IReadOnlyList<string> UnAckedMessageIds — wire straight into SQSBatchResponse.BatchItemFailures

🔌 Lambda Integration (Mode = InvocationLocal)

The kanject-specific value-add: internal re-batching across the records SQS hands a single Lambda invocation, deadline-aware, with partial-batch-failure semantics.

// [CloudFunction] handler — kanject's existing Lambda primitive
public async Task<SQSBatchResponse> HandleAsync(SQSEvent evt, ILambdaContext ctx)
{
    using var scope = _sp.CreateAsyncScope();
    using var cts = CancellationTokenSource.CreateLinkedTokenSource(default);
    cts.CancelAfter(ctx.RemainingTime - TimeSpan.FromMilliseconds(500));

    var writer = scope.ServiceProvider
        .GetRequiredService<IOrderHandlerHandleBatchChannelWriter>();

    foreach (var record in evt.Records)
        await writer.WriteAsync(
            JsonSerializer.Deserialize<OrderEvent>(record.Body)!, cts.Token);
    await writer.CompleteAsync();

    var drain = await OrderHandlerHandleBatchChannelDrain.DrainAsync(
        scope.ServiceProvider, cts.Token);

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

Lambda timeout behaviour

Three layers of defence, all compile-time wired:

Layer 1 — Soft deadline. Generated glue creates a linked CTS firing RemainingTime − SafetyMarginMs before the hard kill. Drainer observes the token at:

• WaitToReadAsync (no new batch starts)
• the handler(batch, ct) boundary (handler gets a cooperative-cancel chance)
• between batches, never mid-batch — a batch is either fully executed or fully un-acked.

Layer 2 — Partial-batch failures. Un-handled items go into BatchItemFailures so SQS re-delivers only the tail. Successfully handled messages are gone for good. Requires ReportBatchItemFailures on the event source — KANCHN016 (project-level analyzer) warns if missing.

Layer 3 — Hard kill. If the safety margin was wrong, Lambda kills the process; un-acked items are redelivered after the visibility timeout. At-least-once preserved at the cost of duplicates — this is the seam where future [Idempotent] plugs in.

🔍 Diagnostics — KANCHN001…KANCHN016

🔌 Adapter seam: port/adapters split

Producer/consumer transport interfaces are split:

public interface IChannelProducerTransport
{
    string Name { get; }
    ValueTask SendAsync<T>(string resource, T item, CancellationToken ct);
    // Provider-specific batch methods may be exposed via type-test (see README notes)
}

public interface IChannelConsumerTransport
{
    string Name { get; }
    ChannelReader<T> Subscribe<T>(string resource, ChannelSubscriptionOptions options);
}

Each adapter package provides its own generator and analyzer.

Conformance contract: Every adapter must pass the universal ChannelConformanceTests suite:

Every provider project must subclass the conformance suite and pass all tests before publishing.

🗺️ Roadmap

⏳ Phase 1.0 — InProcess mode end-to-end

• ChannelAttribute, ChannelProducerAttribute, ChannelMode enum
• ChannelDispatcher engine
• ChannelGenerator (5 outputs)
• ChannelMethodAnalyzer (KANCHN001–007, 011, 014, 015)
• Sample console fan-out app

⏳ Phase 1.5 — InvocationLocal drainer

• InvocationLocalDrainer engine
• KANCHN010 wired
• Default-mode generator-time heuristic via Compilation.ReferencedAssemblyNames
• Lambda + SQS sample

⏳ Phase 2.0 — Distributed transport seam + AwsSqs adapter package

• IChannelProducerTransport / IChannelConsumerTransport interfaces
• Kanject.Core.Channel.Provider.AwsSqs (adapter package)
• ChannelTransportBindingGenerator emits DI extension, provider-specific analyzers, conformance suite

⏳ Phase 2.5 — EventBridge adapter package

• Kanject.Core.Channel.Provider.EventBridge (adapter package)
• EventBridge-rule → SQS-pipe → consumer-Lambda end-to-end sample

⏳ Phase 3.0 — [CircuitBreaker] integration

• Open breaker pauses drain; surfaces via OnOpen callback

⏳ Phase 3.5 — [Channel] × [Parallel] composition (lifts KANCHN011)

• Each batch dispatched via ParallelLoop.ForEachChunkAsync
• Recurring × Parallel × Channels triple-composition sample

⏳ Phase 4.0 — Kinesis provider

• Ordered, sharded distributed transport

✏️ Design decisions worth knowing

• Default mode is generator-time, not runtime. The generator inspects the project's referenced assemblies (Compilation.ReferencedAssemblyNames) — no reflection at runtime, no startup cost, no surprise behaviour swaps when a dependency is added at deploy time.
• Producer/consumer transports are split. SQS-receive does its work via host wiring (event source mapping), not application code; a unified IChannelTransport would force half-empty Subscribe implementations. The split mirrors how AWS actually works, and matches the port/adapter boundary in hexagonal.
• Adapter selection is at the composition root. The domain never names its adapter; the handler is annotated transport-agnostically, and the adapter is chosen in DI wiring. This keeps the domain assembly free of SDK references and enables true swap-ability.
• Conformance contract is enforced. Every adapter must subclass and pass the universal conformance suite before publishing, ensuring observable behavior is uniform.
• Generator topology is two-tier. The domain generator emits only transport-agnostic code; each adapter package ships its own generator and analyzer for provider-specific wiring and diagnostics.
• Adapter scorecard is published. Each adapter's cold-start, steady-state allocation, and throughput are benchmarked and published in the README; regressions block release.
• Soft-deadline cancellation respects batch boundaries. A batch is either fully executed or fully un-acked; cancellation never tears mid-batch. This is what makes partial-batch-failure ack semantics safe to rely on.
• FullMode = DropOldest / DropWrite warns by default. Silent message loss is rarely what callers actually want; KANCHN013 surfaces it loudly.
• Predicate parity from day one. ChannelMethodAnalyzer and ChannelGenerator share a single ChannelTransformHelper for ForAttributeWithMetadataName extraction — no repeat of the schema-vs-code-generator divergence the repo memory documents for the Dynamo pipeline.
• Partial enforcement ships in Phase 1.0. No KANPAR011-style after-the-fact retrofit; KANCHN001 + KANCHN002 are in the v1 diagnostic set.
• Test fixtures partial from the first commit. Every [Channel]-annotated test type is public sealed partial class — the Parallel test-suite migration is not repeated here.
• EmitCompilerGeneratedFiles trap pre-documented in the sample csproj. Same MSBuild comment block as the Parallel sample.