BUG#33761268: skip connection requests on unavailable pool

When a connection pool is full, any new connection request is retried
within a sliding window of time until it reaches the time specified by
the "queueTimeout" option. By default, the connection request will be
retried forever, until the connection is effectively acquired.

Currently, if the pool is closed in the meantime, all connection
requests that have been queued will still be retried in the future, as
long as the application process does not terminate. This is bad and
might result in dangling sockets and formal MySQL server connections,
which is a waste of resources and a potential source of memory leaks.

This patch introduces the changes necessary to ensure all queued
connection requests in a pool are skipped when the pool is closed or is
in the process of being closed (where it should still no longer be
available in order to avoid parallelism issues).

Author: ruiquelhas

lib/DevAPI/Client.js

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018, 2021, Oracle and/or its affiliates.
+ * Copyright (c) 2018, 2022, Oracle and/or its affiliates.
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License, version 2.0, as
@@ -127,6 +127,13 @@ function Client (options = {}) {
 
             return state.pool.getConnection()
                 .then(con => {
+                    // If the connection does not exist, probably because
+                    // the pool was closed in the meantime, there is nothing
+                    // else to do.
+                    if (!con) {
+                        return;
+                    }
+
                     return session(con);
                 });
         }

lib/DevAPI/ConnectionPool.js

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018, 2021, Oracle and/or its affiliates.
+ * Copyright (c) 2018, 2022, Oracle and/or its affiliates.
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License, version 2.0, as
@@ -77,6 +77,9 @@ function ConnectionPool (options = {}) {
         // Contains the list of active connections i.e. that have not been
         // closed neither by the client nor the server.
         active: [],
+        // Informs if the pool is avaiable or not. After it starts to be
+        // destroyed, it should become unavailable.
+        available: true,
         // Contains the list of connections that have expired i.e. where
         // maxIdleTime was exceeded or have been closed by the server.
         expired: [],
@@ -113,6 +116,7 @@ function ConnectionPool (options = {}) {
          */
         create ({ active = [], expired = [], idle = [] } = {}) {
             state.active = active;
+            state.available = true;
             state.expired = expired;
             state.idle = idle;
             state.tickets = active.length;
@@ -128,10 +132,13 @@ function ConnectionPool (options = {}) {
          * @returns {Promise}
          */
         destroy () {
+            // We should put the pool in a state that prevents queued
+            // connection requests from being fulfilled.
+            state.available = false;
             // Expired connections have already been destroyed by this point.
             return Promise.all(state.idle.concat(state.active).map(con => con.destroy()))
                 .then(() => {
-                    this.reset();
+                    return this.reset();
                 })
                 .catch(err => {
                     this.reset();
@@ -168,6 +175,16 @@ function ConnectionPool (options = {}) {
         getConnection (requestedAt = Date.now()) {
             return this.update()
                 .then(() => {
+                    // If previous connection requests were queued and the
+                    // pool is closed or being closed, we should not try to
+                    // create, refurbish or re-use connections at the risk of
+                    // leaving a system socket open which will be not cleaned
+                    // up after the process finishes, leading to a resource
+                    // leak. There is nothing else left to do.
+                    if (!this.isAvailable()) {
+                        return;
+                    }
+
                     const elapsedTime = Date.now() - requestedAt;
                     const queueTimeout = options.pooling.queueTimeout;
 
@@ -247,6 +264,17 @@ function ConnectionPool (options = {}) {
             return state.idle;
         },
 
+        /**
+         * Checks if the pool is being closed.
+         * @private
+         * @function
+         * @name module:ConnectionPool#isAvailable
+         * @returns {boolean}
+         */
+        isAvailable () {
+            return state.available;
+        },
+
         /**
          * Checks if a pool is full i.e. a total of "maxSize" number of
          * connections are currently active.
@@ -270,6 +298,9 @@ function ConnectionPool (options = {}) {
          */
         reset () {
             state.active = [];
+            // The original pool instance will never be available again.
+            // A new one should be created by the client wrapper.
+            state.available = false;
             state.expired = [];
             state.idle = [];
             state.tickets = 0;

test/functional/default/connection/pool.js

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2021, Oracle and/or its affiliates.
+ * Copyright (c) 2021, 2022, Oracle and/or its affiliates.
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License, version 2.0, as
@@ -657,6 +657,24 @@ describe('connection pool', () => {
                     return pool.close();
                 });
         });
+
+        it('queued connection requests are not fullfilled', () => {
+            const poolingConfig = Object.assign({}, config, baseConfig);
+            const pool = mysqlx.getClient(poolingConfig, { pooling: { maxSize: 2 } });
+
+            return Promise.all([pool.getSession(), pool.getSession()])
+                .then(() => {
+                    return new Promise(resolve => {
+                        // We need to ensure the test only finishes after the
+                        // queued connection request is processed.
+                        pool.getSession().then(resolve);
+                        pool.close();
+                    });
+                })
+                .then(session => {
+                    return expect(session).to.not.exist;
+                });
+        });
     });
 
     context('when acquiring connections in parallel', () => {
@@ -675,6 +693,7 @@ describe('connection pool', () => {
             const maxSize = totalRequestsInQueue / 2;
             const queueTimeout = 500;
             const pool = mysqlx.getClient(poolingConfig, { pooling: { enabled: true, maxSize, queueTimeout } });
+            const connections = [...Array(totalRequestsInQueue)];
 
             const connectAndWaitBeforeClose = () => {
                 return pool.getSession()
@@ -691,9 +710,7 @@ describe('connection pool', () => {
                     });
             };
 
-            const work = [...Array(totalRequestsInQueue)].map(() => connectAndWaitBeforeClose());
-
-            return Promise.all(work)
+            return Promise.all(connections.map(() => connectAndWaitBeforeClose()))
                 .then(() => {
                     return pool.close();
                 });
@@ -704,6 +721,7 @@ describe('connection pool', () => {
             const maxSize = totalRequestsInQueue / 2;
             const queueTimeout = 500;
             const pool = mysqlx.getClient(poolingConfig, { pooling: { enabled: true, maxSize, queueTimeout } });
+            const connections = [...Array(totalRequestsInQueue)];
             const error = `Could not retrieve a connection from the pool. Timeout of ${queueTimeout} ms was exceeded.`;
 
             const connectAndWaitBeforeClose = () => {
@@ -721,9 +739,7 @@ describe('connection pool', () => {
                     });
             };
 
-            const work = [...Array(totalRequestsInQueue)].map(() => connectAndWaitBeforeClose());
-
-            return Promise.all(work)
+            return Promise.all(connections.map(() => connectAndWaitBeforeClose()))
                 .then(() => {
                     return expect.fail();
                 })
@@ -799,4 +815,53 @@ describe('connection pool', () => {
                 });
         });
     });
+
+    context('when the pool has been closed', () => {
+        it('fulfills new connection requests', () => {
+            const poolingConfig = Object.assign({}, config, baseConfig);
+            const pool = mysqlx.getClient(poolingConfig, { pooling: { maxSize: 2 } });
+
+            return Promise.all([pool.getSession(), pool.getSession()])
+                .then(() => {
+                    return pool.close();
+                })
+                .then(() => {
+                    return pool.getSession();
+                })
+                .then(session => {
+                    // eslint-disable-next-line no-unused-expressions
+                    expect(session).to.exist;
+                    return pool.close();
+                });
+        });
+
+        it('queued connection requests are not fullfilled', () => {
+            const poolingConfig = Object.assign({}, config, baseConfig);
+            const pool = mysqlx.getClient(poolingConfig, { pooling: { maxSize: 2 } });
+
+            return Promise.all([pool.getSession(), pool.getSession()])
+                .then(() => {
+                    return new Promise(resolve => {
+                        // We need to ensure the test only finishes after the
+                        // queued connection request is processed.
+                        pool.getSession().then(resolve);
+                        pool.close();
+                    });
+                })
+                .then(() => {
+                    return Promise.all([pool.getSession(), pool.getSession()]);
+                })
+                .then(() => {
+                    return new Promise(resolve => {
+                        // We need to ensure the test only finishes after the
+                        // queued connection request is processed.
+                        pool.getSession().then(resolve);
+                        pool.close();
+                    });
+                })
+                .then(session => {
+                    return expect(session).to.not.exist;
+                });
+        });
+    });
 });

test/unit/DevAPI/Client.js

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018, 2021, Oracle and/or its affiliates.
+ * Copyright (c) 2018, 2022, Oracle and/or its affiliates.
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License, version 2.0, as
@@ -89,6 +89,21 @@ describe('DevAPI Client', () => {
                     expect(res).to.equal('qux');
                 });
         });
+
+        it('does not create a session if the pool is not able to return a valid connection', () => {
+            const options = { foo: 'bar', pooling: { enabled: true } };
+            const create = td.function();
+            const getConnection = td.function();
+
+            td.when(pool(options)).thenReturn({ create });
+            td.when(create()).thenReturn({ getConnection });
+            td.when(getConnection()).thenResolve('baz');
+
+            return client(options).getSession()
+                .then(res => {
+                    return expect(res).to.not.exist;
+                });
+        });
     });
 
     context('close()', () => {