Tutorial5/5

5. Idempotency and business incidents

Wire two platform services into the send pipeline:

Idempotency Service, so redelivered orders do not repeat side effects.
Business Incident Service, so business failures are routed.

empty=true scaffolded none of this wiring. This chapter adds it and tests the duplicate and incident paths.

Register the service clients

In src/Configuration/Composition.cs, add these usings at the top of the file:

using Intropy.IdempotencyService.Client;
using Intropy.BusinessIncidentService.Client;

Add the registrations after the Dapr clients:

services.AddSingleton(_ => new DaprClientBuilder().Build());
services.AddDaprPubSubClient();

// Registered after the DaprClient above on purpose: AddIdempotencyServiceClient
// contributes its own DaprClient registration (camelCase JSON + string enums),
// and the last registration wins for GetRequiredService<DaprClient>().
services.AddIdempotencyServiceClient(c =>
{
    c.AppId = configuration["PlatformServices:IdempotencyServiceAppId"]
              ?? "idempotency-service.services";
});
services.AddBusinessIncidentServiceClient(c =>
{
    c.AppId = configuration["PlatformServices:BusinessIncidentServiceAppId"]
              ?? "business-incident-service.services";
});

The Dapr app ids come from PlatformServices:* in configuration. The defaults match the platform-services deployment and work under the local host, so no appsettings entry is needed here.

Wire the send pipeline

The send pipeline is defined in src/Send/SendRoute.cs. Add these usings at the top of the file:

using Intropy.Contracts.IdempotencyService;
using Intropy.Contracts.BusinessIncidentService;

Resolve the clients and extend the chain inside Configure:

public static SendPipelineBuilder<In, Out, Context> Configure(
    SendPipelineBuilder<In, Out, Context> builder, IServiceProvider provider)
{
    var destinationAdapter = provider.GetRequiredKeyedService<IFileAdapter>(Constants.DestinationPort);
    var idempotencyClient = provider.GetRequiredService<IIdempotencyServiceClient>();
    var incidentClient = provider.GetRequiredService<IBusinessIncidentServiceClient>();

    return builder
        .WithDeserializer(new Deserializer())
        .WithIdempotency(
            idempotencyClient,
            idExtractor: (order, _) => order.OrderId,
            dateExtractor: (order, _) => order.OrderDate)
        .WithExtractor(new Extractor())
        .WithValidator(new Validator())
        .WithTransformer(new Transformer())
        .WithSerializer(new Serializer())
        .WithSender(new Sender(destinationAdapter))
        .WithBusinessIncidents(
            incidentClient,
            messageIdExtractor: ctx => ctx.Metadata.GetValueOrDefault(Constants.ContextKeyOrderId, "unknown"),
            subjectExtractor: ctx => ctx.Metadata.GetValueOrDefault(Constants.ContextKeyOrderId, "unknown"));
}

Use stable business values for idempotency:

idExtractor: OrderId
dateExtractor: OrderDate

Do not use ProcessedAt; it changes per run. Incident routing uses the order id from context.

The receive side needs no wiring. It is framework code, and a file it cannot read or publish stays in the source folder for the next run.

Teach the tests the new edges

The integration tests need fake service clients. In test/OrderSync.Test.Integration/FakeEdges.cs, add these usings at the top of the file:

using Intropy.Contracts.IdempotencyService;
using Intropy.Contracts.BusinessIncidentService;
using Intropy.Framework.Testing.Services;

Add the two fake properties below DestinationFiles:

public FakeIdempotencyServiceClient Idempotency { get; } = new();
public FakeBusinessIncidentServiceClient Incidents { get; } = new();

In CreateServices, replace the production service clients with the fakes. Add the four lines after the existing adapter swaps — the adapter wiring stays, or the tests see empty SourceFiles and DestinationFiles:

var services = Composition.ConfigureServices(new ServiceCollection(), "Development");
UseInternalQueue(services);
services.RemoveAll<EnqueueStep<Context>>();
services.AddSingleton<EnqueueStep<Context>>(Enqueue);

services.RemoveAllKeyed<IFileAdapter>(Constants.SourcePort);
services.AddKeyedSingleton<IFileAdapter>(Constants.SourcePort, (_, _) => SourceFiles);
services.RemoveAllKeyed<IFileAdapter>(Constants.DestinationPort);
services.AddKeyedSingleton<IFileAdapter>(Constants.DestinationPort, (_, _) => DestinationFiles);
services.RemoveAll<IIdempotencyServiceClient>();
services.AddSingleton<IIdempotencyServiceClient>(Idempotency);
services.RemoveAll<IBusinessIncidentServiceClient>();
services.AddSingleton<IBusinessIncidentServiceClient>(Incidents);
return services;

Add two tests to SendPipelineIntegrationTests.cs, one for each new path. Add these usings at the top of the file — the alias is required because Action would otherwise resolve to System.Action:

using Intropy.Contracts.IdempotencyService;
using Intropy.Framework.Testing.Services;
using Action = Intropy.Contracts.IdempotencyService.Action;

Add the tests above NewContext:

[Fact]
public async Task Execute_WithDuplicate_WritesNothingAndDoesNotCommit()
{
    // Arrange: the idempotency service has already seen this order.
    var edges = new FakeEdges();
    edges.Idempotency.NextStatus = new StatusResponse(Action.Ignore, Reason.SameData);
    var provider = edges.CreateServices().BuildServiceProvider();
    var pipeline = FakeEdges.ResolveSendPipeline(provider);

    // Act
    var (result, _) = await pipeline.Execute(
        Encoding.UTF8.GetBytes(ValidRecordJson), NewContext(), CancellationToken.None);

    // Assert: the duplicate cancels as a traceable no-op — consumed, but
    // never written or committed again.
    Assert.IsType<StepResult<string>.Cancelled>(result);
    Assert.Empty(edges.DestinationFiles.Files);
    Assert.Empty(edges.Idempotency.Committed);
}

[Fact]
public async Task Execute_WithValidationFailure_RoutesIncidentAndConsumesWithoutCommit()
{
    // Arrange: an order with no lines fails the Validator.
    var edges = new FakeEdges();
    var provider = edges.CreateServices().BuildServiceProvider();
    var pipeline = FakeEdges.ResolveSendPipeline(provider);
    const string noLines =
        """{"orderId":"ORD-1001","customerId":"CUST-42","orderDate":"2026-01-15T09:30:00+00:00","lines":[]}""";

    // Act
    var (result, _) = await pipeline.Execute(
        Encoding.UTF8.GetBytes(noLines), NewContext(), CancellationToken.None);

    // Assert: consumed with an incident routed; nothing written or committed.
    Assert.IsType<StepResult<string>.Success>(result);
    Assert.Empty(edges.DestinationFiles.Files);
    Assert.Empty(edges.Idempotency.Committed);
    var incident = Assert.Single(edges.Incidents.Incidents);
    Assert.Equal("Order must contain at least one line", incident.Data.Description);
}

A business failure is consumed, not retried, and never commits idempotency, so a corrected redelivery is not cancelled as already seen.

Check it

dotnet test

Passed!  - Failed:     0, Passed:    10, Skipped:     0, Total:    10 - OrderSync.Test.Unit.dll (net10.0)
Passed!  - Failed:     0, Passed:     7, Skipped:     0, Total:     7 - OrderSync.Test.Integration.dll (net10.0)

All unit and integration tests should now pass.

From order-sync-host/, seed and start as in chapter 4:

mkdir -p test/order-sync-source
cp sample-data/sample-order-status-0.json sample-data/sample-order-status-1.json test/order-sync-source/
dotnet run

task seed PORT_DIR=./test/order-sync-source runs the seeding commands if you have Task installed.

ORD-1001.json and ORD-1002.json land in test/order-sync-destination/ as before. The local mock is stateless, so a repeated local run processes the orders again. The duplicate test covers the cancel path with a fake that can return “seen it”.