TCIS.EventBus 1.0.0-rc.20

TCIS.EventBus

Transactional messaging for the TCIS ecosystem: an outbox that writes the message in the same transaction as your business data, a background sender, and a consumer side with retries and a poison-message path.

Transport bindings live in separate packages — see TCIS.EventBus.Kafka. For choosing between mediator.Publish and the EventBus, see TCIS.Mediator.

Table of contents

1. Registration

builder.Services.AddTEventBus(options =>
{
    builder.Configuration.GetSection("TCIS:EventBus").Bind(options);

    options.UseSqlServer(connectionString);      // outbox lives next to your data
    options.UseKafka(kafka =>
        builder.Configuration.GetSection("TCIS:EventBus:Kafka").Bind(kafka));
});

Three things must be registered or startup throws: the bus itself, one storage provider, and one transport provider.

{
  "TCIS": {
    "EventBus": {
      "DefaultGroupName": "tcis.queue.vessel-service",
      "Version": "v1",
      "FailedRetryCount": 10,
      "FailedRetryInterval": 30,
      "SucceedMessageExpiredAfter": 86400,
      "ConsumerShutdownTimeoutSeconds": 5,
      "Kafka": {
        "Servers": "broker:9092",
        "MainConfig": { "group.id": "tcis.subscriber.vessel-group" }
      }
    }
  }
}

2. Publishing

public sealed class BerthVesselHandler(IEventPublisher bus, VesselDbContext db)
{
    public async Task Handle(BerthVessel cmd, CancellationToken ct)
    {
        // The bus joins the EF transaction: the outbox row and the business row commit together.
        using var tx = db.Database.BeginTransaction(bus, autoCommit: false);

        visit.Status = "BERTHED";
        await db.SaveChangesAsync(ct);

        await bus.PublishAsync("vessel.berthed", new VesselBerthed(visit.Id, visit.BerthCode));

        await tx.CommitAsync(ct);
    }
}

BeginTransaction(publisher, autoCommit) is an extension shipped by the storage package (TCIS.EventBus.SqlServer, TCIS.EventBus.PostgreSql); overloads exist for DatabaseFacade and for a raw IDbConnection.

The point of the outbox is that PublishAsync does not talk to the broker. It writes a row in the same transaction. If the transaction rolls back, the message was never published. If the process dies after commit but before send, a background processor picks the row up and sends it.

Publishing outside a transaction is allowed and sends immediately after persisting — use it only when there is no business data to keep in step.

3. Subscribing

public sealed class VesselSubscriber(IVesselService service) : IEventSubscribe
{
    [EventSubscribe("vessel.berthed")]
    public async Task OnBerthed(VesselBerthed evt, CancellationToken ct)
    {
        await service.RecordBerthingAsync(evt.VisitId, evt.BerthCode, ct);
    }
}

builder.Services.AddScoped<VesselSubscriber>();   // discovery reads DI registrations

Subscribers are discovered from the service collection, so the class must be registered — an unregistered subscriber is not an error, it simply never runs.

Each subscriber runs in its own DI scope, not the publisher's. A throwing subscriber does not affect the publisher; it affects only its own message.

4. Headers and context propagation

Ambient context travels in headers, so subscribers do not need it in the payload:

Keep the DTO clean — do not copy TenantId or SiteCode into the payload.

5. ⚠️ What happens to a message that fails

Three distinct outcomes. Confusing them is the usual reason an incident takes hours instead of minutes.

The subscriber throws

Retried in place up to 3 times, then handed to the background retry processor, which retries every FailedRetryInterval seconds up to FailedRetryCount times. After that the message stays Failed in storage and FailedThresholdCallback fires.

The message cannot be routed or deserialized

No subscriber matches the topic+group, or the payload does not parse. Retrying cannot help — the next attempt fails identically. The message is stored as an exception message, the offset is committed, and the stream continues. One Error log line records the reason:

Message parked and skipped: it could not be routed or deserialized, so the offset was committed
to let the stream continue. Id:…, Name:…, Group:…, Reason:SubscriberNotFoundException-->…

The payload parses but does not match the DTO

Nothing is reported. System.Text.Json ignores unknown JSON properties and leaves absent ones at their default, so the subscriber receives a half-empty object and the message is marked Succeeded. Property names are matched case-insensitively — but a genuinely different name (vessel_name versus VesselName) still will not bind.

This one has no library-level defence. Validate required fields inside the subscriber:

if (string.IsNullOrWhiteSpace(evt.VesselName))
    throw new TValidationException("VALIDATION_MISSING_VESSEL_NAME", "vessel_name is empty");

That converts a silent empty row into a parked message with a log line.

6. Consuming from a third-party producer

External producers do not emit TCIS headers. The transport fills in what is missing — see the Kafka README for how t-msg-id is derived and why it stays stable across redeliveries.

Two things to get right on the TCIS side:

1. Do not set TopicNamePrefix. It would rewrite the partner's topic name into prefix.partner.topic, which does not exist.
2. Match the payload explicitly. Declare [JsonPropertyName("...")] on the DTO rather than hoping the partner's naming matches yours. When the schema is not yet known, take a JsonElement first and log raw.GetRawText() to see exactly what arrives.

7. Storage choices

InMemory is for development and for stateless gateways where losing an un-sent message is acceptable. Anything with business meaning belongs in a durable store.

Two InMemory specifics worth knowing:

• Messages are held for SucceedMessageExpiredAfter (24 h by default) before the collector removes them. That is 24 hours of messages resident in RAM — lower it for a high-volume consumer.
• The dictionaries are static, so every host in the same process shares one store. This is fine for a single application, but it means two hosts side by side (or parallel integration tests) see each other's messages.

8. Delivery guarantees

At-least-once, with one condition. The offset is committed only after the subscriber finishes, so a crash mid-processing means the broker redelivers.

That condition breaks when EnableSubscriberParallelExecute = true: messages are then handed to an in-memory channel and the offset is committed straight away. A crash loses whatever is still in the channel. Enable it only when the throughput matters more than those messages do.

Because delivery is at-least-once, subscribers must be idempotent. Use t-msg-id as the deduplication key.

9. Options reference

10. Pitfalls

11. Provenance — fork point

This package and its transport/storage siblings are a fork of DotNetCore.CAP, not original TCIS code. The fork exists because TCIS needed hooks inside CAP's consumer lifecycle that CAP does not expose — see §11.3.

11.1. The fork point

Reproduce the comparison with:

git clone https://github.com/dotnetcore/CAP.git && cd CAP
git worktree add --detach ../CAPfork d3f53648
git log --oneline d3f53648..origin/master        # what we are behind

11.2. State as measured on 2026-08-18

The 10 upstream commits are documentation, the Kubernetes dashboard, an AmazonSQS null-response fix, sample removal, and dependency bumps. None of them touches code we carry.

The only upstream change that concerns us is the dependency baseline:

Plus one commit removing System.Linq.Async for .NET 10 support.

11.3. Why the fork exists

Three needs, none of which CAP exposes an extension point for:

IConsumerContextInitializer is deliberately neutral — it does not know the word "tenant" — so the multi-tenant behaviour lives outside this package, in TCIS.EventBus.MultiTenancy. That package is original TCIS code, not forked.

Everything else in the diff is one of two things:

• Modernisation — C# 12 primary constructors, ILogger → ILogWriter, TGuard at constructor boundaries. Wide, mechanical, no behavioural change. Recognisable by touching many lines while adding no comments.
• Defect fixes — 18 of them, August 2026. Recognisable by code and Vietnamese explanatory comments arriving together. Concentrated in IConsumerRegister.Default.cs, IDispatcher.Default.cs, ISubscribeExector.Default.cs and KafkaConsumerClient.cs.

None of the 18 fixes exists upstream. As of v10.0.2, ISubscribeExector.Default.cs on master still contains:

catch (OperationCanceledException)
{
    //ignore          // a cancelled handler is still recorded as Succeeded — silent data loss
}

11.4. How to work with this fork

• Hitting a bug? Check whether upstream already fixed it before debugging: compare the file against d3f53648..origin/master. Today that range is empty for the core, so the answer is almost certainly "no" — but re-check, because that will not stay true.
• Upgrading a transport/storage dependency? Take the version upstream is on (table in §11.2) rather than picking one yourself; upstream tests against those combinations.
• Security advisories for Confluent.Kafka, Npgsql, Microsoft.Data.SqlClient, RabbitMQ.Client and CAP itself apply to us and must be ported by hand.
• Do not renumber or rename further. The identifier rename (Cap* → EventBus*/T*) is already the main cost of any future re-base. It is mechanical and reversible with a script; widening it is not.
• Keep this section current. Re-run the two commands in §11.1 when you touch this package, and update §11.2. A fork whose distance from upstream is unknown cannot be maintained.