fix(email): reconcile abandoned sends
Age stale pre-delivery attempts to failed and in-delivery attempts to uncertain without provider I/O. Notify each owner with content-free guidance and keep recovery idempotent across replicas and restarts.
This commit is contained in:
@@ -58,12 +58,49 @@ public sealed class EmailSendAttemptStoreTests
|
|||||||
property.Name.Contains("Recipient", StringComparison.OrdinalIgnoreCase));
|
property.Name.Contains("Recipient", StringComparison.OrdinalIgnoreCase));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task Restart_recovery_is_tenant_visible_idempotent_and_never_requeues_delivery()
|
||||||
|
{
|
||||||
|
await using var fixture = await Fixture.CreateAsync();
|
||||||
|
var ownerOne = fixture.Store("user-1");
|
||||||
|
var ownerTwo = fixture.Store("user-2");
|
||||||
|
var oldSending = await ownerOne.CreateAsync(new(fixture.JobId, "gmail", Guid.NewGuid().ToString(), Hash('e')), default);
|
||||||
|
var oldPending = await ownerTwo.CreateAsync(new(fixture.OtherJobId, "microsoft", Guid.NewGuid().ToString(), Hash('f')), default);
|
||||||
|
Assert.Equal(1, await ownerOne.BeginAsync(oldSending.Attempt.Id, default));
|
||||||
|
|
||||||
|
fixture.Time.Advance(EmailSendAttemptStore.AbandonedAge + TimeSpan.FromSeconds(1));
|
||||||
|
var freshPending = await ownerOne.CreateAsync(new(fixture.JobId, "gmail", Guid.NewGuid().ToString(), Hash('1')), default);
|
||||||
|
var freshSending = await ownerTwo.CreateAsync(new(fixture.OtherJobId, "microsoft", Guid.NewGuid().ToString(), Hash('2')), default);
|
||||||
|
Assert.Equal(1, await ownerTwo.BeginAsync(freshSending.Attempt.Id, default));
|
||||||
|
|
||||||
|
var recovered = await fixture.Store(null).ReconcileAbandonedAsync(default);
|
||||||
|
var repeated = await fixture.Store(null).ReconcileAbandonedAsync(default);
|
||||||
|
|
||||||
|
Assert.Equal(new EmailSendAttemptRecoveryResult(1, 1), recovered);
|
||||||
|
Assert.Equal(new EmailSendAttemptRecoveryResult(0, 0), repeated);
|
||||||
|
Assert.Equal(EmailSendStatuses.Uncertain, (await ownerOne.GetAsync(oldSending.Attempt.Id, default))!.Status);
|
||||||
|
Assert.Equal(EmailSendStatuses.Failed, (await ownerTwo.GetAsync(oldPending.Attempt.Id, default))!.Status);
|
||||||
|
Assert.Equal(EmailSendStatuses.Pending, (await ownerOne.GetAsync(freshPending.Attempt.Id, default))!.Status);
|
||||||
|
Assert.Equal(EmailSendStatuses.Sending, (await ownerTwo.GetAsync(freshSending.Attempt.Id, default))!.Status);
|
||||||
|
|
||||||
|
var userOneNotifications = await fixture.Notifications("user-1").ListAsync(10, default);
|
||||||
|
var userTwoNotifications = await fixture.Notifications("user-2").ListAsync(10, default);
|
||||||
|
Assert.Single(userOneNotifications);
|
||||||
|
Assert.Single(userTwoNotifications);
|
||||||
|
Assert.Equal("email.send.uncertain", userOneNotifications[0].Kind);
|
||||||
|
Assert.Equal("email.send.stopped", userTwoNotifications[0].Kind);
|
||||||
|
Assert.DoesNotContain("gmail", userOneNotifications[0].Message, StringComparison.OrdinalIgnoreCase);
|
||||||
|
Assert.DoesNotContain("microsoft", userTwoNotifications[0].Message, StringComparison.OrdinalIgnoreCase);
|
||||||
|
}
|
||||||
|
|
||||||
private static string Hash(char value) => new(value, 64);
|
private static string Hash(char value) => new(value, 64);
|
||||||
|
|
||||||
private sealed class Fixture(SqliteConnection connection, DbContextOptions<JobTrackerContext> options, int jobId) : IAsyncDisposable
|
private sealed class Fixture(SqliteConnection connection, DbContextOptions<JobTrackerContext> options, int jobId, int otherJobId) : IAsyncDisposable
|
||||||
{
|
{
|
||||||
private readonly List<JobTrackerContext> contexts = new();
|
private readonly List<JobTrackerContext> contexts = new();
|
||||||
|
public ManualTimeProvider Time { get; } = new(new DateTimeOffset(2026, 8, 10, 12, 0, 0, TimeSpan.Zero));
|
||||||
public int JobId { get; } = jobId;
|
public int JobId { get; } = jobId;
|
||||||
|
public int OtherJobId { get; } = otherJobId;
|
||||||
|
|
||||||
public static async Task<Fixture> CreateAsync()
|
public static async Task<Fixture> CreateAsync()
|
||||||
{
|
{
|
||||||
@@ -78,17 +115,31 @@ public sealed class EmailSendAttemptStoreTests
|
|||||||
var job = new JobApplication { JobTitle = "Backend", CompanyId = company.Id, OwnerUserId = "user-1" };
|
var job = new JobApplication { JobTitle = "Backend", CompanyId = company.Id, OwnerUserId = "user-1" };
|
||||||
db.JobApplications.Add(job);
|
db.JobApplications.Add(job);
|
||||||
await db.SaveChangesAsync();
|
await db.SaveChangesAsync();
|
||||||
return new Fixture(connection, options, job.Id);
|
await using var otherDb = CreateDb(options, "user-2");
|
||||||
|
var otherCompany = new Company { Name = "Other", OwnerUserId = "user-2" };
|
||||||
|
otherDb.Companies.Add(otherCompany);
|
||||||
|
await otherDb.SaveChangesAsync();
|
||||||
|
var otherJob = new JobApplication { JobTitle = "Frontend", CompanyId = otherCompany.Id, OwnerUserId = "user-2" };
|
||||||
|
otherDb.JobApplications.Add(otherJob);
|
||||||
|
await otherDb.SaveChangesAsync();
|
||||||
|
return new Fixture(connection, options, job.Id, otherJob.Id);
|
||||||
}
|
}
|
||||||
|
|
||||||
public EmailSendAttemptStore Store(string userId)
|
public EmailSendAttemptStore Store(string? userId)
|
||||||
{
|
{
|
||||||
var db = CreateDb(options, userId);
|
var db = CreateDb(options, userId);
|
||||||
contexts.Add(db);
|
contexts.Add(db);
|
||||||
return new EmailSendAttemptStore(db, TimeProvider.System);
|
return new EmailSendAttemptStore(db, Time);
|
||||||
}
|
}
|
||||||
|
|
||||||
private static JobTrackerContext CreateDb(DbContextOptions<JobTrackerContext> options, string userId)
|
public UserNotificationStore Notifications(string userId)
|
||||||
|
{
|
||||||
|
var db = CreateDb(options, userId);
|
||||||
|
contexts.Add(db);
|
||||||
|
return new UserNotificationStore(db, Time);
|
||||||
|
}
|
||||||
|
|
||||||
|
private static JobTrackerContext CreateDb(DbContextOptions<JobTrackerContext> options, string? userId)
|
||||||
{
|
{
|
||||||
var currentUser = new Mock<ICurrentUserService>();
|
var currentUser = new Mock<ICurrentUserService>();
|
||||||
currentUser.SetupGet(service => service.UserId).Returns(userId);
|
currentUser.SetupGet(service => service.UserId).Returns(userId);
|
||||||
@@ -101,4 +152,11 @@ public sealed class EmailSendAttemptStoreTests
|
|||||||
await connection.DisposeAsync();
|
await connection.DisposeAsync();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private sealed class ManualTimeProvider(DateTimeOffset now) : TimeProvider
|
||||||
|
{
|
||||||
|
public DateTimeOffset UtcNow { get; private set; } = now;
|
||||||
|
public override DateTimeOffset GetUtcNow() => UtcNow;
|
||||||
|
public void Advance(TimeSpan duration) => UtcNow += duration;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -168,6 +168,7 @@ builder.Services.AddHostedService<DailyExportHostedService>();
|
|||||||
builder.Services.AddHostedService<JobEnrichmentHostedService>();
|
builder.Services.AddHostedService<JobEnrichmentHostedService>();
|
||||||
builder.Services.AddHostedService<SummarizerProbeHostedService>();
|
builder.Services.AddHostedService<SummarizerProbeHostedService>();
|
||||||
builder.Services.AddHostedService<AiOperationHostedService>();
|
builder.Services.AddHostedService<AiOperationHostedService>();
|
||||||
|
builder.Services.AddHostedService<EmailSendAttemptRecoveryHostedService>();
|
||||||
|
|
||||||
builder.Services.AddHttpClient("jobimport")
|
builder.Services.AddHttpClient("jobimport")
|
||||||
.ConfigurePrimaryHttpMessageHandler(() => new HttpClientHandler
|
.ConfigurePrimaryHttpMessageHandler(() => new HttpClientHandler
|
||||||
|
|||||||
@@ -0,0 +1,44 @@
|
|||||||
|
namespace JobTrackerApi.Services;
|
||||||
|
|
||||||
|
public sealed class EmailSendAttemptRecoveryHostedService(
|
||||||
|
IServiceScopeFactory scopes,
|
||||||
|
IStartupReadiness startupReadiness,
|
||||||
|
ILogger<EmailSendAttemptRecoveryHostedService> logger) : BackgroundService
|
||||||
|
{
|
||||||
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||||
|
{
|
||||||
|
await startupReadiness.WaitUntilReadyAsync(stoppingToken);
|
||||||
|
while (!stoppingToken.IsCancellationRequested)
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
await using var scope = scopes.CreateAsyncScope();
|
||||||
|
var result = await scope.ServiceProvider.GetRequiredService<EmailSendAttemptStore>()
|
||||||
|
.ReconcileAbandonedAsync(stoppingToken);
|
||||||
|
if (result.FailedPending > 0 || result.UncertainSending > 0)
|
||||||
|
{
|
||||||
|
logger.LogWarning(
|
||||||
|
"Recovered abandoned email attempts: failedPending={FailedPending}, uncertainSending={UncertainSending}. No provider retry was attempted.",
|
||||||
|
result.FailedPending, result.UncertainSending);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
||||||
|
{
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
catch (Exception ex)
|
||||||
|
{
|
||||||
|
logger.LogError(ex, "Email send-attempt recovery failed; provider delivery was not attempted.");
|
||||||
|
}
|
||||||
|
|
||||||
|
try
|
||||||
|
{
|
||||||
|
await Task.Delay(TimeSpan.FromMinutes(5), stoppingToken);
|
||||||
|
}
|
||||||
|
catch (OperationCanceledException)
|
||||||
|
{
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -6,11 +6,13 @@ namespace JobTrackerApi.Services;
|
|||||||
|
|
||||||
public sealed record CreateEmailSendAttempt(int JobApplicationId, string Provider, string ClientRequestId, string PayloadHash);
|
public sealed record CreateEmailSendAttempt(int JobApplicationId, string Provider, string ClientRequestId, string PayloadHash);
|
||||||
public sealed record EmailSendAttemptCreation(EmailSendAttempt Attempt, bool Created);
|
public sealed record EmailSendAttemptCreation(EmailSendAttempt Attempt, bool Created);
|
||||||
|
public sealed record EmailSendAttemptRecoveryResult(int FailedPending, int UncertainSending);
|
||||||
|
|
||||||
public sealed class EmailSendConflictException(string message) : InvalidOperationException(message);
|
public sealed class EmailSendConflictException(string message) : InvalidOperationException(message);
|
||||||
|
|
||||||
public sealed class EmailSendAttemptStore(JobTrackerContext db, TimeProvider timeProvider)
|
public sealed class EmailSendAttemptStore(JobTrackerContext db, TimeProvider timeProvider)
|
||||||
{
|
{
|
||||||
|
public static readonly TimeSpan AbandonedAge = TimeSpan.FromMinutes(15);
|
||||||
private DateTime UtcNow => timeProvider.GetUtcNow().UtcDateTime;
|
private DateTime UtcNow => timeProvider.GetUtcNow().UtcDateTime;
|
||||||
|
|
||||||
public Task<EmailSendAttempt?> GetAsync(Guid id, CancellationToken cancellationToken)
|
public Task<EmailSendAttempt?> GetAsync(Guid id, CancellationToken cancellationToken)
|
||||||
@@ -74,6 +76,58 @@ public sealed class EmailSendAttemptStore(JobTrackerContext db, TimeProvider tim
|
|||||||
public Task<int> MarkUncertainAsync(Guid id, string failureCategory, CancellationToken cancellationToken) =>
|
public Task<int> MarkUncertainAsync(Guid id, string failureCategory, CancellationToken cancellationToken) =>
|
||||||
CompleteAsync(id, EmailSendStatuses.Uncertain, null, failureCategory, cancellationToken);
|
CompleteAsync(id, EmailSendStatuses.Uncertain, null, failureCategory, cancellationToken);
|
||||||
|
|
||||||
|
public async Task<EmailSendAttemptRecoveryResult> ReconcileAbandonedAsync(CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
var now = UtcNow;
|
||||||
|
var cutoff = now - AbandonedAge;
|
||||||
|
var candidates = await db.EmailSendAttempts.IgnoreQueryFilters().AsNoTracking()
|
||||||
|
.Where(item =>
|
||||||
|
(item.Status == EmailSendStatuses.Pending && item.CreatedAtUtc <= cutoff) ||
|
||||||
|
(item.Status == EmailSendStatuses.Sending && (item.StartedAtUtc ?? item.CreatedAtUtc) <= cutoff))
|
||||||
|
.Select(item => new { item.Id, item.OwnerUserId, item.Status })
|
||||||
|
.ToListAsync(cancellationToken);
|
||||||
|
if (candidates.Count == 0) return new EmailSendAttemptRecoveryResult(0, 0);
|
||||||
|
|
||||||
|
await using var transaction = db.Database.IsRelational()
|
||||||
|
? await db.Database.BeginTransactionAsync(cancellationToken)
|
||||||
|
: null;
|
||||||
|
var failedPending = 0;
|
||||||
|
var uncertainSending = 0;
|
||||||
|
foreach (var candidate in candidates)
|
||||||
|
{
|
||||||
|
var failed = candidate.Status == EmailSendStatuses.Pending;
|
||||||
|
var terminalStatus = failed ? EmailSendStatuses.Failed : EmailSendStatuses.Uncertain;
|
||||||
|
var failureCategory = failed ? "process_stopped_before_delivery" : "process_interrupted";
|
||||||
|
var affected = await db.EmailSendAttempts.IgnoreQueryFilters()
|
||||||
|
.Where(item => item.Id == candidate.Id && item.Status == candidate.Status &&
|
||||||
|
(failed ? item.CreatedAtUtc <= cutoff : (item.StartedAtUtc ?? item.CreatedAtUtc) <= cutoff))
|
||||||
|
.ExecuteUpdateAsync(setters => setters
|
||||||
|
.SetProperty(item => item.Status, terminalStatus)
|
||||||
|
.SetProperty(item => item.FailureCategory, failureCategory)
|
||||||
|
.SetProperty(item => item.CompletedAtUtc, now), cancellationToken);
|
||||||
|
if (affected != 1) continue;
|
||||||
|
|
||||||
|
if (failed) failedPending++;
|
||||||
|
else uncertainSending++;
|
||||||
|
db.UserNotifications.Add(new UserNotification
|
||||||
|
{
|
||||||
|
Id = Guid.NewGuid(),
|
||||||
|
OwnerUserId = candidate.OwnerUserId,
|
||||||
|
Kind = failed ? "email.send.stopped" : "email.send.uncertain",
|
||||||
|
Title = failed ? "Email send stopped" : "Check email delivery",
|
||||||
|
Message = failed
|
||||||
|
? "An email attempt stopped before provider delivery. Review the draft before trying again."
|
||||||
|
: "Email delivery could not be confirmed after an interruption. Check the provider Sent folder and do not resend automatically.",
|
||||||
|
LinkPath = "/correspondence",
|
||||||
|
CreatedAtUtc = now,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
await db.SaveChangesAsync(cancellationToken);
|
||||||
|
if (transaction is not null) await transaction.CommitAsync(cancellationToken);
|
||||||
|
return new EmailSendAttemptRecoveryResult(failedPending, uncertainSending);
|
||||||
|
}
|
||||||
|
|
||||||
private Task<int> CompleteAsync(Guid id, string status, string? providerMessageId, string? failureCategory, CancellationToken cancellationToken)
|
private Task<int> CompleteAsync(Guid id, string status, string? providerMessageId, string? failureCategory, CancellationToken cancellationToken)
|
||||||
{
|
{
|
||||||
EnsureOwnerScope();
|
EnsureOwnerScope();
|
||||||
|
|||||||
Reference in New Issue
Block a user