G1 & G2: Storage Governor and Aggregated retention pruning

This commit is contained in:
Richard
2026-07-19 10:56:12 +02:00
parent 7045002ca3
commit 8a39b912a6
5 changed files with 156 additions and 5 deletions
@@ -0,0 +1,28 @@
using Predictalytics.Application.Services;
using Xunit;
namespace Predictalytics.Application.Tests.Services;
public class StorageGovernorTests
{
[Theory]
[InlineData(10.0, 90.0, 90, 30, 90)] // far below budget -> full window
[InlineData(72.0, 90.0, 90, 30, 90)] // exactly at 80% -> still full
[InlineData(90.0, 90.0, 90, 30, 30)] // at budget -> min window
[InlineData(100.0, 90.0, 90, 30, 30)] // over budget -> min window
[InlineData(72.9, 90.0, 90, 30, 87)] // 81% of budget -> slightly tightened
[InlineData(81.0, 90.0, 90, 30, 60)] // 90% of budget -> halfway tightened
public void ComputeEffectiveRetentionDays_Interpolates(
double sizeGb, double maxGb, int configured, int min, int expected)
{
Assert.Equal(expected, StorageGovernor.ComputeEffectiveRetentionDays(sizeGb, maxGb, configured, min));
}
[Fact]
public void HalfwayIntoZone_IsBetweenMinAndConfigured()
{
// 85 GB of 90 (softStart 72) -> t ~0.72 -> days between 30 and 90
var days = StorageGovernor.ComputeEffectiveRetentionDays(85.0, 90.0, 90, 30);
Assert.InRange(days, 31, 89);
}
}
@@ -137,4 +137,67 @@ public class TradeRetentionWorkerTests
Assert.Equal(0.45m, pos.AvgCost); Assert.Equal(0.45m, pos.AvgCost);
} }
} }
[Fact]
public async Task RunOptimizationAsync_PrunesAggregatedTraderTrades()
{
using var connection = new SqliteConnection("DataSource=:memory:");
connection.Open();
var options = new DbContextOptionsBuilder<AppDbContext>()
.UseSqlite(connection)
.Options;
var baseDate = DateTime.UtcNow.Date;
using (var setup = new AppDbContext(options))
{
setup.Database.EnsureCreated();
var ev = new Event { Id = 1, Platform = PlatformType.Polymarket, Slug = "e", Title = "E" };
setup.Set<Event>().Add(ev);
var market = new Market { Id = 10, EventId = 1, PlatformMarketId = 1L, Question = "Q?" };
market.Outcomes.Add(new MarketOutcome { Id = 100, MarketId = 10, Label = "Yes", TokenId = "t100", CurrentPrice = 0.45m });
setup.Markets.Add(market);
// Aggregated trader - target of G2 retention pruning
setup.Traders.Add(new Trader { Id = 2, PlatformUserId = "0x2", DisplayName = "AggregatedTrader", IngestMode = IngestMode.Aggregated });
// Old trade older than retention cut-off (e.g. 200 days old when retention is 90 days)
setup.Trades.Add(new Trade
{
Id = 20, TraderId = 2, DbMarketId = 10, MarketOutcomeId = 100, PlatformTradeId = "tx20",
Side = TradeSide.Buy, Price = 0.40m, Size = 50m, Amount = 20m,
ExecutedAt = baseDate.AddDays(-200)
});
setup.SaveChanges();
}
var services = new ServiceCollection();
services.AddScoped(_ => new AppDbContext(options));
using var provider = services.BuildServiceProvider();
var config = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary<string, string?>
{
["RetentionSettings:RetentionDays"] = "90",
["RetentionSettings:CompactionDays"] = "14",
["RetentionSettings:MaxDatabaseSizeGb"] = "90.0",
["RetentionSettings:MinRetentionDays"] = "30"
}).Build();
var worker = new TradeRetentionWorker(provider, config, NullLogger<TradeRetentionWorker>.Instance);
var method = typeof(TradeRetentionWorker).GetMethod("RunOptimizationAsync", BindingFlags.Instance | BindingFlags.NonPublic);
Assert.NotNull(method);
// Act: Run optimization
await (Task)method!.Invoke(worker, new object[] { CancellationToken.None })!;
// Assert: The old trade must be deleted
using (var assertCtx = new AppDbContext(options))
{
var oldTradeExists = await assertCtx.Trades.AnyAsync(t => t.Id == 20);
Assert.False(oldTradeExists, "The old trade of the Aggregated trader should have been pruned.");
}
}
} }
@@ -0,0 +1,30 @@
namespace Predictalytics.Application.Services;
/// <summary>
/// Pure helper: shrinks the retention window as the database approaches its size budget.
/// No DB access here — fully unit-testable.
/// </summary>
public static class StorageGovernor
{
/// <summary>
/// Returns the retention window (in days) to actually use.
/// - Below 80% of the budget: the configured window (no tightening).
/// - Between 80% and 100%: linearly interpolated down towards minRetentionDays.
/// - At or above 100%: minRetentionDays.
/// </summary>
public static int ComputeEffectiveRetentionDays(
double currentSizeGb, double maxSizeGb, int configuredRetentionDays, int minRetentionDays)
{
if (maxSizeGb <= 0 || configuredRetentionDays <= minRetentionDays)
return configuredRetentionDays;
double softStart = 0.80 * maxSizeGb;
if (currentSizeGb <= softStart) return configuredRetentionDays;
if (currentSizeGb >= maxSizeGb) return minRetentionDays;
// linear interpolation between softStart (=configured) and maxSizeGb (=min)
double t = (currentSizeGb - softStart) / (maxSizeGb - softStart); // 0..1
double days = configuredRetentionDays - t * (configuredRetentionDays - minRetentionDays);
return (int)System.Math.Round(days);
}
}
@@ -37,5 +37,12 @@
"AuthRequired": false, "AuthRequired": false,
"AllowedOrigins": [ "http://localhost:5000" ], "AllowedOrigins": [ "http://localhost:5000" ],
"ReadOnlyDatabase": false "ReadOnlyDatabase": false
},
"RetentionSettings": {
"Enabled": true,
"RetentionDays": 180,
"CompactionDays": 14,
"MaxDatabaseSizeGb": 90.0,
"MinRetentionDays": 30
} }
} }
@@ -71,25 +71,48 @@ public class TradeRetentionWorker : BackgroundService
var retentionDays = _config.GetValue("RetentionSettings:RetentionDays", 180); var retentionDays = _config.GetValue("RetentionSettings:RetentionDays", 180);
var compactionDays = _config.GetValue("RetentionSettings:CompactionDays", 14); var compactionDays = _config.GetValue("RetentionSettings:CompactionDays", 14);
var maxSizeGb = _config.GetValue("RetentionSettings:MaxDatabaseSizeGb", 90.0);
var minRetention = _config.GetValue("RetentionSettings:MinRetentionDays", 30);
_logger.LogInformation("🧹 TradeRetentionWorker: Starting optimization. RetentionDays={Retention}, CompactionDays={Compaction}", double currentSizeGb = 0;
retentionDays, compactionDays); try
{
currentSizeGb = (await db.Database.SqlQueryRaw<double>(
"SELECT COALESCE(SUM(data_length + index_length),0) / 1073741824.0 AS Value " +
"FROM information_schema.tables WHERE table_schema = DATABASE()").ToListAsync(ct)).FirstOrDefault();
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Could not read DB size; using configured retention.");
}
var effectiveRetentionDays = Predictalytics.Application.Services.StorageGovernor.ComputeEffectiveRetentionDays(
currentSizeGb, maxSizeGb, retentionDays, minRetention);
if (effectiveRetentionDays != retentionDays)
{
_logger.LogWarning("Storage governor: DB {Size:F1} GB, retention tightened {From}d -> {To}d",
currentSizeGb, retentionDays, effectiveRetentionDays);
}
_logger.LogInformation("🧹 TradeRetentionWorker: Starting optimization. RetentionDays={Retention} (Effective={Effective}), CompactionDays={Compaction}",
retentionDays, effectiveRetentionDays, compactionDays);
var utcNow = DateTime.UtcNow; var utcNow = DateTime.UtcNow;
var retentionCutoff = utcNow.Date.AddDays(-retentionDays); var retentionCutoff = utcNow.Date.AddDays(-effectiveRetentionDays);
var compactionCutoff = utcNow.Date.AddDays(-compactionDays); var compactionCutoff = utcNow.Date.AddDays(-compactionDays);
// Exclude trades if the trader is on any active Watchlist // Exclude trades if the trader is on any active Watchlist
_logger.LogInformation("Pruning trades older than {Cutoff}...", retentionCutoff); _logger.LogInformation("Pruning trades older than {Cutoff}...", retentionCutoff);
var deletedTrades = await db.Trades var deletedTrades = await db.Trades
.Where(t => t.ExecutedAt < retentionCutoff && !t.Trader.WatchlistEntries.Any() && t.Trader.IngestMode == IngestMode.Full) .Where(t => t.ExecutedAt < retentionCutoff && !t.Trader.WatchlistEntries.Any() && t.Trader.IngestMode != IngestMode.SnapshotOnly)
.Select(t => new { t.TraderId, t.MarketOutcomeId }) .Select(t => new { t.TraderId, t.MarketOutcomeId })
.Distinct() .Distinct()
.ToListAsync(ct); .ToListAsync(ct);
var deletedCount = await db.Trades var deletedCount = await db.Trades
.Where(t => t.ExecutedAt < retentionCutoff && !t.Trader.WatchlistEntries.Any() && t.Trader.IngestMode == IngestMode.Full) .Where(t => t.ExecutedAt < retentionCutoff && !t.Trader.WatchlistEntries.Any() && t.Trader.IngestMode != IngestMode.SnapshotOnly)
.ExecuteDeleteAsync(ct); .ExecuteDeleteAsync(ct);
if (deletedCount > 0 && deletedTrades.Any()) if (deletedCount > 0 && deletedTrades.Any())