Add sql.listen() and sql.notify() for PostgreSQL LISTEN/NOTIFY - #32089
Conversation
|
Updated 5:55 PM PT - Aug 13th, 2026
❌ @robobun, your commit 915989f has 1 failures in
🧪 To try this PR locally: bunx bun-pr 32089That installs a local version of the PR into your bun-32089 --bun |
|
Preview deployment for your docs. Learn more about Mintlify Previews.
💡 Tip: Enable Workflows to automatically generate PRs for you. |
|
Found 1 issue this PR may fix:
🤖 Generated with Claude Code |
|
This PR may be a duplicate of:
🤖 Generated with Claude Code |
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughThis PR implements PostgreSQL LISTEN/NOTIFY support: it adds SQL.listen, SQL.unlisten, and SQL.notify types/docs, JS instance methods with validation and default delegations, a PostgresAdapter dedicated listen connection with reconnect/backoff and per-channel dispatch, JSC/Rust bindings to forward notifications and expose connection metadata, and comprehensive tests. ChangesPostgreSQL LISTEN/NOTIFY pub/sub
Suggested reviewers
🚥 Pre-merge checks | ✅ 4✅ Passed checks (4 passed)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/js/internal/sql/postgres.ts (1)
1151-1157:⚠️ Potential issue | 🟠 Major | ⚡ Quick winValidate
timeoutbefore tearing down LISTEN state.
close()calls#closeListen()before it range-checksoptions.timeout. Passing an invalid timeout throws, but by then the dedicated listen connection, channel maps, and reconnect timer have already been cleared, leaving the adapter open but silently unsubscribed.Proposed fix
async close(options?: { timeout?: number }): Promise<void> { if (this.closed) { return; } - - this.#closeListen(); let timeout = options?.timeout; if (timeout) { timeout = Number(timeout); if (timeout > 2 ** 31 || timeout < 0 || timeout !== timeout) { throw $ERR_INVALID_ARG_VALUE("options.timeout", timeout, "must be a non-negative integer less than 2^31"); } + this.#closeListen(); this.closed = true; if (timeout === 0 || !this.hasPendingQueries()) { await this.#close(); return; } @@ } else { + this.#closeListen(); this.closed = true; if (!this.hasPendingQueries()) { await this.#close(); return; }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/js/internal/sql/postgres.ts` around lines 1151 - 1157, The close() method currently calls `#closeListen`() before validating options.timeout, so an invalid timeout throws after listen state, channel maps and reconnect timer are cleared; move the timeout validation for options.timeout (including Number conversion and the 0 <= timeout < 2**31 and NaN check that throws $ERR_INVALID_ARG_VALUE) to occur before invoking `#closeListen`() so that invalid arguments are rejected without tearing down the dedicated listen connection or clearing channel maps/timers; ensure the validation logic references options.timeout and the same $ERR_INVALID_ARG_VALUE error path used now.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/js/internal/sql/postgres.ts`:
- Around line 1525-1527: The current bug is that `#listenOnlistenCallbacks` stores
onlisten callbacks per channel in a Set, losing the mapping between an
individual onnotify subscription and its corresponding onlisten handler, so
unlisten(channel, onnotify) cannot remove the paired onlisten and it keeps
firing on reconnect via `#scheduleListenReconnect`(); fix by changing
`#listenOnlistenCallbacks` to store a Map<string, Map<onnotifyFn, onlistenFn>> (or
similar) so each channel maps onnotify→onlisten, update the code paths that add
entries (where listen registers both callbacks) to insert the onnotify→onlisten
pair, and update unlisten(channel, onnotify) to look up and remove the specific
onlisten callback for that onnotify (and clean up empty maps/entries) so
`#scheduleListenReconnect`() only re-registers remaining pairs.
- Around line 1801-1804: When creating the inFlight task that awaits
this.#ensureListenConnection() and then calls
this.#runListenQuery(...LISTEN...), add a guard after the await to skip calling
`#runListenQuery` if the channel was removed meanwhile by unlisten(). Concretely:
after the await this.#ensureListenConnection() and before
this.#runListenQuery(conn, `LISTEN ${this.#quoteChannel(channel)}`) check the
module's listener registry (the concrete structure used by listen()/unlisten(),
e.g. this.#listeners or whatever field tracks active channels) to see if the
channel is still registered; if it is not, simply return/resolve and do not send
the LISTEN. This ensures the early unlisten() prevents stale LISTEN being sent.
---
Outside diff comments:
In `@src/js/internal/sql/postgres.ts`:
- Around line 1151-1157: The close() method currently calls `#closeListen`()
before validating options.timeout, so an invalid timeout throws after listen
state, channel maps and reconnect timer are cleared; move the timeout validation
for options.timeout (including Number conversion and the 0 <= timeout < 2**31
and NaN check that throws $ERR_INVALID_ARG_VALUE) to occur before invoking
`#closeListen`() so that invalid arguments are rejected without tearing down the
dedicated listen connection or clearing channel maps/timers; ensure the
validation logic references options.timeout and the same $ERR_INVALID_ARG_VALUE
error path used now.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 7f3c7bfd-ece2-40ff-ba3a-3dce9a05cac2
📒 Files selected for processing (8)
docs/runtime/sql.mdxpackages/bun-types/sql.d.tssrc/js/bun/sql.tssrc/js/internal/sql/postgres.tssrc/runtime/api/sql.classes.tssrc/sql_jsc/jsc.rssrc/sql_jsc/postgres/PostgresSQLConnection.rstest/js/sql/sql.test.ts
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/js/internal/sql/postgres.ts (1)
1527-1536:⚠️ Potential issue | 🟠 Major | 🏗️ Heavy liftDuplicate
listen()calls with the same callback collapse into one subscription.The registry keys subscriptions by
onnotifyfunction identity (Set/Map), and each returned handle later unsubscribes viaunlisten(channel, onnotify). A secondlisten("ch", sameFn)therefore shares the same backing entry, so the first handle removes the second one too. This needs a per-listen token or refcounted subscription record rather than treating the callback as the subscription key. Based on the documentedsub.unlisten()contract indocs/runtime/sql.mdx, eachlisten()result should remove only its own listener.Also applies to: 1798-1805, 1857-1864
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/js/internal/sql/postgres.ts` around lines 1527 - 1536, The current listen/unlisten logic collapses multiple listen(channel, onnotify) calls because `#listenChannels` and `#listenOnlistenCallbacks` key by the onnotify function identity; change this to track subscriptions per-listen call instead of by callback identity: assign a unique token/id (or maintain a refcounted subscription record) for each call in listen(), store records in `#listenChannels` (e.g., Map<string, Map<id, {onnotify}>>) and in `#listenOnlistenCallbacks` (map by channel → map by id → onlisten set), return an unlisten handle that removes only that id record, and ensure existing LISTEN/UNLISTEN RPCs toggle actual DB LISTEN only when the aggregate active-count for a channel goes from 0→1 or 1→0; update listen(), unlisten(), and the code that iterates `#listenChannels/`#listenOnlistenCallbacks to use the new id-based records.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/js/internal/sql/postgres.ts`:
- Around line 1162-1165: Set the adapter's closed flag before tearing down the
dedicated listen connection: move the assignment this.closed = true to occur
before calling this.#closeListen() (and apply the same change at the other
occurrence around the alternate close path). This ensures
`#createListenConnection`() will reject any late connects (preventing a late
resolution from repopulating this.#listenConnection or allowing listen() to
succeed after close()).
- Around line 1153-1156: The code treats timeout=0 as falsy because hasTimeout
is set via !!timeout; change the presence check to detect undefined instead
(e.g., const hasTimeout = options?.timeout !== undefined) and only call timeout
= Number(timeout) when hasTimeout is true so close({ timeout: 0 }) is honored
and converted to numeric 0; update the logic around the timeout and hasTimeout
variables in the same block (referencing timeout, hasTimeout, and
options?.timeout) accordingly.
---
Outside diff comments:
In `@src/js/internal/sql/postgres.ts`:
- Around line 1527-1536: The current listen/unlisten logic collapses multiple
listen(channel, onnotify) calls because `#listenChannels` and
`#listenOnlistenCallbacks` key by the onnotify function identity; change this to
track subscriptions per-listen call instead of by callback identity: assign a
unique token/id (or maintain a refcounted subscription record) for each call in
listen(), store records in `#listenChannels` (e.g., Map<string, Map<id,
{onnotify}>>) and in `#listenOnlistenCallbacks` (map by channel → map by id →
onlisten set), return an unlisten handle that removes only that id record, and
ensure existing LISTEN/UNLISTEN RPCs toggle actual DB LISTEN only when the
aggregate active-count for a channel goes from 0→1 or 1→0; update listen(),
unlisten(), and the code that iterates `#listenChannels/`#listenOnlistenCallbacks
to use the new id-based records.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 3bda564e-85be-4f54-bca1-41764c061140
📒 Files selected for processing (2)
src/js/internal/sql/postgres.tstest/js/sql/sql.test.ts
👮 Files not reviewed due to content moderation or server errors (1)
- test/js/sql/sql.test.ts
|
Re the out-of-diff note about duplicate The public surface is keyed by callback identity: Per-call token records would make the same function fire N times per NOTIFY when registered N times, and would leave |
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/js/internal/sql/postgres.ts (1)
1153-1159:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winValidate the timer's millisecond bound, not the seconds input.
timeoutis multiplied by1000on Line 1179, so the current2 ** 31guard is off by three orders of magnitude. Large values can pass validation here and then overflow the actual timer range, which makes the forced-close path run almost immediately instead of waiting for the requested timeout.Proposed fix
let timeout = options?.timeout; const hasTimeout = !!timeout; + let timeoutMs = 0; if (hasTimeout) { timeout = Number(timeout); - if (timeout > 2 ** 31 || timeout < 0 || timeout !== timeout) { + timeoutMs = timeout * 1000; + if (timeoutMs > 2 ** 31 - 1 || timeout < 0 || timeout !== timeout) { throw $ERR_INVALID_ARG_VALUE("options.timeout", timeout, "must be a non-negative integer less than 2^31"); } } @@ - }, timeout * 1000); + }, timeoutMs);Also applies to: 1175-1179
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/js/internal/sql/postgres.ts` around lines 1153 - 1159, The validation currently checks the seconds value against 2**31 but the code multiplies timeout by 1000 later, so convert and validate in milliseconds: coerce options.timeout to Number into the existing timeout variable, ensure it's a non-negative integer (timeout === Math.floor(timeout) and not NaN), compute const timeoutMs = timeout * 1000, and validate timeoutMs is <= 2**31 - 1 (and >= 0) before using it (or replace later multiplication to use timeoutMs). Update references that rely on the multiplied value to use the validated timeoutMs so the timer bound matches Node's millisecond limit.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@src/js/internal/sql/postgres.ts`:
- Around line 1153-1159: The validation currently checks the seconds value
against 2**31 but the code multiplies timeout by 1000 later, so convert and
validate in milliseconds: coerce options.timeout to Number into the existing
timeout variable, ensure it's a non-negative integer (timeout ===
Math.floor(timeout) and not NaN), compute const timeoutMs = timeout * 1000, and
validate timeoutMs is <= 2**31 - 1 (and >= 0) before using it (or replace later
multiplication to use timeoutMs). Update references that rely on the multiplied
value to use the validated timeoutMs so the timer bound matches Node's
millisecond limit.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 6f44ec4f-664e-4450-837f-88f6ebab9303
📒 Files selected for processing (1)
src/js/internal/sql/postgres.ts
|
Re the out-of-diff note on the timeout bound (2^31 checked in seconds, timer armed in milliseconds): that guard is pre-existing main code shared by both adapters, not introduced here, so I am not changing it in this PR. Tracked in #32091 together with the timeout: 0 handling. |
|
CI status: the diff is green everywhere it runs. Two consecutive builds failed only on the
The Windows |
|
This should be an async resource and work with "using" @robobun |
|
Done in d7074dc: the subscription returned by {
await using sub = await sql.listen("user_events", payload => console.log(payload));
// ...
} // scope exit unlistensTypes, docs, and tests updated (a mock-server test asserts UNLISTEN is sent on scope exit, and a docker-suite test asserts delivery stops). |
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
src/js/internal/sql/postgres.ts (3)
1800-1808:⚠️ Potential issue | 🟠 Major | 🏗️ Heavy liftSame
onnotifyfunction cannot back two independentlisten()handles.The registry is keyed only by
onnotify, so twolisten(channel, sameFn, ...)calls collapse into one entry. When either returned handle callsunlisten(), Lines 1886-1891 remove that shared entry and tear down the other live subscription too. That breaks the per-call cleanup contract implied by the returnedunlisten/[Symbol.asyncDispose]handle.This needs per-subscription bookkeeping or a refcount that is separate from the callback identity.
Also applies to: 1859-1868, 1886-1891
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/js/internal/sql/postgres.ts` around lines 1800 - 1808, The current listen() bookkeeping keys registrations solely by the onnotify function, so multiple listen(channel, sameFn, ...) calls share the same entry and unlisten() from one handle removes the other; change the registry to track subscriptions per-listen call (either by using unique subscription IDs or a per-subscription wrapper object) instead of using onnotify as the sole key. Update `#listenChannels` and `#listenOnlistenCallbacks` to map channel -> Map<subscriptionId, {onnotify, onlistenSet}> (or maintain a refcount for identical handlers) so that each listen() returns an unlisten/Symbol.asyncDispose that only removes its subscriptionId, and adjust the unlisten() logic (and the logic at the spots you noted around lines 1859-1868 and 1886-1891) to only teardown the actual channel listener when no subscriptionIds remain.
1627-1627:⚠️ Potential issue | 🟠 Major | ⚡ Quick winClose the dedicated LISTEN socket when the last subscription is removed.
Line 1627
ref()s the dedicated listen connection, but Lines 1903-1908 only clear the reconnect timer. After the finalunlisten(), the adapter still holds an idle backend session open and can keep the process alive untilclose()is called.Also applies to: 1903-1908
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/js/internal/sql/postgres.ts` at line 1627, The dedicated LISTEN connection (conn) is being ref()'d but never closed when the last subscription is removed; update the unlisten() / subscription-removal path so that when the final subscription is dropped you both clear the reconnect timer (existing lines ~1903-1908) and explicitly tear down the dedicated listen connection: call the appropriate close/release on conn (conn.end() or conn.release() depending on how conn was acquired) and undo the ref by calling conn.unref() (or the equivalent) so the idle backend session is closed and the process can exit; make the same change in the reconnect-timer-clear branch so that clearing the timer also closes/unrefs conn when there are no subscriptions left.
1153-1159:⚠️ Potential issue | 🟠 Major | ⚡ Quick winClamp
close({ timeout })in milliseconds, not seconds.The validation accepts values up to
2 ** 31, but Line 1179 multipliestimeoutby1000before handing it tosetTimeout(). Any value above about2_147_483seconds passes validation and then overflows/clamps the actual timer delay, so a long graceful close can fire much earlier than requested.Suggested fix
let timeout = options?.timeout; const hasTimeout = !!timeout; if (hasTimeout) { timeout = Number(timeout); - if (timeout > 2 ** 31 || timeout < 0 || timeout !== timeout) { + const timeoutMs = timeout * 1000; + if ( + timeout < 0 || + timeout !== timeout || + timeoutMs > 2 ** 31 - 1 + ) { throw $ERR_INVALID_ARG_VALUE("options.timeout", timeout, "must be a non-negative integer less than 2^31"); } } @@ - const timer = setTimeout(() => { + const timer = setTimeout(() => { // timeout is reached, lets close and probably fail some queries this.#close().finally(resolve); - }, timeout * 1000); + }, timeout * 1000);Also applies to: 1175-1179
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/js/internal/sql/postgres.ts` around lines 1153 - 1159, The timeout validation currently treats the value as seconds but later multiplies by 1000 before setTimeout, causing overflow; update the logic so timeout is clamped/validated as milliseconds (a non-negative integer <= 2**31 - 1) and ensure the value passed to setTimeout is that validated millisecond value (adjust/remove the later *1000). Target the code handling options.timeout (variables timeout and hasTimeout) and the close/setTimeout usage so the same validated timeout (in ms) is used end-to-end.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@src/js/internal/sql/postgres.ts`:
- Around line 1800-1808: The current listen() bookkeeping keys registrations
solely by the onnotify function, so multiple listen(channel, sameFn, ...) calls
share the same entry and unlisten() from one handle removes the other; change
the registry to track subscriptions per-listen call (either by using unique
subscription IDs or a per-subscription wrapper object) instead of using onnotify
as the sole key. Update `#listenChannels` and `#listenOnlistenCallbacks` to map
channel -> Map<subscriptionId, {onnotify, onlistenSet}> (or maintain a refcount
for identical handlers) so that each listen() returns an
unlisten/Symbol.asyncDispose that only removes its subscriptionId, and adjust
the unlisten() logic (and the logic at the spots you noted around lines
1859-1868 and 1886-1891) to only teardown the actual channel listener when no
subscriptionIds remain.
- Line 1627: The dedicated LISTEN connection (conn) is being ref()'d but never
closed when the last subscription is removed; update the unlisten() /
subscription-removal path so that when the final subscription is dropped you
both clear the reconnect timer (existing lines ~1903-1908) and explicitly tear
down the dedicated listen connection: call the appropriate close/release on conn
(conn.end() or conn.release() depending on how conn was acquired) and undo the
ref by calling conn.unref() (or the equivalent) so the idle backend session is
closed and the process can exit; make the same change in the
reconnect-timer-clear branch so that clearing the timer also closes/unrefs conn
when there are no subscriptions left.
- Around line 1153-1159: The timeout validation currently treats the value as
seconds but later multiplies by 1000 before setTimeout, causing overflow; update
the logic so timeout is clamped/validated as milliseconds (a non-negative
integer <= 2**31 - 1) and ensure the value passed to setTimeout is that
validated millisecond value (adjust/remove the later *1000). Target the code
handling options.timeout (variables timeout and hasTimeout) and the
close/setTimeout usage so the same validated timeout (in ms) is used end-to-end.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 92d37c30-faa4-4ad4-9001-b393b001e9c7
📒 Files selected for processing (4)
docs/runtime/sql.mdxpackages/bun-types/sql.d.tssrc/js/internal/sql/postgres.tstest/js/sql/sql.test.ts
👮 Files not reviewed due to content moderation or server errors (1)
- test/js/sql/sql.test.ts
|
Re the three out-of-diff findings from the latest review:
|
|
CI status update for build 61927 (sha b116124): three failure signals, none attributable to this PR.
The retrigger budget for this branch is spent (one was used after build 61871's macOS runner infra failures), so I am leaving CI as is rather than pushing empty commits. Every lane that exercises this PR's code paths is green. |
|
Pushed 36adfc0, which fixes #32127 (sql.listen() subscription does not keep the process alive). Root cause: the adapter calls conn.ref() on the dedicated listen connection, but the request was not sticky. The tail of The fix tracks the explicit JS ref()/unref() in a new KEEP_ALIVE_REQUESTED connection flag and has the idle unref path honor it. Close and failure paths still release the poll ref unconditionally, so "last unlisten closes the connection and frees the process" keeps working (covered by the existing exit test). New mock-backed test: the notification is sent only after the child prints SUBSCRIBED, so the child can only receive it if the subscription held the process open. On the unfixed build the child exits first and the test fails. Note: MySQLConnection's ref() has the same non-sticky behavior, left unchanged here since nothing calls it (LISTEN/NOTIFY is Postgres-only) and there is no observable behavior to test. |
|
CI status for build 61931 (sha 36adfc0): one hard failure, the same Evidence it is not caused by this PR:
The other annotated failure ( |
36adfc0 to
3b8717c
Compare
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
…en the Channel comment
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
|
Triage note: the two other open LISTEN/NOTIFY implementations, #25511 (structwafel) and #29710 (versecafe), are now closed in favor of this PR. Both modify the Zig files removed by the Rust port and no longer apply to main. Nothing from either remained to be folded in: #29710 is the branch this PR grew out of (its design is the basis here and versecafe is co-author on the commits), and the parts of #25511 not already covered (listen() resolving to a bare unlisten function, sql.unlisten(channel), a reserved pool connection instead of a dedicated one) are the shape the review above moved away from. The one open item on this PR is still cirospaciari's question about whether the methods should be visibly scoped as Postgres-only, which is with @alii. |
… sent With the listen connection already live, a close() in the same tick as a listen() cleared the channel table while #subscribe was still on its first await; #subscribe then returned without sending anything and listen() resolved with an inert subscription. Since a caller's own registration is the only thing that keeps an entry in the table, an entry that is gone after the round trip means the client was closed, so listen() now rejects with the connection-closed error like the other close-during-listen paths.
|
@robobun check tests from postgres.js github so we port then |
|
Looking at postgres.js's listen/notify tests now; will port the ones that apply and list what each maps to. |
The ten listen/notify tests from postgres.js's tests/index.js, adapted to the subscription object and with their delay() calls replaced by waiting for the deliveries themselves. They replace the two docker tests they subsumed (unlisten stopping delivery, reconnecting after the listening backend is terminated); the payload and transaction tests stay. Verified against a local PostgreSQL 17 as well.
|
@cirospaciari Ported in db9546a. postgres.js has eleven listen/notify tests in
Two adaptations apply throughout: |
|
@alii if you think the DX is ok (look ok for me) feel free to approve and merge it |
|
@robobun fix conflicts |
|
@cirospaciari Conflicts fixed in 06ec075 (merged main). The only conflict was |
…38628) ### What does this PR do? `main` does not compile: #32089 (sql.listen/notify) calls `JSGlobalObject::report_active_exception_as_unhandled`, which #37275 removed (`bun_sql_jsc`: "no method named `report_active_exception_as_unhandled`", ×2 in `PostgresSQLConnection.rs`). A NOTIFY channel/payload string the VM could not make (OOM, a worker's termination) is now handled the way every other JS conversion failure inside `PostgresSQLConnection::on()` is: `on_notification` returns `AnyPostgresError::JSError`, and the connection's error path takes the pending exception and fails the connection with it — which is also #37275's rule that an `Err` always leaves its exception pending for exactly one taker. ### How did you verify your code works? `cargo check -p bun_sql_jsc` and `cargo check -p bun_bin` (full crate graph) pass. The LISTEN/NOTIFY tests need a Postgres server (CI's docker lane).
5 participants
What does this PR do?
Adds
sql.listen()andsql.notify()for PostgreSQL (LISTEN/NOTIFY). Originally #29710 by @versecafe; this is a from-scratch rewrite of both the native and the JS side, rebased onto current main (design and numbers below, the earlier implementation is referred to as v1).listen(channel, onnotify, onlisten?)resolves once the server acknowledged theLISTEN, with aListenSubscription(channel,unlisten(),Symbol.asyncDispose) that removes exactly the registration that call made. Every call is its own registration; several on one channel share one server-sideLISTEN.onlisten()runs on the acknowledgement and again after every reconnect (the place to catch up on missed notifications). Shape per the review discussion below; there is nosql.unlisten()and no pid/secret exposure.notify(channel, payload?)isSELECT pg_notify($1, $2)on the handle it is called on, sotx.notify()is delivered on commit and dropped on rollback. Reserved and transaction handles have both methods.listen()and closed when the last subscription is removed; while anything is subscribed it keeps the process alive (sql.listen() subscription does not keep the process alive while waiting for notifications #32127). If it drops, it reconnects with jittered exponential backoff (250ms to 32s) and re-subscribes every channel.sql.close()tears it down, including mid-handshake.Design
Hot path (one
NotificationResponse), native side inPostgresSQLConnection.rs:NotificationResponse::decodeyields the channel and payload as slices into the read buffer (string_within), nothing is copied. v1 copied both intoVecs.JSStrings (Strong, at most 256 entries), so after the first notification on a channel the channel name costs a scan of that table and no allocation. v1 created a new string every time.JSString(create_utf8_for_js, the string owns the only copy).run_callback, the same way query results are delivered; exceptions surface as uncaught. v1 queued a microtask per notification.ref()on a connection now setsKEEP_ALIVE_REQUESTEDso the idle-unref at the end ofon_dataleaves the listen connection ref'd.JS side (
src/js/internal/sql/postgres.ts): oneListenConnectionper adapter holding aMap<channel, Channel>plus the connection, sweep timer and backoff fields; aListenSubscriptiononly remembers its channel and callbacks and hands them back onunlisten(). Dispatch isMap.getfollowed by a direct call (a lone listener is stored as the bare function; several are stored in an array that is replaced rather than mutated, solisten()/unlisten()from inside a callback cannot disturb the dispatch in progress). AChannelentry's identity scopes itsLISTENround trip: removing the last registration deletes the entry synchronously, and anything resuming after an await compares its entry against the map, which removes the re-add / unlisten-during-ack races v1 needed separate bookkeeping for. There is exactly one repair mechanism: a connection drop, connect failure or rejectedLISTENnulls the affected entries'readypromise and arms#scheduleSweep(), which re-issuesLISTENfor those entries. v1 kept five channel-keyed maps and sets (listeners, onlisten callbacks, in-flight round trips, channels registered on the current connection, failure counters) consistent by hand for the same job.Numbers
Both builds are release builds of this branch's base with the respective implementation applied, measured against a scripted backend in a separate process that writes pre-built frames (it sustains over 10M frames/s into a sink, so the client is what is measured). 4 channels, medians of 5 runs of 1M notifications, two passes each.
JS heap objects per notification (
bun:jscheapStats()delta across a blast with no collection in between): v1 allocated 2 strings plus a Set iterator (until the JIT elides it), this PR allocates the 1 payload string. Native allocations per notification went from twoVecs plus two string buffers plus a microtask entry to one string buffer.How did you verify your code works?
test/js/sql/postgres-listen-notify.test.ts, 56 tests:LISTENack, fail one, or drop the connection on demand) or in a subprocess: routing and ordering, thousands of notifications in one read, UTF-8 and quoting, the 63-byte limit, listeners added or removed from inside a callback, shared round trips, the same callback registered twice being two subscriptions,onlistentiming,unlisten()andawait usingremoving exactly their own registration (idempotent, and a no-op on a handle that outlivedclose()), a re-listen while the UNLISTEN is in flight, reconnect (re-subscribe, an unlistened subscription'sonlistenno longer firing, retry with warning, cancelled by unlistening, a shared listen() whose LISTEN is rejected still repairing the other listeners),close()during a round trip, during the handshake and in the same tick as a listen() on the live connection, argument validation on Postgres and SQLite,notify()going through the pool, and in a subprocess: the subscription keeping the process alive (also across the backoff),close()letting it exit, throwingonnotifyandonlistencallbacks reachinguncaughtExceptionwithout affecting the other listeners, and an RSS check over 2 x 24 MiB of payloads that fails when one string per notification is leaked (verified by injecting that leak).All of them fail on main (
sql.listen is not a function).Fixes #18214
Fixes #32127
[review] gate passed · iteration 30 · 11 files touched
fails on main (without fix)
passes on PR (with fix)
diff hotspot
gate history · 8 passed · 1 rejected · iteration 30
evidence per changed file