Fix sql.begin() sending BEGIN on an unreserved connection

src/connection.js

@@ -170,11 +170,16 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
         : (query = q, query.active = true)
 
       build(q)
-      return write(toBuffer(q))
+      const written = write(toBuffer(q))
+      // Run the hook whenever the bytes were written, even if the pipeline is
+      // full or socket.write() reported backpressure. sql.begin() relies on it
+      // to reserve the connection, and its falsy return keeps the connection
+      // out of the busy queue until BEGIN completes.
+      return (!q.options.onexecute || q.options.onexecute(connection))
+        && written
         && !q.describeFirst
         && !q.cursorFn
         && sent.length < max_pipeline
-        && (!q.options.onexecute || q.options.onexecute(connection))
     } catch (error) {
       sent.length === 0 && write(Sync)
       errored(error)

tests/index.js

@@ -300,6 +300,33 @@ t('Many transactions at beginning of connection', async() => {
   return [100, xs.length]
 })
 
+t('Transaction at pipeline boundary is reserved', async() => {
+  const sql = postgres({ ...options, max: 2, max_pipeline: 1, fetch_types: false })
+  await Promise.all([sql`select 1`, sql`select 1`])
+  const inflight = [sql`select pg_sleep(0.1)`.execute(), sql`select pg_sleep(0.1)`.execute()]
+  const x = await sql.begin(sql => sql`select 1 as x`).then(x => x[0].x, x => x.code)
+  await Promise.all(inflight)
+  return [1, x, await sql.end()]
+})
+
+t('Transaction is reserved with pipelining disabled', async() => {
+  const sql = postgres({ ...options, max: 2, max_pipeline: 0, fetch_types: false })
+  const x = await sql.begin(sql => sql`select 1 as x`).then(x => x[0].x, x => x.code)
+  return [1, x, await sql.end()]
+})
+
+t('Query issued while BEGIN is in flight does not join the transaction', async() => {
+  const sql = postgres({ ...options, max: 1, fetch_types: false })
+  await sql`create table test (a int)`
+  const tx = sql.begin(async sql => {
+    await sql`select 1`
+    throw new Error('rollback')
+  }).catch(() => {})
+  const insert = sql`insert into test values (1)`
+  await Promise.all([tx, insert])
+  return [1, (await sql`select count(*)::int as n from test`)[0].n, await sql`drop table test`, await sql.end()]
+})
+
 t('Transactions array', async() => {
   await sql`create table test (a int)`