Batch Notifications - #470
Merged
Merged
Conversation
A transaction that sends a NOTIFY takes a global PostgreSQL lock, so emitting
one per row written serializes writes against every other notifying transaction
in the database. Queue the payload instead and flush it on an interval: one
notifying transaction per channel per interval, however many rows were written.
dbos-transact-py#774 measures maximum stream throughput with concurrent readers
going from ~3K to ~55K QPS on this change.
Ports dbos-transact-py#774 and #778 (equivalently dbos-transact-ts#1309 and
dbos-transact-go#427), the last SDK to take it.
- Notifier accumulates {workflowId}::{key} into a per-channel set and flushes
with one `SELECT pg_notify(?, p) FROM unnest(?::text[]) AS p`. A failed batch
is dropped rather than requeued: a payload over pg_notify's 8000-byte limit
would otherwise stall the notifier forever, and every waiter still makes
progress on its poll.
- Waiters in the writing process are woken directly, with no round trip. This
also gives them an immediate wake-up where there is no LISTEN/NOTIFY at all,
such as CockroachDB, which previously had only the one-second poll.
- The notifications channel keeps its database trigger. Per #778 a message can
be sent from a process with no notifier running, and the trigger fires inside
the writing transaction so a recv is never woken before the row it would read
has committed.
- DBOSConfig.notificationCoalesceInterval configures the flush interval,
defaulting to 10ms and rejecting anything under 1ms.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Java's migration history stopped at 41 while Python and TypeScript reached 47. A shared system database carries one dbos_migrations version, so whichever SDK migrates it first decides the schema every peer then skips: a Java-migrated database left a TypeScript peer without the debounce columns it writes on every debounced enqueue. - 42 adds workflow_status.debounce_deadline_epoch_ms and is_debounced. Java's debouncer does not use them; they exist so a peer's does. It is also what makes 43 and 44 reachable, since the migration list is positional. - 43 and 44 drop the streams and workflow_events NOTIFY triggers. The preceding commit moved those wake-ups off the write path, and until the triggers go the write transaction still takes the async-notify lock per row, so the two belong together. Unlike the migrations that create these triggers, the drops are not gated on useListenNotify. Both statements are IF EXISTS no-ops where the objects were never created, on PostgreSQL and CockroachDB alike, and gating them would leave a hole: a process running without LISTEN/NOTIFY that happened to be the one advancing the version past here would skip the drop, and no later process would retry it, so the triggers would survive forever on that database -- still sending a notification inside every write transaction. - 45-47 add the partitioned-queue dequeue index and retire its first version. The surviving index is the one QueuesDAO.startQueuedWorkflows already reads through: it equality-matches queue_name, status and queue_partition_key and orders by priority and created_at. Java does not have the batched multi-partition claim these were introduced alongside (py#802), but the index serves its own dequeue regardless. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
A reader that found nothing at its offset checked whether the producer was still active and, if not, ended the stream. Those are two statements, so the producer can commit its last value between them: the read misses it, the status check sees terminal, and the value is dropped. Cancellation and timeout make this reachable without a race at all, since both set the status from outside while the workflow is still writing. Once the producer is terminal every one of its writes is committed, so make one more read pass before ending the stream. Python avoids the window by reading the value and the status in one statement; Go takes this shape, with the same regression test (read_stream_drain_test.go). Java was the only SDK dropping the value. The test drives the interleaving through readStream's createSubscription parameter, which is called once per pass: the second call is exactly the window between the first read and the drain pass, so the producer's late write is committed there. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
recv and getEvent re-queried every second whether or not a listener was running. The listener signals them promptly, so with one running that re-check is only a safety net -- against a notification dropped on the wire or missed across a reconnect, and, since the preceding commits, against a writer that died between committing and its notifier flushing. Python drops to a 60s re-check in that case (_event_recheck_interval) and Go to the same 60s ticker (_NOTIFICATION_FALLBACK_RECHECK_INTERVAL); Java polled 60x more often than either for no added delivery. Java keeps the fallback on all three channels rather than following Go, which skips it on the notifications channel: Go wakes every waiter after a listener reconnect and Java's listener does not, so without the re-check a recv would hang until its timeout across a reconnect. Two waits deliberately keep the one-second interval, both because nothing pushes what they are waiting for: - awaitWorkflowResult, which no channel carries at all. - A stream read, which is watching for the producer to terminate as well as for a value to arrive. Only the value is pushed. Python and Go draw the line in the same place (_notification_listener_polling_interval_sec, _READ_STREAM_POLL_INTERVAL), and StreamTest.streamTerminationWhileReaderBlocked catches it: on the 60s interval a blocked reader took 60s to notice. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Python and TS read the stream value and the owning workflow's status in a single LEFT JOIN on every pass, and raise DBOSNonExistentWorkflowError as soon as the status comes back null. Java reads the status separately and only on a miss, and folded "no such workflow" into the same branch as "workflow is no longer active" -- so reading a stream on a mistyped workflow ID silently yielded an empty iterator instead of reporting the mistake. Split the two cases. A null status now throws DBOSNonExistentWorkflowException, matching the other SDKs; a terminal status still sets up the drain pass. The check stays on the miss path, which reaches the same conclusion for a workflow that never existed, and stays behind the finalRead short-circuit so a workflow garbage-collected mid-drain still ends the stream cleanly rather than throwing.
Wait operations -- awaiting a workflow result, recv, getEvent, and reading a stream -- re-query on an interval, holding a connection for the query but not across the wait. One waiter is cheap; a few thousand ticking on the same interval are not, because they all reach for the pool at the same moment and the control plane (enqueue and dequeue, status writes, recovery, cancellation) queues behind the burst. Python bounds this with PollingLimiter and TS with pollLimiter; Java had no equivalent anywhere. Add PollingLimiter, a counting semaphore held for the duration of a polling read's connection, and take a permit at each of the four wait loops. The permit covers only the query: every loop already closes its connection before waiting, so nothing blocks while holding one, and no permit-holder acquires a second. Defaults to half the pool with a minimum of one, matching Python and TS, and is configurable through DBOSConfig.withDatabasePollingConcurrency -- non-positive removes the cap. While here, readStream now checks the workflow's status on the connection it already holds instead of checking out a second one underneath the first.
ChaosTest started failing at "Poll for notifications at the same intervals as the other SDKs" with NoSuchElementException from getEvent(...).orElseThrow(). It kills every backend on the database, the listener's included, and NOTIFY is not queued for absent listeners: a wake-up delivered while this process had no connection is gone for good, and re-establishing LISTEN only catches later ones. The row sits committed in the database with nothing left to point at it. Recovering that was left entirely to the periodic re-check, which is why raising the interval to 60s broke it. A wait computes min(interval, remaining), so a five second getEvent now runs one query and sleeps out its whole timeout instead of polling five times. That is not specific to the interval, though: the same wait would have missed the notification at any interval longer than its own timeout. The listener knew it had been disconnected and said nothing. So say something. After LISTEN is re-established, raise every outstanding signal, so each waiter re-queries once and finds whatever landed while the connection was down. This closes the hole regardless of when a wait started, which tracking the connection's liveness would not: a wait already sleeping when the connection drops has already chosen its interval. The thundering herd this could cause on a large process is bounded by the polling limiter. Python's Postgres listener has the same structure and the same gap, and is worth fixing there too; no other SDK has a test that kills connections, which is why this surfaced here first. ListenNotifySource now takes the SignalMap rather than a raiseSignal callback, since it needs two operations from it and the indirection bought nothing.
kraftp
approved these changes
Aug 19, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
fixes #469
Update DBOS Java to move event and stream notications out of db triggers and into the application. The notifications channel deliberately keeps its trigger, matching py#778: messages can be sent from a process with no background flusher, and the trigger fires in-transaction so recv is never woken before the row it would read has committed.
Other Changes
DBOSNonExistentWorkflowExceptioninstead of returning an empty iterator