store

package
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Aug 10, 2026 License: MIT Imports: 26 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	KeyScopePermanent = "permanent"
	KeyScopeLive      = "live"
)

Execution key scopes. Permanent keys identify one execution forever and rediscover it idempotently; live keys are held only while their execution is non-terminal and are released at settlement.

View Source 
const (
	// MaxReadKeys bounds every by-keys batch read before duplicate removal.
	MaxReadKeys = 200
	// MaxReadPageLimit is the largest store read, including the public layer's
	// one-row lookahead used to decide whether a next page exists.
	MaxReadPageLimit = 1001
)
View Source 
const MaxCommandEventWaits = 256
View Source 
const MaxExecutionListLimit = 201

MaxExecutionListLimit includes the one-row look-ahead used to construct a public page cursor. Public pages remain capped at 200 executions.

View Source 
const MaxHistoryLimit = 1000

Variables

View Source 
var ErrLockUnavailable = errors.New("flow store: execution lock unavailable")
View Source 
var ErrTransient = errors.New("flow store: transient database error")

Functions

func MapError 

func MapError(operation string, err error) error

func NotificationChannel 

func NotificationChannel(schema, database string) string

NotificationChannel returns the database-local channel used for bounded wake hints. PostgreSQL already isolates channels by database; including the database and schema identities in the digest also makes the name stable and collision-resistant for diagnostics and tests.

func ParseNotificationHint 

func ParseNotificationHint(payload string) (uuid.UUID, bool)

ParseNotificationHint validates the deliberately tiny, versioned payload. A hint is never durable work and is safe to discard; polling remains the correctness mechanism for malformed or future versions.

Types

type ApplicationEvent 

type ApplicationEvent struct {
	ID   uuid.UUID
	Name string
	Key  string
	Body canonical.Value
}

type ApplyResult 

type ApplyResult struct {
	Journal []JournalRow
}

type AttemptOwnership 

type AttemptOwnership string
const (
	AttemptOwnershipStillOwned AttemptOwnership = "running"
	AttemptOwnershipConcluded  AttemptOwnership = "concluded"
	AttemptOwnershipLost       AttemptOwnership = "lost"
)

type CancelResult 

type CancelResult struct {
	Created bool
}

type ClaimBatchResult 

type ClaimBatchResult struct {
	Commands   []ClaimedCommand
	Progressed bool
}

type ClaimResult 

type ClaimResult struct {
	Command    *ClaimedCommand
	Progressed bool
}

type ClaimedCommand 

type ClaimedCommand struct {
	CommandID              uuid.UUID
	ExecutionID            uuid.UUID
	CommandKey             string
	Name                   string
	Version                int
	Queue                  string
	Args                   []byte
	EventInputs            []ClaimedEventInput
	RetryMaxElapsed        *time.Duration
	AttemptTimeout         time.Duration
	CreatedAt              time.Time
	BudgetStartedAt        time.Time
	ExecutionDeadline      *time.Time
	Attempt                int
	ConsumedAttempts       int
	AttemptID              uuid.UUID
	LeaseToken             uuid.UUID
	DBNow                  time.Time
	LeaseExpiresAt         time.Time
	LocalLeaseExpiresAt    time.Time
	AttemptStartedPosition int64
}

type ClaimedEventInput 

type ClaimedEventInput struct {
	Name     string
	Key      string
	Position int64
	Payload  []byte
}

type CommandCandidate 

type CommandCandidate struct {
	CommandID   uuid.UUID
	ExecutionID uuid.UUID
	Queue       string
	Name        string
	Version     int
	NextRunAt   time.Time
}

type CommandConclusion 

type CommandConclusion struct {
	Claim          ClaimedCommand
	Classification retrypolicy.ErrorClass
	ExplicitDelay  *time.Duration
	Failure        failure.Value
}

type CommandCreate 

type CommandCreate struct {
	ID                     uuid.UUID
	Key                    string
	Name                   string
	Version                int
	Args                   canonical.Value
	DeclarationFingerprint [32]byte
	ParentCommandID        *uuid.UUID
	Required               bool
	Queue                  string
	AttemptTimeout         time.Duration
	RetryPolicy            canonical.Value
	InitialDelay           time.Duration
	Waits                  []EventWaitCreate
	Within                 time.Duration
}

type CommandKind 

type CommandKind struct {
	Name    string
	Version int
}

type CommandProbeCursor 

type CommandProbeCursor struct {
	NextRunAt time.Time
	Queue     string
	CommandID uuid.UUID
}

type CommandSuccess 

type CommandSuccess struct {
	Claim    ClaimedCommand
	Result   canonical.Value
	Events   []ApplicationEvent
	Children []CommandCreate
	Commit   func(pgx.Tx) error
}

type CommitFunctionError 

type CommitFunctionError struct{ Err error }

func (*CommitFunctionError) Error 

func (e *CommitFunctionError) Error() string

func (*CommitFunctionError) Unwrap 

func (e *CommitFunctionError) Unwrap() error

type DBError 

type DBError struct {
	Category   error
	Operation  string
	SQLState   string
	Constraint string
}

func (*DBError) Error 

func (e *DBError) Error() string

func (*DBError) Unwrap 

func (e *DBError) Unwrap() error

type DeadlineSpec 

type DeadlineSpec struct {
	Mode     string
	Duration time.Duration
}

type EntryKind 

type EntryKind string
const (
	ExecutionStarted EntryKind = "execution_started"
	ExecutionFailing EntryKind = "execution_failing"
	CommandCreated   EntryKind = "command_created"
	AttemptStarted   EntryKind = "attempt_started"
	AttemptConcluded EntryKind = "attempt_concluded"
	EventRecorded    EntryKind = "event_recorded"
)

type EventWaitCreate 

type EventWaitCreate struct {
	Name string
	Key  string
}

type ExecutionHead 

type ExecutionHead struct {
	ID           uuid.UUID
	Status       string
	FailFast     bool
	MaxCommands  int
	CommandCount int
	OpenCommands int
}

type ExecutionListFilter 

type ExecutionListFilter struct {
	DefinitionName string
	KeyPrefix      string
	Statuses       []string
	CreatedAfter   *time.Time
	CreatedBefore  *time.Time
	Metadata       []byte
	CursorCreated  *time.Time
	CursorID       *uuid.UUID
	Limit          int
}

type ExecutionRow 

type ExecutionRow struct {
	ID                uuid.UUID
	DefinitionName    string
	DefinitionVersion int
	Key               string
	RootCommandID     *uuid.UUID
	Status            string
	FailFast          bool
	MaxCommands       int
	CommandCount      int
	OpenCommands      int
	DeadlineAt        *time.Time
	Failure           *failure.Value
	CreatedAt         time.Time
	UpdatedAt         time.Time
	StatusAt          time.Time
	FinishedAt        *time.Time
	Metadata          []byte
}

type ExistingEvent 

type ExistingEvent struct {
	ID    uuid.UUID
	Body  []byte
	Found bool
}

type ExpiredLeaseCandidate 

type ExpiredLeaseCandidate struct {
	CommandID   uuid.UUID
	ExecutionID uuid.UUID
}

type ExpiredWaitCandidate 

type ExpiredWaitCandidate struct {
	CommandID   uuid.UUID
	ExecutionID uuid.UUID
}

type JournalEntry 

type JournalEntry struct {
	EntryID             uuid.UUID
	Kind                EntryKind
	CausationPosition   *int64
	CausationBatchIndex *int
	CommandID           *uuid.UUID
	AttemptID           *uuid.UUID
	EventID             *uuid.UUID
	EventNamespace      *string
	EventName           *string
	EventKey            *string
	EventClass          *string
	TerminalStatus      *string
	Body                canonical.Value
}

func NewJournalEntry 

func NewJournalEntry(kind EntryKind, body any) (JournalEntry, error)

type JournalRow 

type JournalRow struct {
	ExecutionID       uuid.UUID
	Position          int64
	EntryID           uuid.UUID
	Kind              EntryKind
	RecordedAt        time.Time
	CausationPosition *int64
	CommandID         *uuid.UUID
	AttemptID         *uuid.UUID
	EventID           *uuid.UUID
	EventNamespace    *string
	EventName         *string
	EventKey          *string
	EventClass        *string
	TerminalStatus    *string
	Body              []byte
	BodyHash          [sha256.Size]byte
}

type KeyedHistoryCursor 

type KeyedHistoryCursor struct {
	ExecutionKey       string
	DefinitionName     string
	ExecutionCreatedAt time.Time
	ExecutionID        uuid.UUID
	Position           int64
}

KeyedHistoryCursor is the last immutable ordering tuple returned by a keyed-history list.

type KeyedHistoryListFilter 

type KeyedHistoryListFilter struct {
	Keys   []string
	Limit  int
	Cursor *KeyedHistoryCursor
}

KeyedHistoryListFilter selects a bounded keyset page of retained history.

type KeyedJournalRow 

type KeyedJournalRow struct {
	DefinitionName     string
	ExecutionKey       string
	KeyScope           string
	ExecutionCreatedAt time.Time
	Entry              JournalRow
}

KeyedJournalRow is a journal entry carrying its execution's identity, for key-addressed reads that span executions.

type LeaseRenewal 

type LeaseRenewal struct {
	CommandID uuid.UUID
	AttemptID uuid.UUID
	Token     uuid.UUID
}

type LeaseRenewalOutcome 

type LeaseRenewalOutcome string
const (
	LeaseRenewed   LeaseRenewalOutcome = "renewed"
	LeaseLost      LeaseRenewalOutcome = "lost"
	LeaseUncertain LeaseRenewalOutcome = "uncertain"
)

type LeaseRenewalResult 

type LeaseRenewalResult struct {
	CommandID      uuid.UUID
	AttemptID      uuid.UUID
	Outcome        LeaseRenewalOutcome
	LeaseExpiresAt *time.Time
}

type LiveWorkCursor 

type LiveWorkCursor struct {
	ExecutionKey       string
	DefinitionName     string
	ExecutionCreatedAt time.Time
	ExecutionID        uuid.UUID
	CommandID          uuid.UUID
}

LiveWorkCursor is the last immutable ordering tuple returned by a live-work list. It is internal store state; the public API binds it to its key filter.

type LiveWorkListFilter 

type LiveWorkListFilter struct {
	Keys   []string
	Limit  int
	Cursor *LiveWorkCursor
}

LiveWorkListFilter selects a bounded keyset page of live work.

type LiveWorkRow 

type LiveWorkRow struct {
	ExecutionID        uuid.UUID
	DefinitionName     string
	ExecutionKey       string
	KeyScope           string
	ExecutionStatus    string
	ExecutionCreatedAt time.Time
	CommandID          uuid.UUID
	CommandKey         string
	CommandName        string
	Queue              string
	QueueState         string
	NextRunAt          time.Time
	LeaseOwner         *string
	LeaseExpiresAt     *time.Time
	AttemptOrdinal     int
	CommandCreatedAt   time.Time
}

LiveWorkRow is one queued (or leased) command of a non-terminal execution, carrying the execution's identity for key-addressed batch reads.

type LockMode 

type LockMode uint8
const (
	LockBlocking LockMode = iota
	LockSkipLocked
)

type LockOrder 

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

LockOrder is transaction-local state used by caller-owned transactions. It rejects reverse execution locking and any return to Flow after the application-write phase begins, before issuing SQL.

func (*LockOrder) BeforeExecution 

func (o *LockOrder) BeforeExecution(id uuid.UUID) error

func (*LockOrder) BeforeFlowOperation 

func (o *LockOrder) BeforeFlowOperation() error

BeforeFlowOperation rejects a Flow operation after application-owned writes have begun without registering an execution lock that the operation has not resolved yet.

func (*LockOrder) BeginApplicationPhase 

func (o *LockOrder) BeginApplicationPhase() error

type PersistedChangeSet 

type PersistedChangeSet struct {
	Journal []JournalEntry
}

type QueueDepthRow 

type QueueDepthRow struct {
	Ready          int64
	Delayed        int64
	Running        int64
	OldestReadyFor time.Duration
}

QueueDepthRow is a point-in-time projection of one queue lane's operational depth, derived from flow_command_queue.

type SemanticTx 

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

func (*SemanticTx) Apply 

func (tx *SemanticTx) Apply(ctx context.Context, changes PersistedChangeSet) (ApplyResult, error)

Apply appends exactly one deterministically ordered semantic batch through a SemanticTx value. Internal batch operations may derive another value over the same owned PostgreSQL transaction and DBNow after the first application.

func (*SemanticTx) Commit 

func (tx *SemanticTx) Commit(ctx context.Context) error

func (*SemanticTx) DBNow 

func (tx *SemanticTx) DBNow() time.Time

func (*SemanticTx) ExecutionID 

func (tx *SemanticTx) ExecutionID() uuid.UUID

func (*SemanticTx) NotifyRunnableCommands 

func (tx *SemanticTx) NotifyRunnableCommands(ctx context.Context) error

NotifyRunnableCommands emits one transactional latency hint after a store operation has created work that is runnable at DBNow. Polling remains the correctness mechanism, and callers must not use this for journal-only or future-scheduled transitions.

func (*SemanticTx) PGX 

func (tx *SemanticTx) PGX() pgx.Tx

func (*SemanticTx) Rollback 

func (tx *SemanticTx) Rollback(ctx context.Context) error

type SettleResult 

type SettleResult struct {
	Retry         bool
	Terminal      bool
	NextAttemptAt *time.Time
	Status        string
}

type StartRequest 

type StartRequest struct {
	ID                uuid.UUID
	DefinitionName    string
	DefinitionVersion int
	Key               string
	KeyScope          string
	StartFingerprint  [32]byte
	Input             canonical.Value
	Metadata          canonical.Value
	FailFast          bool
	Deadline          DeadlineSpec
	MaxCommands       int
	Root              *CommandCreate
}

type StartResult 

type StartResult struct {
	Row     ExecutionRow
	Created bool
}

StartResult carries the accepted execution row as of durable acceptance. Created is false when an idempotent start rediscovered an existing execution.

type Store 

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

func New 

func New(db *pgkit.DB, schema string, notifications bool) (*Store, error)

New constructs the PostgreSQL store. Notifications controls transactional wake hints; correctness never depends on it.

func (*Store) AdoptSemantic 

func (s *Store) AdoptSemantic(tx pgx.Tx, id uuid.UUID, dbNow time.Time) (*SemanticTx, error)

AdoptSemantic wraps a newly inserted execution row that the supplied transaction already owns. It is used only by the start path after the insert has established row ownership and database time has been captured.

func (*Store) AttachSemantic 

func (s *Store) AttachSemantic(ctx context.Context, tx pgx.Tx, id uuid.UUID, mode LockMode) (*SemanticTx, error)

AttachSemantic acquires an execution-first semantic lock inside a caller-owned transaction. The returned value never takes ownership of the transaction; callers that use this entry point remain responsible for its final commit or rollback.

func (*Store) BeginSemantic 

func (s *Store) BeginSemantic(ctx context.Context, id uuid.UUID, mode LockMode) (*SemanticTx, error)

func (*Store) CancelCommandLocked 

func (s *Store) CancelCommandLocked(ctx context.Context, semantic *SemanticTx, commandID uuid.UUID, reason string) (CancelResult, error)

func (*Store) CancelExecutionLocked 

func (s *Store) CancelExecutionLocked(ctx context.Context, semantic *SemanticTx, reason string) (CancelResult, error)

func (*Store) ClaimCommand 

func (s *Store) ClaimCommand(
	ctx context.Context,
	candidate CommandCandidate,
	lease time.Duration,
	owner string,
	hook fault.Hook,
) (ClaimResult, error)

func (*Store) ClaimCommands 

func (s *Store) ClaimCommands(
	ctx context.Context,
	candidates []CommandCandidate,
	lease time.Duration,
	owner string,
	hook fault.Hook,
) (ClaimBatchResult, error)

ClaimCommands claims a bounded set of candidates from one execution under one execution lock and one commit. Candidate rows remain individually skip-locked, so a busy sibling does not make the batch wait.

func (*Store) EmitLocked 

func (s *Store) EmitLocked(ctx context.Context, semantic *SemanticTx, event ApplicationEvent) (bool, error)

func (*Store) ExpireCommandWait 

func (s *Store) ExpireCommandWait(ctx context.Context, candidate ExpiredWaitCandidate) (bool, error)

func (*Store) ExpireExecution 

func (s *Store) ExpireExecution(ctx context.Context, id uuid.UUID, reason string) (bool, error)

func (*Store) GetExecutionInTx 

func (s *Store) GetExecutionInTx(ctx context.Context, tx pgx.Tx, id uuid.UUID) (ExecutionRow, error)

func (*Store) History 

func (s *Store) History(ctx context.Context, id uuid.UUID, after uint64, limit int) ([]JournalRow, error)

func (*Store) HistoryInTx 

func (s *Store) HistoryInTx(ctx context.Context, tx pgx.Tx, id uuid.UUID, after uint64, limit int) ([]JournalRow, error)

HistoryInTx reads history through tx when supplied so transaction-scoped inspection can observe its own uncommitted Flow writes.

func (*Store) ListExecutionsInTx 

func (s *Store) ListExecutionsInTx(ctx context.Context, tx pgx.Tx, filter ExecutionListFilter) ([]ExecutionRow, error)

func (*Store) ListJournalByKeysInTx 

func (s *Store) ListJournalByKeysInTx(ctx context.Context, tx pgx.Tx, filter KeyedHistoryListFilter) ([]KeyedJournalRow, error)

ListJournalByKeysInTx returns one bounded keyset page of retained journal entries for executions that ever held one of the keys, reading through tx when supplied.

func (*Store) ListLiveWorkInTx 

func (s *Store) ListLiveWorkInTx(ctx context.Context, tx pgx.Tx, filter LiveWorkListFilter) ([]LiveWorkRow, error)

ListLiveWorkInTx returns one bounded keyset page of queued commands for non-terminal executions, reading through tx when supplied.

func (*Store) LoadExecutionHead 

func (s *Store) LoadExecutionHead(ctx context.Context, semantic *SemanticTx) (ExecutionHead, error)

func (*Store) LookupApplicationEvent 

func (s *Store) LookupApplicationEvent(ctx context.Context, tx pgx.Tx, executionID uuid.UUID, name, key string) (ExistingEvent, error)

func (*Store) LookupCommandExecution 

func (s *Store) LookupCommandExecution(ctx context.Context, tx pgx.Tx, commandID uuid.UUID) (uuid.UUID, error)

func (*Store) LookupLiveExecutionInTx 

func (s *Store) LookupLiveExecutionInTx(ctx context.Context, tx pgx.Tx, definitionName, key string) (ExecutionRow, bool, error)

LookupLiveExecutionInTx finds the non-terminal execution holding a live-scoped key for one definition. The live-key partial unique index guarantees at most one match.

func (*Store) NotificationChannel 

func (s *Store) NotificationChannel() string

NotificationChannel returns this store's stable LISTEN/NOTIFY channel.

func (*Store) ProbeCommands 

func (s *Store) ProbeCommands(ctx context.Context, kinds []CommandKind, limit int) ([]CommandCandidate, error)

func (*Store) ProbeCommandsExcluding 

func (s *Store) ProbeCommandsExcluding(
	ctx context.Context,
	kinds []CommandKind,
	limit int,
	excludedExecutionIDs []uuid.UUID,
	excludedQueues []string,
	after *CommandProbeCursor,
) ([]CommandCandidate, error)

ProbeCommandsExcluding returns runnable candidates while omitting executions already found to be busy and queues with no process-local lane capacity during the caller's current scheduling pass. This lets a bounded probe make room for other work without broadening the database transaction that tests an execution fence.

func (*Store) ProbeExpiredCommandLeases 

func (s *Store) ProbeExpiredCommandLeases(ctx context.Context, limit int) ([]ExpiredLeaseCandidate, error)

func (*Store) ProbeExpiredCommandWaits 

func (s *Store) ProbeExpiredCommandWaits(ctx context.Context, limit int) ([]ExpiredWaitCandidate, error)

func (*Store) ProbeExpiredExecutions 

func (s *Store) ProbeExpiredExecutions(ctx context.Context, limit int) ([]uuid.UUID, error)

func (*Store) QueueDepthInTx 

func (s *Store) QueueDepthInTx(ctx context.Context, tx pgx.Tx, queue string) (QueueDepthRow, error)

QueueDepthInTx counts the lane's deliverable, scheduled, and leased commands. Ready commands are claimable now; Delayed commands wait out a retry backoff or start delay; Running commands hold a lease.

func (*Store) RecoverExpiredCommandLease 

func (s *Store) RecoverExpiredCommandLease(ctx context.Context, candidate ExpiredLeaseCandidate) (bool, error)

func (*Store) RenewCommandLeases 

func (s *Store) RenewCommandLeases(ctx context.Context, leases []LeaseRenewal, duration time.Duration) ([]LeaseRenewalResult, error)

func (*Store) ResolveCommandAttempt 

func (s *Store) ResolveCommandAttempt(ctx context.Context, commandID, attemptID, token uuid.UUID) (AttemptOwnership, error)

func (*Store) SettleCommandConclusion 

func (s *Store) SettleCommandConclusion(ctx context.Context, request CommandConclusion, hook fault.Hook) (SettleResult, error)

func (*Store) SettleCommandSuccess 

func (s *Store) SettleCommandSuccess(ctx context.Context, request CommandSuccess, hook fault.Hook) (SettleResult, error)

func (*Store) StartInTx 

func (s *Store) StartInTx(ctx context.Context, tx pgx.Tx, request StartRequest, order *LockOrder) (StartResult, error)

StartInTx creates or idempotently rediscovers an execution inside tx. It never commits or rolls back tx.

func (*Store) TraceOperationalInTx 

func (s *Store) TraceOperationalInTx(ctx context.Context, tx pgx.Tx, id uuid.UUID) (TraceOperationalRows, error)

type TraceCommandRow 

type TraceCommandRow struct {
	ID               uuid.UUID
	State            string
	UnsatisfiedWaits int
	BudgetStartedAt  *time.Time
	NextAttemptAt    *time.Time
	WaitStartedAt    *time.Time
	WaitDeadlineAt   *time.Time
	AttemptOrdinal   int
	ConsumedAttempts int
	LastError        *failure.Value
	CreatedAt        time.Time
	UpdatedAt        time.Time
	StatusAt         time.Time
	FinishedAt       *time.Time
	DeliveryState    string
	LeaseOwner       string
	LeaseStartedAt   *time.Time
	LeaseExpiresAt   *time.Time
}

type TraceEventWaitRow 

type TraceEventWaitRow struct {
	CommandID         uuid.UUID
	Name              string
	Key               string
	SatisfiedPosition *int64
}

type TraceOperationalRows 

type TraceOperationalRows struct {
	Commands []TraceCommandRow
	Waits    []TraceEventWaitRow
}

Source Files

View all Source files

Directories

Path Synopsis
Package journalcodec owns version validation for internal journal bodies.
 Package journalcodec owns version validation for internal journal bodies.

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.