opusdev/vector-similarity-api
1
1import { Binary, type Document, Long, type Timestamp } from './bson';2import type { CommandOptions, Connection } from './cmap/connection';3import { ConnectionPoolMetrics } from './cmap/metrics';4import { type MongoDBResponse } from './cmap/wire_protocol/responses';5import { PINNED, UNPINNED } from './constants';6import type { AbstractCursor } from './cursor/abstract_cursor';7import {8 type AnyError,9 isRetryableWriteError,10 MongoAPIError,11 MongoCompatibilityError,12 MONGODB_ERROR_CODES,13 type MongoDriverError,14 MongoError,15 MongoErrorLabel,16 MongoExpiredSessionError,17 MongoInvalidArgumentError,18 MongoRuntimeError,19 MongoServerError,20 MongoTransactionError,21 MongoWriteConcernError22} from './error';23import type { MongoClient, MongoOptions } from './mongo_client';24import { TypedEventEmitter } from './mongo_types';25import { executeOperation } from './operations/execute_operation';26import { RunCommandOperation } from './operations/run_command';27import { ReadConcernLevel } from './read_concern';28import { ReadPreference } from './read_preference';29import { type AsyncDisposable, configureResourceManagement } from './resource_management';30import { _advanceClusterTime, type ClusterTime, TopologyType } from './sdam/common';31import { TimeoutContext } from './timeout';32import {33 isTransactionCommand,34 Transaction,35 type TransactionOptions,36 TxnState37} from './transactions';38import {39 ByteUtils,40 calculateDurationInMs,41 commandSupportsReadConcern,42 isPromiseLike,43 List,44 MongoDBNamespace,45 noop,46 now,47 squashError,48 uuidV449} from './utils';50import { WriteConcern, type WriteConcernOptions, type WriteConcernSettings } from './write_concern';51 52/** @public */53export interface ClientSessionOptions {54 /** Whether causal consistency should be enabled on this session */55 causalConsistency?: boolean;56 /** Whether all read operations should be read from the same snapshot for this session (NOTE: not compatible with `causalConsistency=true`) */57 snapshot?: boolean;58 /** The default TransactionOptions to use for transactions started on this session. */59 defaultTransactionOptions?: TransactionOptions;60 /**61 * @public62 * @experimental63 * An overriding timeoutMS value to use for a client-side timeout.64 * If not provided the session uses the timeoutMS specified on the MongoClient.65 */66 defaultTimeoutMS?: number;67 68 /** @internal */69 owner?: symbol | AbstractCursor;70 /** @internal */71 explicit?: boolean;72 /** @internal */73 initialClusterTime?: ClusterTime;74}75 76/** @public */77export type WithTransactionCallback<T = any> = (session: ClientSession) => Promise<T>;78 79/** @public */80export type ClientSessionEvents = {81 ended(session: ClientSession): void;82};83 84/** @public */85export interface EndSessionOptions {86 /**87 * An optional error which caused the call to end this session88 * @internal89 */90 error?: AnyError;91 force?: boolean;92 forceClear?: boolean;93 94 /** Specifies the time an operation will run until it throws a timeout error */95 timeoutMS?: number;96}97 98/**99 * A class representing a client session on the server100 *101 * NOTE: not meant to be instantiated directly.102 * @public103 */104export class ClientSession105 extends TypedEventEmitter<ClientSessionEvents>106 implements AsyncDisposable107{108 /** @internal */109 client: MongoClient;110 /** @internal */111 sessionPool: ServerSessionPool;112 hasEnded: boolean;113 clientOptions: MongoOptions;114 supports: { causalConsistency: boolean };115 clusterTime?: ClusterTime;116 operationTime?: Timestamp;117 explicit: boolean;118 /** @internal */119 owner?: symbol | AbstractCursor;120 defaultTransactionOptions: TransactionOptions;121 /** @deprecated - Will be made internal in the next major release */122 transaction: Transaction;123 /**124 * @internal125 * Keeps track of whether or not the current transaction has attempted to be committed. Is126 * initially undefined. Gets set to false when startTransaction is called. When commitTransaction is sent to server, if the commitTransaction succeeds, it is then set to undefined, otherwise, set to true127 */128 private commitAttempted?: boolean;129 public readonly snapshotEnabled: boolean;130 131 /** @internal */132 private _serverSession: ServerSession | null;133 /** @internal */134 public snapshotTime?: Timestamp;135 /** @internal */136 public pinnedConnection?: Connection;137 /** @internal */138 public txnNumberIncrement: number;139 /**140 * @experimental141 * Specifies the time an operation in a given `ClientSession` will run until it throws a timeout error142 */143 timeoutMS?: number;144 145 /** @internal */146 public timeoutContext: TimeoutContext | null = null;147 148 /**149 * Create a client session.150 * @internal151 * @param client - The current client152 * @param sessionPool - The server session pool (Internal Class)153 * @param options - Optional settings154 * @param clientOptions - Optional settings provided when creating a MongoClient155 */156 constructor(157 client: MongoClient,158 sessionPool: ServerSessionPool,159 options: ClientSessionOptions,160 clientOptions: MongoOptions161 ) {162 super();163 this.on('error', noop);164 165 if (client == null) {166 // TODO(NODE-3483)167 throw new MongoRuntimeError('ClientSession requires a MongoClient');168 }169 170 if (sessionPool == null || !(sessionPool instanceof ServerSessionPool)) {171 // TODO(NODE-3483)172 throw new MongoRuntimeError('ClientSession requires a ServerSessionPool');173 }174 175 options = options ?? {};176 177 this.snapshotEnabled = options.snapshot === true;178 if (options.causalConsistency === true && this.snapshotEnabled) {179 throw new MongoInvalidArgumentError(180 'Properties "causalConsistency" and "snapshot" are mutually exclusive'181 );182 }183 184 this.client = client;185 this.sessionPool = sessionPool;186 this.hasEnded = false;187 this.clientOptions = clientOptions;188 this.timeoutMS = options.defaultTimeoutMS ?? client.s.options?.timeoutMS;189 190 this.explicit = !!options.explicit;191 this._serverSession = this.explicit ? this.sessionPool.acquire() : null;192 this.txnNumberIncrement = 0;193 194 const defaultCausalConsistencyValue = this.explicit && options.snapshot !== true;195 this.supports = {196 // if we can enable causal consistency, do so by default197 causalConsistency: options.causalConsistency ?? defaultCausalConsistencyValue198 };199 200 this.clusterTime = options.initialClusterTime;201 202 this.operationTime = undefined;203 this.owner = options.owner;204 this.defaultTransactionOptions = { ...options.defaultTransactionOptions };205 this.transaction = new Transaction();206 }207 208 /** The server id associated with this session */209 get id(): ServerSessionId | undefined {210 return this.serverSession?.id;211 }212 213 get serverSession(): ServerSession {214 let serverSession = this._serverSession;215 if (serverSession == null) {216 if (this.explicit) {217 throw new MongoRuntimeError('Unexpected null serverSession for an explicit session');218 }219 if (this.hasEnded) {220 throw new MongoRuntimeError('Unexpected null serverSession for an ended implicit session');221 }222 serverSession = this.sessionPool.acquire();223 this._serverSession = serverSession;224 }225 return serverSession;226 }227 228 get loadBalanced(): boolean {229 return this.client.topology?.description.type === TopologyType.LoadBalanced;230 }231 232 /** @internal */233 pin(conn: Connection): void {234 if (this.pinnedConnection) {235 throw TypeError('Cannot pin multiple connections to the same session');236 }237 238 this.pinnedConnection = conn;239 conn.emit(240 PINNED,241 this.inTransaction() ? ConnectionPoolMetrics.TXN : ConnectionPoolMetrics.CURSOR242 );243 }244 245 /** @internal */246 unpin(options?: { force?: boolean; forceClear?: boolean; error?: AnyError }): void {247 if (this.loadBalanced) {248 return maybeClearPinnedConnection(this, options);249 }250 251 this.transaction.unpinServer();252 }253 254 get isPinned(): boolean {255 return this.loadBalanced ? !!this.pinnedConnection : this.transaction.isPinned;256 }257 258 /**259 * Frees any client-side resources held by the current session. If a session is in a transaction,260 * the transaction is aborted.261 *262 * Does not end the session on the server.263 *264 * @param options - Optional settings. Currently reserved for future use265 */266 async endSession(options?: EndSessionOptions): Promise<void> {267 try {268 if (this.inTransaction()) {269 await this.abortTransaction({ ...options, throwTimeout: true });270 }271 } catch (error) {272 // spec indicates that we should ignore all errors for `endSessions`273 if (error.name === 'MongoOperationTimeoutError') throw error;274 squashError(error);275 } finally {276 if (!this.hasEnded) {277 const serverSession = this.serverSession;278 if (serverSession != null) {279 // release the server session back to the pool280 this.sessionPool.release(serverSession);281 // Store a clone of the server session for reference (debugging)282 this._serverSession = new ServerSession(serverSession);283 }284 // mark the session as ended, and emit a signal285 this.hasEnded = true;286 this.emit('ended', this);287 }288 maybeClearPinnedConnection(this, { force: true, ...options });289 }290 }291 /**292 * @beta293 * @experimental294 * An alias for {@link ClientSession.endSession|ClientSession.endSession()}.295 */296 declare [Symbol.asyncDispose]: () => Promise<void>;297 /** @internal */298 async asyncDispose() {299 await this.endSession({ force: true });300 }301 302 /**303 * Advances the operationTime for a ClientSession.304 *305 * @param operationTime - the `BSON.Timestamp` of the operation type it is desired to advance to306 */307 advanceOperationTime(operationTime: Timestamp): void {308 if (this.operationTime == null) {309 this.operationTime = operationTime;310 return;311 }312 313 if (operationTime.greaterThan(this.operationTime)) {314 this.operationTime = operationTime;315 }316 }317 318 /**319 * Advances the clusterTime for a ClientSession to the provided clusterTime of another ClientSession320 *321 * @param clusterTime - the $clusterTime returned by the server from another session in the form of a document containing the `BSON.Timestamp` clusterTime and signature322 */323 advanceClusterTime(clusterTime: ClusterTime): void {324 if (!clusterTime || typeof clusterTime !== 'object') {325 throw new MongoInvalidArgumentError('input cluster time must be an object');326 }327 if (!clusterTime.clusterTime || clusterTime.clusterTime._bsontype !== 'Timestamp') {328 throw new MongoInvalidArgumentError(329 'input cluster time "clusterTime" property must be a valid BSON Timestamp'330 );331 }332 if (333 !clusterTime.signature ||334 clusterTime.signature.hash?._bsontype !== 'Binary' ||335 (typeof clusterTime.signature.keyId !== 'bigint' &&336 typeof clusterTime.signature.keyId !== 'number' &&337 clusterTime.signature.keyId?._bsontype !== 'Long') // apparently we decode the key to number?338 ) {339 throw new MongoInvalidArgumentError(340 'input cluster time must have a valid "signature" property with BSON Binary hash and BSON Long keyId'341 );342 }343 344 _advanceClusterTime(this, clusterTime);345 }346 347 /**348 * Used to determine if this session equals another349 *350 * @param session - The session to compare to351 */352 equals(session: ClientSession): boolean {353 if (!(session instanceof ClientSession)) {354 return false;355 }356 357 if (this.id == null || session.id == null) {358 return false;359 }360 361 return ByteUtils.equals(this.id.id.buffer, session.id.id.buffer);362 }363 364 /**365 * Increment the transaction number on the internal ServerSession366 *367 * @privateRemarks368 * This helper increments a value stored on the client session that will be369 * added to the serverSession's txnNumber upon applying it to a command.370 * This is because the serverSession is lazily acquired after a connection is obtained371 */372 incrementTransactionNumber(): void {373 this.txnNumberIncrement += 1;374 }375 376 /** @returns whether this session is currently in a transaction or not */377 inTransaction(): boolean {378 return this.transaction.isActive;379 }380 381 /**382 * Starts a new transaction with the given options.383 *384 * @remarks385 * **IMPORTANT**: Running operations in parallel is not supported during a transaction. The use of `Promise.all`,386 * `Promise.allSettled`, `Promise.race`, etc to parallelize operations inside a transaction is387 * undefined behaviour.388 *389 * @param options - Options for the transaction390 */391 startTransaction(options?: TransactionOptions): void {392 if (this.snapshotEnabled) {393 throw new MongoCompatibilityError('Transactions are not supported in snapshot sessions');394 }395 396 if (this.inTransaction()) {397 throw new MongoTransactionError('Transaction already in progress');398 }399 400 if (this.isPinned && this.transaction.isCommitted) {401 this.unpin();402 }403 404 this.commitAttempted = false;405 // increment txnNumber406 this.incrementTransactionNumber();407 // create transaction state408 this.transaction = new Transaction({409 readConcern:410 options?.readConcern ??411 this.defaultTransactionOptions.readConcern ??412 this.clientOptions?.readConcern,413 writeConcern:414 options?.writeConcern ??415 this.defaultTransactionOptions.writeConcern ??416 this.clientOptions?.writeConcern,417 readPreference:418 options?.readPreference ??419 this.defaultTransactionOptions.readPreference ??420 this.clientOptions?.readPreference,421 maxCommitTimeMS: options?.maxCommitTimeMS ?? this.defaultTransactionOptions.maxCommitTimeMS422 });423 424 this.transaction.transition(TxnState.STARTING_TRANSACTION);425 }426 427 /**428 * Commits the currently active transaction in this session.429 *430 * @param options - Optional options, can be used to override `defaultTimeoutMS`.431 */432 async commitTransaction(options?: { timeoutMS?: number }): Promise<void> {433 if (this.transaction.state === TxnState.NO_TRANSACTION) {434 throw new MongoTransactionError('No transaction started');435 }436 437 if (438 this.transaction.state === TxnState.STARTING_TRANSACTION ||439 this.transaction.state === TxnState.TRANSACTION_COMMITTED_EMPTY440 ) {441 // the transaction was never started, we can safely exit here442 this.transaction.transition(TxnState.TRANSACTION_COMMITTED_EMPTY);443 return;444 }445 446 if (this.transaction.state === TxnState.TRANSACTION_ABORTED) {447 throw new MongoTransactionError(448 'Cannot call commitTransaction after calling abortTransaction'449 );450 }451 452 const command: {453 commitTransaction: 1;454 writeConcern?: WriteConcernSettings;455 recoveryToken?: Document;456 maxTimeMS?: number;457 } = { commitTransaction: 1 };458 459 const timeoutMS =460 typeof options?.timeoutMS === 'number'461 ? options.timeoutMS462 : typeof this.timeoutMS === 'number'463 ? this.timeoutMS464 : null;465 466 const wc = this.transaction.options.writeConcern ?? this.clientOptions?.writeConcern;467 if (wc != null) {468 if (timeoutMS == null && this.timeoutContext == null) {469 WriteConcern.apply(command, { wtimeoutMS: 10000, w: 'majority', ...wc });470 } else {471 const wcKeys = Object.keys(wc);472 if (wcKeys.length > 2 || (!wcKeys.includes('wtimeoutMS') && !wcKeys.includes('wTimeoutMS')))473 // if the write concern was specified with wTimeoutMS, then we set both wtimeoutMS and wTimeoutMS, guaranteeing at least two keys, so if we have more than two keys, then we can automatically assume that we should add the write concern to the command. If it has 2 or fewer keys, we need to check that those keys aren't the wtimeoutMS or wTimeoutMS options before we add the write concern to the command474 WriteConcern.apply(command, { ...wc, wtimeoutMS: undefined });475 }476 }477 478 if (this.transaction.state === TxnState.TRANSACTION_COMMITTED || this.commitAttempted) {479 if (timeoutMS == null && this.timeoutContext == null) {480 WriteConcern.apply(command, { wtimeoutMS: 10000, ...wc, w: 'majority' });481 } else {482 WriteConcern.apply(command, { w: 'majority', ...wc, wtimeoutMS: undefined });483 }484 }485 486 if (typeof this.transaction.options.maxTimeMS === 'number') {487 command.maxTimeMS = this.transaction.options.maxTimeMS;488 }489 490 if (this.transaction.recoveryToken) {491 command.recoveryToken = this.transaction.recoveryToken;492 }493 494 const operation = new RunCommandOperation(new MongoDBNamespace('admin'), command, {495 session: this,496 readPreference: ReadPreference.primary,497 bypassPinningCheck: true498 });499 500 const timeoutContext =501 this.timeoutContext ??502 (typeof timeoutMS === 'number'503 ? TimeoutContext.create({504 serverSelectionTimeoutMS: this.clientOptions.serverSelectionTimeoutMS,505 socketTimeoutMS: this.clientOptions.socketTimeoutMS,506 timeoutMS507 })508 : null);509 510 try {511 await executeOperation(this.client, operation, timeoutContext);512 this.commitAttempted = undefined;513 return;514 } catch (firstCommitError) {515 this.commitAttempted = true;516 if (firstCommitError instanceof MongoError && isRetryableWriteError(firstCommitError)) {517 // SPEC-1185: apply majority write concern when retrying commitTransaction518 WriteConcern.apply(command, { wtimeoutMS: 10000, ...wc, w: 'majority' });519 // per txns spec, must unpin session in this case520 this.unpin({ force: true });521 522 try {523 await executeOperation(524 this.client,525 new RunCommandOperation(new MongoDBNamespace('admin'), command, {526 session: this,527 readPreference: ReadPreference.primary,528 bypassPinningCheck: true529 }),530 timeoutContext531 );532 return;533 } catch (retryCommitError) {534 // If the retry failed, we process that error instead of the original535 if (shouldAddUnknownTransactionCommitResultLabel(retryCommitError)) {536 retryCommitError.addErrorLabel(MongoErrorLabel.UnknownTransactionCommitResult);537 }538 539 if (shouldUnpinAfterCommitError(retryCommitError)) {540 this.unpin({ error: retryCommitError });541 }542 543 throw retryCommitError;544 }545 }546 547 if (shouldAddUnknownTransactionCommitResultLabel(firstCommitError)) {548 firstCommitError.addErrorLabel(MongoErrorLabel.UnknownTransactionCommitResult);549 }550 551 if (shouldUnpinAfterCommitError(firstCommitError)) {552 this.unpin({ error: firstCommitError });553 }554 555 throw firstCommitError;556 } finally {557 this.transaction.transition(TxnState.TRANSACTION_COMMITTED);558 }559 }560 561 /**562 * Aborts the currently active transaction in this session.563 *564 * @param options - Optional options, can be used to override `defaultTimeoutMS`.565 */566 async abortTransaction(options?: { timeoutMS?: number }): Promise<void>;567 /** @internal */568 async abortTransaction(options?: { timeoutMS?: number; throwTimeout?: true }): Promise<void>;569 async abortTransaction(options?: { timeoutMS?: number; throwTimeout?: true }): Promise<void> {570 if (this.transaction.state === TxnState.NO_TRANSACTION) {571 throw new MongoTransactionError('No transaction started');572 }573 574 if (this.transaction.state === TxnState.STARTING_TRANSACTION) {575 // the transaction was never started, we can safely exit here576 this.transaction.transition(TxnState.TRANSACTION_ABORTED);577 return;578 }579 580 if (this.transaction.state === TxnState.TRANSACTION_ABORTED) {581 throw new MongoTransactionError('Cannot call abortTransaction twice');582 }583 584 if (585 this.transaction.state === TxnState.TRANSACTION_COMMITTED ||586 this.transaction.state === TxnState.TRANSACTION_COMMITTED_EMPTY587 ) {588 throw new MongoTransactionError(589 'Cannot call abortTransaction after calling commitTransaction'590 );591 }592 593 const command: {594 abortTransaction: 1;595 writeConcern?: WriteConcernOptions;596 recoveryToken?: Document;597 } = { abortTransaction: 1 };598 599 const timeoutMS =600 typeof options?.timeoutMS === 'number'601 ? options.timeoutMS602 : this.timeoutContext?.csotEnabled()603 ? this.timeoutContext.timeoutMS // refresh timeoutMS for abort operation604 : typeof this.timeoutMS === 'number'605 ? this.timeoutMS606 : null;607 608 const timeoutContext =609 timeoutMS != null610 ? TimeoutContext.create({611 timeoutMS,612 serverSelectionTimeoutMS: this.clientOptions.serverSelectionTimeoutMS,613 socketTimeoutMS: this.clientOptions.socketTimeoutMS614 })615 : null;616 617 const wc = this.transaction.options.writeConcern ?? this.clientOptions?.writeConcern;618 if (wc != null && timeoutMS == null) {619 WriteConcern.apply(command, { wtimeoutMS: 10000, w: 'majority', ...wc });620 }621 622 if (this.transaction.recoveryToken) {623 command.recoveryToken = this.transaction.recoveryToken;624 }625 626 const operation = new RunCommandOperation(new MongoDBNamespace('admin'), command, {627 session: this,628 readPreference: ReadPreference.primary,629 bypassPinningCheck: true630 });631 632 try {633 await executeOperation(this.client, operation, timeoutContext);634 this.unpin();635 return;636 } catch (firstAbortError) {637 this.unpin();638 639 if (firstAbortError.name === 'MongoRuntimeError') throw firstAbortError;640 if (options?.throwTimeout && firstAbortError.name === 'MongoOperationTimeoutError') {641 throw firstAbortError;642 }643 644 if (firstAbortError instanceof MongoError && isRetryableWriteError(firstAbortError)) {645 try {646 await executeOperation(this.client, operation, timeoutContext);647 return;648 } catch (secondAbortError) {649 if (secondAbortError.name === 'MongoRuntimeError') throw secondAbortError;650 if (options?.throwTimeout && secondAbortError.name === 'MongoOperationTimeoutError') {651 throw secondAbortError;652 }653 // we do not retry the retry654 }655 }656 657 // The spec indicates that if the operation times out or fails with a non-retryable error, we should ignore all errors on `abortTransaction`658 } finally {659 this.transaction.transition(TxnState.TRANSACTION_ABORTED);660 if (this.loadBalanced) {661 maybeClearPinnedConnection(this, { force: false });662 }663 }664 }665 666 /**667 * This is here to ensure that ClientSession is never serialized to BSON.668 */669 toBSON(): never {670 throw new MongoRuntimeError('ClientSession cannot be serialized to BSON.');671 }672 673 /**674 * Starts a transaction and runs a provided function, ensuring the commitTransaction is always attempted when all operations run in the function have completed.675 *676 * **IMPORTANT:** This method requires the function passed in to return a Promise. That promise must be made by `await`-ing all operations in such a way that rejections are propagated to the returned promise.677 *678 * **IMPORTANT:** Running operations in parallel is not supported during a transaction. The use of `Promise.all`,679 * `Promise.allSettled`, `Promise.race`, etc to parallelize operations inside a transaction is680 * undefined behaviour.681 *682 * **IMPORTANT:** When running an operation inside a `withTransaction` callback, if it is not683 * provided the explicit session in its options, it will not be part of the transaction and it will not respect timeoutMS.684 *685 *686 * @remarks687 * - If all operations successfully complete and the `commitTransaction` operation is successful, then the provided function will return the result of the provided function.688 * - If the transaction is unable to complete or an error is thrown from within the provided function, then the provided function will throw an error.689 * - If the transaction is manually aborted within the provided function it will not throw.690 * - If the driver needs to attempt to retry the operations, the provided function may be called multiple times.691 *692 * Checkout a descriptive example here:693 * @see https://www.mongodb.com/blog/post/quick-start-nodejs--mongodb--how-to-implement-transactions694 *695 * If a command inside withTransaction fails:696 * - It may cause the transaction on the server to be aborted.697 * - This situation is normally handled transparently by the driver.698 * - However, if the application catches such an error and does not rethrow it, the driver will not be able to determine whether the transaction was aborted or not.699 * - The driver will then retry the transaction indefinitely.700 *701 * To avoid this situation, the application must not silently handle errors within the provided function.702 * If the application needs to handle errors within, it must await all operations such that if an operation is rejected it becomes the rejection of the callback function passed into withTransaction.703 *704 * @param fn - callback to run within a transaction705 * @param options - optional settings for the transaction706 * @returns A raw command response or undefined707 */708 async withTransaction<T = any>(709 fn: WithTransactionCallback<T>,710 options?: TransactionOptions & {711 /**712 * Configures a timeoutMS expiry for the entire withTransactionCallback.713 *714 * @remarks715 * - The remaining timeout will not be applied to callback operations that do not use the ClientSession.716 * - Overriding timeoutMS for operations executed using the explicit session inside the provided callback will result in a client-side error.717 */718 timeoutMS?: number;719 }720 ): Promise<T> {721 const MAX_TIMEOUT = 120000;722 723 const timeoutMS = options?.timeoutMS ?? this.timeoutMS ?? null;724 this.timeoutContext =725 timeoutMS != null726 ? TimeoutContext.create({727 timeoutMS,728 serverSelectionTimeoutMS: this.clientOptions.serverSelectionTimeoutMS,729 socketTimeoutMS: this.clientOptions.socketTimeoutMS730 })731 : null;732 733 const startTime = this.timeoutContext?.csotEnabled() ? this.timeoutContext.start : now();734 735 let committed = false;736 let result: any;737 738 try {739 while (!committed) {740 this.startTransaction(options); // may throw on error741 742 try {743 const promise = fn(this);744 if (!isPromiseLike(promise)) {745 throw new MongoInvalidArgumentError(746 'Function provided to `withTransaction` must return a Promise'747 );748 }749 750 result = await promise;751 752 if (753 this.transaction.state === TxnState.NO_TRANSACTION ||754 this.transaction.state === TxnState.TRANSACTION_COMMITTED ||755 this.transaction.state === TxnState.TRANSACTION_ABORTED756 ) {757 // Assume callback intentionally ended the transaction758 return result;759 }760 } catch (fnError) {761 if (!(fnError instanceof MongoError) || fnError instanceof MongoInvalidArgumentError) {762 await this.abortTransaction();763 throw fnError;764 }765 766 if (767 this.transaction.state === TxnState.STARTING_TRANSACTION ||768 this.transaction.state === TxnState.TRANSACTION_IN_PROGRESS769 ) {770 await this.abortTransaction();771 }772 773 if (774 fnError.hasErrorLabel(MongoErrorLabel.TransientTransactionError) &&775 (this.timeoutContext != null || now() - startTime < MAX_TIMEOUT)776 ) {777 continue;778 }779 780 throw fnError;781 }782 783 while (!committed) {784 try {785 /*786 * We will rely on ClientSession.commitTransaction() to787 * apply a majority write concern if commitTransaction is788 * being retried (see: DRIVERS-601)789 */790 await this.commitTransaction();791 committed = true;792 } catch (commitError) {793 /*794 * Note: a maxTimeMS error will have the MaxTimeMSExpired795 * code (50) and can be reported as a top-level error or796 * inside writeConcernError, ex.797 * { ok:0, code: 50, codeName: 'MaxTimeMSExpired' }798 * { ok:1, writeConcernError: { code: 50, codeName: 'MaxTimeMSExpired' } }799 */800 if (801 !isMaxTimeMSExpiredError(commitError) &&802 commitError.hasErrorLabel(MongoErrorLabel.UnknownTransactionCommitResult) &&803 (this.timeoutContext != null || now() - startTime < MAX_TIMEOUT)804 ) {805 continue;806 }807 808 if (809 commitError.hasErrorLabel(MongoErrorLabel.TransientTransactionError) &&810 (this.timeoutContext != null || now() - startTime < MAX_TIMEOUT)811 ) {812 break;813 }814 815 throw commitError;816 }817 }818 }819 return result;820 } finally {821 this.timeoutContext = null;822 }823 }824}825 826configureResourceManagement(ClientSession.prototype);827 828const NON_DETERMINISTIC_WRITE_CONCERN_ERRORS = new Set([829 'CannotSatisfyWriteConcern',830 'UnknownReplWriteConcern',831 'UnsatisfiableWriteConcern'832]);833 834function shouldUnpinAfterCommitError(commitError: Error) {835 if (commitError instanceof MongoError) {836 if (837 isRetryableWriteError(commitError) ||838 commitError instanceof MongoWriteConcernError ||839 isMaxTimeMSExpiredError(commitError)840 ) {841 if (isUnknownTransactionCommitResult(commitError)) {842 // per txns spec, must unpin session in this case843 return true;844 }845 } else if (commitError.hasErrorLabel(MongoErrorLabel.TransientTransactionError)) {846 return true;847 }848 }849 return false;850}851 852function shouldAddUnknownTransactionCommitResultLabel(commitError: MongoError) {853 let ok = isRetryableWriteError(commitError);854 ok ||= commitError instanceof MongoWriteConcernError;855 ok ||= isMaxTimeMSExpiredError(commitError);856 ok &&= isUnknownTransactionCommitResult(commitError);857 return ok;858}859 860function isUnknownTransactionCommitResult(err: MongoError): err is MongoError {861 const isNonDeterministicWriteConcernError =862 err instanceof MongoServerError &&863 err.codeName &&864 NON_DETERMINISTIC_WRITE_CONCERN_ERRORS.has(err.codeName);865 866 return (867 isMaxTimeMSExpiredError(err) ||868 (!isNonDeterministicWriteConcernError &&869 err.code !== MONGODB_ERROR_CODES.UnsatisfiableWriteConcern &&870 err.code !== MONGODB_ERROR_CODES.UnknownReplWriteConcern)871 );872}873 874export function maybeClearPinnedConnection(875 session: ClientSession,876 options?: EndSessionOptions877): void {878 // unpin a connection if it has been pinned879 const conn = session.pinnedConnection;880 const error = options?.error;881 882 if (883 session.inTransaction() &&884 error &&885 error instanceof MongoError &&886 error.hasErrorLabel(MongoErrorLabel.TransientTransactionError)887 ) {888 return;889 }890 891 const topology = session.client.topology;892 // NOTE: the spec talks about what to do on a network error only, but the tests seem to893 // to validate that we don't unpin on _all_ errors?894 if (conn && topology != null) {895 const servers = Array.from(topology.s.servers.values());896 const loadBalancer = servers[0];897 898 if (options?.error == null || options?.force) {899 loadBalancer.pool.checkIn(conn);900 session.pinnedConnection = undefined;901 conn.emit(902 UNPINNED,903 session.transaction.state !== TxnState.NO_TRANSACTION904 ? ConnectionPoolMetrics.TXN905 : ConnectionPoolMetrics.CURSOR906 );907 908 if (options?.forceClear) {909 loadBalancer.pool.clear({ serviceId: conn.serviceId });910 }911 }912 }913}914 915function isMaxTimeMSExpiredError(err: MongoError): boolean {916 if (err == null || !(err instanceof MongoServerError)) {917 return false;918 }919 920 return (921 err.code === MONGODB_ERROR_CODES.MaxTimeMSExpired ||922 err.writeConcernError?.code === MONGODB_ERROR_CODES.MaxTimeMSExpired923 );924}925 926/** @public */927export type ServerSessionId = { id: Binary };928 929/**930 * Reflects the existence of a session on the server. Can be reused by the session pool.931 * WARNING: not meant to be instantiated directly. For internal use only.932 * @public933 */934export class ServerSession {935 id: ServerSessionId;936 lastUse: number;937 txnNumber: number;938 isDirty: boolean;939 940 /** @internal */941 constructor(cloned?: ServerSession | null) {942 if (cloned != null) {943 const idBytes = Buffer.allocUnsafe(16);944 idBytes.set(cloned.id.id.buffer);945 this.id = { id: new Binary(idBytes, cloned.id.id.sub_type) };946 this.lastUse = cloned.lastUse;947 this.txnNumber = cloned.txnNumber;948 this.isDirty = cloned.isDirty;949 return;950 }951 this.id = { id: new Binary(uuidV4(), Binary.SUBTYPE_UUID) };952 this.lastUse = now();953 this.txnNumber = 0;954 this.isDirty = false;955 }956 957 /**958 * Determines if the server session has timed out.959 *960 * @param sessionTimeoutMinutes - The server's "logicalSessionTimeoutMinutes"961 */962 hasTimedOut(sessionTimeoutMinutes: number): boolean {963 // Take the difference of the lastUse timestamp and now, which will result in a value in964 // milliseconds, and then convert milliseconds to minutes to compare to `sessionTimeoutMinutes`965 const idleTimeMinutes = Math.round(966 ((calculateDurationInMs(this.lastUse) % 86400000) % 3600000) / 60000967 );968 969 return idleTimeMinutes > sessionTimeoutMinutes - 1;970 }971}972 973/**974 * Maintains a pool of Server Sessions.975 * For internal use only976 * @internal977 */978export class ServerSessionPool {979 client: MongoClient;980 sessions: List<ServerSession>;981 982 constructor(client: MongoClient) {983 if (client == null) {984 throw new MongoRuntimeError('ServerSessionPool requires a MongoClient');985 }986 987 this.client = client;988 this.sessions = new List<ServerSession>();989 }990 991 /**992 * Acquire a Server Session from the pool.993 * Iterates through each session in the pool, removing any stale sessions994 * along the way. The first non-stale session found is removed from the995 * pool and returned. If no non-stale session is found, a new ServerSession is created.996 */997 acquire(): ServerSession {998 const sessionTimeoutMinutes = this.client.topology?.logicalSessionTimeoutMinutes ?? 10;999 1000 let session: ServerSession | null = null;1001 1002 // Try to obtain from session pool1003 while (this.sessions.length > 0) {1004 const potentialSession = this.sessions.shift();1005 if (1006 potentialSession != null &&1007 (!!this.client.topology?.loadBalanced ||1008 !potentialSession.hasTimedOut(sessionTimeoutMinutes))1009 ) {1010 session = potentialSession;1011 break;1012 }1013 }1014 1015 // If nothing valid came from the pool make a new one1016 if (session == null) {1017 session = new ServerSession();1018 }1019 1020 return session;1021 }1022 1023 /**1024 * Release a session to the session pool1025 * Adds the session back to the session pool if the session has not timed out yet.1026 * This method also removes any stale sessions from the pool.1027 *1028 * @param session - The session to release to the pool1029 */1030 release(session: ServerSession): void {1031 const sessionTimeoutMinutes = this.client.topology?.logicalSessionTimeoutMinutes ?? 10;1032 1033 if (this.client.topology?.loadBalanced && !sessionTimeoutMinutes) {1034 this.sessions.unshift(session);1035 }1036 1037 if (!sessionTimeoutMinutes) {1038 return;1039 }1040 1041 this.sessions.prune(session => session.hasTimedOut(sessionTimeoutMinutes));1042 1043 if (!session.hasTimedOut(sessionTimeoutMinutes)) {1044 if (session.isDirty) {1045 return;1046 }1047 1048 // otherwise, readd this session to the session pool1049 this.sessions.unshift(session);1050 }1051 }1052}1053 1054/**1055 * Optionally decorate a command with sessions specific keys1056 *1057 * @param session - the session tracking transaction state1058 * @param command - the command to decorate1059 * @param options - Optional settings passed to calling operation1060 *1061 * @internal1062 */1063export function applySession(1064 session: ClientSession,1065 command: Document,1066 options: CommandOptions1067): MongoDriverError | undefined {1068 if (session.hasEnded) {1069 return new MongoExpiredSessionError();1070 }1071 1072 // May acquire serverSession here1073 const serverSession = session.serverSession;1074 if (serverSession == null) {1075 return new MongoRuntimeError('Unable to acquire server session');1076 }1077 1078 if (options.writeConcern?.w === 0) {1079 if (session && session.explicit) {1080 // Error if user provided an explicit session to an unacknowledged write (SPEC-1019)1081 return new MongoAPIError('Cannot have explicit session with unacknowledged writes');1082 }1083 return;1084 }1085 1086 // mark the last use of this session, and apply the `lsid`1087 serverSession.lastUse = now();1088 command.lsid = serverSession.id;1089 1090 const inTxnOrTxnCommand = session.inTransaction() || isTransactionCommand(command);1091 const isRetryableWrite = !!options.willRetryWrite;1092 1093 if (isRetryableWrite || inTxnOrTxnCommand) {1094 serverSession.txnNumber += session.txnNumberIncrement;1095 session.txnNumberIncrement = 0;1096 // TODO(NODE-2674): Preserve int64 sent from MongoDB1097 command.txnNumber = Long.fromNumber(serverSession.txnNumber);1098 }1099 1100 if (!inTxnOrTxnCommand) {1101 if (session.transaction.state !== TxnState.NO_TRANSACTION) {1102 session.transaction.transition(TxnState.NO_TRANSACTION);1103 }1104 1105 if (1106 session.supports.causalConsistency &&1107 session.operationTime &&1108 commandSupportsReadConcern(command)1109 ) {1110 command.readConcern = command.readConcern || {};1111 Object.assign(command.readConcern, { afterClusterTime: session.operationTime });1112 } else if (session.snapshotEnabled) {1113 command.readConcern = command.readConcern || { level: ReadConcernLevel.snapshot };1114 if (session.snapshotTime != null) {1115 Object.assign(command.readConcern, { atClusterTime: session.snapshotTime });1116 }1117 }1118 1119 return;1120 }1121 1122 // now attempt to apply transaction-specific sessions data1123 1124 // `autocommit` must always be false to differentiate from retryable writes1125 command.autocommit = false;1126 1127 if (session.transaction.state === TxnState.STARTING_TRANSACTION) {1128 session.transaction.transition(TxnState.TRANSACTION_IN_PROGRESS);1129 command.startTransaction = true;1130 1131 const readConcern =1132 session.transaction.options.readConcern || session?.clientOptions?.readConcern;1133 if (readConcern) {1134 command.readConcern = readConcern;1135 }1136 1137 if (session.supports.causalConsistency && session.operationTime) {1138 command.readConcern = command.readConcern || {};1139 Object.assign(command.readConcern, { afterClusterTime: session.operationTime });1140 }1141 }1142 return;1143}1144 1145export function updateSessionFromResponse(session: ClientSession, document: MongoDBResponse): void {1146 if (document.$clusterTime) {1147 _advanceClusterTime(session, document.$clusterTime);1148 }1149 1150 if (document.operationTime && session && session.supports.causalConsistency) {1151 session.advanceOperationTime(document.operationTime);1152 }1153 1154 if (document.recoveryToken && session && session.inTransaction()) {1155 session.transaction._recoveryToken = document.recoveryToken;1156 }1157 1158 if (session?.snapshotEnabled && session.snapshotTime == null) {1159 // find and aggregate commands return atClusterTime on the cursor1160 // distinct includes it in the response body1161 const atClusterTime = document.atClusterTime;1162 if (atClusterTime) {1163 session.snapshotTime = atClusterTime;1164 }1165 }1166}1167