The Notifier Pattern for Applications That Use Postgres

Why Postgres Listen/Notify Deserves a Dedicated Connection

Postgres's LISTEN/NOTIFY mechanism remains one of the database's standout features—and one that competing databases still lack. The concept is simple: clients subscribe to topics, and messages sent on those topics are delivered to every subscriber. But applications built on Postgres often use this feature inefficiently, consuming scarce database connections and ignoring failure modes.

The core problem is that a LISTEN is bound to a specific connection. That connection must remain open for the subscription to receive messages. If multiple components in an application each want to listen to different topics, the naive approach is one connection per topic per program. Given that Postgres connections are a finite resource, this becomes wasteful quickly.

The notifier pattern solves this by consolidating all listen/notify activity into a single Postgres connection per process. That connection handles subscriptions for any number of topics, waits for incoming notifications, and distributes them to the components that registered interest. Components that want to listen on a topic simply ask the notifier to add a subscription—they never touch the database connection directly.

This reduces connection overhead from one-per-topic to one-per-program. In languages like Go, where in-process concurrency is cheap, the practical cost of listen/notify drops to almost nothing.

Key Implementation Details

A notifier is straightforward to build. The essential elements are a way to register subscriptions and a loop that waits for notifications and fans them out to subscribers. A few details matter more than others:

• Buffered channels with non-blocking sends. A notifier can receive a high volume of notifications. If it blocked on every subscriber acknowledging receipt, it could fall behind. Using a buffered channel (e.g., make(chan string, 100)) with non-blocking sends means a notification is discarded if a subscriber's buffer is full. Each component is responsible for draining its inbox fast enough; one slow subscriber won't degrade the whole system.
• One LISTEN per topic, regardless of subscriber count. Multiple components may want the same topic. The notifier tracks subscriptions by topic and only issues a LISTEN the first time a topic is requested. Subsequent subscriptions to that topic just add another entry to the internal distribution list.
• An "established" signal for testability. Subscriptions should expose a channel that's closed once the LISTEN has actually been issued. Without this, tests that fire pg_notify before the notifier is listening will lose messages, leading to intermittent failures that are notoriously hard to debug.

// EstablishedC is a channel that's closed after the notifier's successfully
// established a connection. This is especially useful in test cases, where it
// can be used to wait for confirmation that not only that the listener is
// started, but that it's successfully established started listening on a
// channel before continuing. For a new subscription on an already established
// channel, EstablishedC is already closed, so it's always safe to wait on it.
//
// There's no full guarantee that the notifier can ever successfully establish a
// listen, so callers will usually want to `select` on it combined with a
// context done, a stop channel, and/or a timeout.
//
// The channel is always closed as a notifier is stopping.
func (s *Subscription) EstablishedC() <-chan struct{} { return s.establishedChan }

Making the Wait Loop Interruptible

Drivers typically provide a blocking call for waiting on notifications—for example, Pgx's WaitForNotification. That's a problem for a single-connection design: what happens when the notifier is blocked in this wait, and another component requests a new subscription that requires issuing a LISTEN?

The solution is to make the wait interruptible. One pattern is to wrap the driver's wait call in a closure that uses a context with a default timeout—say, 30 seconds—so the wait cycles periodically. Store the cancellation function, and when a new subscription comes in, invoke it to break the wait immediately, process the new LISTENs, then reenter the wait loop.

func (l *Notifier) runOnce(ctx context.Context) error {
    if err := l.processChannelChanges(ctx); err != nil {
        return err
    }

    // WaitForNotification is a blocking function, but since we want to wake
    // occasionally to process new `LISTEN`/`UNLISTEN` operations, we put a
    // context deadline on the listen, and as it expires don't treat it as an
    // error unless it's unrelated to context expiration.
    notification, err := func() (*pgconn.Notification, error) {
        const listenTimeout = 30 * time.Second

        ctx, cancel := context.WithTimeout(ctx, listenTimeout)
        defer cancel()

        // Provides a way for the blocking wait to be cancelled in case a new
        // subscription change comes in.
        l.mu.Lock()
        l.waitForNotificationCancel = cancel
        l.mu.Unlock()

        notification, err := l.conn.WaitForNotification(ctx)
        if err != nil {
            return nil, xerrors.Errorf("error waiting for notification: %w", err)
        }

        return notification, nil
    }()
    if err != nil {
        // If the error was a cancellation or the deadline being exceeded but
        // there's no error in the parent context, return no error.
        if (errors.Is(err, context.Canceled) ||
            errors.Is(err, context.DeadlineExceeded)) && ctx.Err() == nil {
            return nil
        }

        return err
    }

    l.mu.RLock()
    defer l.mu.RUnlock()

    // Notify subscribers (this is a no-op if no subs/empty slice).
    for _, sub := range l.subscriptions[notification.Channel] {
        sub.listenChan <- notification.Payload
    }

    return nil
}

Handling Connection Failure

With a single connection handling all notifications, its health is critical. If it dies, every listen/notify consumer in the program loses service at once.

The obvious recovery path is to close the dead connection, grab a new one from the pool, reissue LISTEN for all active subscriptions, and resume the wait loop. But resetting state cleanly can be fiddly. An alternative is the "let it crash" approach: if the connection becomes irreconcilably unhealthy, stop the whole program and let normal startup restore a healthy state.

// If the notifier gets unhealthy, restart the worker. This will generally
// never happen as the notifier has a built-in retry loop that try its best
// to keep established before giving up.
notifier.AddUnhealthyCallback(closeShutdown)

In practice, catastrophic connection failures for a notifier are rare. When they do happen, letting the process crash is a simple, robust response that avoids subtle state-reset bugs.

PgBouncer Compatibility

A notifier's single connection fits naturally alongside PgBouncer, but only if you understand the pooling modes. LISTEN requires session pooling because notifications are delivered only to the original session that issued the LISTEN. Transaction pooling breaks this assumption.

The notifier pattern works well in this setup: one connection per program is dedicated to listen/notify and held directly, while the rest of the application can use PgBouncer in transaction or statement pooling mode for maximum connection efficiency.