feat(engine): RetentionService — hourly executions sweep
This commit is contained in:
52
src/VolcanicCore.Api/Engine/RetentionService.cs
Normal file
52
src/VolcanicCore.Api/Engine/RetentionService.cs
Normal file
@@ -0,0 +1,52 @@
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using VolcanicCore.Persistence.Repositories;
|
||||
|
||||
namespace VolcanicCore.Api.Engine;
|
||||
|
||||
public sealed class RetentionService : BackgroundService
|
||||
{
|
||||
private static readonly TimeSpan SweepInterval = TimeSpan.FromHours(1);
|
||||
|
||||
private readonly ExecutionRepository _executions;
|
||||
private readonly int _retentionDays;
|
||||
private readonly int _keepPerWorkflow;
|
||||
private readonly ILogger<RetentionService> _log;
|
||||
|
||||
public RetentionService(
|
||||
ExecutionRepository executions, int retentionDays, int keepPerWorkflow, ILogger<RetentionService> log)
|
||||
{
|
||||
_executions = executions;
|
||||
_retentionDays = retentionDays;
|
||||
_keepPerWorkflow = keepPerWorkflow;
|
||||
_log = log;
|
||||
}
|
||||
|
||||
// Public so tests can invoke it deterministically.
|
||||
public void RunSweep()
|
||||
{
|
||||
try
|
||||
{
|
||||
var oldCutoff = DateTimeOffset.UtcNow.AddDays(-_retentionDays);
|
||||
var byAge = _executions.PruneOlderThan(oldCutoff);
|
||||
var overCap = _executions.PruneOldestPerWorkflow(_keepPerWorkflow);
|
||||
_log.LogInformation("retention sweep: pruned {ByAge} by age, {OverCap} by per-workflow cap",
|
||||
byAge, overCap);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_log.LogError(ex, "retention sweep failed");
|
||||
}
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
RunSweep();
|
||||
while (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
try { await Task.Delay(SweepInterval, stoppingToken); }
|
||||
catch (TaskCanceledException) { break; }
|
||||
RunSweep();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -125,6 +125,12 @@ builder.Services.AddSingleton<TriggerHost>();
|
||||
builder.Services.AddSingleton<IWorkflowLifecycle>(sp => sp.GetRequiredService<TriggerHost>());
|
||||
builder.Services.AddHostedService(sp => sp.GetRequiredService<TriggerHost>());
|
||||
|
||||
builder.Services.AddHostedService(sp => new RetentionService(
|
||||
sp.GetRequiredService<ExecutionRepository>(),
|
||||
settings.ExecutionRetentionDays,
|
||||
settings.ExecutionKeepPerWorkflow,
|
||||
sp.GetRequiredService<ILogger<RetentionService>>()));
|
||||
|
||||
var app = builder.Build();
|
||||
|
||||
// Apply migrations at boot.
|
||||
|
||||
43
tests/VolcanicCore.Tests/RetentionServiceTests.cs
Normal file
43
tests/VolcanicCore.Tests/RetentionServiceTests.cs
Normal file
@@ -0,0 +1,43 @@
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using VolcanicCore.Api.Engine;
|
||||
using VolcanicCore.Persistence;
|
||||
using VolcanicCore.Persistence.Migrations;
|
||||
using VolcanicCore.Persistence.Repositories;
|
||||
|
||||
namespace VolcanicCore.Tests;
|
||||
|
||||
public class RetentionServiceTests
|
||||
{
|
||||
[Fact]
|
||||
public void Sweep_prunes_old_rows_and_over_cap_rows()
|
||||
{
|
||||
var path = Path.Combine(Path.GetTempPath(), $"vc-{Guid.NewGuid():N}.db");
|
||||
var f = new SqliteConnectionFactory(path);
|
||||
using (var conn = f.Open())
|
||||
new Migrator(new IMigration[]
|
||||
{
|
||||
new Migration001Initial(), new Migration002Connections(), new Migration003Workflows(),
|
||||
new Migration004Executions(), new Migration005NodeExecutions(),
|
||||
}).Apply(conn);
|
||||
var er = new ExecutionRepository(f);
|
||||
|
||||
// 3 old finished, 5 recent finished (keep=3 → prune 2 of the recent + all 3 old)
|
||||
for (var i = 0; i < 3; i++)
|
||||
{
|
||||
var id = er.Create("wf", "manual", "{}");
|
||||
er.SetStatus(id, "completed", null, DateTimeOffset.UtcNow.AddDays(-30));
|
||||
}
|
||||
for (var i = 0; i < 5; i++)
|
||||
{
|
||||
var id = er.Create("wf", "manual", "{}");
|
||||
er.SetStatus(id, "completed", null, DateTimeOffset.UtcNow.AddSeconds(i));
|
||||
}
|
||||
Assert.Equal(8, er.ListForWorkflow("wf", 100).Count);
|
||||
|
||||
var svc = new RetentionService(er, retentionDays: 7, keepPerWorkflow: 3,
|
||||
NullLogger<RetentionService>.Instance);
|
||||
svc.RunSweep();
|
||||
|
||||
Assert.Equal(3, er.ListForWorkflow("wf", 100).Count);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user