diff --git a/src/Netclaw.Actors.Tests/Channels/ChannelBotFilterOrderTests.cs b/src/Netclaw.Actors.Tests/Channels/ChannelBotFilterOrderTests.cs index 22369332e5..cbfe5e65ee 100644 --- a/src/Netclaw.Actors.Tests/Channels/ChannelBotFilterOrderTests.cs +++ b/src/Netclaw.Actors.Tests/Channels/ChannelBotFilterOrderTests.cs @@ -116,14 +116,20 @@ public async Task Mattermost_person_denied_by_ACL_is_dropped() await AssertDroppedNotFiltered(ChannelType.Mattermost); } + // RecordEventDropped and RecordEventFiltered bump the snapshot total first. The meter listener then runs and + // records the reason. So a wait on the total does not prove that the reason exists. Each helper waits on the + // reason itself, and only then checks the negative side. private async Task AssertFilteredNotDropped(ChannelType channelType) { var metrics = ChannelTelemetry.For(channelType); await AwaitAssertAsync( - () => Assert.Equal(1, metrics.GetSnapshot().EventsFiltered), + () => + { + Assert.Equal(1, metrics.GetSnapshot().EventsFiltered); + Assert.Equal(["bot_message"], Reasons("events.filtered")); + }, cancellationToken: TestContext.Current.CancellationToken); Assert.Equal(0, metrics.GetSnapshot().EventsDropped); - Assert.Equal(["bot_message"], Reasons("events.filtered")); Assert.Empty(Reasons("events.dropped")); } @@ -131,10 +137,13 @@ private async Task AssertDroppedNotFiltered(ChannelType channelType) { var metrics = ChannelTelemetry.For(channelType); await AwaitAssertAsync( - () => Assert.Equal(1, metrics.GetSnapshot().EventsDropped), + () => + { + Assert.Equal(1, metrics.GetSnapshot().EventsDropped); + Assert.Equal([AclDenyReasons.UserNotAllowed], Reasons("events.dropped")); + }, cancellationToken: TestContext.Current.CancellationToken); Assert.Equal(0, metrics.GetSnapshot().EventsFiltered); - Assert.Equal([AclDenyReasons.UserNotAllowed], Reasons("events.dropped")); Assert.Empty(Reasons("events.filtered")); } diff --git a/src/Netclaw.Actors.Tests/Sessions/ChildParkSqliteRecoveryTests.cs b/src/Netclaw.Actors.Tests/Sessions/ChildParkSqliteRecoveryTests.cs index d89a09c091..f8e0280ba9 100644 --- a/src/Netclaw.Actors.Tests/Sessions/ChildParkSqliteRecoveryTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/ChildParkSqliteRecoveryTests.cs @@ -5,7 +5,6 @@ // ----------------------------------------------------------------------- using Akka.Hosting; using Akka.Persistence.Sql.Hosting; -using Microsoft.Data.Sqlite; using Netclaw.Actors.Protocol; using Netclaw.Actors.SubAgents; using Netclaw.Tests.Utilities; @@ -32,9 +31,14 @@ protected override void ConfigureSessionPersistence(AkkaConfigurationBuilder bui protected override void DeleteOwnedDirectories() { - SqliteConnection.ClearAllPools(); - _sqliteDirectory?.DisposeAsync().AsTask().GetAwaiter().GetResult(); - base.DeleteOwnedDirectories(); + try + { + SqlPersistencePlugins.WaitForStopThenDelete(Sys, _sqliteDirectory); + } + finally + { + base.DeleteOwnedDirectories(); + } } /// @@ -48,7 +52,8 @@ public async Task A_parked_run_survives_a_SQLite_journal_and_snapshot_restart() { // Arrange var id = new SessionId("park/sqlite"); - await SqlPersistencePlugins.StartOneAtATimeAsync(this, ColdRecoveryCeiling); + await SqlPersistencePlugins.StartOneAtATimeAsync(this, ColdRecoveryCeiling, + Path.Combine(_sqliteDirectory!.Path, "journal.db")); ScriptParent(); PlanChild("park-sqlite", [[Probe("p1")]]); await JoinRecorderAsync(id); diff --git a/src/Netclaw.Actors.Tests/Sessions/LegacyCoordinationSqliteRecoveryTests.cs b/src/Netclaw.Actors.Tests/Sessions/LegacyCoordinationSqliteRecoveryTests.cs index c0f84583e0..76f65ea5bb 100644 --- a/src/Netclaw.Actors.Tests/Sessions/LegacyCoordinationSqliteRecoveryTests.cs +++ b/src/Netclaw.Actors.Tests/Sessions/LegacyCoordinationSqliteRecoveryTests.cs @@ -61,7 +61,7 @@ await manager.Ask(new SendUserMessage { SessionId = sessionId, Content = "Continue after the recovered child result." - }, TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + }, ColdRecoveryCeiling, TestContext.Current.CancellationToken); await subscriber.FishForMessageAsync(_ => true, TimeSpan.FromSeconds(10), cancellationToken: TestContext.Current.CancellationToken); @@ -111,7 +111,7 @@ await manager.Ask(new SendUserMessage { SessionId = sessionId, Content = "Continue after the quarantined legacy snapshot." - }, TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + }, ColdRecoveryCeiling, TestContext.Current.CancellationToken); await subscriber.FishForMessageAsync(_ => true, TimeSpan.FromSeconds(10), cancellationToken: TestContext.Current.CancellationToken); @@ -157,7 +157,7 @@ await manager.Ask(new SendUserMessage { SessionId = sessionId, Content = "Continue after the notice." - }, TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + }, ColdRecoveryCeiling, TestContext.Current.CancellationToken); await subscriber.FishForMessageAsync(_ => true, TimeSpan.FromSeconds(10), cancellationToken: TestContext.Current.CancellationToken); var later = _chatClient.ReceivedMessages.Last(); diff --git a/src/Netclaw.Actors.Tests/Sessions/RecordedJournalTestBase.cs b/src/Netclaw.Actors.Tests/Sessions/RecordedJournalTestBase.cs index 03daa8d601..e588888687 100644 --- a/src/Netclaw.Actors.Tests/Sessions/RecordedJournalTestBase.cs +++ b/src/Netclaw.Actors.Tests/Sessions/RecordedJournalTestBase.cs @@ -56,10 +56,7 @@ protected sealed override void ConfigureSessionSerialization(AkkaConfigurationBu .WithStrictSerialization(); protected override void DeleteOwnedDirectories() - { - SqliteConnection.ClearAllPools(); - _sqliteDirectory?.DisposeAsync().AsTask().GetAwaiter().GetResult(); - } + => SqlPersistencePlugins.WaitForStopThenDelete(Sys, _sqliteDirectory); /// Writes the named corpus records into the journal of one persistence id, in order. protected async Task SeedJournalAsync(string persistenceId, string seederName, params string[] recordNames) @@ -96,13 +93,18 @@ private async Task StopSeederAsync(IActorRef seeder) /// private async Task StartSeederAsync(string persistenceId, string actorName) { - await SqlPersistencePlugins.StartOneAtATimeAsync(this, ColdRecoveryCeiling); + await SqlPersistencePlugins.StartOneAtATimeAsync(this, ColdRecoveryCeiling, SqliteDatabasePath()); var recovered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var seeder = Sys.ActorOf(Props.Create(() => new RecordedRowSeeder(persistenceId, recovered)), actorName); await recovered.Task.WaitAsync(ColdRecoveryCeiling, TestContext.Current.CancellationToken); return seeder; } + private string SqliteDatabasePath() + => Path.Combine( + _sqliteDirectory?.Path ?? throw new InvalidOperationException("The test SQLite directory was not configured."), + "journal.db"); + private void BackupAndRestoreSqliteDatabase() { var directory = _sqliteDirectory?.Path @@ -258,17 +260,89 @@ internal static class SqlPersistencePlugins /// fight. That timeout is equal to the cold ceiling, so a long fight fails the start wait with no cause (#2599). /// One plugin at a time has no fight. A warm plugin answers at once. /// - public static async Task StartOneAtATimeAsync(Akka.TestKit.TestKitBase kit, TimeSpan ceiling) + public static async Task StartOneAtATimeAsync(Akka.TestKit.TestKitBase kit, TimeSpan ceiling, string databasePath) { const string unusedId = "recorded-journal-plugin-start"; var persistence = AkkaPersistence.Instance.Apply(kit.Sys); var probe = kit.CreateTestProbe(); - persistence.JournalFor(null).Tell(new ReplayMessages(0, long.MaxValue, long.MaxValue, unusedId, probe.Ref)); - await probe.ExpectMsgAsync(ceiling, - cancellationToken: TestContext.Current.CancellationToken); - persistence.SnapshotStoreFor(null).Tell( + var journal = persistence.JournalFor(null); + journal.Tell(new ReplayMessages(0, long.MaxValue, long.MaxValue, unusedId, probe.Ref)); + await ExpectStartReplyAsync(probe, ceiling, "journal", journal, databasePath); + var snapshotStore = persistence.SnapshotStoreFor(null); + snapshotStore.Tell( new LoadSnapshot(unusedId, SnapshotSelectionCriteria.Latest, long.MaxValue), probe.Ref); - await probe.ExpectMsgAsync(ceiling, - cancellationToken: TestContext.Current.CancellationToken); + await ExpectStartReplyAsync(probe, ceiling, "snapshot store", snapshotStore, databasePath); + } + + // The plugin that does not reply is the one that failed to start. The message names it and the database file, + // so a stall shows where it happened. A plain timeout would not say which plugin blocked. + private static async Task ExpectStartReplyAsync( + Akka.TestKit.TestProbe probe, TimeSpan ceiling, string plugin, IActorRef pluginActor, string databasePath) + { + try + { + await probe.ExpectMsgAsync(ceiling, cancellationToken: TestContext.Current.CancellationToken); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + throw new InvalidOperationException( + $"The SQL {plugin} plugin at {pluginActor.Path} did not send {typeof(TReply).Name} within " + + $"{ceiling.TotalSeconds:0} s. The plugins start in this order: journal, snapshot store. " + + $"Database: {databasePath} (file exists: {File.Exists(databasePath)}).", ex); + } + } + + // A fault ceiling for the stop of the actor system. It is not an orchestration delay. + private static readonly TimeSpan TerminationCeiling = TimeSpan.FromSeconds(15); + + private const int DeleteAttempts = 3; + + /// + /// Waits for the actor system to terminate, and then deletes the SQLite folder. The journal and the snapshot store + /// are actors of the system. Each one holds a connection until it stops. A plugin that blocks in a SQL call + /// cannot stop, and its connection stays open. Windows then refuses the delete with "journal.db in use". The wait is + /// bounded. If it ends first, the method still tries the delete, and then reports the stall. The seeders of the + /// base run in the same system, so this one wait covers them. + /// + public static void WaitForStopThenDelete(ActorSystem system, TestSessionTempDirectory? directory) + { + TimeoutException? stall = null; + if (!system.WhenTerminated.Wait(TerminationCeiling)) + { + stall = new TimeoutException( + $"The actor system did not terminate within {TerminationCeiling.TotalSeconds:0} s. " + + "A SQL persistence plugin actor probably still holds its connection."); + } + + try + { + DeleteWithPoolClear(directory); + } + catch (IOException ex) when (stall is not null) + { + throw new AggregateException(stall, ex); + } + + if (stall is not null) + throw stall; + } + + // Each attempt clears the pools first. A connection that a plugin returns late goes back to the pool, and only + // a new clear closes it. DisposableTempDir retries the delete between the attempts. + [SlopwatchSuppress("SW003", "A failed delete attempt is retried; the last attempt throws.")] + private static void DeleteWithPoolClear(TestSessionTempDirectory? directory) + { + for (var attempt = 1; attempt <= DeleteAttempts; attempt++) + { + SqliteConnection.ClearAllPools(); + try + { + directory?.DisposeAsync().AsTask().GetAwaiter().GetResult(); + return; + } + catch (IOException) when (attempt < DeleteAttempts) + { + } + } } } diff --git a/src/Netclaw.Cli.MutationTests/HeadlessStopRuleMutationTests.cs b/src/Netclaw.Cli.MutationTests/HeadlessStopRuleMutationTests.cs index 677724d765..0a9af2fde4 100644 --- a/src/Netclaw.Cli.MutationTests/HeadlessStopRuleMutationTests.cs +++ b/src/Netclaw.Cli.MutationTests/HeadlessStopRuleMutationTests.cs @@ -30,8 +30,13 @@ public sealed class HeadlessStopRuleMutationTests private const string Session = "fake/session"; // A fault ceiling. A run that never stops fails here. No test orchestrates with this wait. + // It starts after the send. Stryker pays it in full for each mutant that never stops, so keep it short. private static readonly TimeSpan StopCeiling = TimeSpan.FromSeconds(10); + // A ceiling for the cold start of the first run in the process. It covers the session create, the first + // log file, and the first send. A loaded Windows runner needs more than 10 s for this work. + private static readonly TimeSpan ColdStartCeiling = TimeSpan.FromSeconds(30); + [Fact] public async Task An_open_run_keeps_the_wait_until_a_turn_reports_none() { @@ -103,10 +108,12 @@ private static async Task RunAsync(params SessionOutputDto[] outputs) { var directory = Path.Combine(Path.GetTempPath(), "netclaw-headless-stop-rule-" + Guid.NewGuid().ToString("N")); var transport = new FakeDaemonHubTransport(); + var sent = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); transport.VoidInvokeHook = (method, _, _) => { if (method == "SendMessage") { + sent.TrySetResult(); // The client pump delivers the outputs in order, so the channel sees every turn. foreach (var output in outputs) transport.PushOutput(output); @@ -137,7 +144,28 @@ private static async Task RunAsync(params SessionOutputDto[] outputs) try { await channel.StartAsync(CancellationToken.None); - await stopped.Task.WaitAsync(StopCeiling); + // Phase one: the cold start. The ceiling of the stop rule does not cover it. + try + { + await sent.Task.WaitAsync(ColdStartCeiling); + } + catch (TimeoutException) + { + throw new TimeoutException( + $"Cold start phase: the channel did not send the message within {ColdStartCeiling.TotalSeconds:0} s."); + } + + // Phase two: the stop rule. The outputs are already queued, so the run stops soon after the send. + try + { + await stopped.Task.WaitAsync(StopCeiling); + } + catch (TimeoutException) + { + throw new TimeoutException( + $"Stop phase: the run did not stop within {StopCeiling.TotalSeconds:0} s after the send."); + } + await channel.StopAsync(CancellationToken.None); Console.SetOut(originalOut); Console.SetError(originalError);