ledger

package
v19.4.0 Latest Latest
Warning

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

 Go to latest Published: Sep 16, 2026 License: MIT Imports: 17 Imported by: 0

Documentation

Overview

Package ledger is the loadtest driver's durable record of executed sessions: the authoritative client-side observation a later verifier consumes. Every session row carries the expectation, and every planned drive op gets an intent row before the op is issued and an outcome row update after it returns, so an op without an outcome is indeterminate rather than failed. The ledger lives in a dedicated logical database with its own migration set.

Index

Constants

View Source 
const (
	// DefaultGroupCommitWindow bounds how long a SessionStarted call waits for
	// peers to share its insert transaction.
	DefaultGroupCommitWindow = 10 * time.Millisecond
	// DefaultBatchCap flushes a batch before its group-commit window elapses
	// once it holds this many items.
	DefaultBatchCap = 512
	// DefaultBufferSize is the capacity of the async entry buffer. Entries
	// arriving at a full buffer are dropped and counted; their rows stay
	// indeterminate.
	DefaultBufferSize = 8192
)
View Source 
const MigrationName = "loadtest_ledger"

MigrationName scopes goose's migration tracking table (goose_<name>). Goose's default advisory lock serializes concurrent migrators per database, which suffices because the ledger owns its logical database.

Variables

This section is empty.

Functions

func MigrationsFS 

func MigrationsFS() fs.FS

MigrationsFS returns the ledger's migration set, exported so tests outside the package can build a pre-migrated template database.

Types

type DisallowedTerminalState added in v19.4.0 

type DisallowedTerminalState struct {
	SessionID              uuid.UUID
	UnitName               string
	WorkflowKey            *string
	ExpectedTerminalStates []int16
	ObservedState          int16
}

DisallowedTerminalState is a session observed in a terminal state outside its expected set. States hold flowtest.TerminalState numeric values.

type MisbehavedOp added in v19.4.0 

type MisbehavedOp struct {
	SessionID   uuid.UUID
	UnitName    string
	WorkflowKey *string
	OpIndex     int32
	Error       string
}

MisbehavedOp is one expected-error op whose recorded outcome contradicts its expectation: Error carries the driver's inverted verdict (the engine accepted the op, or refused it with a different error). Ops without a recorded outcome are out of scope: indeterminate is never guessed.

type Observation 

type Observation struct {
	SessionID uuid.UUID
	State     *int16
	Error     string
}

Observation is one resolved session outcome written by the observer. State holds a flowtest.TerminalState numeric value; nil means the observation itself failed (e.g. the workflow was not found) and Error carries the reason.

type OpKind 

type OpKind int16

OpKind identifies the kind of a planned drive op, stored in loadtest_op.op_kind. 

const (
	OpKindStart  OpKind = 1
	OpKindSend   OpKind = 2
	OpKindCancel OpKind = 3
)

type OpOutcome 

type OpOutcome struct {
	SessionID uuid.UUID
	OpIndex   int32
	Outcome   Outcome
	Error     string
}

OpOutcome is the result of one issued op. Error is empty for OutcomeOK.

type OpRecord 

type OpRecord struct {
	Index   int32
	Kind    OpKind
	Channel string
	// ExpectError marks an op the engine was supposed to refuse: its outcome
	// arrives already inverted, so OutcomeOK means the expected refusal
	// happened and OutcomeError means the engine misbehaved.
	ExpectError bool
}

OpRecord is one planned drive op of a session. Channel is only set for send ops.

type OpStats added in v19.4.0 

type OpStats struct {
	PastDeadline      int64
	Indeterminate     int64
	ExpectErrorJudged int64
}

OpStats are aggregate op counts, each with its own scope. PastDeadline and Indeterminate cover the ops of sessions whose deadline_at lies before the evaluation instant, Indeterminate narrowing that to ops whose intent was recorded but whose outcome never made it to the ledger. ExpectErrorJudged spans the ops of every session in the window instead: it counts expected-error ops whose outcome is recorded, because a verdict is in as soon as it lands, and an op without one stays unjudged (indeterminate is never guessed).

type Outcome 

type Outcome int16

Outcome is the result of an issued op, stored in loadtest_op.outcome. An op row without an outcome is indeterminate: the intent was recorded but the result never made it to the ledger. 

const (
	OutcomeOK    Outcome = 1
	OutcomeError Outcome = 2
)

type SessionProvenance added in v19.4.0 

type SessionProvenance struct {
	Source           string
	UnitName         string
	Seed             *int64
	GeneratorVersion string
	GeneratorConfig  string
}

SessionProvenance is a recorded session's generator provenance, everything a re-render of its unit needs: the source and unit name identify what the session ran, the seed, version and config reproduce it.

type SessionRecord 

type SessionRecord struct {
	ID       uuid.UUID
	Source   string
	UnitName string
	// Seed, GeneratorVersion and GeneratorConfig are the unit's generator
	// provenance: a generated unit is re-renderable from the three together. Nil
	// seed and empty version and config mark a hand-written unit.
	Seed                   *int64
	GeneratorVersion       string
	GeneratorConfig        string
	IdempotencyKey         string
	ExpectedTerminalStates []int16
	// ExpectedTokens are the side-effect tokens the workflow must record via
	// the testkit module, exactly once each; empty means the unit expects none.
	ExpectedTokens []string
	DeadlineAt     time.Time
	Ops            []OpRecord
}

SessionRecord is one driver session: the expectation written before the workflow starts plus one intent row per planned drive op. ExpectedTerminalStates and the observed state passed to SetObservedBatch hold flowtest.TerminalState numeric values; the ledger cannot use the type directly because the driver package imports the ledger.

type SessionStats added in v19.4.0 

type SessionStats struct {
	Total                        int64
	Observed                     int64
	ObservedExpectingTokens      int64
	Acked                        int64
	AckedPastDeadline            int64
	AckedPastDeadlineDelivered   int64
	AckedPastDeadlineUndelivered int64
	Keyed                        int64
	Resolved                     int64
}

SessionStats are aggregate session counts the verifier uses as check populations. Acked counts sessions whose start op (op_index 0) recorded OutcomeOK; AckedPastDeadline narrows that to sessions whose deadline_at lies before the evaluation instant, and AckedPastDeadlineDelivered plus AckedPastDeadlineUndelivered split it by send delivery: whether every send op of the session recorded OutcomeOK (sessions without send ops count as delivered) or at least one failed or stayed indeterminate. ObservedExpectingTokens narrows Observed to sessions whose expectation names side-effect tokens. Keyed counts sessions holding a workflow key; Resolved counts sessions the observer has visited (observed_at set), whatever the resolution.

type SessionWorkflowKey 

type SessionWorkflowKey struct {
	SessionID   uuid.UUID
	WorkflowKey string
}

SessionWorkflowKey links a session to the workflow key the target minted for it.

type Store 

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

Store owns all queries against the ledger's dedicated logical database.

func Open 

func Open(ctx context.Context, log *slog.Logger, dsn string, mp metric.MeterProvider, tp trace.TracerProvider) (*Store, error)

Open connects to the ledger database. The caller owns the Store and must Close it.

func (*Store) Close 

func (s *Store) Close()

func (*Store) CountUnobservedOlderThan 

func (s *Store) CountUnobservedOlderThan(ctx context.Context, createdBefore time.Time) (int64, error)

CountUnobservedOlderThan returns the number of started sessions without an observation that fell out of the observation window: they aged past createdBefore, so the observer stopped polling them and the verifier decides their fate.

func (*Store) EarliestSessionCreation added in v19.4.0 

func (s *Store) EarliestSessionCreation(ctx context.Context) (time.Time, error)

EarliestSessionCreation returns the created_at of the oldest session, or the zero time when the ledger holds no sessions.

func (*Store) IncrementCounter added in v19.4.0 

func (s *Store) IncrementCounter(ctx context.Context, key string) (int64, error)

IncrementCounter atomically increments the counter named by key, creating it at 1, and returns the new value. Counters are testkit scratch state that no invariant reads.

func (*Store) InsertSessions 

func (s *Store) InsertSessions(ctx context.Context, sessions []*SessionRecord) error

InsertSessions writes the session rows and all their op intent rows in one transaction. The whole drive script is known up front, so writing every intent at session insert satisfies intent-before-op for all ops.

func (*Store) ListDisallowedTerminalStates added in v19.4.0 

func (s *Store) ListDisallowedTerminalStates(ctx context.Context, since, until time.Time, limit int) ([]DisallowedTerminalState, int64, error)

ListDisallowedTerminalStates returns up to limit sessions created in [since, until) (zero bounds are unbounded) observed in a terminal state outside their expected set, oldest first, plus the exact total.

func (*Store) ListLostSessions added in v19.4.0 

func (s *Store) ListLostSessions(ctx context.Context, now, since, until time.Time, limit int) ([]UnresolvedSession, int64, error)

ListLostSessions returns up to limit acked sessions created in [since, until) (zero bounds are unbounded) whose send ops all recorded OutcomeOK and that are past their deadline at now without an observed terminal state, oldest first, plus the exact total. A session waiting for a send the driver never delivered is not lost; ListUndeliveredSendSessions owns those.

func (*Store) ListMisbehavedOps added in v19.4.0 

func (s *Store) ListMisbehavedOps(ctx context.Context, since, until time.Time, limit int) ([]MisbehavedOp, int64, error)

ListMisbehavedOps returns up to limit expected-error ops of sessions created in [since, until) (zero bounds are unbounded) whose outcome contradicts their expectation, oldest session first, plus the exact total.

func (*Store) ListTokenMismatches added in v19.4.0 

func (s *Store) ListTokenMismatches(ctx context.Context, since, until time.Time, limit int) ([]TokenMismatch, int64, error)

ListTokenMismatches returns up to limit differences between the expected and recorded side-effect token sets of token-expecting terminal sessions created in [since, until) (zero bounds are unbounded), oldest session first, plus the exact total.

func (*Store) ListUndeliveredSendSessions added in v19.4.0 

func (s *Store) ListUndeliveredSendSessions(ctx context.Context, now, since, until time.Time, limit int) ([]UnresolvedSession, int64, error)

ListUndeliveredSendSessions returns up to limit acked sessions created in [since, until) (zero bounds are unbounded) with at least one send op that failed or stayed indeterminate and that are past their deadline at now without an observed terminal state, oldest first, plus the exact total.

func (*Store) ListUnkeyedSessions added in v19.4.0 

func (s *Store) ListUnkeyedSessions(ctx context.Context, since, until time.Time, limit int) ([]UnkeyedSession, int64, error)

ListUnkeyedSessions returns up to limit sessions created in [since, until) (zero bounds are unbounded) whose start op was acknowledged but that have no workflow key recorded, oldest first, plus the exact total.

func (*Store) ListUnobserved 

func (s *Store) ListUnobserved(ctx context.Context, createdAfter, createdBefore time.Time, limit int) ([]UnobservedSession, error)

ListUnobserved returns up to limit started sessions (workflow key known) without an observation yet, created in (createdAfter, createdBefore), oldest first.

func (*Store) OpStats added in v19.4.0 

func (s *Store) OpStats(ctx context.Context, now, since, until time.Time) (OpStats, error)

OpStats returns the aggregate op counts over the ops of sessions created in [since, until) (zero bounds are unbounded), with deadline comparisons evaluated against now.

func (*Store) RecordToken added in v19.4.0 

func (s *Store) RecordToken(ctx context.Context, sessionID uuid.UUID, token string) error

RecordToken records one side-effect token under the session it names. The token is the destination dedup key: a redelivered action invocation collapses into the row it already wrote.

func (*Store) RunMigrations 

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

RunMigrations applies the ledger's migration set, tracked in the ledger's own goose table.

func (*Store) SessionProvenance added in v19.4.0 

func (s *Store) SessionProvenance(ctx context.Context, sessionID uuid.UUID) (*SessionProvenance, error)

SessionProvenance returns the generator provenance of one session, so a recorded session's unit can be re-rendered from its ledger row alone.

func (*Store) SessionStats added in v19.4.0 

func (s *Store) SessionStats(ctx context.Context, now, since, until time.Time) (SessionStats, error)

SessionStats returns the aggregate session counts in one scan over the sessions created in [since, until) (zero bounds are unbounded), with deadline comparisons evaluated against now.

func (*Store) SetObservedBatch 

func (s *Store) SetObservedBatch(ctx context.Context, observations []Observation) error

SetObservedBatch records the outcomes of many sessions in one round trip. A NULL state with a non-empty error records a workflow that could not be resolved.

func (*Store) SetOpOutcome 

func (s *Store) SetOpOutcome(ctx context.Context, outcomes []OpOutcome) error

SetOpOutcome records the results of issued ops in one batched update. Outcomes for rows that were never inserted (e.g. their session batch failed) update nothing; those ops stay indeterminate.

func (*Store) SetSessionWorkflowKeys 

func (s *Store) SetSessionWorkflowKeys(ctx context.Context, keys []SessionWorkflowKey) error

SetSessionWorkflowKeys records the workflow keys the target minted for the sessions in one batched update, linking the ledger rows to the workflows the observer will resolve.

type TokenMismatch added in v19.4.0 

type TokenMismatch struct {
	SessionID   uuid.UUID
	UnitName    string
	WorkflowKey *string
	Token       string
	Unexpected  bool
}

TokenMismatch is one difference between a terminal session's expected and recorded side-effect token sets. Unexpected marks a token recorded for the session but outside its expected set; otherwise the token was expected but never recorded.

type UnkeyedSession added in v19.4.0 

type UnkeyedSession struct {
	SessionID uuid.UUID
	UnitName  string
}

UnkeyedSession is a session whose start op was acknowledged but that has no workflow key recorded.

type UnobservedSession 

type UnobservedSession struct {
	SessionID   uuid.UUID
	WorkflowKey string
}

UnobservedSession identifies a started session whose terminal state has not been observed yet.

type UnresolvedSession added in v19.4.0 

type UnresolvedSession struct {
	SessionID     uuid.UUID
	UnitName      string
	WorkflowKey   *string
	DeadlineAt    time.Time
	ObservedError *string
}

UnresolvedSession is an acked session past its deadline without an observed terminal state: either never observed at all, or observed with no state (ObservedError then carries the reason, e.g. the workflow was not found).

type Writer 

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

Writer is the batching front of the ledger Store. Session inserts are group-committed: SessionStarted blocks until its batch's transaction commits, so intents are durable before any op is issued, while the batching keeps the open-loop driver's insert rate off the database. All later updates are buffered and flushed asynchronously; under sustained overload they are dropped rather than applying backpressure, leaving the affected rows indeterminate.

func NewWriter 

func NewWriter(log *slog.Logger, store writerStore, meter metric.Meter, opts ...WriterOption) (*Writer, error)

func (*Writer) OpDone 

func (w *Writer) OpDone(sessionID uuid.UUID, opIndex int32, err error)

OpDone records the outcome of an issued op. Non-blocking; dropped under overload, which leaves the op indeterminate.

func (*Writer) Run 

func (w *Writer) Run(ctx context.Context)

Run is the single flusher goroutine. It flushes batches until ctx is canceled, then drains and writes out what is already queued so buffered outcomes survive shutdown.

func (*Writer) SessionStarted 

func (w *Writer) SessionStarted(ctx context.Context, rec *SessionRecord) error

SessionStarted records the session's expectation and all its op intents, blocking until the group-committed insert transaction resolves. Call it before issuing any of the session's ops; a nil return means every intent row is durable.

func (*Writer) WorkflowKeyKnown 

func (w *Writer) WorkflowKeyKnown(sessionID uuid.UUID, key string)

WorkflowKeyKnown records the workflow key the target minted for the session. Non-blocking; dropped under overload.

type WriterOption 

type WriterOption func(*writerConfig)

func WithBatchCap 

func WithBatchCap(n int) WriterOption

WithBatchCap overrides DefaultBatchCap.

func WithBufferSize 

func WithBufferSize(n int) WriterOption

WithBufferSize overrides DefaultBufferSize.

func WithGroupCommitWindow 

func WithGroupCommitWindow(d time.Duration) WriterOption

WithGroupCommitWindow overrides DefaultGroupCommitWindow.

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.