Transaction rejects with rethrown error - fixes #289

cjs/src/index.js

@@ -198,22 +198,25 @@ function Postgres(a, b) {
     async function scope(c, fn, name) {
       const sql = Sql(handler, true)
       sql.savepoint = savepoint
-      let errored
+      let uncaughtError
       name && await sql`savepoint ${ sql(name) }`
       try {
         const result = await new Promise((resolve, reject) => {
-          errored = reject
           const x = fn(sql)
           Promise.resolve(Array.isArray(x) ? Promise.all(x) : x).then(resolve, reject)
         })
+
+        if (uncaughtError)
+          throw uncaughtError
+
         !name && await sql`commit`
         return result
       } catch (e) {
         await (name
           ? sql`rollback to ${ sql(name) }`
           : sql`rollback`
         )
-        throw e
+        throw e instanceof PostgresError && e.code === '25P02' && uncaughtError || e
       }
 
       function savepoint(name, fn) {
@@ -225,7 +228,7 @@ function Postgres(a, b) {
       }
 
       function handler(q) {
-        errored && q.catch(errored)
+        q.catch(e => uncaughtError || (uncaughtError = e))
         c.state === 'full'
           ? queries.push(q)
           : c.execute(q) || (c.state = 'full', full.push(c))

cjs/src/query.js

@@ -31,17 +31,19 @@ const Query = module.exports.Query = class Query extends Promise {
     this.executed = false
     this.signature = ''
 
-    this[originError] = handler.debug || !this.tagged
+    this[originError] = this.handler.debug
       ? new Error()
-      : cachedError(this.strings)
+      : this.tagged && cachedError(this.strings)
   }
 
   get origin() {
-    return this.handler.debug || !this.tagged
+    return this.handler.debug
       ? this[originError].stack
-      : originStackCache.has(this.strings)
-        ? originStackCache.get(this.strings)
-        : originStackCache.set(this.strings, this[originError].stack).get(this.strings)
+      : this.tagged
+        ? originStackCache.has(this.strings)
+          ? originStackCache.get(this.strings)
+          : originStackCache.set(this.strings, this[originError].stack).get(this.strings)
+        : ''
   }
 
   static get [Symbol.species]() {

cjs/src/subscribe.js

@@ -8,6 +8,7 @@ module.exports = Subscribe;function Subscribe(postgres, options) {
 
     options.max = 1
     options.onclose = onclose
+    options.fetch_types = false
     options.connection = {
       ...options.connection,
       replication: 'database'

cjs/tests/index.js

@@ -254,6 +254,17 @@ t('Uncaught transaction request errors bubbles to transaction', async() => [
   )).catch(e => e.code))
 ])
 
+t('Transaction rejects with rethrown error', async() => [
+  'WAT',
+  await sql.begin(async sql => {
+    try {
+      await sql`select exception`
+    } catch (ex) {
+      throw new Error('WAT')
+    }
+  }).catch(e => e.message)
+])
+
 t('Parallel transactions', async() => {
   await sql`create table test (a int)`
   return ['11', (await Promise.all([

deno/src/index.js

@@ -199,22 +199,25 @@ function Postgres(a, b) {
     async function scope(c, fn, name) {
       const sql = Sql(handler, true)
       sql.savepoint = savepoint
-      let errored
+      let uncaughtError
       name && await sql`savepoint ${ sql(name) }`
       try {
         const result = await new Promise((resolve, reject) => {
-          errored = reject
           const x = fn(sql)
           Promise.resolve(Array.isArray(x) ? Promise.all(x) : x).then(resolve, reject)
         })
+
+        if (uncaughtError)
+          throw uncaughtError
+
         !name && await sql`commit`
         return result
       } catch (e) {
         await (name
           ? sql`rollback to ${ sql(name) }`
           : sql`rollback`
         )
-        throw e
+        throw e instanceof PostgresError && e.code === '25P02' && uncaughtError || e
       }
 
       function savepoint(name, fn) {
@@ -226,7 +229,7 @@ function Postgres(a, b) {
       }
 
       function handler(q) {
-        errored && q.catch(errored)
+        q.catch(e => uncaughtError || (uncaughtError = e))
         c.state === 'full'
           ? queries.push(q)
           : c.execute(q) || (c.state = 'full', full.push(c))

deno/src/query.js

@@ -31,17 +31,19 @@ export class Query extends Promise {
     this.executed = false
     this.signature = ''
 
-    this[originError] = handler.debug || !this.tagged
+    this[originError] = this.handler.debug
       ? new Error()
-      : cachedError(this.strings)
+      : this.tagged && cachedError(this.strings)
   }
 
   get origin() {
-    return this.handler.debug || !this.tagged
+    return this.handler.debug
       ? this[originError].stack
-      : originStackCache.has(this.strings)
-        ? originStackCache.get(this.strings)
-        : originStackCache.set(this.strings, this[originError].stack).get(this.strings)
+      : this.tagged
+        ? originStackCache.has(this.strings)
+          ? originStackCache.get(this.strings)
+          : originStackCache.set(this.strings, this[originError].stack).get(this.strings)
+        : ''
   }
 
   static get [Symbol.species]() {

deno/tests/index.js

@@ -255,6 +255,17 @@ t('Uncaught transaction request errors bubbles to transaction', async() => [
   )).catch(e => e.code))
 ])
 
+t('Transaction rejects with rethrown error', async() => [
+  'WAT',
+  await sql.begin(async sql => {
+    try {
+      await sql`select exception`
+    } catch (ex) {
+      throw new Error('WAT')
+    }
+  }).catch(e => e.message)
+])
+
 t('Parallel transactions', async() => {
   await sql`create table test (a int)`
   return ['11', (await Promise.all([

src/index.js

@@ -198,22 +198,25 @@ function Postgres(a, b) {
     async function scope(c, fn, name) {
       const sql = Sql(handler, true)
       sql.savepoint = savepoint
-      let errored
+      let uncaughtError
       name && await sql`savepoint ${ sql(name) }`
       try {
         const result = await new Promise((resolve, reject) => {
-          errored = reject
           const x = fn(sql)
           Promise.resolve(Array.isArray(x) ? Promise.all(x) : x).then(resolve, reject)
         })
+
+        if (uncaughtError)
+          throw uncaughtError
+
         !name && await sql`commit`
         return result
       } catch (e) {
         await (name
           ? sql`rollback to ${ sql(name) }`
           : sql`rollback`
         )
-        throw e
+        throw e instanceof PostgresError && e.code === '25P02' && uncaughtError || e
       }
 
       function savepoint(name, fn) {
@@ -225,7 +228,7 @@ function Postgres(a, b) {
       }
 
       function handler(q) {
-        errored && q.catch(errored)
+        q.catch(e => uncaughtError || (uncaughtError = e))
         c.state === 'full'
           ? queries.push(q)
           : c.execute(q) || (c.state = 'full', full.push(c))

tests/index.js

@@ -254,6 +254,17 @@ t('Uncaught transaction request errors bubbles to transaction', async() => [
   )).catch(e => e.code))
 ])
 
+t('Transaction rejects with rethrown error', async() => [
+  'WAT',
+  await sql.begin(async sql => {
+    try {
+      await sql`select exception`
+    } catch (ex) {
+      throw new Error('WAT')
+    }
+  }).catch(e => e.message)
+])
+
 t('Parallel transactions', async() => {
   await sql`create table test (a int)`
   return ['11', (await Promise.all([