feat(NODE-2843): implement sessions advanceClusterTime method (#2920)

src/sdam/common.ts

@@ -66,16 +66,16 @@ export interface ClusterTime {
   };
 }
 
-/** Shared function to determine clusterTime for a given topology */
-export function resolveClusterTime(
-  topology: Topology | ClientSession,
+/** Shared function to determine clusterTime for a given topology or session */
+export function _advanceClusterTime(
+  entity: Topology | ClientSession,
   $clusterTime: ClusterTime
 ): void {
-  if (topology.clusterTime == null) {
-    topology.clusterTime = $clusterTime;
+  if (entity.clusterTime == null) {
+    entity.clusterTime = $clusterTime;
   } else {
-    if ($clusterTime.clusterTime.greaterThan(topology.clusterTime.clusterTime)) {
-      topology.clusterTime = $clusterTime;
+    if ($clusterTime.clusterTime.greaterThan(entity.clusterTime.clusterTime)) {
+      entity.clusterTime = $clusterTime;
     }
   }
 }

src/sdam/topology.ts

@@ -28,7 +28,7 @@ import {
   ServerType,
   ClusterTime,
   TimerQueue,
-  resolveClusterTime,
+  _advanceClusterTime,
   drainTimerQueue,
   clearAndRemoveTimerFrom,
   STATE_CLOSED,
@@ -681,7 +681,7 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
     // value of the clusterTime embedded field."
     const clusterTime = serverDescription.$clusterTime;
     if (clusterTime) {
-      resolveClusterTime(this, clusterTime);
+      _advanceClusterTime(this, clusterTime);
     }
 
     // If we already know all the information contained in this updated description, then

src/sessions.ts

@@ -2,7 +2,7 @@ import { PromiseProvider } from './promise_provider';
 import { Binary, Long, Timestamp, Document } from './bson';
 import { ReadPreference } from './read_preference';
 import { isTransactionCommand, TxnState, Transaction, TransactionOptions } from './transactions';
-import { resolveClusterTime, ClusterTime } from './sdam/common';
+import { _advanceClusterTime, ClusterTime } from './sdam/common';
 import { isSharded } from './cmap/wire_protocol/shared';
 import {
   MongoError,
@@ -249,6 +249,34 @@ export class ClientSession extends TypedEventEmitter<ClientSessionEvents> {
     }
   }
 
+  /**
+   * Advances the clusterTime for a ClientSession to the provided clusterTime of another ClientSession
+   *
+   * @param clusterTime - the $clusterTime returned by the server from another session in the form of a document containing the `BSON.Timestamp` clusterTime and signature
+   */
+  advanceClusterTime(clusterTime: ClusterTime): void {
+    if (!clusterTime || typeof clusterTime !== 'object') {
+      throw new MongoInvalidArgumentError('input cluster time must be an object');
+    }
+    if (!clusterTime.clusterTime || clusterTime.clusterTime._bsontype !== 'Timestamp') {
+      throw new MongoInvalidArgumentError(
+        'input cluster time "clusterTime" property must be a valid BSON Timestamp'
+      );
+    }
+    if (
+      !clusterTime.signature ||
+      clusterTime.signature.hash?._bsontype !== 'Binary' ||
+      (typeof clusterTime.signature.keyId !== 'number' &&
+        clusterTime.signature.keyId?._bsontype !== 'Long') // apparently we decode the key to number?
+    ) {
+      throw new MongoInvalidArgumentError(
+        'input cluster time must have a valid "signature" property with BSON Binary hash and BSON Long keyId'
+      );
+    }
+
+    _advanceClusterTime(this, clusterTime);
+  }
+
   /**
    * Used to determine if this session equals another
    *
@@ -886,7 +914,7 @@ export function applySession(
 
 export function updateSessionFromResponse(session: ClientSession, document: Document): void {
   if (document.$clusterTime) {
-    resolveClusterTime(session, document.$clusterTime);
+    _advanceClusterTime(session, document.$clusterTime);
   }
 
   if (document.operationTime && session && session.supports.causalConsistency) {