Documentation
¶
Overview ¶
Package scheduled is the durable backing store for the chassis `scheduled` inlet and the txco://schedule op: a table of future events ("run this payload, not before schedule_at") that the scheduled personality polls, claims, and fires.
Storage is dialect-aware (registry.Dialect, the same seam the auth registry uses); the bundled backend is a SQLite file. The CLAIM is the coordination — a node flips a due row pending→claimed with one conditional UPDATE and treats rows-affected==1 as "I won", so a single claimer fires each event even when several pollers share one table.
Index ¶
- func NewEventID() string
- func Register(name string, c Constructor)
- type Claimed
- type Config
- type Constructor
- type Store
- func (s *Store) Cancel(ctx context.Context, tenant, idempotencyKey string) (bool, error)
- func (s *Store) ClaimDue(ctx context.Context, node string, limit int) ([]Claimed, error)
- func (s *Store) Close() error
- func (s *Store) Enqueue(ctx context.Context, tenant, idempotencyKey string, at time.Time, ...) (string, error)
- func (s *Store) EnqueueTx(ctx context.Context, tx *sql.Tx, id, tenant, idempotencyKey string, ...) error
- func (s *Store) EnsureSchema(ctx context.Context) error
- func (s *Store) MarkDone(ctx context.Context, id string) error
- func (s *Store) MarkFailed(ctx context.Context, id string) error
- func (s *Store) Purge(ctx context.Context, retention time.Duration) (int64, error)
- func (s *Store) ReclaimStale(ctx context.Context, staleAfter time.Duration) (int64, error)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func NewEventID ¶ added in v0.2.30
func NewEventID() string
NewEventID mints a scheduled_events row id (`sched_<hxid>`), for a caller that needs the id before EnqueueTx.
func Register ¶
func Register(name string, c Constructor)
Register adds a backend constructor. Called from a backend package's init().
Types ¶
type Claimed ¶
type Claimed struct {
ID string
Tenant string
IdempotencyKey string
Payload json.RawMessage
}
Claimed is a row this node won at ClaimDue: the bits the firing path needs to build the `_scheduled/0` envelope.
type Config ¶
type Config struct {
// DBPath is the bundled SQLite backend's file path (--scheduled-db-path).
DBPath string
}
Config carries backend-selecting options resolved from chassis config. The bundled "sqlite" backend uses DBPath. Adding a field here doesn't affect existing backends (same posture as cron.Config / vector.Config).
type Constructor ¶
Constructor builds a Store from resolved config. It is expected to open its backing DB and call Store.EnsureSchema before returning.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store is the thin façade over the scheduled_events table. It carries the dialect (for `?`→`$n` rebinding + Postgres-only tweaks) and a clock seam for tests, mirroring secrets.Store.
func NewStore ¶
NewStore builds a Store over the opened scheduled DB and its dialect. A nil dialect defaults to SQLite (the in-tree default).
func Open ¶
Open constructs the named backend. Unknown name is a startup error listing what is available (so a misconfigured --scheduled-store fails loudly).
func (*Store) Cancel ¶
Cancel deletes a still-PENDING event for (tenant, idempotency_key). Returns true if a row was removed; false (no error) if it was already fired or never existed. A claimed/done/failed event is immutable and is left untouched.
func (*Store) ClaimDue ¶
ClaimDue claims up to limit due pending rows (schedule_at <= now) in ONE statement and returns the rows this call won. The subselect picks the due set, the UPDATE flips it, RETURNING hands back the claimed rows.
The previous shape — SELECT candidates, then one conditional UPDATE per row — was correct but paid one network round trip per candidate on a shared Postgres store (~5s at 200 due rows, all before the first event fired), and competing pollers each burned the full round-trip count racing for the same rows. Now: on Postgres the subselect takes FOR UPDATE SKIP LOCKED, so concurrent pollers partition the due set without waiting; on SQLite the clause is empty and the single statement is atomic under the single writer. The redundant status guard on the outer UPDATE is a belt for any plan shape — a row can never be claimed twice.
func (*Store) Enqueue ¶
func (s *Store) Enqueue(ctx context.Context, tenant, idempotencyKey string, at time.Time, payload json.RawMessage) (string, error)
Enqueue inserts a pending event, or — if a PENDING row already exists for (tenant, idempotency_key) — reschedules it (new schedule_at + payload). A row that has already left 'pending' (claimed/done/failed) is immutable: the ON CONFLICT update no-ops via the status guard, so a spent key can't be resurrected. Returns the row id (the proposed id on a no-op conflict).
func (*Store) EnqueueTx ¶ added in v0.2.30
func (s *Store) EnqueueTx(ctx context.Context, tx *sql.Tx, id, tenant, idempotencyKey string, at time.Time, payload json.RawMessage) error
EnqueueTx is Enqueue inside the caller's transaction: the row is written on tx and lands (or not) with the caller's commit. id is the row id the caller minted (NewEventID). Same validation and ON CONFLICT semantics as Enqueue.
func (*Store) EnsureSchema ¶
EnsureSchema creates the scheduled_events table + due index if absent. The DDL is portable (TEXT timestamps RFC3339, JSON as TEXT, native partial index); a backend calls it once at construction.
func (*Store) MarkFailed ¶
MarkFailed moves a claimed row to the terminal 'failed' state. Used when a dispatched event's pipeline errored or its response timed out — an at-most-once bias (it likely ran; don't risk a double-fire by retrying). A true crash never reaches here: the row stays 'claimed' for ReclaimStale.
func (*Store) Purge ¶
Purge deletes terminal (done/failed) rows older than retention so the table doesn't grow without bound. Returns the deleted count.
func (*Store) ReclaimStale ¶
ReclaimStale resets rows stuck in 'claimed' past staleAfter back to 'pending' so another node retries — crash recovery for a node that died after claiming but before MarkDone/MarkFailed. Returns the reset count. (RFC3339 UTC timestamps compare lexicographically === chronologically.)
Source Files