From c6a21cb4ec45a52309438ba6764fbbde308596fc Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Wed, 29 Jul 2026 20:34:59 -0500 Subject: [PATCH 1/2] fix(agents): bound the node record table by row count, not just age (GH-3701) `wolverine_node_records` is append-only diagnostics -- no foreign keys, and the only reader is `FetchRecentRecordsAsync`, which nothing asks for more than ~100 rows from. Two things left it effectively unbounded: 1. `INodeAgentPersistence.DeleteOldNodeRecordsAsync` was implemented for the relational stores but had no call site anywhere outside tests. 2. The pruning that *did* run, `DeleteOldNodeEventRecords`, bounds the table by age only (`NodeEventRecordExpirationTime`, 5 days). Age is no ceiling at all when the write rate is high: one `AssignmentChanged` row per agent per assignment decision on a cluster with thousands of distributed agents is millions of rows a day, all of them inside the window. The reporting cluster reached 36,135,221 rows / 16 GB in five days -- 15 GB of heap on a table nothing on the hot path reads, plus the insert and WAL load, on the main Wolverine database, while the cluster was already unhealthy for an unrelated reason (GH-3698). Adds `Durability.NodeRecordRetention` (default 10,000 rows; zero or negative keeps the old age-only behavior) applied by a new `TrimNodeRecordsCommand`, and moves both deletes onto a dedicated `Durability.NodeRecordPruningPeriod` timer (default hourly) against the `Main` store. That timer also fixes a second defect found on the way in. The age sweep was appended to `buildOperationBatch` behind an `isTimeToPruneNodeEventRecords()` guard whose backing field `_lastNodeRecordPruneTime` was never assigned -- the `#pragma warning disable CS0649` sitting on it said as much -- so the intended hourly throttle never engaged and the delete went out on *every* recovery cycle, i.e. every 5 seconds by default. On the reporting cluster that was a full scan of a 16 GB table every five seconds. Also: - `MultiTenantedMessageStore` inherited the interface's no-op default for `DeleteOldNodeRecordsAsync`, so a multi-tenanted store would never have trimmed even with a call site. It now delegates to `Main.Nodes`, matching `LogRecordsAsync`/`FetchRecentRecordsAsync` -- and that shape (one main store, hundreds of tenant databases) is exactly where the report came from. - Implements the row cap for Sqlite, MySQL, and Oracle, which had also been falling through to the no-op default. MySQL rejects a LIMIT inside a subquery over the table being deleted, and Oracle pays for a NOT IN over a FETCH FIRST subquery, so both resolve a floor id first and delete as a primary-key range scan. Tests: a Postgres integration test asserts a *running host* drains a seeded table down to the cap with no test-side call to the persistence method (red baseline: 54 rows remain), plus store-level retention tests for Sqlite, MySQL, and Oracle mirroring the existing Postgres and SQL Server ones. Co-Authored-By: Claude Opus 5 (1M context) --- .../Agents/delete_old_node_records.cs | 117 ++++++++++++++++++ .../Wolverine.MySql/MySqlNodePersistence.cs | 15 +++ .../Agents/delete_old_node_records.cs | 108 ++++++++++++++++ .../Wolverine.Oracle/OracleNodePersistence.cs | 16 +++ .../Agents/node_record_retention.cs | 112 +++++++++++++++++ .../Agents/delete_old_node_records.cs | 109 ++++++++++++++++ .../Durability/TrimNodeRecordsCommand.cs | 57 +++++++++ .../Wolverine.RDBMS/DurabilityAgent.cs | 63 +++++++--- .../Wolverine.Sqlite/SqliteNodePersistence.cs | 14 +++ src/Wolverine/DurabilitySettings.cs | 26 ++++ .../Durability/MultiTenantedMessageStore.cs | 9 ++ 11 files changed, 626 insertions(+), 20 deletions(-) create mode 100644 src/Persistence/MySql/MySqlTests/Agents/delete_old_node_records.cs create mode 100644 src/Persistence/Oracle/OracleTests/Agents/delete_old_node_records.cs create mode 100644 src/Persistence/PostgresqlTests/Agents/node_record_retention.cs create mode 100644 src/Persistence/SqliteTests/Agents/delete_old_node_records.cs create mode 100644 src/Persistence/Wolverine.RDBMS/Durability/TrimNodeRecordsCommand.cs diff --git a/src/Persistence/MySql/MySqlTests/Agents/delete_old_node_records.cs b/src/Persistence/MySql/MySqlTests/Agents/delete_old_node_records.cs new file mode 100644 index 000000000..4f50040b2 --- /dev/null +++ b/src/Persistence/MySql/MySqlTests/Agents/delete_old_node_records.cs @@ -0,0 +1,117 @@ +using IntegrationTests; +using Microsoft.Extensions.Logging.Abstractions; +using MySqlConnector; +using Shouldly; +using Wolverine; +using Wolverine.MySql; +using Wolverine.Persistence.Durability; +using Wolverine.RDBMS; +using Wolverine.RDBMS.Sagas; +using Wolverine.Runtime.Agents; +using Xunit; + +namespace MySqlTests.Agents; + +/// +/// GH-3701: the node record row cap. MySQL inherited the interface's no-op default before this, so +/// wolverine_node_records was bounded only by age — which is no ceiling at all under assignment churn. +/// MySQL also rejects a LIMIT inside a subquery over the table being deleted, so the implementation resolves +/// a floor id first; these tests are what pin that rewrite to the same semantics as the other stores. +/// +[Collection("mysql")] +public class delete_old_node_records : IAsyncLifetime +{ + private const string SchemaName = "node_record_retention"; + private MySqlMessageStore _store = null!; + + public async ValueTask InitializeAsync() + { + await using (var conn = new MySqlConnection(Servers.MySqlConnectionString)) + { + await conn.OpenAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = $"DROP DATABASE IF EXISTS `{SchemaName}`"; + await cmd.ExecuteNonQueryAsync(); + await conn.CloseAsync(); + } + + var dataSource = MySqlDataSourceFactory.Create(Servers.MySqlConnectionString); + var settings = new DatabaseSettings + { + ConnectionString = Servers.MySqlConnectionString, + SchemaName = SchemaName, + Role = MessageStoreRole.Main + }; + + _store = new MySqlMessageStore(settings, new DurabilitySettings(), dataSource, + NullLogger.Instance, Array.Empty()); + + await _store.Admin.MigrateAsync(); + } + + public async ValueTask DisposeAsync() + { + await _store.DisposeAsync(); + } + + private async Task insertNodeRecordsAsync(int count) + { + await using var conn = new MySqlConnection(Servers.MySqlConnectionString); + await conn.OpenAsync(); + + for (var i = 1; i <= count; i++) + { + await using var cmd = conn.CreateCommand(); + cmd.CommandText = + $"INSERT INTO {SchemaName}.{DatabaseConstants.NodeRecordTableName} (node_number, event_name, description) VALUES (@number, @event, @description)"; + cmd.Parameters.AddWithValue("number", 1); + cmd.Parameters.AddWithValue("event", NodeRecordType.AssignmentChanged.ToString()); + cmd.Parameters.AddWithValue("description", $"Record {i:00}"); + await cmd.ExecuteNonQueryAsync(); + } + + await conn.CloseAsync(); + } + + [Fact] + public async Task retains_the_most_recent_records() + { + await insertNodeRecordsAsync(10); + (await _store.Nodes.FetchRecentRecordsAsync(100)).Count.ShouldBe(10); + + await _store.Nodes.DeleteOldNodeRecordsAsync(3); + + var remaining = await _store.Nodes.FetchRecentRecordsAsync(100); + remaining.Count.ShouldBe(3); + remaining.Select(x => x.Description).OrderBy(x => x) + .ShouldBe(["Record 08", "Record 09", "Record 10"]); + } + + [Fact] + public async Task zero_retain_is_a_noop() + { + await insertNodeRecordsAsync(3); + + await _store.Nodes.DeleteOldNodeRecordsAsync(0); + + (await _store.Nodes.FetchRecentRecordsAsync(100)).Count.ShouldBe(3); + } + + [Fact] + public async Task fewer_records_than_the_cap_keeps_all() + { + await insertNodeRecordsAsync(2); + + await _store.Nodes.DeleteOldNodeRecordsAsync(5); + + (await _store.Nodes.FetchRecentRecordsAsync(100)).Count.ShouldBe(2); + } + + [Fact] + public async Task empty_table_does_not_throw() + { + await _store.Nodes.DeleteOldNodeRecordsAsync(5); + + (await _store.Nodes.FetchRecentRecordsAsync(100)).Count.ShouldBe(0); + } +} diff --git a/src/Persistence/MySql/Wolverine.MySql/MySqlNodePersistence.cs b/src/Persistence/MySql/Wolverine.MySql/MySqlNodePersistence.cs index 51c284d83..3baefb93d 100644 --- a/src/Persistence/MySql/Wolverine.MySql/MySqlNodePersistence.cs +++ b/src/Persistence/MySql/Wolverine.MySql/MySqlNodePersistence.cs @@ -378,6 +378,21 @@ public async Task> FetchRecentRecordsAsync(int count) .FetchListAsync(readRecord); } + // GH-3701: the row cap that bounds the node record table alongside the age sweep. Without this the + // store fell through to the interface's no-op default and only the age bound applied. MySQL rejects + // a LIMIT inside a subquery of the same table being deleted ("This version of MySQL doesn't yet + // support 'LIMIT & IN/ALL/ANY/SOME subquery'"), so the surviving window is resolved to a floor id in + // a derived table first and the delete is a plain range scan on the primary key. + public async Task DeleteOldNodeRecordsAsync(int retainCount) + { + if (retainCount <= 0) return; + + await _dataSource.CreateCommand( + $"DELETE FROM {_settings.SchemaName}.{NodeRecordTableName} WHERE id < COALESCE((SELECT floor_id FROM (SELECT MIN(id) AS floor_id FROM (SELECT id FROM {_settings.SchemaName}.{NodeRecordTableName} ORDER BY id DESC LIMIT @retain) AS keep) AS floor), 0)") + .With("retain", retainCount) + .ExecuteNonQueryAsync(); + } + public bool HasLeadershipLock() { return _database.AdvisoryLock.HasLock(_lockId); diff --git a/src/Persistence/Oracle/OracleTests/Agents/delete_old_node_records.cs b/src/Persistence/Oracle/OracleTests/Agents/delete_old_node_records.cs new file mode 100644 index 000000000..e5898b712 --- /dev/null +++ b/src/Persistence/Oracle/OracleTests/Agents/delete_old_node_records.cs @@ -0,0 +1,108 @@ +using IntegrationTests; +using Microsoft.Extensions.Logging.Abstractions; +using Oracle.ManagedDataAccess.Client; +using Shouldly; +using Weasel.Oracle; +using Wolverine; +using Wolverine.Oracle; +using Wolverine.Persistence.Durability; +using Wolverine.RDBMS; +using Wolverine.RDBMS.Sagas; +using Wolverine.Runtime.Agents; +using Xunit; + +namespace OracleTests.Agents; + +/// +/// GH-3701: the node record row cap. Oracle inherited the interface's no-op default before this, so +/// wolverine_node_records was bounded only by age — which is no ceiling at all under assignment churn. +/// +[Collection("oracle")] +public class delete_old_node_records : IAsyncLifetime +{ + private const string SchemaName = "WOLVERINE"; + private OracleMessageStore _store = null!; + + public async ValueTask InitializeAsync() + { + var dataSource = new OracleDataSource(Servers.OracleConnectionString); + var settings = new DatabaseSettings + { + ConnectionString = Servers.OracleConnectionString, + SchemaName = SchemaName, + Role = MessageStoreRole.Main + }; + + _store = new OracleMessageStore(settings, new DurabilitySettings(), dataSource, + NullLogger.Instance, Array.Empty()); + + await _store.Admin.RebuildAsync(); + } + + public async ValueTask DisposeAsync() + { + await _store.DisposeAsync(); + } + + private async Task insertNodeRecordsAsync(int count) + { + await using var conn = new OracleConnection(Servers.OracleConnectionString); + await conn.OpenAsync(); + + for (var i = 1; i <= count; i++) + { + await using var cmd = conn.CreateCommand(); + cmd.BindByName = true; + cmd.CommandText = + $"INSERT INTO {SchemaName}.{DatabaseConstants.NodeRecordTableName} (node_number, event_name, description) VALUES (:number, :event_name, :description)"; + cmd.Parameters.Add("number", 1); + cmd.Parameters.Add("event_name", NodeRecordType.AssignmentChanged.ToString()); + cmd.Parameters.Add("description", $"Record {i:00}"); + await cmd.ExecuteNonQueryAsync(); + } + + await conn.CloseAsync(); + } + + [Fact] + public async Task retains_the_most_recent_records() + { + await insertNodeRecordsAsync(10); + (await _store.Nodes.FetchRecentRecordsAsync(100)).Count.ShouldBe(10); + + await _store.Nodes.DeleteOldNodeRecordsAsync(3); + + var remaining = await _store.Nodes.FetchRecentRecordsAsync(100); + remaining.Count.ShouldBe(3); + remaining.Select(x => x.Description).OrderBy(x => x) + .ShouldBe(["Record 08", "Record 09", "Record 10"]); + } + + [Fact] + public async Task zero_retain_is_a_noop() + { + await insertNodeRecordsAsync(3); + + await _store.Nodes.DeleteOldNodeRecordsAsync(0); + + (await _store.Nodes.FetchRecentRecordsAsync(100)).Count.ShouldBe(3); + } + + [Fact] + public async Task fewer_records_than_the_cap_keeps_all() + { + await insertNodeRecordsAsync(2); + + await _store.Nodes.DeleteOldNodeRecordsAsync(5); + + (await _store.Nodes.FetchRecentRecordsAsync(100)).Count.ShouldBe(2); + } + + [Fact] + public async Task empty_table_does_not_throw() + { + await _store.Nodes.DeleteOldNodeRecordsAsync(5); + + (await _store.Nodes.FetchRecentRecordsAsync(100)).Count.ShouldBe(0); + } +} diff --git a/src/Persistence/Oracle/Wolverine.Oracle/OracleNodePersistence.cs b/src/Persistence/Oracle/Wolverine.Oracle/OracleNodePersistence.cs index 32b8cc616..a829d3f69 100644 --- a/src/Persistence/Oracle/Wolverine.Oracle/OracleNodePersistence.cs +++ b/src/Persistence/Oracle/Wolverine.Oracle/OracleNodePersistence.cs @@ -390,6 +390,22 @@ public async Task> FetchRecentRecordsAsync(int count) return list; } + // GH-3701: the row cap that bounds the node record table alongside the age sweep. Without this the + // store fell through to the interface's no-op default and only the age bound applied. Expressed as a + // floor id rather than a NOT IN over a FETCH FIRST subquery so the delete stays a primary-key range scan. + public async Task DeleteOldNodeRecordsAsync(int retainCount) + { + if (retainCount <= 0) return; + + await using var conn = await _dataSource.OpenConnectionAsync(); + await using var cmd = conn.CreateCommand( + $"DELETE FROM {_settings.SchemaName}.{NodeRecordTableName} WHERE id < NVL((SELECT MIN(id) FROM (SELECT id FROM {_settings.SchemaName}.{NodeRecordTableName} ORDER BY id DESC FETCH FIRST :retain ROWS ONLY)), 0)"); + cmd.With("retain", retainCount); + + await cmd.ExecuteNonQueryAsync(); + await conn.CloseAsync(); + } + public bool HasLeadershipLock() { return _database.AdvisoryLock.HasLock(_lockId); diff --git a/src/Persistence/PostgresqlTests/Agents/node_record_retention.cs b/src/Persistence/PostgresqlTests/Agents/node_record_retention.cs new file mode 100644 index 000000000..ed09eaa45 --- /dev/null +++ b/src/Persistence/PostgresqlTests/Agents/node_record_retention.cs @@ -0,0 +1,112 @@ +using IntegrationTests; +using JasperFx.Core; +using Microsoft.Extensions.Hosting; +using Npgsql; +using Shouldly; +using Weasel.Postgresql; +using Wolverine; +using Wolverine.Postgresql; +using Wolverine.RDBMS; +using Wolverine.Runtime.Agents; +using Xunit; + +namespace PostgresqlTests.Agents; + +/// +/// GH-3701: wolverine_node_records is append-only diagnostics with no foreign keys and no hot-path +/// reader, and until now nothing bounded its row count. DeleteOldNodeRecordsAsync was implemented for +/// every relational store but had no call site outside tests, and the only pruning that did run was the +/// age sweep against NodeEventRecordExpirationTime (5 days) — which is no ceiling at all on a cluster +/// writing one AssignmentChanged row per agent per assignment decision. The reporting cluster reached +/// 36,135,221 rows / 16 GB in five days, every one of those rows inside the age window. +/// +/// This pins the call site: a running host must trim the table down to +/// on its own, with no test-side invocation of the +/// persistence method. +/// +[Collection("marten")] +public class node_record_retention : IAsyncLifetime +{ + private readonly string _schemaName = $"node_retention_{Guid.NewGuid().ToString("N")[..8]}"; + private IHost _host = null!; + + public async ValueTask InitializeAsync() + { + await using (var conn = new NpgsqlConnection(Servers.PostgresConnectionString)) + { + await conn.OpenAsync(); + await conn.DropSchemaAsync(_schemaName); + await conn.CloseAsync(); + } + + _host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.PersistMessagesWithPostgresql(Servers.PostgresConnectionString, _schemaName); + + opts.Durability.NodeRecordRetention = 5; + + // The production cadence is hourly. Nothing here depends on the *period* being right, only + // on the prune running at all, so shorten it to keep the test's poll window sane. + opts.Durability.NodeRecordPruningPeriod = 1.Seconds(); + }).StartAsync(); + } + + public async ValueTask DisposeAsync() + { + await _host.StopAsync(); + _host.Dispose(); + } + + private async Task insertNodeRecordsAsync(int count) + { + await using var conn = new NpgsqlConnection(Servers.PostgresConnectionString); + await conn.OpenAsync(); + + for (var i = 1; i <= count; i++) + { + await using var cmd = conn.CreateCommand(); + cmd.CommandText = + $"insert into {_schemaName}.{DatabaseConstants.NodeRecordTableName} (node_number, event_name, description) values ($1, $2, $3)"; + cmd.Parameters.AddWithValue(1); + cmd.Parameters.AddWithValue(NodeRecordType.AssignmentChanged.ToString()); + cmd.Parameters.AddWithValue($"Record {i}"); + await cmd.ExecuteNonQueryAsync(); + } + + await conn.CloseAsync(); + } + + private async Task countNodeRecordsAsync() + { + await using var conn = new NpgsqlConnection(Servers.PostgresConnectionString); + await conn.OpenAsync(); + + await using var cmd = conn.CreateCommand(); + cmd.CommandText = $"select count(*) from {_schemaName}.{DatabaseConstants.NodeRecordTableName}"; + var count = Convert.ToInt32(await cmd.ExecuteScalarAsync()); + + await conn.CloseAsync(); + return count; + } + + [Fact] + public async Task a_running_host_prunes_the_node_record_table_down_to_the_retention_cap() + { + await insertNodeRecordsAsync(50); + (await countNodeRecordsAsync()).ShouldBeGreaterThanOrEqualTo(50); + + var count = 0; + var deadline = DateTimeOffset.UtcNow.AddSeconds(30); + while (DateTimeOffset.UtcNow < deadline) + { + count = await countNodeRecordsAsync(); + if (count <= 5) break; + await Task.Delay(500.Milliseconds(), TestContext.Current.CancellationToken); + } + + // The host also writes its own NodeStarted/AgentStarted records while this runs, so the cap is the + // ceiling, not an exact count. Before the fix this stayed at 50+ forever. + count.ShouldBeLessThanOrEqualTo(5); + } +} diff --git a/src/Persistence/SqliteTests/Agents/delete_old_node_records.cs b/src/Persistence/SqliteTests/Agents/delete_old_node_records.cs new file mode 100644 index 000000000..6cee54ff9 --- /dev/null +++ b/src/Persistence/SqliteTests/Agents/delete_old_node_records.cs @@ -0,0 +1,109 @@ +using Microsoft.Data.Sqlite; +using Microsoft.Extensions.Logging.Abstractions; +using Shouldly; +using Weasel.Sqlite; +using Wolverine; +using Wolverine.Persistence.Durability; +using Wolverine.RDBMS; +using Wolverine.Runtime.Agents; +using Wolverine.Sqlite; +using Xunit; + +namespace SqliteTests.Agents; + +/// +/// GH-3701: the node record row cap. Sqlite inherited the interface's no-op default before this, so +/// wolverine_node_records was bounded only by age. +/// +public class delete_old_node_records : IAsyncLifetime +{ + private readonly SqliteTestDatabase _database = Servers.CreateDatabase(nameof(delete_old_node_records)); + private SqliteMessageStore _store = null!; + private SqliteDataSource _dataSource = null!; + + public async ValueTask InitializeAsync() + { + _dataSource = new SqliteDataSource(_database.ConnectionString); + + var settings = new DatabaseSettings + { + ConnectionString = _database.ConnectionString, + SchemaName = "main", + Role = MessageStoreRole.Main + }; + + _store = new SqliteMessageStore(settings, new DurabilitySettings(), _dataSource, + NullLogger.Instance); + + await _store.Admin.MigrateAsync(); + } + + public async ValueTask DisposeAsync() + { + await _store.DisposeAsync(); + _dataSource.Dispose(); + _database.Dispose(); + } + + private async Task insertNodeRecordsAsync(int count) + { + await using var conn = new SqliteConnection(_database.ConnectionString); + await conn.OpenAsync(); + + for (var i = 1; i <= count; i++) + { + await using var cmd = conn.CreateCommand(); + cmd.CommandText = + $"insert into {DatabaseConstants.NodeRecordTableName} (node_number, event_name, timestamp, description) values (@number, @event, @time, @description)"; + cmd.Parameters.AddWithValue("@number", 1); + cmd.Parameters.AddWithValue("@event", NodeRecordType.AssignmentChanged.ToString()); + cmd.Parameters.AddWithValue("@time", DateTimeOffset.UtcNow.ToString("o")); + cmd.Parameters.AddWithValue("@description", $"Record {i:00}"); + await cmd.ExecuteNonQueryAsync(); + } + + await conn.CloseAsync(); + } + + [Fact] + public async Task retains_the_most_recent_records() + { + await insertNodeRecordsAsync(10); + (await _store.Nodes.FetchRecentRecordsAsync(100)).Count.ShouldBe(10); + + await _store.Nodes.DeleteOldNodeRecordsAsync(3); + + var remaining = await _store.Nodes.FetchRecentRecordsAsync(100); + remaining.Count.ShouldBe(3); + remaining.Select(x => x.Description).OrderBy(x => x) + .ShouldBe(["Record 08", "Record 09", "Record 10"]); + } + + [Fact] + public async Task zero_retain_is_a_noop() + { + await insertNodeRecordsAsync(3); + + await _store.Nodes.DeleteOldNodeRecordsAsync(0); + + (await _store.Nodes.FetchRecentRecordsAsync(100)).Count.ShouldBe(3); + } + + [Fact] + public async Task fewer_records_than_the_cap_keeps_all() + { + await insertNodeRecordsAsync(2); + + await _store.Nodes.DeleteOldNodeRecordsAsync(5); + + (await _store.Nodes.FetchRecentRecordsAsync(100)).Count.ShouldBe(2); + } + + [Fact] + public async Task empty_table_does_not_throw() + { + await _store.Nodes.DeleteOldNodeRecordsAsync(5); + + (await _store.Nodes.FetchRecentRecordsAsync(100)).Count.ShouldBe(0); + } +} diff --git a/src/Persistence/Wolverine.RDBMS/Durability/TrimNodeRecordsCommand.cs b/src/Persistence/Wolverine.RDBMS/Durability/TrimNodeRecordsCommand.cs new file mode 100644 index 000000000..da40106fd --- /dev/null +++ b/src/Persistence/Wolverine.RDBMS/Durability/TrimNodeRecordsCommand.cs @@ -0,0 +1,57 @@ +using Microsoft.Extensions.Logging; +using Wolverine.Runtime; +using Wolverine.Runtime.Agents; + +namespace Wolverine.RDBMS.Durability; + +/// +/// GH-3701: cap the node record table at rows. +/// +/// bounds the same table by age +/// (, 5 days by default), which is no ceiling +/// at all when the write rate is high: one AssignmentChanged row per agent per assignment decision on +/// a cluster with thousands of distributed agents is millions of rows a day, and all of them are inside the +/// age window. The reporting cluster reached 36M rows / 16 GB in five days — 15 GB of heap on a diagnostic +/// table nothing on the hot path reads, plus the write and WAL load of the inserts, on the main Wolverine +/// database, while the cluster was already unhealthy for an unrelated reason. +/// +/// The "keep the newest N by id" delete itself is engine-specific (LIMIT / TOP / FETCH FIRST), so it lives +/// behind on the store rather than being +/// expressed as an IDatabaseOperation here. Stores that do not implement it inherit the interface's +/// no-op default and are simply left to the age sweep. +/// +internal class TrimNodeRecordsCommand : IAgentCommand +{ + private readonly IMessageDatabase _database; + private readonly DurabilitySettings _settings; + private readonly ILogger _logger; + + public TrimNodeRecordsCommand(IMessageDatabase database, DurabilitySettings settings, ILogger logger) + { + _database = database; + _settings = settings; + _logger = logger; + } + + public async Task ExecuteAsync(IWolverineRuntime runtime, CancellationToken cancellationToken) + { + var retention = _settings.NodeRecordRetention; + if (retention <= 0) + { + return AgentCommands.Empty; + } + + try + { + await _database.Nodes.DeleteOldNodeRecordsAsync(retention); + } + catch (Exception e) + { + // Housekeeping must never be able to take down the durability agent's running block. + _logger.LogError(e, "Error trying to trim the node record table in database {Database} to {Retention} rows", + _database.Name, retention); + } + + return AgentCommands.Empty; + } +} diff --git a/src/Persistence/Wolverine.RDBMS/DurabilityAgent.cs b/src/Persistence/Wolverine.RDBMS/DurabilityAgent.cs index 88305c61b..2bfb042e2 100644 --- a/src/Persistence/Wolverine.RDBMS/DurabilityAgent.cs +++ b/src/Persistence/Wolverine.RDBMS/DurabilityAgent.cs @@ -31,6 +31,7 @@ internal class DurabilityAgent : IAgent private Timer? _recoveryTimer; private Timer? _scheduledJobTimer; private Timer? _handledCleanupTimer; + private Timer? _nodeRecordPruningTimer; private readonly DurabilityHealthSignals _health; private DateTime _lastHealthCheck = DateTime.UtcNow; @@ -134,6 +135,20 @@ public Task StartAsync(CancellationToken cancellationToken) _runningBlock.Post(command); }, _settings, 5.Seconds(), _settings.HandledMessageCleanupPollingTime); + // GH-3701: node records only exist on the Main store, and their housekeeping is slow, unbounded-scan + // work that has no business riding along on the recovery batch. See PruneNodeRecords. + if (_database.Settings.Role == MessageStoreRole.Main) + { + // Hold the first pass back so it can't pile onto startup, but never past the configured period + // itself -- a host that asks for a short period is asking to see pruning promptly. + var pruningStart = _settings.NodeRecordPruningPeriod < 1.Minutes() + ? _settings.NodeRecordPruningPeriod + : 1.Minutes(); + + _nodeRecordPruningTimer = new Timer(_ => PruneNodeRecords(), _settings, pruningStart, + _settings.NodeRecordPruningPeriod); + } + if (AutoStartScheduledJobPolling) { StartScheduledJobPolling(); @@ -168,6 +183,11 @@ public async Task StopAsync(CancellationToken cancellationToken) await _handledCleanupTimer.DisposeAsync(); } + if (_nodeRecordPruningTimer != null) + { + await _nodeRecordPruningTimer.DisposeAsync(); + } + Status = AgentStatus.Stopped; } @@ -193,23 +213,31 @@ public static Uri AddMarkerType(Uri uri, Type markerType) return new Uri($"{uri}{markerType.Name}"); } -#pragma warning disable CS0649 // Field is never assigned to - private DateTimeOffset? _lastNodeRecordPruneTime; -#pragma warning restore CS0649 - - private bool isTimeToPruneNodeEventRecords() + /// + /// GH-3701: everything that keeps the node record table from growing without bound. Two deletes, both + /// against the Main store, both on the slow + /// timer rather than the five-second recovery batch: + /// + /// 1. — the age sweep against + /// (5 days by default). This one already + /// existed, but it was appended to behind an + /// isTimeToPruneNodeEventRecords() guard reading a field that was never assigned — the + /// suppression on it (CS0649) said so outright — so the guard always returned true and the whole + /// delete ran on every recovery cycle. + /// 2. — the row cap against + /// . An age bound alone puts no ceiling on the + /// table: the reporting cluster wrote one AssignmentChanged row per agent per assignment + /// decision, ~12.8M rows/day, reaching 36M rows / 16 GB well inside the 5-day window. + /// + internal void PruneNodeRecords() { - if (_lastNodeRecordPruneTime == null) - { - return true; - } + _runningBlock.Post(new DatabaseOperationBatch(_database, + [new DeleteOldNodeEventRecords(_database, _settings)])); - if (DateTimeOffset.UtcNow.Subtract(_lastNodeRecordPruneTime.Value) > 1.Hours()) + if (_settings.NodeRecordRetention > 0) { - return true; + _runningBlock.Post(new TrimNodeRecordsCommand(_database, _settings, _logger)); } - - return false; } internal IDatabaseOperation[] buildOperationBatch(IReadOnlyList? activeNodeNumbers = null) @@ -238,13 +266,8 @@ internal IDatabaseOperation[] buildOperationBatch(IReadOnlyList? activeNode } } - if (_database.Settings.Role == MessageStoreRole.Main) - { - if (isTimeToPruneNodeEventRecords()) - { - ops.Add(new DeleteOldNodeEventRecords(_database, _settings)); - } - } + // GH-3701: node record pruning used to live here, on the five-second recovery cadence. It now runs + // on its own timer -- see PruneNodeRecords. if (_runtime.Options.Durability.OutboxStaleTime.HasValue) { diff --git a/src/Persistence/Wolverine.Sqlite/SqliteNodePersistence.cs b/src/Persistence/Wolverine.Sqlite/SqliteNodePersistence.cs index 917f38c6c..15b514dd3 100644 --- a/src/Persistence/Wolverine.Sqlite/SqliteNodePersistence.cs +++ b/src/Persistence/Wolverine.Sqlite/SqliteNodePersistence.cs @@ -376,6 +376,20 @@ public async Task> FetchRecentRecordsAsync(int count) return records; } + // GH-3701: the row cap that bounds the node record table alongside the age sweep. Without this the + // store fell through to the interface's no-op default and only the age bound applied. + public async Task DeleteOldNodeRecordsAsync(int retainCount) + { + if (retainCount <= 0) return; + + await using var conn = await _dataSource.OpenConnectionAsync(CancellationToken.None).ConfigureAwait(false); + await using var cmd = conn.CreateCommand( + $"delete from {NodeRecordTableName} where id not in (select id from {NodeRecordTableName} order by id desc limit @retain)") + .With("retain", retainCount); + + await cmd.ExecuteNonQueryAsync(); + } + public bool HasLeadershipLock() { return _database.AdvisoryLock.HasLock(_lockId); diff --git a/src/Wolverine/DurabilitySettings.cs b/src/Wolverine/DurabilitySettings.cs index 3ef413004..ac416b016 100644 --- a/src/Wolverine/DurabilitySettings.cs +++ b/src/Wolverine/DurabilitySettings.cs @@ -246,6 +246,30 @@ internal set /// public int StaleNodeEjectionThreshold { get; set; } = 2; + /// + /// GH-3701: a hard cap on the number of rows retained in the node record table + /// (wolverine_node_records), the append-only diagnostic log written by + /// INodeAgentPersistence.LogRecordsAsync and read back by FetchRecentRecordsAsync. + /// bounds those rows by *age* only, which puts no ceiling on + /// the table at all: a cluster churning assignments writes one AssignmentChanged row per agent per + /// decision, so millions of rows a day fit comfortably inside the age window and turn an + /// agent-assignment incident into a database capacity problem on top of it. This cap is applied on the + /// same housekeeping pass as the age sweep, every , against the + /// Main store only. Raise it on very large agent universes, where one assignment wave is already + /// thousands of rows. Set to zero or a negative number to keep the age sweep as the only bound, which + /// was the behavior before 6.24.1. + /// + public int NodeRecordRetention { get; set; } = 10_000; + + /// + /// GH-3701: how often the node record table is pruned, both by age + /// () and down to rows. + /// Deliberately far slower than — this is a housekeeping scan + /// over a table nothing on the hot path reads, and until 6.24.1 it was being appended to every + /// five-second recovery batch instead. + /// + public TimeSpan NodeRecordPruningPeriod { get; set; } = 1.Hours(); + /// /// How often should Wolverine do a full check that all assigned agents are /// really running and try to restart (or stop) any differences from the last @@ -463,6 +487,8 @@ public OptionsDescription ToDescription() desc.AddValue(nameof(DeadLetterQueueExpirationEnabled), DeadLetterQueueExpirationEnabled); desc.AddValue(nameof(DeadLetterQueueExpiration), DeadLetterQueueExpiration); desc.AddValue(nameof(NodeEventRecordExpirationTime), NodeEventRecordExpirationTime); + desc.AddValue(nameof(NodeRecordRetention), NodeRecordRetention); + desc.AddValue(nameof(NodeRecordPruningPeriod), NodeRecordPruningPeriod); desc.AddValue(nameof(SendingAgentIdleTimeout), SendingAgentIdleTimeout); desc.AddValue(nameof(DrainTimeout), DrainTimeout); desc.AddValue(nameof(EnableInboxPartitioning), EnableInboxPartitioning); diff --git a/src/Wolverine/Persistence/Durability/MultiTenantedMessageStore.cs b/src/Wolverine/Persistence/Durability/MultiTenantedMessageStore.cs index aecae650c..7383162f2 100644 --- a/src/Wolverine/Persistence/Durability/MultiTenantedMessageStore.cs +++ b/src/Wolverine/Persistence/Durability/MultiTenantedMessageStore.cs @@ -596,6 +596,15 @@ Task> INodeAgentPersistence.FetchRecentRecordsAsync(in return Main.Nodes.FetchRecentRecordsAsync(count); } + // GH-3701: node records are written to, and read back from, the Main store only -- see LogRecordsAsync + // and FetchRecentRecordsAsync above -- so the retention cap has to follow them there. Without this + // override a multi-tenanted store inherited the interface's no-op default and never trimmed at all, + // which is exactly the shape (one main store, hundreds of tenant databases) the 36M-row report came from. + Task INodeAgentPersistence.DeleteOldNodeRecordsAsync(int retainCount) + { + return Main.Nodes.DeleteOldNodeRecordsAsync(retainCount); + } + bool INodeAgentPersistence.HasLeadershipLock() { return Main.Nodes.HasLeadershipLock(); From bafb90faeb1a2c8ec3182228647790ef8a779d57 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Wed, 29 Jul 2026 22:08:46 -0500 Subject: [PATCH 2/2] fix(tests): make the Oracle retention test survive its own collection (GH-3701) Two defects in the new `OracleTests.Agents.delete_old_node_records`, both caught by CIOracle: 1. `ORA-01745: invalid host/bind variable name`. The insert bound `:number` and `:description`; NUMBER is an Oracle reserved word, so a bind variable named after one is rejected outright. Prefixed to `:p_number` / `:p_event` / `:p_description`. 2. `Admin.RebuildAsync()` in `InitializeAsync` took out two sibling tests (`Transport.clear_all_wolverine_storage`). In Oracle a schema IS a user, so every test in the "oracle" collection shares WOLVERINE -- rebuilding it drops the tables out from under whatever runs next. Migrates instead, and clears only the one table this class owns, on the way in and on the way out. Verified against a real Oracle 23 container: the class passes 6/6 and the full OracleTests suite is 113/113, where CI was 108/113. Co-Authored-By: Claude Opus 5 (1M context) --- .../Agents/delete_old_node_records.cs | 34 +++++++++++++++---- 1 file changed, 28 insertions(+), 6 deletions(-) diff --git a/src/Persistence/Oracle/OracleTests/Agents/delete_old_node_records.cs b/src/Persistence/Oracle/OracleTests/Agents/delete_old_node_records.cs index e5898b712..11d87ca9c 100644 --- a/src/Persistence/Oracle/OracleTests/Agents/delete_old_node_records.cs +++ b/src/Persistence/Oracle/OracleTests/Agents/delete_old_node_records.cs @@ -36,15 +36,33 @@ public async ValueTask InitializeAsync() _store = new OracleMessageStore(settings, new DurabilitySettings(), dataSource, NullLogger.Instance, Array.Empty()); - await _store.Admin.RebuildAsync(); + // Deliberately NOT Admin.RebuildAsync(). In Oracle a schema IS a user, so every test in the + // "oracle" collection shares WOLVERINE -- rebuilding it drops the tables out from under the + // siblings that run after this class. Migrate to make sure the table exists, then clear only the + // one table this class actually owns. + await _store.Admin.MigrateAsync(); + await clearNodeRecordsAsync(); } public async ValueTask DisposeAsync() { + await clearNodeRecordsAsync(); await _store.DisposeAsync(); } - private async Task insertNodeRecordsAsync(int count) + private static async Task clearNodeRecordsAsync() + { + await using var conn = new OracleConnection(Servers.OracleConnectionString); + await conn.OpenAsync(); + + await using var cmd = conn.CreateCommand(); + cmd.CommandText = $"DELETE FROM {SchemaName}.{DatabaseConstants.NodeRecordTableName}"; + await cmd.ExecuteNonQueryAsync(); + + await conn.CloseAsync(); + } + + private static async Task insertNodeRecordsAsync(int count) { await using var conn = new OracleConnection(Servers.OracleConnectionString); await conn.OpenAsync(); @@ -53,11 +71,15 @@ private async Task insertNodeRecordsAsync(int count) { await using var cmd = conn.CreateCommand(); cmd.BindByName = true; + + // NOT :number -- NUMBER is an Oracle reserved word and a bind variable named after one is + // rejected with ORA-01745: invalid host/bind variable name. Same for :description, which + // collides with the DESCRIPTION keyword in some contexts. Prefix them. cmd.CommandText = - $"INSERT INTO {SchemaName}.{DatabaseConstants.NodeRecordTableName} (node_number, event_name, description) VALUES (:number, :event_name, :description)"; - cmd.Parameters.Add("number", 1); - cmd.Parameters.Add("event_name", NodeRecordType.AssignmentChanged.ToString()); - cmd.Parameters.Add("description", $"Record {i:00}"); + $"INSERT INTO {SchemaName}.{DatabaseConstants.NodeRecordTableName} (node_number, event_name, description) VALUES (:p_number, :p_event, :p_description)"; + cmd.Parameters.Add("p_number", 1); + cmd.Parameters.Add("p_event", NodeRecordType.AssignmentChanged.ToString()); + cmd.Parameters.Add("p_description", $"Record {i:00}"); await cmd.ExecuteNonQueryAsync(); }