From b76029b84390a25d3c2a2cd3ea6d1504a9d5108c Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Tue, 7 Jul 2026 14:01:38 -0500 Subject: [PATCH] daemonload over pooled sharded databases (#4882, epic jasperfx#486 WS6) --databases N pools the tenants across N scaletest_dl_shard_* databases via MultiTenantedWithShardedDatabases (explicit round-robin placement, one daemon per shard). ShardConnectionSampler groups pg_stat_activity by datname so the --max-connections gate is enforced PER DATABASE, plus per-tenant catch-up verification on every shard and a database-affine placement check (per-tenant sequence in exactly the home shard). Local-scale result (25 tenants x 2 shards x 2 projections, 60s): shard peaks 16/11, all 25 tenants caught up, zero placement violations. Single-database path (--databases 1, the default) unchanged. Addresses #4882 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01XNfeNz73Q3EdiTgSeDbo96 --- .../Commands/DaemonLoadCommand.Sharded.cs | 436 ++++++++++++++++++ .../Commands/DaemonLoadCommand.cs | 31 +- .../Commands/DaemonLoadInput.cs | 5 +- .../Instrumentation/ShardConnectionSampler.cs | 174 +++++++ src/Marten.ScaleTesting/README.md | 30 +- 5 files changed, 665 insertions(+), 11 deletions(-) create mode 100644 src/Marten.ScaleTesting/Commands/DaemonLoadCommand.Sharded.cs create mode 100644 src/Marten.ScaleTesting/Instrumentation/ShardConnectionSampler.cs diff --git a/src/Marten.ScaleTesting/Commands/DaemonLoadCommand.Sharded.cs b/src/Marten.ScaleTesting/Commands/DaemonLoadCommand.Sharded.cs new file mode 100644 index 0000000000..f0e66f61d7 --- /dev/null +++ b/src/Marten.ScaleTesting/Commands/DaemonLoadCommand.Sharded.cs @@ -0,0 +1,436 @@ +using System.Diagnostics; +using System.Text.Json; +using JasperFx.Core; +using JasperFx.Events; +using JasperFx.Events.Daemon; +using JasperFx.Events.Projections; +using Marten.Events; +using Marten.ScaleTesting.Instrumentation; +using Marten.Storage; +using Npgsql; +using Spectre.Console; + +namespace Marten.ScaleTesting.Commands; + +/// +/// marten#4882 (epic jasperfx#486 WS6): the sharded half of daemonload. Pools +/// --tenants tenants across --databases shard databases on the same Postgres server +/// via MultiTenantedWithShardedDatabases (sharded tenancy + per-tenant partitioned events +/// within each shard), runs one projection daemon per shard, appends continuously across every +/// tenant, and samples pg_stat_activity grouped by datname. +/// +/// The WS6 assertions this run makes: +/// +/// O(databases) connections: with the 2.22.0 per-database governors (4 event loads + +/// 4 batch writes + HWM + appenders per database), each shard DB's peak should mirror the +/// single-DB daemonload result — --max-connections gates PER DATABASE +/// Per-tenant catch-up on every shard: every tenant's per-tenant progression rows +/// reach that tenant's own sequence ceiling in its own shard +/// Database-affine placement: each tenant's per-tenant event sequence exists in +/// EXACTLY its assigned shard — no cross-shard bleed +/// +/// +public sealed partial class DaemonLoadCommand +{ + private const string ShardDatabasePrefix = "scaletest_dl_shard_"; + + private async Task ExecuteShardedAsync(DaemonLoadInput input) + { + var totalElapsed = Stopwatch.StartNew(); + var databaseCount = input.DatabasesFlag; + var shardNames = Enumerable.Range(0, databaseCount) + .Select(i => $"{ShardDatabasePrefix}{i}") + .ToArray(); + + AnsiConsole.MarkupLine( + $"[blue]daemonload (sharded): databases=[yellow]{databaseCount}[/] tenants=[yellow]{input.TenantsFlag}[/] " + + $"projections=[yellow]{input.ProjectionsFlag}[/] duration=[yellow]{input.DurationSecondsFlag}s[/] " + + $"rate=[yellow]~{input.AppendRatePerSecondFlag}/s[/][/]"); + + if (input.WipeFlag) + { + await WipeShardedAsync(shardNames).ConfigureAwait(false); + } + + await EnsureShardDatabasesExistAsync(shardNames).ConfigureAwait(false); + + var shardConnectionStrings = shardNames.ToDictionary( + name => name, + name => new NpgsqlConnectionStringBuilder(ConnectionSource.ConnectionString) + { + Database = name + }.ConnectionString); + + var projectionNames = Enumerable.Range(0, Math.Max(1, input.ProjectionsFlag)) + .Select(i => $"LoadRollup{i}") + .ToArray(); + + // ShardedTenancyOptions.ApplicationName stamps every pooled shard connection string, so + // the by-datname sampler counts exactly the store's connections per database. + using var store = Marten.DocumentStore.For(opts => + { + opts.MultiTenantedWithShardedDatabases(x => + { + x.ConnectionString = new NpgsqlConnectionStringBuilder(ConnectionSource.ConnectionString) + { + ApplicationName = ApplicationName + }.ConnectionString; + x.SchemaName = Schema; + x.ApplicationName = ApplicationName; + x.UseExplicitAssignment(); + + foreach (var (name, connectionString) in shardConnectionStrings) + { + x.AddDatabase(name, connectionString); + } + }); + + opts.DisableNpgsqlLogging = true; + opts.DatabaseSchemaName = Schema; + opts.Events.TenancyStyle = TenancyStyle.Conjoined; + opts.Events.UseTenantPartitionedEvents = true; + opts.Events.AppendMode = EventAppendMode.QuickWithServerTimestamps; + opts.Policies.AllDocumentsAreMultiTenanted(); + opts.Events.AddEventType(); + + foreach (var name in projectionNames) + { + opts.Projections.Add(new DaemonLoadRollupProjection(name), ProjectionLifecycle.Async, name); + } + }); + + // Apply the schema to every seeded shard database up front — tenant provisioning + // (per-tenant sequences + partitions) presumes the events schema already exists there. + await store.Storage.ApplyAllConfiguredChangesToDatabaseAsync().ConfigureAwait(false); + + var tenants = Enumerable.Range(0, input.TenantsFlag) + .Select(i => $"tenant_{i:0000}") + .ToArray(); + + // Round-robin, explicit, deterministic placement: tenant i lives on shard i % N. The + // placement map doubles as the expected-affinity baseline for the bleed check below. + var tenantsByShard = tenants + .Select((tenant, i) => (Tenant: tenant, Shard: shardNames[i % databaseCount])) + .GroupBy(x => x.Shard, x => x.Tenant) + .ToDictionary(g => g.Key, g => g.ToArray()); + + AnsiConsole.MarkupLine( + $"[grey]Assigning {tenants.Length} tenants round-robin across {databaseCount} shard databases (per-tenant partition DDL — this can take a bit)...[/]"); + for (var i = 0; i < tenants.Length; i++) + { + await store.Advanced.AddTenantToShardAsync(tenants[i], shardNames[i % databaseCount], CancellationToken.None) + .ConfigureAwait(false); + } + + // One seed event per tenant so every per-tenant sequence + partition is exercised before + // the daemons start and every (projection × tenant) agent has something to fan out for. + foreach (var tenant in tenants) + { + await using var session = store.LightweightSession(tenant); + session.Events.StartStream(Guid.NewGuid(), new DaemonLoadEvent(tenant, 0)); + await session.SaveChangesAsync().ConfigureAwait(false); + } + + using var cts = new CancellationTokenSource(); + + // Sample every shard database plus the master (registry) database, one poll session for + // the whole cluster — pg_stat_activity is cluster-wide. + var masterDatabaseName = new NpgsqlConnectionStringBuilder(ConnectionSource.ConnectionString).Database!; + var sampledDatabases = shardNames.Append(masterDatabaseName).ToArray(); + await using var sampler = ShardConnectionSampler.Start( + ConnectionSource.ConnectionString, ApplicationName, sampledDatabases, + TimeSpan.FromSeconds(Math.Max(0.1, input.SampleSecondsFlag)), + string.IsNullOrWhiteSpace(input.TraceFlag) ? null : input.TraceFlag, + cts.Token); + + // One daemon per shard database. Under sharded tenancy each shard is its own + // MartenDatabase and the default-tenant daemon overload is invalid, so address each + // daemon via a representative tenant assigned to that shard. + AnsiConsole.MarkupLine("[grey]Starting one projection daemon per shard (per-tenant agent fan-out within each)...[/]"); + var daemons = new List(); + try + { + foreach (var shard in shardNames) + { + var daemon = await store.BuildProjectionDaemonAsync(tenantsByShard[shard][0]).ConfigureAwait(false); + daemons.Add(daemon); + await daemon.StartAllAsync().ConfigureAwait(false); + } + + // ---- Continuous append load ------------------------------------------ + + var appended = 0L; + var appendFailures = 0L; + var appendElapsed = Stopwatch.StartNew(); + var writers = Enumerable.Range(0, Math.Max(1, input.WritersFlag)) + .Select(w => Task.Run(() => AppendLoopAsync(store, tenants, w, input, cts.Token, + () => Interlocked.Increment(ref appended), + () => Interlocked.Increment(ref appendFailures)))) + .ToArray(); + + await Task.Delay(TimeSpan.FromSeconds(input.DurationSecondsFlag)).ConfigureAwait(false); + cts.Cancel(); + await Task.WhenAll(writers).ConfigureAwait(false); + appendElapsed.Stop(); + + // ---- Catch-up + placement verification -------------------------------- + + AnsiConsole.MarkupLine( + $"[grey]Appends stopped ({appended:N0} events). Waiting for per-tenant catch-up on every shard...[/]"); + var (caughtUp, stalled) = await WaitForShardedCatchUpAsync( + shardConnectionStrings, tenantsByShard, projectionNames, + TimeSpan.FromSeconds(input.CatchUpTimeoutSecondsFlag)).ConfigureAwait(false); + + var placementViolations = await CheckPlacementAffinityAsync(shardConnectionStrings, tenantsByShard) + .ConfigureAwait(false); + + var perDatabase = sampler.Capture(); + + foreach (var daemon in daemons) + { + await daemon.StopAllAsync().ConfigureAwait(false); + } + + totalElapsed.Stop(); + + // ---- Report ------------------------------------------------------------ + + var appendRate = appended / Math.Max(0.001, appendElapsed.Elapsed.TotalSeconds); + var shardSnapshots = perDatabase.Where(x => x.Database != masterDatabaseName).ToArray(); + var masterSnapshot = perDatabase.FirstOrDefault(x => x.Database == masterDatabaseName); + + var table = new Table().AddColumn("Metric").AddColumn(new TableColumn("Value").RightAligned()); + table.AddRow("Shard databases", databaseCount.ToString("N0")); + table.AddRow("Tenants", tenants.Length.ToString("N0")); + table.AddRow("Async projections", projectionNames.Length.ToString("N0")); + table.AddRow("Tenant agents (projections × tenants)", (tenants.Length * projectionNames.Length).ToString("N0")); + table.AddRow("Events appended", appended.ToString("N0")); + table.AddRow("Append failures", appendFailures.ToString("N0")); + table.AddRow("Sustained append rate (events/sec)", appendRate.ToString("N0")); + foreach (var snapshot in shardSnapshots) + { + table.AddRow($"[bold]{snapshot.Database} peak connections[/]", + $"[bold]{snapshot.MaxTotal:N0}[/] (mean {snapshot.MeanTotal:N1}, busy peak {snapshot.MaxBusy:N0})"); + } + if (masterSnapshot != null) + { + table.AddRow("Master DB peak connections", + $"{masterSnapshot.MaxTotal:N0} (mean {masterSnapshot.MeanTotal:N1})"); + } + table.AddRow("Peak across shards", shardSnapshots.Length > 0 ? shardSnapshots.Max(x => x.MaxTotal).ToString("N0") : "0"); + table.AddRow("Total peak (all shards summed)", shardSnapshots.Sum(x => x.MaxTotal).ToString("N0")); + table.AddRow("Tenants caught up", $"{caughtUp.Count:N0} / {tenants.Length:N0}"); + table.AddRow("Placement violations", placementViolations.Count.ToString("N0")); + table.AddRow("Total elapsed", $"{totalElapsed.Elapsed.TotalSeconds:N1}s"); + AnsiConsole.Write(table); + + if (stalled.Count > 0) + { + AnsiConsole.MarkupLine( + $"[red]{stalled.Count} tenant(s) did not catch up within {input.CatchUpTimeoutSecondsFlag}s: " + + $"{string.Join(", ", stalled.Take(10))}{(stalled.Count > 10 ? ", ..." : "")}[/]"); + } + + if (placementViolations.Count > 0) + { + AnsiConsole.MarkupLine( + $"[red]Cross-shard placement bleed: {string.Join("; ", placementViolations.Take(10))}[/]"); + } + + await WriteShardedMetricsAsync(input, databaseCount, tenants.Length, projectionNames.Length, + appended, appendRate, shardSnapshots, masterSnapshot, caughtUp.Count, stalled, + placementViolations).ConfigureAwait(false); + + var gatePassed = input.MaxConnectionsFlag <= 0 + || shardSnapshots.All(x => x.MaxTotal <= input.MaxConnectionsFlag); + if (!gatePassed) + { + var worst = shardSnapshots.MaxBy(x => x.MaxTotal)!; + AnsiConsole.MarkupLine( + $"[red]GATE FAILED: {worst.Database} peak connections {worst.MaxTotal:N0} > --max-connections {input.MaxConnectionsFlag:N0} (per-database gate).[/]"); + } + + var healthy = appendFailures == 0 && stalled.Count == 0 && placementViolations.Count == 0; + if (!healthy) + { + AnsiConsole.MarkupLine("[red]Run unhealthy — see append failures / stalled tenants / placement bleed above.[/]"); + } + + return gatePassed && healthy; + } + finally + { + foreach (var daemon in daemons) + { + daemon.SafeDispose(); + } + } + } + + /// + /// Wait until, on every shard, every assigned tenant's per-tenant progression rows have + /// reached that tenant's own sequence ceiling. One catch-up query per shard per poll. + /// + private static async Task<(HashSet CaughtUp, List Stalled)> WaitForShardedCatchUpAsync( + IReadOnlyDictionary shardConnectionStrings, + IReadOnlyDictionary tenantsByShard, + string[] projectionNames, + TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + var caughtUp = new HashSet(); + var stalled = new List(); + + while (true) + { + caughtUp.Clear(); + stalled = new List(); + + foreach (var (shard, connectionString) in shardConnectionStrings) + { + var (shardCaughtUp, shardStalled) = await CheckCatchUpOnceAsync( + connectionString, Schema, tenantsByShard[shard], projectionNames).ConfigureAwait(false); + caughtUp.UnionWith(shardCaughtUp); + stalled.AddRange(shardStalled); + } + + if (stalled.Count == 0 || DateTime.UtcNow >= deadline) + { + return (caughtUp, stalled); + } + + await Task.Delay(TimeSpan.FromSeconds(1)).ConfigureAwait(false); + } + } + + /// + /// Database-affine placement sanity: each tenant's per-tenant event sequence must exist in + /// EXACTLY its assigned shard database. A tenant sequence found on a foreign shard (or missing + /// from its home shard) is cross-shard bleed. + /// + private static async Task> CheckPlacementAffinityAsync( + IReadOnlyDictionary shardConnectionStrings, + IReadOnlyDictionary tenantsByShard) + { + var violations = new List(); + + foreach (var (shard, connectionString) in shardConnectionStrings) + { + var found = new HashSet(); + await using (var conn = new NpgsqlConnection(connectionString)) + { + await conn.OpenAsync().ConfigureAwait(false); + await using var cmd = new NpgsqlCommand( + "SELECT replace(sequencename, 'mt_events_sequence_', '') FROM pg_sequences " + + "WHERE schemaname = @schema AND sequencename LIKE 'mt_events_sequence_%'", conn); + cmd.Parameters.AddWithValue("schema", Schema); + await using var reader = await cmd.ExecuteReaderAsync().ConfigureAwait(false); + while (await reader.ReadAsync().ConfigureAwait(false)) + { + found.Add(reader.GetString(0)); + } + } + + var expected = tenantsByShard[shard].ToHashSet(); + foreach (var missing in expected.Where(t => !found.Contains(t))) + { + violations.Add($"{missing} missing from home shard {shard}"); + } + + foreach (var foreign in found.Where(t => !expected.Contains(t))) + { + violations.Add($"{foreign} bled onto foreign shard {shard}"); + } + } + + return violations; + } + + private static async Task EnsureShardDatabasesExistAsync(string[] shardNames) + { + await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); + await conn.OpenAsync().ConfigureAwait(false); + + foreach (var name in shardNames) + { + await using var check = new NpgsqlCommand("SELECT 1 FROM pg_database WHERE datname = @name", conn); + check.Parameters.AddWithValue("name", name); + if (await check.ExecuteScalarAsync().ConfigureAwait(false) == null) + { + AnsiConsole.MarkupLine($"[grey]CREATE DATABASE {name}[/]"); + await using var create = new NpgsqlCommand($"CREATE DATABASE \"{name}\"", conn); + await create.ExecuteNonQueryAsync().ConfigureAwait(false); + } + } + } + + private static async Task WipeShardedAsync(string[] shardNames) + { + await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); + await conn.OpenAsync().ConfigureAwait(false); + + // Registry + assignment tables live in the master database's scaletest schema + await using (var drop = new NpgsqlCommand($"drop schema if exists \"{Schema}\" cascade", conn)) + { + await drop.ExecuteNonQueryAsync().ConfigureAwait(false); + } + + foreach (var name in shardNames) + { + await using var dropDb = new NpgsqlCommand($"DROP DATABASE IF EXISTS \"{name}\" WITH (FORCE)", conn); + await dropDb.ExecuteNonQueryAsync().ConfigureAwait(false); + } + } + + private static async Task WriteShardedMetricsAsync(DaemonLoadInput input, int databases, int tenants, + int projections, long appended, double appendRate, + IReadOnlyList shardSnapshots, + ShardConnectionSampler.DatabaseSnapshot? masterSnapshot, + int caughtUpTenants, List stalledTenants, List placementViolations) + { + if (string.IsNullOrWhiteSpace(input.MetricsFlag)) + { + return; + } + + var doc = new + { + scenario = "daemonload-sharded", + databases, + tenants, + projections, + tenantAgents = tenants * projections, + durationSeconds = input.DurationSecondsFlag, + eventsAppended = appended, + appendRatePerSecond = appendRate, + connectionsPerDatabase = shardSnapshots.ToDictionary( + x => x.Database, + x => new + { + samples = x.SampleCount, + maxTotal = x.MaxTotal, + meanTotal = x.MeanTotal, + maxBusy = x.MaxBusy, + meanBusy = x.MeanBusy + }), + masterDatabase = masterSnapshot == null + ? null + : new + { + database = masterSnapshot.Database, + maxTotal = masterSnapshot.MaxTotal, + meanTotal = masterSnapshot.MeanTotal + }, + peakAcrossShards = shardSnapshots.Count > 0 ? shardSnapshots.Max(x => x.MaxTotal) : 0, + totalPeakSummed = shardSnapshots.Sum(x => x.MaxTotal), + caughtUpTenants, + stalledTenants, + placementViolations + }; + + await File.WriteAllTextAsync(input.MetricsFlag, + JsonSerializer.Serialize(doc, new JsonSerializerOptions { WriteIndented = true })) + .ConfigureAwait(false); + AnsiConsole.MarkupLine($"[grey]Metrics written to {input.MetricsFlag}[/]"); + } +} diff --git a/src/Marten.ScaleTesting/Commands/DaemonLoadCommand.cs b/src/Marten.ScaleTesting/Commands/DaemonLoadCommand.cs index 5edc8b92d0..c90eac101e 100644 --- a/src/Marten.ScaleTesting/Commands/DaemonLoadCommand.cs +++ b/src/Marten.ScaleTesting/Commands/DaemonLoadCommand.cs @@ -36,12 +36,21 @@ namespace Marten.ScaleTesting.Commands; /// turns the report into a pass/fail gate for regression runs. /// [Description("WS2 (jasperfx#486): run the async daemon over N partitioned tenants under continuous append load and sample pg_stat_activity for the store's connection footprint.")] -public sealed class DaemonLoadCommand: JasperFxAsyncCommand +public sealed partial class DaemonLoadCommand: JasperFxAsyncCommand { private const string Schema = "scaletest_daemonload"; private const string ApplicationName = "scaletest-daemonload"; - public override async Task Execute(DaemonLoadInput input) + public override Task Execute(DaemonLoadInput input) + { + // marten#4882: N > 1 pools the tenants across N shard databases (sharded tenancy); + // the original single-database WS2 scenario is untouched at the default of 1. + return input.DatabasesFlag > 1 + ? ExecuteShardedAsync(input) + : ExecuteSingleDatabaseAsync(input); + } + + private async Task ExecuteSingleDatabaseAsync(DaemonLoadInput input) { var totalElapsed = Stopwatch.StartNew(); @@ -246,8 +255,12 @@ private static async Task AppendLoopAsync(IDocumentStore store, string[] tenants /// progression row has reached that tenant's own sequence ceiling /// (last_value of the per-tenant event sequence). /// - private static async Task<(HashSet CaughtUp, List Stalled)> WaitForCatchUpAsync( + private static Task<(HashSet CaughtUp, List Stalled)> WaitForCatchUpAsync( string[] tenants, string[] projectionNames, TimeSpan timeout) + => WaitForCatchUpAsync(ConnectionSource.ConnectionString, Schema, tenants, projectionNames, timeout); + + private static async Task<(HashSet CaughtUp, List Stalled)> WaitForCatchUpAsync( + string connectionString, string schema, string[] tenants, string[] projectionNames, TimeSpan timeout) { var deadline = DateTime.UtcNow + timeout; var stalled = new List(); @@ -255,7 +268,7 @@ private static async Task AppendLoopAsync(IDocumentStore store, string[] tenants while (DateTime.UtcNow < deadline) { - (caughtUp, stalled) = await CheckCatchUpOnceAsync(tenants, projectionNames).ConfigureAwait(false); + (caughtUp, stalled) = await CheckCatchUpOnceAsync(connectionString, schema, tenants, projectionNames).ConfigureAwait(false); if (stalled.Count == 0) { return (caughtUp, stalled); @@ -268,7 +281,7 @@ private static async Task AppendLoopAsync(IDocumentStore store, string[] tenants } private static async Task<(HashSet CaughtUp, List Stalled)> CheckCatchUpOnceAsync( - string[] tenants, string[] projectionNames) + string connectionString, string schema, string[] tenants, string[] projectionNames) { // One round-trip: per tenant, min progression across the per-tenant projection rows vs the // tenant's own sequence ceiling. Sequences are named mt_events_sequence_{tenant} by the @@ -292,14 +305,14 @@ SELECT split_part(name, ':', 3) AS tenant, name, last_seq_id var stalled = new List(); var expectedRows = projectionNames.Length; - await using var conn = new NpgsqlConnection(ConnectionSource.ConnectionString); + await using var conn = new NpgsqlConnection(connectionString); await conn.OpenAsync().ConfigureAwait(false); // Also require the full per-tenant row COUNT so a tenant whose agents never started (zero // progression rows) reads as stalled, not vacuously caught up. var perTenantRows = new Dictionary(); await using (var countCmd = new NpgsqlCommand( - $"select split_part(name, ':', 3), count(*) from {Schema}.mt_event_progression where name like '%:All:%' group by 1", + $"select split_part(name, ':', 3), count(*) from {schema}.mt_event_progression where name like '%:All:%' group by 1", conn)) await using (var countReader = await countCmd.ExecuteReaderAsync().ConfigureAwait(false)) { @@ -309,8 +322,8 @@ SELECT split_part(name, ':', 3) AS tenant, name, last_seq_id } } - await using var cmd = new NpgsqlCommand(sql.Replace("{SCHEMA}", Schema), conn); - cmd.Parameters.AddWithValue("schema", Schema); + await using var cmd = new NpgsqlCommand(sql.Replace("{SCHEMA}", schema), conn); + cmd.Parameters.AddWithValue("schema", schema); await using var reader = await cmd.ExecuteReaderAsync().ConfigureAwait(false); var seen = new HashSet(); diff --git a/src/Marten.ScaleTesting/Commands/DaemonLoadInput.cs b/src/Marten.ScaleTesting/Commands/DaemonLoadInput.cs index cb11412e18..6aeafe9881 100644 --- a/src/Marten.ScaleTesting/Commands/DaemonLoadInput.cs +++ b/src/Marten.ScaleTesting/Commands/DaemonLoadInput.cs @@ -16,6 +16,9 @@ public sealed class DaemonLoadInput: NetCoreInput [Description("Number of tenants to register on the partitioned store. Default: 100.")] public int TenantsFlag { get; set; } = 100; + [Description("marten#4882: number of shard databases to pool the tenants across (sharded tenancy via MultiTenantedWithShardedDatabases). 1 = the original single-database scenario; N > 1 creates scaletest_dl_shard_0..N-1 databases on the same server, assigns tenants round-robin, and runs one daemon per shard. Default: 1.")] + public int DatabasesFlag { get; set; } = 1; + [Description("Number of async projections to register — each fans out one agent per tenant. Default: 2.")] public int ProjectionsFlag { get; set; } = 2; @@ -34,7 +37,7 @@ public sealed class DaemonLoadInput: NetCoreInput [Description("Seconds to wait after appends stop for every tenant's agents to catch up to that tenant's ceiling. Default: 60.")] public int CatchUpTimeoutSecondsFlag { get; set; } = 60; - [Description("Fail (exit 1) if the store's peak concurrent connections exceed this. Default: 0 = report only.")] + [Description("Fail (exit 1) if the store's peak concurrent connections exceed this. With --databases > 1 the gate is enforced PER SHARD DATABASE (the WS6 expectation: each shard's peak mirrors the single-DB governor result, O(databases) total). Default: 0 = report only.")] public int MaxConnectionsFlag { get; set; } [Description("Drop + recreate the dedicated schema before the run. Default: false.")] diff --git a/src/Marten.ScaleTesting/Instrumentation/ShardConnectionSampler.cs b/src/Marten.ScaleTesting/Instrumentation/ShardConnectionSampler.cs new file mode 100644 index 0000000000..7309d9b9d0 --- /dev/null +++ b/src/Marten.ScaleTesting/Instrumentation/ShardConnectionSampler.cs @@ -0,0 +1,174 @@ +using System.Globalization; +using Npgsql; + +namespace Marten.ScaleTesting.Instrumentation; + +/// +/// Multi-database sibling of for the sharded daemonload scenario +/// (marten#4882, epic jasperfx#486 WS6). pg_stat_activity is cluster-wide, so ONE sampling +/// session on the maintenance database sees every shard database's backends; each sample groups +/// the store's Application-Name-attributed connections by datname. The WS6 gate is +/// O(databases): with the 2.22.0 per-database governors, each shard database's peak should mirror +/// the single-DB daemonload result rather than scale with that shard's agent count. +/// +internal sealed class ShardConnectionSampler: IAsyncDisposable +{ + private const string SqlText = @" +SELECT + datname, + count(*) AS total, + count(*) FILTER (WHERE state <> 'idle') AS busy +FROM pg_stat_activity +WHERE pid <> pg_backend_pid() + AND application_name = @app + AND datname = ANY(@dbs) +GROUP BY datname;"; + + private readonly string _connectionString; + private readonly string _applicationName; + private readonly string[] _databases; + private readonly TimeSpan _interval; + private readonly CancellationTokenSource _cts; + private readonly Task _loop; + private readonly List _samples = new(); + private readonly StreamWriter? _traceWriter; + + private ShardConnectionSampler(string connectionString, string applicationName, string[] databases, + TimeSpan interval, string? tracePath, CancellationToken outerCancellation) + { + _connectionString = connectionString; + _applicationName = applicationName; + _databases = databases; + _interval = interval; + _cts = CancellationTokenSource.CreateLinkedTokenSource(outerCancellation); + if (tracePath != null) + { + _traceWriter = new StreamWriter(tracePath, append: false) { AutoFlush = true }; + _traceWriter.WriteLine("timestamp,database,total_connections,busy_connections"); + } + + _loop = Task.Run(LoopAsync, CancellationToken.None); + } + + public static ShardConnectionSampler Start(string connectionString, string applicationName, + string[] databases, TimeSpan interval, string? tracePath, CancellationToken cancellation) + => new(connectionString, applicationName, databases, interval, tracePath, cancellation); + + /// One poll instant: per-database totals. Databases with zero connections at the + /// instant are recorded explicitly so means don't skew optimistic. + private sealed record Sample(DateTimeOffset Timestamp, IReadOnlyDictionary PerDatabase); + + public sealed record DatabaseSnapshot(string Database, int SampleCount, int MaxTotal, double MeanTotal, int MaxBusy, double MeanBusy); + + public IReadOnlyList Capture() + { + lock (_samples) + { + return _databases + .Select(db => + { + var series = _samples + .Select(s => s.PerDatabase.TryGetValue(db, out var counts) ? counts : (Total: 0, Busy: 0)) + .ToArray(); + if (series.Length == 0) + { + return new DatabaseSnapshot(db, 0, 0, 0, 0, 0); + } + + return new DatabaseSnapshot( + db, + series.Length, + series.Max(x => x.Total), + series.Average(x => x.Total), + series.Max(x => x.Busy), + series.Average(x => x.Busy)); + }) + .ToArray(); + } + } + + private async Task LoopAsync() + { + while (!_cts.IsCancellationRequested) + { + try + { + await SampleOnceAsync().ConfigureAwait(false); + } + catch (OperationCanceledException) + { + break; + } + catch (Exception) + { + // Transient sampling failure — skip the sample, keep the loop alive + } + + try + { + await Task.Delay(_interval, _cts.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + break; + } + } + } + + private async Task SampleOnceAsync() + { + await using var conn = new NpgsqlConnection(_connectionString); + await conn.OpenAsync(_cts.Token).ConfigureAwait(false); + await using var cmd = new NpgsqlCommand(SqlText, conn); + cmd.Parameters.AddWithValue("app", _applicationName); + cmd.Parameters.AddWithValue("dbs", _databases); + + var perDatabase = new Dictionary(); + await using (var reader = await cmd.ExecuteReaderAsync(_cts.Token).ConfigureAwait(false)) + { + while (await reader.ReadAsync(_cts.Token).ConfigureAwait(false)) + { + perDatabase[reader.GetString(0)] = + (Convert.ToInt32(reader.GetValue(1)), Convert.ToInt32(reader.GetValue(2))); + } + } + + var sample = new Sample(DateTimeOffset.UtcNow, perDatabase); + lock (_samples) + { + _samples.Add(sample); + } + + if (_traceWriter != null) + { + foreach (var db in _databases) + { + var (total, busy) = perDatabase.TryGetValue(db, out var counts) ? counts : (0, 0); + await _traceWriter.WriteLineAsync(string.Create(CultureInfo.InvariantCulture, + $"{sample.Timestamp:O},{db},{total},{busy}")) + .ConfigureAwait(false); + } + } + } + + public async ValueTask DisposeAsync() + { + _cts.Cancel(); + try + { + await _loop.ConfigureAwait(false); + } + catch + { + // Best-effort teardown + } + + if (_traceWriter != null) + { + await _traceWriter.FlushAsync().ConfigureAwait(false); + _traceWriter.Dispose(); + } + + _cts.Dispose(); + } +} diff --git a/src/Marten.ScaleTesting/README.md b/src/Marten.ScaleTesting/README.md index f06174f02b..c64cc6e28d 100644 --- a/src/Marten.ScaleTesting/README.md +++ b/src/Marten.ScaleTesting/README.md @@ -130,9 +130,37 @@ that tenant's own per-tenant sequence ceiling within `--catch-up-timeout-seconds gate: the WS2 goal is connections O(databases), not O(tenant agents), so run this before and after the daemon command-batching work to quantify the win and then pin it. +### Sharded variant (marten#4882, epic jasperfx#486 WS6) + +`--databases N` (N > 1) pools the tenants across N shard databases on the same server via +`MultiTenantedWithShardedDatabases` — `scaletest_dl_shard_0..N-1` are created on demand +(`--wipe` drops them first), tenants are assigned round-robin with explicit placement, the +harness runs **one projection daemon per shard**, and the `pg_stat_activity` sampler +groups the store's Application-Name-attributed connections by `datname` so every shard +database reports its own peak/mean series (the master/registry database is reported +separately). Three health assertions on top of the single-DB ones: + +* **Per-database gate** — `--max-connections` is enforced per shard database; the WS6 + expectation from the 2.22.0 governors (4 event loads + 4 batch writes + HWM per + database) is that each shard's peak mirrors the single-DB daemonload result, giving + O(databases) total rather than O(agents) +* **Per-tenant catch-up on every shard** — every tenant's per-tenant progression rows + reach that tenant's own sequence ceiling in its own shard database +* **Database-affine placement** — each tenant's per-tenant event sequence exists in + exactly its assigned shard; a sequence on a foreign shard (or missing at home) fails + the run as cross-shard bleed + +```bash +# 100 tenants pooled over 4 shard databases, per-database ceiling of 16 +dotnet run --project src/Marten.ScaleTesting -- daemonload \ + --databases 4 --tenants 100 --projections 2 --duration-seconds 120 --wipe \ + --max-connections 16 --metrics daemonload-sharded.json --trace daemonload-sharded.csv +``` + ## Non-goals * Not a microbenchmark — `src/MartenBenchmarks/` covers per-method timings. * Not a NuGet package — internal tool only. * Not wired into CI. -* Not sharded-PG or distributed. +* Not distributed across machines — the sharded `daemonload` variant shards across + databases on ONE Postgres server.