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

cluster

package
v1.0.0 Latest Latest
Warning

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

 Go to latest Published: Oct 26, 2026 License: MIT Imports: 24 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 (Channel) and are routed to the subscribers by their topic, so that new topics don't require a new LISTEN.

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"

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

Channel is the Postgres LISTEN/NOTIFY channel used by the cluster messages.

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 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

This section is empty.

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>" (with the invalid characters replaced) and it is truncated with a hash suffix if it is longer than 63 characters.

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.

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) 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) 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 the 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) 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) 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) OnReconnect 

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

OnReconnect registers a function that is called after the LISTEN connection was reestablished (the messages published in the meantime are lost, so the subscribers should resync their state) and returns a function that removes it.

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.

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) 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 a single dispatch goroutine, so they should not block for long. The messages published by the current node are delivered too (check Message.NodeId to skip them).

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.

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

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

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

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.