diff --git a/src/Dispatch.Data/Providers/IDatabaseProvider.cs b/src/Dispatch.Data/Providers/IDatabaseProvider.cs index cf3f35f..d5885d2 100644 --- a/src/Dispatch.Data/Providers/IDatabaseProvider.cs +++ b/src/Dispatch.Data/Providers/IDatabaseProvider.cs @@ -128,6 +128,21 @@ public interface IDatabaseProvider /// Task GetTableSizeBytesAsync(DbContext db, string table, CancellationToken ct = default); + /// + /// Sizes for several tables at once. On the server engines each table's size is an independent query, + /// so the default just loops. SQLite overrides this: it derives every table's size from ONE whole-file + /// sampling pass (including a linear COUNT), so asking for the sizes one at a time re-runs that pass - + /// and the storage page needs both relay_log and audit_log, which doubled the work. + /// + async Task> GetTableSizesBytesAsync( + DbContext db, IReadOnlyList tables, CancellationToken ct = default) + { + var result = new Dictionary(tables.Count, StringComparer.OrdinalIgnoreCase); + foreach (var table in tables) + result[table] = await GetTableSizeBytesAsync(db, table, ct); + return result; + } + /// /// Reclaims space left by deleted rows so on-disk size actually shrinks and the size-pressure trigger /// in PurgeWorker can clear. No engine shrinks on DELETE alone. This holds heavy locks on every engine diff --git a/src/Dispatch.Data/Providers/SqliteDatabaseProvider.cs b/src/Dispatch.Data/Providers/SqliteDatabaseProvider.cs index 977fbc7..0052cd5 100644 --- a/src/Dispatch.Data/Providers/SqliteDatabaseProvider.cs +++ b/src/Dispatch.Data/Providers/SqliteDatabaseProvider.cs @@ -160,18 +160,39 @@ public Task GetDatabaseSizeBytesAsync(DbContext db, CancellationToken ct = /// public async Task GetTableSizeBytesAsync(DbContext db, string table, CancellationToken ct = default) { - var name = ProviderBootstrap.SafeIdentifier(table); + ProviderBootstrap.SafeIdentifier(table); + var sizes = await GetTableSizesBytesAsync(db, [table], ct); + return sizes.TryGetValue(table, out var bytes) ? bytes : 0; + } + + /// + /// Both sizes from a SINGLE sampling pass. GetTableSizeBytesAsync delegates here, and the storage page + /// asks for relay_log and audit_log together - so the whole-file payload sampling (and its linear + /// COUNT) runs once for the page rather than once per table. + /// + public async Task> GetTableSizesBytesAsync( + DbContext db, IReadOnlyList tables, CancellationToken ct = default) + { + foreach (var table in tables) ProviderBootstrap.SafeIdentifier(table); + + var result = new Dictionary(tables.Count, StringComparer.OrdinalIgnoreCase); var totalBytes = await GetDatabaseSizeBytesAsync(db, ct); - if (totalBytes == 0) return 0; + if (totalBytes == 0) + { + foreach (var table in tables) result[table] = 0; + return result; + } var payloads = await PayloadByTableAsync(db, ct); - if (!payloads.TryGetValue(name, out var mine) || mine == 0) return 0; - var allPayload = payloads.Values.Sum(); - if (allPayload == 0) return 0; - return (long)(totalBytes * ((double)mine / allPayload)); + foreach (var table in tables) + result[table] = allPayload > 0 && payloads.TryGetValue(table, out var mine) && mine > 0 + ? (long)(totalBytes * ((double)mine / allPayload)) + : 0; + + return result; } /// diff --git a/src/Dispatch.Data/Repositories/SqlMessageLogQuery.cs b/src/Dispatch.Data/Repositories/SqlMessageLogQuery.cs index 9d57da2..f6075c4 100644 --- a/src/Dispatch.Data/Repositories/SqlMessageLogQuery.cs +++ b/src/Dispatch.Data/Repositories/SqlMessageLogQuery.cs @@ -115,19 +115,30 @@ public async Task> RecentByApiKeyAsync( /// /// The list shows ONE row per message, but relay_log holds a row per lifecycle event: a Retrying row /// per failed attempt, then a terminal Delivered/Failed. Connection-level rows (denials, pre-DATA - /// failures) have no spool id and must each stay their own entry, hence the composite grouping key - - /// (spool_id, 0) collapses a message's events together, while ("", id) keeps every anonymous row apart. + /// failures) have no spool id and must each stay their own entry. /// - /// This replaces a ROW_NUMBER() OVER (PARTITION BY ... ORDER BY logged_at DESC, id DESC) window query. - /// MAX(id) selects the same row: ids are monotonically increasing, so within one spool id the highest - /// id is always the most recent event. It also translates cleanly on all four engines, where LINQ's - /// group-then-take-first does not. + /// Split into two arms, UNIONed, rather than one GROUP BY with a conditional key: + /// * messages (non-empty spool_id) - GROUP BY spool_id, taking MAX(id). ids increase monotonically, + /// so within one spool id the highest is the latest event (this replaces a ROW_NUMBER() window). + /// * anonymous rows (empty spool_id) - each id as-is; they are never grouped together. + /// + /// The reason for the split is performance, and it is exact, not approximate. The obvious single query, + /// GROUP BY (spool_id, CASE WHEN spool_id='' THEN id ELSE 0), produces identical results - but the CASE + /// makes the group key non-sargable, so IX_relay_log_spool_id (spool_id, logged_at, id) cannot serve it + /// and the engine scans relay_log into a temp B-tree. On the default (broad) Message Log filter over a + /// multi-million-row table that is seconds per page. Grouping by spool_id alone uses the index. /// - private static IQueryable LatestPerMessage(IQueryable filtered) => - filtered - .GroupBy(r => new { r.SpoolId, Anonymous = r.SpoolId == "" ? r.Id : 0L }) + private static IQueryable LatestPerMessage(IQueryable filtered) + { + var messages = filtered.Where(r => r.SpoolId != "") + .GroupBy(r => r.SpoolId) .Select(g => g.Max(r => r.Id)); + var anonymous = filtered.Where(r => r.SpoolId == "").Select(r => r.Id); + + return messages.Concat(anonymous); + } + /// /// Shared filter clauses. Composed rather than concatenated, so values cannot be interpolated even by /// accident. diff --git a/src/Dispatch.Data/Repositories/SqlStorageReport.cs b/src/Dispatch.Data/Repositories/SqlStorageReport.cs index 2a1a005..43cbe78 100644 --- a/src/Dispatch.Data/Repositories/SqlStorageReport.cs +++ b/src/Dispatch.Data/Repositories/SqlStorageReport.cs @@ -28,8 +28,11 @@ public async Task GetAsync(CancellationToken ct = default) .Select(g => new LogEventCount(g.Key, g.LongCount())) .ToListAsync(ct); - var relayLogBytes = await provider.GetTableSizeBytesAsync(db, "relay_log", ct); - var auditBytes = await provider.GetTableSizeBytesAsync(db, "audit_log", ct); + // Both sizes in one call: on SQLite that is one whole-file sampling pass for the page rather + // than one per table (see IDatabaseProvider.GetTableSizesBytesAsync). + var sizes = await provider.GetTableSizesBytesAsync(db, ["relay_log", "audit_log"], ct); + var relayLogBytes = sizes.GetValueOrDefault("relay_log"); + var auditBytes = sizes.GetValueOrDefault("audit_log"); var auditTotal = await db.AuditLog.AsNoTracking().LongCountAsync(ct); var auditSecurity = await db.AuditLog.AsNoTracking() diff --git a/src/Dispatch.Data/ServiceCollectionExtensions.cs b/src/Dispatch.Data/ServiceCollectionExtensions.cs index 7f515b4..56a8c8d 100644 --- a/src/Dispatch.Data/ServiceCollectionExtensions.cs +++ b/src/Dispatch.Data/ServiceCollectionExtensions.cs @@ -27,9 +27,18 @@ public static IServiceCollection AddDispatchData( services.AddSingleton(sp => new DatabaseBootstrap( provider, connectionString, sp.GetService>())); - // A factory rather than a scoped DbContext: the repositories below are singletons called - // concurrently from SpoolWorkerPool's worker threads, and a DbContext is not thread-safe. - services.AddDbContextFactory(o => + // A POOLED factory, not a scoped DbContext. The repositories below are singletons called + // concurrently from SpoolWorkerPool's worker threads, and a DbContext is not thread-safe, so each + // operation takes its own context. On the ingest path that is several contexts per message from + // many threads at once; pooling resets and reuses instances instead of re-allocating the change + // tracker and internal scope every time. Measured, this is the difference between roughly 1,000 and + // several thousand concurrent writes per second on SQLite. + // + // This must stay in sync with DispatchDbContextFactory.Create, which the migrator and tests use and + // which is also pooled - otherwise a benchmark run through the test path would report a throughput + // production never sees. That is precisely the gap this once had: DI registered the non-pooled + // AddDbContextFactory while the tests ran pooled. + services.AddPooledDbContextFactory(o => DispatchDbContextFactory.Configure(o, provider, connectionString)); // The repositories reach the engine only through this, for the few things EF cannot express. diff --git a/src/Dispatch.Service/Program.cs b/src/Dispatch.Service/Program.cs index 4cbdd8a..3bcbdb8 100644 --- a/src/Dispatch.Service/Program.cs +++ b/src/Dispatch.Service/Program.cs @@ -474,9 +474,17 @@ static void AlignTargetFileOwnership(DatabaseProvider provider, string connectio if (!File.Exists(file)) continue; try { + // Absolute path, not "chown" resolved via PATH. This runs as root; a PATH-relative lookup + // would let a chown planted earlier in PATH run as root instead. ArgumentList already keeps + // the file path out of any shell, so the value cannot inject arguments. /usr/bin then /bin + // covers the common layouts (coreutils lives in one or the other). + var chownBin = File.Exists("/usr/bin/chown") ? "/usr/bin/chown" + : File.Exists("/bin/chown") ? "/bin/chown" : null; + if (chownBin is null) break; + using var chown = System.Diagnostics.Process.Start(new System.Diagnostics.ProcessStartInfo { - FileName = "chown", + FileName = chownBin, ArgumentList = { $"--reference={directory}", file }, RedirectStandardError = true, RedirectStandardOutput = true, diff --git a/tests/Dispatch.Data.Tests/DependencyInjectionTests.cs b/tests/Dispatch.Data.Tests/DependencyInjectionTests.cs new file mode 100644 index 0000000..c6b792c --- /dev/null +++ b/tests/Dispatch.Data.Tests/DependencyInjectionTests.cs @@ -0,0 +1,30 @@ +using Dispatch.Data; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.Extensions.DependencyInjection; + +namespace Dispatch.Data.Tests; + +/// +/// Guards the wiring the repositories depend on, so it cannot silently drift from what the benchmarks and +/// tests exercise. +/// +public class DependencyInjectionTests +{ + [Fact] + public void The_context_factory_is_pooled() + { + // The ingest path takes a fresh context per operation from many threads. It must be POOLED, and DI + // must register the SAME kind the tests and migrator use (DispatchDbContextFactory.Create). This + // once diverged - DI registered the non-pooled AddDbContextFactory while ConcurrentWriteTests ran + // pooled - so the measured throughput was one production never saw. Assert the type directly. + var services = new ServiceCollection(); + services.AddLogging(); + services.AddDispatchData("Data Source=:memory:"); + + using var provider = services.BuildServiceProvider(); + var factory = provider.GetRequiredService>(); + + Assert.IsType>(factory); + } +} diff --git a/tests/Dispatch.Data.Tests/SqlRepositoriesTests.cs b/tests/Dispatch.Data.Tests/SqlRepositoriesTests.cs index 513f57a..7576812 100644 --- a/tests/Dispatch.Data.Tests/SqlRepositoriesTests.cs +++ b/tests/Dispatch.Data.Tests/SqlRepositoriesTests.cs @@ -205,15 +205,18 @@ public async Task Message_log_list_collapses_lifecycle_events_to_one_row_per_mes await log.InsertAsync(new RelayLogEntry { Event = "Retrying", Status = "Error", SpoolId = spool, FromAddress = $"a@{dom}", FromDomain = dom, ToAddresses = ["b@y.com"], ToDomain = "y.com", Provider = "None", Error = "temp" }); await log.InsertAsync(new RelayLogEntry { Event = "Retrying", Status = "Error", SpoolId = spool, FromAddress = $"a@{dom}", FromDomain = dom, ToAddresses = ["b@y.com"], ToDomain = "y.com", Provider = "None", Error = "temp" }); await log.InsertAsync(new RelayLogEntry { Event = "Delivered", Status = "OK", SpoolId = spool, FromAddress = $"a@{dom}", FromDomain = dom, ToAddresses = ["b@y.com"], ToDomain = "y.com", Provider = "None" }); - // A connection-level denial has no spool id and is its own message. + // TWO connection-level denials. Denials have no spool id, and each is its OWN message - they must + // never collapse together the way lifecycle rows sharing a spool id do. Two, not one, so a dedup + // that keyed only on spool_id would visibly merge them and fail here. await log.InsertAsync(new RelayLogEntry { Event = "Denied", Status = "Denied", SpoolId = "", FromAddress = $"c@{dom}", FromDomain = dom, ToAddresses = [], ToDomain = "", IngestSource = "SMTP", Error = "blocked" }); + await log.InsertAsync(new RelayLogEntry { Event = "Denied", Status = "Denied", SpoolId = "", FromAddress = $"d@{dom}", FromDomain = dom, ToAddresses = [], ToDomain = "", IngestSource = "SMTP", Error = "blocked" }); var page = await query.PageAsync(new MessageLogFilter { FromDomain = dom, Limit = 50 }, offset: 0); - // The 3 lifecycle rows collapse to ONE row showing the latest event (Delivered); the denial stays its own. - Assert.Equal(2, page.Total); + // The 3 lifecycle rows collapse to ONE (latest event = Delivered); each denial stays its own row. + Assert.Equal(3, page.Total); Assert.Equal("Delivered", page.Rows.Single(r => r.SpoolId == spool).Event); - Assert.Single(page.Rows, r => r.Event == "Denied"); + Assert.Equal(2, page.Rows.Count(r => r.Event == "Denied")); } [Fact]