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 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) 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