From d6dc899ad296d3ecee19b164b5be6f3aa715d695 Mon Sep 17 00:00:00 2001 From: Jonathan Horowitz Date: Fri, 20 Mar 2026 16:10:40 -0700 Subject: [PATCH 1/7] feat: auto purge stale prepared statements from cache and retry query --- lib/base/connection.js | 79 +++++++++++- .../connection/test-execute-cached.test.mts | 117 ++++++++++++++++++ 2 files changed, 194 insertions(+), 2 deletions(-) diff --git a/lib/base/connection.js b/lib/base/connection.js index 77da3d0f37..ee34411456 100644 --- a/lib/base/connection.js +++ b/lib/base/connection.js @@ -28,6 +28,7 @@ const Packets = require('../packets/index.js'); const Commands = require('../commands/index.js'); const ConnectionConfig = require('../connection_config.js'); const CharsetToEncoding = require('../constants/charset_encodings.js'); +const { ER_UNKNOWN_STMT_HANDLER } = require('../constants/errors.js'); const { traceCallback, tracePromise, @@ -814,13 +815,48 @@ class BaseConnection extends EventEmitter { this.addCommand(executeCommand); }; + // We need to intercept and retry prepareAndExecute if we had a stale prepared statement in the cache that the server already released + const key = BaseConnection.statementKey(options); + const cacheHasKey = this._statements.has(key); + if (executeCommand.onResult) { // Callback mode: traceCallback wraps the callback with tracing lifecycle, or calls through directly when no subscribers are registered const origExecCb = executeCommand.onResult; traceCallback( executeChannel, (wrappedCb) => { - executeCommand.onResult = wrappedCb; + if (cacheHasKey) { + executeCommand.onResult = (err, ...rest) => { + if (err && err.errno === ER_UNKNOWN_STMT_HANDLER) { + const origEmit = executeCommand.emit.bind(executeCommand); + executeCommand.emit = (eventName, ...rest) => { + if (eventName === 'end') { + // Intercept the 'end' event that will be emitted after this 'error' event is emitted + executeCommand.emit = origEmit; + return false; + } + + // In this case there currently will not be any other events emitted before 'end', but leaving + // this here in case that changes in the future... + /* c8 ignore next */ + return origEmit(eventName, ...rest); + }; + + // Listeners may have been added to this execute command, so we re-use it + executeCommand.next = null; + + this._statements.delete(key); + executeCommand.onResult = wrappedCb; + prepareAndExecute(wrappedCb); + return; + } + + wrappedCb(err, ...rest); + }; + } else { + executeCommand.onResult = wrappedCb; + } + prepareAndExecute(wrappedCb); }, 0, @@ -837,7 +873,46 @@ class BaseConnection extends EventEmitter { null, origExecCb ); - } else if (shouldTrace(executeChannel)) { + + return executeCommand; + } + + if (cacheHasKey) { + const origEmit = executeCommand.emit.bind(executeCommand); + executeCommand.emit = (eventName, firstArg, ...rest) => { + if ( + eventName === 'error' && + firstArg && + firstArg.errno === ER_UNKNOWN_STMT_HANDLER + ) { + executeCommand.emit = (eventName, ...rest) => { + if (eventName === 'end') { + // Intercept the 'end' event that will be emitted after this 'error' event is emitted + executeCommand.emit = origEmit; + return false; + } + + // In this case there currently will not be any other events emitted before 'end', but leaving + // this here in case that changes in the future... + /* c8 ignore next */ + return origEmit(eventName, ...rest); + }; + + // Listeners may have been added to this execute command, so we re-use it + executeCommand.next = null; + + this._statements.delete(key); + prepareAndExecute((err) => { + executeCommand.emit('error', err); + }); + return false; + } + + return origEmit(eventName, firstArg, ...rest); + }; + } + + if (shouldTrace(executeChannel)) { // Event-emitter mode: tracePromise wraps the async lifecycle tracePromise( executeChannel, diff --git a/test/integration/connection/test-execute-cached.test.mts b/test/integration/connection/test-execute-cached.test.mts index d8da7abc25..2f226a00a3 100644 --- a/test/integration/connection/test-execute-cached.test.mts +++ b/test/integration/connection/test-execute-cached.test.mts @@ -41,5 +41,122 @@ await describe('Execute Cached', async () => { strict.deepEqual(rows3, [{ test: 126 }]); }); + await it('should discard cached prepared statements that no longer exist on the server and retry automatically (Callback API)', async () => { + // Remove on the server but leave it in the cache + // @ts-expect-error: internal access + connection._statements.get(key).close(); + + // Remember the id + // @ts-expect-error: internal access + const { id: oldStatementId } = connection._statements.get(key); + + // @ts-expect-error: internal access + strict(connection._statements.size === 1); + // @ts-expect-error: internal access + strict(connection._statements.get(key).query === q); + // @ts-expect-error: internal access + strict(connection._statements.get(key).parameters.length === 1); + // @ts-expect-error: internal access + strict(connection._statements.get(key).id === oldStatementId); + + // Callback API + const rows1 = await new Promise((resolve, reject) => { + connection.execute(q, [123], (err, _rows) => + err ? reject(err) : resolve(_rows) + ); + }); + + const rows2 = await new Promise((resolve, reject) => { + connection.execute(q, [124], (err, _rows) => + err ? reject(err) : resolve(_rows) + ); + }); + + // EventEmitter API + const rows3 = await new Promise((resolve, reject) => { + connection.execute(q, [125], (err, _rows) => + err ? reject(err) : resolve(_rows) + ); + }); + + // @ts-expect-error: internal access + strict(connection._statements.size === 1); + // @ts-expect-error: internal access + strict(connection._statements.get(key).query === q); + // @ts-expect-error: internal access + strict(connection._statements.get(key).parameters.length === 1); + // @ts-expect-error: internal access + strict(connection._statements.get(key).id !== oldStatementId); + + strict.deepEqual(rows1, [{ test: 124 }]); + strict.deepEqual(rows2, [{ test: 125 }]); + strict.deepEqual(rows3, [{ test: 126 }]); + }); + + await it('should discard cached prepared statements that no longer exist on the server and retry automatically (EventEmitter API)', async () => { + // Remove on the server but leave it in the cache + // @ts-expect-error: internal access + connection._statements.get(key).close(); + + // Remember the id + // @ts-expect-error: internal access + const { id: oldStatementId } = connection._statements.get(key); + + // @ts-expect-error: internal access + strict(connection._statements.size === 1); + // @ts-expect-error: internal access + strict(connection._statements.get(key).query === q); + // @ts-expect-error: internal access + strict(connection._statements.get(key).parameters.length === 1); + // @ts-expect-error: internal access + strict(connection._statements.get(key).id === oldStatementId); + + const rows1 = await new Promise((resolve, reject) => { + const executeCommand = connection.execute(q, [123]); + const _rows = []; + + executeCommand.once('error', (err) => reject(err)); + executeCommand.on('result', (row) => { + _rows.push(row); + }); + executeCommand.once('end', () => resolve(_rows)); + }); + + const rows2 = await new Promise((resolve, reject) => { + const executeCommand = connection.execute(q, [124]); + const _rows = []; + + executeCommand.once('error', (err) => reject(err)); + executeCommand.on('result', (row) => { + _rows.push(row); + }); + executeCommand.once('end', () => resolve(_rows)); + }); + + const rows3 = await new Promise((resolve, reject) => { + const executeCommand = connection.execute(q, [125]); + const _rows = []; + + executeCommand.once('error', (err) => reject(err)); + executeCommand.on('result', (row) => { + _rows.push(row); + }); + executeCommand.once('end', () => resolve(_rows)); + }); + + // @ts-expect-error: internal access + strict(connection._statements.size === 1); + // @ts-expect-error: internal access + strict(connection._statements.get(key).query === q); + // @ts-expect-error: internal access + strict(connection._statements.get(key).parameters.length === 1); + // @ts-expect-error: internal access + strict(connection._statements.get(key).id !== oldStatementId); + + strict.deepEqual(rows1, [{ test: 124 }]); + strict.deepEqual(rows2, [{ test: 125 }]); + strict.deepEqual(rows3, [{ test: 126 }]); + }); + connection.end(); }); From 73160c177168bf8f8cb966d22b8633f5dd8b2f52 Mon Sep 17 00:00:00 2001 From: Jonathan Horowitz Date: Mon, 23 Mar 2026 23:29:53 -0700 Subject: [PATCH 2/7] fix: typecheck test/integration/connection/test-execute-cached.test.mts --- .../connection/test-execute-cached.test.mts | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/test/integration/connection/test-execute-cached.test.mts b/test/integration/connection/test-execute-cached.test.mts index 2f226a00a3..4171534edf 100644 --- a/test/integration/connection/test-execute-cached.test.mts +++ b/test/integration/connection/test-execute-cached.test.mts @@ -92,7 +92,7 @@ await describe('Execute Cached', async () => { strict.deepEqual(rows2, [{ test: 125 }]); strict.deepEqual(rows3, [{ test: 126 }]); }); - + await it('should discard cached prepared statements that no longer exist on the server and retry automatically (EventEmitter API)', async () => { // Remove on the server but leave it in the cache // @ts-expect-error: internal access @@ -113,10 +113,10 @@ await describe('Execute Cached', async () => { const rows1 = await new Promise((resolve, reject) => { const executeCommand = connection.execute(q, [123]); - const _rows = []; + const _rows: TestRow[] = []; executeCommand.once('error', (err) => reject(err)); - executeCommand.on('result', (row) => { + executeCommand.on('result', (row: TestRow) => { _rows.push(row); }); executeCommand.once('end', () => resolve(_rows)); @@ -124,10 +124,10 @@ await describe('Execute Cached', async () => { const rows2 = await new Promise((resolve, reject) => { const executeCommand = connection.execute(q, [124]); - const _rows = []; + const _rows: TestRow[] = []; executeCommand.once('error', (err) => reject(err)); - executeCommand.on('result', (row) => { + executeCommand.on('result', (row: TestRow) => { _rows.push(row); }); executeCommand.once('end', () => resolve(_rows)); @@ -135,10 +135,10 @@ await describe('Execute Cached', async () => { const rows3 = await new Promise((resolve, reject) => { const executeCommand = connection.execute(q, [125]); - const _rows = []; + const _rows: TestRow[] = []; executeCommand.once('error', (err) => reject(err)); - executeCommand.on('result', (row) => { + executeCommand.on('result', (row: TestRow) => { _rows.push(row); }); executeCommand.once('end', () => resolve(_rows)); From 8997f8bc650c5971de3e46720828d4df3f52c8c4 Mon Sep 17 00:00:00 2001 From: Jonathan Horowitz Date: Mon, 23 Mar 2026 23:37:43 -0700 Subject: [PATCH 3/7] fix: lint test/integration/connection/test-execute-cached.test.mts --- test/integration/connection/test-execute-cached.test.mts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/integration/connection/test-execute-cached.test.mts b/test/integration/connection/test-execute-cached.test.mts index 4171534edf..ac57e578e1 100644 --- a/test/integration/connection/test-execute-cached.test.mts +++ b/test/integration/connection/test-execute-cached.test.mts @@ -92,7 +92,7 @@ await describe('Execute Cached', async () => { strict.deepEqual(rows2, [{ test: 125 }]); strict.deepEqual(rows3, [{ test: 126 }]); }); - + await it('should discard cached prepared statements that no longer exist on the server and retry automatically (EventEmitter API)', async () => { // Remove on the server but leave it in the cache // @ts-expect-error: internal access From 0af6cc81e3a87fe4b4d85c1e1b21c54a817abbd2 Mon Sep 17 00:00:00 2001 From: Jonathan Horowitz Date: Mon, 23 Mar 2026 23:48:13 -0700 Subject: [PATCH 4/7] fix: coverage ignore comments in lib/base/connection.js --- lib/base/connection.js | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/lib/base/connection.js b/lib/base/connection.js index ee34411456..a5d48700d8 100644 --- a/lib/base/connection.js +++ b/lib/base/connection.js @@ -835,10 +835,9 @@ class BaseConnection extends EventEmitter { executeCommand.emit = origEmit; return false; } - + /* c8 ignore next 3 */ // In this case there currently will not be any other events emitted before 'end', but leaving // this here in case that changes in the future... - /* c8 ignore next */ return origEmit(eventName, ...rest); }; @@ -891,10 +890,9 @@ class BaseConnection extends EventEmitter { executeCommand.emit = origEmit; return false; } - + /* c8 ignore next 3 */ // In this case there currently will not be any other events emitted before 'end', but leaving // this here in case that changes in the future... - /* c8 ignore next */ return origEmit(eventName, ...rest); }; From 5ca7ffb95dee90895d90ce9a6eecaadeed3192f5 Mon Sep 17 00:00:00 2001 From: Jonathan Horowitz Date: Tue, 24 Mar 2026 13:55:36 -0700 Subject: [PATCH 5/7] fix: emit end event only once when internal prepare fails in connection.execute --- lib/base/connection.js | 19 ++- .../connection/test-execute-cached.test.mts | 145 +++++++++++++++++- 2 files changed, 154 insertions(+), 10 deletions(-) diff --git a/lib/base/connection.js b/lib/base/connection.js index a5d48700d8..0117b05bdb 100644 --- a/lib/base/connection.js +++ b/lib/base/connection.js @@ -39,6 +39,10 @@ const { connectChannel, } = require('../tracing.js'); +const returnNull = function () { + return null; +}; + let _connectionId = 0; let convertNamedPlaceholders = null; @@ -802,11 +806,8 @@ class BaseConnection extends EventEmitter { const prepareCommand = new Commands.Prepare(options, (err, stmt) => { if (err) { // skip execute command if prepare failed - executeCommand.start = function () { - return null; - }; + executeCommand.next = returnNull; errorCb(err); - executeCommand.emit('end'); return; } executeCommand.statement = stmt; @@ -841,10 +842,13 @@ class BaseConnection extends EventEmitter { return origEmit(eventName, ...rest); }; - // Listeners may have been added to this execute command, so we re-use it + // Listeners may have been added to this execute command, so we reuse it executeCommand.next = null; + // We know that the statement does not exist on the server, so there is no need to close it + executeCommand.statement.close = returnNull; this._statements.delete(key); + executeCommand.onResult = wrappedCb; prepareAndExecute(wrappedCb); return; @@ -896,10 +900,13 @@ class BaseConnection extends EventEmitter { return origEmit(eventName, ...rest); }; - // Listeners may have been added to this execute command, so we re-use it + // Listeners may have been added to this execute command, so we reuse it executeCommand.next = null; + // We know that the statement does not exist on the server, so there is no need to close it + executeCommand.statement.close = returnNull; this._statements.delete(key); + prepareAndExecute((err) => { executeCommand.emit('error', err); }); diff --git a/test/integration/connection/test-execute-cached.test.mts b/test/integration/connection/test-execute-cached.test.mts index ac57e578e1..3bcacd1c81 100644 --- a/test/integration/connection/test-execute-cached.test.mts +++ b/test/integration/connection/test-execute-cached.test.mts @@ -1,5 +1,6 @@ -import type { RowDataPacket } from '../../../index.js'; +import type { QueryError, RowDataPacket } from '../../../index.js'; import { describe, it, strict } from 'poku'; +import { ER_PARSE_ERROR } from '../../../lib/constants/errors.js'; import { createConnection } from '../../common.test.mjs'; type TestRow = RowDataPacket & { test: number }; @@ -10,7 +11,7 @@ await describe('Execute Cached', async () => { const q = 'select 1 + ? as test'; const key = `undefined/undefined/undefined${q}`; - await it('should cache prepared statements', async () => { + await it('should cache prepared statements (Callback API)', async () => { const rows1 = await new Promise((resolve, reject) => { connection.execute(q, [123], (err, _rows) => err ? reject(err) : resolve(_rows) @@ -59,7 +60,6 @@ await describe('Execute Cached', async () => { // @ts-expect-error: internal access strict(connection._statements.get(key).id === oldStatementId); - // Callback API const rows1 = await new Promise((resolve, reject) => { connection.execute(q, [123], (err, _rows) => err ? reject(err) : resolve(_rows) @@ -72,7 +72,6 @@ await describe('Execute Cached', async () => { ); }); - // EventEmitter API const rows3 = await new Promise((resolve, reject) => { connection.execute(q, [125], (err, _rows) => err ? reject(err) : resolve(_rows) @@ -93,6 +92,96 @@ await describe('Execute Cached', async () => { strict.deepEqual(rows3, [{ test: 126 }]); }); + await it('should properly forward prepare command errors to the executeCommand event listeners (Callback API)', async () => { + // Remove on the server but leave it in the cache + // @ts-expect-error: internal access + connection._statements.get(key).close(); + + // @ts-expect-error: internal access + strict(connection._statements.size === 1); + // @ts-expect-error: internal access + strict(connection._statements.get(key).query === q); + // @ts-expect-error: internal access + strict(connection._statements.get(key).parameters.length === 1); + + // Intercept addCommand so we can access the prepare command instance + let numAddCommandCalls = 0; + // @ts-expect-error: internal access + const origAddCommand = connection.addCommand.bind(connection); + // @ts-expect-error: internal access + connection.addCommand = function (command: { query: string }) { + if (++numAddCommandCalls === 3) { + // The third command added will be the prepare command issued while retrying the query + command.query = 'ASDF'; // Force a failure with an invalid query + + // Restore the original addCommand function to the connection + // @ts-expect-error: internal access + connection.addCommand = origAddCommand; + } + + origAddCommand(command); + }; + + const error = await new Promise((resolve, reject) => { + connection.execute(q, [123], (err, _rows) => + err ? resolve(err) : reject(_rows) + ); + }); + + // @ts-expect-error: internal access + strict(connection._statements.size === 0); + // @ts-expect-error: internal access + strict(!connection._statements.has(key)); + + strict(error.errno === ER_PARSE_ERROR); + }); + + await it('should cache prepared statements (EventEmitter API)', async () => { + const rows1 = await new Promise((resolve, reject) => { + const executeCommand = connection.execute(q, [123]); + const _rows: TestRow[] = []; + + executeCommand.once('error', (err) => reject(err)); + executeCommand.on('result', (row: TestRow) => { + _rows.push(row); + }); + executeCommand.once('end', () => resolve(_rows)); + }); + + const rows2 = await new Promise((resolve, reject) => { + const executeCommand = connection.execute(q, [124]); + const _rows: TestRow[] = []; + + executeCommand.once('error', (err) => reject(err)); + executeCommand.on('result', (row: TestRow) => { + _rows.push(row); + }); + executeCommand.once('end', () => resolve(_rows)); + }); + + const rows3 = await new Promise((resolve, reject) => { + const executeCommand = connection.execute(q, [125]); + const _rows: TestRow[] = []; + + executeCommand.once('error', (err) => reject(err)); + executeCommand.on('result', (row: TestRow) => { + _rows.push(row); + }); + executeCommand.once('end', () => resolve(_rows)); + }); + + // @ts-expect-error: internal access + strict(connection._statements.size === 1); + // @ts-expect-error: internal access + strict(connection._statements.get(key).query === q); + // @ts-expect-error: internal access + strict(connection._statements.get(key).parameters.length === 1); + + strict.deepEqual(rows1, [{ test: 124 }]); + strict.deepEqual(rows2, [{ test: 125 }]); + strict.deepEqual(rows3, [{ test: 126 }]); + }); + await it('should discard cached prepared statements that no longer exist on the server and retry automatically (EventEmitter API)', async () => { // Remove on the server but leave it in the cache // @ts-expect-error: internal access @@ -158,5 +247,53 @@ await describe('Execute Cached', async () => { strict.deepEqual(rows3, [{ test: 126 }]); }); + await it('should properly forward prepare command errors to the executeCommand event listeners (EventEmitter API)', async () => { + // Remove on the server but leave it in the cache + // @ts-expect-error: internal access + connection._statements.get(key).close(); + + // @ts-expect-error: internal access + strict(connection._statements.size === 1); + // @ts-expect-error: internal access + strict(connection._statements.get(key).query === q); + // @ts-expect-error: internal access + strict(connection._statements.get(key).parameters.length === 1); + + // Intercept addCommand so we can access the prepare command instance + let numAddCommandCalls = 0; + // @ts-expect-error: internal access + const origAddCommand = connection.addCommand.bind(connection); + // @ts-expect-error: internal access + connection.addCommand = function (command: { query: string }) { + if (++numAddCommandCalls === 3) { + // The third command added will be the prepare command issued while retrying the query + command.query = 'ASDF'; // Force a failure with an invalid query + + // Restore the original addCommand function to the connection + // @ts-expect-error: internal access + connection.addCommand = origAddCommand; + } + + origAddCommand(command); + }; + + const errors = await new Promise((resolve, reject) => { + const executeCommand = connection.execute(q, [123]); + + const errors: QueryError[] = []; + executeCommand.on('error', (err) => errors.push(err)); + executeCommand.once('result', (row) => reject(row)); + executeCommand.once('end', () => resolve(errors)); + }); + + // @ts-expect-error: internal access + strict(connection._statements.size === 0); + // @ts-expect-error: internal access + strict(!connection._statements.has(key)); + + strict(errors.length === 1); + strict(errors[0].errno === ER_PARSE_ERROR); + }); + connection.end(); }); From 0a4aaa24ec181b219b05534c30d0beb88f6942ff Mon Sep 17 00:00:00 2001 From: Jonathan Horowitz Date: Fri, 10 Apr 2026 10:28:45 -0700 Subject: [PATCH 6/7] Update lib/base/connection.js MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Weslley Araújo <46850407+wellwelwel@users.noreply.github.com> --- lib/base/connection.js | 1 - 1 file changed, 1 deletion(-) diff --git a/lib/base/connection.js b/lib/base/connection.js index 0117b05bdb..9725b646d0 100644 --- a/lib/base/connection.js +++ b/lib/base/connection.js @@ -894,7 +894,6 @@ class BaseConnection extends EventEmitter { executeCommand.emit = origEmit; return false; } - /* c8 ignore next 3 */ // In this case there currently will not be any other events emitted before 'end', but leaving // this here in case that changes in the future... return origEmit(eventName, ...rest); From debdc7684ea6de0fbf95f9c3e6193ee460fa3c20 Mon Sep 17 00:00:00 2001 From: Jonathan Horowitz Date: Fri, 10 Apr 2026 10:28:56 -0700 Subject: [PATCH 7/7] Update lib/base/connection.js MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Weslley Araújo <46850407+wellwelwel@users.noreply.github.com> --- lib/base/connection.js | 1 - 1 file changed, 1 deletion(-) diff --git a/lib/base/connection.js b/lib/base/connection.js index 9725b646d0..62d84796f0 100644 --- a/lib/base/connection.js +++ b/lib/base/connection.js @@ -836,7 +836,6 @@ class BaseConnection extends EventEmitter { executeCommand.emit = origEmit; return false; } - /* c8 ignore next 3 */ // In this case there currently will not be any other events emitted before 'end', but leaving // this here in case that changes in the future... return origEmit(eventName, ...rest);