Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions docs/runtime/sql.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -980,6 +980,14 @@ await sql.close({ timeout: 0 }); // close all connections from the pool immediat

`timeout` is in seconds. With `timeout: 0`, or when the timeout expires, queries that are still in flight reject with the adapter's `ERR_*_CONNECTION_CLOSED` error.

`sql.close()` waits for the queries of the pool that started before the call. A query starts in the call of `.then()`, `.catch()`, `.finally()` or `.execute()`. `await query` and `Promise.all([query])` call `.then()` in a later microtask. When `close()` follows them in the same tick, it comes before the start, and the query rejects with the adapter's `ERR_*_CONNECTION_CLOSED` error. The same applies to a library that awaits the query in an async function of its own.

```ts
sql`INSERT INTO users ${sql(alice)}`.catch(report); // started, close() waits for it
const inserted = (async () => await sql`INSERT INTO users ${sql(bob)}`)(); // not started yet
await sql.close(); // `inserted` rejects
```

---

## Reserved Connections
Expand Down
6 changes: 6 additions & 0 deletions packages/bun-types/sql.d.ts
Original file line number Diff line number Diff line change
Expand Up @@ -683,6 +683,12 @@ declare module "bun" {
* Closes the database connection. With `timeout: 0` it closes
* immediately; with no timeout it waits for all queries to finish first.
*
* It waits for the queries of the pool that started before the call. A
* query starts in the call of `.then()`, `.catch()`, `.finally()` or
* `.execute()`. `await query` calls `.then()` in a later microtask, so
* a `close()` that follows it in the same tick comes before the start,
* and the query rejects.
*
* @param options Optional `timeout` in seconds
*
* @example
Expand Down
6 changes: 4 additions & 2 deletions src/js/bun/sql.ts
Original file line number Diff line number Diff line change
Expand Up @@ -993,7 +993,8 @@ const SQL = function SQL(
}

const sqlQuery = pool.getRollbackDistributedSQL(name);
return await sql.unsafe(sqlQuery);
// then() starts the query in this call. `await` alone starts it one promise job later.
return await sql.unsafe(sqlQuery).then();
};

sql.commitDistributed = async function (name: string) {
Expand All @@ -1006,7 +1007,8 @@ const SQL = function SQL(
}

const sqlQuery = pool.getCommitDistributedSQL(name);
return await sql.unsafe(sqlQuery);
// then() starts the query in this call. `await` alone starts it one promise job later.
return await sql.unsafe(sqlQuery).then();
};

sql.beginDistributed = (name: string, fn: TransactionCallback) => {
Expand Down
51 changes: 4 additions & 47 deletions src/js/internal/sql/query.ts
Original file line number Diff line number Diff line change
Expand Up @@ -128,38 +128,6 @@ class Query<T, Handle extends BaseQueryHandle<any>> extends PublicPromise<T> {
}
}

async #runAsync() {
const { [_handler]: handler, [_queryStatus]: status } = this;

if (
status &
(SQLQueryStatus.executed | SQLQueryStatus.error | SQLQueryStatus.cancelled | SQLQueryStatus.invalidHandle)
) {
return;
}

if (this[_flags] & SQLQueryFlags.notTagged) {
this.reject(this[_adapter].notTaggedCallError());
return;
}

this[_queryStatus] |= SQLQueryStatus.executed;
const handle = this.#getQueryHandle();

if (!handle) {
return this;
}

await Promise.$resolve();

try {
return handler(this, handle);
} catch (err) {
this[_queryStatus] |= SQLQueryStatus.error;
this.reject(err as Error);
}
}

get active() {
return (this[_queryStatus] & SQLQueryStatus.active) != 0;
}
Expand Down Expand Up @@ -240,7 +208,7 @@ class Query<T, Handle extends BaseQueryHandle<any>> extends PublicPromise<T> {
throw this[_adapter].notTaggedCallError();
}

await this.#runAsync();
this.#run();
return this;
}

Expand Down Expand Up @@ -271,23 +239,12 @@ class Query<T, Handle extends BaseQueryHandle<any>> extends PublicPromise<T> {
return this;
}

#runAsyncAndCatch() {
const runPromise = this.#runAsync();

if ($isPromise(runPromise) && runPromise !== this) {
runPromise.catch(() => {
// Error is already handled via this.reject() in #runAsync
// This catch is just to prevent unhandled rejection warnings
});
}
}

then<TResult1 = T, TResult2 = never>(
onfulfilled?: ((value: T) => TResult1 | PromiseLike<TResult1>) | null,
onrejected?: ((reason: any) => TResult2 | PromiseLike<TResult2>) | null,
): Promise<TResult1 | TResult2>;
then() {
this.#runAsyncAndCatch();
this.#run();

const result = super.$then.$apply(this, arguments);

Expand All @@ -305,7 +262,7 @@ class Query<T, Handle extends BaseQueryHandle<any>> extends PublicPromise<T> {
throw this[_adapter].notTaggedCallError();
}

this.#runAsyncAndCatch();
this.#run();

const result = super.catch.$apply(this, arguments);
$pokePromiseAsHandled(result);
Expand All @@ -318,7 +275,7 @@ class Query<T, Handle extends BaseQueryHandle<any>> extends PublicPromise<T> {
throw this[_adapter].notTaggedCallError();
}

this.#runAsyncAndCatch();
this.#run();

return super.finally.$apply(this, arguments);
}
Expand Down
Loading
Loading