pglogreplsimple

README

pglogreplsimple

A wrapper around github.com/jackc/pglogrepl. It represents the decoded WAL stream as an iterator and handles feedback to the server and (re)connection behind the scenes.

A more elaborate example can be found in the example directory.

Installation

go get -u github.com/tfoertsch123/pglogreplsimple

Documentation

Overview

Package pglogreplsimple provides a high-level wrapper around the github.com/jackc/pglogrepl package for consuming PostgreSQL logical replication streams.

It manages the full lifecycle of a logical replication receiver: connecting to the database, identifying the system, validating the replication slot, starting replication, receiving WAL messages, and sending periodic standby status updates (feedback) to advance the confirmed-flush LSN.

Basic usage

Create a Receiver with NewReceiver, optionally specifying parameters, accepted plugins, and a starting LSN via the With* option functions. Then iterate over the channel returned by Receiver.Produce to receive WAL messages as they arrive:

r := pglogreplsimple.NewReceiver(
    pglogreplsimple.WithParams(&pglogreplsimple.Param{
        Logger:    log.L(),
        ConnInfo:  "host=localhost dbname=mydb",
        SlotName:  "my_slot",
    }),
)

it, err := r.Produce(ctx)
if err != nil {
	// handle error
}
for msg := range it {
    switch dat := msg.(type) {
    case *pglogrepl.XLogData:
        // process WAL data ...
        r.AckLSN(dat.WALStart)
    case *pglogrepl.PrimaryKeepaliveMessage:
        r.AckLSN(dat.ServerWALEnd)
    case *pgproto3.NoticeResponse:
        // process NOTICE from server
    }
}

A more elaborate example can be found in the "example" directory.

Accepted plugins

The Receiver only connects to slots using a logical-decoding plugin listed in its accepted-plugins map. DefaultPlugins includes sensible defaults for the "wal2json" and "pgoutput" plugins. Override the set with WithAcceptedPlugins.

Reconnection and error handling

On connection or receive errors the Receiver automatically retries after Param.ErrorRetryInterval (default 5 s). If the parent context is canceled or Receiver.Shutdown is called, the Receiver stops and the iterator ends. The terminal error can be retrieved with Receiver.Err.

Reloading parameters at runtime

Parameters can be changed without restarting the process. Call Receiver.RequestReload with a new Param; the updated values are applied at the next loop iteration. Changing ConnInfo or SlotName triggers a reconnection.

Acknowledging WAL positions

The Receiver sends standby status updates to the server at regular intervals (see Param.FeedbackInterval, default 10 s). The confirmed- flush LSN in those updates determines how much WAL the server retains. Use Receiver.AckLSN from within the Produce loop to advance the write, flush, and replay positions reported to the server. Positions are monotonic and never move backwards.

Index

• Constants
• Variables
• type Logger
• type MsgItem
• type Next
  • func (nxt Next) String() string
• type Opt
  • func WithAcceptedPlugins(x map[string][]string) Opt
  • func WithParams(x *Param) Opt
  • func WithStartLSN(x pglogrepl.LSN) Opt
• type Param
• type Receiver
  • func NewReceiver(p ...Opt) *Receiver
  • func (r *Receiver) AckLSN(write pglogrepl.LSN, other ...pglogrepl.LSN)
  • func (r *Receiver) Close() error
  • func (r *Receiver) Err() error
  • func (r *Receiver) Params() Param
  • func (r *Receiver) Plugin() string
  • func (r *Receiver) Produce(ctx context.Context) (iter.Seq[MsgItem], error)
  • func (r *Receiver) RequestReload(p Param)
  • func (r *Receiver) Shutdown(err error)
  • func (r *Receiver) State() Next

Constants

const DefaultErrorRetryInterval = 5 * time.Second

DefaultErrorRetryInterval is the duration waited before retrying a failed connection attempt when no explicit interval is set.

const DefaultFeedbackInterval = 10 * time.Second

DefaultFeedbackInterval is the default interval at which standby status updates (feedback) are sent to the server.

Variables

var (
	// ErrNoLogger is returned when the Receiver is configured without a
	// logger.  A logger must be set before [Receiver.Produce] is called.
	ErrNoLogger = errors.New("Logger not set")

	// ErrReceiverLocked is returned by [Receiver.Produce] or
	// [Receiver.Close] when another goroutine is already running a Produce
	// iterator on the same Receiver.
	ErrReceiverLocked = errors.New("Receiver is locked by another thread")

	// ErrReceiverStopped is returned by [Receiver.Produce] or
	// [Receiver.Close] when the Receiver has already been stopped and
	// cannot be used again.
	ErrReceiverStopped = errors.New("Receiver was stopped before")

	// ErrPlugin is the shutdown cause set when the replication slot's
	// logical-decoding plugin is not in the Receiver's accepted-plugins
	// map.
	ErrPlugin = errors.New("Wrong decoding plugin")

	// ErrConfirmedFlushLSN is the shutdown cause set when the slot's
	// confirmed_flush_lsn is ahead of the requested start LSN, which
	// would mean skipping WAL data that has not yet been acknowledged.
	ErrConfirmedFlushLSN = errors.New("ConfirmedFlushLSN is in the future")
)

Package-level error values returned by the Receiver API or stored as a shutdown cause retrievable via Receiver.Err.

var DefaultPlugins = map[string][]string{
	"wal2json": []string{
		`"format-version" '2'`,
		`"include-types" 'true'`,
		`"include-xids" 'true'`,
		`"include-timestamp" 'true'`,
		`"include-lsn" 'true'`,
		`"include-pk" 'true'`,
		`"numeric-data-types-as-string" 'true'`,
	},
	"pgoutput": []string{
		`"proto_version" '1'`,
		`"messages" 'true'`,
		`"publication_names" 'all_tables'`,
	},
}

DefaultPlugins is the default set of accepted logical-decoding plugins and their option arguments. It supports the "wal2json" and "pgoutput" plugins. Pass it (or a subset) to WithAcceptedPlugins to override the defaults.

Types

type Logger

type Logger interface {
	Error(string)
	Warn(string)
	Info(string)
	Debug(string)
	Debg2(string)
	Debg3(string)
	Debg4(string)
	Debg5(string)
	Errorf(string, ...interface{})
	Warnf(string, ...interface{})
	Infof(string, ...interface{})
	Debugf(string, ...interface{})
	Debg2f(string, ...interface{})
	Debg3f(string, ...interface{})
	Debg4f(string, ...interface{})
	Debg5f(string, ...interface{})
}

Logger is an interface type representing the expected logger interface The github.com/tfoertsch123/log module provides a logger with this interface.

type MsgItem

type MsgItem interface{}

MsgItem is a type that is either a *pglogrepl.PrimaryKeepaliveMessage, a *pglogrepl.XLogData or a *pgproto3.NoticeResponse.

type Next

type Next int8

Next represents the internal state-machine state of a Receiver. It determines what the Receiver.Produce loop does on its next iteration.

const (
	// Connect means the Receiver should (re)establish a connection to the
	// database and start replication.
	Connect Next = iota
	// Recv means the Receiver should wait for and process the next message
	// from the database.
	Recv
	// Break means the iterator's yield function returned false and the
	// Produce loop should exit.  A subsequent Produce call can resume.
	Break
	// Stop means the Receiver has shut down and the Produce loop should
	// exit without the possibility of resuming.
	Stop
)

Next states. The Receiver transitions between these as it connects, receives messages, pauses on error, and shuts down.

func (Next) String

func (nxt Next) String() string

String returns the name of the state for logging and debugging.

type Opt

type Opt func(*rOpts)

Opt is a configuration option applied to a new Receiver. Use the provided option functions (e.g. WithParams, WithAcceptedPlugins, WithStartLSN) to construct an Opt.

func WithAcceptedPlugins

func WithAcceptedPlugins(x map[string][]string) Opt

WithAcceptedPlugins sets the map of accepted logical-decoding plugin names to their plugin-option argument lists. The slot's plugin must appear as a key in this map or the Receiver will refuse to start replication. By default the plugins in DefaultPlugins are accepted. However, this module does not depend on any specific plugin behavior or data format. So, your own plugin should work, as well.

func WithParams

func WithParams(x *Param) Opt

WithParams sets the initial Param values for the Receiver. The Param is copied and applied during the first call to Receiver.Produce.

func WithStartLSN

func WithStartLSN(x pglogrepl.LSN) Opt

WithStartLSN sets the starting LSN used when the Receiver connects for the first time and no confirmed-flush LSN has been recorded yet. If zero, the slot's confirmed_flush_lsn is used.

type Param

type Param struct {
	CloseOnActivation  chan<- struct{}
	ConnInfo           string
	SlotName           string
	ErrorRetryInterval time.Duration
	FeedbackInterval   time.Duration
	Logger             Logger
}

Param holds the configuration parameters for a Receiver. It is supplied via WithParams when the Receiver is created and can be updated at runtime via Receiver.RequestReload. The CloseOnActivation channel will be closed when these parameters are activated. This can be used for instance if the logger is changed as the result of a reload request to close the connected log file.

type Receiver

type Receiver struct {
	// contains filtered or unexported fields
}

Receiver manages a single logical replication connection to a PostgreSQL database. It connects to the server, starts replication on a named slot, and yields WAL messages to the caller via the Receiver.Produce iterator.

A Receiver is created with NewReceiver and used by calling Produce in a range loop. During iteration the caller calls Receiver.AckLSN to advance the confirmed-flush LSN. The Receiver handles reconnection, standby feedback, and graceful shutdown internally.

func NewReceiver

func NewReceiver(p ...Opt) *Receiver

NewReceiver creates a new Receiver configured with the given options. If no WithParams option is supplied, default values are used for the error-retry and feedback intervals; a logger and connection info must be provided via WithParams before calling Receiver.Produce.

Typical usage:

r := NewReceiver(
    WithParams(&Param{Logger: lg, ConnInfo: conninfo, SlotName: slot}),
    WithStartLSN(startLSN),
)
it, err := r.Produce(ctx)
if err != nil {
	// handle error
}
for msg := range it {
    // handle msg ...
    r.AckLSN(lsn)
}

func (*Receiver) AckLSN

func (r *Receiver) AckLSN(write pglogrepl.LSN, other ...pglogrepl.LSN)

AckLSN advances the LSN positions reported to the server in the next standby status update. It should be called from within the Receiver.Produce loop after processing a message.

With a single argument, the write, flush, and replay positions are all set to at least the given LSN. With two arguments, the second sets both the flush and replay positions. With three arguments, the second sets the flush position and the third sets the replay position. Positions are never moved backwards.

func (*Receiver) Close

func (r *Receiver) Close() error

Close shuts down the Receiver. It sets the internal state to Stop and closes the underlying database connection if one is open. Close must not be called concurrently with an active Receiver.Produce iterator; if the iterator is running, use Receiver.Shutdown instead.

The main use of Close is as follows:

it, err := r.Produce(ctx)
if err != nil {
	// handle error
}
for msg := range it {
	// handle msg ...
	if some condition {
		break
	}
}
switch r.State() {
case Stop:
	// r is done and cannot be resumed
case Break:
	// the for loop was exited by "break".
	// at this point another iterator can be created using r.Produce()
	// or the r object can be shut down using
	r.Close()
}

Close returns ErrReceiverLocked if the Receiver is currently producing, ErrReceiverStopped if it has already been stopped, or any error returned by closing the connection.

func (*Receiver) Err

func (r *Receiver) Err() error

Err returns the error that caused the Receiver to shut down, or nil if it stopped without an error. It should be called after the Receiver.Produce iterator has ended.

func (*Receiver) Params

func (r *Receiver) Params() Param

Params returns a copy of the Receiver's current configuration parameters. The returned value reflects the most recently applied Param (either from WithParams or Receiver.RequestReload).

func (*Receiver) Plugin

func (r *Receiver) Plugin() string

Plugin returns the name of the logical-decoding plugin in use by the current connection. It is set after a successful connection and is empty before the first successful connection.

func (*Receiver) Produce

func (r *Receiver) Produce(ctx context.Context) (iter.Seq[MsgItem], error)

Produce starts the replication receive loop and returns an iterator over WAL messages. Each yielded item is either

• a *pglogrepl.PrimaryKeepaliveMessage or
• a *pglogrepl.XLogData or
• a *pgproto3.NoticeResponse (see MsgItem).

If the "for range" iterator style does not fit your needs, the returned iter.Seq can easily be converted into a pull-style iterator.

Postgres expects to be called back from time to time. If it does not get this feedback message it deems the other end dead and closes the connection. See wal_sender_timeout. This limits the time one message must be processed in.

If the "for range" loop is exited with "break", message consumption can be resumed by calling Receiver.Produce again and creating a new iterator. The Receiver.State function can be used to distinguish between this situation and a normal shutdown.

A Receiver in Break state still holds all the resources. If you don't want to resume consumption, you need to Receiver.Close it.

Produce must not be called concurrently with itself or with Receiver.Close. If the Receiver is already producing, it returns ErrReceiverLocked; if the Receiver has been stopped, it returns ErrReceiverStopped.

The provided ctx controls the lifetime of the entire replication session. When ctx is canceled the Receiver shuts down and the iterator ends. Passing nil is equivalent to context.Background().

The caller should call Receiver.AckLSN from within the loop to advance the confirmed-flush LSN sent in standby status updates.

func (*Receiver) RequestReload

func (r *Receiver) RequestReload(p Param)

RequestReload queues a parameter update for the Receiver. The updated Param is applied at the top of the next Receiver.Produce loop iteration. If the connection info or slot name changed, the Receiver reconnects; other fields (intervals, logger) take effect immediately. If the Receiver is currently blocked waiting for a message, the wait is interrupted so the reload is applied without delay. The optional Param.CloseOnActivation channel is closed when the parameter package has been activated. This can be used to close an old logfile for instance.

func (*Receiver) Shutdown

func (r *Receiver) Shutdown(err error)

Shutdown signals the Receiver to stop. The provided err is stored as the cause of the shutdown and can be retrieved later with Receiver.Err. Shutdown is safe to call from any goroutine and is the preferred way to stop a Receiver whose Receiver.Produce iterator is currently running.

func (*Receiver) State

func (r *Receiver) State() Next

State returns the current internal state of the Receiver. This is primarily useful for debugging or monitoring. See Next for the possible values.