pgx Listen and Notify for Change Events in Go Services

The basics of LISTEN and NOTIFY, a trigger, a payload and a loop around WaitForNotification, are covered in LISTEN and NOTIFY from Go, including the facts that make it unreliable: notifications are not stored, they are lost when the listener is away, and they are merged within a transaction. This article takes the next step and builds the version you can run in a service, the one that survives the listener's connection dying. It runs against PostgreSQL 18.6 from Go 1.25.6 with pgx v5.11.0, and proves the recovery by killing the listener's session in the middle of a stream of inserts.

It belongs to the database cornerstone. The same technique, a hint plus a table, drives the outbox relay and the wake up of a Postgres job queue.

The design

Three rules make a listener reliable.

1. A notification is a hint, not data. The payload carries at most an id, and the handler reads the table for everything newer than the last id it processed. If the handler trusted payloads, a lost or merged notification would be a lost event. If it reads the table, a lost notification costs only latency.
2. Subscribe first, then catch up. On every connect: run LISTEN, and only then query for rows newer than the last seen. The reverse order has a window between the query and the LISTEN in which an insert's notification is missed and its row is not in the query result.
3. Reconnect forever. When WaitForNotification returns an error, the connection is gone. Close it, wait briefly, connect again and repeat from rule 2.

The listener must also be a dedicated connection, not one borrowed from a pool for each call: LISTEN is session state, and a pooled connection that is returned and reused carries the subscription to whoever gets it next.

The code

package main

import (
	"context"
	"fmt"
	"log"
	"sync"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

const dsn = "postgres://postgres:[email protected]:55432/lab?sslmode=disable&application_name=listen671"

func must(err error) {
	if err != nil {
		log.Fatal(err)
	}
}

// Listener keeps one connection LISTENing, reconnects when it breaks, and runs catch-up after every (re)connect
// because a notification sent while it was away is gone for good.
type Listener struct {
	dsn      string
	channel  string
	catchUp  func(ctx context.Context) error // reads everything newer than what the handler has seen
	handle   func(payload string)
	mu       sync.Mutex
	connects int
}

func (l *Listener) connections() int {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.connects
}

func (l *Listener) Run(ctx context.Context) {
	for ctx.Err() == nil {
		conn, err := pgx.Connect(ctx, l.dsn)
		if err != nil {
			time.Sleep(100 * time.Millisecond)
			continue
		}
		// Subscribe first, query second, so nothing slips between the two.
		if _, err := conn.Exec(ctx, "listen "+l.channel); err != nil {
			conn.Close(ctx)
			continue
		}
		l.mu.Lock()
		l.connects++
		l.mu.Unlock()
		if err := l.catchUp(ctx); err != nil {
			log.Println("catch up:", err)
		}
		for {
			n, err := conn.WaitForNotification(ctx)
			if err != nil {
				break // the connection is gone: reconnect
			}
			l.handle(n.Payload)
		}
		conn.Close(context.Background())
	}
}

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	pool, err := pgxpool.New(ctx, dsn)
	must(err)
	defer pool.Close()
	for _, q := range []string{
		"drop table if exists events671",
		"create table events671 (id bigserial primary key, body text not null)",
		`create or replace function notify_event671() returns trigger language plpgsql as $$
		 begin perform pg_notify('events671', new.id::text); return new; end $$`,
		"create trigger events671_notify after insert on events671 for each row execute function notify_event671()",
	} {
		_, err := pool.Exec(ctx, q)
		must(err)
	}

	var mu sync.Mutex
	seen := map[int64]string{} // id -> how the handler learned about it
	var last int64
	read := func(how string) error {
		mu.Lock()
		after := last
		mu.Unlock()
		rows, err := pool.Query(ctx, "select id from events671 where id > $1 order by id", after)
		if err != nil {
			return err
		}
		ids, err := pgx.CollectRows(rows, pgx.RowTo[int64])
		if err != nil {
			return err
		}
		mu.Lock()
		defer mu.Unlock()
		for _, id := range ids {
			if _, ok := seen[id]; !ok {
				seen[id] = how
			}
			if id > last {
				last = id
			}
		}
		return nil
	}
	l := &Listener{
		dsn:     dsn,
		channel: "events671",
		catchUp: func(ctx context.Context) error { return read("catch-up query") },
		// A notification is only a hint: read the table instead of trusting the payload.
		handle: func(payload string) { _ = read("notification") },
	}
	go l.Run(ctx)
	insert := func(n int) {
		for i := 0; i < n; i++ {
			_, err := pool.Exec(ctx, "insert into events671 (body) values ('x')")
			must(err)
		}
	}
	count := func(how string) (n int) {
		mu.Lock()
		defer mu.Unlock()
		for _, h := range seen {
			if h == how {
				n++
			}
		}
		return
	}
	time.Sleep(500 * time.Millisecond)
	fmt.Println("1. listener connected, connections made:", l.connections())

	insert(5)
	time.Sleep(300 * time.Millisecond)
	fmt.Printf("2. five inserts while connected: learned by notification %d, by catch-up query %d\n", count("notification"), count("catch-up query"))

	// Kill the listener's backend, insert while it is away, and let it reconnect.
	var killed int
	must(pool.QueryRow(ctx, "select count(*) from (select pg_terminate_backend(pid) from pg_stat_activity where application_name = 'listen671' and query like 'listen%') t").Scan(&killed))
	insert(4)
	time.Sleep(700 * time.Millisecond)
	fmt.Printf("3. listener's backend ended (%d session), 4 rows inserted around it: connections made %d, caught up by the query %d\n", killed, l.connections(), count("catch-up query")-0)

	mu.Lock()
	total := len(seen)
	mu.Unlock()
	var inTable int
	must(pool.QueryRow(ctx, "select count(*) from events671").Scan(&inTable))
	fmt.Printf("4. rows in the table %d, rows the handler knows about %d: nothing was lost\n", inTable, total)

	insert(3)
	time.Sleep(300 * time.Millisecond)
	fmt.Printf("5. after reconnecting, notifications work again: learned by notification %d in total\n", count("notification"))
}

1. listener connected, connections made: 1
2. five inserts while connected: learned by notification 5, by catch-up query 0
3. listener's backend ended (1 session), 4 rows inserted around it: connections made 2, caught up by the query 4
4. rows in the table 9, rows the handler knows about 9: nothing was lost
5. after reconnecting, notifications work again: learned by notification 8 in total

Listener.Run is the loop. It connects with pgx.Connect, a standalone connection outside any pool, runs listen events671, counts the connection, runs the catchUp function and then blocks in WaitForNotification, calling handle for each notification. When that call returns an error, the inner loop ends, the connection is closed and the outer loop connects again after the next iteration. Cancelling the context ends everything.

The handler and the catch-up both call the same function, read, which selects the ids greater than the last one seen and records how each was learned. The notification handler ignores the payload completely. That is the point: whether the hint came from a trigger, a manual notify, or a retry, the handler does the same thing, and a hint that arrives twice or not at all changes nothing about correctness.

The trigger on the table sends pg_notify('events671', new.id::text) after each insert, as in the earlier article. The id in the payload is only informative here.

The experiment

While connected, five inserts produced five notifications and the handler learned all five through them: learned by notification 5, by catch-up query 0. The path works in the normal case, with latency measured in milliseconds.

Then the program ended the listener's session on the server with pg_terminate_backend, finding it by its application_name and its last query beginning with listen. It inserted four rows around that moment. The listener's connection died, WaitForNotification returned an error, and the loop reconnected: the log shows connections made 2. Some of the four rows were committed while no connection was listening, so their notifications were never delivered to anyone. They were learned anyway: the catch-up query after the reconnect returned all 4, because they were in the table with ids above the last seen.

The check at the end compares the table with the handler's knowledge: 9 rows in the table, 9 known. Nothing was lost across the outage. A final batch of three inserts proves the reconnected listener works: the total learned through notifications reached 8 (five before the outage plus three after), and the four from the gap came through the catch-up query.

The experiment is deterministic enough to be a test in a real service: start the listener, insert, kill the session, insert, wait, assert the handler saw every id. It exercises exactly the failure that production produces during a database failover, a deployment of a pooler, or a network blip.

Details that matter

Backoff on reconnect. The loop sleeps 100 milliseconds between failed connection attempts. In production use exponential backoff with a cap and jitter, so a database that is coming back is not hit by every service at once, and log reconnects at warn level with a counter: frequent reconnects mean something is dropping connections.

A periodic catch-up. Add a ticker that runs the catch-up query every 30 seconds or so. It covers the rarest failure: a notification lost on a connection that did not error (a server restart that the client did not notice for a while, or a bug). The cost is one cheap indexed query.

Ordering and duplicates. Reading by id means events are handled in id order, but ids are assigned at insert time and become visible at commit time, so a transaction that took a lower id can commit after one with a higher id. A catch-up that advances last to the highest id seen can skip a row that commits later. For strict correctness, use a visibility safe cursor, such as tracking the transaction id (xmin) with a safe horizon, or accept at least once semantics by re-reading a small window below last and ignoring duplicates through an idempotent handler. The outbox article discusses the same issue.

Pool interplay. Keep the listener's connection outside the pool, as here, and use a pool for the reads. If your pool is behind PgBouncer in transaction mode, the listener must connect directly or through a session pooled port, since LISTEN does not survive transaction pooling. See pgx with PgBouncer.

Several channels and many listeners. One connection can listen to many channels; the payload or the channel name tells the handler what to read. Many service instances can each listen; each receives every notification, so make the handler idempotent or let only one instance act, using a lock or a queue as in advisory locks.

Shutdown. Cancel the context and the WaitForNotification call returns; close the connection with a context that is not cancelled so the UNLISTEN and close complete.

When to use it

Use this for low rate signals that make a service react sooner than a poll would: cache invalidation, waking workers, refreshing a materialised view, pushing a change to connected clients through one process. Do not use it as the only delivery of something that must happen, and never without the table behind it. For events that must be processed exactly in order with guarantees, the answer is a queue table consumed with SKIP LOCKED or a logical replication stream, as in a pgx logical replication client. The WaitForNotification method is documented in the pgx reference, and the server side in the PostgreSQL manual's page on NOTIFY.