Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
- API
  
  Reference documentation for the pkg.go.dev API
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

cluster

package

Go to main page

Versions in this module

v1

v1.1.0

Oct 29, 2026

Changes in this version

+ const ChannelPrefix

+ const DefaultResyncInterval

+ const MaxTransactionEvents

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

type Cluster

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

+ func (c *Cluster) Connected() bool

+ func (c *Cluster) InvalidMessages() int64

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

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

+ func (c *Cluster) Resync()

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

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

type Config

+ ResyncInterval time.Duration

+ Secret func() (string, error)

v1.0.0

Oct 26, 2026

Changes in this version

+ const Channel

+ const CronRunsTable

+ const KVTable

+ const MaxPayloadSize

+ const PublicationName

+ const RateLimitsTable

+ const RealtimeEventsMaxAge

+ const RealtimeEventsTable

+ const RealtimeSourceHooks

+ const RealtimeSourceReplication

+ const RetainedWALWarningSize

+ const TopicRealtimeEvents

+ func LockKey(name string) int64

+ func SlotName(database string) string

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

+ type Cluster struct

+ func New(config Config) *Cluster

+ func (c *Cluster) AddRealtimeFailure()

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

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

+ func (c *Cluster) DeleteExpired() error

+ func (c *Cluster) DroppedMessages() int64

+ func (c *Cluster) Enabled() bool

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

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

+ func (c *Cluster) IsReplicationLeader() bool

+ func (c *Cluster) KV() *KV

+ func (c *Cluster) NodeId() string

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

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

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

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

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

+ func (c *Cluster) RealtimeFailures() int64

+ func (c *Cluster) RealtimeSource() string

+ func (c *Cluster) SetRealtimeSource(source string)

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

+ func (c *Cluster) Start()

+ func (c *Cluster) StartReplication(config ReplicationConfig)

+ func (c *Cluster) Stop()

+ func (c *Cluster) StopReplication()

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

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

+ type Config struct

+ DB func() dbx.Builder

+ DSN func() string

+ Enabled func() bool

+ Logger func() *slog.Logger

+ type EventsDeduplicator struct

+ func NewEventsDeduplicator(max int) *EventsDeduplicator

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

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

+ type Handler func(msg Message)

+ type KV struct

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

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

+ func (kv *KV) DeleteExpired() error

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

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

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

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

+ type Message struct

+ NodeId string

+ Payload string

+ Topic string

+ type RealtimeEvent struct

+ CollectionId string

+ Id string

+ Op string

+ RecordId string

+ Snapshot string

+ Source string

+ type ReplicationConfig struct

+ PrepareTables func() error

+ ResolveCollection func(table string) string

+ type SlotMetrics struct

+ Active bool

+ LagBytes int64

+ RetainedBytes int64

+ Slot string

+ WALStatus string