feat: auto purge stale prepared statements from cache and retry query

Author: Invader444

lib/base/connection.js

@@ -836,6 +836,9 @@ class BaseConnection extends EventEmitter {
                     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);
                 };
 
@@ -889,6 +892,9 @@ class BaseConnection extends EventEmitter {
               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);
           };
 

test/integration/connection/test-execute-cached.test.mts

@@ -41,7 +41,7 @@ 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', async () => {
+  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();
@@ -59,6 +59,7 @@ await describe('Execute Cached', async () => {
     // @ts-expect-error: internal access
     strict(connection._statements.get(key).id === oldStatementId);
 
+    // Callback API
     const rows1 = await new Promise<TestRow[]>((resolve, reject) => {
       connection.execute<TestRow[]>(q, [123], (err, _rows) =>
         err ? reject(err) : resolve(_rows)
@@ -71,6 +72,7 @@ await describe('Execute Cached', async () => {
       );
     });
 
+    // EventEmitter API
     const rows3 = await new Promise<TestRow[]>((resolve, reject) => {
       connection.execute<TestRow[]>(q, [125], (err, _rows) =>
         err ? reject(err) : resolve(_rows)
@@ -91,5 +93,70 @@ 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 (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<TestRow[]>((resolve, reject) => {
+      const executeCommand = connection.execute<TestRow[]>(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<TestRow[]>((resolve, reject) => {
+      const executeCommand = connection.execute<TestRow[]>(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<TestRow[]>((resolve, reject) => {
+      const executeCommand = connection.execute<TestRow[]>(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();
 });