Skip to content

Commit aa96cb0

Browse files
nbbeekenW-A-James
nbbeeken
and
W-A-James
committed
feat(NODE-5682): set maxTimeMS on commands and preempt I/O (#4174)
Co-authored-by: Warren James <[email protected]>
1 parent 26800a9 commit aa96cb0

File tree

16 files changed

+200
-77
lines changed

16 files changed

+200
-77
lines changed

‎src/admin.ts

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -155,7 +155,10 @@ export class Admin {
155155
* @param options - Optional settings for the command
156156
*/
157157
async listDatabases(options?: ListDatabasesOptions): Promise<ListDatabasesResult> {
158-
return await executeOperation(this.s.db.client, new ListDatabasesOperation(this.s.db, options));
158+
return await executeOperation(
159+
this.s.db.client,
160+
new ListDatabasesOperation(this.s.db, { timeoutMS: this.s.db.timeoutMS, ...options })
161+
);
159162
}
160163

161164
/**

‎src/cmap/connection.ts

Lines changed: 58 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ import {
1818
MongoMissingDependencyError,
1919
MongoNetworkError,
2020
MongoNetworkTimeoutError,
21+
MongoOperationTimeoutError,
2122
MongoParseError,
2223
MongoServerError,
2324
MongoUnexpectedServerResponseError
@@ -29,7 +30,7 @@ import { type CancellationToken, TypedEventEmitter } from '../mongo_types';
2930
import { ReadPreference, type ReadPreferenceLike } from '../read_preference';
3031
import { ServerType } from '../sdam/common';
3132
import { applySession, type ClientSession, updateSessionFromResponse } from '../sessions';
32-
import { type TimeoutContext } from '../timeout';
33+
import { type TimeoutContext, TimeoutError } from '../timeout';
3334
import {
3435
BufferPool,
3536
calculateDurationInMs,
@@ -416,6 +417,11 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
416417
...options
417418
};
418419

420+
if (options.timeoutContext?.csotEnabled()) {
421+
const { maxTimeMS } = options.timeoutContext;
422+
if (maxTimeMS > 0 && Number.isFinite(maxTimeMS)) cmd.maxTimeMS = maxTimeMS;
423+
}
424+
419425
const message = this.supportsOpMsg
420426
? new OpMsgRequest(db, cmd, commandOptions)
421427
: new OpQueryRequest(db, cmd, commandOptions);
@@ -430,7 +436,9 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
430436
): AsyncGenerator<MongoDBResponse> {
431437
this.throwIfAborted();
432438

433-
if (typeof options.socketTimeoutMS === 'number') {
439+
if (options.timeoutContext?.csotEnabled()) {
440+
this.socket.setTimeout(0);
441+
} else if (typeof options.socketTimeoutMS === 'number') {
434442
this.socket.setTimeout(options.socketTimeoutMS);
435443
} else if (this.socketTimeoutMS !== 0) {
436444
this.socket.setTimeout(this.socketTimeoutMS);
@@ -439,7 +447,8 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
439447
try {
440448
await this.writeCommand(message, {
441449
agreedCompressor: this.description.compressor ?? 'none',
442-
zlibCompressionLevel: this.description.zlibCompressionLevel
450+
zlibCompressionLevel: this.description.zlibCompressionLevel,
451+
timeoutContext: options.timeoutContext
443452
});
444453

445454
if (options.noResponse) {
@@ -449,7 +458,17 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
449458

450459
this.throwIfAborted();
451460

452-
for await (const response of this.readMany()) {
461+
if (
462+
options.timeoutContext?.csotEnabled() &&
463+
options.timeoutContext.minRoundTripTime != null &&
464+
options.timeoutContext.remainingTimeMS < options.timeoutContext.minRoundTripTime
465+
) {
466+
throw new MongoOperationTimeoutError(
467+
'Server roundtrip time is greater than the time remaining'
468+
);
469+
}
470+
471+
for await (const response of this.readMany({ timeoutContext: options.timeoutContext })) {
453472
this.socket.setTimeout(0);
454473
const bson = response.parse();
455474

@@ -622,7 +641,11 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
622641
*/
623642
private async writeCommand(
624643
command: WriteProtocolMessageType,
625-
options: { agreedCompressor?: CompressorName; zlibCompressionLevel?: number }
644+
options: {
645+
agreedCompressor?: CompressorName;
646+
zlibCompressionLevel?: number;
647+
timeoutContext?: TimeoutContext;
648+
}
626649
): Promise<void> {
627650
const finalCommand =
628651
options.agreedCompressor === 'none' || !OpCompressedRequest.canCompress(command)
@@ -634,8 +657,32 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
634657

635658
const buffer = Buffer.concat(await finalCommand.toBin());
636659

660+
if (options.timeoutContext?.csotEnabled()) {
661+
if (
662+
options.timeoutContext.minRoundTripTime != null &&
663+
options.timeoutContext.remainingTimeMS < options.timeoutContext.minRoundTripTime
664+
) {
665+
throw new MongoOperationTimeoutError(
666+
'Server roundtrip time is greater than the time remaining'
667+
);
668+
}
669+
}
670+
637671
if (this.socket.write(buffer)) return;
638-
return await once(this.socket, 'drain');
672+
673+
const drainEvent = once<void>(this.socket, 'drain');
674+
const timeout = options?.timeoutContext?.timeoutForSocketWrite;
675+
if (timeout) {
676+
try {
677+
return await Promise.race([drainEvent, timeout]);
678+
} catch (error) {
679+
if (TimeoutError.is(error)) {
680+
throw new MongoOperationTimeoutError('Timed out at socket write');
681+
}
682+
throw error;
683+
}
684+
}
685+
return await drainEvent;
639686
}
640687

641688
/**
@@ -647,9 +694,12 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
647694
*
648695
* Note that `for-await` loops call `return` automatically when the loop is exited.
649696
*/
650-
private async *readMany(): AsyncGenerator<OpMsgResponse | OpReply> {
697+
private async *readMany(options: {
698+
timeoutContext?: TimeoutContext;
699+
}): AsyncGenerator<OpMsgResponse | OpReply> {
651700
try {
652-
this.dataEvents = onData(this.messageStream);
701+
this.dataEvents = onData(this.messageStream, options);
702+
653703
for await (const message of this.dataEvents) {
654704
const response = await decompressResponse(message);
655705
yield response;

‎src/cmap/wire_protocol/on_data.ts

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
import { type EventEmitter } from 'events';
22

3+
import { MongoOperationTimeoutError } from '../../error';
4+
import { type TimeoutContext, TimeoutError } from '../../timeout';
35
import { List, promiseWithResolvers } from '../../utils';
46

57
/**
@@ -18,7 +20,10 @@ type PendingPromises = Omit<
1820
* Returns an AsyncIterator that iterates each 'data' event emitted from emitter.
1921
* It will reject upon an error event.
2022
*/
21-
export function onData(emitter: EventEmitter) {
23+
export function onData(
24+
emitter: EventEmitter,
25+
{ timeoutContext }: { timeoutContext?: TimeoutContext }
26+
) {
2227
// Setup pending events and pending promise lists
2328
/**
2429
* When the caller has not yet called .next(), we store the
@@ -86,6 +91,8 @@ export function onData(emitter: EventEmitter) {
8691
// Adding event handlers
8792
emitter.on('data', eventHandler);
8893
emitter.on('error', errorHandler);
94+
// eslint-disable-next-line github/no-then
95+
timeoutContext?.timeoutForSocketRead?.then(undefined, errorHandler);
8996

9097
return iterator;
9198

@@ -97,8 +104,12 @@ export function onData(emitter: EventEmitter) {
97104

98105
function errorHandler(err: Error) {
99106
const promise = unconsumedPromises.shift();
100-
if (promise != null) promise.reject(err);
101-
else error = err;
107+
const timeoutError = TimeoutError.is(err)
108+
? new MongoOperationTimeoutError('Timed out during socket read')
109+
: undefined;
110+
111+
if (promise != null) promise.reject(timeoutError ?? err);
112+
else error = timeoutError ?? err;
102113
void closeHandler();
103114
}
104115

‎src/db.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -277,7 +277,7 @@ export class Db {
277277
this.client,
278278
new RunCommandOperation(this, command, {
279279
...resolveBSONOptions(options),
280-
timeoutMS: options?.timeoutMS,
280+
timeoutMS: options?.timeoutMS ?? this.timeoutMS,
281281
session: options?.session,
282282
readPreference: options?.readPreference
283283
})

‎src/sdam/topology.ts

Lines changed: 10 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -460,29 +460,28 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
460460
}
461461
}
462462

463-
const timeoutMS = this.client.s.options.timeoutMS;
463+
// TODO(NODE-6223): auto connect cannot use timeoutMS
464+
// const timeoutMS = this.client.s.options.timeoutMS;
464465
const serverSelectionTimeoutMS = this.client.s.options.serverSelectionTimeoutMS;
465466
const readPreference = options.readPreference ?? ReadPreference.primary;
466-
467467
const timeoutContext = TimeoutContext.create({
468-
timeoutMS,
468+
timeoutMS: undefined,
469469
serverSelectionTimeoutMS,
470470
waitQueueTimeoutMS: this.client.s.options.waitQueueTimeoutMS
471471
});
472-
473472
const selectServerOptions = {
474473
operationName: 'ping',
475474
...options,
476475
timeoutContext
477476
};
477+
478478
try {
479479
const server = await this.selectServer(
480480
readPreferenceServerSelector(readPreference),
481481
selectServerOptions
482482
);
483-
484483
const skipPingOnConnect = this.s.options[Symbol.for('@@mdb.skipPingOnConnect')] === true;
485-
if (!skipPingOnConnect && server && this.s.credentials) {
484+
if (!skipPingOnConnect && this.s.credentials) {
486485
await server.command(ns('admin.$cmd'), { ping: 1 }, { timeoutContext });
487486
stateTransition(this, STATE_CONNECTED);
488487
this.emit(Topology.OPEN, this);
@@ -623,7 +622,11 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
623622

624623
try {
625624
timeout?.throwIfExpired();
626-
return await (timeout ? Promise.race([serverPromise, timeout]) : serverPromise);
625+
const server = await (timeout ? Promise.race([serverPromise, timeout]) : serverPromise);
626+
if (options.timeoutContext?.csotEnabled() && server.description.minRoundTripTime !== 0) {
627+
options.timeoutContext.minRoundTripTime = server.description.minRoundTripTime;
628+
}
629+
return server;
627630
} catch (error) {
628631
if (TimeoutError.is(error)) {
629632
// Timeout

‎src/timeout.ts

Lines changed: 36 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
import { clearTimeout, setTimeout } from 'timers';
22

3-
import { MongoInvalidArgumentError, MongoRuntimeError } from './error';
3+
import { MongoInvalidArgumentError, MongoOperationTimeoutError, MongoRuntimeError } from './error';
44
import { csotMin, noop } from './utils';
55

66
/** @internal */
@@ -51,7 +51,7 @@ export class Timeout extends Promise<never> {
5151
}
5252

5353
/** Create a new timeout that expires in `duration` ms */
54-
private constructor(executor: Executor = () => null, duration: number, unref = false) {
54+
private constructor(executor: Executor = () => null, duration: number, unref = true) {
5555
let reject!: Reject;
5656

5757
if (duration < 0) {
@@ -164,6 +164,10 @@ export abstract class TimeoutContext {
164164

165165
abstract get clearConnectionCheckoutTimeout(): boolean;
166166

167+
abstract get timeoutForSocketWrite(): Timeout | null;
168+
169+
abstract get timeoutForSocketRead(): Timeout | null;
170+
167171
abstract csotEnabled(): this is CSOTTimeoutContext;
168172
}
169173

@@ -176,13 +180,15 @@ export class CSOTTimeoutContext extends TimeoutContext {
176180
clearConnectionCheckoutTimeout: boolean;
177181
clearServerSelectionTimeout: boolean;
178182

179-
private _maxTimeMS?: number;
180-
181183
private _serverSelectionTimeout?: Timeout | null;
182184
private _connectionCheckoutTimeout?: Timeout | null;
185+
public minRoundTripTime = 0;
186+
private start: number;
183187

184188
constructor(options: CSOTTimeoutContextOptions) {
185189
super();
190+
this.start = Math.trunc(performance.now());
191+
186192
this.timeoutMS = options.timeoutMS;
187193

188194
this.serverSelectionTimeoutMS = options.serverSelectionTimeoutMS;
@@ -194,11 +200,12 @@ export class CSOTTimeoutContext extends TimeoutContext {
194200
}
195201

196202
get maxTimeMS(): number {
197-
return this._maxTimeMS ?? -1;
203+
return this.remainingTimeMS - this.minRoundTripTime;
198204
}
199205

200-
set maxTimeMS(v: number) {
201-
this._maxTimeMS = v;
206+
get remainingTimeMS() {
207+
const timePassed = Math.trunc(performance.now()) - this.start;
208+
return this.timeoutMS <= 0 ? Infinity : this.timeoutMS - timePassed;
202209
}
203210

204211
csotEnabled(): this is CSOTTimeoutContext {
@@ -239,6 +246,20 @@ export class CSOTTimeoutContext extends TimeoutContext {
239246
}
240247
return this._connectionCheckoutTimeout;
241248
}
249+
250+
get timeoutForSocketWrite(): Timeout | null {
251+
const { remainingTimeMS } = this;
252+
if (!Number.isFinite(remainingTimeMS)) return null;
253+
if (remainingTimeMS > 0) return Timeout.expires(remainingTimeMS);
254+
throw new MongoOperationTimeoutError('Timed out before socket write');
255+
}
256+
257+
get timeoutForSocketRead(): Timeout | null {
258+
const { remainingTimeMS } = this;
259+
if (!Number.isFinite(remainingTimeMS)) return null;
260+
if (remainingTimeMS > 0) return Timeout.expires(remainingTimeMS);
261+
throw new MongoOperationTimeoutError('Timed out before socket read');
262+
}
242263
}
243264

244265
/** @internal */
@@ -269,4 +290,12 @@ export class LegacyTimeoutContext extends TimeoutContext {
269290
return Timeout.expires(this.options.waitQueueTimeoutMS);
270291
return null;
271292
}
293+
294+
get timeoutForSocketWrite(): Timeout | null {
295+
return null;
296+
}
297+
298+
get timeoutForSocketRead(): Timeout | null {
299+
return null;
300+
}
272301
}

‎test/integration/client-side-operations-timeout/client_side_operations_timeout.prose.test.ts

Lines changed: 12 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -384,7 +384,7 @@ describe('CSOT spec prose tests', function () {
384384
clock.restore();
385385
});
386386

387-
it('serverSelectionTimeoutMS honored if timeoutMS is not set', async function () {
387+
it.skip('serverSelectionTimeoutMS honored if timeoutMS is not set', async function () {
388388
/**
389389
* 1. Create a MongoClient (referred to as `client`) with URI `mongodb://invalid/?serverSelectionTimeoutMS=10`.
390390
* 1. Using `client`, execute the command `{ ping: 1 }` against the `admin` database.
@@ -416,10 +416,11 @@ describe('CSOT spec prose tests', function () {
416416

417417
await clock.tickAsync(11);
418418
expect(await maybeError).to.be.instanceof(MongoServerSelectionError);
419-
});
419+
}).skipReason =
420+
'TODO(NODE-6223): Auto connect performs extra server selection. Explicit connect throws on invalid host name';
420421
});
421422

422-
it("timeoutMS honored for server selection if it's lower than serverSelectionTimeoutMS", async function () {
423+
it.skip("timeoutMS honored for server selection if it's lower than serverSelectionTimeoutMS", async function () {
423424
/**
424425
* 1. Create a MongoClient (referred to as `client`) with URI `mongodb://invalid/?timeoutMS=10&serverSelectionTimeoutMS=20`.
425426
* 1. Using `client`, run the command `{ ping: 1 }` against the `admin` database.
@@ -440,9 +441,10 @@ describe('CSOT spec prose tests', function () {
440441

441442
expect(maybeError).to.be.instanceof(MongoOperationTimeoutError);
442443
expect(end - start).to.be.lte(15);
443-
});
444+
}).skipReason =
445+
'TODO(NODE-6223): Auto connect performs extra server selection. Explicit connect throws on invalid host name';
444446

445-
it("serverSelectionTimeoutMS honored for server selection if it's lower than timeoutMS", async function () {
447+
it.skip("serverSelectionTimeoutMS honored for server selection if it's lower than timeoutMS", async function () {
446448
/**
447449
* 1. Create a MongoClient (referred to as `client`) with URI `mongodb://invalid/?timeoutMS=20&serverSelectionTimeoutMS=10`.
448450
* 1. Using `client`, run the command `{ ping: 1 }` against the `admin` database.
@@ -462,9 +464,10 @@ describe('CSOT spec prose tests', function () {
462464

463465
expect(maybeError).to.be.instanceof(MongoOperationTimeoutError);
464466
expect(end - start).to.be.lte(15);
465-
});
467+
}).skipReason =
468+
'TODO(NODE-6223): Auto connect performs extra server selection. Explicit connect throws on invalid host name';
466469

467-
it('serverSelectionTimeoutMS honored for server selection if timeoutMS=0', async function () {
470+
it.skip('serverSelectionTimeoutMS honored for server selection if timeoutMS=0', async function () {
468471
/**
469472
* 1. Create a MongoClient (referred to as `client`) with URI `mongodb://invalid/?timeoutMS=0&serverSelectionTimeoutMS=10`.
470473
* 1. Using `client`, run the command `{ ping: 1 }` against the `admin` database.
@@ -484,7 +487,8 @@ describe('CSOT spec prose tests', function () {
484487

485488
expect(maybeError).to.be.instanceof(MongoOperationTimeoutError);
486489
expect(end - start).to.be.lte(15);
487-
});
490+
}).skipReason =
491+
'TODO(NODE-6223): Auto connect performs extra server selection. Explicit connect throws on invalid host name';
488492

489493
it.skip("timeoutMS honored for connection handshake commands if it's lower than serverSelectionTimeoutMS", async function () {
490494
/**

0 commit comments

Comments
 (0)