Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 | 564x 564x 564x 564x 564x 564x 564x 564x 564x 13341419x 13341419x 13337908x 13337908x 7436003x 7436003x 5901905x 5901905x 174x 5901731x 458x 13337276x 13337276x 13337276x 13337276x 2296x 13334980x 440x 13334980x 13334980x 13334980x 13334950x 7436003x 355414x 355414x 2824x 352590x 352590x 352590x 352132x 352132x 352590x 13334980x 611473x 12723507x 5385x 12718122x 13334980x 13333312x 13333312x 13333312x 13333312x 13333312x 13333312x 13333312x 745013x 745013x 13333312x 13333312x 13467649x 134337x 134x 134203x 314x 133889x 16622x 117267x 29902x 87365x 720x 87365x 87365x 13420677x 13420677x 420x 13420677x 410489x 380325x 174x 380151x 380151x 380151x 245814x | import {
isRetryableReadError,
isRetryableWriteError,
MongoCompatibilityError,
MONGODB_ERROR_CODES,
MongoError,
MongoErrorLabel,
MongoExpiredSessionError,
MongoInvalidArgumentError,
MongoNetworkError,
MongoNotConnectedError,
MongoRuntimeError,
MongoServerError,
MongoTransactionError,
MongoUnexpectedServerResponseError
} from '../error';
import type { MongoClient } from '../mongo_client';
import { ReadPreference } from '../read_preference';
import type { ServerDescription } from '../sdam/server_description';
import {
sameServerSelector,
secondaryWritableServerSelector,
type ServerSelector
} from '../sdam/server_selection';
import type { Topology } from '../sdam/topology';
import type { ClientSession } from '../sessions';
import { TimeoutContext } from '../timeout';
import { abortable, supportsRetryableWrites } from '../utils';
import { AbstractOperation, Aspect } from './operation';
const MMAPv1_RETRY_WRITES_ERROR_CODE = MONGODB_ERROR_CODES.IllegalOperation;
const MMAPv1_RETRY_WRITES_ERROR_MESSAGE =
'This MongoDB deployment does not support retryable writes. Please add retryWrites=false to your connection string.';
type ResultTypeFromOperation<TOperation> =
TOperation extends AbstractOperation<infer K> ? K : never;
/**
* Executes the given operation with provided arguments.
* @internal
*
* @remarks
* Allows for a single point of entry to provide features such as implicit sessions, which
* are required by the Driver Sessions specification in the event that a ClientSession is
* not provided.
*
* The expectation is that this function:
* - Connects the MongoClient if it has not already been connected, see {@link autoConnect}
* - Creates a session if none is provided and cleans up the session it creates
* - Tries an operation and retries under certain conditions, see {@link tryOperation}
*
* @typeParam T - The operation's type
* @typeParam TResult - The type of the operation's result, calculated from T
*
* @param client - The MongoClient to execute this operation with
* @param operation - The operation to execute
*/
export async function executeOperation<
T extends AbstractOperation<TResult>,
TResult = ResultTypeFromOperation<T>
>(client: MongoClient, operation: T, timeoutContext?: TimeoutContext | null): Promise<TResult> {
Iif (!(operation instanceof AbstractOperation)) {
// TODO(NODE-3483): Extend MongoRuntimeError
throw new MongoRuntimeError('This method requires a valid operation instance');
}
const topology =
client.topology == null
? await abortable(autoConnect(client), operation.options)
: client.topology;
// The driver sessions spec mandates that we implicitly create sessions for operations
// that are not explicitly provided with a session.
let session = operation.session;
let owner: symbol | undefined;
if (session == null) {
owner = Symbol();
session = client.startSession({ owner, explicit: false });
} else Iif (session.hasEnded) {
throw new MongoExpiredSessionError('Use of expired sessions is not permitted');
} else if (session.snapshotEnabled && !topology.capabilities.supportsSnapshotReads) {
throw new MongoCompatibilityError('Snapshot reads require MongoDB 5.0 or later');
} else if (session.client !== client) {
throw new MongoInvalidArgumentError('ClientSession must be from the same MongoClient');
}
const readPreference = operation.readPreference ?? ReadPreference.primary;
const inTransaction = !!session?.inTransaction();
const hasReadAspect = operation.hasAspect(Aspect.READ_OPERATION);
if (
inTransaction &&
!readPreference.equals(ReadPreference.primary) &&
(hasReadAspect || operation.commandName === 'runCommand')
) {
throw new MongoTransactionError(
`Read preference in a transaction must be primary, not: ${readPreference.mode}`
);
}
if (session?.isPinned && session.transaction.isCommitted && !operation.bypassPinningCheck) {
session.unpin();
}
timeoutContext ??= TimeoutContext.create({
session,
serverSelectionTimeoutMS: client.s.options.serverSelectionTimeoutMS,
waitQueueTimeoutMS: client.s.options.waitQueueTimeoutMS,
timeoutMS: operation.options.timeoutMS
});
try {
return await tryOperation(operation, {
topology,
timeoutContext,
session,
readPreference
});
} finally {
if (session?.owner != null && session.owner === owner) {
await session.endSession();
}
}
}
/**
* Connects a client if it has not yet been connected
* @internal
*/
async function autoConnect(client: MongoClient): Promise<Topology> {
Eif (client.topology == null) {
if (client.s.hasBeenClosed) {
throw new MongoNotConnectedError('Client must be connected before running operations');
}
client.s.options.__skipPingOnConnect = true;
try {
await client.connect();
Iif (client.topology == null) {
throw new MongoRuntimeError(
'client.connect did not create a topology but also did not throw'
);
}
return client.topology;
} finally {
delete client.s.options.__skipPingOnConnect;
}
}
return client.topology;
}
/** @internal */
type RetryOptions = {
session: ClientSession | undefined;
readPreference: ReadPreference;
topology: Topology;
timeoutContext: TimeoutContext;
};
/**
* Executes an operation and retries as appropriate
* @internal
*
* @remarks
* Implements behaviour described in [Retryable Reads](https://github.com/mongodb/specifications/blob/master/source/retryable-reads/retryable-reads.md) and [Retryable
* Writes](https://github.com/mongodb/specifications/blob/master/source/retryable-writes/retryable-writes.md) specification
*
* This function:
* - performs initial server selection
* - attempts to execute an operation
* - retries the operation if it meets the criteria for a retryable read or a retryable write
*
* @typeParam T - The operation's type
* @typeParam TResult - The type of the operation's result, calculated from T
*
* @param operation - The operation to execute
* */
async function tryOperation<
T extends AbstractOperation<TResult>,
TResult = ResultTypeFromOperation<T>
>(
operation: T,
{ topology, timeoutContext, session, readPreference }: RetryOptions
): Promise<TResult> {
let selector: ReadPreference | ServerSelector;
if (operation.hasAspect(Aspect.MUST_SELECT_SAME_SERVER)) {
// GetMore and KillCursor operations must always select the same server, but run through
// server selection to potentially force monitor checks if the server is
// in an unknown state.
selector = sameServerSelector(operation.server?.description);
} else if (operation.trySecondaryWrite) {
// If operation should try to write to secondary use the custom server selector
// otherwise provide the read preference.
selector = secondaryWritableServerSelector(topology.commonWireVersion, readPreference);
} else {
selector = readPreference;
}
let server = await topology.selectServer(selector, {
session,
operationName: operation.commandName,
timeoutContext,
signal: operation.options.signal
});
const hasReadAspect = operation.hasAspect(Aspect.READ_OPERATION);
const hasWriteAspect = operation.hasAspect(Aspect.WRITE_OPERATION);
const inTransaction = session?.inTransaction() ?? false;
const willRetryRead = topology.s.options.retryReads && !inTransaction && operation.canRetryRead;
const willRetryWrite =
topology.s.options.retryWrites &&
!inTransaction &&
supportsRetryableWrites(server) &&
operation.canRetryWrite;
const willRetry =
operation.hasAspect(Aspect.RETRYABLE) &&
session != null &&
((hasReadAspect && willRetryRead) || (hasWriteAspect && willRetryWrite));
if (hasWriteAspect && willRetryWrite && session != null) {
operation.options.willRetryWrite = true;
session.incrementTransactionNumber();
}
const maxTries = willRetry ? (timeoutContext.csotEnabled() ? Infinity : 2) : 1;
let previousOperationError: MongoError | undefined;
let previousServer: ServerDescription | undefined;
for (let tries = 0; tries < maxTries; tries++) {
if (previousOperationError) {
if (hasWriteAspect && previousOperationError.code === MMAPv1_RETRY_WRITES_ERROR_CODE) {
throw new MongoServerError({
message: MMAPv1_RETRY_WRITES_ERROR_MESSAGE,
errmsg: MMAPv1_RETRY_WRITES_ERROR_MESSAGE,
originalError: previousOperationError
});
}
if (operation.hasAspect(Aspect.COMMAND_BATCHING) && !operation.canRetryWrite) {
throw previousOperationError;
}
if (hasWriteAspect && !isRetryableWriteError(previousOperationError))
throw previousOperationError;
if (hasReadAspect && !isRetryableReadError(previousOperationError))
throw previousOperationError;
if (
previousOperationError instanceof MongoNetworkError &&
operation.hasAspect(Aspect.CURSOR_CREATING) &&
session != null &&
session.isPinned &&
!session.inTransaction()
) {
session.unpin({ force: true, forceClear: true });
}
server = await topology.selectServer(selector, {
session,
operationName: operation.commandName,
previousServer,
signal: operation.options.signal
});
Iif (hasWriteAspect && !supportsRetryableWrites(server)) {
throw new MongoUnexpectedServerResponseError(
'Selected server does not support retryable writes'
);
}
}
try {
// If tries > 0 and we are command batching we need to reset the batch.
if (tries > 0 && operation.hasAspect(Aspect.COMMAND_BATCHING)) {
operation.resetBatch();
}
return await operation.execute(server, session, timeoutContext);
} catch (operationError) {
if (!(operationError instanceof MongoError)) throw operationError;
if (
previousOperationError != null &&
operationError.hasErrorLabel(MongoErrorLabel.NoWritesPerformed)
) {
throw previousOperationError;
}
previousServer = server.description;
previousOperationError = operationError;
// Reset timeouts
timeoutContext.clear();
}
}
throw (
previousOperationError ??
new MongoRuntimeError('Tried to propagate retryability error, but no error was found.')
);
}
|