coordination

Documentation

Overview

Package coordination provides the cross-process seams a multi-replica Keyway deployment needs: a shared idempotency store (so a retried write replays the same result on any replica) and a leader gate (so exactly one replica runs the scheduler). Each seam has an in-memory single-node implementation (the default, and all a single daemon needs) and a Postgres implementation that shares state across replicas. The interfaces are the point: turning on HA is choosing a different adapter, not a rewrite (architecture review W5/#6).

Index

• type Coordinator
  • func NewMemory(idemTTL time.Duration) *Coordinator
  • func Open(ctx context.Context, dsn string, idemTTL time.Duration) (*Coordinator, error)
  • func (c *Coordinator) Close()
  • func (c *Coordinator) Idempotency() IdempotencyStore
  • func (c *Coordinator) Leader() Leader
  • func (c *Coordinator) Pool() *pgxpool.Pool
• type IdempotencyStore
  • func NewMemoryIdempotency(ttl time.Duration) IdempotencyStore
• type Leader
  • func NewLocalLeader() Leader
• type Record

Types

type Coordinator

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

Coordinator bundles the coordination seams and owns whatever resources back them (e.g. a Postgres pool). Close releases them.

func NewMemory

func NewMemory(idemTTL time.Duration) *Coordinator

NewMemory returns a single-node coordinator: an in-memory idempotency store and a leader that is always the leader. This is the default and is all a single daemon needs.

func Open

func Open(ctx context.Context, dsn string, idemTTL time.Duration) (*Coordinator, error)

Open builds a Coordinator for the DSN plus a cleanup that Close() invokes. A "memory" DSN yields the single-node in-memory coordinator; anything else is a Postgres DSN whose seams are shared across replicas.

func (*Coordinator) Close

func (c *Coordinator) Close()

Close releases the leader lock and any owned pool.

func (*Coordinator) Idempotency

func (c *Coordinator) Idempotency() IdempotencyStore

func (*Coordinator) Leader

func (c *Coordinator) Leader() Leader

func (*Coordinator) Pool

func (c *Coordinator) Pool() *pgxpool.Pool

Pool returns the operational Postgres pool backing the coordinator, or nil for the in-memory (single-node) coordinator. It lets other operational stores that share the same database (e.g. the Postgres keystore) reuse one pool.

type IdempotencyStore

type IdempotencyStore interface {
	// Get returns the stored record for key, or ok=false if absent/expired.
	Get(ctx context.Context, key string) (Record, bool, error)
	// Put stores a record for key with the store's configured TTL.
	Put(ctx context.Context, key string, rec Record) error
}

IdempotencyStore persists the response to a completed write so a retry with the same idempotency key replays it instead of re-executing. A Postgres-backed implementation shares this across replicas; the in-memory one is per-process.

func NewMemoryIdempotency

func NewMemoryIdempotency(ttl time.Duration) IdempotencyStore

NewMemoryIdempotency returns an in-memory idempotency store with the given TTL and a hard FIFO cap (4096 entries) so a burst of distinct keys cannot exhaust memory.

type Leader

type Leader interface {
	IsLeader(ctx context.Context) bool
	Close() error
}

Leader gates work that must run on exactly one replica. IsLeader reports (and, for durable backends, lazily acquires) leadership; it is cheap to call each scheduler tick. Close releases any held lock.

func NewLocalLeader

func NewLocalLeader() Leader

NewLocalLeader returns a leader that always holds leadership (single node).

type Record

type Record struct {
	Status      int
	Body        []byte
	ContentType string
}

Record is a cached idempotent response.