Fix flaky monitoring and throughput query tests

src/ServiceControl.AcceptanceTests/Monitoring/CustomChecks/When_email_notifications_are_enabled.cs

@@ -24,7 +24,7 @@ public async Task Should_send_custom_check_status_change_emails()
         {
             var emailDropPath = Path.Combine(Path.GetTempPath(), Path.GetRandomFileName());
             Directory.CreateDirectory(emailDropPath);
-            string[] emails = [];
+            string[] emailHeaders = [];
 
             SetSettings = settings =>
             {
@@ -41,23 +41,51 @@ await Define<MyContext>(c =>
                 .WithEndpoint<EndpointWithFailingCustomCheck>()
                 .Done(c =>
                 {
-                    emails = Directory.EnumerateFiles(emailDropPath).ToArray();
-                    return emails.Length > 0;
+                    var emails = Directory.EnumerateFiles(emailDropPath).ToArray();
+
+                    return emails.Length > 0 && TryReadHeaders(emails[0], out emailHeaders);
                 })
                 .Run();
 
-            Assert.That(emails, Is.Not.Empty);
-
-            var emailText = await File.ReadAllLinesAsync(emails[0]);
+            Assert.That(emailHeaders, Is.Not.Empty);
 
             using (Assert.EnterMultipleScope())
             {
-                Assert.That(emailText[0], Is.EqualTo("X-Sender: YouServiceControl@particular.net"));
-                Assert.That(emailText[1], Is.EqualTo("X-Receiver: WhoeverMightBeConcerned@particular.net"));
-                Assert.That(emailText[3], Is.EqualTo("From: YouServiceControl@particular.net"));
-                Assert.That(emailText[4], Is.EqualTo("To: WhoeverMightBeConcerned@particular.net"));
-                Assert.That(emailText[6], Is.EqualTo("Subject: [Particular.ServiceControl] health check failed"));
+                Assert.That(emailHeaders[0], Is.EqualTo("X-Sender: YouServiceControl@particular.net"));
+                Assert.That(emailHeaders[1], Is.EqualTo("X-Receiver: WhoeverMightBeConcerned@particular.net"));
+                Assert.That(emailHeaders[3], Is.EqualTo("From: YouServiceControl@particular.net"));
+                Assert.That(emailHeaders[4], Is.EqualTo("To: WhoeverMightBeConcerned@particular.net"));
+                Assert.That(emailHeaders[6], Is.EqualTo("Subject: [Particular.ServiceControl] health check failed"));
+            }
+        }
+
+        // SmtpClient creates the file in the pickup folder before it writes the message into it, so
+        // the headers can only be read once the blank line that terminates them has been written.
+        static bool TryReadHeaders(string emailFile, out string[] headers)
+        {
+            headers = [];
+
+            string[] lines;
+
+            try
+            {
+                lines = File.ReadAllLines(emailFile);
             }
+            catch (IOException)
+            {
+                return false;
+            }
+
+            var endOfHeaders = Array.IndexOf(lines, string.Empty);
+
+            if (endOfHeaders < 0)
+            {
+                return false;
+            }
+
+            headers = lines[..endOfHeaders];
+
+            return true;
         }
 
         class SetupNotificationSettings(INotificationsDataStore notificationsDataStore) : IHostedService

src/ServiceControl.Transports.PostgreSql.Tests/PostgreSqlQueryTests.cs

@@ -95,29 +95,42 @@ public async Task RunScenario()
         Assert.That(queue, Is.Not.Null);
 
         long total = 0L;
-        using var reset = new ManualResetEventSlim();
 
-        var runScenarioAndAdvanceTime = Task.Run(async () =>
-        {
-            while (!reset.IsSet)
-            {
-                await SendAndReceiveMessages(transportSettings.EndpointName, 1);
-                provider.Advance(TimeSpan.FromHours(1));
-            }
-        }, token);
+        await using var throughputPerDay = query.GetThroughputPerDay(queue, new DateOnly(), token).GetAsyncEnumerator(token);
 
-        await foreach (QueueThroughput queueThroughput in query.GetThroughputPerDay(queue, new DateOnly(), token))
+        // Sending the message before asking for the next hourly value keeps every message inside
+        // exactly one snapshot interval, instead of racing the snapshots from a background task.
+        for (int hour = 0; hour < 24; hour++)
         {
-            total += queueThroughput.TotalThroughput;
-        }
+            await SendAndReceiveMessages(transportSettings.EndpointName, 1);
 
-        reset.Set();
-        await runScenarioAndAdvanceTime.WaitAsync(token);
+            Assert.That(await MoveToNextHour(throughputPerDay, token), Is.True);
+
+            total += throughputPerDay.Current.TotalThroughput;
+        }
 
-        // Asserting that we have one message per hour during 24 hours, the first snapshot is not counted hence the 23 assertion.
+        // The sequence already reports a last_value of 1 before the first message is sent, so that
+        // message is not visible in any of the hourly deltas, hence the 23 instead of 24.
         Assert.That(total, Is.EqualTo(23));
     }
 
+    // The query only takes its next snapshot once an hour has passed on the fake clock, but it
+    // registers that timer asynchronously after the previous value was consumed, so an advance can
+    // land before the timer exists. Keep advancing until the value arrives.
+    async Task<bool> MoveToNextHour(IAsyncEnumerator<QueueThroughput> throughputPerDay, CancellationToken cancellationToken)
+    {
+        var moveNext = throughputPerDay.MoveNextAsync().AsTask();
+
+        while (!moveNext.IsCompleted)
+        {
+            provider.Advance(TimeSpan.FromHours(1));
+
+            await Task.WhenAny(moveNext, Task.Delay(TimeSpan.FromMilliseconds(50), cancellationToken));
+        }
+
+        return await moveNext;
+    }
+
     [Test]
     public async Task NoNegativeThroughputWhenQueueTableIsDeletedBetweenSnapshots()
     {

src/ServiceControl.Transports.SqlServer.Tests/SqlServerQueryTests.cs

@@ -115,26 +115,39 @@ public async Task RunScenario()
         Assert.That(queue, Is.Not.Null);
 
         long total = 0L;
-        using var reset = new ManualResetEventSlim();
 
-        var runScenarioAndAdvanceTime = Task.Run(async () =>
-        {
-            while (!reset.IsSet)
-            {
-                await SendAndReceiveMessages(transportSettings.EndpointName, 1);
-                provider.Advance(TimeSpan.FromHours(1));
-            }
-        }, token);
-
-        await foreach (QueueThroughput queueThroughput in query.GetThroughputPerDay(queue, new DateOnly(), token))
+        await using var throughputPerDay = query.GetThroughputPerDay(queue, new DateOnly(), token).GetAsyncEnumerator(token);
+
+        // Sending the message before asking for the next hourly value keeps every message inside
+        // exactly one snapshot interval, instead of racing the snapshots from a background task.
+        for (int hour = 0; hour < 24; hour++)
         {
-            total += queueThroughput.TotalThroughput;
-        }
+            await SendAndReceiveMessages(transportSettings.EndpointName, 1);
 
-        reset.Set();
-        await runScenarioAndAdvanceTime.WaitAsync(token);
+            Assert.That(await MoveToNextHour(throughputPerDay, token), Is.True);
 
-        // Asserting that we have one message per hour during 24 hours, the first snapshot is not counted hence the 23 assertion. 
+            total += throughputPerDay.Current.TotalThroughput;
+        }
+
+        // IDENT_CURRENT already reports 1 before the first message is sent, so that message is not
+        // visible in any of the hourly deltas, hence the 23 instead of 24.
         Assert.That(total, Is.EqualTo(23));
     }
+
+    // The query only takes its next snapshot once an hour has passed on the fake clock, but it
+    // registers that timer asynchronously after the previous value was consumed, so an advance can
+    // land before the timer exists. Keep advancing until the value arrives.
+    async Task<bool> MoveToNextHour(IAsyncEnumerator<QueueThroughput> throughputPerDay, CancellationToken cancellationToken)
+    {
+        var moveNext = throughputPerDay.MoveNextAsync().AsTask();
+
+        while (!moveNext.IsCompleted)
+        {
+            provider.Advance(TimeSpan.FromHours(1));
+
+            await Task.WhenAny(moveNext, Task.Delay(TimeSpan.FromMilliseconds(50), cancellationToken));
+        }
+
+        return await moveNext;
+    }
 }