From 73f9e6496fbf47d704cb1e166e8b2b0c0d6598b3 Mon Sep 17 00:00:00 2001 From: "Marcelo M. Maciel" <4993482+marcelo-maciel@users.noreply.github.com> Date: Sat, 11 Jul 2026 19:15:56 -0300 Subject: [PATCH] fix(eventing): back off outbox retries and make dead-letters recoverable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit REL-02: a failing outbox message was retried on every dispatch cycle (~10s) with no backoff, hit IsDead after 5 tries in ~50s, and was then lost — no replay path and no metric, only a log line. - Add NextRetryAt with exponential backoff (base 30s, cap 1h; configurable via EventingOptions). The dispatch batch skips messages whose NextRetryAt is in the future. - Add IOutboxStore.GetDeadLetteredAsync + RedriveDeadLettersAsync to inspect and reset dead-lettered messages for another attempt. - Add an OpenTelemetry meter (FSH.Eventing) with dead-letter and redrive counters so the condition is alertable, not just greppable in logs. Additive migration (nullable NextRetryAt column, Identity schema). Redis/infra unchanged. --- .../Eventing/EventingOptions.cs | 11 + .../Eventing/Outbox/EfCoreOutboxStore.cs | 67 +- .../Eventing/Outbox/IOutboxStore.cs | 10 + .../Eventing/Outbox/OutboxDispatcher.cs | 2 + .../Eventing/Outbox/OutboxMessage.cs | 7 + .../Eventing/Telemetry/EventingTelemetry.cs | 27 + .../Observability/OpenTelemetry/Extensions.cs | 4 + ...260711220711_OutboxNextRetryAt.Designer.cs | 842 ++++++++++++++++++ .../20260711220711_OutboxNextRetryAt.cs | 31 + .../IdentityDbContextModelSnapshot.cs | 3 + .../Tests/Eventing/OutboxRetryTests.cs | 94 ++ 11 files changed, 1093 insertions(+), 5 deletions(-) create mode 100644 src/BuildingBlocks/Eventing/Telemetry/EventingTelemetry.cs create mode 100644 src/Host/FSH.Starter.Migrations.PostgreSQL/Identity/20260711220711_OutboxNextRetryAt.Designer.cs create mode 100644 src/Host/FSH.Starter.Migrations.PostgreSQL/Identity/20260711220711_OutboxNextRetryAt.cs create mode 100644 src/Tests/Integration.Tests/Tests/Eventing/OutboxRetryTests.cs diff --git a/src/BuildingBlocks/Eventing/EventingOptions.cs b/src/BuildingBlocks/Eventing/EventingOptions.cs index c421562cbf..e86e379184 100644 --- a/src/BuildingBlocks/Eventing/EventingOptions.cs +++ b/src/BuildingBlocks/Eventing/EventingOptions.cs @@ -20,6 +20,17 @@ public sealed class EventingOptions /// public int OutboxMaxRetries { get; set; } = 5; + /// + /// Base delay (seconds) for exponential retry backoff after a failed dispatch. The n-th retry + /// waits base * 2^(n-1), capped at . + /// + public int OutboxRetryBaseDelaySeconds { get; set; } = 30; + + /// + /// Upper bound (seconds) on the exponential retry backoff. + /// + public int OutboxRetryMaxDelaySeconds { get; set; } = 3600; + /// /// Whether inbox-based idempotent handling is enabled. /// diff --git a/src/BuildingBlocks/Eventing/Outbox/EfCoreOutboxStore.cs b/src/BuildingBlocks/Eventing/Outbox/EfCoreOutboxStore.cs index eedfef4abd..134b4b033c 100644 --- a/src/BuildingBlocks/Eventing/Outbox/EfCoreOutboxStore.cs +++ b/src/BuildingBlocks/Eventing/Outbox/EfCoreOutboxStore.cs @@ -1,6 +1,7 @@ using FSH.Framework.Eventing.Abstractions; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; namespace FSH.Framework.Eventing.Outbox; @@ -15,17 +16,21 @@ public sealed class EfCoreOutboxStore : IOutboxStore private readonly IEventSerializer _serializer; private readonly ILogger> _logger; private readonly TimeProvider _timeProvider; + private readonly EventingOptions _options; public EfCoreOutboxStore( TDbContext dbContext, IEventSerializer serializer, ILogger> logger, - TimeProvider timeProvider) + TimeProvider timeProvider, + IOptions options) { + ArgumentNullException.ThrowIfNull(options); _dbContext = dbContext; _serializer = serializer; _logger = logger; _timeProvider = timeProvider; + _options = options.Value; } public async Task AddAsync(IIntegrationEvent @event, CancellationToken ct = default) @@ -51,8 +56,9 @@ public async Task AddAsync(IIntegrationEvent @event, CancellationToken ct = defa public async Task> GetPendingBatchAsync(int batchSize, CancellationToken ct = default) { + var now = _timeProvider.GetUtcNow().UtcDateTime; return await _dbContext.Set() - .Where(m => !m.IsDead && m.ProcessedOnUtc == null) + .Where(m => !m.IsDead && m.ProcessedOnUtc == null && (m.NextRetryAt == null || m.NextRetryAt <= now)) .OrderBy(m => m.CreatedOnUtc) .Take(batchSize) .ToListAsync(ct) @@ -75,11 +81,62 @@ public async Task MarkAsFailedAsync(OutboxMessage message, string error, bool is message.RetryCount++; message.LastError = error; message.IsDead = isDead; + // Space out retries with exponential backoff so a persistently failing message doesn't + // re-fire every dispatch cycle. A dead message won't be retried, so leave it eligible-null. + message.NextRetryAt = isDead ? null : _timeProvider.GetUtcNow().UtcDateTime.Add(BackoffFor(message.RetryCount)); _dbContext.Set().Update(message); - _logger.LogWarning("Outbox message {MessageId} failed. RetryCount={RetryCount}, IsDead={IsDead}, Error={Error}", - message.Id, message.RetryCount, message.IsDead, error); - await _dbContext.SaveChangesAsync(ct).ConfigureAwait(false); } + + public async Task> GetDeadLetteredAsync(int max, CancellationToken ct = default) + { + if (max <= 0) max = 100; + return await _dbContext.Set() + .Where(m => m.IsDead) + .OrderBy(m => m.CreatedOnUtc) + .Take(max) + .ToListAsync(ct) + .ConfigureAwait(false); + } + + public async Task RedriveDeadLettersAsync(IReadOnlyCollection? ids, CancellationToken ct = default) + { + var query = _dbContext.Set().Where(m => m.IsDead); + if (ids is { Count: > 0 }) + { + query = query.Where(m => ids.Contains(m.Id)); + } + + var dead = await query.ToListAsync(ct).ConfigureAwait(false); + foreach (var message in dead) + { + message.IsDead = false; + message.RetryCount = 0; + message.LastError = null; + message.NextRetryAt = null; + } + + if (dead.Count > 0) + { + await _dbContext.SaveChangesAsync(ct).ConfigureAwait(false); + Telemetry.EventingTelemetry.OutboxRedriven.Add(dead.Count); + if (_logger.IsEnabled(LogLevel.Information)) + { + _logger.LogInformation("Redrove {Count} dead-lettered outbox message(s) for another attempt.", dead.Count); + } + } + + return dead.Count; + } + + private TimeSpan BackoffFor(int retryCount) + { + var baseSeconds = _options.OutboxRetryBaseDelaySeconds > 0 ? _options.OutboxRetryBaseDelaySeconds : 30; + var maxSeconds = _options.OutboxRetryMaxDelaySeconds > 0 ? _options.OutboxRetryMaxDelaySeconds : 3600; + // retryCount is 1 after the first failure → first backoff is exactly baseSeconds. + var exponent = Math.Min(retryCount - 1, 30); // clamp so the shift can't overflow + var seconds = Math.Min((double)baseSeconds * Math.Pow(2, exponent), maxSeconds); + return TimeSpan.FromSeconds(seconds); + } } \ No newline at end of file diff --git a/src/BuildingBlocks/Eventing/Outbox/IOutboxStore.cs b/src/BuildingBlocks/Eventing/Outbox/IOutboxStore.cs index b3e493a1b8..a3df9dead2 100644 --- a/src/BuildingBlocks/Eventing/Outbox/IOutboxStore.cs +++ b/src/BuildingBlocks/Eventing/Outbox/IOutboxStore.cs @@ -14,4 +14,14 @@ public interface IOutboxStore Task MarkAsProcessedAsync(OutboxMessage message, CancellationToken ct = default); Task MarkAsFailedAsync(OutboxMessage message, string error, bool isDead, CancellationToken ct = default); + + /// Lists dead-lettered (exhausted) messages so an operator can inspect them. + Task> GetDeadLetteredAsync(int max, CancellationToken ct = default); + + /// + /// Resets dead-lettered messages for another dispatch attempt (clears the dead flag, retry + /// count, last error and backoff). Pass specific ids, or null to redrive them all. + /// Returns the number of messages redriven. + /// + Task RedriveDeadLettersAsync(IReadOnlyCollection? ids, CancellationToken ct = default); } \ No newline at end of file diff --git a/src/BuildingBlocks/Eventing/Outbox/OutboxDispatcher.cs b/src/BuildingBlocks/Eventing/Outbox/OutboxDispatcher.cs index f6bcd1f653..02c939f027 100644 --- a/src/BuildingBlocks/Eventing/Outbox/OutboxDispatcher.cs +++ b/src/BuildingBlocks/Eventing/Outbox/OutboxDispatcher.cs @@ -1,4 +1,5 @@ using FSH.Framework.Eventing.Abstractions; +using FSH.Framework.Eventing.Telemetry; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; @@ -80,6 +81,7 @@ public async Task DispatchAsync(CancellationToken ct = default) if (isDead) { deadLetterCount++; + EventingTelemetry.OutboxDeadLettered.Add(1); } if (isDead) diff --git a/src/BuildingBlocks/Eventing/Outbox/OutboxMessage.cs b/src/BuildingBlocks/Eventing/Outbox/OutboxMessage.cs index 64a459d840..5699b8577e 100644 --- a/src/BuildingBlocks/Eventing/Outbox/OutboxMessage.cs +++ b/src/BuildingBlocks/Eventing/Outbox/OutboxMessage.cs @@ -33,6 +33,13 @@ public class OutboxMessage : IGlobalEntity public string? LastError { get; set; } public bool IsDead { get; set; } + + /// + /// Earliest UTC time the message is eligible for another dispatch attempt. Set to a + /// backed-off future time after a failure so retries space out instead of re-firing every + /// dispatch cycle; null means immediately eligible (fresh message or after a redrive). + /// + public DateTime? NextRetryAt { get; set; } } public sealed class OutboxMessageConfiguration : IEntityTypeConfiguration diff --git a/src/BuildingBlocks/Eventing/Telemetry/EventingTelemetry.cs b/src/BuildingBlocks/Eventing/Telemetry/EventingTelemetry.cs new file mode 100644 index 0000000000..5610e0df53 --- /dev/null +++ b/src/BuildingBlocks/Eventing/Telemetry/EventingTelemetry.cs @@ -0,0 +1,27 @@ +using System.Diagnostics.Metrics; + +namespace FSH.Framework.Eventing.Telemetry; + +/// +/// OpenTelemetry metrics for the eventing building block. Register with +/// metrics.AddMeter(EventingTelemetry.MeterName) in the OTel setup. +/// +public static class EventingTelemetry +{ + /// Name of the used for eventing metrics. + public const string MeterName = "FSH.Eventing"; + + internal static readonly Meter Meter = new(MeterName); + + /// Outbox messages that exhausted their retries and were dead-lettered. + internal static readonly Counter OutboxDeadLettered = Meter.CreateCounter( + "fsh.eventing.outbox.deadlettered", + unit: "{message}", + description: "Number of outbox messages moved to dead-letter after exhausting retries."); + + /// Dead-lettered outbox messages reset for another dispatch attempt. + internal static readonly Counter OutboxRedriven = Meter.CreateCounter( + "fsh.eventing.outbox.redriven", + unit: "{message}", + description: "Number of dead-lettered outbox messages redriven for another attempt."); +} diff --git a/src/BuildingBlocks/Web/Observability/OpenTelemetry/Extensions.cs b/src/BuildingBlocks/Web/Observability/OpenTelemetry/Extensions.cs index 3ba011848b..6b9b95bb80 100644 --- a/src/BuildingBlocks/Web/Observability/OpenTelemetry/Extensions.cs +++ b/src/BuildingBlocks/Web/Observability/OpenTelemetry/Extensions.cs @@ -105,6 +105,10 @@ private static void ConfigureMetricsAndTracing( // Auditing pipeline metrics (published, dropped, flush, dead-letter). metrics.AddMeter("FSH.Modules.Auditing"); + // Eventing outbox metrics (dead-letter, redrive) — string literal matches + // EventingTelemetry.MeterName; Web does not reference the Eventing project. + metrics.AddMeter("FSH.Eventing"); + foreach (var meterName in options.Metrics.MeterNames ?? Array.Empty()) { metrics.AddMeter(meterName); diff --git a/src/Host/FSH.Starter.Migrations.PostgreSQL/Identity/20260711220711_OutboxNextRetryAt.Designer.cs b/src/Host/FSH.Starter.Migrations.PostgreSQL/Identity/20260711220711_OutboxNextRetryAt.Designer.cs new file mode 100644 index 0000000000..7bbda2f96d --- /dev/null +++ b/src/Host/FSH.Starter.Migrations.PostgreSQL/Identity/20260711220711_OutboxNextRetryAt.Designer.cs @@ -0,0 +1,842 @@ +// +using System; +using FSH.Modules.Identity.Data; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; +using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata; + +#nullable disable + +namespace FSH.Starter.Migrations.PostgreSQL.Identity +{ + [DbContext(typeof(IdentityDbContext))] + [Migration("20260711220711_OutboxNextRetryAt")] + partial class OutboxNextRetryAt + { + /// + protected override void BuildTargetModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasAnnotation("ProductVersion", "10.0.8") + .HasAnnotation("Relational:MaxIdentifierLength", 63); + + NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder); + + modelBuilder.Entity("FSH.Framework.Eventing.Inbox.InboxMessage", b => + { + b.Property("Id") + .HasColumnType("uuid"); + + b.Property("HandlerName") + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.Property("EventType") + .IsRequired() + .HasMaxLength(512) + .HasColumnType("character varying(512)"); + + b.Property("ProcessedOnUtc") + .HasColumnType("timestamp with time zone"); + + b.Property("TenantId") + .HasMaxLength(64) + .HasColumnType("character varying(64)"); + + b.HasKey("Id", "HandlerName"); + + b.ToTable("InboxMessages", "identity"); + }); + + modelBuilder.Entity("FSH.Framework.Eventing.Outbox.OutboxMessage", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("CorrelationId") + .HasMaxLength(128) + .HasColumnType("character varying(128)"); + + b.Property("CreatedOnUtc") + .HasColumnType("timestamp with time zone"); + + b.Property("IsDead") + .HasColumnType("boolean"); + + b.Property("LastError") + .HasColumnType("text"); + + b.Property("NextRetryAt") + .HasColumnType("timestamp with time zone"); + + b.Property("Payload") + .IsRequired() + .HasColumnType("text"); + + b.Property("ProcessedOnUtc") + .HasColumnType("timestamp with time zone"); + + b.Property("RetryCount") + .HasColumnType("integer"); + + b.Property("TenantId") + .HasMaxLength(64) + .HasColumnType("character varying(64)"); + + b.Property("Type") + .IsRequired() + .HasMaxLength(512) + .HasColumnType("character varying(512)"); + + b.HasKey("Id"); + + b.ToTable("OutboxMessages", "identity"); + }); + + modelBuilder.Entity("FSH.Modules.Identity.Domain.FshRole", b => + { + b.Property("Id") + .HasColumnType("text"); + + b.Property("ConcurrencyStamp") + .IsConcurrencyToken() + .HasColumnType("text"); + + b.Property("Description") + .HasColumnType("text"); + + b.Property("Name") + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.Property("NormalizedName") + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.Property("TenantId") + .IsRequired() + .HasColumnType("text"); + + b.HasKey("Id"); + + b.HasIndex("NormalizedName", "TenantId") + .IsUnique() + .HasDatabaseName("RoleNameIndex"); + + b.ToTable("Roles", "identity"); + + b.HasAnnotation("Finbuckle:MultiTenant", true); + }); + + modelBuilder.Entity("FSH.Modules.Identity.Domain.FshRoleClaim", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("ClaimType") + .HasColumnType("text"); + + b.Property("ClaimValue") + .HasColumnType("text"); + + b.Property("CreatedBy") + .HasColumnType("text"); + + b.Property("CreatedOn") + .HasColumnType("timestamp with time zone"); + + b.Property("RoleId") + .IsRequired() + .HasColumnType("text"); + + b.Property("TenantId") + .IsRequired() + .HasColumnType("text"); + + b.HasKey("Id"); + + b.HasIndex("RoleId"); + + b.ToTable("RoleClaims", "identity"); + + b.HasAnnotation("Finbuckle:MultiTenant", true); + }); + + modelBuilder.Entity("FSH.Modules.Identity.Domain.FshUser", b => + { + b.Property("Id") + .HasColumnType("text"); + + b.Property("AccessFailedCount") + .HasColumnType("integer"); + + b.Property("ConcurrencyStamp") + .IsConcurrencyToken() + .HasColumnType("text"); + + b.Property("Email") + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.Property("EmailConfirmed") + .HasColumnType("boolean"); + + b.Property("FirstName") + .HasColumnType("text"); + + b.Property("ImageUrl") + .HasColumnType("text"); + + b.Property("IsActive") + .HasColumnType("boolean"); + + b.Property("LastName") + .HasColumnType("text"); + + b.Property("LastPasswordChangeDate") + .HasColumnType("timestamp with time zone"); + + b.Property("LockoutEnabled") + .HasColumnType("boolean"); + + b.Property("LockoutEnd") + .HasColumnType("timestamp with time zone"); + + b.Property("NormalizedEmail") + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.Property("NormalizedUserName") + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.Property("ObjectId") + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.Property("PasswordHash") + .HasColumnType("text"); + + b.Property("PhoneNumber") + .HasColumnType("text"); + + b.Property("PhoneNumberConfirmed") + .HasColumnType("boolean"); + + b.Property("RefreshToken") + .HasColumnType("text"); + + b.Property("RefreshTokenExpiryTime") + .HasColumnType("timestamp with time zone"); + + b.Property("SecurityStamp") + .HasColumnType("text"); + + b.Property("TenantId") + .IsRequired() + .HasColumnType("text"); + + b.Property("TwoFactorEnabled") + .HasColumnType("boolean"); + + b.Property("UserName") + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.HasKey("Id"); + + b.HasIndex("NormalizedEmail") + .HasDatabaseName("EmailIndex"); + + b.HasIndex("NormalizedUserName", "TenantId") + .IsUnique() + .HasDatabaseName("UserNameIndex"); + + b.ToTable("Users", "identity"); + + b.HasAnnotation("Finbuckle:MultiTenant", true); + }); + + modelBuilder.Entity("FSH.Modules.Identity.Domain.Group", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("CreatedBy") + .HasMaxLength(450) + .HasColumnType("character varying(450)"); + + b.Property("CreatedOnUtc") + .ValueGeneratedOnAdd() + .HasColumnType("timestamp with time zone") + .HasColumnName("CreatedAt") + .HasDefaultValueSql("CURRENT_TIMESTAMP"); + + b.Property("DeletedBy") + .HasMaxLength(450) + .HasColumnType("character varying(450)"); + + b.Property("DeletedOnUtc") + .HasColumnType("timestamp with time zone"); + + b.Property("Description") + .HasMaxLength(1024) + .HasColumnType("character varying(1024)"); + + b.Property("IsDefault") + .HasColumnType("boolean"); + + b.Property("IsDeleted") + .HasColumnType("boolean"); + + b.Property("IsSystemGroup") + .HasColumnType("boolean"); + + b.Property("LastModifiedBy") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("ModifiedBy"); + + b.Property("LastModifiedOnUtc") + .HasColumnType("timestamp with time zone") + .HasColumnName("ModifiedAt"); + + b.Property("Name") + .IsRequired() + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.Property("TenantId") + .IsRequired() + .HasColumnType("text"); + + b.HasKey("Id"); + + b.HasIndex("IsDefault"); + + b.HasIndex("IsDeleted"); + + b.HasIndex("Name"); + + b.ToTable("Groups", "identity"); + + b.HasAnnotation("Finbuckle:MultiTenant", true); + }); + + modelBuilder.Entity("FSH.Modules.Identity.Domain.GroupRole", b => + { + b.Property("GroupId") + .HasColumnType("uuid"); + + b.Property("RoleId") + .HasMaxLength(450) + .HasColumnType("character varying(450)"); + + b.Property("TenantId") + .IsRequired() + .HasColumnType("text"); + + b.HasKey("GroupId", "RoleId"); + + b.HasIndex("GroupId"); + + b.HasIndex("RoleId"); + + b.ToTable("GroupRoles", "identity"); + + b.HasAnnotation("Finbuckle:MultiTenant", true); + }); + + modelBuilder.Entity("FSH.Modules.Identity.Domain.ImpersonationGrant", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("ActorTenantId") + .IsRequired() + .HasMaxLength(64) + .HasColumnType("character varying(64)"); + + b.Property("ActorUserId") + .IsRequired() + .HasMaxLength(64) + .HasColumnType("character varying(64)"); + + b.Property("ActorUserName") + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.Property("ClientId") + .HasMaxLength(128) + .HasColumnType("character varying(128)"); + + b.Property("EndedAtUtc") + .HasColumnType("timestamp with time zone"); + + b.Property("ExpiresAtUtc") + .HasColumnType("timestamp with time zone"); + + b.Property("ImpersonatedTenantId") + .IsRequired() + .HasMaxLength(64) + .HasColumnType("character varying(64)"); + + b.Property("ImpersonatedUserId") + .IsRequired() + .HasMaxLength(64) + .HasColumnType("character varying(64)"); + + b.Property("ImpersonatedUserName") + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.Property("IpAddress") + .HasMaxLength(64) + .HasColumnType("character varying(64)"); + + b.Property("Jti") + .IsRequired() + .HasMaxLength(64) + .HasColumnType("character varying(64)"); + + b.Property("Reason") + .IsRequired() + .HasMaxLength(500) + .HasColumnType("character varying(500)"); + + b.Property("RevokeReason") + .HasMaxLength(500) + .HasColumnType("character varying(500)"); + + b.Property("RevokedAtUtc") + .HasColumnType("timestamp with time zone"); + + b.Property("RevokedByUserId") + .HasMaxLength(64) + .HasColumnType("character varying(64)"); + + b.Property("RevokedByUserName") + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.Property("StartedAtUtc") + .HasColumnType("timestamp with time zone"); + + b.Property("UserAgent") + .HasMaxLength(512) + .HasColumnType("character varying(512)"); + + b.HasKey("Id"); + + b.HasIndex("Jti") + .IsUnique(); + + b.HasIndex("ActorUserId", "StartedAtUtc") + .HasDatabaseName("IX_ImpersonationGrants_ActorUserId_StartedAtUtc"); + + b.HasIndex("ImpersonatedTenantId", "StartedAtUtc") + .HasDatabaseName("IX_ImpersonationGrants_ImpersonatedTenantId_StartedAtUtc"); + + b.ToTable("ImpersonationGrants", "identity"); + }); + + modelBuilder.Entity("FSH.Modules.Identity.Domain.PasswordHistory", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("CreatedAt") + .ValueGeneratedOnAdd() + .HasColumnType("timestamp with time zone") + .HasDefaultValueSql("CURRENT_TIMESTAMP"); + + b.Property("PasswordHash") + .IsRequired() + .HasColumnType("text"); + + b.Property("TenantId") + .IsRequired() + .HasColumnType("text"); + + b.Property("UserId") + .IsRequired() + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.HasKey("Id"); + + b.HasIndex("UserId"); + + b.HasIndex("UserId", "CreatedAt"); + + b.ToTable("PasswordHistory", "identity"); + + b.HasAnnotation("Finbuckle:MultiTenant", true); + }); + + modelBuilder.Entity("FSH.Modules.Identity.Domain.UserGroup", b => + { + b.Property("UserId") + .HasMaxLength(450) + .HasColumnType("character varying(450)"); + + b.Property("GroupId") + .HasColumnType("uuid"); + + b.Property("AddedAt") + .ValueGeneratedOnAdd() + .HasColumnType("timestamp with time zone") + .HasDefaultValueSql("CURRENT_TIMESTAMP"); + + b.Property("AddedBy") + .HasMaxLength(450) + .HasColumnType("character varying(450)"); + + b.Property("TenantId") + .IsRequired() + .HasColumnType("text"); + + b.HasKey("UserId", "GroupId"); + + b.HasIndex("GroupId"); + + b.HasIndex("UserId"); + + b.ToTable("UserGroups", "identity"); + + b.HasAnnotation("Finbuckle:MultiTenant", true); + }); + + modelBuilder.Entity("FSH.Modules.Identity.Domain.UserSession", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("Browser") + .HasMaxLength(100) + .HasColumnType("character varying(100)"); + + b.Property("BrowserVersion") + .HasMaxLength(50) + .HasColumnType("character varying(50)"); + + b.Property("CreatedAt") + .ValueGeneratedOnAdd() + .HasColumnType("timestamp with time zone") + .HasDefaultValueSql("CURRENT_TIMESTAMP"); + + b.Property("DeviceType") + .HasMaxLength(50) + .HasColumnType("character varying(50)"); + + b.Property("ExpiresAt") + .HasColumnType("timestamp with time zone"); + + b.Property("IpAddress") + .IsRequired() + .HasMaxLength(45) + .HasColumnType("character varying(45)"); + + b.Property("IsRevoked") + .HasColumnType("boolean"); + + b.Property("LastActivityAt") + .HasColumnType("timestamp with time zone"); + + b.Property("OperatingSystem") + .HasMaxLength(100) + .HasColumnType("character varying(100)"); + + b.Property("OsVersion") + .HasMaxLength(50) + .HasColumnType("character varying(50)"); + + b.Property("RefreshTokenHash") + .IsRequired() + .HasMaxLength(256) + .HasColumnType("character varying(256)"); + + b.Property("RevokedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("RevokedBy") + .HasMaxLength(450) + .HasColumnType("character varying(450)"); + + b.Property("RevokedReason") + .HasMaxLength(500) + .HasColumnType("character varying(500)"); + + b.Property("TenantId") + .IsRequired() + .HasColumnType("text"); + + b.Property("UserAgent") + .IsRequired() + .HasMaxLength(1024) + .HasColumnType("character varying(1024)"); + + b.Property("UserId") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)"); + + b.HasKey("Id"); + + b.HasIndex("ExpiresAt"); + + b.HasIndex("RefreshTokenHash"); + + b.HasIndex("UserId"); + + b.HasIndex("UserId", "IsRevoked"); + + b.ToTable("UserSessions", "identity"); + + b.HasAnnotation("Finbuckle:MultiTenant", true); + }); + + modelBuilder.Entity("Microsoft.AspNetCore.Identity.IdentityUserClaim", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("ClaimType") + .HasColumnType("text"); + + b.Property("ClaimValue") + .HasColumnType("text"); + + b.Property("TenantId") + .IsRequired() + .HasColumnType("text"); + + b.Property("UserId") + .IsRequired() + .HasColumnType("text"); + + b.HasKey("Id"); + + b.HasIndex("UserId"); + + b.ToTable("UserClaims", "identity"); + + b.HasAnnotation("Finbuckle:MultiTenant", true); + }); + + modelBuilder.Entity("Microsoft.AspNetCore.Identity.IdentityUserLogin", b => + { + b.Property("LoginProvider") + .HasColumnType("text"); + + b.Property("ProviderKey") + .HasColumnType("text"); + + b.Property("ProviderDisplayName") + .HasColumnType("text"); + + b.Property("TenantId") + .IsRequired() + .HasColumnType("text"); + + b.Property("UserId") + .IsRequired() + .HasColumnType("text"); + + b.HasKey("LoginProvider", "ProviderKey"); + + b.HasIndex("UserId"); + + b.ToTable("UserLogins", "identity"); + + b.HasAnnotation("Finbuckle:MultiTenant", true); + }); + + modelBuilder.Entity("Microsoft.AspNetCore.Identity.IdentityUserRole", b => + { + b.Property("UserId") + .HasColumnType("text"); + + b.Property("RoleId") + .HasColumnType("text"); + + b.Property("TenantId") + .IsRequired() + .HasColumnType("text"); + + b.HasKey("UserId", "RoleId"); + + b.HasIndex("RoleId"); + + b.ToTable("UserRoles", "identity"); + + b.HasAnnotation("Finbuckle:MultiTenant", true); + }); + + modelBuilder.Entity("Microsoft.AspNetCore.Identity.IdentityUserToken", b => + { + b.Property("UserId") + .HasColumnType("text"); + + b.Property("LoginProvider") + .HasColumnType("text"); + + b.Property("Name") + .HasColumnType("text"); + + b.Property("TenantId") + .IsRequired() + .HasColumnType("text"); + + b.Property("Value") + .HasColumnType("text"); + + b.HasKey("UserId", "LoginProvider", "Name"); + + b.ToTable("UserTokens", "identity"); + + b.HasAnnotation("Finbuckle:MultiTenant", true); + }); + + modelBuilder.Entity("FSH.Modules.Identity.Domain.FshRoleClaim", b => + { + b.HasOne("FSH.Modules.Identity.Domain.FshRole", null) + .WithMany() + .HasForeignKey("RoleId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + }); + + modelBuilder.Entity("FSH.Modules.Identity.Domain.GroupRole", b => + { + b.HasOne("FSH.Modules.Identity.Domain.Group", "Group") + .WithMany("GroupRoles") + .HasForeignKey("GroupId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.HasOne("FSH.Modules.Identity.Domain.FshRole", "Role") + .WithMany() + .HasForeignKey("RoleId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.Navigation("Group"); + + b.Navigation("Role"); + }); + + modelBuilder.Entity("FSH.Modules.Identity.Domain.PasswordHistory", b => + { + b.HasOne("FSH.Modules.Identity.Domain.FshUser", "User") + .WithMany("PasswordHistories") + .HasForeignKey("UserId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.Navigation("User"); + }); + + modelBuilder.Entity("FSH.Modules.Identity.Domain.UserGroup", b => + { + b.HasOne("FSH.Modules.Identity.Domain.Group", "Group") + .WithMany("UserGroups") + .HasForeignKey("GroupId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.HasOne("FSH.Modules.Identity.Domain.FshUser", "User") + .WithMany() + .HasForeignKey("UserId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.Navigation("Group"); + + b.Navigation("User"); + }); + + modelBuilder.Entity("FSH.Modules.Identity.Domain.UserSession", b => + { + b.HasOne("FSH.Modules.Identity.Domain.FshUser", "User") + .WithMany() + .HasForeignKey("UserId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.Navigation("User"); + }); + + modelBuilder.Entity("Microsoft.AspNetCore.Identity.IdentityUserClaim", b => + { + b.HasOne("FSH.Modules.Identity.Domain.FshUser", null) + .WithMany() + .HasForeignKey("UserId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + }); + + modelBuilder.Entity("Microsoft.AspNetCore.Identity.IdentityUserLogin", b => + { + b.HasOne("FSH.Modules.Identity.Domain.FshUser", null) + .WithMany() + .HasForeignKey("UserId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + }); + + modelBuilder.Entity("Microsoft.AspNetCore.Identity.IdentityUserRole", b => + { + b.HasOne("FSH.Modules.Identity.Domain.FshRole", null) + .WithMany() + .HasForeignKey("RoleId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.HasOne("FSH.Modules.Identity.Domain.FshUser", null) + .WithMany() + .HasForeignKey("UserId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + }); + + modelBuilder.Entity("Microsoft.AspNetCore.Identity.IdentityUserToken", b => + { + b.HasOne("FSH.Modules.Identity.Domain.FshUser", null) + .WithMany() + .HasForeignKey("UserId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + }); + + modelBuilder.Entity("FSH.Modules.Identity.Domain.FshUser", b => + { + b.Navigation("PasswordHistories"); + }); + + modelBuilder.Entity("FSH.Modules.Identity.Domain.Group", b => + { + b.Navigation("GroupRoles"); + + b.Navigation("UserGroups"); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/src/Host/FSH.Starter.Migrations.PostgreSQL/Identity/20260711220711_OutboxNextRetryAt.cs b/src/Host/FSH.Starter.Migrations.PostgreSQL/Identity/20260711220711_OutboxNextRetryAt.cs new file mode 100644 index 0000000000..8864efdb06 --- /dev/null +++ b/src/Host/FSH.Starter.Migrations.PostgreSQL/Identity/20260711220711_OutboxNextRetryAt.cs @@ -0,0 +1,31 @@ +using System; +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace FSH.Starter.Migrations.PostgreSQL.Identity +{ + /// + public partial class OutboxNextRetryAt : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.AddColumn( + name: "NextRetryAt", + schema: "identity", + table: "OutboxMessages", + type: "timestamp with time zone", + nullable: true); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropColumn( + name: "NextRetryAt", + schema: "identity", + table: "OutboxMessages"); + } + } +} diff --git a/src/Host/FSH.Starter.Migrations.PostgreSQL/Identity/IdentityDbContextModelSnapshot.cs b/src/Host/FSH.Starter.Migrations.PostgreSQL/Identity/IdentityDbContextModelSnapshot.cs index 88a14a6c4a..03f737d0ee 100644 --- a/src/Host/FSH.Starter.Migrations.PostgreSQL/Identity/IdentityDbContextModelSnapshot.cs +++ b/src/Host/FSH.Starter.Migrations.PostgreSQL/Identity/IdentityDbContextModelSnapshot.cs @@ -67,6 +67,9 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.Property("LastError") .HasColumnType("text"); + b.Property("NextRetryAt") + .HasColumnType("timestamp with time zone"); + b.Property("Payload") .IsRequired() .HasColumnType("text"); diff --git a/src/Tests/Integration.Tests/Tests/Eventing/OutboxRetryTests.cs b/src/Tests/Integration.Tests/Tests/Eventing/OutboxRetryTests.cs new file mode 100644 index 0000000000..74251a3b1c --- /dev/null +++ b/src/Tests/Integration.Tests/Tests/Eventing/OutboxRetryTests.cs @@ -0,0 +1,94 @@ +using Finbuckle.MultiTenant; +using Finbuckle.MultiTenant.Abstractions; +using FSH.Framework.Eventing.Outbox; +using FSH.Framework.Shared.Multitenancy; +using FSH.Modules.Identity.Data; +using Integration.Tests.Infrastructure; + +namespace Integration.Tests.Tests.Eventing; + +/// +/// Covers audit finding REL-02: a failed outbox message was retried every dispatch cycle with no +/// backoff, then dead-lettered with no way to recover it. Exercises the real +/// (Identity owns the OutboxMessages set) over Postgres. +/// +[Collection(FshCollectionDefinition.Name)] +public sealed class OutboxRetryTests +{ + private readonly FshWebApplicationFactory _factory; + + public OutboxRetryTests(FshWebApplicationFactory factory) + { + _factory = factory; + } + + [Fact] + public async Task MarkAsFailed_NotDead_Should_BackOff_And_ExcludeFromPendingUntilDue() + { + using var scope = await CreateTenantScopeAsync(); + var db = scope.ServiceProvider.GetRequiredService(); + var store = scope.ServiceProvider.GetRequiredService(); + + var message = NewMessage(); + db.Set().Add(message); + await db.SaveChangesAsync(); + + await store.MarkAsFailedAsync(message, "transient boom", isDead: false); + + message.NextRetryAt.ShouldNotBeNull("a non-dead failure must schedule a backed-off retry"); + message.NextRetryAt!.Value.ShouldBeGreaterThan( + DateTime.UtcNow.AddSeconds(20), + "the first retry backs off by the base delay (30s), not the next 10s cycle"); + + var pending = await store.GetPendingBatchAsync(500); + pending.ShouldNotContain( + m => m.Id == message.Id, + "a message whose NextRetryAt is in the future must be excluded from the dispatch batch"); + } + + [Fact] + public async Task RedriveDeadLetters_Should_ResetDeadMessage_And_MakeItPendingAgain() + { + using var scope = await CreateTenantScopeAsync(); + var db = scope.ServiceProvider.GetRequiredService(); + var store = scope.ServiceProvider.GetRequiredService(); + + var message = NewMessage(); + message.IsDead = true; + message.RetryCount = 5; + message.LastError = "exhausted"; + db.Set().Add(message); + await db.SaveChangesAsync(); + + var redriven = await store.RedriveDeadLettersAsync([message.Id]); + + redriven.ShouldBe(1); + (await store.GetDeadLetteredAsync(500)).ShouldNotContain(m => m.Id == message.Id); + (await store.GetPendingBatchAsync(500)).ShouldContain( + m => m.Id == message.Id, + "a redriven message clears IsDead/RetryCount/NextRetryAt and becomes eligible again"); + } + + private static OutboxMessage NewMessage() => new() + { + Id = Guid.NewGuid(), + // Old timestamp so it sorts first in the CreatedOnUtc-ascending pending batch. + CreatedOnUtc = DateTime.UtcNow.AddDays(-1), + Type = "REL-02.OutboxRetryTest", + Payload = "{}", + TenantId = TestConstants.RootTenantId, + RetryCount = 0, + IsDead = false, + }; + + private async Task CreateTenantScopeAsync() + { + var scope = _factory.Services.CreateScope(); + var tenant = await scope.ServiceProvider + .GetRequiredService>() + .GetAsync(TestConstants.RootTenantId); + scope.ServiceProvider.GetRequiredService() + .MultiTenantContext = new MultiTenantContext(tenant); + return scope; + } +}