feat: new Subscribe method with pub/sub for hot-path polling

README.md

@@ -7,7 +7,7 @@ Here's an example of flushing messages on `topic1` by simply printing them to th
 ```go
 type printFlusher struct{}
 
-func (printFlusher) Flush(_ context.Context, msgs []*sqlc.Message) error {
+func (printFlusher) Flush(_ pgoutbox.FlushContext, msgs []*sqlc.Message) error {
 	for _, m := range msgs {
 		fmt.Printf("  flushed id=%d topic=%s payload=%s\n", m.ID, m.Topic, string(m.Payload))
 	}
@@ -97,16 +97,13 @@ outbox.ProcessMessages(ctx, "shipments")
 
 ## Atomic flush and delete
 
-If your flusher writes to Postgres itself (e.g. into a relay table), implement `TxFlusher` instead of `Flusher`. `ProcessMessages` will pass the same transaction it uses to lock and delete messages, so your writes and the outbox delete commit or roll back together:
+If your flusher writes to Postgres itself (e.g. into a relay table), use the transaction exposed by the `FlushContext` passed to `Flush`. It is the same transaction `ProcessMessages` uses to lock and delete messages, so your writes and the outbox delete commit or roll back together:
 
 ```go
-type relayFlusher struct{ db *pgxpool.Pool }
+type relayFlusher struct{}
 
-func (f *relayFlusher) Flush(_ context.Context, _ []*sqlc.Message) error {
-    panic("not used; FlushWithTx is called instead")
-}
-
-func (f *relayFlusher) FlushWithTx(ctx context.Context, tx pgx.Tx, msgs []*sqlc.Message) error {
+func (f *relayFlusher) Flush(ctx pgoutbox.FlushContext, msgs []*sqlc.Message) error {
+    tx := ctx.Tx()
     for _, m := range msgs {
         if _, err := tx.Exec(ctx, "INSERT INTO relay (payload) VALUES ($1)", m.Payload); err != nil {
             return err
@@ -116,9 +113,51 @@ func (f *relayFlusher) FlushWithTx(ctx context.Context, tx pgx.Tx, msgs []*sqlc.
 }
 ```
 
+`FlushContext` embeds `context.Context`, so flushers that don't need the transaction can ignore `Tx()` and treat it as a plain context.
+
+## Continuous processing with `Subscribe`
+
+Instead of calling `ProcessMessages` yourself, `Subscribe` runs it in a loop: it drains the topic, then waits until either the poll interval elapses or a new-message notification arrives (see below), and drains again. It blocks until its context is cancelled:
+
+```go
+go func() {
+    err := outbox.Subscribe(ctx, "orders",
+        pgoutbox.WithPollInterval(5*time.Second),              // default: 5s
+        pgoutbox.WithProcessOpts(pgoutbox.WithBatchSize(500)), // forwarded to every ProcessMessages call
+    )
+    if err != nil && !errors.Is(err, context.Canceled) {
+        panic(err)
+    }
+}()
+```
+
+Processing errors don't kill the loop — they're logged to the `WithLogger` logger and retried on the next wake-up. `Subscribe` returns an error immediately only if no flusher is registered for the topic or the subscription itself can't be established.
+
+### Waking on new messages with `LISTEN`/`NOTIFY`
+
+Without further configuration, `Subscribe` is purely poll-based. To wake subscribers the moment new messages commit, attach a `PubSub` — the built-in implementation rides Postgres `LISTEN`/`NOTIFY`:
+
+```go
+ps, err := pgoutbox.NewPGPubSub(ctx, pool)
+if err != nil {
+    panic(err)
+}
+
+outbox, err := pgoutbox.NewOutbox(ctx, pool, pgoutbox.WithPubSub(ps))
+```
+
+With a `PubSub` attached, `AddMessages` publishes a notification for each staged topic and `Subscribe` wakes on it instead of waiting out the poll interval. The pg-backed `PubSub` publishes *inside the `AddMessages` transaction*, so the notification is delivered exactly when the insert commits — and never for a transaction that rolls back. Postgres deduplicates identical notifications within a transaction, so any number of `AddMessages` calls for a topic in one transaction cost a single wake-up.
+
+Delivery is best-effort by design: if a notification is lost (for example while the listener reconnects), polling picks the messages up within one poll interval. The listener occupies a single dedicated connection (hijacked out of the pool so it doesn't consume a pool slot) no matter how many topics are subscribed.
+
+Two details worth knowing:
+
+- All notifications travel over one NOTIFY channel, `pgoutbox_pubsub` by default. Two outboxes sharing a database (e.g. different schemas) should use distinct channels via `pgoutbox.WithNotifyChannel("my_channel")` to avoid waking each other's subscribers.
+- `PubSub` is an interface, so you can bring your own transport (e.g. Redis, NATS) instead of `LISTEN`/`NOTIFY`. If your implementation also implements `TxPublisher`, notifications are published transactionally as described above; otherwise `AddMessages` publishes best-effort at insert time.
+
 ## Message expiration
 
-Call `Start` to run background maintenance goroutines that delete old messages. Expiration is configured per topic, with an optional default for topics not explicitly named:
+`NewOutbox` starts background maintenance goroutines that delete old messages; they run until the context passed to `NewOutbox` is cancelled. Expiration is configured per topic, with an optional default for topics not explicitly named:
 
 ```go
 ctx, cancel := context.WithCancel(context.Background())
@@ -134,14 +173,14 @@ if err != nil {
 }
 ```
 
-`Start` always runs a background scanner goroutine that polls the `topics` table, but maintenance loops are only launched for topics that actually have an expiration configured. Topics don't need to be declared at startup: the library tracks every topic that receives a message in a `topics` table (via a Postgres trigger) and applies the default expiration automatically.
+The outbox always runs a background scanner goroutine that polls the `topics` table, but maintenance loops are only launched for topics that actually have an expiration configured. Topics don't need to be declared at startup: the library tracks every topic that receives a message in a `topics` table (via a Postgres trigger) and applies the default expiration automatically.
 
 Multiple outbox instances (e.g. replicas of the same service) coordinate cleanup using a per-topic maintenance lease, so only one instance runs the delete at a time.
 
 To log errors from the maintenance goroutines, pass a [zerolog](https://github.com/rs/zerolog) logger:
 
 ```go
-outbox, err := pgoutbox.NewOutbox(pool,
+outbox, err := pgoutbox.NewOutbox(ctx, pool,
     pgoutbox.WithTopicExpiration("orders", 24*time.Hour),
     pgoutbox.WithLogger(logger),
 )
@@ -176,8 +215,30 @@ if errors.Is(err, pgoutbox.ErrExclusiveLeaseHeld) {
 }
 ```
 
+An instance that never acquired the lease (or whose lease has expired) receives `pgoutbox.ErrExclusiveLeaseRequired` instead.
+
 When the holder's context is cancelled, the lease expires naturally (within the lease duration, 30 s by default) and another instance's `AcquireTopic` call unblocks.
 
+### Exclusive subscribers
+
+`Subscribe` composes with exclusive consumers: pass `WithExclusive()` and it manages the lease for you.
+
+```go
+// Exactly one instance across the fleet drains "orders" at a time; the rest
+// wait in line and take over on failure.
+err := outbox.Subscribe(ctx, "orders", pgoutbox.WithExclusive())
+```
+
+With `WithExclusive()`, `Subscribe`:
+
+1. acquires the lease before its first processing pass, blocking while another instance holds it (like `AcquireTopic`);
+2. re-acquires it automatically if the lease is ever lost mid-subscribe (for example, a heartbeat lapse during a database blip); and
+3. releases it on return, so a waiting instance takes over immediately instead of waiting out the lease's grace period.
+
+Several instances calling `Subscribe(..., WithExclusive())` on the same topic therefore form a failover group: one active consumer, the rest hot standbys. Nothing is missed during a handoff — the new holder's first drain pass covers any backlog that accumulated while the lease changed hands.
+
+Without `WithExclusive()`, subscribing to a topic whose exclusive lease is held elsewhere doesn't fail — every pass errors (visible via `WithLogger`) and is retried, so the subscriber sits idle until the lease frees up or is acquired. For an exclusive topic, either call `AcquireTopic` before `Subscribe` or pass `WithExclusive()`.
+
 ## Benchmarks
 
 You can run benchmarks locally; for example, to write and flush 100k messages, you can run:
@@ -186,13 +247,38 @@ You can run benchmarks locally; for example, to write and flush 100k messages, y
 go test -bench=. -benchtime=100000x
 ```
 
-On a local Macbook with an M3 Max core, this results in `8492 msgs/sec`:
+`BenchmarkOutbox_WriteAndPublishThroughput` drains each topic with a busy-polling `ProcessMessages` loop; `BenchmarkOutbox_SubscribeThroughput` drains each topic with a `Subscribe` call woken by `LISTEN`/`NOTIFY` (its poll interval is set far above the benchmark runtime, so throughput is carried entirely by notifications — and each producer commit pays the in-transaction `pg_notify`). Both run a matrix of 1 and 10 topics (messages spread round-robin, one consumer per topic) and producer batch sizes of 1, 10, and 100 messages per `AddMessages` transaction. On a local Macbook with an M3 Max core:
 
 ```
 $ go test -bench=. -benchtime=100000x
 goos: darwin
 goarch: arm64
 pkg: github.com/hatchet-dev/pgoutbox
 cpu: Apple M3 Max
-BenchmarkOutbox_WriteAndPublishThroughput-14              100000            117757 ns/op              8492 msgs/sec
+BenchmarkOutbox_WriteAndPublishThroughput/Flush/topics=1/batch=1-14         	  100000	    125855 ns/op	      7946 msgs/sec
+BenchmarkOutbox_WriteAndPublishThroughput/TxFlush/topics=1/batch=1-14       	  100000	    219546 ns/op	      4555 msgs/sec
+BenchmarkOutbox_WriteAndPublishThroughput/Flush/topics=1/batch=10-14        	  100000	     16328 ns/op	     61244 msgs/sec
+BenchmarkOutbox_WriteAndPublishThroughput/TxFlush/topics=1/batch=10-14      	  100000	    133446 ns/op	      7494 msgs/sec
+BenchmarkOutbox_WriteAndPublishThroughput/Flush/topics=1/batch=100-14       	  100000	      6806 ns/op	    146925 msgs/sec
+BenchmarkOutbox_WriteAndPublishThroughput/TxFlush/topics=1/batch=100-14     	  100000	    150929 ns/op	      6626 msgs/sec
+BenchmarkOutbox_WriteAndPublishThroughput/Flush/topics=10/batch=1-14        	  100000	    293680 ns/op	      3405 msgs/sec
+BenchmarkOutbox_WriteAndPublishThroughput/TxFlush/topics=10/batch=1-14      	  100000	    241538 ns/op	      4140 msgs/sec
+BenchmarkOutbox_WriteAndPublishThroughput/Flush/topics=10/batch=10-14       	  100000	     25111 ns/op	     39822 msgs/sec
+BenchmarkOutbox_WriteAndPublishThroughput/TxFlush/topics=10/batch=10-14     	  100000	     48755 ns/op	     20511 msgs/sec
+BenchmarkOutbox_WriteAndPublishThroughput/Flush/topics=10/batch=100-14      	  100000	      4468 ns/op	    223817 msgs/sec
+BenchmarkOutbox_WriteAndPublishThroughput/TxFlush/topics=10/batch=100-14    	  100000	     36192 ns/op	     27631 msgs/sec
+BenchmarkOutbox_SubscribeThroughput/Flush/topics=1/batch=1-14               	  100000	    209316 ns/op	      4777 msgs/sec
+BenchmarkOutbox_SubscribeThroughput/TxFlush/topics=1/batch=1-14             	  100000	    301901 ns/op	      3312 msgs/sec
+BenchmarkOutbox_SubscribeThroughput/Flush/topics=1/batch=10-14              	  100000	     21391 ns/op	     46749 msgs/sec
+BenchmarkOutbox_SubscribeThroughput/TxFlush/topics=1/batch=10-14            	  100000	    153971 ns/op	      6495 msgs/sec
+BenchmarkOutbox_SubscribeThroughput/Flush/topics=1/batch=100-14             	  100000	      7479 ns/op	    133706 msgs/sec
+BenchmarkOutbox_SubscribeThroughput/TxFlush/topics=1/batch=100-14           	  100000	    141189 ns/op	      7083 msgs/sec
+BenchmarkOutbox_SubscribeThroughput/Flush/topics=10/batch=1-14              	  100000	    289379 ns/op	      3456 msgs/sec
+BenchmarkOutbox_SubscribeThroughput/TxFlush/topics=10/batch=1-14            	  100000	    924623 ns/op	      1082 msgs/sec
+BenchmarkOutbox_SubscribeThroughput/Flush/topics=10/batch=10-14             	  100000	     33516 ns/op	     29836 msgs/sec
+BenchmarkOutbox_SubscribeThroughput/TxFlush/topics=10/batch=10-14           	  100000	     58475 ns/op	     17101 msgs/sec
+BenchmarkOutbox_SubscribeThroughput/Flush/topics=10/batch=100-14            	  100000	      5352 ns/op	    186840 msgs/sec
+BenchmarkOutbox_SubscribeThroughput/TxFlush/topics=10/batch=100-14          	  100000	     39021 ns/op	     25627 msgs/sec
 ```
+
+Batching producer writes is the single biggest lever: staging 100 messages per transaction reaches ~150-220k msgs/sec on the cheap-flusher path, roughly 20× the one-message-per-transaction rate. The `batch=1` cells run long enough to be sensitive to background database noise (checkpoints, autovacuum), so expect swings between runs — the 1082 msgs/sec outlier above measures ~3300 msgs/sec in isolation.

examples/go.mod

@@ -10,13 +10,18 @@ require (
 )
 
 require (
+	github.com/google/uuid v1.6.0 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
 	github.com/jackc/puddle/v2 v2.2.2 // indirect
+	github.com/mattn/go-colorable v0.1.14 // indirect
+	github.com/mattn/go-isatty v0.0.21 // indirect
 	github.com/mfridman/interpolate v0.0.2 // indirect
 	github.com/pressly/goose/v3 v3.27.1 // indirect
+	github.com/rs/zerolog v1.35.1 // indirect
 	github.com/sethvargo/go-retry v0.3.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/sync v0.20.0 // indirect
+	golang.org/x/sys v0.43.0 // indirect
 	golang.org/x/text v0.36.0 // indirect
 )

examples/go.sum

@@ -55,6 +55,8 @@ github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ
 github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I=
 github.com/magiconair/properties v1.8.10 h1:s31yESBquKXCV9a/ScB3ESkOjUYYv+X0rg8SYxI99mE=
 github.com/magiconair/properties v1.8.10/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0=
+github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE=
+github.com/mattn/go-colorable v0.1.14/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8=
 github.com/mattn/go-isatty v0.0.21 h1:xYae+lCNBP7QuW4PUnNG61ffM4hVIfm+zUzDuSzYLGs=
 github.com/mattn/go-isatty v0.0.21/go.mod h1:ZXfXG4SQHsB/w3ZeOYbR0PrPwLy+n6xiMrJlRFqopa4=
 github.com/mfridman/interpolate v0.0.2 h1:pnuTK7MQIxxFz1Gr+rjSIx9u7qVjf5VOoM/u6BbAxPY=
@@ -91,6 +93,8 @@ github.com/pressly/goose/v3 v3.27.1 h1:6uEvcprBybDmW4hcz3gYujhARhye+GoWKhEWyzD5s
 github.com/pressly/goose/v3 v3.27.1/go.mod h1:maruOxsPnIG2yHHyo8UqKWXYKFcH7Q76csUV7+7KYoM=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
+github.com/rs/zerolog v1.35.1 h1:m7xQeoiLIiV0BCEY4Hs+j2NG4Gp2o2KPKmhnnLiazKI=
+github.com/rs/zerolog v1.35.1/go.mod h1:EjML9kdfa/RMA7h/6z6pYmq1ykOuA8/mjWaEvGI+jcw=
 github.com/sethvargo/go-retry v0.3.0 h1:EEt31A35QhrcRZtrYFDTBg91cqZVnFL2navjDrah2SE=
 github.com/sethvargo/go-retry v0.3.0/go.mod h1:mNX17F0C/HguQMyMyJxcnU471gOZGxCLyYaFyAZraas=
 github.com/shirou/gopsutil/v4 v4.26.3 h1:2ESdQt90yU3oXF/CdOlRCJxrP+Am1aBYubTMTfxJ1qc=

examples/main.go

@@ -3,9 +3,11 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"log"
 	"os"
+	"time"
 
 	"github.com/hatchet-dev/pgoutbox"
 	"github.com/hatchet-dev/pgoutbox/sqlc"
@@ -19,7 +21,7 @@ const (
 
 type printFlusher struct{}
 
-func (printFlusher) Flush(_ context.Context, msgs []*sqlc.Message) error {
+func (printFlusher) Flush(_ pgoutbox.FlushContext, msgs []*sqlc.Message) error {
 	for _, m := range msgs {
 		fmt.Printf("  flushed id=%d topic=%s payload=%s\n", m.ID, m.Topic, string(m.Payload))
 	}
@@ -42,12 +44,27 @@ func main() {
 	}
 	defer pool.Close()
 
-	outbox, err := pgoutbox.NewOutbox(pool, pgoutbox.WithSchema(schema))
+	// LISTEN/NOTIFY pubsub: Subscribe wakes as soon as AddMessages commits
+	// instead of waiting out its poll interval.
+	pubsub, err := pgoutbox.NewPGPubSub(ctx, pool)
+	if err != nil {
+		log.Fatalf("create pubsub: %v", err)
+	}
+
+	outbox, err := pgoutbox.NewOutbox(ctx, pool, pgoutbox.WithSchema(schema), pgoutbox.WithPubSub(pubsub))
 	if err != nil {
 		log.Fatalf("create outbox: %v", err)
 	}
 	outbox.AddFlusher(topic, printFlusher{})
 
+	subCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	done := make(chan error, 1)
+	go func() {
+		done <- outbox.Subscribe(subCtx, topic, pgoutbox.WithPollInterval(30*time.Second))
+	}()
+
 	tx, err := pool.Begin(ctx)
 	if err != nil {
 		log.Fatalf("begin: %v", err)
@@ -67,11 +84,15 @@ func main() {
 		log.Fatalf("commit: %v", err)
 	}
 
-	fmt.Printf("staged %d messages on topic %q in schema %q\n", len(msgs), topic, schema)
+	fmt.Printf("staged %d messages on topic %q in schema %q; waiting for the subscriber...\n", len(msgs), topic, schema)
+
+	// The commit above delivers the notification, so the subscriber flushes
+	// well within this window despite the 30s poll interval.
+	time.Sleep(2 * time.Second)
+	cancel()
 
-	fmt.Println("processing...")
-	if _, err := outbox.ProcessMessages(ctx, topic); err != nil {
-		log.Fatalf("process: %v", err)
+	if err := <-done; err != nil && !errors.Is(err, context.Canceled) {
+		log.Fatalf("subscribe: %v", err)
 	}
 	fmt.Println("done")
 }