Help us shape the future of Go! Take the 2026 Go Developer Survey

cluster

package
v1.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 29, 2026 License: MIT Imports: 26 Imported by: 0

Documentation

Overview

Package cluster implements the coordination of the popgres app instances (nodes) that share the same Postgres database.

The nodes exchange short messages with LISTEN/NOTIFY. All messages are sent on a single Postgres channel and are routed to the subscribers by their topic, so that new topics don't require a new LISTEN.

Postgres doesn't restrict LISTEN and NOTIFY with privileges (any role that can connect to the database can use them), so the channel name is derived from a secret stored in the database (see Config.Secret) and each message is signed with it. The messages with an invalid signature are ignored.

The NOTIFY payloads are limited to 8000 bytes by Postgres, so the messages are expected to carry only ids (the larger data should be stored in a table).

Index

Constants

View Source 
const (
	RealtimeSourceHooks       = "hooks"
	RealtimeSourceReplication = "replication"
)

The realtime event sources.

View Source 
const (
	// PublicationName is the Postgres publication with the collection tables.
	PublicationName = "popgres_rt_pub"

	// TopicRealtimeEvents is the cluster topic with the ids of the stored realtime events
	// (comma separated).
	TopicRealtimeEvents = "rt_events"

	// RealtimeEventsTable is the UNLOGGED table of the realtime events (created by the system migrations).
	RealtimeEventsTable = "_realtime_events"

	// MaxTransactionEvents is the max number of realtime events of a single transaction
	// (the changes above it don't send events, eg. a bulk delete of a million rows).
	MaxTransactionEvents = 10000

	// RetainedWALWarningSize is the retained WAL size (in bytes) of the
	// realtime slot above which a warning is logged.
	RetainedWALWarningSize = 1 << 30
)
View Source 
const ChannelPrefix = "popgres_cluster_"

ChannelPrefix is the prefix of the Postgres LISTEN/NOTIFY channel used by the cluster messages (the rest of the name is derived from the cluster secret).

View Source 
const CronRunsTable = "_cron_runs"

CronRunsTable is the UNLOGGED table of the claimed cron job executions (created by the system migrations).

View Source 
const DefaultResyncInterval = time.Minute

DefaultResyncInterval is the default interval of the periodic resync (see Cluster.OnResync).

View Source 
const KVTable = "_kv_ttl"

KVTable is the UNLOGGED table of the shared keys (created by the system migrations).

View Source 
const MaxPayloadSize = 7000

MaxPayloadSize is the max size in bytes of a message payload (the Postgres NOTIFY limit minus the message header).

View Source 
const RateLimitsTable = "_rate_limits"

RateLimitsTable is the UNLOGGED table of the shared rate limit counters (created by the system migrations).

View Source 
const (

	// RealtimeEventsMaxAge is the max age of the published realtime events.
	RealtimeEventsMaxAge = 5 * time.Minute
)

Variables

View Source 
var ErrNotReady = errors.New("the cluster secret is not available yet")

ErrNotReady is returned by Config.Secret when the secret can't be loaded yet (eg. before the system migrations of a new database).

Functions

func LockKey 

func LockKey(name string) int64

LockKey returns the Postgres advisory lock key of the named lock.

The advisory locks are per database, so the same name can be used by multiple popgres databases on the same server.

func SlotName 

func SlotName(database string) string

SlotName returns the name of the realtime logical replication slot of the database.

The name is "popgres_rt_<database>". If the database name has characters other than [a-z0-9_] (they are replaced with "_") or if the name is longer than 63 characters (it is truncated), a hash of the database name is appended, so that eg. the "my-db" and "my_db" databases have different slots.

func XactLock 

func XactLock(db dbx.Builder, name string) error

XactLock acquires the named transaction level advisory lock, waiting if necessary. The lock is released at the end of the transaction.

db is expected to be a transaction (otherwise the lock is released immediately after the statement).

Types

type Cluster 

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

Cluster manages the LISTEN connection of a node and the dispatch of the received messages to the topic subscribers.

The LISTEN connection is dedicated (it is not part of the app pool) and it is automatically reconnected with a backoff in case of an error.

The received messages are dispatched by two goroutines with separate queues: one for the state changes (eg. the collections and settings, see Cluster.Subscribe) and one for the data events that could take longer to handle (eg. the realtime events, see Cluster.SubscribeData).

func New 

func New(config Config) *Cluster

New creates a new Cluster instance.

func (*Cluster) AddRealtimeFailure 

func (c *Cluster) AddRealtimeFailure()

AddRealtimeFailure increments the counter of the realtime replication events that couldn't be broadcasted (see Cluster.RealtimeFailures).

func (*Cluster) Channel added in v1.1.0 

func (c *Cluster) Channel() (string, error)

Channel returns the name of the Postgres LISTEN/NOTIFY channel of the cluster messages.

func (*Cluster) CheckReplicationSupport 

func (c *Cluster) CheckReplicationSupport(ctx context.Context) error

CheckReplicationSupport reports whether the logical replication can be used for the realtime events (wal_level=logical and a role with the REPLICATION attribute).

func (*Cluster) ClaimCronRun 

func (c *Cluster) ClaimCronRun(jobId string, slot time.Time) (bool, error)

ClaimCronRun claims the execution of a cron job for the specified scheduled slot and reports whether the current node should run it.

Only one node can claim the same job and slot.

func (*Cluster) Connected added in v1.1.0 

func (c *Cluster) Connected() bool

Connected reports whether the LISTEN connection of the node is connected (the node doesn't receive the cluster messages while it is disconnected).

func (*Cluster) DeleteExpired 

func (c *Cluster) DeleteExpired() error

DeleteExpired deletes the expired rows of the cluster tables (the old cron runs, the expired keys and rate limits, and the old realtime events).

func (*Cluster) DroppedMessages 

func (c *Cluster) DroppedMessages() int64

DroppedMessages returns the number of the received messages that were dropped because their dispatch queue was full.

func (*Cluster) Enabled 

func (c *Cluster) Enabled() bool

Enabled reports whether the cluster mode is enabled.

The state synchronization of the nodes (eg. the cached collections and the realtime events) works always, while the features with an extra cost per request (eg. the shared rate limits) are used only in cluster mode.

func (*Cluster) FindRealtimeEvents 

func (c *Cluster) FindRealtimeEvents(ids []string) ([]*RealtimeEvent, error)

FindRealtimeEvents returns the stored realtime events with the specified ids (ordered by id).

func (*Cluster) InvalidMessages added in v1.1.0 

func (c *Cluster) InvalidMessages() int64

InvalidMessages returns the number of the received messages that were ignored because of an invalid format or signature.

func (*Cluster) IsLocked 

func (c *Cluster) IsLocked(ctx context.Context, name string) (bool, error)

IsLocked reports whether the named session level advisory lock is held by a session.

func (*Cluster) IsReplicationLeader 

func (c *Cluster) IsReplicationLeader() bool

IsReplicationLeader reports whether the current node consumes the replication slot.

func (*Cluster) KV 

func (c *Cluster) KV() *KV

KV returns the key-value store with expiration shared by all nodes.

func (*Cluster) NodeAlive added in v1.1.0 

func (c *Cluster) NodeAlive(nodeId string) bool

NodeAlive reports whether the specified node is connected to the cluster (its LISTEN connection is active), based on the heartbeats that the nodes store in the key-value store in cluster mode.

It returns false for an invalid node id without querying the database.

func (*Cluster) NodeId 

func (c *Cluster) NodeId() string

NodeId returns the unique id of the current node (app instance).

func (*Cluster) Notify 

func (c *Cluster) Notify(topic string, payload string) error

Notify is similar to Cluster.Publish but the message is sent with the main database pool (it is not part of the current transaction, if any).

func (*Cluster) OnResync added in v1.1.0 

func (c *Cluster) OnResync(fn func()) (unsubscribe func())

OnResync registers a function that resyncs the state of a subscriber and returns a function that removes it.

The messages could be lost, so the resync functions are called:

  • after each (re)connection of the LISTEN connection (the messages published while it was not connected are not received);
  • after a state changes message is dropped (see Cluster.Subscribe);
  • periodically (see Config.ResyncInterval).

They are called in a background goroutine, so they could run concurrently with the message handlers (and with themselves if a resync is requested while running).

func (*Cluster) Publish 

func (c *Cluster) Publish(db dbx.Builder, topic string, payload string) error

Publish sends a message to all nodes (including the current one).

The message is sent with pg_notify using the provided db builder. If the builder is a transaction, the message is delivered only after its commit.

It is no-op if the cluster secret is not available yet (no node can listen without it).

func (*Cluster) RateLimitHit 

func (c *Cluster) RateLimitHit(key string, maxAllowed int, interval time.Duration) (bool, error)

RateLimitHit registers a hit of the specified rate limit key and reports whether it is allowed (aka. the total hits in the current window are <= maxAllowed).

It implements the same fixed window strategy as the in-memory app rate limiter: the window starts with the first hit and resets after interval.

func (*Cluster) Ready 

func (c *Cluster) Ready() <-chan struct{}

Ready returns a channel that is closed after the first successful LISTEN.

func (*Cluster) RealtimeFailures 

func (c *Cluster) RealtimeFailures() int64

RealtimeFailures returns the number of the realtime replication events that couldn't be broadcasted to the local clients.

func (*Cluster) RealtimeSource 

func (c *Cluster) RealtimeSource() string

RealtimeSource returns the realtime source of the current node (RealtimeSourceHooks if not set).

func (*Cluster) Resync added in v1.1.0 

func (c *Cluster) Resync()

Resync calls the resync handlers in the background (see Cluster.OnResync).

func (*Cluster) SetRealtimeSource 

func (c *Cluster) SetRealtimeSource(source string)

SetRealtimeSource sets the realtime source of the current node (see RealtimeSourceHooks).

func (*Cluster) SlotMetrics 

func (c *Cluster) SlotMetrics(ctx context.Context) (*SlotMetrics, error)

SlotMetrics returns the realtime replication slot metrics of the database (or nil if there is no slot).

func (*Cluster) Start 

func (c *Cluster) Start()

Start starts listening for the cluster messages in the background.

It is no-op if the cluster is already started.

func (*Cluster) StartReplication 

func (c *Cluster) StartReplication(config ReplicationConfig)

StartReplication starts competing for the replication leadership in the background.

It is no-op if it is already started.

func (*Cluster) Stop 

func (c *Cluster) Stop()

Stop closes the LISTEN connection and waits for the background goroutines to finish.

It is no-op if the cluster is not started.

func (*Cluster) StopReplication 

func (c *Cluster) StopReplication()

StopReplication stops the replication (releasing the leadership) and waits for it to finish.

func (*Cluster) Subscribe 

func (c *Cluster) Subscribe(topic string, handler Handler) (unsubscribe func())

Subscribe registers a handler for the messages of the specified topic and returns a function that removes it.

The handlers are called sequentially from the state changes dispatch goroutine, so they must not block (see also Cluster.SubscribeData). The messages published by the current node are delivered too (check Message.NodeId to skip them).

If a message can't be dispatched (the queue is full), the resync handlers are called (see Cluster.OnResync).

func (*Cluster) SubscribeData added in v1.1.0 

func (c *Cluster) SubscribeData(topic string, handler Handler) (unsubscribe func())

SubscribeData is similar to Cluster.Subscribe but the handlers are called from a separate dispatch goroutine (and queue) for the data events that could take longer to handle (eg. the realtime events), so that they don't delay the state changes messages.

The data messages that can't be dispatched (the queue is full) are dropped.

func (*Cluster) TryLock 

func (c *Cluster) TryLock(ctx context.Context, name string) (unlock func(), err error)

TryLock tries to acquire the named session level advisory lock without waiting.

On success it returns a function that releases the lock. The lock is held by a dedicated connection of the main pool until it is released (or until the connection is closed, eg. if the app instance crashes).

It returns nil (and no error) if the lock is held by another session.

func (*Cluster) WarnLimited added in v1.1.0 

func (c *Cluster) WarnLimited(kind string, message string, args ...any)

WarnLimited logs a warning at most once per 10 seconds for the same kind (eg. so that a flood of invalid messages doesn't flood the logs).

type Config 

type Config struct {
	// DSN returns the connection string of the shared database
	// (used for the dedicated LISTEN connection).
	DSN func() string

	// DB returns the main database pool (not a transaction).
	DB func() dbx.Builder

	// Secret returns the secret shared by all nodes that authenticates the
	// cluster messages. It must be stored in the database (so that the
	// nodes share it) and it must not be readable by the untrusted roles.
	//
	// It should return [ErrNotReady] if it can't be loaded yet.
	Secret func() (string, error)

	// Logger returns the logger to use.
	Logger func() *slog.Logger

	// Enabled reports whether the cluster mode is enabled
	// (see [Cluster.Enabled]).
	Enabled func() bool

	// ResyncInterval is the interval of the periodic resync
	// (default to [DefaultResyncInterval]).
	ResyncInterval time.Duration
}

Config defines the Cluster dependencies.

They are functions because they could change on app bootstrap.

type EventsDeduplicator 

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

EventsDeduplicator remembers the recently received event ids so that the events delivered more than once (eg. after a leader change) are discarded.

func NewEventsDeduplicator 

func NewEventsDeduplicator(max int) *EventsDeduplicator

NewEventsDeduplicator creates a new deduplicator that remembers the last max ids.

func (*EventsDeduplicator) Filter 

func (d *EventsDeduplicator) Filter(ids []string) []string

Filter returns the ids that were not seen before (sorted) and remembers them.

func (*EventsDeduplicator) Forget 

func (d *EventsDeduplicator) Forget(ids []string)

Forget removes the provided ids from the received ones (eg. the events that failed to broadcast, so that a new publish delivers them).

type Handler 

type Handler func(msg Message)

Handler handles the messages of a subscribed topic.

type KV 

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

KV is a key-value store with expiration that is shared by all nodes.

It replaces the app in-memory store for the temporary state that must be visible to all nodes (eg. the resend timeouts of the auth emails).

The queries are executed with the main database pool (and not with the current transaction, if any) because the state is not part of the transaction data and it is usually written after the transaction completion.

func (*KV) Claim 

func (kv *KV) Claim(key string, value string, ttl time.Duration) (bool, error)

Claim stores the value of the key for the ttl duration only if the key doesn't exist or has expired.

It reports whether the key was stored (aka. claimed by the caller).

func (*KV) Delete 

func (kv *KV) Delete(key string) error

Delete deletes the key (if exists).

func (*KV) DeleteExpired 

func (kv *KV) DeleteExpired() error

DeleteExpired deletes all expired keys.

func (*KV) Get 

func (kv *KV) Get(key string) (string, bool, error)

Get returns the value of the key and whether it exists (the expired keys don't exist).

func (*KV) Has 

func (kv *KV) Has(key string) bool

Has reports whether the key exists (errors are reported as missing key).

func (*KV) Pop 

func (kv *KV) Pop(key string) (string, bool, error)

Pop returns the value of the key and deletes it (it is like Get followed by Delete but only one of the concurrent callers receives the value).

func (*KV) Set 

func (kv *KV) Set(key string, value string, ttl time.Duration) error

Set stores the value of the key (replacing the existing one) for the ttl duration.

type Message 

type Message struct {
	// NodeId is the id of the node that published the message.
	NodeId string

	// Topic is the message topic (eg. "collections").
	Topic string

	// Payload is the optional message data.
	Payload string
}

Message is a cluster message received from a node (including the current one).

type RealtimeEvent 

type RealtimeEvent struct {
	Id           string `db:"id"`
	Source       string `db:"source"`
	Op           string `db:"op"`
	CollectionId string `db:"collection_id"`
	RecordId     string `db:"record_id"`
	Snapshot     string `db:"snapshot"` // json, only for the deleted records
}

RealtimeEvent is a stored realtime event.

type ReplicationConfig 

type ReplicationConfig struct {
	// PrepareTables ensures that the publication contains all collection
	// tables (with REPLICA IDENTITY FULL). It is called by the leader before
	// consuming the slot.
	PrepareTables func() error

	// ResolveCollection returns the collection id of the specified table
	// (or the table name if it can't be resolved).
	ResolveCollection func(table string) string
}

ReplicationConfig defines the callbacks used by the replication leader.

type SlotMetrics 

type SlotMetrics struct {
	Slot string `db:"slot_name" json:"slot"`

	// Active reports whether a node consumes the slot.
	Active bool `db:"active" json:"active"`

	// WALStatus is the Postgres wal_status of the slot (eg. "reserved" or "lost").
	WALStatus string `db:"wal_status" json:"walStatus"`

	// LagBytes is the size of the WAL that is not confirmed by the consumer yet.
	LagBytes int64 `db:"lag_bytes" json:"lagBytes"`

	// RetainedBytes is the size of the WAL retained by the slot.
	RetainedBytes int64 `db:"retained_bytes" json:"retainedBytes"`
}

SlotMetrics are the realtime replication slot metrics.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.