Skip to content

sql: reject unsafe() and file() on a closed transaction or reserved handle #43249

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
robobun wants to merge 3 commits into
base: main
Choose a base branch
from
Open

sql: reject unsafe() and file() on a closed transaction or reserved handle #43249

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
1512d6e
sql: reject unsafe() and file() on a closed transaction or reserved h…
robobun Sep 17, 2026
96b8a92
test(sql): assert statement order, not connection ids, after reserved…
robobun Sep 18, 2026
cb285a4
sql: shorten the comment on unsafeQueryFromHandle to one line
robobun Sep 18, 2026
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
111 changes: 58 additions & 53 deletions src/js/bun/sql.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,13 @@ interface TransactionState {
queries: Set<Query<any, any>>;
}

function acceptsQueries(state: TransactionState) {
return (
(state.connectionState & ReservedConnectionState.closed) === 0 &&
(state.connectionState & ReservedConnectionState.acceptQueries) !== 0
);
}

/// Bound as `this` to both callbacks of a reserve({ signal }) call, so each can
/// reach the other without a per-call closure.
interface ReserveAbortState {
Expand Down Expand Up @@ -220,24 +227,28 @@ const SQL: typeof Bun.SQL = function SQL(
}
}

function unsafeTransactionQueryFlags(values: any[]) {
let flags = connectionInfo.bigint
? SQLQueryFlags.allowUnsafeTransaction | SQLQueryFlags.unsafe | SQLQueryFlags.bigint
: SQLQueryFlags.allowUnsafeTransaction | SQLQueryFlags.unsafe;

if ((values?.length ?? 0) === 0) {
flags |= SQLQueryFlags.simple;
}
return flags;
}

function unsafeQueryFromTransaction(
strings: string | TemplateStringsArray | import("internal/sql/shared.ts").SQLHelper<any> | Query<any, any>,
values: any[],
pooledConnection: PooledPostgresConnection,
transactionQueries: Set<Query<any, any>>,
) {
try {
let flags = connectionInfo.bigint
? SQLQueryFlags.allowUnsafeTransaction | SQLQueryFlags.unsafe | SQLQueryFlags.bigint
: SQLQueryFlags.allowUnsafeTransaction | SQLQueryFlags.unsafe;

if ((values?.length ?? 0) === 0) {
flags |= SQLQueryFlags.simple;
}
const query = new Query(
strings,
values,
flags,
unsafeTransactionQueryFlags(values),
queryFromTransactionHandler.bind(pooledConnection, transactionQueries),
pool,
);
Expand All @@ -248,6 +259,35 @@ const SQL: typeof Bun.SQL = function SQL(
}
}

function rejectConnectionClosed(query: Query<any, any>) {
query.reject(pool.connectionClosedError());
}

// Not for COMMIT and ROLLBACK: close() stops accepting queries and then still has to send ROLLBACK.
function unsafeQueryFromHandle(
state: TransactionState,
pooledConnection: PooledPostgresConnection,
strings: string,
values: any[],
) {
if (acceptsQueries(state)) {
return unsafeQueryFromTransaction(strings, values, pooledConnection, state.queries);
}
// Still a lazy Query, so .values() and use as a fragment work. It rejects when it runs.
return new Query(strings, values, unsafeTransactionQueryFlags(values), rejectConnectionClosed, pool);
Comment thread
robobun marked this conversation as resolved.
}

async function fileQueryFromHandle(
state: TransactionState,
pooledConnection: PooledPostgresConnection,
path: string,
values: any[],
) {
const text = await Bun.file(path).text();
// The handle can close while the file is read, so its state is read here and not before.
return unsafeQueryFromHandle(state, pooledConnection, text, values);
}

function onTransactionDisconnected(this: TransactionState, err: Error) {
const reject = this.reject;
this.connectionState |= ReservedConnectionState.closed;
Expand Down Expand Up @@ -355,10 +395,7 @@ const SQL: typeof Bun.SQL = function SQL(
}

function reserved_sql(strings: string | TemplateStringsArray | SQLHelper<any> | Query<any, any>, ...values: any[]) {
if (
state.connectionState & ReservedConnectionState.closed ||
!(state.connectionState & ReservedConnectionState.acceptQueries)
) {
if (!acceptsQueries(state)) {
return Promise.$reject(pool.connectionClosedError());
}
if ($isArray(strings)) {
Expand All @@ -373,17 +410,8 @@ const SQL: typeof Bun.SQL = function SQL(
return queryFromTransaction(strings, values, pooledConnection, state.queries);
}

reserved_sql.unsafe = (string, args = []) => {
return unsafeQueryFromTransaction(string, args, pooledConnection, state.queries);
};

reserved_sql.file = async (path: string, args = []) => {
return await Bun.file(path)
.text()
.then(text => {
return unsafeQueryFromTransaction(text, args, pooledConnection, state.queries);
});
};
reserved_sql.unsafe = (string, args = []) => unsafeQueryFromHandle(state, pooledConnection, string, args);
reserved_sql.file = (path: string, args = []) => fileQueryFromHandle(state, pooledConnection, path, args);

reserved_sql.connect = () => {
if (state.connectionState & ReservedConnectionState.closed) {
Expand Down Expand Up @@ -433,10 +461,7 @@ const SQL: typeof Bun.SQL = function SQL(
};
reserved_sql.begin = (options_or_fn: string | TransactionCallback, fn?: TransactionCallback) => {
// begin is allowed the difference is that we need to make sure to use the same connection and never release it
if (
state.connectionState & ReservedConnectionState.closed ||
!(state.connectionState & ReservedConnectionState.acceptQueries)
) {
if (!acceptsQueries(state)) {
return Promise.$reject(pool.connectionClosedError());
}
let callback = fn;
Expand Down Expand Up @@ -465,10 +490,7 @@ const SQL: typeof Bun.SQL = function SQL(
};
reserved_sql.close = async (options?: { timeout?: number }) => {
const reserveQueries = state.queries;
if (
state.connectionState & ReservedConnectionState.closed ||
!(state.connectionState & ReservedConnectionState.acceptQueries)
) {
if (!acceptsQueries(state)) {
return Promise.$resolve(undefined);
}
state.connectionState &= ~ReservedConnectionState.acceptQueries;
Comment thread
robobun marked this conversation as resolved.
Expand Down Expand Up @@ -662,10 +684,7 @@ const SQL: typeof Bun.SQL = function SQL(
strings: string | TemplateStringsArray | import("internal/sql/shared.ts").SQLHelper<any> | Query<any, any>,
...values: any[]
) {
if (
state.connectionState & ReservedConnectionState.closed ||
!(state.connectionState & ReservedConnectionState.acceptQueries)
) {
if (!acceptsQueries(state)) {
return Promise.$reject(pool.connectionClosedError());
}
if ($isArray(strings)) {
Expand All @@ -679,16 +698,8 @@ const SQL: typeof Bun.SQL = function SQL(

return queryFromTransaction(strings, values, pooledConnection, state.queries);
}
transaction_sql.unsafe = (string, args = []) => {
return unsafeQueryFromTransaction(string, args, pooledConnection, state.queries);
};
transaction_sql.file = async (path: string, args = []) => {
return await Bun.file(path)
.text()
.then(text => {
return unsafeQueryFromTransaction(text, args, pooledConnection, state.queries);
});
};
transaction_sql.unsafe = (string, args = []) => unsafeQueryFromHandle(state, pooledConnection, string, args);
transaction_sql.file = (path: string, args = []) => fileQueryFromHandle(state, pooledConnection, path, args);
// reserve is allowed to be called inside transaction connection but will return a new reserved connection from the pool and will not be part of the transaction
// this matchs the behavior of the postgres package
transaction_sql.reserve = (options?: { signal?: AbortSignal }) => sql.reserve(options);
Expand Down Expand Up @@ -748,10 +759,7 @@ const SQL: typeof Bun.SQL = function SQL(
};
transaction_sql.close = async function (options?: { timeout?: number }) {
// we dont actually close the connection here, we just set the state to closed and rollback the transaction
if (
state.connectionState & ReservedConnectionState.closed ||
!(state.connectionState & ReservedConnectionState.acceptQueries)
) {
if (!acceptsQueries(state)) {
return Promise.$resolve(undefined);
}
state.connectionState &= ~ReservedConnectionState.acceptQueries;
Expand Down Expand Up @@ -833,10 +841,7 @@ const SQL: typeof Bun.SQL = function SQL(
transaction_sql.savepoint = async (fn: TransactionCallback, name?: string): Promise<any> => {
let savepoint_callback = fn;

if (
state.connectionState & ReservedConnectionState.closed ||
!(state.connectionState & ReservedConnectionState.acceptQueries)
) {
if (!acceptsQueries(state)) {
throw pool.connectionClosedError();
}

Expand Down
28 changes: 28 additions & 0 deletions test/js/sql/postgres-listen-notify.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -769,6 +769,34 @@ describe("notify()", () => {
expect((await settled).map(result => result.status)).toEqual(["rejected", "rejected"]);
expect(server.liveConnections).toBe(1);
});

// notify() on a handle sends through that handle's unsafe().
test("rejects on a released reserved handle and on a settled transaction handle, and sends nothing", async () => {
await using server = await mockServer();
await using sql = client(server.url);
const reserved = await sql.reserve();
await reserved.release();
let settledTransaction!: Bun.TransactionSQL;
await sql.begin(async tx => {
settledTransaction = tx;
});

const rejectionCode = (promise: Promise<void>) =>
promise.then(
() => "resolved",
err => err.code ?? err.message,
);
expect({
reserved: await rejectionCode(reserved.notify("ch", "payload")),
transaction: await rejectionCode(settledTransaction.notify("ch", "payload")),
}).toEqual({
reserved: "ERR_POSTGRES_CONNECTION_CLOSED",
transaction: "ERR_POSTGRES_CONNECTION_CLOSED",
});
// The pool has one connection, so a notify that was sent would arrive before this query.
await sql.unsafe("SELECT 'barrier'");
expect(server.queries).toEqual(["BEGIN", "COMMIT", "SELECT 'barrier'"]);
});
});

describe("in a subprocess", () => {
Expand Down
Loading
Loading