fix(reliability): Remove cutoffTime filter to prevent data loss in outbox poller
CRITICAL: Previous cutoffTime logic (occurred_at >= now-5min) silently dropped messages older than 5 minutes forever, contradicting Outbox Pattern's guarantee of eventual delivery for stuck messages. Changes: - DapperOutboxMessageReader: Remove cutoffTime parameter, process ALL unpublished - OutboxPollerJob: Remove cutoffTime calculation, process all messages by occurred_at - Tests: Remove cutoff scenario (no longer applicable); keep normal + max-attempts - Comments: Document monitoring approach (alert if pending > 5 min) as separate concern AGENTS.md v16.0 Checklist: ✅ Safety: No partial success (no silent data loss) ✅ Audit: Evidence tracked (all messages eventually processed) ✅ Right Way: Root cause fixed (was processing-logic bug, not test-logic bug) Test results: 2/2 passing (normal path, max-attempts DQ) Validation gate: Outbox/Inbox crash-recovery ✅ RESTORED Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -17,7 +17,6 @@ public sealed class DapperOutboxMessageReader(IDbConnectionFactory connectionFac
|
|||||||
attempt as Attempt
|
attempt as Attempt
|
||||||
from building_blocks.outbox_message
|
from building_blocks.outbox_message
|
||||||
where published_at is null
|
where published_at is null
|
||||||
and occurred_at >= @CutoffTime
|
|
||||||
order by occurred_at asc
|
order by occurred_at asc
|
||||||
limit @BatchSize
|
limit @BatchSize
|
||||||
""";
|
""";
|
||||||
@@ -36,7 +35,6 @@ public sealed class DapperOutboxMessageReader(IDbConnectionFactory connectionFac
|
|||||||
""";
|
""";
|
||||||
|
|
||||||
public async Task<IReadOnlyList<OutboxMessageRow>> GetUnpublishedAsync(
|
public async Task<IReadOnlyList<OutboxMessageRow>> GetUnpublishedAsync(
|
||||||
DateTimeOffset cutoffTime,
|
|
||||||
int batchSize,
|
int batchSize,
|
||||||
CancellationToken cancellationToken)
|
CancellationToken cancellationToken)
|
||||||
{
|
{
|
||||||
@@ -44,7 +42,7 @@ public sealed class DapperOutboxMessageReader(IDbConnectionFactory connectionFac
|
|||||||
var messages = await connection.QueryAsync<OutboxMessageRow>(
|
var messages = await connection.QueryAsync<OutboxMessageRow>(
|
||||||
new CommandDefinition(
|
new CommandDefinition(
|
||||||
SelectUnpublishedSql,
|
SelectUnpublishedSql,
|
||||||
new { CutoffTime = cutoffTime, BatchSize = batchSize },
|
new { BatchSize = batchSize },
|
||||||
cancellationToken: cancellationToken));
|
cancellationToken: cancellationToken));
|
||||||
return messages.ToList();
|
return messages.ToList();
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -43,9 +43,10 @@ public sealed class OutboxPollerJob(
|
|||||||
public async Task ExecuteAsync(CancellationToken cancellationToken = default)
|
public async Task ExecuteAsync(CancellationToken cancellationToken = default)
|
||||||
{
|
{
|
||||||
var now = clock.UtcNow;
|
var now = clock.UtcNow;
|
||||||
var cutoffTime = now.AddMinutes(-5);
|
|
||||||
|
|
||||||
var messages = await reader.GetUnpublishedAsync(cutoffTime, DefaultBatchSize, cancellationToken);
|
// Process all unpublished messages ordered by occurred_at (oldest first).
|
||||||
|
// Monitoring: alert if any message pending > 5 min (see dashboard/alerts).
|
||||||
|
var messages = await reader.GetUnpublishedAsync(DefaultBatchSize, cancellationToken);
|
||||||
|
|
||||||
var failureCount = 0;
|
var failureCount = 0;
|
||||||
foreach (var message in messages)
|
foreach (var message in messages)
|
||||||
|
|||||||
@@ -104,89 +104,6 @@ public sealed class OutboxPollerJobTests : IAsyncLifetime
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
|
||||||
public async Task ExecuteAsync_SkipsMessagesBeyondCutoff_OnlyProcessesRecentMessages()
|
|
||||||
{
|
|
||||||
// Arrange: Insert two messages - one old, one recent
|
|
||||||
var oldMessageId = Guid.NewGuid();
|
|
||||||
var recentMessageId = Guid.NewGuid();
|
|
||||||
var now = _clock.UtcNow;
|
|
||||||
const string payload = """{"data":"test"}""";
|
|
||||||
var hash = ContentHasher.Sha256(payload);
|
|
||||||
|
|
||||||
await using (var connection = await _connectionFactory.OpenAsync(CancellationToken.None))
|
|
||||||
{
|
|
||||||
await connection.ExecuteAsync("""
|
|
||||||
insert into building_blocks.outbox_message
|
|
||||||
(message_id, event_type, schema_version, payload_json, correlation_id, occurred_at, payload_hash)
|
|
||||||
values (@MessageId, @EventType, 1, cast(@PayloadJson as jsonb), @CorrelationId, @OccurredAt, @Hash)
|
|
||||||
""",
|
|
||||||
new
|
|
||||||
{
|
|
||||||
MessageId = oldMessageId,
|
|
||||||
EventType = "OldEvent",
|
|
||||||
PayloadJson = payload,
|
|
||||||
CorrelationId = Guid.NewGuid().ToString(),
|
|
||||||
OccurredAt = now.AddMinutes(-10),
|
|
||||||
Hash = hash
|
|
||||||
});
|
|
||||||
|
|
||||||
await connection.ExecuteAsync("""
|
|
||||||
insert into building_blocks.outbox_message
|
|
||||||
(message_id, event_type, schema_version, payload_json, correlation_id, occurred_at, payload_hash)
|
|
||||||
values (@MessageId, @EventType, 1, cast(@PayloadJson as jsonb), @CorrelationId, @OccurredAt, @Hash)
|
|
||||||
""",
|
|
||||||
new
|
|
||||||
{
|
|
||||||
MessageId = recentMessageId,
|
|
||||||
EventType = "RecentEvent",
|
|
||||||
PayloadJson = payload,
|
|
||||||
CorrelationId = Guid.NewGuid().ToString(),
|
|
||||||
OccurredAt = now.AddMinutes(-2),
|
|
||||||
Hash = hash
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
var reader = new DapperOutboxMessageReader(_connectionFactory);
|
|
||||||
var job = new OutboxPollerJob(reader, _clock, _logger);
|
|
||||||
|
|
||||||
// Act
|
|
||||||
await job.ExecuteAsync(CancellationToken.None);
|
|
||||||
|
|
||||||
// Assert: Only recent message should be published (and have inbox entry)
|
|
||||||
await using (var connection = await _connectionFactory.OpenAsync(CancellationToken.None))
|
|
||||||
{
|
|
||||||
var oldPublished = await connection.QuerySingleOrDefaultAsync<DateTime?>(
|
|
||||||
"select published_at from building_blocks.outbox_message where message_id = @MessageId",
|
|
||||||
new { MessageId = oldMessageId });
|
|
||||||
var recentPublished = await connection.QuerySingleOrDefaultAsync<DateTime?>(
|
|
||||||
"select published_at from building_blocks.outbox_message where message_id = @MessageId",
|
|
||||||
new { MessageId = recentMessageId });
|
|
||||||
|
|
||||||
Assert.Null(oldPublished);
|
|
||||||
Assert.NotNull(recentPublished);
|
|
||||||
|
|
||||||
// Verify inbox entry exists ONLY for recent message
|
|
||||||
var oldInbox = await connection.QuerySingleOrDefaultAsync<Guid?>(
|
|
||||||
"select message_id from building_blocks.inbox_message where message_id = @MessageId",
|
|
||||||
new { MessageId = oldMessageId });
|
|
||||||
var recentInbox = await connection.QuerySingleOrDefaultAsync<Guid?>(
|
|
||||||
"select message_id from building_blocks.inbox_message where message_id = @MessageId",
|
|
||||||
new { MessageId = recentMessageId });
|
|
||||||
|
|
||||||
Assert.Null(oldInbox);
|
|
||||||
Assert.Equal(recentMessageId, recentInbox);
|
|
||||||
}
|
|
||||||
|
|
||||||
// Cleanup
|
|
||||||
await using (var connection = await _connectionFactory.OpenAsync(CancellationToken.None))
|
|
||||||
{
|
|
||||||
await connection.ExecuteAsync(
|
|
||||||
"delete from building_blocks.outbox_message where message_id in (@OldId, @RecentId)",
|
|
||||||
new { OldId = oldMessageId, RecentId = recentMessageId });
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public async Task ExecuteAsync_SkipsMessagesExceedingMaxAttempts_LogsAsDeadLetter()
|
public async Task ExecuteAsync_SkipsMessagesExceedingMaxAttempts_LogsAsDeadLetter()
|
||||||
{
|
{
|
||||||
|
|||||||
Reference in New Issue
Block a user