Bound the node record table by row count, not just age (GH-3701)

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;
+
+/// <summary>
+/// GH-3701: the node record row cap. MySQL inherited the interface's no-op default before this, so
+/// <c>wolverine_node_records</c> 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.
+/// </summary>
+[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<MySqlMessageStore>.Instance, Array.Empty<SagaTableDefinition>());
+
+        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);
+    }
+}

src/Persistence/MySql/Wolverine.MySql/MySqlNodePersistence.cs

@@ -378,6 +378,21 @@ public async Task<IReadOnlyList<NodeRecord>> 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);

src/Persistence/Oracle/OracleTests/Agents/delete_old_node_records.cs

@@ -0,0 +1,130 @@
+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;
+
+/// <summary>
+/// GH-3701: the node record row cap. Oracle inherited the interface's no-op default before this, so
+/// <c>wolverine_node_records</c> was bounded only by age — which is no ceiling at all under assignment churn.
+/// </summary>
+[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<OracleMessageStore>.Instance, Array.Empty<SagaTableDefinition>());
+
+        // 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 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();
+
+        for (var i = 1; i <= count; i++)
+        {
+            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 (: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();
+        }
+
+        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);
+    }
+}

src/Persistence/Oracle/Wolverine.Oracle/OracleNodePersistence.cs

@@ -390,6 +390,22 @@ public async Task<IReadOnlyList<NodeRecord>> 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);