feat(NODE-2993): implement maxConnecting (#3255)

Author: dariakp

src/cmap/connection_pool.ts

@@ -107,7 +107,7 @@ export type ConnectionPoolEvents = {
  */
 export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
  closed: boolean;
- options: Readonly<ConnectionPoolOptions>;
+ options: Readonly<ConnectionPoolOptions & { maxConnecting: number }>;
  /** @internal */
  [kLogger]: Logger;
  /** @internal */
@@ -199,6 +199,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
  connectionType: Connection,
  maxPoolSize: options.maxPoolSize ?? 100,
  minPoolSize: options.minPoolSize ?? 0,
+ maxConnecting: 2,
  maxIdleTimeMS: options.maxIdleTimeMS ?? 0,
  waitQueueTimeoutMS: options.waitQueueTimeoutMS ?? 0,
  autoEncrypter: options.autoEncrypter,
@@ -494,16 +495,29 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
 }
 
 function ensureMinPoolSize(pool: ConnectionPool) {
- if (pool.closed || pool.options.minPoolSize === 0) {
+ const minPoolSize = pool.options.minPoolSize;
+ if (pool.closed || minPoolSize === 0) {
  return;
  }
 
- const minPoolSize = pool.options.minPoolSize;
- for (let i = pool.totalConnectionCount; i < minPoolSize; ++i) {
- createConnection(pool);
+ if (
+ pool.totalConnectionCount < minPoolSize &&
+ pool.pendingConnectionCount < pool.options.maxConnecting
+ ) {
+ // NOTE: ensureMinPoolSize should not try to get all the pending
+ // connection permits because that potentially delays the availability of
+ // the connection to a checkout request
+ createConnection(pool, (err, connection) => {
+ pool[kPending]--;
+ if (!err && connection) {
+ pool[kConnections].push(connection);
+ process.nextTick(processWaitQueue, pool);
+ }
+ pool[kMinPoolSizeTimer] = setTimeout(() => ensureMinPoolSize(pool), 10);
+ });
+ } else {
+ pool[kMinPoolSizeTimer] = setTimeout(() => ensureMinPoolSize(pool), 100);
  }
-
- pool[kMinPoolSizeTimer] = setTimeout(() => ensureMinPoolSize(pool), 10);
 }
 
 function connectionIsStale(pool: ConnectionPool, connection: Connection) {
@@ -521,7 +535,7 @@ function connectionIsIdle(pool: ConnectionPool, connection: Connection) {
  return !!(pool.options.maxIdleTimeMS && connection.idleTime > pool.options.maxIdleTimeMS);
 }
 
-function createConnection(pool: ConnectionPool, callback?: Callback<Connection>) {
+function createConnection(pool: ConnectionPool, callback: Callback<Connection>) {
  const connectOptions: ConnectionOptions = {
  ...pool.options,
  id: pool[kConnectionCounter].next().value,
@@ -530,14 +544,16 @@ function createConnection(pool: ConnectionPool, callback?: Callback<Connection>)
  };
 
  pool[kPending]++;
+ // This is our version of a "virtual" no-I/O connection as the spec requires
+ pool.emit(
+ ConnectionPool.CONNECTION_CREATED,
+ new ConnectionCreatedEvent(pool, { id: connectOptions.id })
+ );
+
  connect(connectOptions, (err, connection) => {
  if (err || !connection) {
- pool[kPending]--;
  pool[kLogger].debug(`connection attempt failed with error [${JSON.stringify(err)}]`);
- if (typeof callback === 'function') {
- callback(err);
- }
-
+ callback(err);
  return;
  }
 
@@ -553,8 +569,6 @@ function createConnection(pool: ConnectionPool, callback?: Callback<Connection>)
  connection.on(event, (e: any) => pool.emit(event, e));
  }
 
- pool.emit(ConnectionPool.CONNECTION_CREATED, new ConnectionCreatedEvent(pool, connection));
-
  if (pool.loadBalanced) {
  connection.on(Connection.PINNED, pinType => pool[kMetrics].markPinned(pinType));
  connection.on(Connection.UNPINNED, pinType => pool[kMetrics].markUnpinned(pinType));
@@ -575,16 +589,8 @@ function createConnection(pool: ConnectionPool, callback?: Callback<Connection>)
  connection.markAvailable();
  pool.emit(ConnectionPool.CONNECTION_READY, new ConnectionReadyEvent(pool, connection));
 
- // if a callback has been provided, hand off the connection immediately
- if (typeof callback === 'function') {
- callback(undefined, connection);
- return;
- }
-
- // otherwise add it to the pool for later acquisition, and try to process the wait queue
- pool[kConnections].push(connection);
- pool[kPending]--;
- process.nextTick(processWaitQueue, pool);
+ callback(undefined, connection);
+ return;
  });
 }
 
@@ -642,44 +648,45 @@ function processWaitQueue(pool: ConnectionPool) {
  }
  }
 
- const maxPoolSize = pool.options.maxPoolSize;
- if (pool.waitQueueSize && (maxPoolSize <= 0 || pool.totalConnectionCount < maxPoolSize)) {
+ const { maxPoolSize, maxConnecting } = pool.options;
+ while (
+ pool.waitQueueSize > 0 &&
+ pool.pendingConnectionCount < maxConnecting &&
+ (maxPoolSize === 0 || pool.totalConnectionCount < maxPoolSize)
+ ) {
+ const waitQueueMember = pool[kWaitQueue].shift();
+ if (!waitQueueMember || waitQueueMember[kCancelled]) {
+ continue;
+ }
  createConnection(pool, (err, connection) => {
- const waitQueueMember = pool[kWaitQueue].shift();
- if (!waitQueueMember || waitQueueMember[kCancelled]) {
+ pool[kPending]--;
+ if (waitQueueMember[kCancelled]) {
  if (!err && connection) {
  pool[kConnections].push(connection);
- pool[kPending]--;
+ }
+ } else {
+ if (err) {
+ pool.emit(
+ ConnectionPool.CONNECTION_CHECK_OUT_FAILED,
+ new ConnectionCheckOutFailedEvent(pool, err)
+ );
+ } else if (connection) {
+ pool[kCheckedOut]++;
+ pool.emit(
+ ConnectionPool.CONNECTION_CHECKED_OUT,
+ new ConnectionCheckedOutEvent(pool, connection)
+ );
  }
 
- pool[kProcessingWaitQueue] = false;
- return;
- }
-
- if (err) {
- pool.emit(
- ConnectionPool.CONNECTION_CHECK_OUT_FAILED,
- new ConnectionCheckOutFailedEvent(pool, err)
- );
- } else if (connection) {
- pool[kCheckedOut]++;
- pool[kPending]--;
- pool.emit(
- ConnectionPool.CONNECTION_CHECKED_OUT,
- new ConnectionCheckedOutEvent(pool, connection)
- );
- }
-
- if (waitQueueMember.timer) {
- clearTimeout(waitQueueMember.timer);
+ if (waitQueueMember.timer) {
+ clearTimeout(waitQueueMember.timer);
+ }
+ waitQueueMember.callback(err, connection);
  }
- waitQueueMember.callback(err, connection);
- pool[kProcessingWaitQueue] = false;
- process.nextTick(() => processWaitQueue(pool));
+ process.nextTick(processWaitQueue, pool);
  });
- } else {
- pool[kProcessingWaitQueue] = false;
  }
+ pool[kProcessingWaitQueue] = false;
 }
 
 /**

src/cmap/connection_pool_events.ts

@@ -59,7 +59,7 @@ export class ConnectionCreatedEvent extends ConnectionPoolMonitoringEvent {
  connectionId: number | '<monitor>';
 
  /** @internal */
- constructor(pool: ConnectionPool, connection: Connection) {
+ constructor(pool: ConnectionPool, connection: { id: number | '<monitor>' }) {
  super(pool);
  this.connectionId = connection.id;
  }

test/integration/connection-monitoring-and-pooling/cmap-node-specs/pool-minPoolSize-population-stats.json

@@ -33,16 +33,16 @@
  "address": 42,
  "currentCheckedOutCount": 0,
  "availableConnectionCount": 0,
- "pendingConnectionCount": 3,
- "totalConnectionCount": 3
+ "pendingConnectionCount": 1,
+ "totalConnectionCount": 1
  },
  {
  "type": "ConnectionCreated",
  "connectionId": 42,
  "address": 42,
  "availableConnectionCount": 1,
- "pendingConnectionCount": 2,
- "totalConnectionCount": 3
+ "pendingConnectionCount": 1,
+ "totalConnectionCount": 2
  },
  {
  "type": "ConnectionCreated",

test/integration/connection-monitoring-and-pooling/cmap-node-specs/pool-minPoolSize-replace-removed-connections.json

@@ -0,0 +1,77 @@
+{
+ "version": 1,
+ "style": "unit",
+ "description": "must replace removed connections up to minPoolSize",
+ "poolOptions": {
+ "minPoolSize": 2
+ },
+ "operations": [
+ {
+ "name": "waitForEvent",
+ "event": "ConnectionReady",
+ "count": 2
+ },
+ {
+ "name": "wait",
+ "ms": 1000
+ },
+ {
+ "name": "checkOut",
+ "label": "conn"
+ },
+ {
+ "name": "clear"
+ },
+ {
+ "name": "checkIn",
+ "connection": "conn"
+ },
+ {
+ "name": "waitForEvent",
+ "event": "ConnectionReady",
+ "count": 3
+ }
+ ],
+ "events": [
+ {
+ "type": "ConnectionReady",
+ "address": 42
+ },
+ {
+ "type": "ConnectionReady",
+ "address": 42
+ },
+ {
+ "type": "ConnectionCheckedOut",
+ "address": 42
+ },
+ {
+ "type": "ConnectionPoolCleared",
+ "address": 42
+ },
+ {
+ "type": "ConnectionCheckedIn",
+ "address": 42
+ },
+ {
+ "type": "ConnectionClosed",
+ "reason": "stale",
+ "address": 42,
+ "availableConnectionCount": 1,
+ "pendingConnectionCount": 0,
+ "totalConnectionCount": 1
+ },
+ {
+ "type": "ConnectionReady",
+ "address": 42,
+ "availableConnectionCount": 1,
+ "pendingConnectionCount": 1,
+ "totalConnectionCount": 2
+ }
+ ],
+ "ignore": [
+ "ConnectionPoolCreated",
+ "ConnectionCreated",
+ "ConnectionCheckOutStarted"
+ ]
+}

test/integration/connection-monitoring-and-pooling/connection_monitoring_and_pooling.spec.test.ts

@@ -15,9 +15,14 @@ const LB_SKIP_TESTS: SkipDescription[] = [
 describe('Connection Monitoring and Pooling Spec Tests (Integration)', function () {
  const tests: CmapTest[] = loadSpecTests('connection-monitoring-and-pooling');
 
- runCmapTestSuite(
- // TODO(NODE-2993): unskip integration tests for maxConnecting
- tests.filter(({ style }) => style === 'unit'),
- { testsToSkip: LB_SKIP_TESTS }
- );
+ runCmapTestSuite(tests, {
+ testsToSkip: LB_SKIP_TESTS.concat([
+ {
+ description: 'waiting on maxConnecting is limited by WaitQueueTimeoutMS',
+ skipIfCondition: 'always',
+ skipReason:
+ 'not applicable: waitQueueTimeoutMS limits connection establishment time in our driver'
+ }
+ ])
+ });
 });

test/integration/connection-monitoring-and-pooling/connection_monitoring_and_pooling.test.ts

@@ -6,5 +6,14 @@ describe('Connection Monitoring and Pooling (Node Driver)', function () {
  '../integration/connection-monitoring-and-pooling/cmap-node-specs'
  );
 
- runCmapTestSuite(tests, { injectPoolStats: true });
+ runCmapTestSuite(tests, {
+ injectPoolStats: true,
+ testsToSkip: [
+ {
+ description: 'must replace removed connections up to minPoolSize',
+ skipIfCondition: 'loadBalanced',
+ skipReason: 'cannot run against load balancer due to reliance on pool.clear() command'
+ }
+ ]
+ });
 });

test/tools/cmap_spec_runner.ts

@@ -116,12 +116,14 @@ class Thread {
  await sleep();
  }
 
- queue(op: CmapOperation) {
+ queue(op: CmapOperation, thread?: Thread) {
  if (this.#killed || this.#error) {
  return;
  }
 
- this.#promise = this.#promise.then(() => this._runOperation(op)).catch(e => (this.#error = e));
+ const functionToQueue = () => (!thread ? this._runOperation(op) : thread.queue(op));
+
+ this.#promise = this.#promise.then(functionToQueue).catch(e => (this.#error = e));
  }
 
  async finish() {
@@ -352,9 +354,12 @@ async function runCmapTest(test: CmapTest, threadContext: ThreadContext) {
  const op = operations[idx];
 
  const threadKey = op.name === 'checkOut' ? op.thread || MAIN_THREAD_KEY : MAIN_THREAD_KEY;
- const thread = threadContext.getThread(threadKey);
-
- thread.queue(op);
+ if (threadKey === MAIN_THREAD_KEY) {
+ mainThread.queue(op);
+ } else {
+ const thread = threadContext.getThread(threadKey);
+ mainThread.queue(op, thread);
+ }
  }
 
  await mainThread.finish().catch(e => {
@@ -387,6 +392,7 @@ async function runCmapTest(test: CmapTest, threadContext: ThreadContext) {
  );
 
  expect(actualEvents).to.have.lengthOf(expectedEvents.length);
+
  for (const expected of expectedEvents) {
  const actual = actualEvents.shift();
  const { type: eventType, ...eventPropsToCheck } = expected;
@@ -397,7 +403,7 @@ async function runCmapTest(test: CmapTest, threadContext: ThreadContext) {
 
 export type SkipDescription = {
  description: string;
- skipIfCondition: 'loadBalanced';
+ skipIfCondition: 'loadBalanced' | 'always';
  skipReason: string;
 };
 
@@ -416,10 +422,12 @@ export function runCmapTestSuite(
  ({ description }) => description === test.description
  );
  if (skipDescription) {
+ const alwaysSkip = skipDescription.skipIfCondition === 'always';
  const matchesLoadBalanceSkip =
  skipDescription.skipIfCondition === 'loadBalanced' && this.configuration.isLoadBalanced;
- if (matchesLoadBalanceSkip) {
- (this.currentTest as Mocha.Runnable).skipReason = skipDescription.skipReason;
+
+ if (alwaysSkip || matchesLoadBalanceSkip) {
+ this.currentTest.skipReason = skipDescription.skipReason;
  this.skip();
  }
  }