✨ Outbox pattern

[Hulkstance]

When processing domain events, there are two ways:

• Immediate consistency (before the SaveChangesAsync call), which is useful for cases like creating an order and its corresponding historical order where one doesn't make sense without the other. Domain events are basically part of the same transaction. We've implemented this using separate interfaces IPreDomainEvent (immediate consistency) and IPostDomainEvent (eventual consistency)
• Eventual consistency (after the SaveChangesAsync call), where domain events are basically a separate transaction. The issue we ran into with this in prod (large userbase), was that if the user closed their browser mid-request, some domain events wouldn't fire up. You solved this by queuing them up using the HTTP context while the user is still online, which is great.

That said, I wonder if you guys considered the Outbox pattern? It's kind of related to integration events, i.e. #123, not sure what the current progress on that is

References:

• https://github.com/QuickOrBeDead/TransactionalOutboxPattern/blob/main/TransactionalOutboxPatternApp/Infrastructure/Service/EventLogService.cs
• https://github.com/QuickOrBeDead/TransactionalOutboxPattern/blob/main/TransactionalOutboxPatternApp/Infrastructure/Service/OrderService.cs#L63

It's a bit off topic, but I just wanted to say I'm really impressed by your template. You've solved the same problems we did, but in a different way. I especially like how you handled domain invariants and kinda "unified" EF Core configurations with FluentValidation. Really great job!

[danielmackay]

Hey @Hulkstance, thanks for the feedback and your very kind comments. 🙏

The Outbox pattern is something we've been thinking about. Haven't quite decided to pull the trigger yet. Will update this issue if we put it in.

[danielmackay]

Looking to get this done in the next few weeks

[Hulkstance]

Looking to get this done in the next few weeks

Perfect! Thanks. You can borrow some ideas from https://github.com/MassTransit/MassTransit/blob/develop/src/Persistence/MassTransit.EntityFrameworkCoreIntegration/EntityFrameworkCoreIntegration/BusOutboxDeliveryService.cs#L54

public sealed class OutboxMessageProcessor : BackgroundService
{
	const int PollingIntervalSeconds = 30;
	const int BatchSize = 1000;

	readonly IServiceProvider _serviceProvider;
	readonly ILogger<OutboxMessageProcessor> _logger;
	readonly PeriodicTimer _pollingInterval = new(TimeSpan.FromSeconds(PollingIntervalSeconds));
	readonly string _outboxTopic;
	readonly string _outboxSubscription;
	readonly Func<AppDbContext, DateTime, int, IAsyncEnumerable<OutboxMessage>> _outboxMessagesQuery;

	public OutboxMessageProcessor(IServiceProvider serviceProvider, IConfiguration configuration, ILogger<OutboxMessageProcessor> logger)
	{
		_serviceProvider = serviceProvider;
		_logger = logger;
		
		_outboxTopic = configuration[ConfigurationKeys.AzureServiceBusOutboxTopic]!;
		_outboxSubscription = configuration[ConfigurationKeys.AzureServiceBusOutboxSubscription]!;
		
		// Precompile for performance: avoids EF Core query translation overhead every poll
		_outboxMessagesQuery = EF.CompileAsyncQuery((AppDbContext dbContext, DateTime nextAttemptUtc, int limit) =>
			dbContext.Set<OutboxMessage>()
				.Where(x => x.ProcessedOnUtc == null)
				.Where(x => !x.IgnoreRetry)
				.Where(x => x.NextAttemptUtc == null || x.NextAttemptUtc <= nextAttemptUtc)
				.OrderBy(x => x.OccurredOnUtc)
				.Take(limit));
	}
	
	protected override async Task ExecuteAsync(CancellationToken stoppingToken)
	{
		// Ensure outbox topic & subscription exist with duplicate detection enabled
		await using (var scope = _serviceProvider.CreateAsyncScope())
		{
			var eventBus = scope.ServiceProvider.GetRequiredService<IEventBus>();
			await eventBus.EnsureSubscriptionIsCreated(_outboxTopic, _outboxSubscription, enableDuplicateDetection: true);
		}

		while (await _pollingInterval.WaitForNextTickAsync(stoppingToken))
		{
			try
			{
				await ProcessAllTenantOutboxMessagesAsync(stoppingToken);
			}
			catch (Exception ex)
			{
				_logger.LogError(ex, "An error occurred during processing: {ErrorMessage}", ex.Message);
			}
		}
	}

	async Task ProcessAllTenantOutboxMessagesAsync(CancellationToken cancellationToken)
	{
		await using var scope = _serviceProvider.CreateAsyncScope();
		var magneto = scope.ServiceProvider.GetRequiredService<IMagneto>();
		var tenants = await magneto.QueryAsync(new GetAllTenants(), CacheOption.Default, cancellationToken);

		await Task.WhenAll(tenants.Select(tenant => ProcessTenantSafeAsync(tenant, cancellationToken)));
	}

	async Task ProcessTenantSafeAsync(ApplicationTenant tenant, CancellationToken cancellationToken)
	{
		try
		{
			// Each tenant runs in its own isolated async scope with its own DbContext instance
			// Reason being: you can't use the same AppDbContext concurrently
			await using var tenantScope = _serviceProvider.CreateAsyncScope();
			var provider = tenantScope.ServiceProvider;
			await ProcessTenantOutboxMessagesAsync(tenant, provider, cancellationToken);
		}
		// Keep this catch, otherwise Task.WhenAll would throw when any tenant fails, which is not what we want
		catch (Exception ex)
		{
			_logger.LogError(ex, "Error while processing tenant {TenantId}", tenant.Id);
		}
	}
	
	async Task ProcessTenantOutboxMessagesAsync(ApplicationTenant tenant, IServiceProvider serviceProvider, CancellationToken cancellationToken)
	{
		var dbContext = serviceProvider.GetRequiredService<AppDbContext>();
		var eventBus = serviceProvider.GetRequiredService<IEventBus>();

		// Set tenant for EF Core global filter ("WHERE TenantId = TenancyContext.Tenant.Id"); normally set via TenancyMiddleware, but background jobs lack HTTP context
		dbContext.TenancyContext.Tenant = tenant;

		// Retry on transient database failures (e.g. deadlocks, Azure SQL throttling, or network blips)
		var strategy = dbContext.Database.CreateExecutionStrategy();
		await strategy.ExecuteAsync(async () =>
		{
			var unprocessedMessages = await _outboxMessagesQuery(dbContext, DateTime.UtcNow, BatchSize)
				.ToListAsync(cancellationToken);

			if (unprocessedMessages.Count == 0)
				return;

			// Batch messages into a single request (because it doesn't make sense to publish messages one by one)
			await ProcessMessagesInBatchAsync(dbContext, unprocessedMessages, eventBus);
			await dbContext.SaveChangesAsync(cancellationToken);
		});
	}

	async Task ProcessMessagesInBatchAsync(AppDbContext dbContext, List<OutboxMessage> messages, IEventBus eventBus)
	{
		using var activity = OrderServiceTelemetry.ActivitySource.StartActivity("Outbox.PublishBatch");
		activity?.SetTag("tenant.id", dbContext.TenancyContext.Tenant.Id);
		activity?.SetTag("message.count", messages.Count);
		
		try
		{
			var integrationEvents = messages.Select(m => new OutboxIntegrationEvent(
				id: m.Id, // used for MessageId (idempotency key)
				entityType: m.Type,
				action: m.Action,
				content: m.Content,
				tenantId: dbContext.TenancyContext.Tenant.Id
			)).ToList();

			await eventBus.PublishBatch(_outboxTopic, integrationEvents);

			foreach (var message in messages)
			{
				message.MarkAsProcessed();
			}
		}
		catch (Exception ex)
		{
			_logger.LogError(ex, "Failed to publish batch. Tenant: {TenantId}", dbContext.TenancyContext.Tenant.Id);

			activity?.SetStatus(ActivityStatusCode.Error, ex.Message);
			
			var isTransient = IsTransient(ex);
			foreach (var message in messages)
			{
				message.MarkAsFailed($"Service Bus batch error: {ex.Message}", ignoreRetry: !isTransient);
			}
		}
	}

	static bool IsTransient(Exception ex) =>
		ex is ServiceBusException { IsTransient: true }
			or ServiceBusException { Reason: ServiceBusFailureReason.MessagingEntityNotFound }
			or RequestFailedException { Status: 429 or 503 or 500 };

	public override void Dispose()
	{
		_pollingInterval.Dispose();
		base.Dispose();
	}
}

public sealed class OutboxMessage : ITenantable
{
	const int BaseDelayMinutes = 1;
	const int MaxBackoffMinutes = 60;
	
	[UsedImplicitly]
	OutboxMessage() { }

	public OutboxMessage(string type, string action, string content, Guid tenantId)
	{
		Id = IdentityGenerator.NewSequentialGuid();
		Type = type;
		Action = action;
		Content = content;
		TenantId = tenantId;
		OccurredOnUtc = DateTime.UtcNow;
	}
	
	public Guid Id { get; private set; }
	public string Type { get; private set; } = null!;
	public string Action { get; private set; } = null!;
	public string Content { get; private set; } = null!;
	public DateTime OccurredOnUtc { get; private set; }
	public DateTime? ProcessedOnUtc { get; private set; }
	public string? Error { get; private set; }
	
	// We assign this manually in the OutboxSaveChangesInterceptor.
	// Reason being: the EF Core tenant filter is applied after interceptors run, so this value must be set explicitly.
	public Guid TenantId { get; private set; }

	// Retry/backoff
	public int RetryCount { get; private set; }
	public DateTime? LastAttemptOnUtc { get; private set; }
	public DateTime? NextAttemptUtc { get; private set; }
	public bool IgnoreRetry { get; private set; }

	public void MarkAsProcessed()
	{
		ProcessedOnUtc = DateTime.UtcNow;
		Error = null;
		RetryCount = 0;
		NextAttemptUtc = null;
		LastAttemptOnUtc = DateTime.UtcNow;
	}

	// Mark the message as failed and schedule the next retry using exponential backoff.
	// Retry intervals double each time (1m, 2m, 4m, 8m, 16m, 32m, 60m...) and are capped at one hour.
	// Retries continue indefinitely unless ignoreRetry is set to true.
	public void MarkAsFailed(string error, bool ignoreRetry = false)
	{
		Error = error;
		IgnoreRetry = ignoreRetry;
		LastAttemptOnUtc = DateTime.UtcNow;
		RetryCount++;

		if (ignoreRetry)
		{
			NextAttemptUtc = null;
			return;
		}

		var backoffMinutes = Math.Min(
			BaseDelayMinutes * Math.Pow(2, RetryCount - 1),
			MaxBackoffMinutes);

		NextAttemptUtc = DateTime.UtcNow.AddMinutes(backoffMinutes);
	}
}

public static class OutboxActions
{
	public const string Created = nameof(Created);
	public const string Updated = nameof(Updated);
	public const string Deleted = nameof(Deleted);
}

// The problem we're solving here is when writing a Commodity to the DB, 
// we'd also need to send an integration event to notify the Webhook Service (Function App) about the same change. 
// The problem is the DB write and integration event publish are not atomic — one could succeed while the other fails.
//
// To solve this, we use the Transactional Outbox pattern. 
// We write the Commodity and the Outbox record within the same transaction, 
// so both the data and the event are saved together.
//
// A background process later reads those Outbox records and publishes them to Azure Service Bus.
// Azure Service Bus guarantees at-least-once delivery, which means retries may cause duplicate messages.
// Our consumer in the Function App should therefore be idempotent to handle possible duplicates.
//
// The core idea:
// 1. The CreateCommodityCommandHandle does one thing (e.g. saves a Commodity to the DB).
// 2. In the same transaction, it also writes an OutboxMessage (JSON event, metadata).
// 3. A background worker picks up unprocessed Outbox messages and publishes them to
// Azure Service Bus, and marks them as processed.
// 4. If publishing fails, it retries with backoff.
// 5. Consumers are idempotent (or use the Inbox pattern) to avoid double-processing.
public class OutboxSaveChangesInterceptor : SaveChangesInterceptor
{
	static readonly JsonSerializerOptions JsonSerializerOptions = new()
	{
		WriteIndented = false,
		Converters = { new JsonStringEnumConverter() }
	};
	
	public override InterceptionResult<int> SavingChanges(
		DbContextEventData eventData,
		InterceptionResult<int> result)
	{
		InsertOutboxMessages(eventData.Context);
		return base.SavingChanges(eventData, result);
	}

	public override ValueTask<InterceptionResult<int>> SavingChangesAsync(
		DbContextEventData eventData,
		InterceptionResult<int> result,
		CancellationToken cancellationToken = default)
	{
		InsertOutboxMessages(eventData.Context);
		return base.SavingChangesAsync(eventData, result, cancellationToken);
	}

	static void InsertOutboxMessages(DbContext? context)
	{
		if (context is not AppDbContext dbContext)
			return;

		if (dbContext.TenancyContext.Tenant is null)
			return;
		
		var outboxMessages = context.ChangeTracker
			.Entries<IOutboxTracked>()
			.Where(x => x.State is EntityState.Added or EntityState.Modified or EntityState.Deleted)
			.Select(entry =>
			{
				var changeType = entry.State switch
				{
					EntityState.Added => OutboxActions.Created,
					EntityState.Modified => OutboxActions.Updated,
					EntityState.Deleted => OutboxActions.Deleted,
					_ => throw new UnreachableException()
				};
				
				return new OutboxMessage(
					type: entry.Entity.GetType().Name,
					action: changeType,
					content: JsonSerializer.Serialize(entry.Entity, entry.Entity.GetType(), JsonSerializerOptions),
					tenantId: dbContext.TenancyContext.Tenant.Id
				);
			})
			.ToList();

		if (outboxMessages.Count == 0)
			return;

		context.Set<OutboxMessage>().AddRange(outboxMessages);
	}
}

Here is how I’ve implemented it in the past, but I still see some valuable ideas worth exploring in the MassTransit codebase, such as EF Core optimistic concurrency control (RowVersion) and adaptive batching, as well as the Inbox pattern

Hopefully that helps