feat(NODE-5825): add minRoundTripTime to ServerDescription and change roundTripTime to a moving average

src/cmap/connection.ts

@@ -28,6 +28,7 @@ import { MongoLoggableComponent, type MongoLogger, SeverityLevel } from '../mong
 import { type CancellationToken, TypedEventEmitter } from '../mongo_types';
 import { ReadPreference, type ReadPreferenceLike } from '../read_preference';
 import { ServerType } from '../sdam/common';
+import { type Monitor } from '../sdam/monitor';
 import { applySession, type ClientSession, updateSessionFromResponse } from '../sessions';
 import {
   BufferPool,
@@ -125,6 +126,8 @@ export interface ConnectionOptions
   extendedMetadata: Promise<Document>;
   /** @internal */
   mongoLogger?: MongoLogger | undefined;
+  /** @internal */
+  parent?: Monitor;
 }
 
 /** @public */

src/index.ts

@@ -552,6 +552,7 @@ export type {
   List,
   MongoDBCollectionNamespace,
   MongoDBNamespace,
+  MovingWindow,
   TimeoutController
 } from './utils';
 export type { W, WriteConcernOptions, WriteConcernSettings } from './write_concern';

src/sdam/monitor.ts

@@ -8,8 +8,15 @@ import { LEGACY_HELLO_COMMAND } from '../constants';
 import { MongoError, MongoErrorLabel, MongoNetworkTimeoutError } from '../error';
 import { MongoLoggableComponent } from '../mongo_logger';
 import { CancellationToken, TypedEventEmitter } from '../mongo_types';
-import type { Callback, EventEmitterWithState } from '../utils';
-import { calculateDurationInMs, makeStateMachine, now, ns } from '../utils';
+import {
+  calculateDurationInMs,
+  type Callback,
+  type EventEmitterWithState,
+  makeStateMachine,
+  MovingWindow,
+  now,
+  ns
+} from '../utils';
 import { ServerType, STATE_CLOSED, STATE_CLOSING } from './common';
 import {
   ServerHeartbeatFailedEvent,
@@ -25,8 +32,6 @@ const kServer = Symbol('server');
 const kMonitorId = Symbol('monitorId');
 /** @internal */
 const kCancellationToken = Symbol('cancellationToken');
-/** @internal */
-const kRoundTripTime = Symbol('roundTripTime');
 
 const STATE_IDLE = 'idle';
 const STATE_MONITORING = 'monitoring';
@@ -100,6 +105,8 @@ export class Monitor extends TypedEventEmitter<MonitorEvents> {
   rttPinger?: RTTPinger;
   /** @internal */
   override component = MongoLoggableComponent.TOPOLOGY;
+  /** @internal */
+  rttSamplesMS: MovingWindow;
 
   constructor(server: Server, options: MonitorOptions) {
     super();
@@ -121,6 +128,7 @@ export class Monitor extends TypedEventEmitter<MonitorEvents> {
     });
     this.isRunningInFaasEnv = getFAASEnv() != null;
     this.mongoLogger = this[kServer].topology.client?.mongoLogger;
+    this.rttSamplesMS = new MovingWindow(10);
 
     const cancellationToken = this[kCancellationToken];
     // TODO: refactor this to pull it directly from the pool, requires new ConnectionPool integration
@@ -135,7 +143,8 @@ export class Monitor extends TypedEventEmitter<MonitorEvents> {
       useBigInt64: false,
       promoteLongs: true,
       promoteValues: true,
-      promoteBuffers: true
+      promoteBuffers: true,
+      parent: this
     };
 
     // ensure no authentication is used for monitoring
@@ -203,6 +212,26 @@ export class Monitor extends TypedEventEmitter<MonitorEvents> {
     this.emit('close');
     stateTransition(this, STATE_CLOSED);
   }
+
+  get roundTripTime(): number {
+    return this.rttSamplesMS.average();
+  }
+
+  get minRoundTripTime(): number {
+    return this.rttSamplesMS.min();
+  }
+
+  get latestRTT(): number {
+    return this.rttSamplesMS.last ?? 0; // FIXME: Check if this is acceptable
+  }
+
+  addRttSample(rtt: number) {
+    this.rttSamplesMS.addSample(rtt);
+  }
+
+  clearRttSamples() {
+    this.rttSamplesMS.clear();
+  }
 }
 
 function resetMonitorState(monitor: Monitor) {
@@ -216,6 +245,8 @@ function resetMonitorState(monitor: Monitor) {
 
   monitor.connection?.destroy();
   monitor.connection = null;
+
+  monitor.clearRttSamples();
 }
 
 function useStreamingProtocol(monitor: Monitor, topologyVersion: TopologyVersion | null): boolean {
@@ -249,7 +280,6 @@ function checkServer(monitor: Monitor, callback: Callback<Document | null>) {
   function onHeartbeatFailed(err: Error) {
     monitor.connection?.destroy();
     monitor.connection = null;
-
     monitor.emitAndLogHeartbeat(
       Server.SERVER_HEARTBEAT_FAILED,
       monitor[kServer].topology.s.id,
@@ -275,11 +305,15 @@ function checkServer(monitor: Monitor, callback: Callback<Document | null>) {
       hello.isWritablePrimary = hello[LEGACY_HELLO_COMMAND];
     }
 
+    // NOTE: here we use the latestRTT as this measurment corresponds with the value
+    // obtained for this successful heartbeat
     const duration =
       isAwaitable && monitor.rttPinger
-        ? monitor.rttPinger.roundTripTime
+        ? monitor.rttPinger.latestRTT ?? calculateDurationInMs(start)
         : calculateDurationInMs(start);
 
+    monitor.addRttSample(duration);
+
     monitor.emitAndLogHeartbeat(
       Server.SERVER_HEARTBEAT_SUCCEEDED,
       monitor[kServer].topology.s.id,
@@ -329,6 +363,7 @@ function checkServer(monitor: Monitor, callback: Callback<Document | null>) {
 
     if (isAwaitable && monitor.rttPinger == null) {
       monitor.rttPinger = new RTTPinger(
+        monitor,
         monitor[kCancellationToken],
         Object.assign(
           { heartbeatFrequencyMS: monitor.options.heartbeatFrequencyMS },
@@ -377,6 +412,8 @@ function checkServer(monitor: Monitor, callback: Callback<Document | null>) {
         connection.destroy();
         return;
       }
+      const duration = calculateDurationInMs(start);
+      monitor.addRttSample(duration);
 
       monitor.connection = connection;
       monitor.emitAndLogHeartbeat(
@@ -385,7 +422,7 @@ function checkServer(monitor: Monitor, callback: Callback<Document | null>) {
         connection.hello?.connectionId,
         new ServerHeartbeatSucceededEvent(
           monitor.address,
-          calculateDurationInMs(start),
+          duration,
           connection.hello,
           useStreamingProtocol(monitor, connection.hello?.topologyVersion)
         )
@@ -458,23 +495,30 @@ export class RTTPinger {
   /** @internal */
   [kCancellationToken]: CancellationToken;
   /** @internal */
-  [kRoundTripTime]: number;
-  /** @internal */
   [kMonitorId]: NodeJS.Timeout;
+  /** @internal */
+  monitor: Monitor;
   closed: boolean;
+  /** @internal */
+  latestRTT?: number;
 
-  constructor(cancellationToken: CancellationToken, options: RTTPingerOptions) {
+  constructor(monitor: Monitor, cancellationToken: CancellationToken, options: RTTPingerOptions) {
     this.connection = undefined;
     this[kCancellationToken] = cancellationToken;
-    this[kRoundTripTime] = 0;
     this.closed = false;
+    this.monitor = monitor;
+    this.latestRTT = monitor.latestRTT;
 
     const heartbeatFrequencyMS = options.heartbeatFrequencyMS;
     this[kMonitorId] = setTimeout(() => measureRoundTripTime(this, options), heartbeatFrequencyMS);
   }
 
   get roundTripTime(): number {
-    return this[kRoundTripTime];
+    return this.monitor.roundTripTime;
+  }
+
+  get minRoundTripTime(): number {
+    return this.monitor.minRoundTripTime;
   }
 
   close(): void {
@@ -486,42 +530,48 @@ export class RTTPinger {
   }
 }
 
-function measureRoundTripTime(rttPinger: RTTPinger, options: RTTPingerOptions) {
-  const start = now();
-  options.cancellationToken = rttPinger[kCancellationToken];
-  const heartbeatFrequencyMS = options.heartbeatFrequencyMS;
-
+function measureAndReschedule(
+  rttPinger: RTTPinger,
+  options: RTTPingerOptions,
+  start?: number,
+  conn?: Connection
+) {
+  if (start == null) {
+    start = now();
+  }
   if (rttPinger.closed) {
+    conn?.destroy();
     return;
   }
 
-  function measureAndReschedule(conn?: Connection) {
-    if (rttPinger.closed) {
-      conn?.destroy();
-      return;
-    }
+  if (rttPinger.connection == null) {
+    rttPinger.connection = conn;
+  }
 
-    if (rttPinger.connection == null) {
-      rttPinger.connection = conn;
-    }
+  rttPinger.latestRTT = calculateDurationInMs(start);
+  rttPinger[kMonitorId] = setTimeout(
+    () => measureRoundTripTime(rttPinger, options),
+    options.heartbeatFrequencyMS
+  );
+}
 
-    rttPinger[kRoundTripTime] = calculateDurationInMs(start);
-    rttPinger[kMonitorId] = setTimeout(
-      () => measureRoundTripTime(rttPinger, options),
-      heartbeatFrequencyMS
-    );
+function measureRoundTripTime(rttPinger: RTTPinger, options: RTTPingerOptions) {
+  const start = now();
+  options.cancellationToken = rttPinger[kCancellationToken];
+
+  if (rttPinger.closed) {
+    return;
   }
 
   const connection = rttPinger.connection;
   if (connection == null) {
     // eslint-disable-next-line github/no-then
     connect(options).then(
       connection => {
-        measureAndReschedule(connection);
+        measureAndReschedule(rttPinger, options, start, connection);
       },
       () => {
         rttPinger.connection = undefined;
-        rttPinger[kRoundTripTime] = 0;
       }
     );
     return;
@@ -531,11 +581,10 @@ function measureRoundTripTime(rttPinger: RTTPinger, options: RTTPingerOptions) {
     connection.serverApi?.version || connection.helloOk ? 'hello' : LEGACY_HELLO_COMMAND;
   // eslint-disable-next-line github/no-then
   connection.command(ns('admin.$cmd'), { [commandName]: 1 }, undefined).then(
-    () => measureAndReschedule(),
+    () => measureAndReschedule(rttPinger, options),
     () => {
       rttPinger.connection?.destroy();
       rttPinger.connection = undefined;
-      rttPinger[kRoundTripTime] = 0;
       return;
     }
   );

src/sdam/server.ts

@@ -175,7 +175,8 @@ export class Server extends TypedEventEmitter<ServerEvents> {
       this.emit(
         Server.DESCRIPTION_RECEIVED,
         new ServerDescription(this.description.hostAddress, event.reply, {
-          roundTripTime: calculateRoundTripTime(this.description.roundTripTime, event.duration)
+          roundTripTime: this.monitor?.roundTripTime,
+          minRoundTripTime: this.monitor?.minRoundTripTime
         })
       );
 
@@ -467,15 +468,6 @@ export class Server extends TypedEventEmitter<ServerEvents> {
   }
 }
 
-function calculateRoundTripTime(oldRtt: number, duration: number): number {
-  if (oldRtt === -1) {
-    return duration;
-  }
-
-  const alpha = 0.2;
-  return alpha * duration + (1 - alpha) * oldRtt;
-}
-
 function markServerUnknown(server: Server, error?: MongoError) {
   // Load balancer servers can never be marked unknown.
   if (server.loadBalanced) {

src/sdam/server_description.ts

@@ -33,8 +33,10 @@ export interface ServerDescriptionOptions {
   /** An Error used for better reporting debugging */
   error?: MongoError;
 
-  /** The round trip time to ping this server (in ms) */
+  /** The average round trip time to ping this server (in ms) */
   roundTripTime?: number;
+  /** The minimum round trip time to ping this server over the past 10 samples(in ms) */
+  minRoundTripTime?: number;
 
   /** If the client is in load balancing mode. */
   loadBalanced?: boolean;
@@ -58,6 +60,7 @@ export class ServerDescription {
   minWireVersion: number;
   maxWireVersion: number;
   roundTripTime: number;
+  minRoundTripTime: number;
   lastUpdateTime: number;
   lastWriteDate: number;
   me: string | null;
@@ -98,6 +101,7 @@ export class ServerDescription {
     this.minWireVersion = hello?.minWireVersion ?? 0;
     this.maxWireVersion = hello?.maxWireVersion ?? 0;
     this.roundTripTime = options?.roundTripTime ?? -1;
+    this.minRoundTripTime = options?.minRoundTripTime ?? 0;
     this.lastUpdateTime = now();
     this.lastWriteDate = hello?.lastWrite?.lastWriteDate ?? 0;
     this.error = options.error ?? null;

src/sdam/server_selection.ts

@@ -223,9 +223,8 @@ function latencyWindowReducer(
   servers: ServerDescription[]
 ): ServerDescription[] {
   const low = servers.reduce(
-    (min: number, server: ServerDescription) =>
-      min === -1 ? server.roundTripTime : Math.min(server.roundTripTime, min),
-    -1
+    (min: number, server: ServerDescription) => Math.min(server.roundTripTime, min),
+    Infinity
   );
 
   const high = low + topologyDescription.localThresholdMS;