Documentation
¶
Overview ¶
Package ledger implements the durable-task-admission block: idempotency-keyed admission, lease-based ownership with fencing, and dependency-driven blocking on failure.
Map: task_state.go = IdempotencyKey, OwnerID, Sequence, FenceToken, the five Status constants, TaskState, Validate; store.go = Store, MemStore, NewMemStore; ledger.go = Ledger, New, Admit, State, Blocked; claim.go = Claim, Renew, Release, Takeover; complete.go = Complete and the dependency-blocking walk; snapshot.go = Snapshot, Validate, Restore; wire.go = Encode, Decode; events.go = the emitted event names; errors.go = the sentinel errors; sqlite_store.go (behind the ledger_sqlite build tag) = SQLiteStore, NewSQLiteStore, Close, the row-marshal helpers it shares with wire.go, a modernc.org/sqlite-backed Store. Rationale: ../docs/plans/ledger.md. Contribution rules: ../AGENTS.md.
Index ¶
- Constants
- Variables
- type Actor
- type FenceToken
- type IdempotencyKey
- type Ledger
- func (l *Ledger) Admit(ctx context.Context, actor Actor, key IdempotencyKey, seq Sequence, task any, ...) (bool, error)
- func (l *Ledger) Blocked(ctx context.Context, key IdempotencyKey) (IdempotencyKey, bool, error)
- func (l *Ledger) Claim(ctx context.Context, actor Actor, key IdempotencyKey, owner OwnerID, ...) (FenceToken, error)
- func (l *Ledger) Complete(ctx context.Context, actor Actor, key IdempotencyKey, owner OwnerID, ...) error
- func (l *Ledger) Release(ctx context.Context, actor Actor, key IdempotencyKey, owner OwnerID, ...) error
- func (l *Ledger) Renew(ctx context.Context, actor Actor, key IdempotencyKey, owner OwnerID, ...) error
- func (l *Ledger) Restore(ctx context.Context, s Snapshot) error
- func (l *Ledger) Snapshot(ctx context.Context) (Snapshot, error)
- func (l *Ledger) State(ctx context.Context, key IdempotencyKey) (TaskState, bool, error)
- func (l *Ledger) Takeover(ctx context.Context, actor Actor, key IdempotencyKey, owner OwnerID, ...) (FenceToken, error)
- type MemStore
- type MemStoreOptions
- type OwnerID
- type Sequence
- type Snapshot
- type Store
- type TaskState
Constants ¶
const AdmittedEvent events.Name = "ledger.admitted"
AdmittedEvent fires once per successful Admit.
const BlockedEvent events.Name = "ledger.blocked"
BlockedEvent fires once per dependent a failed Complete blocks.
const ClaimedEvent events.Name = "ledger.claimed"
ClaimedEvent fires once per successful Claim.
const CompletedEvent events.Name = "ledger.completed"
CompletedEvent fires once per successful Complete.
const ReleasedEvent events.Name = "ledger.released"
ReleasedEvent fires once per successful Release.
const RenewedEvent events.Name = "ledger.renewed"
RenewedEvent fires once per successful Renew.
const StatusBlocked machine.Status = "blocked"
StatusBlocked marks a record whose named dependency failed.
const StatusClaimed machine.Status = "claimed"
StatusClaimed marks a record owned under a live or stale lease.
const StatusCompleted machine.Status = "completed"
StatusCompleted marks a record whose task finished successfully.
const StatusFailed machine.Status = "failed"
StatusFailed marks a record whose task finished unsuccessfully.
const StatusPending machine.Status = "pending"
StatusPending marks a record admitted but not yet claimed.
const TakenOverEvent events.Name = "ledger.taken_over"
TakenOverEvent fires once per successful Takeover.
Variables ¶
var ErrEmptyOwner = errors.New("ledger: owner must not be empty")
ErrEmptyOwner is returned by Claim or Takeover when owner is empty.
var ErrFenced = errors.New("ledger: fence token is stale")
ErrFenced is returned by Renew, Release, or Complete when the caller's fence token no longer matches the stored record.
var ErrInvalidLease = errors.New("ledger: lease must be positive")
ErrInvalidLease is returned by Claim, Renew, or Takeover when lease is not positive. A lease at or below zero closes the moment it opens, so the record it would write is stale on return.
var ErrInvalidMaxEntries = errors.New("ledger: MaxEntries must not be negative")
ErrInvalidMaxEntries is returned by NewMemStoreWithOptions when MemStoreOptions.MaxEntries is negative.
var ErrLeaseActive = errors.New("ledger: lease is still active")
ErrLeaseActive is returned by Claim when the stored LeaseUntil is still after now, whichever owner holds the lease.
var ErrNoKey = errors.New("ledger: key has no record")
ErrNoKey is returned by Claim, Renew, Release, Takeover, or Complete when the key has no admitted record.
var ErrNotClaimed = errors.New("ledger: record is not claimed")
ErrNotClaimed is returned by Claim, Renew, Release, Complete, or Takeover for a record the caller may not hold. Renew, Release, and Complete return it when the stored record's Status is not StatusClaimed. Claim returns it for a terminal or blocked record, and Takeover for a StatusPending or terminal record. Claim and Takeover also return it for an otherwise eligible record, including a StatusPending one, when a key in its transitive Needs closure holds StatusFailed or StatusBlocked.
var ErrNotStale = errors.New("ledger: lease is not stale")
ErrNotStale is returned by Takeover when the current lease has not yet reached its LeaseUntil deadline.
var ErrUnknownStatus = errors.New("ledger: status must be StatusCompleted or StatusFailed")
ErrUnknownStatus is returned by Complete when status is neither StatusCompleted nor StatusFailed.
Functions ¶
This section is empty.
Types ¶
type Actor ¶
type Actor string
Actor is the caller-chosen identity of whoever performs a write: an external user ID, an agent ID, or any other identifier the caller finds meaningful. Ledger does not validate its shape.
type FenceToken ¶
type FenceToken uint64
FenceToken is a monotonic counter Claim and Takeover return. Renew, Release, and Complete reject a stale token, so a dispossessed owner's late call never mutates the record.
type IdempotencyKey ¶
type IdempotencyKey string
IdempotencyKey is the caller-chosen key that dedupes a task across retries and duplicate submissions.
type Ledger ¶
type Ledger struct {
// contains filtered or unexported fields
}
Ledger is the durable-task-admission handle: admission, lease ownership, fencing, dependency blocking, and snapshot persistence over one Store.
func New ¶
New builds a Ledger over store. A nil store defaults to NewMemStore. A nil bus disables events, matching the flow and machine emit contract.
func (*Ledger) Admit ¶
func (l *Ledger) Admit(ctx context.Context, actor Actor, key IdempotencyKey, seq Sequence, task any, now time.Time, needs ...IdempotencyKey) (bool, error)
Admit records a task once per key. It CAS-admits a StatusPending record when the key is absent or the stored sequence is lower and the stored Status is StatusPending or StatusClaimed. A need whose record already holds StatusFailed or StatusBlocked lands the new record StatusBlocked, with BlockedBy naming that need, so a dependent that arrives after its dependency's failure never claims: late admission blocks, like Complete's dependent scan. Admit validates the record before it writes: a record TaskState.Validate rejects, such as one naming itself in Needs, returns false and that error. A Store fault while reading a need fails Admit; admission never guesses between pending and blocked. After the record inserts, Admit re-reads its needs and blocks the record when a need failed in that window; see recheckNeeds. A Store fault on that re-read returns the error and leaves the record StatusPending. It returns false, nil, not an error, when the key already holds a record at or above seq, or when the stored record is terminal: a duplicate or late-arriving submission is a no-op against a finished task, not a failure. On first insert, Admit sets CreatedBy and CreatedAt from actor and now; on a rebase over an existing non-terminal record, it carries CreatedBy/CreatedAt forward unchanged. Every successful write sets UpdatedBy to actor and UpdatedAt to now. A rebase carries Fence forward from the stored record unchanged, and clears Owner and LeaseUntil, so the next Claim bumps past a dispossessed owner's token.
func (*Ledger) Blocked ¶
func (l *Ledger) Blocked(ctx context.Context, key IdempotencyKey) (IdempotencyKey, bool, error)
Blocked returns the blocking ancestor when key's status is StatusBlocked. The bool means "key is currently blocked"; it is false both for a never-admitted key and for an admitted, unblocked key. A caller who needs to tell those two apart calls State first.
func (*Ledger) Claim ¶
func (l *Ledger) Claim(ctx context.Context, actor Actor, key IdempotencyKey, owner OwnerID, lease time.Duration, now time.Time) (FenceToken, error)
Claim claims a StatusPending record, or a StatusClaimed record whose LeaseUntil is at or before now. LeaseUntil versus now is the only staleness signal Claim reads; ledger keeps no heartbeat or other liveness state. Claim bumps FenceToken and sets LeaseUntil to now.Add(lease).
Claim first calls Store.Load. It returns ErrNoKey when the key has no record, checked before any status precondition and before any CompareAndSwap call: a never-admitted key is never eligible for Claim. It returns ErrLeaseActive when the stored LeaseUntil is still after now, whichever owner holds it; Claim makes no owner comparison, so an owner extends its own lease with Renew. A record in a terminal or blocked status is also ineligible; Claim reports that with ErrNotClaimed, matching Takeover's vocabulary for a non-claimable status. It returns ErrNotClaimed last when a key in the record's transitive Needs closure holds StatusFailed or StatusBlocked, checked after the ErrLeaseActive check and before any CompareAndSwap call. That refusal writes: it moves the record to StatusBlocked through blockOne, naming the nearest blocking ancestor in BlockedBy. See blockingAncestor.
Claim returns ErrInvalidLease when lease is at or below zero. That check runs after the ErrEmptyOwner check and before any Store call. Claim returns the TaskState.Validate error when the record it would write is invalid. That check runs immediately before CompareAndSwap, mirroring Admit.
func (*Ledger) Complete ¶
func (l *Ledger) Complete(ctx context.Context, actor Actor, key IdempotencyKey, owner OwnerID, fence FenceToken, status machine.Status, now time.Time) error
Complete accepts only StatusCompleted or StatusFailed. It returns ErrUnknownStatus when status is neither, checked first, before any Store call: an invalid status argument is a caller error independent of the record's state.
Once status is valid, Complete calls Store.Load. It returns ErrNoKey when the key has no record, checked next, before the ErrFenced and ErrNotClaimed checks and before any CompareAndSwap call. It returns ErrFenced on a stale token. It returns ErrNotClaimed when the record's Status is not StatusClaimed, including a record Complete already moved to a terminal status: a second call against a terminal record never mutates it, even when the fence still matches.
On StatusFailed, Complete walks the dependency graph and sets StatusBlocked, with BlockedBy set to the failed key, on every record that transitively names it in Needs. See blockDependents.
func (*Ledger) Release ¶
func (l *Ledger) Release(ctx context.Context, actor Actor, key IdempotencyKey, owner OwnerID, fence FenceToken, now time.Time) error
Release returns a claimed record to StatusPending. Release first calls Store.Load. It returns ErrNoKey when the key has no record, checked before the ErrFenced and ErrNotClaimed checks and before any CompareAndSwap call. It returns ErrFenced on a stale token. It returns ErrNotClaimed when the record's Status is not StatusClaimed. On a losing CompareAndSwap whose fresh reload still shows the caller's fence owning a StatusClaimed record, Release retries; a fresh reload that fails either check returns the matching sentinel error instead.
func (*Ledger) Renew ¶
func (l *Ledger) Renew(ctx context.Context, actor Actor, key IdempotencyKey, owner OwnerID, fence FenceToken, lease time.Duration, now time.Time) error
Renew extends LeaseUntil to now.Add(lease). Renew first calls Store.Load. It returns ErrNoKey when the key has no record, checked before the ErrNotClaimed and ErrFenced checks and before any CompareAndSwap call. It returns ErrNotClaimed when the record's Status is not StatusClaimed. It returns ErrFenced when fence does not match the current record. On a losing CompareAndSwap whose fresh reload still shows the caller's fence owning a StatusClaimed record, Renew retries; a fresh reload that fails either check returns the matching sentinel error instead.
Renew returns ErrInvalidLease when lease is at or below zero. That check runs before any Store call. Renew returns the TaskState.Validate error when the record it would write is invalid. That check runs immediately before CompareAndSwap, mirroring Admit.
func (*Ledger) Restore ¶
Restore inserts every Snapshot record into Store through CompareAndSwap. It validates each record first; a record failing TaskState.Validate fails the restore with the key named. It is meant for MemStore cold-start or test setup. It returns an error the first time a key fails validation or already has a record; earlier inserts stay in place.
func (*Ledger) Snapshot ¶
Snapshot gathers a point-in-time copy of every record in Store through Range.
func (*Ledger) State ¶
State returns the current record for key. The bool is a found signal: true when key has a record, false when it does not. State never returns an error for a missing key; only Load failing against the Store returns an error.
func (*Ledger) Takeover ¶
func (l *Ledger) Takeover(ctx context.Context, actor Actor, key IdempotencyKey, owner OwnerID, lease time.Duration, now time.Time) (FenceToken, error)
Takeover claims a StatusClaimed record whose LeaseUntil is at or before now: the same staleness signal Claim reads, applied through the same Store.CompareAndSwap call. Takeover bumps FenceToken past the prior value, fencing the dispossessed owner's token.
Takeover first calls Store.Load. It returns ErrNoKey when the key has no record, checked before the ErrNotClaimed and ErrNotStale checks and before any CompareAndSwap call. It returns ErrNotClaimed for a StatusPending or terminal record, checked next: Takeover never admits or claims a never-claimed record, and a caller uses Claim for that. It returns ErrNotStale when LeaseUntil is still after now. It returns ErrNotClaimed last when a key in the record's transitive Needs closure holds StatusFailed or StatusBlocked, checked after the ErrNotStale check and before any CompareAndSwap call. That refusal writes: it moves the record to StatusBlocked through blockOne, naming the nearest blocking ancestor in BlockedBy. See blockingAncestor.
Takeover returns ErrInvalidLease when lease is at or below zero. That check runs after the ErrEmptyOwner check and before any Store call. Takeover returns the TaskState.Validate error when the record it would write is invalid. That check runs immediately before CompareAndSwap, mirroring Admit.
type MemStore ¶
type MemStore struct {
// contains filtered or unexported fields
}
MemStore is the shipped in-memory Store, mutex-guarded. It is the default backend when New receives a nil Store.
With a positive MemStoreOptions.MaxEntries, MemStore deletes records to hold its entry count near the cap. It never deletes a StatusClaimed record whose LeaseUntil is after MemStoreOptions.Now. See MemStoreOptions.MaxEntries and store_eviction.go.
evictQueue is a permutation of the keys of tasks: CompareAndSwap appends on the insert branch only, and eviction deletes a key from both structures together.
func NewMemStore ¶
func NewMemStore() *MemStore
NewMemStore builds an empty, unbounded MemStore ready for use.
func NewMemStoreWithOptions ¶
func NewMemStoreWithOptions(opts MemStoreOptions) (*MemStore, error)
NewMemStoreWithOptions builds an empty MemStore honoring opts. It returns a wrapped ErrInvalidMaxEntries for a negative MaxEntries. A nil opts.Now resolves to time.Now.
func (*MemStore) CompareAndSwap ¶
func (m *MemStore) CompareAndSwap(ctx context.Context, key IdempotencyKey, old TaskState, new TaskState) (bool, error)
CompareAndSwap compares old against the stored record's (Sequence, Status, Fence, Rev) tuple and, on a match, stores new with Rev set to one more than the prior stored Rev. A zero-value old against an absent key inserts new at Rev zero. Any other mismatch, including old against an absent key when old is not the zero value, fails with ok false and no error. An insert raises new.Fence to the store-wide fence floor, so a key's fence never decreases across deletion and re-admission.
type MemStoreOptions ¶
type MemStoreOptions struct {
// MaxEntries caps the number of records MemStore holds. Zero
// means unbounded, matching NewMemStore's existing behavior
// exactly. A positive MaxEntries deletes a record once the entry
// count exceeds the cap. A deleted record is gone: Load and Range
// report found false for its key.
//
// Deletion has three consequences. Idempotency becomes a bounded
// window, because Admit accepts a deleted key again and the task
// can run a second time. A deleted failed or blocked need stops
// blocking its dependents, because a need that is not found
// blocks nothing. A record can be deleted between Admit and
// Claim, or after its lease expired while its owner still works,
// so Claim, Renew, Release, Takeover, and Complete can return
// ErrNoKey for a key the caller admitted.
//
// MaxEntries bounds the records that hold no live lease. It does
// not bound the records that hold one: a StatusClaimed record
// whose LeaseUntil is after Now is never deleted. A caller who
// needs a hard memory bound must bound its own concurrency. A
// caller who needs permanent idempotency must leave MaxEntries at
// zero or use a durable Store.
MaxEntries int
// Now supplies the clock eviction reads to decide whether a
// record's lease is live. A nil Now resolves to time.Now.
// MemStore calls Now while it holds its own lock; Now must not
// call back into the same MemStore, or the call deadlocks.
Now func() time.Time
}
MemStoreOptions configures a MemStore built through NewMemStoreWithOptions.
type Sequence ¶
type Sequence uint64
Sequence is the watermark a caller assigns per submission. Admit rejects a sequence at or below the recorded one.
type Snapshot ¶
type Snapshot struct {
Tasks []TaskState
}
Snapshot is a point-in-time copy of every record in a Store.
func Decode ¶
Decode parses JSON into a Snapshot and validates the result. It rejects malformed JSON and an out-of-range Status on any entry.
type Store ¶
type Store interface {
Load(ctx context.Context, key IdempotencyKey) (TaskState, bool, error)
CompareAndSwap(ctx context.Context, key IdempotencyKey, old TaskState, new TaskState) (bool, error)
Range(ctx context.Context, fn func(TaskState) bool) error
}
Store is the pluggable record backend for TaskState rows. A conforming Store compares old against the stored record on (Sequence, Status, Fence, Rev), not on full-struct or Task-field equality: Task is a caller-owned any value with no defined equality across implementations, and a real backend keys its compare-and-swap off a version or sequence column, not a full-row comparison.
CompareAndSwap with a zero-value old means insert-if-absent: it succeeds only when the key has no stored record yet. On every successful CompareAndSwap, a conforming Store sets the stored record's Rev to one more than the stored record's prior Rev (a newly inserted record starts at Rev zero), regardless of which other fields the write changed. This closes the blind spot a (Sequence, Status, Fence) compare key leaves for Renew: two concurrent Renew calls on the same key and fence would otherwise read the identical triple and both succeed, silently dropping the first writer's lease extension.
Range supports the dependent scan and Snapshot. fn must not call any other Store method on the same Store from inside the callback: Range may hold a lock for the duration of the iteration, and a reentrant Load or CompareAndSwap call from within fn can deadlock against it. fn returns false to stop the iteration early.
type TaskState ¶
type TaskState struct {
Key IdempotencyKey
Status machine.Status
Sequence Sequence
Owner OwnerID
Fence FenceToken
LeaseUntil time.Time
Needs []IdempotencyKey
BlockedBy IdempotencyKey
Task any
Rev uint64
CreatedBy Actor
CreatedAt time.Time
UpdatedBy Actor
UpdatedAt time.Time
}
TaskState is the full record for one idempotency key. Task is caller-owned, like machine.InOut.Input; ledger never inspects it. Rev is a Store-assigned revision counter; a Ledger method reads Rev off the loaded record and forwards it unchanged inside the old argument to Store.CompareAndSwap. Ledger never sets or interprets Rev itself.
func (TaskState) Validate ¶
Validate checks the field rules on a TaskState record. It rejects an empty Key, a Status outside the five declared constants, a Needs entry equal to Key, a non-empty BlockedBy when Status is not StatusBlocked, an empty BlockedBy when Status is StatusBlocked, and a StatusClaimed record with an empty Owner or a zero LeaseUntil.
Source Files