Documentation
¶
Overview ¶
Package pgstore is the Postgres backend for running several halo-server and halo-shadow replicas (ADR-0021): versioned forward-only migrations, the shared record logs and KV behind internal/storage, the hash-chained audit table, fleet reports, rollout documents, leader leases with fencing tokens, and halo-shadow's pair table and global budget. It speaks database/sql over the pgx stdlib driver and stores opaque bytes: the packages that own the types encode them.
Index ¶
- Constants
- Variables
- func CheckTLS(dsn string) error
- func HolderID() string
- func SchemaVersion() int
- type AuditLink
- type BudgetCounter
- type DB
- func (d *DB) Acquire(ctx context.Context, name, holder string, ttl time.Duration) (int64, bool, error)
- func (d *DB) Add(ctx context.Context, ns, key string, val []byte, exp time.Time) (bool, error)
- func (d *DB) AppendAudit(ctx context.Context, ...) error
- func (d *DB) AppendPair(ctx context.Context, id string, at time.Time, row []byte) error
- func (d *DB) ApplyBudget(ctx context.Context, batch string, incs []BudgetCounter) (duplicate bool, err error)
- func (d *DB) AuditEntries(ctx context.Context) ([][]byte, error)
- func (d *DB) AuditHead(ctx context.Context) (uint64, string, error)
- func (d *DB) BudgetCounters(ctx context.Context, keys [][2]string) ([]BudgetCounter, error)
- func (d *DB) CheckSchema(ctx context.Context) error
- func (d *DB) ClaimPending(ctx context.Context, dedupe, requestID string, ...) (bool, error)
- func (d *DB) Close() error
- func (d *DB) CountAuthFailure(ctx context.Context, key string, window time.Duration) (int64, error)
- func (d *DB) Counts(ctx context.Context) (map[string]int64, error)
- func (d *DB) Delete(ctx context.Context, ns, key string) error
- func (d *DB) DeleteReportsWhere(ctx context.Context, match func(report []byte) bool, dry bool) (latest, history int64, err error)
- func (d *DB) Docs(ctx context.Context, kind string) ([][]byte, error)
- func (d *DB) EachAudit(ctx context.Context, from uint64, fn func(seq uint64, entry []byte) bool) error
- func (d *DB) EachAuditDesc(ctx context.Context, fn func(seq uint64, entry []byte) bool) error
- func (d *DB) Errors() int64
- func (d *DB) Fenced(f Fence) *Fenced
- func (d *DB) Get(ctx context.Context, ns, key string) ([]byte, bool, error)
- func (d *DB) GetDoc(ctx context.Context, kind, name string) ([]byte, bool, error)
- func (d *DB) Import(ctx context.Context, fn func(*Importer) error) error
- func (d *DB) Leased(ctx context.Context, name string, ttl time.Duration, ...)
- func (d *DB) Migrate(ctx context.Context) (applied []int, err error)
- func (d *DB) OpenFencedLog(ctx context.Context, name string, replay func([]byte), f Fence) (storage.Log, error)
- func (d *DB) OpenLog(ctx context.Context, name string, replay func([]byte)) (storage.Log, error)
- func (d *DB) Pairs(ctx context.Context, fn func(row []byte) error) error
- func (d *DB) Ping(ctx context.Context) error
- func (d *DB) PruneAudit(ctx context.Context, through uint64) (int64, error)
- func (d *DB) PruneBudgets(ctx context.Context, retention time.Duration) (int64, error)
- func (d *DB) PrunePairs(ctx context.Context, cutoff time.Time) (int, error)
- func (d *DB) PruneReports(ctx context.Context, cutoff time.Time) (int64, error)
- func (d *DB) Put(ctx context.Context, ns, key string, val []byte, exp time.Time) error
- func (d *DB) PutDoc(ctx context.Context, kind, name string, body []byte) error
- func (d *DB) PutReport(ctx context.Context, key, hostname string, report []byte, keep int) error
- func (d *DB) PutRevocation(ctx context.Context, r Revocation) error
- func (d *DB) PutServiceToken(ctx context.Context, t ServiceToken, maxActive int, now time.Time) error
- func (d *DB) Raise(ctx context.Context, ns, key string, at int64, exp time.Time) error
- func (d *DB) RedactAudit(ctx context.Context, bodies map[uint64][2][]byte) error
- func (d *DB) Release(ctx context.Context, name, holder string) error
- func (d *DB) ReleasePending(ctx context.Context, dedupe, requestID string) error
- func (d *DB) ReportHistory(ctx context.Context, key string, limit int) ([][]byte, error)
- func (d *DB) Reports(ctx context.Context) ([][]byte, error)
- func (d *DB) Reserve(ctx context.Context, scope, id string, amount, limit float64, ...) (bool, error)
- func (d *DB) Revocations(ctx context.Context) ([]Revocation, error)
- func (d *DB) RevocationsFor(ctx context.Context, jtis, subjects []string) ([]Revocation, error)
- func (d *DB) RevokeServiceToken(ctx context.Context, id string, record []byte) (ServiceToken, bool, error)
- func (d *DB) ServiceToken(ctx context.Context, id string) (ServiceToken, bool, error)
- func (d *DB) ServiceTokens(ctx context.Context) ([]ServiceToken, error)
- func (d *DB) Settle(ctx context.Context, scope, id string, amount, charged float64) error
- func (d *DB) Spent(ctx context.Context, scope string) (float64, error)
- func (d *DB) Take(ctx context.Context, ns, key string) ([]byte, bool, error)
- func (d *DB) TryLock(ctx context.Context, ns, key, owner string, ttl time.Duration) (bool, error)
- func (d *DB) Unlock(ctx context.Context, ns, key, owner string) error
- func (d *DB) WriteMetrics(w io.Writer)
- type Elector
- type Fence
- type Fenced
- type Importer
- func (im *Importer) AddBudget(incs []BudgetCounter) error
- func (im *Importer) AppendAudit(l AuditLink) error
- func (im *Importer) AppendJournal(stream string, body []byte) error
- func (im *Importer) Counts() (map[string]int64, error)
- func (im *Importer) EachAudit(fn func(seq uint64, entry []byte) bool) error
- func (im *Importer) PutDoc(kind, name string, body []byte) error
- func (im *Importer) PutReport(key, hostname string, report []byte) error
- func (im *Importer) PutRevocation(r Revocation) error
- func (im *Importer) PutServiceToken(t ServiceToken) error
- type LeaseStore
- type Options
- type Revocation
- type ServiceToken
Constants ¶
const DefaultMaxConns = 16
DefaultMaxConns is the pool size when Options.MaxConns is 0.
Variables ¶
var ErrAuditChanged = errors.New("pgstore: an audit entry changed while it was being redacted")
ErrAuditChanged: a body to redact is no longer the one the caller read.
var ErrFenced = storage.ErrFenced
ErrFenced means a fenced write was refused: the writer no longer holds the lease whose token it was opened with (another replica leads now). It is storage.ErrFenced, so callers outside pgstore can test for it.
var ErrInsecureTLS = errors.New("pgstore: database URL does not verify the server's certificate")
ErrInsecureTLS is CheckTLS's refusal.
var ErrNotEmpty = errors.New("pgstore: target database is not empty")
ErrNotEmpty means an import's target already holds state.
var ErrSchema = errors.New("pgstore: schema mismatch")
ErrSchema means the database schema does not match this binary.
var ErrServiceTokenLimit = errors.New("service token limit reached")
ErrServiceTokenLimit is PutServiceToken's refusal at the active-token cap.
Functions ¶
func CheckTLS ¶
CheckTLS refuses a URL whose connection to a non-loopback host is not TLS with a verified certificate and host name (sslmode=verify-full): sslmode disable, allow, prefer and require (no certificate check), and verify-ca (no host-name check) all leave the connection open to a man-in-the-middle on the network. Loopback hosts and Unix sockets never leave the machine and are accepted.
func HolderID ¶
func HolderID() string
HolderID is a replica-unique lease holder id: the host (pod) name and a random suffix, so two processes on one host never share a lease.
func SchemaVersion ¶
func SchemaVersion() int
SchemaVersion is the newest migration this binary knows.
Types ¶
type BudgetCounter ¶
type BudgetCounter struct {
Budget, Key string
Bucket int64 // bucket start, unix seconds
USD float64
Tokens int64
}
BudgetCounter is one bucket of one budget counter.
type DB ¶
type DB struct {
// contains filtered or unexported fields
}
DB is a Postgres handle shared by every store of one process.
func (*DB) Acquire ¶
func (d *DB) Acquire(ctx context.Context, name, holder string, ttl time.Duration) (int64, bool, error)
Acquire implements LeaseStore on halo_leases.
func (*DB) Add ¶
Add implements storage.KV: the upsert only overwrites an expired row, and Postgres's row lock on conflict makes concurrent Adds of one key race-free.
func (*DB) AppendAudit ¶
func (d *DB) AppendAudit(ctx context.Context, chain func(headSeq uint64, headHash string) (AuditLink, error)) error
AppendAudit appends one entry to the hash chain. The chain head row is locked (SELECT ... FOR UPDATE) for the whole transaction, so concurrent writers on any replica append strictly one after another; chain receives the head and returns the new link, which must extend it.
func (*DB) AppendPair ¶
AppendPair stores one encoded pair row (idempotent on id).
func (*DB) ApplyBudget ¶
func (d *DB) ApplyBudget(ctx context.Context, batch string, incs []BudgetCounter) (duplicate bool, err error)
ApplyBudget adds incs to the counters unless batch was already applied, in one transaction: the batch id's primary key decides, so of two replicas receiving the same retry exactly one counts it. The upserts run in (budget, key, bucket) order, so two transactions touching the same rows lock them in the same order and cannot deadlock. Nothing is pruned here: see PruneBudgets.
func (*DB) AuditEntries ¶
AuditEntries returns every stored entry body, oldest first. ponytail: reads the whole chain like the file backend; page by seq if it gets big.
func (*DB) BudgetCounters ¶
BudgetCounters returns the buckets of the given (budget, key) counters, or of every counter when keys is nil.
func (*DB) CheckSchema ¶
CheckSchema returns ErrSchema unless this binary can run against the database: at exactly its schema version, or newer when every newer migration still accepts it (a rolling upgrade's old pods, or a binary rollback). Behind means run `halo-server migrate` (or --auto-migrate).
func (*DB) ClaimPending ¶
func (d *DB) ClaimPending(ctx context.Context, dedupe, requestID string, holder func(id string) (known, pending bool)) (bool, error)
ClaimPending makes requestID the one pending request for dedupe; ok=false when another pending request holds it. The primary key decides between concurrent claims. holder reports what the caller's request log says about the current holder: a known request that is no longer pending (decided) is replaced at once; an unknown one only once its claim is orphanClaimAge old. Replacement is a compare-and-swap on the holder's id.
func (*DB) CountAuthFailure ¶ added in v0.4.0
CountAuthFailure adds one to key's failure count and returns it, in one upsert: a live row (younger than window on the database clock) is incremented, an expired or absent one restarts at 1 with a new window.
func (*DB) Counts ¶
Counts returns row counts: "journal/<stream>" per log stream, "audit", "reports" and "docs/<kind>".
func (*DB) DeleteReportsWhere ¶ added in v0.6.0
func (d *DB) DeleteReportsWhere(ctx context.Context, match func(report []byte) bool, dry bool) (latest, history int64, err error)
DeleteReportsWhere deletes every latest report and history row whose body match selects (ADR-0040 erasure) and returns how many of each it deleted (with dry, would delete). ponytail: reads every report; erasure is a rare admin action.
func (*DB) EachAudit ¶
func (d *DB) EachAudit(ctx context.Context, from uint64, fn func(seq uint64, entry []byte) bool) error
EachAudit streams stored audit entries with seq >= from, in seq order, until fn returns false.
func (*DB) EachAuditDesc ¶
EachAuditDesc streams stored audit entries newest first, in pages of auditDescPage keyed on seq (ORDER BY seq DESC LIMIT), until fn returns false: a tail read touches only the rows it returns, not the whole table.
func (*DB) Import ¶
Import runs fn in one transaction that it first checks is starting from an empty database (every table Counts covers). fn returning an error, or a failed commit, leaves the database as it was.
func (*DB) Leased ¶
func (d *DB) Leased(ctx context.Context, name string, ttl time.Duration, fn func(ctx context.Context, view *Fenced))
Leased runs fn on exactly one replica at a time: every replica calls it with the same name, the one holding lease name runs fn with a context that ends when the lease is lost, and the document writes of the view fn gets are fenced by that lease. Leased returns when ctx ends. ttl 0 = the default (15s).
func (*DB) Migrate ¶
Migrate applies every pending migration in one transaction, under an advisory lock so concurrent replicas apply each exactly once, and returns the versions it applied. A database newer than this binary is fine when its newer migrations still accept this binary (see minBinaryHeader), and an error otherwise.
func (*DB) OpenFencedLog ¶
func (d *DB) OpenFencedLog(ctx context.Context, name string, replay func([]byte), f Fence) (storage.Log, error)
OpenFencedLog is OpenLog whose appends fail with ErrFenced once f's lease has moved on, so a deposed leader cannot write behind the new one.
func (*DB) PruneAudit ¶ added in v0.6.0
PruneAudit deletes entries with seq <= through (ADR-0040 retention). The head row is untouched, so appends continue the chain.
func (*DB) PruneBudgets ¶
PruneBudgets drops what no budget can still count, judged on the database clock and nothing else: buckets that started more than retention ago (the caller passes the longest window policy allows), and batch ids older than budgetBatchMemory. It never consults a replica's policy, so a replica on a stale policy cannot delete the counters of a budget it does not know yet; a removed budget's rows simply age out. Safe on any replica, any number of times.
func (*DB) PrunePairs ¶
PrunePairs deletes pairs older than cutoff.
func (*DB) PruneReports ¶ added in v0.6.0
PruneReports deletes reporters whose report was last seen (the lastSeen the server stamped, its clock) before cutoff, with their history (ADR-0040).
func (*DB) PutReport ¶
PutReport upserts the latest report for key and appends it to key's history, keeping the newest keep entries.
func (*DB) PutRevocation ¶ added in v0.4.0
func (d *DB) PutRevocation(ctx context.Context, r Revocation) error
PutRevocation records r in one upsert that never weakens a live row: the epoch and the expiry only move forward (GREATEST), and the record follows the larger epoch. Expired rows are pruned.
func (*DB) PutServiceToken ¶ added in v0.5.0
func (d *DB) PutServiceToken(ctx context.Context, t ServiceToken, maxActive int, now time.Time) error
PutServiceToken inserts a new token unless maxActive tokens (not revoked, expiring after now) already exist (ErrServiceTokenLimit): the count and the insert are one transaction under an advisory lock, so concurrent creations on any replicas never exceed it. An existing id is an error (ids are random: a collision is a bug, never an overwrite). Long-expired rows are pruned.
func (*DB) Raise ¶
Raise implements storage.KV with one upsert: a live row keeps the greater of its value and at, and the later expiry.
func (*DB) RedactAudit ¶ added in v0.6.0
RedactAudit replaces the stored bodies of existing entries (ADR-0040 erasure): seq -> {body read, redacted body}. Each row is swapped only if it still holds the body read (compare and swap), all or none, so two erasures racing cannot undo each other. seq, hash and prev never change.
func (*DB) ReleasePending ¶
ReleasePending drops dedupe's claim if requestID still holds it.
func (*DB) ReportHistory ¶
ReportHistory returns up to limit reports for key, newest first.
func (*DB) Reserve ¶
func (d *DB) Reserve(ctx context.Context, scope, id string, amount, limit float64, ttl time.Duration) (bool, error)
Reserve books amount against scope's budget of limit USD, globally across replicas: one conditional UPDATE (spent+reserved+amount <= limit) decides. Reservations older than ttl (a worker that died before settling) are first reaped and charged in full, since their upstream calls may have been paid.
func (*DB) Revocations ¶ added in v0.4.0
func (d *DB) Revocations(ctx context.Context) ([]Revocation, error)
Revocations returns every live revocation, ordered by kind and key.
func (*DB) RevocationsFor ¶ added in v0.4.0
RevocationsFor returns the live revocations of the given jtis and subjects: one indexed query, for the per-request check.
func (*DB) RevokeServiceToken ¶ added in v0.5.0
func (d *DB) RevokeServiceToken(ctx context.Context, id string, record []byte) (ServiceToken, bool, error)
RevokeServiceToken marks id revoked and replaces its record, unless it is already revoked (the first revocation's record stands). It returns the row as stored and whether it exists.
func (*DB) ServiceToken ¶ added in v0.5.0
ServiceToken returns the row with id: one primary-key lookup, for the per-request check.
func (*DB) ServiceTokens ¶ added in v0.5.0
func (d *DB) ServiceTokens(ctx context.Context) ([]ServiceToken, error)
ServiceTokens lists every row, oldest expiry first.
func (*DB) Settle ¶
Settle swaps reservation id for the charged amount. If the reservation was already reaped (and charged in full) only the excess is added.
func (*DB) Take ¶
Take implements storage.KV: DELETE ... RETURNING is atomic, so of concurrent Takes exactly one sees the row.
func (*DB) TryLock ¶
TryLock implements storage.KV: the owner is the value, the expiry is computed by the database (now() + ttl), and the upsert only takes a row that is expired or already the owner's.
func (*DB) WriteMetrics ¶
WriteMetrics writes the database metrics in Prometheus text format.
type Elector ¶
type Elector struct {
Store LeaseStore
Name string // lease name, e.g. "controller"
Holder string // unique per replica (pod name + random suffix)
// TTL is the lease length (default 15s); Renew the renewal period
// (default TTL/3). A leader that has not renewed by TTL-Renew after its
// last successful renewal started steps down, so it is gone before the
// lease can expire on the database clock.
TTL, Renew time.Duration
Log *slog.Logger
Now func() time.Time // test seam; default time.Now
// contains filtered or unexported fields
}
Elector runs a singleton loop on exactly one replica. Every replica runs an Elector on the same lease; the one holding it runs lead, and stops it (cancels its context and waits) when the lease is lost or cannot be renewed in time, before any other replica can take the lease over.
func (*Elector) Run ¶
Run campaigns until ctx ends. lead runs while this replica leads, with a context cancelled on step-down and the lease's fencing token; if it returns on its own it is restarted on the next round while the lease is still held. On exit Run stops lead and releases the lease so a peer takes over at once.
func (*Elector) WriteMetrics ¶
WriteMetrics writes halo_leader (1 when this replica leads) and leadership changes.
type Fence ¶
Fence ties a log's appends to a leader lease: an append commits only while lease Name is held with fencing token Token.
type Fenced ¶
type Fenced struct {
// contains filtered or unexported fields
}
Fenced is the view of d a leader writes through: its logs (OpenLog) and document writes (PutDoc) commit only while f's lease is still held at f's token, so a deposed leader finishing a tick cannot overwrite its successor.
type Importer ¶
type Importer struct {
// contains filtered or unexported fields
}
Importer writes a whole data-dir import inside one transaction: either all of it commits, after the caller verified it through the same transaction, or none of it does and the import can simply be run again.
func (*Importer) AddBudget ¶
func (im *Importer) AddBudget(incs []BudgetCounter) error
AddBudget adds counter buckets.
func (*Importer) AppendAudit ¶
AppendAudit appends an already-chained entry; it must extend the head.
func (*Importer) AppendJournal ¶
AppendJournal appends one record to stream.
func (*Importer) PutRevocation ¶ added in v0.4.0
func (im *Importer) PutRevocation(r Revocation) error
PutRevocation stores a revocation as DB.PutRevocation does (no pruning).
func (*Importer) PutServiceToken ¶ added in v0.5.0
func (im *Importer) PutServiceToken(t ServiceToken) error
PutServiceToken stores a token as DB.PutServiceToken does (no pruning).
type LeaseStore ¶
type LeaseStore interface {
// Acquire takes or renews lease name for holder for ttl. ok is false while
// another holder's lease is unexpired. token is the fencing token: it is
// unchanged across renewals and grows whenever the lease is (re)taken.
Acquire(ctx context.Context, name, holder string, ttl time.Duration) (token int64, ok bool, err error)
// Release expires holder's lease now so another replica can take it.
Release(ctx context.Context, name, holder string) error
}
LeaseStore grants named leases. Expiry is judged on the store's clock (for Postgres, the database's now()), never on a replica's.
type Options ¶
type Options struct {
// MaxConns caps the pool (default DefaultMaxConns). Every replica holds
// up to this many connections: size it against the server's max_connections.
MaxConns int
// InsecureTLS accepts a connection to a non-loopback host without
// verified TLS (sslmode below verify-full). Off by default: the database
// holds the audit chain, device token hashes and the kill list.
InsecureTLS bool
}
Options tunes Open.
type Revocation ¶ added in v0.4.0
Revocation is one credential revocation row (ADR-0026). Kind is "jti" or "subject"; Epoch is a subject's revoked-before time (unix seconds, 0 for a jti); Record is opaque to this package.
Source Files
Directories