postgres: LISTEN/NOTIFY for reactive task polling - #488
Open
concernedrat wants to merge 11 commits into
Open
Conversation
- Add database migration with triggers for workflow and activity tasks - Implement notification listener with LISTEN/NOTIFY support - Add WithNotifications option to enable reactive polling - Integrate listener with GetWorkflowTask and GetActivityTask - Add comprehensive tests for notification functionality - Fix race conditions with proper synchronization Co-authored-by: cschleiden <2201819+cschleiden@users.noreply.github.com>
- Create postgres-notify sample demonstrating the feature - Add comprehensive README for the sample - Update Postgres backend README with LISTEN/NOTIFY documentation - Demonstrate 9x performance improvement with notifications enabled Co-authored-by: cschleiden <2201819+cschleiden@users.noreply.github.com>
- Define constants for listener reconnect intervals and ping interval - Add comments explaining dual driver imports (pgx for SQL, pq for LISTEN/NOTIFY) - Add comment in trigger explaining SELECT is efficient due to unique index - Improve code maintainability with named constants Co-authored-by: cschleiden <2201819+cschleiden@users.noreply.github.com>
…ible_at deadline Three gaps in the notification support needed for production use: - WithListenerDSN: LISTEN needs a session-level connection, so deployments whose SQL DSN points at a transaction-pooling proxy (PgBouncer) must aim the listener directly at Postgres. Falls back to the SQL DSN when unset. - Single pq.Listener multiplexing both channels: one connection per process instead of two. A nil notification (pq reconnect signal, previously dropped) now conservatively wakes both consumers since notifications may have been missed while disconnected. - visible_at wait deadline: triggers only fire on INSERT, so an event scheduled with a future visible_at (workflow timers, retry backoff) becomes due without any NOTIFY. GetWorkflowTask/GetActivityTask now bound the notification wait by MIN(visible_at) so deferred work fires on schedule instead of stalling until the caller's poll timeout. Tests: timer fires at its visible_at with no new INSERT (2s, not the 10s context); notifications delivered through a dedicated listener DSN; DSN fallback.
Resolve conflict in backend/postgres/postgres.go: keep the ownsConnection field and close semantics from upstream alongside the notification listener. Close the listener before honoring ownsConnection. Wire the listener into both constructors; the WithDB constructor requires an explicit listener DSN.
A NOTIFY can be lost around process start (listener session not yet established) or during a reconnect. With nothing scheduled, the poller wait was unbounded, so already-pending tasks could sit idle until an unrelated notification arrived. Cap every wait at safetyPollInterval (60s) so a lost wake-up costs at most one interval; idle load stays at one cheap query per poller per minute.
Author
|
bump |
The activity goroutine decremented runningActivities in a defer that runs AFTER it enqueues its completion callback. The callback channel is buffered, so the send returns immediately and the Execute loop can drain the callback, run a full workflow task, and reach fireTimer before the goroutine is rescheduled to decrement. newTimerMode then reads runningActivities > 0 for an activity that has already finished and returns TM_WallClock, so a retry timer scheduled in that window goes on the real clock. The test blocks for the actual backoff interval and dies on TestTimeout with 'No new events generated during workflow execution and no pending timers, workflow blocked?'. Raising TestTimeout cannot fix it: the wall-clock waits follow the retry policy (30s, 60s, 120s, ...), so a larger deadline just moves the panic to a later attempt. The decrement now happens inside the callback, on the Execute loop's own goroutine, ordered before any timer-mode decision. The remaining defer covers only the paths that panic before enqueueing. The window in which an activity counts as running gets longer, not shorter (schedule until the completion event is applied, rather than until the goroutine exits), so this cannot cause the inverse bug of time-travelling past an activity that is genuinely in flight. Reproduced on a 24-core machine with 48 busy loops of CPU contention: 3/3 runs panic before the patch, 400 iterations clean after it. Not reproducible at GOMAXPROCS=1, where the sender runs straight through the buffered send to the decrement.
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
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.
Based on the draft PR #458 (#458), continued and completed. (Let me know if you prefer the commits on the original PR)
Adds LISTEN/NOTIFY support to the Postgres backend so workers wake up immediately when new work is enqueued instead of waiting on the poll interval.
What changed
NOTIFYonworkflow_tasks/activity_tasks(migration000002_add_notify_triggers).notificationListenermultiplexes both channels over one session, so a process holds exactly one LISTEN connection regardless of how many workers it runs.visible_atdeadline so delayed/scheduled tasks are not woken early.WithNotifications(true); default behavior is unchanged.WithListenerDSNlets the listener target Postgres directly when the regular DSN points at a transaction-pooling proxy (PgBouncer in transaction mode) that does not support LISTEN.Tests
notify_test.goandnotify_listener_test.gocover trigger firing, listener reconnect, and the visible_at deadline.Sample
samples/postgres-notifydemonstrates the setup, plus README docs inbackend/postgres.