perf(store-and-forward): bound retry sweep with SweepBatchLimit (default 500) on the due-rows query
This commit is contained in:
@@ -22,4 +22,11 @@ public class StoreAndForwardOptions
|
||||
|
||||
/// <summary>Interval for the background retry timer sweep.</summary>
|
||||
public TimeSpan RetryTimerInterval { get; set; } = TimeSpan.FromSeconds(10);
|
||||
|
||||
/// <summary>
|
||||
/// Maximum due rows loaded per retry sweep. Bounds sweep memory and duration
|
||||
/// under backlog; the remainder is picked up on the next tick (10 s).
|
||||
/// 0 = unlimited (legacy).
|
||||
/// </summary>
|
||||
public int SweepBatchLimit { get; set; } = 500;
|
||||
}
|
||||
|
||||
@@ -601,7 +601,7 @@ public class StoreAndForwardService
|
||||
}
|
||||
}
|
||||
|
||||
var messages = await _storage.GetMessagesForRetryAsync();
|
||||
var messages = await _storage.GetMessagesForRetryAsync(_options.SweepBatchLimit);
|
||||
if (messages.Count == 0) return;
|
||||
|
||||
_logger.LogDebug("Retry sweep: {Count} messages due for retry", messages.Count);
|
||||
|
||||
@@ -180,7 +180,7 @@ public class StoreAndForwardStorage
|
||||
/// Gets all messages that are due for retry (Pending status, last attempt older than retry interval).
|
||||
/// </summary>
|
||||
/// <returns>A task that resolves to the list of messages due for retry, ordered by creation time ascending.</returns>
|
||||
public async Task<List<StoreAndForwardMessage>> GetMessagesForRetryAsync()
|
||||
public async Task<List<StoreAndForwardMessage>> GetMessagesForRetryAsync(int limit = 0)
|
||||
{
|
||||
await using var connection = new SqliteConnection(_connectionString);
|
||||
await connection.OpenAsync();
|
||||
@@ -197,6 +197,13 @@ public class StoreAndForwardStorage
|
||||
OR (julianday('now') - julianday(last_attempt_at)) * 86400000 >= retry_interval_ms)
|
||||
ORDER BY created_at ASC";
|
||||
|
||||
// Bound the sweep batch (oldest-first). 0 = unlimited (legacy).
|
||||
if (limit > 0)
|
||||
{
|
||||
cmd.CommandText += "\n LIMIT @limit";
|
||||
cmd.Parameters.AddWithValue("@limit", limit);
|
||||
}
|
||||
|
||||
cmd.Parameters.AddWithValue("@pending", (int)StoreAndForwardMessageStatus.Pending);
|
||||
|
||||
return await ReadMessagesAsync(cmd);
|
||||
|
||||
@@ -34,4 +34,8 @@ public class StoreAndForwardOptionsTests
|
||||
Assert.Equal(TimeSpan.FromMinutes(5), options.DefaultRetryInterval);
|
||||
Assert.Equal(100, options.DefaultMaxRetries);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void SweepBatchLimit_Defaults_To_500() =>
|
||||
Assert.Equal(500, new StoreAndForwardOptions().SweepBatchLimit);
|
||||
}
|
||||
|
||||
@@ -587,6 +587,33 @@ public class StoreAndForwardStorageTests : IAsyncLifetime, IDisposable
|
||||
Assert.Null(retrieved!.ParentExecutionId);
|
||||
}
|
||||
|
||||
// ── Task 7 (arch review 02, Performance): bounded due-rows query ──
|
||||
|
||||
[Fact]
|
||||
public async Task GetMessagesForRetry_HonorsLimit_OldestFirst()
|
||||
{
|
||||
var baseTime = DateTimeOffset.UtcNow.AddMinutes(-10);
|
||||
for (var i = 0; i < 5; i++)
|
||||
{
|
||||
await _storage.EnqueueAsync(new StoreAndForwardMessage
|
||||
{
|
||||
Id = $"m{i}",
|
||||
Category = StoreAndForwardCategory.ExternalSystem,
|
||||
Target = "t",
|
||||
PayloadJson = "{}",
|
||||
RetryCount = 0,
|
||||
MaxRetries = 50,
|
||||
RetryIntervalMs = 0, // always due
|
||||
CreatedAt = baseTime.AddSeconds(i),
|
||||
Status = StoreAndForwardMessageStatus.Pending
|
||||
});
|
||||
}
|
||||
|
||||
var page = await _storage.GetMessagesForRetryAsync(limit: 3);
|
||||
Assert.Equal(3, page.Count);
|
||||
Assert.Equal(new[] { "m0", "m1", "m2" }, page.Select(m => m.Id));
|
||||
}
|
||||
|
||||
private static StoreAndForwardMessage CreateMessage(string id, StoreAndForwardCategory category)
|
||||
{
|
||||
return new StoreAndForwardMessage
|
||||
|
||||
Reference in New Issue
Block a user