feat(sql) transactions, savepoints, connection pooling and reserve (#16381)

packages/bun-types/bun.d.ts

@@ -1995,6 +1995,298 @@ declare module "bun" {
      */
     stat(path: string, options?: S3Options): Promise<S3Stats>;
   };
+  /**
+   * Configuration options for SQL client connection and behavior
+   *  @example
+   * const config: SQLOptions = {
+   *   host: 'localhost',
+   *   port: 5432,
+   *   user: 'dbuser',
+   *   password: 'secretpass',
+   *   database: 'myapp',
+   *   idleTimeout: 30000,
+   *   max: 20,
+   *   onconnect: (client) => {
+   *     console.log('Connected to database');
+   *   }
+   * };
+   */
+  type SQLOptions = {
+    /** Connection URL (can be string or URL object) */
+    url: URL | string;
+    /** Database server hostname */
+    host: string;
+    /** Database server port number */
+    port: number | string;
+    /** Database user for authentication */
+    user: string;
+    /** Database password for authentication */
+    password: string;
+    /** Name of the database to connect to */
+    database: string;
+    /** Database adapter/driver to use */
+    adapter: string;
+    /** Maximum time in milliseconds to wait for connection to become available */
+    idleTimeout: number;
+    /** Maximum time in milliseconds to wait when establishing a connection */
+    connectionTimeout: number;
+    /** Maximum lifetime in milliseconds of a connection */
+    maxLifetime: number;
+    /** Whether to use TLS/SSL for the connection */
+    tls: boolean;
+    /** Callback function executed when a connection is established */
+    onconnect: (client: SQL) => void;
+    /** Callback function executed when a connection is closed */
+    onclose: (client: SQL) => void;
+    /** Maximum number of connections in the pool */
+    max: number;
+  };
+
+  /**
+   * Represents a SQL query that can be executed, with additional control methods
+   * Extends Promise to allow for async/await usage
+   */
+  interface SQLQuery extends Promise<any> {
+    /** Indicates if the query is currently executing */
+    active: boolean;
+    /** Indicates if the query has been cancelled */
+    cancelled: boolean;
+    /** Cancels the executing query */
+    cancel(): SQLQuery;
+    /** Executes the query */
+    execute(): SQLQuery;
+    /** Returns the raw query result */
+    raw(): SQLQuery;
+    /** Returns only the values from the query result */
+    values(): SQLQuery;
+  }
+
+  /**
+   * Callback function type for transaction contexts
+   * @param sql Function to execute SQL queries within the transaction
+   */
+  type SQLContextCallback = (sql: (strings: string, ...values: any[]) => SQLQuery | Array<SQLQuery>) => Promise<any>;
+
+  /**
+   * Main SQL client interface providing connection and transaction management
+   */
+  interface SQL {
+    /** Creates a new SQL client instance
+     * @example
+     * const sql = new SQL("postgres://localhost:5432/mydb");
+     * const sql = new SQL(new URL("postgres://localhost:5432/mydb"));
+     */
+    new (connectionString: string | URL): SQL;
+    /** Creates a new SQL client instance with options
+     * @example
+     * const sql = new SQL("postgres://localhost:5432/mydb", { idleTimeout: 1000 });
+     */
+    new (connectionString: string | URL, options: SQLOptions): SQL;
+    /** Creates a new SQL client instance with options
+     * @example
+     * const sql = new SQL({ url: "postgres://localhost:5432/mydb", idleTimeout: 1000 });
+     */
+    new (options?: SQLOptions): SQL;
+    /** Executes a SQL query using template literals
+     * @example
+     * const [user] = await sql`select * from users where id = ${1}`;
+     */
+    (strings: string, ...values: any[]): SQLQuery;
+    /** Commits a distributed transaction also know as prepared transaction in postgres or XA transaction in MySQL
+     * @example
+     * await sql.commitDistributed("my_distributed_transaction");
+     */
+    commitDistributed(name: string): Promise<undefined>;
+    /** Rolls back a distributed transaction also know as prepared transaction in postgres or XA transaction in MySQL
+     * @example
+     * await sql.rollbackDistributed("my_distributed_transaction");
+     */
+    rollbackDistributed(name: string): Promise<undefined>;
+    /** Waits for the database connection to be established
+     * @example
+     * await sql.connect();
+     */
+    connect(): Promise<SQL>;
+    /** Closes the database connection with optional timeout in seconds
+     * @example
+     * await sql.close({ timeout: 1 });
+     */
+    close(options?: { timeout?: number }): Promise<undefined>;
+    /** Closes the database connection with optional timeout in seconds
+     * @alias close
+     * @example
+     * await sql.end({ timeout: 1 });
+     */
+    end(options?: { timeout?: number }): Promise<undefined>;
+    /** Flushes any pending operations */
+    flush(): void;
+    /**  The reserve method pulls out a connection from the pool, and returns a client that wraps the single connection.
+     *   This can be used for running queries on an isolated connection.
+     *   Calling reserve in a reserved Sql will return a new reserved connection, not the same connection (behavior matches postgres package).
+     * @example
+     * const reserved = await sql.reserve();
+     * await reserved`select * from users`;
+     * await reserved.release();
+     * // with in a production scenario would be something more like
+     * const reserved = await sql.reserve();
+     * try {
+     *   // ... queries
+     * } finally {
+     *   await reserved.release();
+     * }
+     * //To make it simpler bun supportsSymbol.dispose and Symbol.asyncDispose
+     * {
+     * // always release after context (safer)
+     * using reserved = await sql.reserve()
+     * await reserved`select * from users`
+     * }
+     */
+    reserve(): Promise<ReservedSQL>;
+    /** Begins a new transaction
+     * Will reserve a connection for the transaction and supply a scoped sql instance for all transaction uses in the callback function. sql.begin will resolve with the returned value from the callback function.
+     * BEGIN is automatically sent with the optional options, and if anything fails ROLLBACK will be called so the connection can be released and execution can continue.
+     * @example
+     * const [user, account] = await sql.begin(async sql => {
+     *   const [user] = await sql`
+     *     insert into users (
+     *       name
+     *     ) values (
+     *       'Murray'
+     *     )
+     *     returning *
+     *   `
+     *   const [account] = await sql`
+     *     insert into accounts (
+     *       user_id
+     *     ) values (
+     *       ${ user.user_id }
+     *     )
+     *     returning *
+     *   `
+     *   return [user, account]
+     * })
+     */
+    begin(fn: SQLContextCallback): Promise<any>;
+    /** Begins a new transaction with options
+     * Will reserve a connection for the transaction and supply a scoped sql instance for all transaction uses in the callback function. sql.begin will resolve with the returned value from the callback function.
+     * BEGIN is automatically sent with the optional options, and if anything fails ROLLBACK will be called so the connection can be released and execution can continue.
+     * @example
+     * const [user, account] = await sql.begin("read write", async sql => {
+     *   const [user] = await sql`
+     *     insert into users (
+     *       name
+     *     ) values (
+     *       'Murray'
+     *     )
+     *     returning *
+     *   `
+     *   const [account] = await sql`
+     *     insert into accounts (
+     *       user_id
+     *     ) values (
+     *       ${ user.user_id }
+     *     )
+     *     returning *
+     *   `
+     *   return [user, account]
+     * })
+     */
+    begin(options: string, fn: SQLContextCallback): Promise<any>;
+    /** Alternative method to begin a transaction
+     * Will reserve a connection for the transaction and supply a scoped sql instance for all transaction uses in the callback function. sql.transaction will resolve with the returned value from the callback function.
+     * BEGIN is automatically sent with the optional options, and if anything fails ROLLBACK will be called so the connection can be released and execution can continue.
+     * @alias begin
+     * @example
+     * const [user, account] = await sql.transaction(async sql => {
+     *   const [user] = await sql`
+     *     insert into users (
+     *       name
+     *     ) values (
+     *       'Murray'
+     *     )
+     *     returning *
+     *   `
+     *   const [account] = await sql`
+     *     insert into accounts (
+     *       user_id
+     *     ) values (
+     *       ${ user.user_id }
+     *     )
+     *     returning *
+     *   `
+     *   return [user, account]
+     * })
+     */
+    transaction(fn: SQLContextCallback): Promise<any>;
+    /** Alternative method to begin a transaction with options
+     * Will reserve a connection for the transaction and supply a scoped sql instance for all transaction uses in the callback function. sql.transaction will resolve with the returned value from the callback function.
+     * BEGIN is automatically sent with the optional options, and if anything fails ROLLBACK will be called so the connection can be released and execution can continue.
+     * @alias begin
+     * @example
+     * const [user, account] = await sql.transaction("read write", async sql => {
+     *   const [user] = await sql`
+     *     insert into users (
+     *       name
+     *     ) values (
+     *       'Murray'
+     *     )
+     *     returning *
+     *   `
+     *   const [account] = await sql`
+     *     insert into accounts (
+     *       user_id
+     *     ) values (
+     *       ${ user.user_id }
+     *     )
+     *     returning *
+     *   `
+     *   return [user, account]
+     * })
+     */
+    transaction(options: string, fn: SQLContextCallback): Promise<any>;
+    /** Begins a distributed transaction
+     * Also know as Two-Phase Commit, in a distributed transaction, Phase 1 involves the coordinator preparing nodes by ensuring data is written and ready to commit, while Phase 2 finalizes with nodes committing or rolling back based on the coordinator's decision, ensuring durability and releasing locks.
+     * In PostgreSQL and MySQL distributed transactions persist beyond the original session, allowing privileged users or coordinators to commit/rollback them, ensuring support for distributed transactions, recovery, and administrative tasks.
+     * beginDistributed will automatic rollback if any exception are not caught, and you can commit and rollback later if everything goes well.
+     * PostgreSQL natively supports distributed transactions using PREPARE TRANSACTION, while MySQL uses XA Transactions, and MSSQL also supports distributed/XA transactions. However, in MSSQL, distributed transactions are tied to the original session, the DTC coordinator, and the specific connection.
+     * These transactions are automatically committed or rolled back following the same rules as regular transactions, with no option for manual intervention from other sessions, in MSSQL distributed transactions are used to coordinate transactions using Linked Servers.
+     * @example
+     * await sql.beginDistributed("numbers", async sql => {
+     *   await sql`create table if not exists numbers (a int)`;
+     *   await sql`insert into numbers values(1)`;
+     * });
+     * // later you can call
+     * await sql.commitDistributed("numbers");
+     * // or await sql.rollbackDistributed("numbers");
+     */
+    beginDistributed(name: string, fn: SQLContextCallback): Promise<any>;
+    /** Alternative method to begin a distributed transaction
+     * @alias beginDistributed
+     */
+    distributed(name: string, fn: SQLContextCallback): Promise<any>;
+    /** Current client options */
+    options: SQLOptions;
+  }
+
+  /**
+   * Represents a reserved connection from the connection pool
+   * Extends SQL with additional release functionality
+   */
+  interface ReservedSQL extends SQL {
+    /** Releases the client back to the connection pool */
+    release(): void;
+  }
+
+  /**
+   * Represents a client within a transaction context
+   * Extends SQL with savepoint functionality
+   */
+  interface TransactionSQL extends SQL {
+    /** Creates a savepoint within the current transaction */
+    savepoint(name: string, fn: SQLContextCallback): Promise<undefined>;
+  }
+
+  var sql: SQL;
 
   /**
    *   This lets you use macros as regular imports

src/bun.js/bindings/BunObject.cpp

@@ -292,7 +292,7 @@ static JSValue constructPluginObject(VM& vm, JSObject* bunObject)
     return pluginFunction;
 }
 
-static JSValue constructBunSQLObject(VM& vm, JSObject* bunObject)
+static JSValue defaultBunSQLObject(VM& vm, JSObject* bunObject)
 {
     auto scope = DECLARE_THROW_SCOPE(vm);
     auto* globalObject = defaultGlobalObject(bunObject->globalObject());
@@ -301,6 +301,16 @@ static JSValue constructBunSQLObject(VM& vm, JSObject* bunObject)
     return sqlValue.getObject()->get(globalObject, vm.propertyNames->defaultKeyword);
 }
 
+static JSValue constructBunSQLObject(VM& vm, JSObject* bunObject)
+{
+    auto scope = DECLARE_THROW_SCOPE(vm);
+    auto* globalObject = defaultGlobalObject(bunObject->globalObject());
+    JSValue sqlValue = globalObject->internalModuleRegistry()->requireId(globalObject, vm, InternalModuleRegistry::BunSql);
+    RETURN_IF_EXCEPTION(scope, {});
+    auto clientData = WebCore::clientData(vm);
+    return sqlValue.getObject()->get(globalObject, clientData->builtinNames().SQLPublicName());
+}
+
 extern "C" JSC::EncodedJSValue JSPasswordObject__create(JSGlobalObject*);
 
 static JSValue constructPasswordObject(VM& vm, JSObject* bunObject)
@@ -745,7 +755,8 @@ JSC_DEFINE_HOST_FUNCTION(functionFileURLToPath, (JSC::JSGlobalObject * globalObj
     revision                                       constructBunRevision                                                ReadOnly|DontDelete|PropertyCallback
     semver                                         BunObject_getter_wrap_semver                                        ReadOnly|DontDelete|PropertyCallback
     s3                                             BunObject_callback_s3                                               DontDelete|Function 1
-    sql                                            constructBunSQLObject                                               DontDelete|PropertyCallback
+    sql                                            defaultBunSQLObject                                               DontDelete|PropertyCallback
+    SQL                                            constructBunSQLObject                                               DontDelete|PropertyCallback
     serve                                          BunObject_callback_serve                                            DontDelete|Function 1
     sha                                            BunObject_callback_sha                                              DontDelete|Function 1
     shrink                                         BunObject_callback_shrink                                           DontDelete|Function 1

src/bun.js/bindings/ErrorCode.ts

@@ -167,6 +167,9 @@ const errors: ErrorCodeMapping = [
   ["ERR_POSTGRES_IDLE_TIMEOUT", Error, "PostgresError"],
   ["ERR_POSTGRES_CONNECTION_TIMEOUT", Error, "PostgresError"],
   ["ERR_POSTGRES_LIFETIME_TIMEOUT", Error, "PostgresError"],
+  ["ERR_POSTGRES_INVALID_TRANSACTION_STATE", Error, "PostgresError"],
+  ["ERR_POSTGRES_QUERY_CANCELLED", Error, "PostgresError"],
+  ["ERR_POSTGRES_UNSAFE_TRANSACTION", Error, "PostgresError"],
 
   // S3
   ["ERR_S3_MISSING_CREDENTIALS", Error],

src/bun.js/event_loop.zig

@@ -898,7 +898,6 @@ pub const EventLoop = struct {
     pub fn runCallback(this: *EventLoop, callback: JSC.JSValue, globalObject: *JSC.JSGlobalObject, thisValue: JSC.JSValue, arguments: []const JSC.JSValue) void {
         this.enter();
         defer this.exit();
-
         _ = callback.call(globalObject, thisValue, arguments) catch |err|
             globalObject.reportActiveExceptionAsUnhandled(err);
     }

src/bun.js/module_loader.zig

@@ -2517,14 +2517,7 @@ pub const ModuleLoader = struct {
 
                 // These are defined in src/js/*
                 .@"bun:ffi" => return jsSyntheticModule(.@"bun:ffi", specifier),
-                .@"bun:sql" => {
-                    if (!Environment.isDebug) {
-                        if (!is_allowed_to_use_internal_testing_apis and !bun.FeatureFlags.postgresql)
-                            return null;
-                    }
 
-                    return jsSyntheticModule(.@"bun:sql", specifier);
-                },
                 .@"bun:sqlite" => return jsSyntheticModule(.@"bun:sqlite", specifier),
                 .@"detect-libc" => return jsSyntheticModule(if (!Environment.isLinux) .@"detect-libc" else if (!Environment.isMusl) .@"detect-libc/linux" else .@"detect-libc/musl", specifier),
                 .@"node:assert" => return jsSyntheticModule(.@"node:assert", specifier),
@@ -2732,7 +2725,6 @@ pub const HardcodedModule = enum {
     @"bun:jsc",
     @"bun:main",
     @"bun:test", // usually replaced by the transpiler but `await import("bun:" + "test")` has to work
-    @"bun:sql",
     @"bun:sqlite",
     @"detect-libc",
     @"node:assert",
@@ -2819,7 +2811,6 @@ pub const HardcodedModule = enum {
             .{ "bun:test", HardcodedModule.@"bun:test" },
             .{ "bun:sqlite", HardcodedModule.@"bun:sqlite" },
             .{ "bun:internal-for-testing", HardcodedModule.@"bun:internal-for-testing" },
-            .{ "bun:sql", HardcodedModule.@"bun:sql" },
             .{ "detect-libc", HardcodedModule.@"detect-libc" },
             .{ "node-fetch", HardcodedModule.@"node-fetch" },
             .{ "isomorphic-fetch", HardcodedModule.@"isomorphic-fetch" },
@@ -3059,7 +3050,6 @@ pub const HardcodedModule = enum {
             .{ "bun:ffi", .{ .path = "bun:ffi" } },
             .{ "bun:jsc", .{ .path = "bun:jsc" } },
             .{ "bun:sqlite", .{ .path = "bun:sqlite" } },
-            .{ "bun:sql", .{ .path = "bun:sql" } },
             .{ "bun:wrap", .{ .path = "bun:wrap" } },
             .{ "bun:internal-for-testing", .{ .path = "bun:internal-for-testing" } },
             .{ "ffi", .{ .path = "bun:ffi" } },

src/js/builtins/BunBuiltinNames.h

@@ -259,6 +259,7 @@ using namespace JSC;
     macro(written) \
     macro(napiDlopenHandle) \
     macro(napiWrappedContents) \
+    macro(SQL) \
     BUN_ADDITIONAL_BUILTIN_NAMES(macro)
 // --- END of BUN_COMMON_PRIVATE_IDENTIFIERS_EACH_PROPERTY_NAME ---
 

src/sql/postgres.zig

@@ -217,8 +217,11 @@ pub const PostgresSQLQuery = struct {
     binary: bool = false,
 
     pub usingnamespace JSC.Codegen.JSPostgresSQLQuery;
-
+    const log = bun.Output.scoped(.PostgresSQLQuery, false);
     pub fn getTarget(this: *PostgresSQLQuery, globalObject: *JSC.JSGlobalObject) JSC.JSValue {
+        if (this.thisValue == .zero) {
+            return .zero;
+        }
         const target = PostgresSQLQuery.targetGetCached(this.thisValue) orelse return .zero;
         PostgresSQLQuery.targetSetCached(this.thisValue, globalObject, .zero);
         return target;
@@ -325,10 +328,13 @@ pub const PostgresSQLQuery = struct {
             return;
         }
 
-        // TODO: error handling
         var vm = JSC.VirtualMachine.get();
         const function = vm.rareData().postgresql_context.onQueryRejectFn.get().?;
-        globalObject.queueMicrotask(function, &[_]JSValue{ targetValue, err.toJS(globalObject) });
+        const event_loop = vm.eventLoop();
+        event_loop.runCallback(function, globalObject, thisValue, &.{
+            targetValue,
+            err.toJS(globalObject),
+        });
     }
 
     const CommandTag = union(enum) {
@@ -484,9 +490,14 @@ pub const PostgresSQLQuery = struct {
 
     pub fn call(globalThis: *JSC.JSGlobalObject, callframe: *JSC.CallFrame) bun.JSError!JSC.JSValue {
         const arguments = callframe.arguments_old(4).slice();
-        const query = arguments[0];
-        const values = arguments[1];
-        const columns = arguments[3];
+        var args = JSC.Node.ArgumentsSlice.init(globalThis.bunVM(), arguments);
+        defer args.deinit();
+        const query = args.nextEat() orelse {
+            return globalThis.throw("query must be a string", .{});
+        };
+        const values = args.nextEat() orelse {
+            return globalThis.throw("values must be an array", .{});
+        };
 
         if (!query.isString()) {
             return globalThis.throw("query must be a string", .{});
@@ -496,7 +507,9 @@ pub const PostgresSQLQuery = struct {
             return globalThis.throw("values must be an array", .{});
         }
 
-        const pending_value = arguments[2];
+        const pending_value = args.nextEat() orelse .undefined;
+        const columns = args.nextEat() orelse .undefined;
+
         if (!pending_value.jsType().isArrayLike()) {
             return globalThis.throwInvalidArgumentType("query", "pendingValue", "Array");
         }
@@ -573,10 +586,11 @@ pub const PostgresSQLQuery = struct {
                 signature.deinit();
 
                 if (has_params and this.statement.?.status == .parsing) {
+
                     // if it has params, we need to wait for ParamDescription to be received before we can write the data
                 } else {
                     this.binary = this.statement.?.fields.len > 0;
-
+                    log("bindAndExecute", .{});
                     PostgresRequest.bindAndExecute(globalObject, this.statement.?, binding_value, columns_value, PostgresSQLConnection.Writer, writer) catch |err| {
                         if (!globalObject.hasException())
                             return globalObject.throwError(err, "failed to bind and execute query");
@@ -2447,7 +2461,7 @@ pub const PostgresSQLConnection = struct {
                     DataCell.Putter.put,
                 );
 
-                const pending_value = PostgresSQLQuery.pendingValueGetCached(request.thisValue) orelse .zero;
+                const pending_value = if (request.thisValue == .zero) .zero else PostgresSQLQuery.pendingValueGetCached(request.thisValue) orelse .zero;
                 pending_value.ensureStillAlive();
                 const result = putter.toJS(this.globalObject, pending_value, statement.structure(this.js_value, this.globalObject), statement.fields_flags);
 

test/js/sql/sql.test.ts

@@ -1,4 +1,5 @@
-import { postgres, sql } from "bun:sql";
+import { sql } from "bun";
+const postgres = (...args) => new sql(...args);
 import { expect, test, mock } from "bun:test";
 import { $ } from "bun";
 import { bunExe, isCI, withoutAggressiveGC } from "harness";
@@ -382,168 +383,240 @@ if (!isCI && hasPsql) {
     }
   });
 
-  // t('Throws on illegal transactions', async() => {
-  //   const sql = postgres({ ...options, max: 2, fetch_types: false })
-  //   const error = await sql`begin`.catch(e => e)
-  //   return [
-  //     error.code,
-  //     'UNSAFE_TRANSACTION'
-  //   ]
-  // })
+  test("Throws on illegal transactions", async () => {
+    const sql = postgres({ ...options, max: 2, fetch_types: false });
+    const error = await sql`begin`.catch(e => e);
+    return expect(error.code).toBe("ERR_POSTGRES_UNSAFE_TRANSACTION");
+  });
 
-  // t('Transaction throws', async() => {
-  //   await sql`create table test (a int)`
-  //   return ['22P02', await sql.begin(async sql => {
-  //     await sql`insert into test values(1)`
-  //     await sql`insert into test values('hej')`
-  //   }).catch(x => x.code), await sql`drop table test`]
-  // })
+  test("Transaction throws", async () => {
+    await sql`create table if not exists test (a int)`;
+    try {
+      expect(
+        await sql
+          .begin(async sql => {
+            await sql`insert into test values(1)`;
+            await sql`insert into test values('hej')`;
+          })
+          .catch(e => e.errno),
+      ).toBe(22);
+    } finally {
+      await sql`drop table test`;
+    }
+  });
 
-  // t('Transaction rolls back', async() => {
-  //   await sql`create table test (a int)`
-  //   await sql.begin(async sql => {
-  //     await sql`insert into test values(1)`
-  //     await sql`insert into test values('hej')`
-  //   }).catch(() => { /* ignore */ })
-  //   return [0, (await sql`select a from test`).count, await sql`drop table test`]
-  // })
+  test("Transaction rolls back", async () => {
+    await sql`create table if not exists test (a int)`;
 
-  // t('Transaction throws on uncaught savepoint', async() => {
-  //   await sql`create table test (a int)`
+    try {
+      await sql
+        .begin(async sql => {
+          await sql`insert into test values(1)`;
+          await sql`insert into test values('hej')`;
+        })
+        .catch(() => {
+          /* ignore */
+        });
 
-  //   return ['fail', (await sql.begin(async sql => {
-  //     await sql`insert into test values(1)`
-  //     await sql.savepoint(async sql => {
-  //       await sql`insert into test values(2)`
-  //       throw new Error('fail')
-  //     })
-  //   }).catch((err) => err.message)), await sql`drop table test`]
-  // })
+      expect((await sql`select a from test`).count).toBe(0);
+    } finally {
+      await sql`drop table test`;
+    }
+  });
 
-  // t('Transaction throws on uncaught named savepoint', async() => {
-  //   await sql`create table test (a int)`
+  test("Transaction throws on uncaught savepoint", async () => {
+    await sql`create table test (a int)`;
+    try {
+      expect(
+        await sql
+          .begin(async sql => {
+            await sql`insert into test values(1)`;
+            await sql.savepoint(async sql => {
+              await sql`insert into test values(2)`;
+              throw new Error("fail");
+            });
+          })
+          .catch(err => err.message),
+      ).toBe("fail");
+    } finally {
+      await sql`drop table test`;
+    }
+  });
 
-  //   return ['fail', (await sql.begin(async sql => {
-  //     await sql`insert into test values(1)`
-  //     await sql.savepoit('watpoint', async sql => {
-  //       await sql`insert into test values(2)`
-  //       throw new Error('fail')
-  //     })
-  //   }).catch(() => 'fail')), await sql`drop table test`]
-  // })
+  test("Transaction throws on uncaught named savepoint", async () => {
+    await sql`create table test (a int)`;
+    try {
+      expect(
+        await sql
+          .begin(async sql => {
+            await sql`insert into test values(1)`;
+            await sql.savepoit("watpoint", async sql => {
+              await sql`insert into test values(2)`;
+              throw new Error("fail");
+            });
+          })
+          .catch(() => "fail"),
+      ).toBe("fail");
+    } finally {
+      await sql`drop table test`;
+    }
+  });
 
-  // t('Transaction succeeds on caught savepoint', async() => {
-  //   await sql`create table test (a int)`
-  //   await sql.begin(async sql => {
-  //     await sql`insert into test values(1)`
-  //     await sql.savepoint(async sql => {
-  //       await sql`insert into test values(2)`
-  //       throw new Error('please rollback')
-  //     }).catch(() => { /* ignore */ })
-  //     await sql`insert into test values(3)`
-  //   })
+  test("Transaction succeeds on caught savepoint", async () => {
+    try {
+      await sql`create table test (a int)`;
+      await sql.begin(async sql => {
+        await sql`insert into test values(1)`;
+        await sql
+          .savepoint(async sql => {
+            await sql`insert into test values(2)`;
+            throw new Error("please rollback");
+          })
+          .catch(() => {
+            /* ignore */
+          });
+        await sql`insert into test values(3)`;
+      });
+      expect((await sql`select count(1) from test`)[0].count).toBe("2");
+    } finally {
+      await sql`drop table test`;
+    }
+  });
 
-  //   return ['2', (await sql`select count(1) from test`)[0].count, await sql`drop table test`]
-  // })
+  test("Savepoint returns Result", async () => {
+    let result;
+    await sql.begin(async t => {
+      result = await t.savepoint(s => s`select 1 as x`);
+    });
+    expect(result[0]?.x).toBe(1);
+  });
 
-  // t('Savepoint returns Result', async() => {
-  //   let result
-  //   await sql.begin(async sql => {
-  //     result = await sql.savepoint(sql =>
-  //       sql`select 1 as x`
-  //     )
-  //   })
-
-  //   return [1, result[0].x]
-  // })
-
-  // t('Prepared transaction', async() => {
-  //   await sql`create table test (a int)`
+  // test("Prepared transaction", async () => {
+  //   await sql`create table test (a int)`;
 
   //   await sql.begin(async sql => {
-  //     await sql`insert into test values(1)`
-  //     await sql.prepare('tx1')
-  //   })
+  //     await sql`insert into test values(1)`;
+  //     await sql.prepare("tx1");
+  //   });
 
-  //   await sql`commit prepared 'tx1'`
-
-  //   return ['1', (await sql`select count(1) from test`)[0].count, await sql`drop table test`]
-  // })
-
-  // t('Transaction requests are executed implicitly', async() => {
-  //   const sql = postgres({ debug: true, idle_timeout: 1, fetch_types: false })
-  //   return [
-  //     'testing',
-  //     (await sql.begin(sql => [
-  //       sql`select set_config('bun_sql.test', 'testing', true)`,
-  //       sql`select current_setting('bun_sql.test') as x`
-  //     ]))[1][0].x
-  //   ]
-  // })
-
-  // t('Uncaught transaction request errors bubbles to transaction', async() => [
-  //   '42703',
-  //   (await sql.begin(sql => [
-  //     sql`select wat`,
-  //     sql`select current_setting('bun_sql.test') as x, ${ 1 } as a`
-  //   ]).catch(e => e.code))
-  // ])
-
-  // t('Fragments in transactions', async() => [
-  //   true,
-  //   (await sql.begin(sql => sql`select true as x where ${ sql`1=1` }`))[0].x
-  // ])
-
-  // 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([
-  //     sql.begin(sql => sql`select 1`),
-  //     sql.begin(sql => sql`select 1`)
-  //   ])).map(x => x.count).join(''), await sql`drop table test`]
-  // })
-
-  // t("Many transactions at beginning of connection", async () => {
-  //   const sql = postgres(options);
-  //   const xs = await Promise.all(Array.from({ length: 100 }, () => sql.begin(sql => sql`select 1`)));
-  //   return [100, xs.length];
+  //   await sql`commit prepared 'tx1'`;
+  //   try {
+  //     expect((await sql`select count(1) from test`)[0].count).toBe("1");
+  //   } finally {
+  //     await sql`drop table test`;
+  //   }
   // });
 
-  // t('Transactions array', async() => {
-  //   await sql`create table test (a int)`
+  test("Prepared transaction", async () => {
+    await sql`create table test (a int)`;
 
-  //   return ['11', (await sql.begin(sql => [
-  //     sql`select 1`.then(x => x),
-  //     sql`select 1`
-  //   ])).map(x => x.count).join(''), await sql`drop table test`]
-  // })
+    try {
+      await sql.beginDistributed("tx1", async sql => {
+        await sql`insert into test values(1)`;
+      });
 
-  // t('Transaction waits', async() => {
-  //   await sql`create table test (a int)`
-  //   await sql.begin(async sql => {
-  //     await sql`insert into test values(1)`
-  //     await sql.savepoint(async sql => {
-  //       await sql`insert into test values(2)`
-  //       throw new Error('please rollback')
-  //     }).catch(() => { /* ignore */ })
-  //     await sql`insert into test values(3)`
-  //   })
+      await sql.commitDistributed("tx1");
+      expect((await sql`select count(1) from test`)[0].count).toBe("1");
+    } finally {
+      await sql`drop table test`;
+    }
+  });
 
-  //   return ['11', (await Promise.all([
-  //     sql.begin(sql => sql`select 1`),
-  //     sql.begin(sql => sql`select 1`)
-  //   ])).map(x => x.count).join(''), await sql`drop table test`]
-  // })
+  test("Transaction requests are executed implicitly", async () => {
+    const sql = postgres({ ...options, debug: true, idle_timeout: 1, fetch_types: false });
+    expect(
+      (
+        await sql.begin(sql => [
+          sql`select set_config('bun_sql.test', 'testing', true)`,
+          sql`select current_setting('bun_sql.test') as x`,
+        ])
+      )[1][0].x,
+    ).toBe("testing");
+  });
+
+  test("Uncaught transaction request errosó rs bubbles to transaction", async () => {
+    const sql = postgres({ ...options, debug: true, idle_timeout: 1, fetch_types: false });
+    expect(
+      await sql
+        .begin(sql => [sql`select wat`, sql`select current_setting('bun_sql.test') as x, ${1} as a`])
+        .catch(e => e.errno),
+    ).toBe(42703);
+  });
+
+  // test.only("Fragments in transactions", async () => {
+  //   const sql = postgres({ ...options, debug: true, idle_timeout: 1, fetch_types: false });
+  //   expect((await sql.begin(sql => sql`select true as x where ${sql`1=1`}`))[0].x).toBe(true);
+  // });
+
+  test("Transaction rejects with rethrown error", async () => {
+    await using sql = postgres({ ...options });
+    expect(
+      await sql
+        .begin(async sql => {
+          try {
+            await sql`select exception`;
+          } catch (ex) {
+            throw new Error("WAT");
+          }
+        })
+        .catch(e => e.message),
+    ).toBe("WAT");
+  });
+
+  test("Parallel transactions", async () => {
+    await sql`create table test (a int)`;
+    expect(
+      (await Promise.all([sql.begin(sql => sql`select 1 as count`), sql.begin(sql => sql`select 1 as count`)]))
+        .map(x => x[0].count)
+        .join(""),
+    ).toBe("11");
+    await sql`drop table test`;
+  });
+
+  test("Many transactions at beginning of connection", async () => {
+    await using sql = postgres(options);
+    const xs = await Promise.all(Array.from({ length: 100 }, () => sql.begin(sql => sql`select 1`)));
+    return expect(xs.length).toBe(100);
+  });
+
+  test("Transactions array", async () => {
+    await using sql = postgres(options);
+    await sql`create table test (a int)`;
+    try {
+      expect(
+        (await sql.begin(sql => [sql`select 1 as count`, sql`select 1 as count`])).map(x => x[0].count).join(""),
+      ).toBe("11");
+    } finally {
+      await sql`drop table test`;
+    }
+  });
+
+  test("Transaction waits", async () => {
+    await using sql = postgres({ ...options });
+    await sql`create table test (a int)`;
+    try {
+      await sql.begin(async sql => {
+        await sql`insert into test values(1)`;
+        await sql
+          .savepoint(async sql => {
+            await sql`insert into test values(2)`;
+            throw new Error("please rollback");
+          })
+          .catch(() => {
+            /* ignore */
+          });
+        await sql`insert into test values(3)`;
+      });
+
+      expect(
+        (await Promise.all([sql.begin(sql => sql`select 1 as count`), sql.begin(sql => sql`select 1 as count`)]))
+          .map(x => x[0].count)
+          .join(""),
+      ).toBe("11");
+    } finally {
+      await sql`drop table test`;
+    }
+  });
 
   // t('Helpers in Transaction', async() => {
   //   return ['1', (await sql.begin(async sql =>