Documentation
¶
Overview ¶
Package drain is the CONSUMER side of the metering durability ladder. The emitter (internal/emit) XADDs metering events to a Valkey Stream as the hot buffer; the drainer reads them via a consumer group and writes each event durably to Postgres, the system of record (pre-rating raw counts).
Delivery is at-least-once: an entry is XACK'd only AFTER its Postgres write commits. A crash before ACK re-delivers the entry, which is safe because the Postgres upsert is idempotent on request_id (INSERT ... ON CONFLICT DO NOTHING). At-least-once + idempotent write = effectively-once.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
// ValkeyAddr is the Valkey/Redis address, e.g. "localhost:6379". Required:
// the drainer has no fallback source — it is the consumer of the stream.
ValkeyAddr string
// StreamName is the stream the emitter XADDs to. MUST match emit.Config's
// StreamName, e.g. "phoebe:metering".
StreamName string
// Group is the consumer-group name. All drainer replicas share one group so
// each stream entry is delivered to exactly one consumer in the group.
Group string
// Consumer is this replica's unique name within the group (e.g. the pod
// name). Pending entries are tracked per-consumer; XAUTOCLAIM reclaims
// entries stranded by a dead consumer.
Consumer string
// BatchSize is the max entries read per XREADGROUP / written per upsert
// batch. Larger batches amortise the Postgres round-trip.
BatchSize int
// BlockTimeout is how long XREADGROUP blocks waiting for new entries before
// returning empty (so the loop can check for shutdown / run reclaim).
BlockTimeout time.Duration
// ClaimMinIdle is the minimum time an entry must sit unacknowledged in
// another consumer's PEL before XAUTOCLAIM may steal it. Set comfortably
// above the expected processing time so we don't reclaim in-flight work.
ClaimMinIdle time.Duration
// ClaimInterval is how often (in drain-loop iterations elapsed) reclaim is
// attempted. Reclaim runs opportunistically; it need not be on every pass.
ClaimInterval time.Duration
// DatabaseURL is the Postgres DSN. Read from the DATABASE_URL env var by the
// binary (Atlas convention). The pgx stdlib driver accepts the standard
// libpq URL form, e.g. "postgres://user:pass@host:5432/db?sslmode=require".
DatabaseURL string
// MaxOpenConns / MaxIdleConns / ConnMaxLifetime tune the database/sql pool.
MaxOpenConns int
MaxIdleConns int
ConnMaxLifetime time.Duration
}
Config holds the drainer's tunable knobs. It mirrors the YAML-free, wire-at-call-site pattern used by emit.Config: the binary's main.go reads a settings file (or env) and translates it into this struct, so this package has no dependency on internal/config.
func DefaultConfig ¶
func DefaultConfig() Config
DefaultConfig returns a Config with safe, production-ready defaults. The caller MUST set ValkeyAddr and DatabaseURL; everything else has a sane value.
type Drainer ¶
type Drainer struct {
// contains filtered or unexported fields
}
Drainer consumes the metering stream via a consumer group and writes each event to the Store. One Drainer == one consumer in the group.
func New ¶
New constructs a Drainer. The group is created lazily on Run (XGROUP CREATE ... MKSTREAM) so the drainer can start before the emitter has produced anything.
func (*Drainer) Poisoned ¶
Poisoned returns the number of events dropped as store-poison (failed their own single-row INSERT while the store was reachable) since start.
func (*Drainer) Run ¶
Run drives the drain loop until ctx is cancelled (SIGTERM). On each pass it:
- opportunistically reclaims entries stranded by a dead consumer (XAUTOCLAIM),
- blocks on XREADGROUP for a batch of new entries,
- stores the batch durably, THEN
- XACKs the batch (never before the store commits).
On ctx cancellation it returns after finishing any in-flight batch — an entry that was read but not yet stored is simply left un-ACK'd and redelivers to the next run (safe: idempotent store).
type PostgresStore ¶
type PostgresStore struct {
// contains filtered or unexported fields
}
PostgresStore writes metering events to the billing_event table. It assumes the table already exists — the drainer does NOT run migrations (the schema is owned by the shared Atlas Alembic chain; see migrations/README.md). Upsert surfaces a clear, semantic error if the table is missing.
func NewPostgresStore ¶
func NewPostgresStore(db *sql.DB) *PostgresStore
NewPostgresStore wraps an existing *sql.DB. Used by tests (sqlmock) and by callers that want to own the pool's lifecycle.
func OpenPostgres ¶
func OpenPostgres(ctx context.Context, cfg Config) (*PostgresStore, error)
OpenPostgres opens a *sql.DB against the given DSN using the pgx stdlib driver, applies the pool settings, and Pings once (pool_pre_ping equivalent) so a misconfigured DSN fails fast at startup rather than on first event.
func (*PostgresStore) Close ¶
func (s *PostgresStore) Close() error
func (*PostgresStore) Upsert ¶
Upsert writes a batch of events in a single transaction with a multi-row INSERT ... ON CONFLICT (request_id) DO NOTHING.
DO NOTHING (not DO UPDATE) is deliberate: a metering Event is immutable once emitted (one record per request, keyed by request_id). A redelivery carries the SAME data, so there is nothing to update — the first write wins and the duplicate is a no-op. This is the cheapest correct idempotency policy and avoids gratuitous row churn / WAL writes on every redelivery.
The whole batch is one transaction so the caller's "XACK only after commit" contract holds at batch granularity: either all rows are durable (ACK the batch) or none are (don't ACK; redeliver).
type Store ¶
type Store interface {
Upsert(ctx context.Context, events []metering.Event) error
// Ping verifies connectivity (pool_pre_ping equivalent on startup).
Ping(ctx context.Context) error
Close() error
}
Store is the durable sink for metering events. It is an interface so the drain loop can be unit-tested against a fake, and so the Postgres SQL can be tested in isolation via sqlmock. The contract is:
- Upsert is IDEMPOTENT on request_id: writing the same event (same request_id) twice produces exactly one row and never errors. This is the load-bearing invariant that makes at-least-once delivery safe.
- Upsert is all-or-nothing per call: it either commits every event in the batch or returns an error having committed none, so the caller can withhold the XACK and let the whole batch redeliver.
Source Files