src/QueryStream.ts

@@ -63,15 +63,10 @@ export class QueryStream extends Readable {
     this.cursor.submit(connection);
   }
 
-  public close (callback: Function) {
+  public _destroy (error: Error, callback: Function) {
     this._closed = true;
 
-    // eslint-disable-next-line unicorn/consistent-function-scoping
-    const close = () => {
-      this.emit('close');
-    };
-
-    this.cursor.close(callback || close);
+    this.cursor.close(callback);
   }
 
   public _read (size: number) {

src/connectionMethods/stream.ts

@@ -1,4 +1,6 @@
-import type Stream from 'stream';
+import {
+  type Readable,
+} from 'stream';
 import through from 'through2';
 import {
   QueryStream,
@@ -21,7 +23,7 @@ export const stream: InternalStreamFunction = async (connectionLogger, connectio
     async (finalConnection, finalSql, finalValues, executionContext, actualQuery) => {
       const query = new QueryStream(finalSql, finalValues, options);
 
-      const queryStream: Stream = finalConnection.query(query);
+      const queryStream: Readable = finalConnection.query(query);
 
       const rowTransformers: Array<NonNullable<Interceptor['transformRow']>> = [];
 
@@ -66,6 +68,10 @@ export const stream: InternalStreamFunction = async (connectionLogger, connectio
 
         // Invoked if stream is destroyed using transformedStream.destroy().
         transformedStream.on('close', () => {
+          if (!queryStream.destroyed) {
+            queryStream.destroy();
+          }
+
           resolve({
             command: 'SELECT',
             fields: [],
@@ -75,6 +81,10 @@ export const stream: InternalStreamFunction = async (connectionLogger, connectio
           });
         });
 
+        transformedStream.on('error', (error: Error) => {
+          queryStream.destroy(error);
+        });
+
         streamHandler(transformedStream);
       });
     },

test/slonik/integration/pg.ts

@@ -337,3 +337,39 @@ test('copies from binary stream', async (t) => {
 
   await pool.end();
 });
+
+test('frees connection after destroying a stream', async (t) => {
+  const pool = await createPool(t.context.dsn);
+
+  await pool.stream(sql`
+    SELECT * FROM GENERATE_SERIES(1, 100)
+  `, (stream) => {
+    stream.destroy();
+  });
+
+  t.deepEqual(await pool.anyFirst(sql`
+    SELECT TRUE
+  `), [
+    true,
+  ]);
+
+  await pool.end();
+});
+
+test('frees connection after destroying a stream with an error', async (t) => {
+  const pool = await createPool(t.context.dsn);
+
+  await pool.stream(sql`
+    SELECT * FROM GENERATE_SERIES(1, 100)
+  `, (stream) => {
+    stream.destroy(new Error('Foo'));
+  });
+
+  t.deepEqual(await pool.anyFirst(sql`
+    SELECT TRUE
+  `), [
+    true,
+  ]);
+
+  await pool.end();
+});