postgres

Documentation

Overview

Package postgres implements storage.Storage using PostgreSQL via pgx.

Index

• Variables
• type Storage
  • func NewStorage(pool *pgxpool.Pool) *Storage
  • func (s *Storage) AcquireLock(ctx context.Context, key int64) (func(), error)
  • func (s *Storage) ActivateSnapshot(ctx context.Context, collection, version string) error
  • func (s *Storage) AppliedInstances(ctx context.Context, collection, version, status string) ([]string, error)
  • func (s *Storage) CountApplied(ctx context.Context, collection, version string) (int, error)
  • func (s *Storage) DeleteOldSnapshots(ctx context.Context, olderThan time.Time) (int, error)
  • func (s *Storage) FailSnapshot(ctx context.Context, collection, version string) error
  • func (s *Storage) GetActiveSnapshot(ctx context.Context, collection string) (*storage.Snapshot, error)
  • func (s *Storage) GetActiveVersion(ctx context.Context, collection string) (string, error)
  • func (s *Storage) GetSnapshot(ctx context.Context, collection, version string) (*storage.Snapshot, error)
  • func (s *Storage) LogApply(ctx context.Context, instanceID, collection, version, status string) error
  • func (s *Storage) Migrate(ctx context.Context) error
  • func (s *Storage) ResetApplyLog(ctx context.Context, collection, version string) error
  • func (s *Storage) SaveSnapshot(ctx context.Context, collection, version string, content []byte) error

Variables

var MigrationSQL string

Types

type Storage

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

Storage implements storage.Storage using PostgreSQL.

func NewStorage

func NewStorage(pool *pgxpool.Pool) *Storage

NewStorage creates a new PostgreSQL-backed Storage.

func (*Storage) AcquireLock

func (s *Storage) AcquireLock(ctx context.Context, key int64) (func(), error)

AcquireLock attempts to acquire a Postgres session-level advisory lock. Returns a release function if acquired. The caller must call release when done. Returns storage.ErrLockNotAcquired if the lock is already held by another session.

func (*Storage) ActivateSnapshot

func (s *Storage) ActivateSnapshot(ctx context.Context, collection, version string) error

ActivateSnapshot marks the given snapshot as active and demotes any previously active snapshot for the same collection to inactive.

func (*Storage) AppliedInstances

func (s *Storage) AppliedInstances(ctx context.Context, collection, version, status string) ([]string, error)

AppliedInstances returns the instance IDs that logged the given status for (collection, version).

func (*Storage) CountApplied

func (s *Storage) CountApplied(ctx context.Context, collection, version string) (int, error)

CountApplied returns the number of instances that successfully applied a version.

func (*Storage) DeleteOldSnapshots

func (s *Storage) DeleteOldSnapshots(ctx context.Context, olderThan time.Time) (int, error)

DeleteOldSnapshots removes snapshots created before olderThan, except those in the 'active' status (which are preserved regardless of age so the cluster can always recover the current authoritative version).

In the same transaction, apply-log rows for the deleted snapshot versions are removed. Returns the number of snapshots deleted.

func (*Storage) FailSnapshot

func (s *Storage) FailSnapshot(ctx context.Context, collection, version string) error

FailSnapshot marks a snapshot as failed.

func (*Storage) GetActiveSnapshot

func (s *Storage) GetActiveSnapshot(ctx context.Context, collection string) (*storage.Snapshot, error)

GetActiveSnapshot returns the currently active snapshot for a collection.

func (*Storage) GetActiveVersion

func (s *Storage) GetActiveVersion(ctx context.Context, collection string) (string, error)

GetActiveVersion returns just the version string of the active snapshot, without fetching the full content. Used by followers for cheap version comparison during self-heal checks.

func (*Storage) GetSnapshot

func (s *Storage) GetSnapshot(ctx context.Context, collection, version string) (*storage.Snapshot, error)

GetSnapshot returns a specific snapshot by collection and version.

func (*Storage) LogApply

func (s *Storage) LogApply(ctx context.Context, instanceID, collection, version, status string) error

LogApply records that an instance has applied (or failed to apply) a config version. Uses upsert so repeated calls are idempotent.

func (*Storage) Migrate

func (s *Storage) Migrate(ctx context.Context) error

Migrate runs the DDL migration to create required tables and indexes.

func (*Storage) ResetApplyLog

func (s *Storage) ResetApplyLog(ctx context.Context, collection, version string) error

ResetApplyLog deletes all apply-log rows for (collection, version).

func (*Storage) SaveSnapshot

func (s *Storage) SaveSnapshot(ctx context.Context, collection, version string, content []byte) error

SaveSnapshot inserts a new snapshot in pending state. If a snapshot with the same collection+version already exists, this is a no-op.