feat: new Subscribe method with pub/sub for hot-path polling - #5
Conversation
There was a problem hiding this comment.
Pull request overview
Adds a long-running Outbox.Subscribe API to continuously drain topics without hot-path polling, optionally woken by best-effort pub/sub notifications (with a built-in Postgres LISTEN/NOTIFY implementation). This integrates notification publishing into AddMessages (transactionally when supported) and documents/benchmarks the new workflow.
Changes:
- Add
Subscribeloop with options for poll interval, forwardedProcessMessagesoptions, and exclusive-lease management (WithExclusive). - Introduce
PubSub/TxPublisherabstractions plusNewPGPubSub(PostgresLISTEN/NOTIFY) and wire notifications intoAddMessages. - Add E2E tests, update docs/examples, and expand benchmarks to compare polling vs notification-driven draining.
Reviewed changes
Copilot reviewed 9 out of 10 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| subscribe.go | Implements Outbox.Subscribe, options, and helper logic for draining and wake-ups. |
| subscribe_e2e_test.go | Adds E2E coverage for pubsub wake-ups, polling fallback, and exclusive failover/reacquire behavior. |
| README.md | Documents FlushContext, Subscribe, and pubsub wake-up behavior; updates examples/bench notes. |
| pubsub.go | Defines PubSubMessage, PubSub, and TxPublisher interfaces used by the outbox. |
| pgpubsub.go | Implements PubSub/TxPublisher via Postgres LISTEN/NOTIFY with reconnect logic and fan-out. |
| outbox.go | Wires Subscribe, WithPubSub, AddMessages notification publishing, and adds ErrExclusiveLeaseRequired. |
| outbox_bench_test.go | Expands benchmarks to a matrix and adds a notification-driven (Subscribe) benchmark. |
| examples/main.go | Updates example to use Subscribe + NewPGPubSub and the new FlushContext signature. |
| examples/go.mod | Updates example module indirect deps to match new imports/usage. |
| examples/go.sum | Updates sums for newly referenced indirect dependencies. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| select { | ||
| case <-ctx.Done(): | ||
| return ctx.Err() | ||
| case <-time.After(so.pollInterval): | ||
| case _, ok := <-notifications: | ||
| if !ok { | ||
| // The PubSub shut down before our ctx did; degrade to polling. | ||
| notifications = nil | ||
| } | ||
| } |
| for ch := range p.subscribers[msg.Topic] { | ||
| select { | ||
| case ch <- msg: | ||
| default: | ||
| // Subscriber buffer full — drop rather than block the listener. | ||
| } | ||
| } |
|
Cursor generated me this nice visual |
igor-kupczynski
left a comment
igor-kupczynski
left a comment
There was a problem hiding this comment.
Really good docs! I've raise two comments -- not blocking, but I think important enough to think through before merging.
| // extra processing pass on an empty topic is a no-op). | ||
| type PubSub interface { | ||
| // Pub publishes payload to topic. | ||
| Pub(ctx context.Context, topic string, payload []byte) error |
There was a problem hiding this comment.
🤔 do we need payloads on this interface at all?
IIUC the intent is for the outbox table row to carry the payload, this is just a notification after the row is written.
If we drop it, we can simplify the PR. WDYT?
There was a problem hiding this comment.
You're definitely right in the case of this usage, but I'm hoping we can re-use the pg_listen / pg_notify pub sub implementation in https://github.com/hatchet-dev/hatchet, which would require the payload.
| return nil | ||
| } | ||
|
|
||
| if txp, ok := o.pubsub.(TxPublisher); ok { |
There was a problem hiding this comment.
🤔 This type cast decides the semantics per call: TxPublisher and notify-at-commit, or best-effort, notify right away;
I think a nicer design would do that once at construction time (in WithPubSub or NewOutbox). You could check the interface there and set a publish func or a mode flag. WDYT?
There was a problem hiding this comment.
This is a good catch, the semantics only make sense if we call this after commit (since we're notifying to poll for new messages). I think perhaps AddMessage needs to return a post-commit callback, because there's no way to do this using pgx.Tx (which would be ideal). Will refactor this.
There was a problem hiding this comment.
AddMessage needs to return a post-commit callback, because there's no way to do this using pgx.Tx (which would be ideal).
agree, i was thinking about this as well, that sending a message DURING the transaction is not ideal
There was a problem hiding this comment.
I added a way to hook into a notifier after AddMessages is called - it's a little bit of a strange design pattern, but I didn't want to return a NotifyFunc after every call to AddMessages, or be too invasive in the case we're just using pg listen/notify, so I thought this was a good middle ground. Let me know what you think!
Description
Adds a new
Subscribemethod which improves the experience of using the outbox so that callers don't need to continuously callProcessMessages. That API is still available for more custom use-cases (for example, slow-polling).The outbox now accepts a
PubSubinterface which notifies consumers when a new message has been published.Type of change