state

package
v0.2.45 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 7, 2026 License: MPL-2.0 Imports: 13 Imported by: 0

Documentation

Overview

Package state is the chassis's durable state-record primitive: a named record (tenant, machine, id) with a state, a version and opaque JSON data; an atomic compare-and-swap transition that checks BOTH the current state and the expected version; and a durable event for every committed transition, written in the same transaction (a transactional outbox). The `state` personality presents each event at least once into the tenant's `_state/0` stack.

It is not a workflow engine: no machine definitions, no timers, no retries of the caller's work, no watchers, no deletion. See docs/advanced/protocols/state.md.

Index

Constants

View Source 
const (
	MaxMachineBytes = 64
	MaxIDBytes      = 256
	MaxStateBytes   = 64
	MaxTenantBytes  = 128

	// DefaultMaxDataBytes caps `data` unless SetMaxDataBytes changes it.
	DefaultMaxDataBytes = 64 * 1024
)

Grammar. `machine` plays the role a KV namespace does: a lowercase dotted family name the consumer owns (`onepony.task`). `id` is free text inside the tenant's machine (`acme:t_123`), shaped like a notebook name segment. States are short tokens.

View Source 
const (
	StatusPending = "pending"
	StatusClaimed = "claimed"
	StatusDone    = "done"
	StatusSkipped = "skipped"
	StatusDead    = "dead"
)

Event statuses. pending → claimed → done | skipped | dead; a claim that did not reach acceptance goes back to pending (with a next_attempt_at) or, past the attempt cap, to dead.

View Source 
const AtLayout = "2006-01-02T15:04:05.000000000Z"

AtLayout is the fixed-width UTC timestamp stored in every *_at column (the notebook store's), so text comparison is chronological on both engines.

Variables

This section is empty.

Functions

func ErrorCode 

func ErrorCode(err error) string

ErrorCode returns the code carried anywhere in err's chain, or "".

func Register 

func Register(name string, c Constructor)

Register adds a backend constructor. Called from a backend package's init().

func ValidID 

func ValidID(id string) error

ValidID reports whether id is a record id: 1–MaxIDBytes bytes of UTF-8 with no '/' and no control characters.

func ValidMachine 

func ValidMachine(m string) error

ValidMachine reports whether m is a machine name.

func ValidState 

func ValidState(field, st string) error

ValidState reports whether st is a state token.

Types

type Cause 

type Cause struct {
	Source string `json:"source"` // the committing run's @src (web, lmtp, scheduled, state, …)
	Stack  string `json:"stack"`  // the committing run's stack
	Trace  string `json:"trace"`  // the committing run's rid
	Run    string `json:"run"`    // its continuation run id, when it had one
}

Cause is the provenance of a transition: which run committed it. It is filled by the op layer from trusted context, never from arguments, and rides on the event into `_state/0` as `@state.cause.*`.

type CodedError 

type CodedError interface {
	error
	Code() string
}

CodedError carries a stable txco_state_* code. The op layer surfaces it as `<into>.error.{code,message}` with a nil Go error, so a stack branches on the code and the run continues: a conflict is ordinary control flow, not a failure.

type Config 

type Config struct {
	// DBPath is the bundled SQLite backend's file path (--state-db-path).
	DBPath string
}

Config carries backend-selecting options resolved from chassis config. The bundled "sqlite" backend uses DBPath (same posture as notebook.Config).

type ConflictError 

type ConflictError struct {
	Reason  string
	Current Record
}

ConflictError: the compare-and-swap did not match. Reason is "version" when ExpectedVersion is stale (checked first — a stale actor, doc §10) and "state" when the version matched but the record is not in From. Current is the record as it is now.

func (*ConflictError) Code 

func (e *ConflictError) Code() string

func (*ConflictError) Error 

func (e *ConflictError) Error() string

type Constructor 

type Constructor func(Config) (*Store, error)

Constructor builds a Store from resolved config. It is expected to open its backing DB and call Store.EnsureSchema before returning.

type CreateReq 

type CreateReq struct {
	Tenant  string
	Machine string
	ID      string
	State   string
	Data    json.RawMessage
}

CreateReq creates a record at version 1. Nil Data stores `{}`.

type Event 

type Event struct {
	ID        string    `json:"event_id"` // stev_<hxid>
	Tenant    string    `json:"-"`
	Machine   string    `json:"machine"`
	RecordID  string    `json:"id"`
	Version   int64     `json:"version"` // the version the transition produced
	From      string    `json:"from"`
	To        string    `json:"to"`
	Cause     Cause     `json:"cause"`
	CreatedAt time.Time `json:"created_at"`

	Status string `json:"status"`
	// Attempts counts presentations that consumed an attempt: an accept
	// timeout or a claim reclaimed from a dead node. A denial or a
	// refused handoff does not count. The stack sees Attempts+1 as
	// `@state.attempt`.
	Attempts      int       `json:"attempts"`
	NextAttemptAt time.Time `json:"-"`
	ClaimedBy     string    `json:"-"`
	ClaimedAt     time.Time `json:"-"`
	DeliveredAt   time.Time `json:"-"`
	DeliveredRid  string    `json:"-"`
	LastError     string    `json:"-"`
}

Event is one committed transition and its delivery bookkeeping.

type ExistsError 

type ExistsError struct{ Current Record }

ExistsError: Create found a record already there. Current is what is stored, so the caller can decide whether that is the record it wanted.

func (*ExistsError) Code 

func (e *ExistsError) Code() string

func (*ExistsError) Error 

func (e *ExistsError) Error() string

type InvalidArgError 

type InvalidArgError struct{ Reason string }

InvalidArgError flags a malformed request: a name outside its grammar, data that is not JSON or not UTF-8 or over the cap, a version below 1.

func (*InvalidArgError) Code 

func (e *InvalidArgError) Code() string

func (*InvalidArgError) Error 

func (e *InvalidArgError) Error() string

type NotFoundError 

type NotFoundError struct{ Machine, ID string }

NotFoundError: no record at (tenant, machine, id).

func (*NotFoundError) Code 

func (e *NotFoundError) Code() string

func (*NotFoundError) Error 

func (e *NotFoundError) Error() string

type Record 

type Record struct {
	Tenant    string          `json:"-"`
	Machine   string          `json:"machine"`
	ID        string          `json:"id"`
	State     string          `json:"state"`
	Version   int64           `json:"version"`
	Data      json.RawMessage `json:"data"`
	CreatedAt time.Time       `json:"created_at"`
	UpdatedAt time.Time       `json:"updated_at"`
}

Record is one durable state record.

type Store 

type Store struct {
	// contains filtered or unexported fields
}

Store is the façade over the two tables. It carries the dialect (for `?`→`$n` rebinding), the data cap, and a clock seam for tests, mirroring notebook.Store.

func NewStore 

func NewStore(db *sql.DB, d registry.Dialect) *Store

NewStore builds a Store over the opened DB and its dialect. A nil dialect defaults to SQLite (the in-tree default).

func Open 

func Open(name string, cfg Config) (*Store, error)

Open constructs the named backend. An unknown name is a startup error listing what is available (so a misconfigured --state-store fails loudly).

func (*Store) ClaimDue 

func (s *Store) ClaimDue(ctx context.Context, node string, limit int) ([]Event, error)

ClaimDue atomically moves up to limit due pending events to 'claimed' for node and returns them — one statement, so concurrent claimers (SKIP LOCKED on Postgres; the single writer on SQLite) partition the due set instead of double-claiming.

Presentation order per record: an event is due only when no earlier version of the same record is still pending or claimed. Version 9 is never presented until version 8 has reached a terminal status (done, skipped or dead), even across nodes. Their RUNS may still overlap, because acceptance is not completion; handlers compare versions.

Attempts are not consumed here: the outcome methods decide what counted (see Event.Attempts).

func (*Store) Close 

func (s *Store) Close() error

Close releases the underlying DB handle.

func (*Store) Create 

func (s *Store) Create(ctx context.Context, req CreateReq) (Record, error)

Create stores a record at version 1. A record already at (tenant, machine, id) is an ExistsError carrying what is there; nothing is changed and no event is written (creation is not a transition).

func (*Store) DB 

func (s *Store) DB() *sql.DB

DB exposes the handle for tests and backends. Not for request paths.

func (*Store) EnsureSchema 

func (s *Store) EnsureSchema(ctx context.Context) error

EnsureSchema creates the tables + indexes if absent. Portable DDL (the notebook store's rules): TEXT ids, TEXT fixed-width timestamps, JSON as TEXT, BIGINT counters, native partial indexes — one statement set serves SQLite and Postgres. There is no migration runner for these stores; changes must be additive.

func (*Store) EventsForRecord 

func (s *Store) EventsForRecord(ctx context.Context, tenant, machine, id string) ([]Event, error)

EventsForRecord lists a record's events in version order (tests, inspection).

func (*Store) Get 

func (s *Store) Get(ctx context.Context, tenant, machine, id string) (Record, error)

Get reads a record; NotFoundError when there is none.

func (*Store) GetEvent 

func (s *Store) GetEvent(ctx context.Context, eventID string) (Event, error)

GetEvent reads one event by id; NotFoundError when there is none.

func (*Store) MarkDone 

func (s *Store) MarkDone(ctx context.Context, eventID, rid string) error

MarkDone closes a claim: the chassis accepted the event for execution in the tenant (rid is the run's). Terminal.

func (*Store) MarkSkipped 

func (s *Store) MarkSkipped(ctx context.Context, eventID, reason string) error

MarkSkipped closes a claim without a presentation: the tenant has no active _state stack, or the run never left _sys. Terminal; not replayed if a stack appears later.

func (*Store) Purge 

func (s *Store) Purge(ctx context.Context, retention time.Duration) (int64, error)

Purge deletes terminal events (done, skipped, dead) created before retention ago. Pending and claimed rows are never purged.

func (*Store) ReclaimStale 

func (s *Store) ReclaimStale(ctx context.Context, staleAfter time.Duration, maxAttempts int) (int64, error)

ReclaimStale returns claims older than staleAfter to pending — crash recovery for a node that died after claiming and before an outcome. Each reclaim consumes an attempt (a node that dies on the same event every time must not loop forever), and at maxAttempts the event is dead. Returns the number of rows touched.

func (*Store) Retry 

func (s *Store) Retry(ctx context.Context, eventID string, backoff time.Duration, reason string, consumeAttempt bool, maxAttempts int) error

Retry returns a claim to pending, due again after backoff. With consumeAttempt the presentation counted (an accept timeout: the run may have accepted late, so this is where duplicates come from) and, at maxAttempts, the event is dead instead. Without it (an admission denial, a refused handoff) nothing was presented and nothing is counted.

func (*Store) SetClock 

func (s *Store) SetClock(now func() time.Time)

SetClock replaces the clock (tests).

func (*Store) SetMaxDataBytes 

func (s *Store) SetMaxDataBytes(n int)

SetMaxDataBytes installs the node's cap on `data` (0 = uncapped).

func (*Store) Transition 

func (s *Store) Transition(ctx context.Context, req TransitionReq) (Record, Event, error)

Transition is the compare-and-swap. In one write transaction it (1) updates the record only where state = From AND version = ExpectedVersion, (2) classifies a miss against the current row (NotFoundError; ConflictError "version" first, then "state"), (3) inserts the event for the new version, and (4) commits. The event and the new version are visible together or not at all: a crash between them leaves neither.

Under Postgres READ COMMITTED a second UPDATE racing on the same version blocks on the row, re-evaluates against the winner's new version and matches nothing; on SQLite the _txlock=immediate DSN serializes the two transactions outright. The CAS holds without SERIALIZABLE.

type StoreError 

type StoreError struct {
	Op  string
	Err error
}

StoreError wraps a database or driver failure.

func (*StoreError) Code 

func (e *StoreError) Code() string

func (*StoreError) Error 

func (e *StoreError) Error() string

func (*StoreError) Unwrap 

func (e *StoreError) Unwrap() error

type TransitionReq 

type TransitionReq struct {
	Tenant          string
	Machine         string
	ID              string
	From            string
	To              string
	ExpectedVersion int64
	Data            json.RawMessage
	Cause           Cause
}

TransitionReq moves a record From → To when its current state is From AND its version is ExpectedVersion. Nil Data keeps the stored data; any other value (including `{}`) replaces it. From may equal To: a data-bearing self-transition still bumps the version and emits an event.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.