Fix swallowed cursor error between batches in async iterator

cf/src/query.js

@@ -84,9 +84,15 @@ export class Query extends Promise {
       return (this.cursorFn = fn, this)
 
     let prev
+      , error = null
     return {
       [Symbol.asyncIterator]: () => ({
         next: () => {
+          if (error) {
+            const err = error
+            error = null
+            return Promise.reject(err)
+          }
           if (this.executed && !this.active)
             return { done: true }
 
@@ -97,7 +103,7 @@ export class Query extends Promise {
               return new Promise(r => prev = r)
             }
             this.resolve = () => (this.active = false, resolve({ done: true }))
-            this.reject = x => (this.active = false, reject(x))
+            this.reject = x => (this.active = false, error = x, reject(x))
           })
           this.execute()
           return promise

cjs/src/query.js

@@ -84,9 +84,15 @@ const Query = module.exports.Query = class Query extends Promise {
       return (this.cursorFn = fn, this)
 
     let prev
+      , error = null
     return {
       [Symbol.asyncIterator]: () => ({
         next: () => {
+          if (error) {
+            const err = error
+            error = null
+            return Promise.reject(err)
+          }
           if (this.executed && !this.active)
             return { done: true }
 
@@ -97,7 +103,7 @@ const Query = module.exports.Query = class Query extends Promise {
               return new Promise(r => prev = r)
             }
             this.resolve = () => (this.active = false, resolve({ done: true }))
-            this.reject = x => (this.active = false, reject(x))
+            this.reject = x => (this.active = false, error = x, reject(x))
           })
           this.execute()
           return promise

cjs/tests/index.js

@@ -1457,6 +1457,21 @@ t('Async Iterator Cursor custom with less results than batch size', async() => {
   return ['20', order.join(',')]
 })
 
+t('Async Iterator Cursor surfaces error arriving between batches', async() => {
+  const query = sql`select * from generate_series(1,5) as x`
+  const iterator = query.cursor(1)[Symbol.asyncIterator]()
+  await iterator.next()
+  query.reject(new Error('between-batch'))
+  let caught
+  try {
+    await iterator.next()
+  } catch (err) {
+    caught = err.message
+  }
+  await iterator.return()
+  return ['between-batch', caught]
+})
+
 t('Transform row', async() => {
   const sql = postgres({
     ...options,

deno/src/query.js

@@ -84,9 +84,15 @@ export class Query extends Promise {
       return (this.cursorFn = fn, this)
 
     let prev
+      , error = null
     return {
       [Symbol.asyncIterator]: () => ({
         next: () => {
+          if (error) {
+            const err = error
+            error = null
+            return Promise.reject(err)
+          }
           if (this.executed && !this.active)
             return { done: true }
 
@@ -97,7 +103,7 @@ export class Query extends Promise {
               return new Promise(r => prev = r)
             }
             this.resolve = () => (this.active = false, resolve({ done: true }))
-            this.reject = x => (this.active = false, reject(x))
+            this.reject = x => (this.active = false, error = x, reject(x))
           })
           this.execute()
           return promise

deno/tests/index.js

@@ -1459,6 +1459,21 @@ t('Async Iterator Cursor custom with less results than batch size', async() => {
   return ['20', order.join(',')]
 })
 
+t('Async Iterator Cursor surfaces error arriving between batches', async() => {
+  const query = sql`select * from generate_series(1,5) as x`
+  const iterator = query.cursor(1)[Symbol.asyncIterator]()
+  await iterator.next()
+  query.reject(new Error('between-batch'))
+  let caught
+  try {
+    await iterator.next()
+  } catch (err) {
+    caught = err.message
+  }
+  await iterator.return()
+  return ['between-batch', caught]
+})
+
 t('Transform row', async() => {
   const sql = postgres({
     ...options,

src/query.js

@@ -84,9 +84,15 @@ export class Query extends Promise {
       return (this.cursorFn = fn, this)
 
     let prev
+      , error = null
     return {
       [Symbol.asyncIterator]: () => ({
         next: () => {
+          if (error) {
+            const err = error
+            error = null
+            return Promise.reject(err)
+          }
           if (this.executed && !this.active)
             return { done: true }
 
@@ -97,7 +103,7 @@ export class Query extends Promise {
               return new Promise(r => prev = r)
             }
             this.resolve = () => (this.active = false, resolve({ done: true }))
-            this.reject = x => (this.active = false, reject(x))
+            this.reject = x => (this.active = false, error = x, reject(x))
           })
           this.execute()
           return promise

tests/index.js

@@ -1457,6 +1457,21 @@ t('Async Iterator Cursor custom with less results than batch size', async() => {
   return ['20', order.join(',')]
 })
 
+t('Async Iterator Cursor surfaces error arriving between batches', async() => {
+  const query = sql`select * from generate_series(1,5) as x`
+  const iterator = query.cursor(1)[Symbol.asyncIterator]()
+  await iterator.next()
+  query.reject(new Error('between-batch'))
+  let caught
+  try {
+    await iterator.next()
+  } catch (err) {
+    caught = err.message
+  }
+  await iterator.return()
+  return ['between-batch', caught]
+})
+
 t('Transform row', async() => {
   const sql = postgres({
     ...options,