CSHARP-4831: Emit heartbeat started event before connection establishment in ServerMonitor (#1253)

Author: BorisDog

src/MongoDB.Driver.Core/Core/Servers/ServerMonitor.cs

@@ -202,18 +202,24 @@ private CommandWireProtocol<BsonDocument> InitializeHelloProtocol(IConnection co
         private IConnection InitializeConnection(CancellationToken cancellationToken) // called setUpConnection in spec
         {
             var connection = _connectionFactory.CreateConnection(_serverId, _endPoint);
+            _eventLoggerSdam.LogAndPublish(new ServerHeartbeatStartedEvent(connection.ConnectionId, false));
 
             var stopwatch = Stopwatch.StartNew();
             try
             {
                 // if we are cancelling, it's because the server has
                 // been shut down and we really don't need to wait.
                 connection.Open(cancellationToken);
+
+                _eventLoggerSdam.LogAndPublish(new ServerHeartbeatSucceededEvent(connection.ConnectionId, stopwatch.Elapsed, false, connection.Description.HelloResult.Wrapped));
             }
-            catch
+            catch (Exception exception)
             {
                 // dispose it here because the _connection is not initialized yet
                 try { connection.Dispose(); } catch { }
+
+                _eventLoggerSdam.LogAndPublish(new ServerHeartbeatFailedEvent(connection.ConnectionId, stopwatch.Elapsed, exception, false));
+
                 throw;
             }
             stopwatch.Stop();

tests/MongoDB.Driver.Core.Tests/Core/Servers/ServerMonitorTests.cs

@@ -146,7 +146,9 @@ public void DescriptionChanged_should_be_raised_during_initial_handshake()
             changes[0].OldServerDescription.State.Should().Be(ServerState.Disconnected);
             changes[0].NewServerDescription.State.Should().Be(ServerState.Connected);
 
-            capturedEvents.Any().Should().BeFalse();
+            // ServerHeartbeatStartedEvent and ServerHeartbeatSucceededEvent events should be emitted during initial handshake
+            capturedEvents.Next().Should().BeOfType<ServerHeartbeatStartedEvent>();
+            capturedEvents.Next().Should().BeOfType<ServerHeartbeatSucceededEvent>();
         }
 #endif
 
@@ -163,8 +165,9 @@ public void Description_should_be_connected_after_successful_heartbeat()
             subject.Description.State.Should().Be(ServerState.Connected);
             subject.Description.Type.Should().Be(ServerType.Standalone);
 
-            // no ServerHeartbeat events should be triggered during initial handshake
-            capturedEvents.Any().Should().BeFalse();
+            // ServerHeartbeatStartedEvent and ServerHeartbeatSucceededEvent events should be emitted during initial handshake
+            capturedEvents.Next().Should().BeOfType<ServerHeartbeatStartedEvent>();
+            capturedEvents.Next().Should().BeOfType<ServerHeartbeatSucceededEvent>();
         }
 
         [Fact]
@@ -229,12 +232,13 @@ public void Heartbeat_should_make_immediate_next_attempt_for_streaming_protocol(
                 : 2;
             capturedEvents.WaitForOrThrowIfTimeout(
                 events =>
-                    events.Count(e => e is ServerDescriptionChangedEvent) >= expectedServerDescriptionChangedEventCount,  // the connection has been initialized and the first heatbeat event has been fired
+                    events.Count(e => e is ServerDescriptionChangedEvent) >= expectedServerDescriptionChangedEventCount,  // the connection has been initialized and the first heartbeat event has been fired
                 TimeSpan.FromSeconds(10));
 
+            capturedEvents.Next().Should().BeOfType<ServerHeartbeatSucceededEvent>(); // heartbeat succeeded before connection initialized
             capturedEvents.Next().Should().BeOfType<ServerDescriptionChangedEvent>(); // connection initialized
             AssertHeartbeatAttempt();
-            capturedEvents.Any().Should().BeFalse(); // the next attempt will be in 10 seconds because the second stremable respone has 10 seconds delay
+            capturedEvents.Any().Should().BeFalse(); // the next attempt will be in 10 seconds because the second streamable response has 10 seconds delay
 
             void AssertHeartbeatAttempt()
             {
@@ -248,7 +252,10 @@ void AssertHeartbeatAttempt()
                     var serverDescriptionChangedEvent = capturedEvents.Next().Should().BeOfType<ServerDescriptionChangedEvent>().Subject;
                     serverDescriptionChangedEvent.NewDescription.HeartbeatException.Should().Be(exception);
 
-                    serverDescriptionChangedEvent = capturedEvents.Next().Should().BeOfType<ServerDescriptionChangedEvent>().Subject;  // when we catch exceptions, we close the current connection, so opening connection will trigger one more ServerDescriptionChangedEvent
+                    // when we catch exceptions, we close the current connection,
+                    // so opening connection will trigger one more ServerHeartbeatSucceededEvent and ServerDescriptionChangedEvent
+                    capturedEvents.Next().Should().BeOfType<ServerHeartbeatSucceededEvent>();
+                    serverDescriptionChangedEvent = capturedEvents.Next().Should().BeOfType<ServerDescriptionChangedEvent>().Subject;
                     serverDescriptionChangedEvent.OldDescription.HeartbeatException.Should().Be(exception);
                     serverDescriptionChangedEvent.NewDescription.HeartbeatException.Should().BeNull();
                 }

tests/MongoDB.Driver.Tests/Specifications/connection-monitoring-and-pooling/ConnectionMonitoringAndPoolingTestRunner.cs

@@ -702,6 +702,13 @@ or ConnectionCreatedEvent
                                 return EndPointHelper.Equals(serverId.EndPoint, server.EndPoint);
                             }
 
+                            if (o is ServerHeartbeatStartedEvent ||
+                                o is ServerHeartbeatSucceededEvent ||
+                                o is ServerHeartbeatFailedEvent)
+                            {
+                                return false;
+                            }
+
                             return true;
                         };
 

tests/MongoDB.Driver.Tests/Specifications/server-discovery-and-monitoring/ServerDiscoveryAndMonitoringProseTests.cs

@@ -16,6 +16,7 @@
 using System;
 using System.Collections.Concurrent;
 using System.Diagnostics;
+using System.IO;
 using System.Linq;
 using System.Net;
 using System.Threading;
@@ -25,12 +26,14 @@
 using MongoDB.Driver.Core;
 using MongoDB.Driver.Core.Bindings;
 using MongoDB.Driver.Core.Clusters.ServerSelectors;
+using MongoDB.Driver.Core.Connections;
 using MongoDB.Driver.Core.Events;
 using MongoDB.Driver.Core.Misc;
 using MongoDB.Driver.Core.Servers;
 using MongoDB.Driver.Core.TestHelpers;
 using MongoDB.Driver.Core.TestHelpers.XunitExtensions;
 using MongoDB.Driver.TestHelpers;
+using Moq;
 using Xunit;
 
 namespace MongoDB.Driver.Tests.Specifications.server_discovery_and_monitoring
@@ -41,12 +44,12 @@ public class ServerDiscoveryAndMonitoringProseTests
         [Fact]
         public void Heartbeat_should_work_as_expected()
         {
-            var heartbeatSuceededTimestamps = new ConcurrentQueue<DateTime>();
+            var heartbeatSucceededTimestamps = new ConcurrentQueue<DateTime>();
             var eventCapturer = new EventCapturer()
                 .Capture<ServerHeartbeatSucceededEvent>(
                     (@event) =>
                     {
-                        heartbeatSuceededTimestamps.Enqueue(DateTime.UtcNow);
+                        heartbeatSucceededTimestamps.Enqueue(DateTime.UtcNow);
                         return true;
                     }
                 );
@@ -64,21 +67,79 @@ public void Heartbeat_should_work_as_expected()
                     });
             }
 
-            var heartbeatSuceededTimestampsList = heartbeatSuceededTimestamps.ToList();
+            var heartbeatSucceededTimestampsList = heartbeatSucceededTimestamps.ToList();
             // we have at least 3 items here
             // Skip the first event because we have nothing to compare it to
-            for (int i = 1; i < heartbeatSuceededTimestampsList.Count; i++)
+            for (int i = 1; i < heartbeatSucceededTimestampsList.Count; i++)
             {
-                var attemptDuration = heartbeatSuceededTimestampsList[i] - heartbeatSuceededTimestampsList[i - 1];
+                var attemptDuration = heartbeatSucceededTimestampsList[i] - heartbeatSucceededTimestampsList[i - 1];
                 attemptDuration
                     .Should()
                     .BeLessThan(TimeSpan.FromSeconds(2));
                 // Assert the client processes heartbeat replies more frequently than 10 secs (approximately every 500ms)
             }
         }
 
+        // Heartbeat prose tests
+        // https://github.com/mongodb/specifications/blob/c5771dce88eed54386690039f76142e1d741d83f/source/server-discovery-and-monitoring/tests/README.rst?plain=1#L268-L289
         [Fact]
-        public void Monitor_sleep_at_least_minHeartbeatFreqencyMS_between_checks()
+        public void Heartbeat_should_be_emitted_before_connection_open()
+        {
+            const string HelloReceivedEvent = "client hello received";
+            const string ClientConnectedEvent = "client connected";
+
+            var events = new ConcurrentQueue<string>();
+            var eventCapturer = new EventCapturer()
+                .Capture<ServerHeartbeatStartedEvent>(e => EnqueueEvent(nameof(ServerHeartbeatStartedEvent)))
+                .Capture<ServerHeartbeatSucceededEvent>(e => EnqueueEvent(nameof(ServerHeartbeatSucceededEvent)))
+                .Capture<ServerHeartbeatFailedEvent>(e => EnqueueEvent(nameof(ServerHeartbeatFailedEvent)));
+
+            var mockStream = new Mock<Stream>();
+            mockStream
+                .Setup(s => s.Write(It.IsAny<byte[]>(), It.IsAny<int>(), It.IsAny<int>()))
+                .Callback(() => EnqueueEvent(HelloReceivedEvent))
+                .Throws(new Exception("Stream is closed."));
+
+            var mockStreamFactory = new Mock<IStreamFactory>();
+            mockStreamFactory
+                .Setup(s => s.CreateStream(It.IsAny<EndPoint>(), It.IsAny<CancellationToken>()))
+                .Callback(() => EnqueueEvent(ClientConnectedEvent))
+                .Returns(mockStream.Object);
+
+            var settings = new MongoClientSettings();
+            settings.ServerSelectionTimeout = TimeSpan.FromMilliseconds(500);
+            settings.HeartbeatInterval = TimeSpan.FromMilliseconds(5000);
+            settings.ClusterConfigurator = c => c
+                .Subscribe(eventCapturer)
+                .RegisterStreamFactory(f => mockStreamFactory.Object);
+
+            using var client = DriverTestConfiguration.CreateDisposableClient(settings);
+
+            eventCapturer.WaitForEventOrThrowIfTimeout<ServerHeartbeatFailedEvent>(TimeSpan.FromSeconds(5));
+
+            events.ShouldAllBeEquivalentTo(new[]
+            {
+                nameof(ServerHeartbeatStartedEvent),
+                HelloReceivedEvent,
+                ClientConnectedEvent,
+                nameof(ServerHeartbeatFailedEvent),
+            });
+
+            bool EnqueueEvent(string @event)
+            {
+                // Ignore RTT events
+                if (Environment.StackTrace.Contains(nameof(RoundTripTimeMonitor)))
+                {
+                    return false;
+                }
+
+                events.Enqueue(@event);
+                return true;
+            }
+        }
+
+        [Fact]
+        public void Monitor_sleep_at_least_minHeartbeatFrequencyMS_between_checks()
         {
             var minVersion = new SemanticVersion(4, 9, 0, "");
             RequireServer.Check().VersionGreaterThanOrEqualTo(minVersion);
@@ -232,6 +293,7 @@ public void ConnectionPool_cleared_on_failed_hello()
                 {
                     typeof(ServerHeartbeatFailedEvent),
                     typeof(ConnectionPoolClearedEvent),
+                    typeof(ServerHeartbeatSucceededEvent),
                     typeof(ConnectionPoolReadyEvent),
                     typeof(ServerHeartbeatSucceededEvent),
                 },