TransitGremlin.Sqs 1.0.0

TransitGremlin

Target framework: net10.0 · Language: C# · Test runner: xUnit

PayloadGremlin tests bad payloads. TransitGremlin tests bad message delivery.

TransitGremlin is a lightweight chaos testing library for message-driven .NET applications. It simulates realistic broker and delivery problems — duplicate messages, dropped messages, delayed delivery, missing correlation IDs, lock loss, offset skips, and more.

TransitGremlin does not inspect, parse, serialize, or mutate message bodies. The body is treated as opaque and passed through untouched by all built-in mutations.

What TransitGremlin is

• A chaos testing library for message transport problems
• A fluent configuration API for delivery mutations
• An in-memory harness for fast unit tests
• Thin broker adapters for Azure Service Bus, RabbitMQ, Kafka, SQS, and MassTransit
• A capability model that reports unsupported behaviour honestly

What TransitGremlin is not

• A message broker replacement
• A replacement for Azure Service Bus, RabbitMQ, Kafka, SQS, MassTransit, or Wolverine
• A payload mutation library (that is PayloadGremlin)
• A JSON fuzzer or schema validator

Why message delivery fails in production

Message-driven systems fail in ways unit tests rarely cover:

• Brokers duplicate messages under retries and network partitions
• Messages arrive out of order across partitions
• Correlation and trace headers go missing across service boundaries
• Visibility timeouts expire while handlers are still running
• Lock tokens are lost on peek-lock consumers
• Old messages are replayed from retention logs
• Delivery counts climb until dead-letter thresholds are hit
• Offsets rewind or skip during consumer rebalances

TransitGremlin helps you test whether your application survives these problems.

Supported packages

Installation

dotnet add package TransitGremlin
dotnet add package TransitGremlin.InMemory

Add broker adapters as needed:

dotnet add package TransitGremlin.AzureServiceBus
dotnet add package TransitGremlin.RabbitMq
dotnet add package TransitGremlin.Kafka
dotnet add package TransitGremlin.Sqs
dotnet add package TransitGremlin.MassTransit

Basic usage

builder.Services.AddTransitGremlin(options =>
{
    options.Enabled = builder.Environment.IsDevelopment();
    options.Seed = 42; // deterministic for sequential mutation draws; concurrent publishing may interleave draws

    options.For<PaymentRequested>()
        .Duplicate(times: 2, probability: 0.10)
        .Drop(probability: 0.01)
        .Delay(TimeSpan.FromSeconds(10), probability: 0.10)
        .Replay(probability: 0.05)
        .Reorder(probability: 0.05)
        .RemoveCorrelationId(probability: 0.20)
        .RemoveTraceParent(probability: 0.10)
        .DuplicateMessageId(probability: 0.10)
        .LockLost(probability: 0.05)
        .MaxDeliveryExceeded(probability: 0.02);
});

builder.Services.AddTransitGremlinAzureServiceBus();

Publish and consume through the gremlin:

await gremlin.PublishAsync(envelope, async (mutated, ct) =>
{
    await publisher.SendAsync(mutated, ct);
}, cancellationToken);

await gremlin.ConsumeAsync(envelope, async (mutated, ct) =>
{
    await handler.HandleAsync(mutated.Body, ct);
}, cancellationToken);

In-memory harness

builder.Services.AddTransitGremlinInMemory();

var harness = new TransitGremlinHarness(gremlin);

await harness.PublishAsync(new TransitEnvelope<OrderSubmitted> { Body = order, MessageId = "message-1" });
await harness.DeliverToAsync<OrderSubmitted>(async message =>
{
    await handler.HandleAsync(message.Body);
});

Assert.Contains("Duplicate", harness.MutationLog);

Supported mutations

Delivery

Duplicate, Drop, Delay, Replay, Reorder, Poison, DeadLetter, Redeliver, RedeliveryStorm, Expire, PartialBatchFailure

Metadata and headers

RemoveMessageId, DuplicateMessageId, RemoveCorrelationId, RemoveCausationId, RemoveTraceParent, RemoveIdempotencyKey, ChangeDeliveryCount, ChangePartitionKey, ChangeSessionId, ChangeSubject, ChangeContentType, ChangeTenantId, ChangeEnqueuedAt, ChangeExpiresAt

Broker semantics

AckFailure, NackFailure, LockLost, VisibilityTimeoutExpired, MaxDeliveryExceeded, OffsetRewind, OffsetSkip

Most fluent methods default to sensible pipeline directions: publish/send mutations default to Outgoing, while consumer and broker-semantic mutations default to Incoming. Override with the optional direction parameter:

options.For<OrderSubmitted>()
    .RemoveCorrelationId(probability: 0.20) // Incoming by default
    .Drop(probability: 0.01, direction: TransitMutationDirection.Incoming);

Unsupported mutation behaviour

When a mutation requires capabilities the active adapter does not support:

• Skip — silently skip the mutation
• Warn — log a warning and skip (default)
• Throw — raise UnsupportedTransitMutationException

options.UnsupportedMutationBehaviour = UnsupportedMutationBehaviour.Throw;

Custom mutations

public sealed class RemoveTenantHeaderMutation<T> : ITransitMutation<T>
{
    public string Name => "RemoveTenantHeader";
    public TransitAdapterCapabilities RequiredCapabilities => TransitAdapterCapabilities.Headers;

    public bool CanApply(TransitEnvelope<T> envelope, TransitMutationContext context) =>
        envelope.Headers.ContainsKey("tenant-id");

    public ValueTask<IReadOnlyList<TransitEnvelope<T>>> ApplyAsync(
        TransitEnvelope<T> envelope,
        TransitMutationContext context,
        CancellationToken cancellationToken = default)
    {
        var headers = envelope.Headers.ToDictionary();
        headers.Remove("tenant-id");
        return ValueTask.FromResult<IReadOnlyList<TransitEnvelope<T>>>([envelope with { Headers = headers }]);
    }
}

options.For<OrderSubmitted>()
    .UseMutation(new RemoveTenantHeaderMutation<OrderSubmitted>(), probability: 0.20);

// Global custom mutations via factory:
options.ForAll()
    .UseMutationFactory(
        type => Activator.CreateInstance(typeof(RemoveCorrelationIdMutation<>).MakeGenericType(type))!,
        probability: 0.10);

Built-in mutations always preserve the message body. Custom mutations must preserve the body as well; EnvelopeGuard enforces this at runtime.

Broker adapters

Each adapter registers ITransitGremlinAdapter, ITransitOutgoingAdapter, and ITransitIncomingAdapter via TryAddSingleton. The first registered adapter wins; register only the adapter matching your transport.

Azure Service Bus

builder.Services.AddTransitGremlinAzureServiceBus();

Supports duplicate send, scheduled delay, metadata/header mutation, sessions, and expiry mapping via TimeToLive. Consumer-side dead-letter, lock, and redelivery semantics are simulated through metadata markers in the in-memory harness rather than advertised as transport mapper capabilities.

RabbitMQ

builder.Services.AddTransitGremlinRabbitMq(options =>
{
    options.DelayedDeliverySupported = true; // when delayed message plugin is configured
});

Supports duplicate publish, drop, header mutation, ack/nack simulation, poison loops, and dead-letter exchange scenarios.

Kafka

builder.Services.AddTransitGremlinKafka();

Kafka is log-based. TransitGremlin supports offset rewind/skip, replay, duplicate processing, header/key mutation, and out-of-order simulation. Kafka does not provide broker-native scheduled delay, so Delay is not advertised as a capability.

Amazon SQS

builder.Services.AddTransitGremlinSqs();

Supports duplicate send, message delay (capped at 900 seconds), and header mutation on send. Visibility timeout, receive count, and dead-letter/redrive semantics are harness/metadata simulations and are not advertised as send-mapper capabilities. SQS assigns message IDs on send, so message ID mutation is not advertised.

MassTransit

builder.Services.AddTransitGremlinMassTransit();

// In MassTransit bus configuration:
// endpointConfigurator.UsePublishFilter(typeof(TransitGremlinPublishFilter<>), context);
// endpointConfigurator.UseConsumeFilter(typeof(TransitGremlinConsumeFilter<>), context);

Thin middleware integration for outgoing publish and incoming consume. Publish filters fan out duplicate messages via IPublishEndpoint using a pipeline payload (not a transport header) to avoid re-entering the gremlin filter. Consume filters attach mutated envelopes to the pipeline via TransitGremlinConsumePayload<T> and MassTransitEnvelopeMapper.TryGetTransitEnvelope. Incoming duplicate/redelivery fan-out reuses the same ConsumeContext for simulation; against a real transport this can interact with acknowledgement semantics, so prefer the in-memory harness for ack/redelivery testing. Does not replace MassTransit retry, scheduling, outbox, or transport abstractions. Scheduled delay is not advertised because NotBefore is not mapped to MassTransit scheduling.

Publish and consume through adapters:

await outgoingAdapter.PublishAsync(gremlin, envelope, async (mutated, ct) => { ... }, ct);
await incomingAdapter.ConsumeAsync(gremlin, envelope, async (mutated, ct) => { ... }, ct);

Adapter capability model

Each adapter exposes TransitAdapterCapabilities flags. The in-memory adapter supports all mutations. Broker adapters support what is realistic for that transport and report the rest through the unsupported mutation behaviour setting.

Global rules

options.ForAll()
    .Duplicate(probability: 0.05)
    .Delay(TimeSpan.FromSeconds(5), probability: 0.10)
    .RemoveCorrelationId(probability: 0.10);

Roadmap

• Optional Testcontainers-based integration tests (disabled by default)
• Additional adapters (Wolverine, Rebus, NServiceBus)
• Publish/consume pipeline source generators
• Dashboard for mutation audit records

License

MIT