pgoutbox

A service that helps implement the Event-Driven architecture.

To maintain the consistency of data in the system, we will use transactional messaging - publishing events in a single transaction with a domain model change.

The service allows you to subscribe to changes in the PostgreSQL database using its logical decoding capability and publish them to the NATS Streaming server.

Logic of work

To receive events about data changes in our PostgreSQL DB   we use the standard logic decoding module (pgoutput) This module converts changes read from the WAL into a logical replication protocol.   And we already consume all this information on our side. Then we filter out only the events we need and publish them in the queue

Event publishing

As the message broker will be used is of your choice:

• NATS JetStream [type=nats];
• Apache Kafka [type=kafka];
• RabbitMQ [type=rabbitmq].
• Google Pub/Sub [type=google_pubsub].

Service publishes the following structure. The name of the topic for subscription to receive messages is formed from the prefix of the topic, the name of the database and the name of the table prefix + schema_table.

{
	ID        uuid.UUID       # unique ID
	Schema    string
	Table     string
	Action    string
	Data      map[string]any
	DataOld   map[string]any  # old data (see DB-settings note #1)
	EventTime time.Time       # commit time
}

Messages are published to the broker at least once!

Filter configuration example

databases:
  filter:
    tables:
      users:
        - insert
        - update

This filter means that we only process events occurring with the users table, and in particular insert and update data.

Topic mapping

By default, output NATS topic name consist of prefix, DB schema, and DB table name, but if you want to send all update in one topic you should be configured the topic map:

topicsMap:
  main_users: "notifier"
  main_customers: "notifier"

DB setting

You must make the following settings in the db configuration (postgresql.conf)

• wal_level >= “logical”
• max_replication_slots >= 1

The publication & slot created automatically when the service starts (for all tables and all actions). You can delete the default publication and create your own (name: pgoutbox) with the necessary filtering conditions, and then the filtering will occur at the database level and not at the application level.

https://www.postgresql.org/docs/current/sql-createpublication.html

If you change the publication, do not forget to change the slot name or delete the current one.

Notes:

1. To receive DataOld field you need to change REPLICA IDENTITY to FULL as described here: #SQL-ALTERTABLE-REPLICA-IDENTITY

Connection poolers (pgpool-II, PgBouncer)

Point database.host at the PostgreSQL primary, not at a pooler. On Kubernetes that means the primary Service, whose endpoints follow a promotion.

A logical replication stream cannot go through a protocol-aware pooler. pgpool-II implements neither the replication=database startup option — it forwards the startup packet to every backend, standbys included — nor the CopyBoth mode that START_REPLICATION switches the connection into, so the stream stalls and the pooler eventually tears the connection down. Slot and publication statements are unsafe through a pooler too, because read/write splitting sends reads about pg_replication_slots to standbys, where the slot state differs.

For deployments that are already aimed at pgpool, pgoutbox bypasses it rather than failing in that confusing way: it runs SHOW pool_nodes, takes the backend that PostgreSQL itself reports as the primary (pg_role, falling back to pgpool's own role, which lags a failover), and dials that node for both the queries and the replication stream. A plain PostgreSQL endpoint is detected and used as is, so database.host accepts either.

Whichever way the address was obtained, the node is then asked pg_is_in_recovery() before pgoutbox proceeds. A standby answers yes and the connection is retried, because a plain host may point at a replica and pgpool refreshes its view of the primary on a background poll, so it can name a backend that has already been demoted.

Failover

Resolution and verification happen on every connection attempt, nothing is cached across restarts. When the primary changes:

• the connections to the old node break, ending the stream;
• if instead they stay open — a demoted node holding the socket, or a black holed network path — the pg_is_in_recovery() check in the connection loop notices within refreshConnection (15s by default);
• pgoutbox exits non-zero, and the restart resolves the new primary.

Resuming without a gap needs listener.failover: true together with PostgreSQL 17's sync_replication_slots=on, which keeps the slot in step on the standbys. Otherwise the new primary has no slot, a fresh one is created at the current LSN, and changes committed before the promotion but not yet published are lost.

Service configuration

listener:
  slotName: myslot_1
  refreshConnection: 15s
  heartbeatInterval: 10s
  filter:
    tables:
      seasons:
        - insert
        - update
  topicsMap:
    schema_table_name: "notifier"
logger:
  level: info
  fmt: json
database:
  host: localhost # the PostgreSQL primary, or a pgpool endpoint to discover it through
  port: 5432
  name: my_db
  user: postgres
  password: postgres
  debug: false
publisher:
   type: nats
   address: localhost:4222
   topic: "wal_listener"
   topicPrefix: ""
monitoring:
  sentryDSN: "dsn string"
  promAddr: ":2112"

We are using Viper; it means you can override each value via env variables with WAL_ prefix.

for instance: WAL_DATABASE_PORT=5433

Monitoring

Sentry

If you specify an DSN-string for the Sentry project, the next level errors will be posted there via a hook:

• Panic
• Fatal
• Error

Prometheus

You can take metrics by specifying an endpoint for Prometheus in the configuration.

Available metrics

name	description	fields
published_events_total	the total number of published events	subject, table
filter_skipped_events_total	the total number of skipped events	table

Kubernetes

Application initializes a web server (if a port is specified in the configuration) with two endpoints for readiness /ready and liveness /healthz probes.

Docker

You can start the container from the project folder (configuration file is required).

See ./config_example.yml for an example configuration. Be sure to copy the file to the docker image in the Dockerfile prior to running after the build setp ex:

COPY /config.yml .

Сontainer preparation is carried out with the help of a multi-stage build, which creates after itself auxiliary images of a large size. Please don't forget to delete them:

docker image prune --filter label=stage=builder