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 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) 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 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 selects due pending rows (schedule_at <= now) and claims each with an atomic conditional UPDATE; it returns only the rows this call won (rows-affected==1). Two pollers racing the same candidate is safe: the loser's UPDATE matches 0 rows. attempts is bumped on each claim.
Correctness here comes purely from the conditional UPDATE, which is dialect-neutral and holds whether one or many pollers run. A SELECT … FOR UPDATE SKIP LOCKED could cut contention but would need the SELECT + UPDATE in one tx; deferred as an optimisation (correct-but-simple first).
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) 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