ledger

package
v19.3.0 Latest Latest
Warning

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

 Go to latest Published: Aug 20, 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 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
}

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

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 SessionRecord 

type SessionRecord struct {
	ID                     uuid.UUID
	Source                 string
	UnitName               string
	IdempotencyKey         string
	ExpectedTerminalStates []int16
	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 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 otelmetric.MeterProvider, tp oteltrace.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) 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) 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) 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) 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 UnobservedSession 

type UnobservedSession struct {
	SessionID   uuid.UUID
	WorkflowKey string
}

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

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 otelmetric.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.