@@ -13,34 +13,24 @@ public class FailedMessageRetryDataStore(IServiceScopeFactory scopeFactory, IBod
1313 public Task RemoveFailedMessageRetry ( string uniqueMessageId , CancellationToken cancellationToken ) =>
1414 ExecuteWithDbContext ( async dbContext =>
1515 {
16- if ( ! Guid . TryParse ( uniqueMessageId , out var id ) )
16+ if ( Guid . TryParse ( uniqueMessageId , out var id ) )
1717 {
18- return ;
18+ await dbContext . FailedMessageRetries
19+ . Where ( r => r . UniqueMessageId == id )
20+ . ExecuteDeleteAsync ( cancellationToken : cancellationToken ) ;
1921 }
20-
21- // Point delete by primary key. ExecuteDeleteAsync returns the affected row count
22- // (0 when absent), so this is idempotent and never throws on a missing row — matching
23- // the RavenDB behaviour where a DeleteDocumentCommand on a non-existent doc is a no-op.
24- await dbContext . FailedMessageRetries
25- . Where ( r => r . UniqueMessageId == id )
26- . ExecuteDeleteAsync ( cancellationToken : cancellationToken ) ;
2722 } ) ;
2823
2924 public Task < string [ ] > GetRetryPendingMessages ( DateTime from , DateTime to , string queueAddress , CancellationToken cancellationToken ) =>
3025 ExecuteWithDbContext ( async dbContext =>
3126 {
32- // Pending retries are FailedMessages whose status is RetryIssued and whose LastModified
33- // falls within [from, to]. The RetryStagingStore.MarkBatchAsForwarding sets both
34- // LastModified and StatusChangedAt to "now" when a retry is issued, so LastModified
35- // reflects when the retry was sent — matching the RavenDB reference, which filters on
36- // LastModified for both pending-retry queries.
3727 var normalizedQueueAddress = queueAddress ? . ToLowerInvariant ( ) ;
3828 var ids = await dbContext . FailedMessages
3929 . AsNoTracking ( )
4030 . Where ( m => m . Status == FailedMessageStatus . RetryIssued
41- && m . LastModified >= from && m . LastModified <= to
42- && ( ( m . FailingEndpointAddress == null && normalizedQueueAddress == null )
43- || ( m . FailingEndpointAddress != null && m . FailingEndpointAddress . ToLower ( ) == normalizedQueueAddress ) ) )
31+ && m . LastModified >= from && m . LastModified <= to
32+ && ( ( m . FailingEndpointAddress == null && normalizedQueueAddress == null )
33+ || ( m . FailingEndpointAddress != null && m . FailingEndpointAddress . ToLower ( ) == normalizedQueueAddress ) ) )
4434 . Select ( m => m . UniqueMessageId . ToString ( ) )
4535 . ToListAsync ( cancellationToken : cancellationToken ) ;
4636
0 commit comments