Documentation
¶
Overview ¶
Package pgnotify turns Postgres LISTEN/NOTIFY into a wake-up signal for a poller.
It is not a message transport and must never be used as one. NOTIFY is at-most-once and connection-scoped: a listener that is reconnecting misses everything sent in the gap, and there is no replay. What it is good for is telling a loop that already knows how to find its work that there is work now, so an idle queue drains in milliseconds instead of a poll interval and an idle process stops issuing a query per tick.
The shape ¶
A Listener holds one dedicated connection, issues LISTEN, and converts every notification into a send on a coalescing channel:
listener, err := pgnotify.NewListener(ctx, &pgnotify.Config{
ConnectionString: dsn,
Channel: "outbox",
}, pgnotify.WithLogger(logger), pgnotify.WithMetricsProvider(metricsProvider))
if err != nil {
return err
}
go listener.Run()
defer func() { _ = listener.Close(ctx) }()
relay, err := outbox.NewRelay(ctx, cfg, client, provider,
outbox.WithRelayWakeup(listener.Signal()))
The consumer never learns Postgres is involved: it takes a <-chan struct{} and can be tested with a bare channel. The producer side is equally thin — one pg_notify with an empty payload, run on whatever executor is already in hand, which is what outbox.WithWriterNotifyChannel and workqueue's NotifyChannel emit.
Signal, not stream ¶
Signal is edge-triggered and level-collapsed. The channel has capacity one and the send is non-blocking, so a burst of notifications produces at most one pending wake, and a consumer that reads it once has absorbed all of them. That is the correct semantic for a poller: it re-reads the table when it wakes, so the number of notifications it missed does not matter — only that it wakes.
Payloads are ignored entirely. Notifications carry none, which makes Postgres collapse duplicate (channel, payload) pairs within a transaction for free, and means nothing in the system can come to depend on the contents of a signal that is allowed to be lost.
Every reconnect is a gap ¶
A session begins with an unconditional signal, the first connect included. The listener cannot know what was sent while it was away, so it does not try: it wakes the consumer, and the consumer's own query is what establishes the truth.
The listener connection does nothing else ¶
Postgres buffers undelivered notifications in a cluster-wide 8 GB async queue. A listener that is connected but not draining fills it, and at that point *every committing transaction in the cluster* begins to fail. A slow listener is a cluster-wide write outage, which is why the loop here hands off to a channel with a non-blocking send and immediately waits again. It never does work inline, and a consumer that is behind loses wakes rather than backing up the server.
Deployment constraints ¶
**PgBouncer in transaction or statement pooling mode breaks LISTEN outright.** The session is not yours between transactions, so the LISTEN is issued on a connection that is handed to somebody else. This is the most common way this feature silently does not work. Use session pooling or a direct connection.
**NOTIFY does not cross replication.** A listener must reach the primary; a replica DSN will connect, listen, and never hear anything.
**The connection is held for the process's lifetime.** Config takes its own connection string rather than borrowing from an existing pool, because postgres.PgxAccess's pools cap the union of the native and database/sql surfaces — a borrowed listener connection would permanently cost the application one pool slot.
Channel names ¶
A channel name is validated with dialect.ValidIdentifier and bounded by MaxChannelLength, and the LISTEN it renders is quoted. That quoting is a correctness requirement, not just an injection guard: pg_notify takes its channel as text and compares byte-for-byte, while an unquoted LISTEN identifier would be down-cased. Quoting both sides is what makes "Outbox" on the producer match "Outbox" on the listener. Prefer lowercase names regardless.
Watching it ¶
Pass WithMetricsProvider. postgres_listener_notifications_received against postgres_listener_wakes_coalesced tells you how much a burst is actually collapsing, and postgres_listener_reconnects against postgres_listener_connect_errors is how you learn the listener is flapping — which, because every reconnect fires a catch-up cycle, otherwise looks like a consumer that is simply busier than usual.
Individual notifications are not traced. A root span per wake would be one span per enqueue for the whole fleet, and the span that matters — the cycle the wake triggered — belongs to the consumer.
Index ¶
Constants ¶
const ( // MaxChannelLength bounds a channel name. It is Postgres' NAMEDATALEN - 1: // the server truncates anything longer, and a truncated name is a name that // may no longer be the one the producer is notifying. MaxChannelLength = 63 // DefaultMinReconnectBackoff is how long the listener waits before its first // reconnect attempt. DefaultMinReconnectBackoff = 100 * time.Millisecond // DefaultMaxReconnectBackoff caps the reconnect delay. It is deliberately // short for a backoff ceiling: while the listener is away the consumer is // running on its poll interval alone, so this bounds how long a cluster // failover degrades latency rather than how long a doomed connection is // retried. DefaultMaxReconnectBackoff = 30 * time.Second )
Variables ¶
var ( // ErrNilConfig indicates a nil Config was passed to NewListener. It wraps // errors.ErrNilInputParameter, so a caller may check either. ErrNilConfig = platformerrors.Wrap(platformerrors.ErrNilInputParameter, "nil postgres listener config") // ErrChannelTooLong indicates a channel name longer than MaxChannelLength. // It is reported rather than left to the server, which would truncate it // silently — and truncate the producer's name to something that may or may // not still match. ErrChannelTooLong = platformerrors.New("postgres notification channel name is too long") )
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
// ConnectionString is the DSN the listener dials. It must reach the primary:
// NOTIFY does not cross replication, so a replica connects, listens, and
// hears nothing forever.
//
// It must also not pass through a pooler in transaction or statement
// pooling mode, which breaks LISTEN outright. See the package
// documentation.
ConnectionString string `env:"CONNECTION_STRING" json:"connectionString,omitempty" yaml:"connectionString,omitempty"`
// Channel is the NOTIFY channel to listen on. It is validated as a SQL
// identifier because LISTEN is a utility statement and cannot bind
// parameters — the name is rendered into the statement, quoted, rather than
// bound.
//
// It must match the name the producer notifies byte for byte; prefer
// lowercase.
Channel string `env:"CHANNEL" json:"channel,omitempty" yaml:"channel,omitempty"`
// MinReconnectBackoff is the delay before the first reconnect attempt,
// doubling up to MaxReconnectBackoff and resetting once a session is
// established.
MinReconnectBackoff time.Duration `env:"MIN_RECONNECT_BACKOFF" json:"minReconnectBackoff,omitempty" yaml:"minReconnectBackoff,omitempty"`
// MaxReconnectBackoff caps the reconnect delay.
MaxReconnectBackoff time.Duration `env:"MAX_RECONNECT_BACKOFF" json:"maxReconnectBackoff,omitempty" yaml:"maxReconnectBackoff,omitempty"`
}
Config configures a Listener.
There is no pool here, and no database.Client. LISTEN pins a session for the lifetime of the process, and postgres.PgxAccess's pools cap the union of the native and database/sql surfaces — so a listener that borrowed a connection would permanently cost the application one pool slot. It dials its own.
func (*Config) EnsureDefaults ¶
func (cfg *Config) EnsureDefaults()
EnsureDefaults fills unset knobs with the package defaults.
type Listener ¶
type Listener struct {
// contains filtered or unexported fields
}
Listener holds one dedicated Postgres connection, keeps a LISTEN standing on it across reconnects, and turns every notification into a wake on a coalescing channel.
It owns a goroutine started by Run and stopped by Close.
func NewListener ¶
NewListener builds a Listener. It does not connect and does not start it; call Run.
ctx is used to validate the config and is not retained — Run takes its own, because a listener tied to a request or startup context would stop listening the moment that context was cancelled.
func (*Listener) Close ¶
Close stops the listener and waits for its connection to be released. Safe to call more than once, and safe to call on a Listener that was never run.
func (*Listener) Run ¶
func (l *Listener) Run()
Run is the listener loop: connect, LISTEN, wake on every notification, and reconnect with backoff when the session drops.
Like outbox.Relay.Run it takes no context. A listener exists to shorten the latency of a loop that outlives any single request, and Close is how it stops.
Run returns only after Close.
func (*Listener) Signal ¶
func (l *Listener) Signal() <-chan struct{}
Signal is the wake channel, and is safe to read before Run has started.
It is edge-triggered and level-collapsed: a send never blocks, and a pending wake absorbs every notification that arrives before it is read. A reader therefore learns only that something happened, never how much — which is all a poller needs, since it re-reads the table on waking.
Read it from exactly one loop. A wake is delivered to one receiver, so several loops sharing a Listener would each see an arbitrary subset; give each its own Listener, or fan the signal out yourself.
type Option ¶
type Option func(*options)
Option configures a Listener. The zero configuration works: an absent logger logs nowhere, an absent tracer provider traces nowhere, and an absent metrics provider records nothing.
func WithLogger ¶
WithLogger attaches a logger. Nothing here fails loudly — a dropped connection is retried, a coalesced wake is discarded — so without one, a listener that has been flapping for an hour is visible only in metrics.
func WithMetricsProvider ¶
WithMetricsProvider attaches a metrics provider. An absent provider records nothing.
func WithRand ¶
WithRand replaces the source that spreads reconnect backoff across the upper half of its interval. fn must return a value in [0,1]; a value of 1 yields the un-jittered backoff.
The default draws from math/rand/v2 and needs no seeding. A nil fn is ignored.
func WithTracerProvider ¶
WithTracerProvider attaches a tracer provider, which names the span covering each connect attempt. Individual notifications are not traced: a root span per wake is one span per enqueue across the whole fleet, and the work the wake causes belongs to the consumer's trace, not the listener's.
Source Files