Documentation
¶
Overview ¶
Package store opens Ether's database and applies each component's migrations. Components own their tables: they register migrations under their own name (table names prefixed with it) and receive *sql.DB.
func init() {
store.Register("catalog", 1, `CREATE TABLE catalog_titles (...)`)
}
A standalone server uses SQLite in its data directory. A clustered node uses an owner-supplied PostgreSQL database; component SQL is written once with `?` placeholders (rebound to $n by the "ether-pgx" driver) and migrations are translated from SQLite DDL (see TranslateDDL). DialectOf tells a component which engine it is talking to for the few constructs that differ.
Index ¶
- Constants
- Variables
- func AppliedMigrations(ctx context.Context, db *sql.DB) (map[string]int, error)
- func Contains(d Dialect, haystack, needle string) string
- func ExportLogical(ctx context.Context, db *sql.DB, w io.Writer) error
- func ImportLogical(ctx context.Context, db *sql.DB, r io.Reader) (err error)
- func IsPermanent(err error) bool
- func IsUnavailable(err error) bool
- func JSONInt(d Dialect, col, key string) string
- func JSONRemove(d Dialect, col, key string) string
- func JSONText(d Dialect, col, key string) string
- func KnownMigrations() map[string]int
- func Like(d Dialect) string
- func LockKey(ctx context.Context, d Dialect, tx *sql.Tx, key string) error
- func Maintain(ctx context.Context, db *sql.DB) error
- func Migrate(ctx context.Context, db *sql.DB) error
- func Open(ctx context.Context, dataDir string) (*sql.DB, error)
- func OpenMemory(ctx context.Context) (*sql.DB, error)
- func OpenPath(ctx context.Context, path string) (*sql.DB, error)
- func OpenTest(t TB) *sql.DB
- func OpenTestPostgres(ctx context.Context, rawURL string) (*sql.DB, error)
- func Rebind(q string) (string, int)
- func RedactURL(raw string) string
- func Register(component string, version int, sql string)
- func RegisterDialect(component string, version int, d Dialect, sql string)
- func SessionLock(ctx context.Context, db *sql.DB, key string) (unlock func(), err error)
- func TranslateDDL(sqliteSQL string) string
- func UnixDay(d Dialect, expr string) string
- func UnixHour(d Dialect, expr string) string
- func UnknownApplied(ctx context.Context, db *sql.DB) ([]string, error)
- type Availability
- func (a *Availability) Down() bool
- func (a *Availability) Fail(err error)
- func (a *Availability) OK()
- func (a *Availability) Report(err error) bool
- func (a *Availability) State() AvailabilityState
- func (a *Availability) Subscribe(fn func(reachable bool)) (unsubscribe func())
- func (a *Availability) Up() <-chan struct{}
- type AvailabilityState
- type Dialect
- type LogicalHeader
- type MigrationGate
- type Options
- type Outbox
- type OutboxOptions
- type OutboxStatus
- type Querier
- type RetryOptions
- type TB
- type TestProxy
Constants ¶
const ( DefaultOutboxCap = 10000 DefaultOutboxTimeout = 2 * time.Second )
Outbox defaults.
const PostgresDriverName = "ether-pgx"
PostgresDriverName is the database/sql driver registered by this package: pgx's stdlib driver with `?` placeholders rebound to $n at execute time, so components keep one SQL text for both engines.
const PreMigrateSuffix = ".pre-migrate.bak"
PreMigrateSuffix names the snapshot taken next to a SQLite database before pending migrations change its schema (ether.db.pre-migrate.bak). Only the latest one is kept: each upgrade that migrates replaces it.
const TestPostgresEnv = "ETHER_TEST_POSTGRES"
TestPostgresEnv names the environment variable that points package tests at a PostgreSQL server instead of in-memory SQLite, e.g.
ETHER_TEST_POSTGRES='postgres://ether:ether@127.0.0.1:5432/ether?sslmode=disable' go test ./...
Each test database is a fresh, randomly named schema that is dropped when the *sql.DB is closed.
Variables ¶
var ErrMigrationDeferred = errors.New("store: schema migration deferred")
ErrMigrationDeferred means this node must not migrate the shared schema yet (another live node runs a binary that does not know the pending migrations and the owner asked to wait for it).
var ErrNoLastInsertID = errors.New("store: LastInsertId is not supported on PostgreSQL; use RETURNING")
ErrNoLastInsertID is returned by Result.LastInsertId on PostgreSQL; use INSERT ... RETURNING instead.
var ErrNotEmpty = errors.New("store: restore target database is not empty")
ErrNotEmpty means a logical restore target already holds data.
Functions ¶
func AppliedMigrations ¶
AppliedMigrations returns the recorded (component -> highest version).
func Contains ¶
Contains is a literal (non-pattern) substring test of needle in haystack, both SQL expressions.
func ExportLogical ¶
ExportLogical writes a consistent logical dump of every table in db to w.
func ImportLogical ¶
ImportLogical loads a logical dump into db, which must be migrated and hold no rows in any table the dump contains. It runs in one transaction.
func IsPermanent ¶
IsPermanent reports whether a PostgreSQL connection error will never succeed on retry: wrong password or user, a database or schema that does not exist, or missing privileges. A node starting against such a URL should stop and tell the owner instead of retrying forever.
func IsUnavailable ¶
IsUnavailable reports whether err means the database server could not be reached or dropped the connection — a condition that clears on its own (network outage, server restart or failover) — rather than a problem with the statement or the data. Deadline errors count: a write bounded by a short timeout that expires is treated as an outage.
func JSONInt ¶
JSONInt extracts the top-level key from the JSON text column col as an integer (NULL when absent). key is not escaped.
func JSONRemove ¶
JSONRemove is the JSON text column col without the top-level key, as text. key is not escaped.
func JSONText ¶
JSONText extracts the top-level key from the JSON text column col as text. key must be a constant identifier-like name; it is not escaped.
func KnownMigrations ¶
KnownMigrations is the highest migration version this binary registers for each component.
func Like ¶
Like is the case-insensitive pattern operator for user-facing search: SQLite's LIKE already ignores ASCII case, PostgreSQL needs ILIKE.
func LockKey ¶
LockKey serialises transactions that use the same key until tx ends: a check-then-insert ("count a user's streams, then reserve one") done inside the lock is atomic across every node sharing the database. d is the database's dialect (DialectOf; *sql.Tx does not expose its driver).
On PostgreSQL it takes pg_advisory_xact_lock on a 64-bit hash of key (the single-key space, which never overlaps the cluster's two-key leader lock); the lock is released at commit or rollback. On SQLite it is a no-op: Ether opens SQLite with _txlock=immediate, so every transaction already holds the database's write lock from BEGIN.
func Maintain ¶
Maintain is the periodic (daily) upkeep for a long-running database: it checkpoints and truncates the WAL so it cannot creep upward while readers keep pinning it, then refreshes query-planner statistics. It deliberately never VACUUMs; that rewrites the whole file and is left to the owner.
On PostgreSQL the server manages its own WAL and autovacuum; Maintain only refreshes planner statistics with ANALYZE.
func OpenMemory ¶
OpenMemory is a migrated, empty database for tests: in-memory SQLite (a single connection), or a fresh PostgreSQL schema when TestPostgresEnv is set.
func OpenPath ¶
OpenPath opens a database file; ":memory:" works for tests (single conn). Like OpenMemory, ":memory:" is a fresh PostgreSQL schema instead when TestPostgresEnv is set.
func OpenTest ¶
OpenTest is OpenMemory for a test: it fails the test on error and closes (on PostgreSQL: drops) the database when the test ends.
func OpenTestPostgres ¶
OpenTestPostgres creates a uniquely named schema on the server at rawURL, migrates it and returns a *sql.DB whose connections use it as their search_path. Closing the DB drops the schema.
func Rebind ¶
Rebind rewrites `?` placeholders as PostgreSQL's $1..$n (a numbered `?NNN` keeps its number), leaving any `?` inside string literals, quoted identifiers and comments untouched. The second result is the number of parameters.
func RedactURL ¶
RedactURL hides the password (and any password query parameter) in a database URL so it can be logged or shown. Non-URL text that looks like a key=value DSN has its password value masked.
func Register ¶
Register adds a migration written in SQLite's dialect. Versions are per component and start at 1. Call from init(); a migration must never change after release. On PostgreSQL the DDL is translated by TranslateDDL unless RegisterDialect supplied an override.
func RegisterDialect ¶
RegisterDialect overrides one registered migration for dialect d, for DDL the mechanical translation cannot express. It may be called before or after the matching Register (both from init()).
func SessionLock ¶
SessionLock holds the lock for key across every node sharing a PostgreSQL database until unlock is called, for work that spans more than one transaction (a site login, for example). It uses the same key space as LockKey on a connection reserved for the lock; if releasing fails, that connection is discarded, which drops the lock with the session. Waiting honours ctx. On SQLite it returns nil, nil: there is one node, so callers keep their in-process lock.
func TranslateDDL ¶
Migrations are written once, in SQLite's dialect. TranslateDDL turns them into PostgreSQL mechanically; a migration the rules cannot express gets a hand-written override through RegisterDialect.
Rules, applied only outside literals and comments, case-insensitively:
- `INTEGER PRIMARY KEY AUTOINCREMENT` becomes `BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY`.
- A bare `id INTEGER PRIMARY KEY` (SQLite's rowid alias used as a generated id) becomes the same identity column. When it is followed by a CHECK (a singleton row such as `CHECK(id=1)`), or the column has any other name (a natural key such as rating_key or user_id that callers always supply), it becomes `BIGINT PRIMARY KEY`.
- INTEGER becomes BIGINT, REAL becomes DOUBLE PRECISION and BLOB becomes BYTEA (SQLite integers are 64-bit, reals are doubles).
Everything else - TEXT, CHECK constraints, DEFAULTs, partial and DESC indexes, REFERENCES ... ON DELETE CASCADE, ALTER TABLE ADD COLUMN and seed INSERTs - is already valid PostgreSQL and passes through unchanged.
Types ¶
type Availability ¶
type Availability struct {
// contains filtered or unexported fields
}
Availability is a node's shared view of whether the database is reachable. Writers that can defer their work (Outbox) and the cluster heartbeat report outcomes to it; once one of them sees an outage the others queue at once instead of each waiting out its own timeout, and all of them flush as soon as any write succeeds again.
A nil *Availability is always available and ignores reports.
func NewAvailability ¶
func NewAvailability(now func() time.Time) *Availability
NewAvailability returns a tracker that starts out available. now may be nil (time.Now).
func (*Availability) Down ¶
func (a *Availability) Down() bool
Down reports whether the database is currently believed unreachable.
func (*Availability) Fail ¶
func (a *Availability) Fail(err error)
Fail marks the database unreachable.
func (*Availability) Report ¶
func (a *Availability) Report(err error) bool
Report records the outcome of a database operation: nil marks the database reachable; an error that IsUnavailable marks it unreachable; any other error says nothing about reachability. It returns whether err was an outage.
func (*Availability) State ¶
func (a *Availability) State() AvailabilityState
State snapshots the tracker.
func (*Availability) Subscribe ¶
func (a *Availability) Subscribe(fn func(reachable bool)) (unsubscribe func())
Subscribe calls fn(reachable) on every transition, synchronously from the reporting goroutine (keep it quick).
func (*Availability) Up ¶
func (a *Availability) Up() <-chan struct{}
Up returns a channel that is closed while (or as soon as) the database is reachable.
type AvailabilityState ¶
type AvailabilityState struct {
Reachable bool `json:"reachable"`
// DownSince is when the current outage was first observed.
DownSince time.Time `json:"downSince,omitzero"`
LastError string `json:"lastError,omitempty"`
}
AvailabilityState is a snapshot for health reports.
type Dialect ¶
type Dialect string
Dialect names the SQL engine behind a *sql.DB. Components write portable SQL with `?` placeholders; the few constructs that differ between engines go through the helpers below.
func DialectOf ¶
DialectOf reports which engine db talks to. Databases opened through this package's PostgreSQL connector report Postgres; everything else SQLite.
func OpenWith ¶
OpenWith opens and migrates the database the options select and reports its dialect.
func OpenWithRetry ¶
OpenWithRetry is OpenWith that keeps retrying a PostgreSQL database that cannot be reached (refused, timed out, shutting down), with backoff, until it opens or ctx ends. It gives up at once on errors retrying cannot fix: an invalid URL, failed authentication, a database that does not exist, an unsupported server or a failing migration. SQLite is opened once, as by OpenWith.
func ParseDialect ¶
ParseDialect maps a configured driver name to a Dialect; anything that is not recognisably PostgreSQL is SQLite.
type LogicalHeader ¶
type LogicalHeader struct {
Format string `json:"format"`
Version int `json:"version"`
Dialect Dialect `json:"dialect"`
Created time.Time `json:"created"`
Migrations map[string]int `json:"migrations"`
}
LogicalHeader is the first line of a logical dump.
func ValidateLogical ¶
func ValidateLogical(r io.Reader) (LogicalHeader, error)
ValidateLogical reads a whole dump and checks its structure and trailer without touching a database. It returns the header.
type MigrationGate ¶
MigrationGate is consulted while the cluster-wide migration lock is held and before any of pending ("component/version", in apply order) is applied. Returning an error wrapping ErrMigrationDeferred makes OpenWithRetry wait and try again; any other error fails the open. q reads inside the migration transaction.
type Options ¶
type Options struct {
// DataDir holds ether.db for SQLite.
DataDir string
// Driver is "sqlite" (default) or "postgres".
Driver string
// URL is the PostgreSQL connection URL (postgres://...). A postgres://
// or postgresql:// URL implies Driver "postgres".
URL string
// Logger receives open-time warnings; nil uses slog.Default().
Logger *slog.Logger
// MigrationGate, when set, decides on PostgreSQL whether this node may
// apply its pending migrations now (see MigrationGate).
MigrationGate MigrationGate
}
Options selects and configures the database backend.
type Outbox ¶
type Outbox[K comparable, V any] struct { // contains filtered or unexported fields }
Outbox makes deferrable writes survive a database outage. A write goes straight to the database while it is reachable. When the database cannot be reached (an error IsUnavailable, or the attempt outlives Timeout) the write is kept in memory instead — only the latest value per key, merged with Merge — and replayed oldest key first, with backoff, once the database answers again. While the database is known to be down new writes are queued without being attempted, so callers (HTTP handlers) return at once.
Errors that are not outages (a constraint violation, a bug) are returned to the caller from Write, or logged and dropped during a replay.
func NewOutbox ¶
func NewOutbox[K comparable, V any](o OutboxOptions[K, V]) *Outbox[K, V]
NewOutbox starts an outbox's replay goroutine; Close stops it.
func (*Outbox[K, V]) Close ¶
Close stops the replay goroutine, then makes one last attempt (bounded by ctx) to write what is still held. It returns an error naming how many writes were lost.
func (*Outbox[K, V]) Status ¶
func (ob *Outbox[K, V]) Status() OutboxStatus
Status snapshots the outbox.
type OutboxOptions ¶
type OutboxOptions[K comparable, V any] struct { // Name identifies the outbox in logs and status ("watch progress"). Name string // Apply performs one write. It must be safe to run again for the same // key (a write whose outcome was lost to a timeout is retried). Apply func(ctx context.Context, key K, v V) error // Merge combines a queued value with a newer one for the same key; // nil keeps only the newer value. Merge func(old, new V) V // Cap bounds the number of queued keys (DefaultOutboxCap); when full // the oldest key is dropped. Cap int // Timeout bounds every write attempt (DefaultOutboxTimeout), so a // caller never waits on an outage longer than this. Timeout time.Duration // MinBackoff / MaxBackoff space retries while the database is // unreachable (1s doubling to 30s). MinBackoff, MaxBackoff time.Duration // Availability is shared with the node's other outboxes and the // cluster heartbeat; nil gives this outbox a private tracker. Availability *Availability Logger *slog.Logger }
OutboxOptions configure an Outbox. Apply is required.
type OutboxStatus ¶
type OutboxStatus struct {
Name string `json:"name"`
Depth int `json:"depth"`
Dropped int64 `json:"dropped,omitempty"`
// Oldest is when the oldest held write was queued.
Oldest time.Time `json:"oldest,omitzero"`
LastError string `json:"lastError,omitempty"`
}
OutboxStatus is an outbox's state for health reports.
type Querier ¶
type Querier interface {
QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
}
Querier is the read side of *sql.DB and *sql.Tx.
type RetryOptions ¶
type RetryOptions struct {
MinBackoff time.Duration // first wait (1s), doubling
MaxBackoff time.Duration // longest wait (30s)
// OnRetry is called before every wait with the failed attempt's error
// (attempt counts from 1).
OnRetry func(attempt int, wait time.Duration, err error)
}
RetryOptions configure OpenWithRetry. Zero durations take the defaults in parentheses.
type TestProxy ¶
type TestProxy struct {
// contains filtered or unexported fields
}
TestProxy is an in-process TCP proxy for outage tests: clients connect to Addr and are forwarded to the target until the test cuts or pauses the link. It never touches the database server itself.
func NewTestProxy ¶
NewTestProxy listens on 127.0.0.1 and forwards to target (host:port).
func (*TestProxy) Cut ¶
func (p *TestProxy) Cut()
Cut drops every connection and refuses new ones until Restore.
func (*TestProxy) Pause ¶
func (p *TestProxy) Pause()
Pause stops all traffic (existing connections hang, new ones are accepted and hang) until Restore.
Source Files