inmemory

package
v0.2.0 Latest Latest
Warning

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

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

Documentation

Overview

Package inmemory provides the public concurrency-safe, process-lifetime implementation of the workflow runtime storage contracts.

Store preserves the same CAS, idempotency, defensive-copy, recovery-query, and append-only event semantics required of durable adapters. Its data exists only for the life of one Store value: it does not survive process restart, coordinate across processes, or provide a reopen contract. It is suitable for embedded/offline execution, contract qualification, examples, and tests; a host that promises crash recovery must supply durable storage instead.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Store

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

Store is a concurrency-safe process-lifetime StateStore. It implements the core lifecycle and storage semantics, but no scheduling policy or durable restart guarantee.

func NewStore

func NewStore() *Store

NewStore returns an empty process-lifetime StateStore.

func (*Store) AppendEvent

AppendEvent atomically allocates a monotonic sequence within request.RunID.

func (*Store) ApplyExternalOperation

ApplyExternalOperation atomically records pending progress or a terminal observation. Terminal application also closes the unfinished attempt and moves the waiting aggregate node to its selected outcome.

func (*Store) BindPin

func (*Store) ClaimNode

ClaimNode atomically attempts to acquire a ready node under claim-generation CAS. Non-ready nodes and live leases produce replayable negative results without exposing another owner's lease.

func (*Store) CreateNodeInvocation

CreateNodeInvocation implements runtime.StateStore.

func (*Store) CreateRun

CreateRun implements runtime.StateStore.

func (*Store) FinishNodeAttempt

FinishNodeAttempt atomically closes the latest unfinished running attempt, updates the aggregate node, clears its lease, and appends the derived event.

func (*Store) GetCacheEntry

func (s *Store) GetCacheEntry(ctx context.Context, key string, now time.Time) (workflowruntime.CacheEntry, bool, error)

GetCacheEntry implements runtime.StateStore.

func (*Store) GetPinnedValue

func (s *Store) GetPinnedValue(ctx context.Context, key string, now time.Time) (workflowruntime.PinnedValue, bool, error)

GetPinnedValue implements runtime.StateStore.

func (*Store) ListAttempts

ListAttempts implements runtime.StateStore in attempt-number order.

func (*Store) ListChildRuns

func (s *Store) ListChildRuns(ctx context.Context, parent workflowruntime.RunID) ([]workflowruntime.ChildRunLink, error)

func (*Store) ListCompensationEntries

func (s *Store) ListCompensationEntries(ctx context.Context, runID workflowruntime.RunID) ([]workflowruntime.CompensationEntrySnapshot, error)

func (*Store) ListEvents

func (s *Store) ListEvents(ctx context.Context, query workflowruntime.EventQuery) ([]workflowruntime.Event, error)

ListEvents implements runtime.StateStore in ascending per-run sequence.

func (*Store) ListPinnedValues

func (s *Store) ListPinnedValues(ctx context.Context, now time.Time) ([]workflowruntime.PinnedValue, error)

ListPinnedValues implements runtime.StateStore in key order.

func (*Store) ListRunInvocations

func (s *Store) ListRunInvocations(ctx context.Context, runID workflowruntime.RunID) ([]workflowruntime.NodeInvocationSnapshot, error)

func (*Store) LoadAttempt

LoadAttempt implements runtime.StateStore.

func (*Store) LoadCompensationLedger

func (s *Store) LoadCompensationLedger(ctx context.Context, runID workflowruntime.RunID) (workflowruntime.CompensationLedgerSnapshot, error)

func (*Store) LoadExternalOperation

LoadExternalOperation returns the durable operation bound to one attempt.

func (*Store) LoadFanOutItemResults

func (s *Store) LoadFanOutItemResults(ctx context.Context, parent workflowruntime.NodeInvocationID) ([]workflowruntime.FanOutItemResult, error)

func (*Store) LoadMemoEntry

func (s *Store) LoadMemoEntry(ctx context.Context, key string) (workflowruntime.MemoEntry, error)

func (*Store) LoadNodeInvocation

LoadNodeInvocation implements runtime.StateStore.

func (*Store) LoadPlan

func (s *Store) LoadPlan(ctx context.Context, digest string) (workflowruntime.PlanRef, error)

LoadPlan implements runtime.StateStore.

func (*Store) LoadReplayProvenance

func (s *Store) LoadReplayProvenance(ctx context.Context, runID workflowruntime.RunID) (workflowruntime.ReplayProvenance, error)

func (*Store) LoadRetryActivation

func (s *Store) LoadRetryActivation(ctx context.Context, id string) (workflowruntime.RetryActivationSnapshot, error)

func (*Store) LoadRun

LoadRun implements runtime.StateStore.

func (*Store) LoadRunPolicyDecision

func (s *Store) LoadRunPolicyDecision(ctx context.Context, runID workflowruntime.RunID) (workflowruntime.RunPolicyDecisionSnapshot, error)

func (*Store) LoadTerminalIntent

func (*Store) LoadValueRecord

func (s *Store) LoadValueRecord(ctx context.Context, ref values.ValueSetRef) (workflowruntime.ValueRecord, error)

func (*Store) LoadValues

func (s *Store) LoadValues(ctx context.Context, ref values.ValueSetRef) (values.ValueSet, error)

LoadValues implements runtime.StateStore.

func (*Store) LoadWait

LoadWait implements runtime.StateStore.

func (*Store) LoadWaitContinuation

func (s *Store) LoadWaitContinuation(ctx context.Context, id workflowruntime.AttemptID) (workflowruntime.WaitSnapshot, error)

LoadWaitContinuation returns the most recently resolved resumed wait bound to exactly id. Open and other terminal states are never continuations.

func (*Store) PutCacheEntry

func (s *Store) PutCacheEntry(ctx context.Context, entry workflowruntime.CacheEntry) error

PutCacheEntry implements runtime.StateStore.

func (*Store) PutPinnedValue

func (s *Store) PutPinnedValue(ctx context.Context, pin workflowruntime.PinnedValue) error

PutPinnedValue implements runtime.StateStore.

func (*Store) RecordChildRun

func (s *Store) RecordChildRun(ctx context.Context, link workflowruntime.ChildRunLink) error

func (*Store) RecordExternalActivation

RecordExternalActivation implements runtime.StateStore.

func (*Store) RecordPlan

func (s *Store) RecordPlan(ctx context.Context, plan workflowruntime.PlanRef) error

RecordPlan implements runtime.StateStore.

func (*Store) RecoverCompensation

func (s *Store) RecoverCompensation(ctx context.Context, limit int) ([]workflowruntime.CompensationLedgerSnapshot, error)

func (*Store) RecoverExternalOperations

RecoverExternalOperations returns pending work in deterministic storage order without assigning observation fairness or ownership policy.

func (*Store) RecoverOpenWaits

func (s *Store) RecoverOpenWaits(ctx context.Context, query workflowruntime.OpenWaitQuery) ([]workflowruntime.WaitSnapshot, error)

RecoverOpenWaits returns storage state only; callers own scheduling policy.

func (*Store) RecoverServices

func (*Store) Recovery

Recovery implements runtime.StateStore with deterministic category ordering.

func (*Store) ReleaseNodeClaim

func (s *Store) ReleaseNodeClaim(ctx context.Context, request workflowruntime.ReleaseClaimRequest) error

ReleaseNodeClaim implements runtime.StateStore.

func (*Store) RenewNodeLease

RenewNodeLease implements runtime.StateStore.

func (*Store) RequestExternalOperationCancel

RequestExternalOperationCancel persists cancel intent before an adapter call.

func (*Store) ResumeNodeWait

ResumeNodeWait is the common atomic data-plane mutation used by every wake source after the host authorizer has accepted the responder.

func (*Store) SaveNodeInvocation

SaveNodeInvocation implements runtime.StateStore.

func (*Store) SaveRun

SaveRun implements runtime.StateStore.

func (*Store) SaveValues

SaveValues implements runtime.StateStore.

func (*Store) StartNodeAttempt

StartNodeAttempt atomically starts exactly LatestAttempt+1.

func (*Store) SuspendExternalOperation

SuspendExternalOperation atomically persists adapter-owned work, moves the aggregate node to waiting, and releases the worker lease.

func (*Store) SuspendNodeWait

SuspendNodeWait atomically materializes an open wait, changes the running node to waiting, releases its lease, and appends both durable facts.

func (*Store) TimeoutWait

TimeoutWait atomically marks an open wait, its waiting attempt, and its node timed out while appending all derived events under one store lock.

func (*Store) TransitionNode

TransitionNode atomically validates and persists a non-attempt lifecycle edge and its derived event.

func (*Store) TransitionRun

TransitionRun atomically validates and persists a run status change and its derived lifecycle event.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL