Tutorial5/5

5. Idempotency and business incidents

In this chapter you wire the pipeline to the two platform services: the Idempotency Service, so the same order can arrive twice and its side effect happens once, and the Business Incident Service, so the Validator’s failures from chapter 3 reach a process owner instead of stopping at a trace.

Redeliveries are normal operation, not edge cases. Your own integration suite already says so: the scaffolded test Execute_WhenDeleteFails_ItemIsEnqueuedButFileLeftForRedelivery pins that a source file whose delete fails is processed and left in place — the next run will deliver it again, and the send side must absorb the duplicate. Idempotency explains the model; here you wire it in. empty=true scaffolded none of this wiring, which is also why chapter 1’s last red test is still red.

Register the service clients

Both clients register in src/Configuration/Composition.cs, right after the Dapr clients. Add two usings at the top (Intropy.IdempotencyService.Client and Intropy.BusinessIncidentService.Client), then the registrations:

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

// AddIdempotencyServiceClient contributes its own DaprClient registration
// (camelCase JSON + string enums), and the last registration wins — so the
// clients go after the DaprClient above.
services.AddIdempotencyServiceClient();      // Dapr app id: idempotency-service
services.AddBusinessIncidentServiceClient(); // Dapr app id: business-incident-service

The app ids are the same service names your system host already declares and mocks: Services.cs and the mocks/ folder in order-sync-host/ have been waiting for this chapter.

Wire the send pipeline

In src/Send/ServiceCollectionExtensions.cs, add two usings (Intropy.Contracts.IdempotencyService and Intropy.Contracts.BusinessIncidentService), then resolve the clients and extend the builder chain:

services.AddSendPipeline<In, Out, Context>("OrderSync.Send", (builder, provider) =>
{
    var destinationAdapter = provider.GetRequiredKeyedService<IFileAdapter>(Constants.DestinationFileAdapterKey);
    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"));
});

The two idempotency extractor functions are the decision that matters:

idExtractor must return the same value every time the same business object arrives. OrderId has that property; a UUID minted per delivery does not; it would make the check pass everything.
dateExtractor must return the business timestamp, not the processing time. It’s how the service tells a genuine update (re-run it) from a stale redelivery (ignore it). ProcessedAt from the context would be fresh on every run and defeat the check.

The check runs right after deserialization: an unseen order proceeds, an unchanged redelivery cancels the pipeline as a traceable no-op, and after a successful send a finalizer commits the order so the next delivery is recognised. What counts as “changed” is a content hash you can customise; Add an idempotency check covers the options.

The incident extractors key everything a process owner sees on the order id from context, which is why chapter 2 promoted it there: by the time an incident is routed, the typed message may be long gone.

Wire the receive pipeline

The receive pipeline gets incident routing too, keyed on the source file name: the only identity a file that can’t be read has. In src/Receive/ServiceCollectionExtensions.cs, add the Intropy.Contracts.BusinessIncidentService using and extend the chain:

services.AddReceivePipeline<Context>("OrderSync.Receive", (builder, provider) =>
{
    var sourceAdapter = provider.GetRequiredKeyedService<IFileAdapter>(Constants.SourceFileAdapterKey);
    var daprClient = provider.GetRequiredService<DaprClient>();
    var tiOptions = provider.GetRequiredService<TransactionalIntegrationOptions>();
    var frameworkOptions = provider.GetRequiredService<FrameworkOptions>();
    var incidentClient = provider.GetRequiredService<IBusinessIncidentServiceClient>();
    var enqueuer = enqueuerOverride?.Invoke(provider)
        ?? new Enqueuer(daprClient, tiOptions, frameworkOptions);

    return builder
        .WithReceiver(new Receiver(sourceAdapter))
        .WithEnqueuer(enqueuer)
        .WithCompleter(new Completer(sourceAdapter))
        .WithBusinessIncidents(
            incidentClient,
            messageIdExtractor: ctx => ctx.Metadata.GetValueOrDefault(Constants.ContextKeyFileName, "unknown"),
            subjectExtractor: ctx => ctx.Metadata.GetValueOrDefault(Constants.ContextKeyFileName, "unknown"));
});

Teach the tests the new edges

The integration suite builds the production DI graph, which now resolves two service clients, so FakeEdges must swap them for fakes, the same way it already swaps the file adapters. Add three usings to test/OrderSync.Test.Integration/FakeEdges.cs (Intropy.Contracts.BusinessIncidentService, Intropy.Contracts.IdempotencyService, Intropy.Framework.Testing.Services), plus two properties next to the adapters:

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

And in CreateServices, before the return, remove-then-add so a reordered composition can never silently shadow a fake:

services.RemoveAll<IIdempotencyServiceClient>();
services.AddSingleton<IIdempotencyServiceClient>(Idempotency);
services.RemoveAll<IBusinessIncidentServiceClient>();
services.AddSingleton<IBusinessIncidentServiceClient>(Incidents);

Now collect the payoff in the tests. First, the scaffold’s last red pin: Execute_WithUnreadableFile_RoutesIncidentKeyedOnFileName goes green the moment the receive pipeline routes incidents. Its comment told you to enable the guarded assertion when you got here. Replace the two comment lines at the end of the test:

// Assert: incident routed, nothing enqueued, source file left in place.
Assert.IsType<StepResult<SourceItem>.Success>(result);
Assert.Equal(0, edges.Enqueue!.Count);
var incident = Assert.Single(edges.Incidents.Incidents);
Assert.Equal("corrupt.json", incident.Data.Context!["fileName"]);

Second, pin the cancel behaviour you just bought. In test/OrderSync.Test.Integration/SendPipelineIntegrationTests.cs, add two usings (Intropy.Contracts.IdempotencyService and using Action = Intropy.Contracts.IdempotencyService.Action;) and one test:

[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 = provider.GetRequiredService<ISendPipeline<Context>>();

    // 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);
}

Check it

dotnet test

Passed!  - Failed:     0, Passed:    12, Skipped:     0, Total:    12 - OrderSync.Test.Unit.dll
Passed!  - Failed:     0, Passed:     9, Skipped:     0, Total:     9 - OrderSync.Test.Integration.dll

Everything is green for the first time, and honestly so: the unreadable-file pin from chapter 1 now sees its incident routed, and the new duplicate test proves an already-seen order cancels without writing or committing.

Then confirm the wired component still runs. From order-sync-host/, seed and start as in chapter 4:

task seed PORT_DIR=./test/order-sync-source
dotnet run

ORD-1001.json lands in test/order-sync-destination/ as before. But this run made two calls chapter 4’s run didn’t: an idempotency check after deserialization and a commit after the send, both answered by the mocked service the host runs for you.

Be clear about what the local mock can and cannot show. It is an OpenAPI example server, not a real service: every /status call answers Proceed / NoPreviousData, and it remembers nothing. Stop the host, seed the same order again, start it again: the order is processed again, with a fresh ProcessedAt. That’s the mock being stateless, not the wiring being wrong; the cancel path is exactly what your Execute_WithDuplicate_WritesNothingAndDoesNotCommit test pins against a fake that can say “seen it”. Against the real, stateful Idempotency Service in a deployed environment, that redelivered order cancels; how incidents reach their owners there is covered in Route business incidents.