workflow

package module
v1.0.0 Latest Latest
Warning

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

 Go to latest Published: Aug 25, 2026 License: MIT Imports: 14 Imported by: 0

README

workflow

CI CodeQL Coverage Mutation Documentation Go Reference Release Go License

workflow provides explicit building blocks for durable business workflows and sagas. Definitions have stable names and immutable versions, and instances are expected to persist the exact definition name, version, and fingerprint that created them.

The module is under active development. The current API covers definition compilation, explicit version migrations, immutable lifecycle history, and deterministic replay. It also defines bounded explicit activity attempts and unknown-outcome semantics, including replay of persisted attempt starts, outcomes, and bounded retry admission times. Durable work can be atomically claimed with bounded leases, monotonically increasing fencing tokens, crash recovery after lease expiry, renewal, retry admission, completion, and explicit dead-letter handling. Bounded workers add tenant-fair admission, lease renewal, stale-owner cancellation, graceful draining, deterministic clocks, explicit retry/dead-letter decisions, and synchronous lifecycle hooks. Durable timer schedules atomically create due work, timer workers persist firing before lease completion, and bounded inbound signals become idempotent transitions that must commit before acknowledgement. Audited lifecycle operator commands atomically record the authorized caller identity and reason before pause, resume, cancel, or terminate. Fenced activity and compensation processors persist attempt starts before handlers and preserve unknown outcomes across redelivery. Ordered orchestration can schedule activities and timers, wait for signals and audited human approvals, atomically admit bounded parallel activity branches, join their persisted outcomes, select and persist bounded signal or approval race winners, schedule version-pinned child workflows, and persist known terminal outcomes. A fenced child-start processor persists each creation attempt before invoking a caller-owned idempotent adapter, records known creation, known absence, or uncertainty, and durably admits policy retries only after a known-absent failure. The PostgreSQL adapter exposes stable unresolved dead-letter pages and audited, idempotent, token-fenced retry or discard commands. Optional composition uses the sibling CloudEvents adapter, outbox PostgreSQL writer and Kafka or queue publishers; core workflow code imports none of them.

Transition is the persistence boundary: its contiguous history events and bounded due-work records must commit atomically. TransitionStore exposes that contract without choosing a database driver, and commit failures distinguish not-committed, committed, and unknown durable outcomes. Callers must reconcile unknown outcomes by transition ID before retrying.

NewActivitySchedule atomically persists bounded activity input with the first due-work record. A worker commits NewActivityAttemptStart before invoking the external handler, then commits an explicit success, known failure, or unknown outcome. NewActivityRetry records the deterministic backoff decision and the next semantic attempt together; work redelivery retains the same attempt idempotency key while a policy retry receives a new one.

The postgres package is the first durable adapter. Its immutable ordered migrations create instance, transition, history, due-work, and dead-letter resolution tables in a caller-owned schema. A commit uses optimistic sequence checks and one PostgreSQL transaction for the transition identity, contiguous history, due work, and current instance position. Exact transition replay is idempotent; conflicting identity reuse is rejected. History reads use a bounded stable forward cursor. A transport error from COMMIT is deliberately classified as unknown rather than retried as if nothing happened. Instance lists use immutable creation-time and identity cursors across active or archived views, and uncertain transitions can be reconciled as missing, exact committed, or conflicting identities. Due-work claims use atomic locked admission with stable ordering. Lease expiry never exceeds the persisted work deadline, and every retry or crash recovery increments the attempt and fencing token so a stale owner cannot complete or release work. A terminal transition archives its instance in the same database transaction, so active and archived list views cannot lag the durable outcome. Unresolved dead letters use failure-time and work-identity keyset pagination. Retry or discard locks the exact work fencing token and writes the authorized actor, reason, action, and complete command fingerprint in the same transaction as retry readmission. Exact command replay is idempotent; conflicting command reuse or a stale work token is rejected, and an uncertain commit must be reconciled by replaying that command identity.

postgres.Store.Stage writes a transition through a caller-owned pgx.Tx without committing or rolling it back. This is the explicit composition point for application state and optional transactional outbox records: stage every record in one transaction, commit it once, and allow no externally observable progression before commit succeeds. A commit transport error remains unknown; reconcile the transition identity before deciding whether the same transaction intent is safe to retry. Stage returning nil means staged or exact replay, not durably committed.

tx, err := pool.Begin(ctx)
if err != nil {
	return err
}
defer tx.Rollback(ctx)

if err := store.Stage(ctx, tx, transition); err != nil {
	return err
}
if err := outboxWriter.Insert(ctx, tx, envelope); err != nil {
	return err
}
if err := tx.Commit(ctx); err != nil {
	// The outcome is unknown: reconcile transition.ID() before retrying.
	return err
}

A WorkProcessor must honor cancellation and stop all of its goroutines before returning. It must persist the workflow transition represented by a work item before returning WorkComplete. If an external activity outcome is unknown, it must first persist unknown-outcome/reconciliation state; returning an error does not make an uncertain side effect safe to redispatch. Worker shutdown stops new claims, cancels active processors, preserves any already-known disposition, and waits for processors to exit. Synchronous worker hooks report bounded claim, readmission, processing, lease-heartbeat, completion, retry, dead-letter, and failure kinds. A readmission may follow an explicit retry or lease-expiry recovery; the package exposes the durable attempt rather than guessing the cause. Work and tenant identities remain event data and must not become metric labels.

StepRace currently accepts signal and approval branches, so selecting a winner cannot imply cancellation of an already-started external side effect. The earliest persisted receive time wins; definition order breaks an equal-time tie. EventRaceWon must commit before later steps advance, and replay never recomputes a different winner from signals accepted afterward.

StepChild pins a complete DefinitionReference. NewChildSchedule commits the parent decision and WorkChild admission atomically. ChildWorkProcessor then records EventChildStartAttempted before a caller-owned adapter creates the child instance. DecodeChildDispatch preserves the exact child identity, semantic attempt, and idempotency key across redelivery. A redelivered in-flight attempt becomes EventChildStartUnknown without calling the adapter again. Only a known-absent retryable failure can create later WorkChild; NewChildOutcome records a known child terminal result before parent orchestration advances. Creating a child remains an idempotent external operation and an uncertain start requires reconciliation; a durably observed terminal child outcome is also sufficient evidence that the child existed.

NewTimerSchedule binds a timer-history decision and its WorkTimer record in one transition. A timer processor persists NewTimerFire and only then returns WorkComplete. NewSignalAcceptance uses the inbound message identity as the transition idempotency boundary; a queue or broker adapter must acknowledge the message only after TransitionStore.Commit succeeds or confirms an exact idempotent replay. Optimistic conflicts require reloading history and deciding whether the signal is already accepted or no longer applicable.

Compensation is explicit durable workflow state rather than an implied rollback. NewCompensationSchedule atomically records the schedule decision with WorkCompensation, and NewCompensationAttemptStart records the exact attempt and idempotency key before the compensating side effect begins. Compensation input inherits the activity step input bound. NewCompensationAttemptOutcome preserves success, known failure, or an unknown outcome, while NewCompensationRetry persists the independent retry decision and its next semantic attempt together. Replay preserves schedule order and manual resolutions. A manual resolution is reported as such; it is never represented as a successful rollback.

NewOperatorLifecycleCommand accepts an already-authorized actor and produces one idempotent optimistic transition containing the audit record followed by the matching lifecycle decision. Replay rejects orphaned or mismatched audit records. Authentication and authorization remain application policy; the package does not infer privileges from actor names. InspectInstance performs bounded deterministic replay over stable history pages, while ExportHistory streams owned pages to a caller sink without accumulating unbounded history or acknowledging external work.

for _, migration := range postgres.SchemaMigrations() {
	if _, err := pool.Exec(ctx, migration.Up); err != nil {
		return err
	}
}

store, err := postgres.New(pool, postgres.Config{}) // schema: workflow

The caller creates and owns the schema, applies migrations in order, rolls them back in reverse order when explicitly authorized, owns the pool, and authorizes every operator actor. The adapter does not publish or acknowledge external messages.

The package does not claim exactly-once external side effects. Applications must make activities idempotent and treat unknown outcomes as requiring reconciliation before retry.

Definition example

definition, err := workflow.NewDefinition(workflow.DefinitionSpec{
	Name:    "order.fulfillment",
	Version: "1",
	Mode:    workflow.Orchestration,
	Steps: []workflow.StepSpec{{
		Name:        "reserve",
		Kind:        workflow.StepActivity,
		Target:      "inventory.reserve",
		Timeout:     time.Minute,
		InputLimit:  16 << 10,
		ResultLimit: 16 << 10,
		Retry: workflow.RetryPolicy{
			MaxAttempts:  3,
			InitialDelay: time.Second,
			MaxDelay:     time.Minute,
		},
	}},
})
if err != nil {
	return err
}

registry, err := workflow.CompileDefinitions(definition)

Definitions are copied at construction and registry compilation rejects a duplicate name/version pair. A new version never changes a running instance; applications must select and durably persist an explicit migration edge.

Documentation

  • Architecture and guarantees explains saga theory, orchestration, choreography, idempotency, versioning, and package boundaries.
  • Operations covers schemas, deployment, recovery, operator commands, reconciliation, capacity, archival, and security.
  • Verification maps failure boundaries to executable evidence and distinguishes local gates from environment-owned drills.
  • Security policy records the trust model and reporting path.
  • Example_durableOrchestration is a compiling end-to-end planning example.

FAQ

Does workflow provide exactly-once activity execution?

No. It persists attempt identity before an activity call and records known or unknown outcomes afterward. The activity implementation must be idempotent and an unknown outcome must be reconciled before redispatch.

Does compensation roll back an external transaction?

No. Compensation is another explicit side effect with its own attempts, timeouts, retry policy, unknown outcomes, and manual-resolution state. A failed compensation never becomes a successful rollback.

Who runs and restarts workers?

The package owns durable claim, fencing, renewal, retry, and shutdown semantics. Kubernetes, ECS, systemd, or another process supervisor owns process lifetime.

Must applications use the sibling queue, outbox, or scheduler modules?

No. Core contracts use explicit values and small interfaces. Applications may compose those modules or equivalent implementations. There is no implicit event bus, scheduler, service container, or package-initialization registration.

Ecosystem

Use the Golib documentation portal to choose companion packages, supported stacks, recipes, and operations guidance.

Documentation

Overview

Package workflow provides explicit durable workflow and saga primitives.

Example (DurableOrchestration) 
package main

import (
	"fmt"
	"time"

	workflow "github.com/faustbrian/go-workflow"
)

func main() {
	now := time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC)
	definition, _ := workflow.NewDefinition(workflow.DefinitionSpec{
		Name: "orders", Version: "1", Mode: workflow.Orchestration,
		Steps: []workflow.StepSpec{{
			Name: "reserve", Kind: workflow.StepActivity, Target: "inventory.reserve",
			Timeout: time.Minute, InputLimit: 1024, ResultLimit: 1024,
			Retry: workflow.RetryPolicy{
				MaxAttempts: 3, InitialDelay: time.Second, MaxDelay: time.Minute,
			},
		}},
	})
	registry, _ := workflow.CompileDefinitions(definition)
	started, _ := workflow.NewHistoryEvent(workflow.HistoryEventSpec{
		Sequence: 1, InstanceID: "order-42", Kind: workflow.EventInstanceStarted,
		OccurredAt: now, Definition: definition.Reference(),
	})
	instance, _ := workflow.Replay(registry, []workflow.HistoryEvent{started})
	decision, _ := workflow.NewOrchestrationDecision(workflow.OrchestrationDecisionSpec{
		TransitionID: "schedule-reserve-42", WorkID: "reserve-work-42",
		Instance: instance, Definition: definition, DecidedAt: now.Add(time.Second),
		Deadline: now.Add(time.Hour), IdempotencyKey: "reserve-attempt-42",
		Input: []byte("order-42"), TenantID: "tenant-1", CorrelationID: "request-1",
	})

	fmt.Println(decision.Kind(), decision.StepName())
	fmt.Println(decision.Transition().Events()[0].Kind())
	fmt.Println(decision.Transition().Work()[0].ID())
}

Output:
1 reserve
11
reserve-work-42

Index

Examples

Constants

View Source 
const (
	// MaxPayloadBytes bounds persisted workflow inputs and results.
	MaxPayloadBytes = 16 << 20
	// MaxFanOut bounds one definition's parallel admission request.
	MaxFanOut = 10_000
)
View Source 
const (
	// MaxTransitionEvents bounds one atomic history append.
	MaxTransitionEvents = 100
	// MaxTransitionWork bounds due-work records created by one atomic append.
	MaxTransitionWork = 1_000
	// MaxTransitionBytes bounds aggregate event and work payload in one append.
	MaxTransitionBytes = MaxPayloadBytes
)
View Source 
const (
	// MaxWorkClaimItems bounds one atomic durable-work admission batch.
	MaxWorkClaimItems uint32 = 100
	// MaxWorkLeaseDuration bounds one ownership interval. Owners must renew
	// before expiry; process lifetime never implies durable ownership.
	MaxWorkLeaseDuration = 15 * time.Minute
)
View Source 
const (
	// MaxWorkerConcurrency bounds goroutines owned by one worker.
	MaxWorkerConcurrency uint32 = 1_000
	// MaxWorkerPollInterval bounds idle and recoverable-error polling.
	MaxWorkerPollInterval = time.Minute
)
View Source 
const (
	// MaxActivityDispatchBytes bounds internal durable dispatch metadata.
	MaxActivityDispatchBytes = 1 << 10
)
View Source 
const (
	// MaxChildDispatchBytes bounds internal version-pinned child metadata.
	MaxChildDispatchBytes = 1 << 10
)
View Source 
const (
	// MaxCompensationDispatchBytes bounds internal durable dispatch metadata.
	MaxCompensationDispatchBytes = 1 << 10
)
View Source 
const (
	// MaxDeadLetterPageItems bounds one stable operator inspection page.
	MaxDeadLetterPageItems uint32 = 100
)
View Source 
const (
	// MaxHistoryPageEvents bounds one stable history page.
	MaxHistoryPageEvents = 1_000
)
View Source 
const (
	// MaxInspectionHistoryEvents bounds one inspection or export operation.
	// Larger histories must be segmented with continue-as-new or read through
	// explicit History pages.
	MaxInspectionHistoryEvents uint32 = 100_000
)
View Source 
const (
	// MaxInstanceListItems bounds one stable instance page.
	MaxInstanceListItems uint32 = 100
)
View Source 
const (
	// MaxOperatorAuditBytes bounds persisted operator identity and reason data.
	MaxOperatorAuditBytes = 1 << 10
)

Variables

View Source 
var (
	// ErrInvalidActivityRequest classifies incomplete or unbounded attempt input.
	ErrInvalidActivityRequest = errors.New("invalid workflow activity request")
	// ErrInvalidActivityOutcome classifies ambiguous or oversized activity
	// results. Unknown external outcomes must use ActivityUnknown explicitly.
	ErrInvalidActivityOutcome = errors.New("invalid workflow activity outcome")
	// ErrInvalidActivity classifies malformed explicit activity registrations.
	ErrInvalidActivity = errors.New("invalid workflow activity")
	// ErrDuplicateActivity classifies duplicate explicit activity names.
	ErrDuplicateActivity = errors.New("duplicate workflow activity")
	// ErrActivityNotFound classifies an unavailable explicitly named activity.
	ErrActivityNotFound = errors.New("workflow activity not found")
)
View Source 
var (
	// ErrInvalidDefinition classifies malformed or unsafe definitions.
	ErrInvalidDefinition = errors.New("invalid workflow definition")
	// ErrDuplicateDefinition classifies duplicate immutable definition keys.
	ErrDuplicateDefinition = errors.New("duplicate workflow definition")
	// ErrDefinitionNotFound classifies an unavailable pinned definition.
	ErrDefinitionNotFound = errors.New("workflow definition not found")
	// ErrInvalidMigration classifies incomplete or incoherent migrations.
	ErrInvalidMigration = errors.New("invalid workflow migration")
	// ErrDuplicateMigration classifies duplicate explicit migration edges.
	ErrDuplicateMigration = errors.New("duplicate workflow migration")
	// ErrMigrationNotFound classifies an unavailable explicit migration edge.
	ErrMigrationNotFound = errors.New("workflow migration not found")
)
View Source 
var (
	// ErrInvalidDefinitionReference classifies malformed persisted definition
	// identities.
	ErrInvalidDefinitionReference = errors.New("invalid workflow definition reference")
	// ErrInvalidHistoryEvent classifies malformed durable history records.
	ErrInvalidHistoryEvent = errors.New("invalid workflow history event")
	// ErrEmptyHistory reports that replay has no persisted instance start.
	ErrEmptyHistory = errors.New("workflow history is empty")
	// ErrHistoryConflict classifies gaps, mixed instances, or non-monotonic time.
	ErrHistoryConflict = errors.New("workflow history conflict")
	// ErrInvalidTransition classifies an event that is illegal for current state.
	ErrInvalidTransition = errors.New("invalid workflow transition")
	// ErrDefinitionMismatch reports silent behavior reinterpretation for a pinned
	// name and version.
	ErrDefinitionMismatch = errors.New("workflow definition fingerprint mismatch")
)
View Source 
var (
	// ErrInvalidStoreRequest classifies malformed or unbounded store input.
	ErrInvalidStoreRequest = errors.New("invalid workflow store request")
	// ErrStoreNotFound reports an unavailable workflow instance or work item.
	ErrStoreNotFound = errors.New("workflow store record not found")
	// ErrStoreConflict reports an optimistic sequence mismatch.
	ErrStoreConflict = errors.New("workflow store sequence conflict")
	// ErrDuplicateTransition reports reuse of a transition identity with
	// different content. An exact idempotent replay is not an error.
	ErrDuplicateTransition = errors.New("workflow transition identity conflict")
	// ErrStaleWorkLease reports a stale owner or fencing token.
	ErrStaleWorkLease = errors.New("stale workflow work lease")
)
View Source 
var (
	// ErrInvalidPendingWork classifies incomplete or unbounded durable work.
	ErrInvalidPendingWork = errors.New("invalid workflow pending work")
	// ErrInvalidTransitionPlan classifies a non-atomic or incoherent append plan.
	ErrInvalidTransitionPlan = errors.New("invalid workflow transition plan")
)
View Source 
var (
	// ErrHistoryLimitExceeded reports that inspection or export reached its
	// caller-selected bound while more durable history remained.
	ErrHistoryLimitExceeded = errors.New("workflow history traversal limit exceeded")
)
View Source 
var (
	// ErrInvalidActivityProcessor classifies malformed activity processor
	// configuration or durable work that cannot represent a valid attempt.
	ErrInvalidActivityProcessor = errors.New("invalid workflow activity processor")
)
View Source 
var (
	// ErrInvalidActivityTransition classifies malformed or state-incoherent
	// durable activity progression.
	ErrInvalidActivityTransition = errors.New("invalid workflow activity transition")
)
View Source 
var (
	// ErrInvalidChildProcessor classifies malformed processor configuration or
	// poison durable child work.
	ErrInvalidChildProcessor = errors.New("invalid workflow child processor")
)
View Source 
var (
	// ErrInvalidChildStart classifies malformed child-start requests or
	// outcomes. A starter must return an explicit outcome even when creation
	// may have succeeded.
	ErrInvalidChildStart = errors.New("invalid workflow child start")
)
View Source 
var (
	// ErrInvalidChildTransition classifies malformed or incoherent child plans.
	ErrInvalidChildTransition = errors.New("invalid workflow child transition")
)
View Source 
var (
	// ErrInvalidCompensation classifies malformed compensation persistence or
	// dispatch requests.
	ErrInvalidCompensation = errors.New("invalid workflow compensation")
)
View Source 
var (
	// ErrInvalidCompensationProcessor classifies malformed compensation
	// processor configuration or poison durable compensation work.
	ErrInvalidCompensationProcessor = errors.New("invalid workflow compensation processor")
)
View Source 
var (
	// ErrInvalidOperatorCommand classifies malformed, unauthorized-by-state, or
	// unbounded operator command input. Callers must authorize actors before
	// constructing a command.
	ErrInvalidOperatorCommand = errors.New("invalid workflow operator command")
)
View Source 
var (
	// ErrInvalidOrchestration classifies malformed orchestration input or a
	// definition step whose durable execution semantics are unsupported.
	ErrInvalidOrchestration = errors.New("invalid workflow orchestration decision")
)
View Source 
var (
	// ErrInvalidWait classifies malformed timer or signal persistence plans.
	ErrInvalidWait = errors.New("invalid workflow wait")
)
View Source 
var (
	// ErrInvalidWorkLease classifies malformed or unbounded claim and fencing input.
	ErrInvalidWorkLease = errors.New("invalid workflow work lease")
)
View Source 
var (
	// ErrInvalidWorker classifies malformed worker configuration, decisions, or
	// adapter output.
	ErrInvalidWorker = errors.New("invalid workflow worker")
)

Functions

func ExportHistory 

func ExportHistory(
	ctx context.Context,
	reader HistoryReader,
	spec HistoryExportSpec,
	sink HistoryExportSink,
) error

ExportHistory streams owned stable pages without accumulating an unbounded in-memory export. It performs no external acknowledgement.

func NewStoreCommitError 

func NewStoreCommitError(outcome StoreCommitOutcome, cause error) error

NewStoreCommitError classifies one store commit failure.

Types

type Activity 

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

Activity is one explicit named handler without reflection or global registration.

func NewActivity 

func NewActivity(name string, handler ActivityHandler) (Activity, error)

NewActivity validates one explicit stable activity registration.

func (Activity) Execute 

func (activity Activity) Execute(ctx context.Context, request ActivityRequest) (ActivityOutcome, error)

Execute derives the persisted deadline, checks cancellation before external work, invokes the handler synchronously, and validates its bounded outcome. It does not recover panics or claim that context cancellation stopped an arbitrary external operation.

func (Activity) Name 

func (activity Activity) Name() string

Name returns the stable explicit activity name.

type ActivityAttemptOutcomeSpec 

type ActivityAttemptOutcomeSpec struct {
	TransitionID string
	Instance     Instance
	Definition   Definition
	StepName     string
	Attempt      uint32
	OccurredAt   time.Time
	Outcome      ActivityOutcome
}

ActivityAttemptOutcomeSpec supplies one known or unknown persisted result.

type ActivityAttemptStartSpec 

type ActivityAttemptStartSpec struct {
	TransitionID string
	Lease        WorkLease
	Instance     Instance
	Definition   Definition
	StartedAt    time.Time
}

ActivityAttemptStartSpec supplies the persisted decision required before an activity side effect begins.

type ActivityDispatch 

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

ActivityDispatch is immutable durable metadata for one semantic attempt. Redelivery of one work record retains this identity.

func DecodeActivityDispatch 

func DecodeActivityDispatch(payload []byte) (ActivityDispatch, error)

DecodeActivityDispatch validates bounded durable work metadata.

func (ActivityDispatch) Attempt 

func (dispatch ActivityDispatch) Attempt() uint32

Attempt returns the one-based semantic activity attempt.

func (ActivityDispatch) IdempotencyKey 

func (dispatch ActivityDispatch) IdempotencyKey() string

IdempotencyKey returns the stable external side-effect identity.

func (ActivityDispatch) StepName 

func (dispatch ActivityDispatch) StepName() string

StepName returns the definition activity step.

type ActivityExecutionStore 

type ActivityExecutionStore interface {
	TransitionStore
	ReconcileTransition(context.Context, TransitionReconciliation) (TransitionReconciliationOutcome, error)
}

ActivityExecutionStore is the narrow durable contract needed to execute an activity. Commit and reconciliation must address the same durable store.

type ActivityHandler 

type ActivityHandler func(context.Context, ActivityRequest) ActivityOutcome

ActivityHandler performs one explicitly bounded external activity attempt. It must return ActivityUnknown when cancellation, timeout, or transport loss leaves external commitment uncertain.

type ActivityOutcome 

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

ActivityOutcome is one immutable explicit external-operation classification.

func NewActivityOutcome 

func NewActivityOutcome(spec ActivityOutcomeSpec) (ActivityOutcome, error)

NewActivityOutcome validates and owns one explicit activity result.

func (ActivityOutcome) Code 

func (outcome ActivityOutcome) Code() string

Code returns the stable safe application failure or reconciliation code.

func (ActivityOutcome) Data 

func (outcome ActivityOutcome) Data() []byte

Data returns an owned copy of result or safe persisted failure details.

func (ActivityOutcome) Kind 

func (outcome ActivityOutcome) Kind() ActivityOutcomeKind

Kind returns the explicit external-operation classification.

func (ActivityOutcome) Retryable 

func (outcome ActivityOutcome) Retryable() bool

Retryable reports whether a known failure permits definition-policy retry. It is always false for unknown outcomes.

type ActivityOutcomeKind 

type ActivityOutcomeKind uint8

ActivityOutcomeKind distinguishes success, known failure, and an outcome that may have committed externally and therefore cannot be blindly retried. 

const (
	// ActivitySucceeded records a known successful external outcome.
	ActivitySucceeded ActivityOutcomeKind = 1
	// ActivityFailed records a known failed external outcome.
	ActivityFailed ActivityOutcomeKind = 2
	// ActivityUnknown records that an external side effect may have committed.
	ActivityUnknown ActivityOutcomeKind = 3
)

type ActivityOutcomeSpec 

type ActivityOutcomeSpec struct {
	Kind      ActivityOutcomeKind
	Code      string
	Retryable bool
	Data      []byte
}

ActivityOutcomeSpec supplies one explicit bounded activity result.

type ActivityProgress 

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

ActivityProgress is immutable state reconstructed only from persisted events.

func (ActivityProgress) Attempt 

func (progress ActivityProgress) Attempt() uint32

Attempt returns the latest one-based attempt, or zero before the first attempt.

func (ActivityProgress) Code 

func (progress ActivityProgress) Code() string

Code returns the latest known-failure or unknown-outcome code.

func (ActivityProgress) DueAt 

func (progress ActivityProgress) DueAt() time.Time

DueAt returns the current attempt deadline or retry admission time.

func (ActivityProgress) IdempotencyKey 

func (progress ActivityProgress) IdempotencyKey() string

IdempotencyKey returns the latest externally visible attempt key.

func (ActivityProgress) Input 

func (progress ActivityProgress) Input() []byte

Input returns an owned copy of the durably scheduled activity input.

func (ActivityProgress) Result 

func (progress ActivityProgress) Result() []byte

Result returns an owned copy of result or safe failure details.

func (ActivityProgress) Retryable 

func (progress ActivityProgress) Retryable() bool

Retryable reports the persisted known-failure retry classification.

func (ActivityProgress) Status 

func (progress ActivityProgress) Status() ActivityProgressStatus

Status returns the durable activity progress state.

func (ActivityProgress) StepName 

func (progress ActivityProgress) StepName() string

StepName returns the stable definition step name.

type ActivityProgressStatus 

type ActivityProgressStatus uint8

ActivityProgressStatus identifies replayed durable activity progress. 

const (
	// ActivityProgressReady is durable and eligible for attempt admission.
	ActivityProgressReady ActivityProgressStatus = 1
	// ActivityProgressRunning has one externally observable in-flight attempt.
	ActivityProgressRunning ActivityProgressStatus = 2
	// ActivityProgressSucceeded is a known successful terminal step outcome.
	ActivityProgressSucceeded ActivityProgressStatus = 3
	// ActivityProgressFailed is a known failed attempt awaiting policy action.
	ActivityProgressFailed ActivityProgressStatus = 4
	// ActivityProgressUnknown requires reconciliation or operator resolution.
	ActivityProgressUnknown ActivityProgressStatus = 5
	// ActivityProgressRetryWaiting has a persisted next-attempt admission time.
	ActivityProgressRetryWaiting ActivityProgressStatus = 6
)

type ActivityRegistry 

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

ActivityRegistry is an immutable explicit activity registry.

func CompileActivities 

func CompileActivities(activities ...Activity) (*ActivityRegistry, error)

CompileActivities validates activities and rejects duplicate stable names.

func (*ActivityRegistry) Resolve 

func (registry *ActivityRegistry) Resolve(name string) (Activity, error)

Resolve returns one explicitly registered activity.

type ActivityRequest 

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

ActivityRequest is immutable attempt input. Its exact idempotency key is stable for one attempt; an unknown outcome must be reconciled before a caller starts another key or attempt.

func NewActivityRequest 

func NewActivityRequest(spec ActivityRequestSpec) (ActivityRequest, error)

NewActivityRequest validates and owns one bounded activity attempt.

func (ActivityRequest) Attempt 

func (request ActivityRequest) Attempt() uint32

Attempt returns the one-based attempt number.

func (ActivityRequest) CorrelationID 

func (request ActivityRequest) CorrelationID() string

CorrelationID returns optional propagated correlation identity.

func (ActivityRequest) Deadline 

func (request ActivityRequest) Deadline() time.Time

Deadline returns the persisted attempt deadline.

func (ActivityRequest) Definition 

func (request ActivityRequest) Definition() DefinitionReference

Definition returns the exact behavior identity that scheduled the attempt.

func (ActivityRequest) IdempotencyKey 

func (request ActivityRequest) IdempotencyKey() string

IdempotencyKey returns the stable application-visible attempt key.

func (ActivityRequest) Input 

func (request ActivityRequest) Input() []byte

Input returns an owned copy of bounded activity input.

func (ActivityRequest) InputLimit 

func (request ActivityRequest) InputLimit() uint32

InputLimit returns the immutable maximum input size.

func (ActivityRequest) InstanceID 

func (request ActivityRequest) InstanceID() string

InstanceID returns the durable workflow instance identity.

func (ActivityRequest) MaxAttempts 

func (request ActivityRequest) MaxAttempts() uint32

MaxAttempts returns the immutable definition retry bound.

func (ActivityRequest) ResultLimit 

func (request ActivityRequest) ResultLimit() uint32

ResultLimit returns the immutable maximum result or failure-detail size.

func (ActivityRequest) StartedAt 

func (request ActivityRequest) StartedAt() time.Time

StartedAt returns canonical persisted attempt-start time.

func (ActivityRequest) StepName 

func (request ActivityRequest) StepName() string

StepName returns the stable definition step name.

func (ActivityRequest) TenantID 

func (request ActivityRequest) TenantID() string

TenantID returns optional propagated tenant identity. It is data, not a metric-label recommendation.

type ActivityRequestSpec 

type ActivityRequestSpec struct {
	InstanceID     string
	Definition     DefinitionReference
	StepName       string
	Attempt        uint32
	MaxAttempts    uint32
	IdempotencyKey string
	StartedAt      time.Time
	Deadline       time.Time
	Input          []byte
	InputLimit     uint32
	ResultLimit    uint32
	TenantID       string
	CorrelationID  string
}

ActivityRequestSpec supplies bounded persisted activity-attempt metadata.

type ActivityRetrySpec 

type ActivityRetrySpec struct {
	TransitionID   string
	WorkID         string
	Instance       Instance
	Definition     Definition
	StepName       string
	IdempotencyKey string
	ScheduledAt    time.Time
	Deadline       time.Time
	TenantID       string
	CorrelationID  string
}

ActivityRetrySpec supplies one persisted retry decision and next-attempt work.

type ActivityScheduleSpec 

type ActivityScheduleSpec struct {
	TransitionID   string
	WorkID         string
	Instance       Instance
	Definition     Definition
	StepName       string
	Attempt        uint32
	IdempotencyKey string
	ScheduledAt    time.Time
	Deadline       time.Time
	Input          []byte
	TenantID       string
	CorrelationID  string
}

ActivityScheduleSpec supplies atomic history and first-attempt due work.

type ActivityWorkProcessor 

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

ActivityWorkProcessor executes leased activity work without claiming exactly-once external effects. It persists attempt start before invocation and converts an already-running redelivery to an unknown durable outcome.

func NewActivityWorkProcessor 

func NewActivityWorkProcessor(config ActivityWorkProcessorConfig) (*ActivityWorkProcessor, error)

NewActivityWorkProcessor validates one bounded explicit processor.

func (*ActivityWorkProcessor) Process 

func (processor *ActivityWorkProcessor) Process(ctx context.Context, lease WorkLease) (WorkDecision, error)

Process persists each externally observable activity boundary. Poison work is dead-lettered; store failures retain the lease for fenced recovery.

type ActivityWorkProcessorConfig 

type ActivityWorkProcessorConfig struct {
	Store            ActivityExecutionStore
	Definitions      *Registry
	Activities       *ActivityRegistry
	Clock            Clock
	PageSize         uint32
	MaxHistoryEvents uint32
}

ActivityWorkProcessorConfig supplies explicit bounded activity execution dependencies. Definitions and activities are immutable explicit registries.

type AdministrationStore 

type AdministrationStore interface {
	ListInstances(context.Context, InstanceListQuery) (InstanceListPage, error)
	ReconcileTransition(context.Context, TransitionReconciliation) (TransitionReconciliationOutcome, error)
}

AdministrationStore exposes stable list and uncertain-commit reconciliation.

type ChildDispatch 

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

ChildDispatch is immutable durable metadata for starting one pinned child.

func DecodeChildDispatch 

func DecodeChildDispatch(payload []byte) (ChildDispatch, error)

DecodeChildDispatch validates bounded durable work metadata without starting or otherwise observing the child workflow.

func (ChildDispatch) Attempt 

func (dispatch ChildDispatch) Attempt() uint32

Attempt returns the one-based semantic child-start attempt.

func (ChildDispatch) ChildID 

func (dispatch ChildDispatch) ChildID() string

ChildID returns the stable child instance identity.

func (ChildDispatch) Definition 

func (dispatch ChildDispatch) Definition() DefinitionReference

Definition returns the exact child behavior identity.

func (ChildDispatch) IdempotencyKey 

func (dispatch ChildDispatch) IdempotencyKey() string

IdempotencyKey returns the stable key for this semantic attempt.

func (ChildDispatch) StepName 

func (dispatch ChildDispatch) StepName() string

StepName returns the parent definition step.

type ChildOutcomeSpec 

type ChildOutcomeSpec struct {
	TransitionID string
	Instance     Instance
	Definition   Definition
	StepName     string
	ChildID      string
	CompletedAt  time.Time
	Result       []byte
	FailureCode  string
}

ChildOutcomeSpec supplies one known terminal child result observed by the parent. FailureCode selects failure; an empty code selects success.

type ChildProgress 

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

ChildProgress is immutable state reconstructed only from persisted events.

func (ChildProgress) Attempt 

func (progress ChildProgress) Attempt() uint32

Attempt returns the latest one-based child-start attempt.

func (ChildProgress) ChildID 

func (progress ChildProgress) ChildID() string

ChildID returns the stable child instance identity.

func (ChildProgress) Code 

func (progress ChildProgress) Code() string

Code returns the stable known-failure code, or empty for other states.

func (ChildProgress) Definition 

func (progress ChildProgress) Definition() DefinitionReference

Definition returns the exact pinned child behavior identity.

func (ChildProgress) DueAt 

func (progress ChildProgress) DueAt() time.Time

DueAt returns the attempt deadline or retry admission time.

func (ChildProgress) IdempotencyKey 

func (progress ChildProgress) IdempotencyKey() string

IdempotencyKey returns the latest persisted child-start key.

func (ChildProgress) Input 

func (progress ChildProgress) Input() []byte

Input returns an owned copy of the persisted child input.

func (ChildProgress) Result 

func (progress ChildProgress) Result() []byte

Result returns an owned copy of the known child terminal result.

func (ChildProgress) Retryable 

func (progress ChildProgress) Retryable() bool

Retryable reports whether a known-absent creation failure permits retry.

func (ChildProgress) Status 

func (progress ChildProgress) Status() ChildProgressStatus

Status returns the replayed durable child state.

func (ChildProgress) StepName 

func (progress ChildProgress) StepName() string

StepName returns the stable parent definition step.

type ChildProgressStatus 

type ChildProgressStatus uint8

ChildProgressStatus identifies replayed durable child-workflow progress. 

const (
	// ChildScheduled has durable dispatch work but no terminal child outcome.
	ChildScheduled ChildProgressStatus = 1
	// ChildSucceeded is a known successful child terminal outcome.
	ChildSucceeded ChildProgressStatus = 2
	// ChildFailed is a known failed child terminal outcome.
	ChildFailed ChildProgressStatus = 3
	// ChildStartRunning has one externally observable creation attempt.
	ChildStartRunning ChildProgressStatus = 4
	// ChildActive means the pinned child is known to exist and is nonterminal.
	ChildActive ChildProgressStatus = 5
	// ChildStartFailedStatus is a known-absent creation failure.
	ChildStartFailedStatus ChildProgressStatus = 6
	// ChildStartUnknownStatus requires reconciliation before redispatch.
	ChildStartUnknownStatus ChildProgressStatus = 7
	// ChildStartRetryWaiting has a persisted next-attempt admission time.
	ChildStartRetryWaiting ChildProgressStatus = 8
)

type ChildScheduleSpec 

type ChildScheduleSpec struct {
	TransitionID  string
	WorkID        string
	ChildID       string
	Instance      Instance
	Definition    Definition
	StepName      string
	ScheduledAt   time.Time
	Deadline      time.Time
	Input         []byte
	TenantID      string
	CorrelationID string
}

ChildScheduleSpec supplies atomic parent history and durable child work.

type ChildStartAttemptOutcomeSpec 

type ChildStartAttemptOutcomeSpec struct {
	TransitionID string
	Instance     Instance
	Definition   Definition
	StepName     string
	ChildID      string
	Attempt      uint32
	OccurredAt   time.Time
	Outcome      ChildStartOutcome
}

ChildStartAttemptOutcomeSpec supplies one explicit creation result.

type ChildStartAttemptSpec 

type ChildStartAttemptSpec struct {
	TransitionID string
	Lease        WorkLease
	Instance     Instance
	Definition   Definition
	StartedAt    time.Time
}

ChildStartAttemptSpec supplies the durable pre-creation boundary.

type ChildStartFunc 

type ChildStartFunc func(context.Context, ChildStartRequest) ChildStartOutcome

ChildStartFunc adapts an explicit function without registration or reflection-driven discovery.

func (ChildStartFunc) Start 

func (start ChildStartFunc) Start(ctx context.Context, request ChildStartRequest) ChildStartOutcome

Start invokes the adapted child starter.

type ChildStartOutcome 

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

ChildStartOutcome is an immutable known or uncertain creation result.

func NewChildStartOutcome 

func NewChildStartOutcome(spec ChildStartOutcomeSpec) (ChildStartOutcome, error)

NewChildStartOutcome validates one explicit creation result.

func (ChildStartOutcome) Code 

func (outcome ChildStartOutcome) Code() string

Code returns a stable known-failure or uncertainty classification.

func (ChildStartOutcome) Kind 

func (outcome ChildStartOutcome) Kind() ChildStartOutcomeKind

Kind returns the explicit creation result.

func (ChildStartOutcome) Retryable 

func (outcome ChildStartOutcome) Retryable() bool

Retryable reports whether a known absence permits policy retry.

type ChildStartOutcomeKind 

type ChildStartOutcomeKind uint8

ChildStartOutcomeKind classifies the observable result of one idempotent child-creation attempt. 

const (
	// ChildStarted means the pinned child instance is known to exist.
	ChildStarted ChildStartOutcomeKind = 1
	// ChildStartFailed means creation is known not to have occurred.
	ChildStartFailed ChildStartOutcomeKind = 2
	// ChildStartUnknown means creation may have occurred and must not be
	// repeated without reconciliation.
	ChildStartUnknown ChildStartOutcomeKind = 3
)

type ChildStartOutcomeSpec 

type ChildStartOutcomeSpec struct {
	Kind      ChildStartOutcomeKind
	Code      string
	Retryable bool
}

ChildStartOutcomeSpec supplies one explicit bounded creation result.

type ChildStartRequest 

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

ChildStartRequest is immutable attempt metadata supplied to a caller-owned idempotent child creator.

func NewChildStartRequest 

func NewChildStartRequest(spec ChildStartRequestSpec) (ChildStartRequest, error)

NewChildStartRequest validates and owns one bounded start request.

func (ChildStartRequest) Attempt 

func (request ChildStartRequest) Attempt() uint32

Attempt returns the one-based semantic creation attempt.

func (ChildStartRequest) ChildDefinition 

func (request ChildStartRequest) ChildDefinition() DefinitionReference

ChildDefinition returns the exact child behavior identity.

func (ChildStartRequest) ChildID 

func (request ChildStartRequest) ChildID() string

ChildID returns the stable child identity and natural deduplication key.

func (ChildStartRequest) CorrelationID 

func (request ChildStartRequest) CorrelationID() string

CorrelationID returns caller-supplied trace and message correlation metadata.

func (ChildStartRequest) Deadline 

func (request ChildStartRequest) Deadline() time.Time

Deadline returns the persisted attempt deadline.

func (ChildStartRequest) IdempotencyKey 

func (request ChildStartRequest) IdempotencyKey() string

IdempotencyKey returns the stable key for this semantic attempt.

func (ChildStartRequest) Input 

func (request ChildStartRequest) Input() []byte

Input returns an owned copy of the persisted child input.

func (ChildStartRequest) MaxAttempts 

func (request ChildStartRequest) MaxAttempts() uint32

MaxAttempts returns the immutable parent policy bound.

func (ChildStartRequest) ParentDefinition 

func (request ChildStartRequest) ParentDefinition() DefinitionReference

ParentDefinition returns the exact parent behavior identity.

func (ChildStartRequest) ParentInstanceID 

func (request ChildStartRequest) ParentInstanceID() string

ParentInstanceID returns the durable parent identity.

func (ChildStartRequest) StartedAt 

func (request ChildStartRequest) StartedAt() time.Time

StartedAt returns the persisted attempt start time.

func (ChildStartRequest) StepName 

func (request ChildStartRequest) StepName() string

StepName returns the stable parent child-step name.

func (ChildStartRequest) TenantID 

func (request ChildStartRequest) TenantID() string

TenantID returns caller-supplied routing metadata. It must not be used as an unbounded metric label.

type ChildStartRequestSpec 

type ChildStartRequestSpec struct {
	ParentInstanceID string
	ParentDefinition DefinitionReference
	StepName         string
	ChildID          string
	ChildDefinition  DefinitionReference
	Attempt          uint32
	MaxAttempts      uint32
	IdempotencyKey   string
	StartedAt        time.Time
	Deadline         time.Time
	Input            []byte
	InputLimit       uint32
	TenantID         string
	CorrelationID    string
}

ChildStartRequestSpec supplies one version-pinned bounded creation attempt.

type ChildStartRetrySpec 

type ChildStartRetrySpec struct {
	TransitionID  string
	WorkID        string
	Instance      Instance
	Definition    Definition
	StepName      string
	ScheduledAt   time.Time
	Deadline      time.Time
	TenantID      string
	CorrelationID string
}

ChildStartRetrySpec supplies one deterministic retry decision and due work.

type ChildStarter 

type ChildStarter interface {
	Start(context.Context, ChildStartRequest) ChildStartOutcome
}

ChildStarter creates or observes one pinned child using the supplied stable identity. Implementations must be idempotent and explicitly report unknown outcomes.

type ChildWorkProcessor 

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

ChildWorkProcessor starts pinned child instances without claiming exactly once creation. It persists the start boundary before invoking the adapter.

func NewChildWorkProcessor 

func NewChildWorkProcessor(config ChildWorkProcessorConfig) (*ChildWorkProcessor, error)

NewChildWorkProcessor validates one explicit bounded processor.

func (*ChildWorkProcessor) Process 

func (processor *ChildWorkProcessor) Process(ctx context.Context, lease WorkLease) (WorkDecision, error)

Process persists every child-creation boundary. Poison work is dead-lettered and an in-flight redelivery becomes an explicit unknown outcome.

type ChildWorkProcessorConfig 

type ChildWorkProcessorConfig struct {
	Store            ActivityExecutionStore
	Definitions      *Registry
	Starter          ChildStarter
	Clock            Clock
	PageSize         uint32
	MaxHistoryEvents uint32
}

ChildWorkProcessorConfig supplies explicit bounded child-start dependencies.

type Clock 

type Clock interface {
	Now() time.Time
	NewTimer(time.Duration) ClockTimer
}

Clock supplies deterministic persisted decision and admission time.

type ClockTimer 

type ClockTimer interface {
	C() <-chan time.Time
	Stop() bool
}

ClockTimer is one caller-owned deterministic timer.

type CompensationAttemptOutcomeSpec 

type CompensationAttemptOutcomeSpec struct {
	TransitionID string
	Instance     Instance
	Definition   Definition
	StepName     string
	Attempt      uint32
	OccurredAt   time.Time
	Outcome      ActivityOutcome
}

CompensationAttemptOutcomeSpec supplies one known or unknown persisted compensation result.

type CompensationAttemptStartSpec 

type CompensationAttemptStartSpec struct {
	TransitionID string
	Lease        WorkLease
	Instance     Instance
	Definition   Definition
	StartedAt    time.Time
}

CompensationAttemptStartSpec supplies the persisted decision required before executing one leased compensating side effect.

type CompensationDispatch 

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

CompensationDispatch is immutable durable metadata for one semantic compensation attempt. Lease redelivery retains the same idempotency key.

func DecodeCompensationDispatch 

func DecodeCompensationDispatch(payload []byte) (CompensationDispatch, error)

DecodeCompensationDispatch validates bounded durable work metadata.

func (CompensationDispatch) Attempt 

func (dispatch CompensationDispatch) Attempt() uint32

Attempt returns the one-based semantic compensation attempt.

func (CompensationDispatch) IdempotencyKey 

func (dispatch CompensationDispatch) IdempotencyKey() string

IdempotencyKey returns the stable external side-effect identity.

func (CompensationDispatch) StepName 

func (dispatch CompensationDispatch) StepName() string

StepName returns the compensated activity step.

type CompensationProgress 

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

CompensationProgress is immutable state reconstructed from persisted decisions only.

func (CompensationProgress) Attempt 

func (progress CompensationProgress) Attempt() uint32

Attempt returns the latest one-based attempt.

func (CompensationProgress) Code 

func (progress CompensationProgress) Code() string

Code returns the latest failure, unknown-outcome, or manual resolution code.

func (CompensationProgress) DueAt 

func (progress CompensationProgress) DueAt() time.Time

DueAt returns the attempt deadline or retry admission time.

func (CompensationProgress) IdempotencyKey 

func (progress CompensationProgress) IdempotencyKey() string

IdempotencyKey returns the latest externally observable attempt identity.

func (CompensationProgress) Input 

func (progress CompensationProgress) Input() []byte

Input returns an owned copy of scheduled compensation input.

func (CompensationProgress) Result 

func (progress CompensationProgress) Result() []byte

Result returns an owned copy of result, failure detail, or manual evidence.

func (CompensationProgress) Retryable 

func (progress CompensationProgress) Retryable() bool

Retryable reports the persisted known-failure retry classification.

func (CompensationProgress) ScheduledSequence 

func (progress CompensationProgress) ScheduledSequence() uint64

ScheduledSequence returns persisted compensation ordering.

func (CompensationProgress) Status 

func (progress CompensationProgress) Status() CompensationProgressStatus

Status returns the durable compensation state.

func (CompensationProgress) StepName 

func (progress CompensationProgress) StepName() string

StepName returns the compensated activity step.

type CompensationProgressStatus 

type CompensationProgressStatus uint8

CompensationProgressStatus identifies durable compensating activity state. 

const (
	// CompensationReady is scheduled and eligible for attempt admission.
	CompensationReady CompensationProgressStatus = 1
	// CompensationRunning has one externally observable in-flight attempt.
	CompensationRunning CompensationProgressStatus = 2
	// CompensationSucceeded is a known successful compensating side effect.
	CompensationSucceeded CompensationProgressStatus = 3
	// CompensationFailed is a known failed attempt awaiting policy action.
	CompensationFailed CompensationProgressStatus = 4
	// CompensationUnknown requires reconciliation or manual resolution.
	CompensationUnknown CompensationProgressStatus = 5
	// CompensationRetryWaiting has a persisted next-attempt admission time.
	CompensationRetryWaiting CompensationProgressStatus = 6
	// CompensationManuallyResolved is an explicit operator disposition and is
	// never equivalent to successful rollback.
	CompensationManuallyResolved CompensationProgressStatus = 7
)

type CompensationRetrySpec 

type CompensationRetrySpec struct {
	TransitionID   string
	WorkID         string
	Instance       Instance
	Definition     Definition
	StepName       string
	IdempotencyKey string
	ScheduledAt    time.Time
	Deadline       time.Time
	TenantID       string
	CorrelationID  string
}

CompensationRetrySpec supplies one persisted retry decision and the next semantic compensation attempt.

type CompensationScheduleSpec 

type CompensationScheduleSpec struct {
	TransitionID   string
	WorkID         string
	Instance       Instance
	Definition     Definition
	StepName       string
	Attempt        uint32
	IdempotencyKey string
	ScheduledAt    time.Time
	Deadline       time.Time
	Input          []byte
	TenantID       string
	CorrelationID  string
}

CompensationScheduleSpec supplies atomic history and durable dispatch for the first attempt of one explicit compensation.

type CompensationSpec 

type CompensationSpec struct {
	Target      string
	Timeout     time.Duration
	ResultLimit uint32
	Retry       RetryPolicy
}

CompensationSpec defines an independently retryable compensating activity. A failed compensation remains a failed compensation; it is never translated into a successful rollback.

type CompensationWorkProcessor 

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

CompensationWorkProcessor executes leased compensating side effects. It never translates a failed or unknown compensation into successful rollback.

func NewCompensationWorkProcessor 

func NewCompensationWorkProcessor(config CompensationWorkProcessorConfig) (*CompensationWorkProcessor, error)

NewCompensationWorkProcessor validates one bounded explicit processor.

func (*CompensationWorkProcessor) Process 

func (processor *CompensationWorkProcessor) Process(ctx context.Context, lease WorkLease) (WorkDecision, error)

Process persists compensation attempt start before handler invocation and persists an explicit result before returning WorkComplete.

type CompensationWorkProcessorConfig 

type CompensationWorkProcessorConfig struct {
	Store            ActivityExecutionStore
	Definitions      *Registry
	Compensations    *ActivityRegistry
	Clock            Clock
	PageSize         uint32
	MaxHistoryEvents uint32
}

CompensationWorkProcessorConfig supplies explicit bounded compensation execution dependencies. Compensations is an explicit activity registry resolved from each immutable CompensationSpec target.

type DeadLetterCursor 

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

DeadLetterCursor is an immutable exclusive failure-time and identity cursor.

func NewDeadLetterCursor 

func NewDeadLetterCursor(spec DeadLetterCursorSpec) (DeadLetterCursor, error)

NewDeadLetterCursor validates one non-zero external cursor. The zero cursor is represented by DeadLetterCursor{} for an initial query.

func (DeadLetterCursor) FailedAt 

func (cursor DeadLetterCursor) FailedAt() time.Time

FailedAt returns the cursor failure time.

func (DeadLetterCursor) WorkID 

func (cursor DeadLetterCursor) WorkID() string

WorkID returns the cursor work identity.

type DeadLetterCursorSpec 

type DeadLetterCursorSpec struct {
	FailedAt time.Time
	WorkID   string
}

DeadLetterCursorSpec supplies a decoded stable failure-time cursor.

type DeadLetterPage 

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

DeadLetterPage is one immutable stable unresolved-letter page.

func NewDeadLetterPage 

func NewDeadLetterPage(
	query DeadLetterQuery,
	items []DeadLetterRecord,
	hasMore bool,
) (DeadLetterPage, error)

NewDeadLetterPage validates adapter output against its originating query.

func (DeadLetterPage) HasMore 

func (page DeadLetterPage) HasMore() bool

HasMore reports whether the adapter observed another unresolved letter.

func (DeadLetterPage) Items 

func (page DeadLetterPage) Items() []DeadLetterRecord

Items returns an owned ordered page.

func (DeadLetterPage) NextCursor 

func (page DeadLetterPage) NextCursor() DeadLetterCursor

NextCursor returns the exclusive cursor for the next page.

type DeadLetterQuery 

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

DeadLetterQuery is one immutable stable unresolved-letter page request.

func NewDeadLetterQuery 

func NewDeadLetterQuery(spec DeadLetterQuerySpec) (DeadLetterQuery, error)

NewDeadLetterQuery validates one bounded query.

func (DeadLetterQuery) After 

func (query DeadLetterQuery) After() DeadLetterCursor

After returns the exclusive stable cursor.

func (DeadLetterQuery) Limit 

func (query DeadLetterQuery) Limit() uint32

Limit returns the maximum page size.

func (DeadLetterQuery) Valid 

func (query DeadLetterQuery) Valid() bool

Valid reports whether the query is bounded and coherent.

type DeadLetterQuerySpec 

type DeadLetterQuerySpec struct {
	After DeadLetterCursor
	Limit uint32
}

DeadLetterQuerySpec supplies one bounded stable unresolved-letter request.

type DeadLetterRecord 

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

DeadLetterRecord is immutable unresolved poison-work state.

func NewDeadLetterRecord 

func NewDeadLetterRecord(spec DeadLetterRecordSpec) (DeadLetterRecord, error)

NewDeadLetterRecord validates and owns one adapter-decoded dead letter.

func (DeadLetterRecord) Attempt 

func (record DeadLetterRecord) Attempt() uint32

Attempt returns the durable claim attempt that produced the dead letter.

func (DeadLetterRecord) FailedAt 

func (record DeadLetterRecord) FailedAt() time.Time

FailedAt returns the persisted dead-letter decision time.

func (DeadLetterRecord) FailureCode 

func (record DeadLetterRecord) FailureCode() string

FailureCode returns the bounded poison-work classification.

func (DeadLetterRecord) Token 

func (record DeadLetterRecord) Token() uint64

Token returns the fence required by an operator resolution.

func (DeadLetterRecord) Work 

func (record DeadLetterRecord) Work() PendingWork

Work returns an owned copy of the unresolved durable work item.

type DeadLetterRecordSpec 

type DeadLetterRecordSpec struct {
	Work        PendingWork
	Attempt     uint32
	Token       uint64
	FailureCode string
	FailedAt    time.Time
}

DeadLetterRecordSpec supplies one adapter-decoded unresolved dead letter.

type DeadLetterResolution 

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

DeadLetterResolution is one immutable idempotent fenced operator command. A store commit error uses StoreCommitOutcomeOf to expose unknown outcomes.

func NewDeadLetterResolution 

func NewDeadLetterResolution(spec DeadLetterResolutionSpec) (DeadLetterResolution, error)

NewDeadLetterResolution validates one bounded operator command and computes its stable complete-content idempotency fingerprint.

func (DeadLetterResolution) Action 

func (resolution DeadLetterResolution) Action() DeadLetterResolutionAction

Action returns the explicit retry or discard disposition.

func (DeadLetterResolution) Actor 

func (resolution DeadLetterResolution) Actor() string

Actor returns the already-authorized caller-supplied principal identity.

func (DeadLetterResolution) CommandID 

func (resolution DeadLetterResolution) CommandID() string

CommandID returns the stable idempotency and audit identity.

func (DeadLetterResolution) Deadline 

func (resolution DeadLetterResolution) Deadline() time.Time

Deadline returns the replacement work deadline, or zero for discard.

func (DeadLetterResolution) Fingerprint 

func (resolution DeadLetterResolution) Fingerprint() string

Fingerprint returns the complete stable command digest used for exact replay.

func (DeadLetterResolution) OccurredAt 

func (resolution DeadLetterResolution) OccurredAt() time.Time

OccurredAt returns the deterministic operator decision time.

func (DeadLetterResolution) Reason 

func (resolution DeadLetterResolution) Reason() string

Reason returns the bounded caller-supplied audit reason.

func (DeadLetterResolution) RetryAt 

func (resolution DeadLetterResolution) RetryAt() time.Time

RetryAt returns retry admission time, or zero for discard.

func (DeadLetterResolution) Token 

func (resolution DeadLetterResolution) Token() uint64

Token returns the expected dead-letter fencing token.

func (DeadLetterResolution) Valid 

func (resolution DeadLetterResolution) Valid() bool

Valid reports whether the command remains internally coherent.

func (DeadLetterResolution) WorkID 

func (resolution DeadLetterResolution) WorkID() string

WorkID returns the dead-lettered durable work identity.

type DeadLetterResolutionAction 

type DeadLetterResolutionAction uint8

DeadLetterResolutionAction selects one explicit operator disposition. 

const (
	// DeadLetterRetry returns the exact fenced work item to due admission with
	// an explicit new deadline.
	DeadLetterRetry DeadLetterResolutionAction = 1
	// DeadLetterDiscard records an audited decision that the fenced work item
	// must remain unavailable.
	DeadLetterDiscard DeadLetterResolutionAction = 2
)

func (DeadLetterResolutionAction) String 

func (action DeadLetterResolutionAction) String() string

String returns the stable persisted action name.

type DeadLetterResolutionSpec 

type DeadLetterResolutionSpec struct {
	CommandID  string
	WorkID     string
	Token      uint64
	Action     DeadLetterResolutionAction
	Actor      string
	Reason     string
	OccurredAt time.Time
	RetryAt    time.Time
	Deadline   time.Time
}

DeadLetterResolutionSpec supplies one caller-authorized audited command. Callers remain responsible for authenticating and authorizing Actor.

type DeadLetterStore 

type DeadLetterStore interface {
	ListDeadLetters(context.Context, DeadLetterQuery) (DeadLetterPage, error)
	ResolveDeadLetter(context.Context, DeadLetterResolution) error
}

DeadLetterStore exposes stable unresolved-letter inspection and audited, idempotent fenced resolution. Callers must authorize resolution actors.

type Definition 

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

Definition is one validated immutable workflow behavior version. Its zero value is invalid.

func NewDefinition 

func NewDefinition(spec DefinitionSpec) (Definition, error)

NewDefinition validates, owns, and fingerprints one immutable definition.

func (Definition) Deprecated 

func (definition Definition) Deprecated() bool

Deprecated reports whether new instance creation should be refused by a caller. Existing pinned instances remain resolvable.

func (Definition) Fingerprint 

func (definition Definition) Fingerprint() string

Fingerprint returns the deterministic behavior digest used to detect silent reinterpretation of an immutable name and version.

func (Definition) Mode 

func (definition Definition) Mode() ExecutionMode

Mode returns the definition execution model.

func (Definition) Name 

func (definition Definition) Name() string

Name returns the stable definition name.

func (Definition) Reference 

func (definition Definition) Reference() DefinitionReference

Reference returns the exact immutable behavior identity that instances must persist with their history.

func (Definition) Steps 

func (definition Definition) Steps() []StepSpec

Steps returns a deep copy of the ordered definition steps.

func (Definition) Version 

func (definition Definition) Version() string

Version returns the immutable behavior version.

type DefinitionReference 

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

DefinitionReference is a comparable exact immutable behavior identity. Its zero value is invalid.

func NewDefinitionReference 

func NewDefinitionReference(name, version, fingerprint string) (DefinitionReference, error)

NewDefinitionReference validates one persisted definition identity.

func (DefinitionReference) Fingerprint 

func (reference DefinitionReference) Fingerprint() string

Fingerprint returns the exact behavior digest.

func (DefinitionReference) Name 

func (reference DefinitionReference) Name() string

Name returns the stable definition name.

func (DefinitionReference) Version 

func (reference DefinitionReference) Version() string

Version returns the immutable definition version.

type DefinitionSpec 

type DefinitionSpec struct {
	Name       string
	Version    string
	Mode       ExecutionMode
	Deprecated bool
	Steps      []StepSpec
}

DefinitionSpec supplies one stable immutable workflow definition version.

type EventKind 

type EventKind uint8

EventKind identifies one persisted instance-lifecycle decision. 

const (
	// EventInstanceStarted creates one version-pinned instance.
	EventInstanceStarted EventKind = 1
	// EventInstancePaused stops new workflow progression.
	EventInstancePaused EventKind = 2
	// EventInstanceResumed permits workflow progression after a pause.
	EventInstanceResumed EventKind = 3
	// EventCancellationRequested begins durable cancellation.
	EventCancellationRequested EventKind = 4
	// EventInstanceCancelled records completed cancellation.
	EventInstanceCancelled EventKind = 5
	// EventInstanceCompleted records a successful terminal outcome.
	EventInstanceCompleted EventKind = 6
	// EventInstanceFailed records a known failed terminal outcome.
	EventInstanceFailed EventKind = 7
	// EventInstanceTerminated records forced operator termination.
	EventInstanceTerminated EventKind = 8
	// EventDefinitionMigrated records already-produced migrated persisted state.
	EventDefinitionMigrated EventKind = 9
	// EventContinuedAsNew closes history in favor of an explicit successor.
	EventContinuedAsNew EventKind = 10
	// EventActivityScheduled records durable activity input before dispatch.
	EventActivityScheduled EventKind = 11
	// EventActivityAttemptStarted records one externally observable attempt.
	EventActivityAttemptStarted EventKind = 12
	// EventActivityAttemptSucceeded records a known successful attempt result.
	EventActivityAttemptSucceeded EventKind = 13
	// EventActivityAttemptFailed records a known failed attempt result.
	EventActivityAttemptFailed EventKind = 14
	// EventActivityAttemptUnknown records an attempt that may have committed.
	EventActivityAttemptUnknown EventKind = 15
	// EventActivityRetryScheduled records the deterministic next-attempt time.
	EventActivityRetryScheduled EventKind = 16
	// EventTimerScheduled records a durable timer deadline before admission.
	EventTimerScheduled EventKind = 17
	// EventTimerFired records the persisted observation that a timer became due.
	EventTimerFired EventKind = 18
	// EventSignalReceived records one durably accepted deduplicated signal.
	EventSignalReceived EventKind = 19
	// EventCompensationScheduled records an explicit compensating activity.
	EventCompensationScheduled EventKind = 20
	// EventCompensationAttemptStarted records one compensating side-effect attempt.
	EventCompensationAttemptStarted EventKind = 21
	// EventCompensationAttemptSucceeded records known successful compensation.
	EventCompensationAttemptSucceeded EventKind = 22
	// EventCompensationAttemptFailed records known failed compensation.
	EventCompensationAttemptFailed EventKind = 23
	// EventCompensationAttemptUnknown records an uncertain compensation outcome.
	EventCompensationAttemptUnknown EventKind = 24
	// EventCompensationRetryScheduled records deterministic retry admission.
	EventCompensationRetryScheduled EventKind = 25
	// EventCompensationManuallyResolved records explicit operator resolution.
	EventCompensationManuallyResolved EventKind = 26
	// EventOperatorCommandRecorded audits one caller-authorized intervention.
	EventOperatorCommandRecorded EventKind = 27
	// EventRaceWon records the persisted winner of one explicit race.
	EventRaceWon EventKind = 28
	// EventChildScheduled records a version-pinned child before dispatch.
	EventChildScheduled EventKind = 29
	// EventChildCompleted records a known successful child outcome.
	EventChildCompleted EventKind = 30
	// EventChildFailed records a known failed child outcome.
	EventChildFailed EventKind = 31
	// EventChildStartAttempted records persistence before external creation.
	EventChildStartAttempted EventKind = 32
	// EventChildStarted records that the pinned child is known to exist.
	EventChildStarted EventKind = 33
	// EventChildStartFailed records a known-absent child creation failure.
	EventChildStartFailed EventKind = 34
	// EventChildStartUnknown records creation that may have succeeded.
	EventChildStartUnknown EventKind = 35
	// EventChildStartRetryScheduled records deterministic retry admission.
	EventChildStartRetryScheduled EventKind = 36
)

type ExecutionMode 

type ExecutionMode uint8

ExecutionMode distinguishes central orchestration from explicit external choreography. Choreography does not imply or install a global event bus. 

const (
	// Orchestration advances through decisions made by one workflow definition.
	Orchestration ExecutionMode = 1
	// Choreography advances through explicitly supplied durable external events.
	Choreography ExecutionMode = 2
)

type HistoryEvent 

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

HistoryEvent is one validated immutable persisted decision.

func NewHistoryEvent 

func NewHistoryEvent(spec HistoryEventSpec) (HistoryEvent, error)

NewHistoryEvent validates and owns one durable history record.

func (HistoryEvent) Attempt 

func (event HistoryEvent) Attempt() uint32

Attempt returns the one-based activity attempt selected by an attempt event.

func (HistoryEvent) Code 

func (event HistoryEvent) Code() string

Code returns a stable known-failure or unknown-outcome code.

func (HistoryEvent) Data 

func (event HistoryEvent) Data() []byte

Data returns an owned copy of persisted event data.

func (HistoryEvent) Definition 

func (event HistoryEvent) Definition() DefinitionReference

Definition returns the target definition for start, migration, or continue-as-new decisions.

func (HistoryEvent) DueAt 

func (event HistoryEvent) DueAt() time.Time

DueAt returns a persisted attempt, retry, or timer deadline.

func (HistoryEvent) IdempotencyKey 

func (event HistoryEvent) IdempotencyKey() string

IdempotencyKey returns the persisted external-attempt or signal identity.

func (HistoryEvent) InstanceID 

func (event HistoryEvent) InstanceID() string

InstanceID returns the durable instance identity.

func (HistoryEvent) Kind 

func (event HistoryEvent) Kind() EventKind

Kind returns the persisted transition kind.

func (HistoryEvent) OccurredAt 

func (event HistoryEvent) OccurredAt() time.Time

OccurredAt returns canonical UTC persisted decision time.

func (HistoryEvent) Retryable 

func (event HistoryEvent) Retryable() bool

Retryable reports the persisted known-failure retry classification.

func (HistoryEvent) Sequence 

func (event HistoryEvent) Sequence() uint64

Sequence returns the contiguous instance-history position.

func (HistoryEvent) StepName 

func (event HistoryEvent) StepName() string

StepName returns the definition step selected by a step event.

func (HistoryEvent) SuccessorID 

func (event HistoryEvent) SuccessorID() string

SuccessorID returns the explicit continue-as-new successor identity.

type HistoryEventSpec 

type HistoryEventSpec struct {
	Sequence       uint64
	InstanceID     string
	Kind           EventKind
	OccurredAt     time.Time
	Definition     DefinitionReference
	SuccessorID    string
	StepName       string
	Attempt        uint32
	IdempotencyKey string
	DueAt          time.Time
	Code           string
	Retryable      bool
	Data           []byte
}

HistoryEventSpec supplies one immutable durable history record.

type HistoryExportSink 

type HistoryExportSink func(context.Context, []HistoryEvent) error

HistoryExportSink consumes one owned stable forward page. Returning an error stops export without acknowledging or mutating durable state.

type HistoryExportSpec 

type HistoryExportSpec struct {
	InstanceID string
	PageSize   uint32
	MaxEvents  uint32
}

HistoryExportSpec supplies one bounded streaming history-export request.

type HistoryPage 

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

HistoryPage is one immutable stable forward history page.

func NewHistoryPage 

func NewHistoryPage(query HistoryQuery, events []HistoryEvent, hasMore bool) (HistoryPage, error)

NewHistoryPage validates adapter output against the originating query.

func (HistoryPage) Events 

func (page HistoryPage) Events() []HistoryEvent

Events returns an owned copy of the ordered page.

func (HistoryPage) HasMore 

func (page HistoryPage) HasMore() bool

HasMore reports whether the adapter observed a later event.

func (HistoryPage) NextAfterSequence 

func (page HistoryPage) NextAfterSequence() uint64

NextAfterSequence returns the cursor for the next request.

type HistoryQuery 

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

HistoryQuery is an immutable forward page request. AfterSequence is an exclusive stable cursor; zero begins at the first event.

func NewHistoryQuery 

func NewHistoryQuery(spec HistoryQuerySpec) (HistoryQuery, error)

NewHistoryQuery validates one bounded stable history page request.

func (HistoryQuery) AfterSequence 

func (query HistoryQuery) AfterSequence() uint64

AfterSequence returns the exclusive stable history cursor.

func (HistoryQuery) InstanceID 

func (query HistoryQuery) InstanceID() string

InstanceID returns the selected workflow instance.

func (HistoryQuery) Limit 

func (query HistoryQuery) Limit() uint32

Limit returns the maximum number of events in the page.

func (HistoryQuery) Valid 

func (query HistoryQuery) Valid() bool

Valid reports whether the query is bounded and internally coherent.

type HistoryQuerySpec 

type HistoryQuerySpec struct {
	InstanceID    string
	AfterSequence uint64
	Limit         uint32
}

HistoryQuerySpec supplies one stable bounded instance-history page request.

type HistoryReader 

type HistoryReader interface {
	History(context.Context, HistoryQuery) (HistoryPage, error)
}

HistoryReader is the narrow read contract consumed by inspection and export.

type Instance 

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

Instance is an immutable replay result derived only from persisted history.

func InspectInstance 

func InspectInstance(
	ctx context.Context,
	reader HistoryReader,
	registry *Registry,
	spec InstanceInspectionSpec,
) (Instance, error)

InspectInstance reconstructs one instance from bounded stable history pages. Replay decisions depend only on persisted history and the pinned registry.

func Replay 

func Replay(registry *Registry, events []HistoryEvent) (Instance, error)

Replay deterministically reconstructs one instance from validated persisted decisions. It resolves definitions and migration edges but never executes workflow or migration code.

func (Instance) Activities 

func (instance Instance) Activities() []ActivityProgress

Activities returns replayed activity progress in stable step-name order.

func (Instance) Activity 

func (instance Instance) Activity(stepName string) (ActivityProgress, bool)

Activity returns immutable replayed progress for one definition activity.

func (Instance) Child 

func (instance Instance) Child(stepName string) (ChildProgress, bool)

Child returns immutable replayed progress for one child-workflow step.

func (Instance) Children 

func (instance Instance) Children() []ChildProgress

Children returns child progress in stable step-name order.

func (Instance) Compensation 

func (instance Instance) Compensation(stepName string) (CompensationProgress, bool)

Compensation returns immutable replayed progress for one activity's explicit compensating action.

func (Instance) Compensations 

func (instance Instance) Compensations() []CompensationProgress

Compensations returns replayed compensation progress in persisted schedule order.

func (Instance) Definition 

func (instance Instance) Definition() DefinitionReference

Definition returns the currently pinned exact behavior identity.

func (Instance) ID 

func (instance Instance) ID() string

ID returns the durable instance identity.

func (Instance) Input 

func (instance Instance) Input() []byte

Input returns an owned copy of current persisted workflow state.

func (Instance) OperatorActions 

func (instance Instance) OperatorActions() []OperatorActionRecord

OperatorActions returns an owned ordered audit trail of interventions.

func (Instance) Race 

func (instance Instance) Race(stepName string) (RaceProgress, bool)

Race returns the durably selected winner for one definition race.

func (Instance) Races 

func (instance Instance) Races() []RaceProgress

Races returns persisted race winners in stable control-step order.

func (Instance) Result 

func (instance Instance) Result() []byte

Result returns an owned copy of terminal result or failure data.

func (Instance) Sequence 

func (instance Instance) Sequence() uint64

Sequence returns the final contiguous history position.

func (Instance) Signal 

func (instance Instance) Signal(stepName string) (SignalProgress, bool)

Signal returns immutable replayed progress for one definition signal wait.

func (Instance) Signals 

func (instance Instance) Signals() []SignalProgress

Signals returns replayed signal progress in stable step-name order.

func (Instance) SnapshotDigest 

func (instance Instance) SnapshotDigest() string

SnapshotDigest returns a deterministic digest of reconstructed persisted state for diagnostics and replay comparison.

func (Instance) StartedAt 

func (instance Instance) StartedAt() time.Time

StartedAt returns canonical persisted creation time.

func (Instance) Status 

func (instance Instance) Status() InstanceStatus

Status returns the reconstructed lifecycle state.

func (Instance) SuccessorID 

func (instance Instance) SuccessorID() string

SuccessorID returns the explicit continue-as-new successor, when present.

func (Instance) Timer 

func (instance Instance) Timer(stepName string) (TimerProgress, bool)

Timer returns immutable replayed progress for one definition timer.

func (Instance) Timers 

func (instance Instance) Timers() []TimerProgress

Timers returns replayed timer progress in stable step-name order.

func (Instance) UpdatedAt 

func (instance Instance) UpdatedAt() time.Time

UpdatedAt returns canonical time of the last persisted decision.

type InstanceInspectionSpec 

type InstanceInspectionSpec struct {
	InstanceID string
	PageSize   uint32
	MaxEvents  uint32
}

InstanceInspectionSpec supplies one bounded deterministic replay request.

type InstanceListCursor 

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

InstanceListCursor is an immutable stable creation-time and identity cursor.

func NewInstanceListCursor 

func NewInstanceListCursor(spec InstanceListCursorSpec) (InstanceListCursor, error)

NewInstanceListCursor validates one non-zero external pagination cursor. The zero cursor is represented by InstanceListCursor{} for an initial query.

func (InstanceListCursor) CreatedAt 

func (cursor InstanceListCursor) CreatedAt() time.Time

CreatedAt returns the immutable creation-time cursor component.

func (InstanceListCursor) InstanceID 

func (cursor InstanceListCursor) InstanceID() string

InstanceID returns the immutable identity cursor component.

type InstanceListCursorSpec 

type InstanceListCursorSpec struct {
	CreatedAt  time.Time
	InstanceID string
}

InstanceListCursorSpec supplies a decoded external pagination cursor.

type InstanceListPage 

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

InstanceListPage is one immutable stable page and continuation cursor.

func NewInstanceListPage 

func NewInstanceListPage(
	query InstanceListQuery,
	items []InstanceRecord,
	hasMore bool,
) (InstanceListPage, error)

NewInstanceListPage validates adapter output against its originating query.

func (InstanceListPage) HasMore 

func (page InstanceListPage) HasMore() bool

HasMore reports whether the adapter observed another matching instance.

func (InstanceListPage) Items 

func (page InstanceListPage) Items() []InstanceRecord

Items returns an owned ordered page.

func (InstanceListPage) NextCursor 

func (page InstanceListPage) NextCursor() InstanceListCursor

NextCursor returns the exclusive cursor for the next page.

type InstanceListQuery 

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

InstanceListQuery is an immutable stable list request.

func NewInstanceListQuery 

func NewInstanceListQuery(spec InstanceListQuerySpec) (InstanceListQuery, error)

NewInstanceListQuery validates one stable bounded instance request.

func (InstanceListQuery) After 

func (query InstanceListQuery) After() InstanceListCursor

After returns the exclusive stable cursor.

func (InstanceListQuery) Limit 

func (query InstanceListQuery) Limit() uint32

Limit returns the maximum number of records in one page.

func (InstanceListQuery) Selection 

func (query InstanceListQuery) Selection() InstanceListSelection

Selection returns the archive selection policy.

func (InstanceListQuery) Valid 

func (query InstanceListQuery) Valid() bool

Valid reports whether the query is bounded and internally coherent.

type InstanceListQuerySpec 

type InstanceListQuerySpec struct {
	Selection InstanceListSelection
	After     InstanceListCursor
	Limit     uint32
}

InstanceListQuerySpec supplies one bounded stable list request.

type InstanceListSelection 

type InstanceListSelection uint8

InstanceListSelection selects active, archived, or all durable instances. 

const (
	// ListActiveInstances excludes archived instances.
	ListActiveInstances InstanceListSelection = 1
	// ListArchivedInstances includes only archived instances.
	ListArchivedInstances InstanceListSelection = 2
	// ListAllInstances includes active and archived instances.
	ListAllInstances InstanceListSelection = 3
)

type InstanceRecord 

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

InstanceRecord is immutable durable instance metadata for list operations.

func NewInstanceRecord 

func NewInstanceRecord(spec InstanceRecordSpec) (InstanceRecord, error)

NewInstanceRecord validates one adapter-decoded durable instance record.

func (InstanceRecord) ArchivedAt 

func (record InstanceRecord) ArchivedAt() time.Time

ArchivedAt returns the archive time, or zero while active.

func (InstanceRecord) CreatedAt 

func (record InstanceRecord) CreatedAt() time.Time

CreatedAt returns the immutable instance creation time.

func (InstanceRecord) Definition 

func (record InstanceRecord) Definition() DefinitionReference

Definition returns the exact pinned definition identity.

func (InstanceRecord) InstanceID 

func (record InstanceRecord) InstanceID() string

InstanceID returns the durable workflow identity.

func (InstanceRecord) Sequence 

func (record InstanceRecord) Sequence() uint64

Sequence returns the latest committed history sequence.

func (InstanceRecord) UpdatedAt 

func (record InstanceRecord) UpdatedAt() time.Time

UpdatedAt returns the latest committed transition time.

type InstanceRecordSpec 

type InstanceRecordSpec struct {
	InstanceID string
	Definition DefinitionReference
	Sequence   uint64
	CreatedAt  time.Time
	UpdatedAt  time.Time
	ArchivedAt time.Time
}

InstanceRecordSpec supplies one validated durable adapter record.

type InstanceStatus 

type InstanceStatus uint8

InstanceStatus identifies one durable instance lifecycle state. 

const (
	// StatusRunning permits normal workflow progression.
	StatusRunning InstanceStatus = 1
	// StatusPaused prevents normal progression until explicitly resumed.
	StatusPaused InstanceStatus = 2
	// StatusCancelling records a durable cancellation request in progress.
	StatusCancelling InstanceStatus = 3
	// StatusCompleted is a successful terminal outcome.
	StatusCompleted InstanceStatus = 4
	// StatusFailed is a known failed terminal outcome.
	StatusFailed InstanceStatus = 5
	// StatusCancelled is a completed cancellation terminal outcome.
	StatusCancelled InstanceStatus = 6
	// StatusTerminated is a forced terminal outcome.
	StatusTerminated InstanceStatus = 7
	// StatusContinuedAsNew is a terminal outcome with a named successor.
	StatusContinuedAsNew InstanceStatus = 8
)

type Migration 

type Migration struct {
	Name        string
	FromVersion string
	ToVersion   string
	Apply       func(MigrationState) (MigrationState, error)
}

Migration declares one explicit directed version edge.

type MigrationState 

type MigrationState struct {
	Data []byte
}

MigrationState is opaque application-owned persisted state passed through an explicit version migration. Registry callers must persist the migrated state and target version atomically.

type OperatorAction 

type OperatorAction uint8

OperatorAction identifies one audited lifecycle intervention. 

const (
	// OperatorPause stops new progression of a running instance.
	OperatorPause OperatorAction = 1
	// OperatorResume resumes a paused instance.
	OperatorResume OperatorAction = 2
	// OperatorCancel requests durable cooperative cancellation.
	OperatorCancel OperatorAction = 3
	// OperatorTerminate forcibly ends a non-terminal instance.
	OperatorTerminate OperatorAction = 4
	// OperatorRetryActivity durably admits an explicit activity retry.
	OperatorRetryActivity OperatorAction = 5
	// OperatorCompensate durably schedules an explicit compensation.
	OperatorCompensate OperatorAction = 6
	// OperatorResolveCompensation records explicit manual reconciliation without
	// reporting successful rollback.
	OperatorResolveCompensation OperatorAction = 7
	// OperatorApprove records a caller-authorized human approval decision.
	OperatorApprove OperatorAction = 8
)

func (OperatorAction) String 

func (action OperatorAction) String() string

String returns the stable persisted action name.

type OperatorActionRecord 

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

OperatorActionRecord is immutable audit state reconstructed from history.

func (OperatorActionRecord) Action 

func (record OperatorActionRecord) Action() OperatorAction

Action returns the requested lifecycle intervention.

func (OperatorActionRecord) Actor 

func (record OperatorActionRecord) Actor() string

Actor returns the caller-authorized principal identity supplied to the command.

func (OperatorActionRecord) CommandID 

func (record OperatorActionRecord) CommandID() string

CommandID returns the idempotent operator command identity.

func (OperatorActionRecord) OccurredAt 

func (record OperatorActionRecord) OccurredAt() time.Time

OccurredAt returns the persisted command time.

func (OperatorActionRecord) Reason 

func (record OperatorActionRecord) Reason() string

Reason returns the bounded caller-supplied audit reason code.

type OperatorActivityRetrySpec 

type OperatorActivityRetrySpec struct {
	CommandID      string
	WorkID         string
	Instance       Instance
	Definition     Definition
	StepName       string
	IdempotencyKey string
	Actor          string
	Reason         string
	OccurredAt     time.Time
	Deadline       time.Time
	TenantID       string
	CorrelationID  string
}

OperatorActivityRetrySpec supplies one caller-authorized audited activity retry command. The command records audit history and retry due work in one optimistic transition.

type OperatorApprovalSpec 

type OperatorApprovalSpec struct {
	CommandID  string
	Instance   Instance
	Definition Definition
	StepName   string
	Actor      string
	Reason     string
	OccurredAt time.Time
	Payload    []byte
}

OperatorApprovalSpec supplies caller-authorized audit identity and bounded approval evidence for one human-approval step.

type OperatorCompensationResolutionSpec 

type OperatorCompensationResolutionSpec struct {
	CommandID  string
	Instance   Instance
	Definition Definition
	StepName   string
	Actor      string
	Reason     string
	Code       string
	Evidence   []byte
	OccurredAt time.Time
}

OperatorCompensationResolutionSpec supplies caller-authorized audit data and bounded manual reconciliation evidence for one failed or unknown compensation.

type OperatorCompensationSpec 

type OperatorCompensationSpec struct {
	CommandID      string
	WorkID         string
	Instance       Instance
	Definition     Definition
	StepName       string
	Attempt        uint32
	IdempotencyKey string
	Actor          string
	Reason         string
	OccurredAt     time.Time
	Deadline       time.Time
	Input          []byte
	TenantID       string
	CorrelationID  string
}

OperatorCompensationSpec supplies one caller-authorized audited explicit compensation command.

type OperatorLifecycleCommandSpec 

type OperatorLifecycleCommandSpec struct {
	CommandID  string
	Instance   Instance
	Action     OperatorAction
	Actor      string
	Reason     string
	OccurredAt time.Time
}

OperatorLifecycleCommandSpec supplies one caller-authorized audited command. Authorization remains the caller's policy boundary and is not inferred by this package.

type OrchestrationBranchSpec 

type OrchestrationBranchSpec struct {
	StepName       string
	WorkID         string
	IdempotencyKey string
	Input          []byte
}

OrchestrationBranchSpec supplies one bounded parallel branch dispatch. The branch name must exactly match its enclosing immutable control step.

type OrchestrationDecision 

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

OrchestrationDecision is one immutable persisted plan or explicit wait.

func NewOrchestrationDecision 

func NewOrchestrationDecision(spec OrchestrationDecisionSpec) (OrchestrationDecision, error)

NewOrchestrationDecision deterministically selects the first incomplete ordered step from replayed persisted progress. It does not execute side effects and rejects choreography or unsupported control-flow steps.

func (OrchestrationDecision) Kind 

func (decision OrchestrationDecision) Kind() OrchestrationDecisionKind

Kind returns the durable decision classification.

func (OrchestrationDecision) StepName 

func (decision OrchestrationDecision) StepName() string

StepName returns the affected definition step, or empty on completion.

func (OrchestrationDecision) Transition 

func (decision OrchestrationDecision) Transition() Transition

Transition returns the atomic plan. It is invalid for OrchestrationWaiting.

type OrchestrationDecisionKind 

type OrchestrationDecisionKind uint8

OrchestrationDecisionKind classifies one deterministic next-step decision. 

const (
	// OrchestrationScheduled persists the next due activity or timer work.
	OrchestrationScheduled OrchestrationDecisionKind = 1
	// OrchestrationWaiting means persisted progress is awaiting an activity,
	// timer, or external signal and no transition should be committed.
	OrchestrationWaiting OrchestrationDecisionKind = 2
	// OrchestrationCompleted persists a successful terminal outcome.
	OrchestrationCompleted OrchestrationDecisionKind = 3
	// OrchestrationFailed persists a known failed terminal outcome.
	OrchestrationFailed OrchestrationDecisionKind = 4
	// OrchestrationRecorded persists a durable control-flow decision without
	// creating externally executable work.
	OrchestrationRecorded OrchestrationDecisionKind = 5
)

type OrchestrationDecisionSpec 

type OrchestrationDecisionSpec struct {
	TransitionID   string
	WorkID         string
	Instance       Instance
	Definition     Definition
	DecidedAt      time.Time
	Deadline       time.Time
	IdempotencyKey string
	Input          []byte
	Result         []byte
	TenantID       string
	CorrelationID  string
	ChildID        string
	Branches       []OrchestrationBranchSpec
}

OrchestrationDecisionSpec supplies caller-owned identities, bounded data, and deterministic time for one ordered orchestration decision.

type PendingWork 

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

PendingWork is immutable work created atomically with workflow history.

func NewPendingWork 

func NewPendingWork(spec PendingWorkSpec) (PendingWork, error)

NewPendingWork validates and owns one bounded durable work record.

func (PendingWork) AvailableAt 

func (work PendingWork) AvailableAt() time.Time

AvailableAt returns the persisted earliest admission time.

func (PendingWork) CorrelationID 

func (work PendingWork) CorrelationID() string

CorrelationID returns optional correlation propagation data.

func (PendingWork) Deadline 

func (work PendingWork) Deadline() time.Time

Deadline returns the persisted execution deadline.

func (PendingWork) ID 

func (work PendingWork) ID() string

ID returns the stable globally unique durable work identity.

func (PendingWork) InstanceID 

func (work PendingWork) InstanceID() string

InstanceID returns the owning workflow instance.

func (PendingWork) Kind 

func (work PendingWork) Kind() WorkKind

Kind returns the explicit durable work classification.

func (PendingWork) Payload 

func (work PendingWork) Payload() []byte

Payload returns an owned copy of bounded work input.

func (PendingWork) Sequence 

func (work PendingWork) Sequence() uint64

Sequence returns the committed history position that created the work.

func (PendingWork) TenantID 

func (work PendingWork) TenantID() string

TenantID returns optional tenant propagation data.

type PendingWorkSpec 

type PendingWorkSpec struct {
	ID            string
	Kind          WorkKind
	InstanceID    string
	Sequence      uint64
	AvailableAt   time.Time
	Deadline      time.Time
	Payload       []byte
	TenantID      string
	CorrelationID string
}

PendingWorkSpec supplies one bounded durable work record.

type RaceProgress 

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

RaceProgress is one immutable durably selected race winner.

func (RaceProgress) DecidedAt 

func (progress RaceProgress) DecidedAt() time.Time

DecidedAt returns when the winner decision became durable history.

func (RaceProgress) StepName 

func (progress RaceProgress) StepName() string

StepName returns the stable race control-step name.

func (RaceProgress) WinnerStepName 

func (progress RaceProgress) WinnerStepName() string

WinnerStepName returns the persisted winning branch name.

type Registry 

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

Registry is an immutable explicit definition and migration registry.

func CompileDefinitions 

func CompileDefinitions(definitions ...Definition) (*Registry, error)

CompileDefinitions constructs a registry without migration edges.

func CompileRegistry 

func CompileRegistry(definitions []Definition, migrations []Migration) (*Registry, error)

CompileRegistry validates all immutable definitions and explicit migrations.

func (*Registry) Migration 

func (registry *Registry) Migration(name, fromVersion, toVersion string) (Migration, error)

Migration returns one explicitly declared direct version edge.

func (*Registry) Resolve 

func (registry *Registry) Resolve(name, version string) (Definition, error)

Resolve returns one exact pinned immutable definition version.

type RetryPolicy 

type RetryPolicy struct {
	MaxAttempts  uint32
	InitialDelay time.Duration
	MaxDelay     time.Duration
}

RetryPolicy is a bounded retry contract. Its zero value is invalid.

type SignalAcceptanceSpec 

type SignalAcceptanceSpec struct {
	InstanceID       string
	ExpectedSequence uint64
	Definition       Definition
	StepName         string
	SignalID         string
	ReceivedAt       time.Time
	Payload          []byte
}

SignalAcceptanceSpec supplies one inbound deduplicated signal transition.

type SignalProgress 

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

SignalProgress is one immutable durably accepted external signal.

func (SignalProgress) Payload 

func (progress SignalProgress) Payload() []byte

Payload returns an owned copy of the bounded signal payload.

func (SignalProgress) ReceivedAt 

func (progress SignalProgress) ReceivedAt() time.Time

ReceivedAt returns the persisted acceptance time.

func (SignalProgress) SignalID 

func (progress SignalProgress) SignalID() string

SignalID returns the inbound deduplication identity.

func (SignalProgress) StepName 

func (progress SignalProgress) StepName() string

StepName returns the stable definition signal step.

type StepKind 

type StepKind uint8

StepKind identifies one durable definition step. 

const (
	// StepActivity requests an idempotent external activity.
	StepActivity StepKind = 1
	// StepSignal waits for a named external signal.
	StepSignal StepKind = 2
	// StepTimer waits for durable time to become due.
	StepTimer StepKind = 3
	// StepChild starts or observes a version-pinned child workflow.
	StepChild StepKind = 4
	// StepParallel admits a bounded set of branches.
	StepParallel StepKind = 5
	// StepJoin waits for an explicitly modeled branch set.
	StepJoin StepKind = 6
	// StepRace selects the first persisted winning branch.
	StepRace StepKind = 7
	// StepApproval waits for an authorized operator or application decision.
	StepApproval StepKind = 8
)

type StepSpec 

type StepSpec struct {
	Name            string
	Kind            StepKind
	Target          string
	ChildDefinition DefinitionReference
	Branches        []string
	Timeout         time.Duration
	InputLimit      uint32
	ResultLimit     uint32
	Retry           RetryPolicy
	FanOutLimit     uint32
	Compensation    *CompensationSpec
}

StepSpec is definition input. NewDefinition validates and deep-copies it. Target names an activity, signal, child definition, or approval policy as selected by Kind. ChildDefinition is required only for StepChild and pins the exact child behavior fingerprint.

type StoreCommitError 

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

StoreCommitError preserves a cause without exposing driver diagnostics and makes transition durability explicit.

func (*StoreCommitError) CommitOutcome 

func (commitError *StoreCommitError) CommitOutcome() StoreCommitOutcome

CommitOutcome returns the explicit durability classification.

func (*StoreCommitError) Error 

func (commitError *StoreCommitError) Error() string

Error implements error without exposing the wrapped cause text.

func (*StoreCommitError) Unwrap 

func (commitError *StoreCommitError) Unwrap() error

Unwrap preserves the cause for errors.Is and errors.As.

type StoreCommitOutcome 

type StoreCommitOutcome uint8

StoreCommitOutcome classifies whether a failed transition reached durable storage. Unknown outcomes require reconciliation before retry. 

const (
	// StoreCommitUnknown means the durable outcome must be reconciled.
	StoreCommitUnknown StoreCommitOutcome = iota
	// StoreCommitNotCommitted means retrying the same transition is safe.
	StoreCommitNotCommitted
	// StoreCommitCommitted means persistence succeeded before a later failure.
	StoreCommitCommitted
)

func StoreCommitOutcomeOf 

func StoreCommitOutcomeOf(err error) StoreCommitOutcome

StoreCommitOutcomeOf returns an error's durability classification. Unclassified errors are conservatively unknown.

type SystemClock 

type SystemClock struct{}

SystemClock uses the process wall clock. Tests and replay-sensitive callers should supply a deterministic implementation.

func (SystemClock) NewTimer 

func (SystemClock) NewTimer(duration time.Duration) ClockTimer

NewTimer creates one caller-owned process timer.

func (SystemClock) Now 

func (SystemClock) Now() time.Time

Now returns the current canonical wall-clock time.

type TimerFireSpec 

type TimerFireSpec struct {
	TransitionID     string
	Lease            WorkLease
	ExpectedSequence uint64
	Definition       Definition
	FiredAt          time.Time
}

TimerFireSpec supplies one persisted due observation from current fenced timer work. The returned transition must commit before the lease completes.

type TimerProgress 

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

TimerProgress is immutable state reconstructed only from persisted events.

func (TimerProgress) DueAt 

func (progress TimerProgress) DueAt() time.Time

DueAt returns the persisted timer deadline.

func (TimerProgress) FiredAt 

func (progress TimerProgress) FiredAt() time.Time

FiredAt returns the persisted due observation, or zero while waiting.

func (TimerProgress) Status 

func (progress TimerProgress) Status() TimerProgressStatus

Status returns the durable timer state.

func (TimerProgress) StepName 

func (progress TimerProgress) StepName() string

StepName returns the stable definition timer step.

type TimerProgressStatus 

type TimerProgressStatus uint8

TimerProgressStatus identifies replayed durable timer progress. 

const (
	// TimerWaiting has a persisted deadline and outstanding durable work.
	TimerWaiting TimerProgressStatus = 1
	// TimerFired has a persisted due-time observation.
	TimerFired TimerProgressStatus = 2
)

type TimerScheduleSpec 

type TimerScheduleSpec struct {
	TransitionID     string
	WorkID           string
	InstanceID       string
	ExpectedSequence uint64
	Definition       Definition
	StepName         string
	ScheduledAt      time.Time
	Deadline         time.Time
	TenantID         string
	CorrelationID    string
}

TimerScheduleSpec supplies one atomic durable timer admission plan.

type Transition 

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

Transition is one immutable atomic persistence plan. A store must append all events and create all work in one transaction or make none of them visible.

func NewActivityAttemptOutcome 

func NewActivityAttemptOutcome(spec ActivityAttemptOutcomeSpec) (Transition, error)

NewActivityAttemptOutcome records an explicit bounded activity outcome.

func NewActivityAttemptStart 

func NewActivityAttemptStart(spec ActivityAttemptStartSpec) (Transition, error)

NewActivityAttemptStart creates the attempt-start transition that must commit before external activity execution.

func NewActivityRetry 

func NewActivityRetry(spec ActivityRetrySpec) (Transition, error)

NewActivityRetry atomically records deterministic retry admission and work.

func NewActivitySchedule 

func NewActivitySchedule(spec ActivityScheduleSpec) (Transition, error)

NewActivitySchedule records bounded activity input before dispatching work.

func NewChildOutcome 

func NewChildOutcome(spec ChildOutcomeSpec) (Transition, error)

NewChildOutcome records a known child terminal result before parent orchestration advances.

func NewChildSchedule 

func NewChildSchedule(spec ChildScheduleSpec) (Transition, error)

NewChildSchedule records the pinned child identity and bounded input before any child-start adapter may act.

func NewChildStartAttempt 

func NewChildStartAttempt(spec ChildStartAttemptSpec) (Transition, error)

NewChildStartAttempt records a fenced child creation attempt before calling a child-start adapter.

func NewChildStartAttemptOutcome 

func NewChildStartAttemptOutcome(spec ChildStartAttemptOutcomeSpec) (Transition, error)

NewChildStartAttemptOutcome persists a known or uncertain creation result.

func NewChildStartRetry 

func NewChildStartRetry(spec ChildStartRetrySpec) (Transition, error)

NewChildStartRetry atomically records retry admission and its next work.

func NewCompensationAttemptOutcome 

func NewCompensationAttemptOutcome(spec CompensationAttemptOutcomeSpec) (Transition, error)

NewCompensationAttemptOutcome records an explicit bounded compensation outcome without translating failure into successful rollback.

func NewCompensationAttemptStart 

func NewCompensationAttemptStart(spec CompensationAttemptStartSpec) (Transition, error)

NewCompensationAttemptStart creates the attempt-start transition that must commit before external compensation begins.

func NewCompensationRetry 

func NewCompensationRetry(spec CompensationRetrySpec) (Transition, error)

NewCompensationRetry atomically records independently retryable compensation admission and its durable work.

func NewCompensationSchedule 

func NewCompensationSchedule(spec CompensationScheduleSpec) (Transition, error)

NewCompensationSchedule atomically schedules compensation history and work.

func NewOperatorActivityRetry 

func NewOperatorActivityRetry(spec OperatorActivityRetrySpec) (Transition, error)

NewOperatorActivityRetry atomically records caller-supplied authorization audit data before admitting a definition-policy activity retry.

func NewOperatorApproval 

func NewOperatorApproval(spec OperatorApprovalSpec) (Transition, error)

NewOperatorApproval atomically records audit identity before accepting one approval decision. Replay treats the approval as persisted step progress.

func NewOperatorCompensation 

func NewOperatorCompensation(spec OperatorCompensationSpec) (Transition, error)

NewOperatorCompensation atomically records audit data before scheduling the requested compensation and its first durable work item.

func NewOperatorCompensationResolution 

func NewOperatorCompensationResolution(spec OperatorCompensationResolutionSpec) (Transition, error)

NewOperatorCompensationResolution records manual reconciliation as its own durable outcome. It never emits CompensationSucceeded.

func NewOperatorLifecycleCommand 

func NewOperatorLifecycleCommand(spec OperatorLifecycleCommandSpec) (Transition, error)

NewOperatorLifecycleCommand atomically records audit identity before the requested lifecycle action. Concurrent commands are serialized by the instance sequence and exact command replay is idempotent by CommandID.

func NewSignalAcceptance 

func NewSignalAcceptance(spec SignalAcceptanceSpec) (Transition, error)

NewSignalAcceptance creates the transition that must commit before an inbound transport acknowledges the signal.

func NewTimerFire 

func NewTimerFire(spec TimerFireSpec) (Transition, error)

NewTimerFire creates the durable timer-fired decision represented by a current timer lease.

func NewTimerSchedule 

func NewTimerSchedule(spec TimerScheduleSpec) (Transition, error)

NewTimerSchedule creates history and due work that must commit atomically.

func NewTransition 

func NewTransition(spec TransitionSpec) (Transition, error)

NewTransition validates and owns one bounded atomic persistence plan.

func (Transition) Definition 

func (transition Transition) Definition() DefinitionReference

Definition returns the exact behavior identity that made this decision.

func (Transition) Events 

func (transition Transition) Events() []HistoryEvent

Events returns an owned ordered copy of the atomic history append.

func (Transition) ExpectedSequence 

func (transition Transition) ExpectedSequence() uint64

ExpectedSequence returns the optimistic-concurrency precondition.

func (Transition) Fingerprint 

func (transition Transition) Fingerprint() string

Fingerprint returns a deterministic digest of the complete ordered plan for distinguishing exact idempotent replay from conflicting ID reuse.

func (Transition) ID 

func (transition Transition) ID() string

ID returns the idempotency identity for this complete persistence plan.

func (Transition) InstanceID 

func (transition Transition) InstanceID() string

InstanceID returns the owning workflow instance.

func (Transition) Valid 

func (transition Transition) Valid() bool

Valid reports whether the transition was constructed with NewTransition.

func (Transition) Work 

func (transition Transition) Work() []PendingWork

Work returns owned durable work created by the atomic append.

type TransitionReconciliation 

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

TransitionReconciliation is an immutable transition identity used to distinguish missing, exact committed, and conflicting durable outcomes.

func NewTransitionReconciliation 

func NewTransitionReconciliation(spec TransitionReconciliationSpec) (TransitionReconciliation, error)

NewTransitionReconciliation validates one uncertain transition identity.

func (TransitionReconciliation) Fingerprint 

func (reconciliation TransitionReconciliation) Fingerprint() string

Fingerprint returns the expected complete transition digest.

func (TransitionReconciliation) TransitionID 

func (reconciliation TransitionReconciliation) TransitionID() string

TransitionID returns the idempotent transition identity.

func (TransitionReconciliation) Valid 

func (reconciliation TransitionReconciliation) Valid() bool

Valid reports whether the reconciliation identity is coherent.

type TransitionReconciliationOutcome 

type TransitionReconciliationOutcome uint8

TransitionReconciliationOutcome classifies an uncertain commit lookup. 

const (
	// TransitionMissing means the transition identity is not durable.
	TransitionMissing TransitionReconciliationOutcome = 1
	// TransitionCommitted means the exact transition fingerprint is durable.
	TransitionCommitted TransitionReconciliationOutcome = 2
	// TransitionConflicting means the identity is durable with different content.
	TransitionConflicting TransitionReconciliationOutcome = 3
)

type TransitionReconciliationSpec 

type TransitionReconciliationSpec struct {
	TransitionID string
	Fingerprint  string
}

TransitionReconciliationSpec supplies one uncertain transition identity.

type TransitionSpec 

type TransitionSpec struct {
	ID               string
	InstanceID       string
	ExpectedSequence uint64
	Definition       DefinitionReference
	Events           []HistoryEvent
	Work             []PendingWork
}

TransitionSpec supplies one idempotent atomic history-and-work append.

type TransitionStore 

type TransitionStore interface {
	Commit(context.Context, Transition) error
	History(context.Context, HistoryQuery) (HistoryPage, error)
}

TransitionStore atomically appends history and creates due work. Commit must enforce ExpectedSequence and Transition.ID idempotency. It must return a StoreCommitError whenever an error follows a possible commit boundary.

type WorkClaimRequest 

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

WorkClaimRequest is an immutable atomic due-work admission request.

func NewWorkClaimRequest 

func NewWorkClaimRequest(spec WorkClaimRequestSpec) (WorkClaimRequest, error)

NewWorkClaimRequest validates one bounded claim request.

func (WorkClaimRequest) LeaseDuration 

func (request WorkClaimRequest) LeaseDuration() time.Duration

LeaseDuration returns the bounded initial ownership interval.

func (WorkClaimRequest) Limit 

func (request WorkClaimRequest) Limit() uint32

Limit returns the maximum number of records claimed atomically.

func (WorkClaimRequest) Now 

func (request WorkClaimRequest) Now() time.Time

Now returns the caller-supplied deterministic admission time.

func (WorkClaimRequest) Owner 

func (request WorkClaimRequest) Owner() string

Owner returns the stable process-specific lease owner identity.

func (WorkClaimRequest) Valid 

func (request WorkClaimRequest) Valid() bool

Valid reports whether the request is bounded and internally coherent.

type WorkClaimRequestSpec 

type WorkClaimRequestSpec struct {
	Owner         string
	Now           time.Time
	LeaseDuration time.Duration
	Limit         uint32
}

WorkClaimRequestSpec supplies one bounded due-work claim operation.

type WorkCompletion 

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

WorkCompletion is an immutable fenced successful terminal mutation.

func NewWorkCompletion 

func NewWorkCompletion(spec WorkCompletionSpec) (WorkCompletion, error)

NewWorkCompletion validates one fenced completion.

func (WorkCompletion) CompletedAt 

func (completion WorkCompletion) CompletedAt() time.Time

CompletedAt returns the persisted completion time.

func (WorkCompletion) Owner 

func (completion WorkCompletion) Owner() string

Owner returns the expected current owner.

func (WorkCompletion) Token 

func (completion WorkCompletion) Token() uint64

Token returns the expected current fencing token.

func (WorkCompletion) Valid 

func (completion WorkCompletion) Valid() bool

Valid reports whether the completion fence is coherent.

func (WorkCompletion) WorkID 

func (completion WorkCompletion) WorkID() string

WorkID returns the durable work identity.

type WorkCompletionSpec 

type WorkCompletionSpec struct {
	WorkID      string
	Owner       string
	Token       uint64
	CompletedAt time.Time
}

WorkCompletionSpec supplies one fenced successful terminal mutation.

type WorkDecision 

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

WorkDecision is an immutable processor disposition. A processor must not return WorkComplete until the workflow transition represented by the work is durable. Unknown external outcomes must first persist reconciliation state.

func NewWorkDecision 

func NewWorkDecision(spec WorkDecisionSpec) (WorkDecision, error)

NewWorkDecision validates one explicit bounded disposition.

func (WorkDecision) Code 

func (decision WorkDecision) Code() string

Code returns the stable retry or dead-letter classification.

func (WorkDecision) Kind 

func (decision WorkDecision) Kind() WorkDecisionKind

Kind returns the durable disposition.

func (WorkDecision) RetryAt 

func (decision WorkDecision) RetryAt() time.Time

RetryAt returns explicit retry admission time, or zero otherwise.

func (WorkDecision) Valid 

func (decision WorkDecision) Valid() bool

Valid reports whether the disposition is unambiguous.

type WorkDecisionKind 

type WorkDecisionKind uint8

WorkDecisionKind selects the fenced durable disposition after processing. 

const (
	// WorkComplete acknowledges work only after its represented durable
	// transition has been persisted by the processor.
	WorkComplete WorkDecisionKind = 1
	// WorkRetryDecision releases a known-safe failure at an explicit time.
	WorkRetryDecision WorkDecisionKind = 2
	// WorkDeadLetterDecision records poison or manually resolvable work without
	// reporting successful completion.
	WorkDeadLetterDecision WorkDecisionKind = 3
)

type WorkDecisionSpec 

type WorkDecisionSpec struct {
	Kind    WorkDecisionKind
	Code    string
	RetryAt time.Time
}

WorkDecisionSpec supplies one explicit processor disposition.

type WorkDisposition 

type WorkDisposition uint8

WorkDisposition selects durable handling for one known work failure. 

const (
	// WorkRetry returns work to due admission at an explicit future time.
	WorkRetry WorkDisposition = 1
	// WorkDeadLetter makes poison work unavailable pending operator resolution.
	WorkDeadLetter WorkDisposition = 2
)

type WorkFailure 

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

WorkFailure is an immutable fenced retry or dead-letter decision.

func NewWorkFailure 

func NewWorkFailure(spec WorkFailureSpec) (WorkFailure, error)

NewWorkFailure validates one explicit known failure decision.

func (WorkFailure) Code 

func (failure WorkFailure) Code() string

Code returns the stable safe failure classification.

func (WorkFailure) Disposition 

func (failure WorkFailure) Disposition() WorkDisposition

Disposition returns retry or dead-letter handling.

func (WorkFailure) FailedAt 

func (failure WorkFailure) FailedAt() time.Time

FailedAt returns the persisted known-failure time.

func (WorkFailure) Owner 

func (failure WorkFailure) Owner() string

Owner returns the expected current owner.

func (WorkFailure) RetryAt 

func (failure WorkFailure) RetryAt() time.Time

RetryAt returns retry admission time, or zero for dead-letter handling.

func (WorkFailure) Token 

func (failure WorkFailure) Token() uint64

Token returns the expected current fencing token.

func (WorkFailure) Valid 

func (failure WorkFailure) Valid() bool

Valid reports whether the failure decision and fence are coherent.

func (WorkFailure) WorkID 

func (failure WorkFailure) WorkID() string

WorkID returns the durable work identity.

type WorkFailureSpec 

type WorkFailureSpec struct {
	WorkID      string
	Owner       string
	Token       uint64
	FailedAt    time.Time
	Code        string
	Disposition WorkDisposition
	RetryAt     time.Time
}

WorkFailureSpec supplies one fenced known failure decision.

type WorkKind 

type WorkKind uint8

WorkKind identifies one durable unit that may progress an instance only after the transition that created it commits. 

const (
	// WorkActivity dispatches one explicit external activity attempt.
	WorkActivity WorkKind = 1
	// WorkTimer observes one durable timer becoming due.
	WorkTimer WorkKind = 2
	// WorkChild progresses one version-pinned child workflow operation.
	WorkChild WorkKind = 3
	// WorkPublication publishes one durably accepted outbound message.
	WorkPublication WorkKind = 4
	// WorkReconciliation resolves an activity with an unknown outcome.
	WorkReconciliation WorkKind = 5
	// WorkCompensation dispatches one explicit compensating activity attempt.
	WorkCompensation WorkKind = 6
)

type WorkLease 

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

WorkLease is immutable claimed work. Token is a monotonically increasing fencing value; every renewal or terminal mutation must match it.

func NewWorkLease 

func NewWorkLease(spec WorkLeaseSpec) (WorkLease, error)

NewWorkLease validates and owns one claimed-work record.

func (WorkLease) Attempt 

func (lease WorkLease) Attempt() uint32

Attempt returns the one-based durable claim attempt.

func (WorkLease) ClaimedAt 

func (lease WorkLease) ClaimedAt() time.Time

ClaimedAt returns the persisted claim time.

func (WorkLease) ExpiresAt 

func (lease WorkLease) ExpiresAt() time.Time

ExpiresAt returns the persisted ownership expiry.

func (WorkLease) Owner 

func (lease WorkLease) Owner() string

Owner returns the current lease owner.

func (WorkLease) Token 

func (lease WorkLease) Token() uint64

Token returns the current fencing token.

func (WorkLease) Valid 

func (lease WorkLease) Valid() bool

Valid reports whether the claimed work and fence are coherent.

func (WorkLease) Work 

func (lease WorkLease) Work() PendingWork

Work returns an owned durable-work value.

type WorkLeaseRenewal 

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

WorkLeaseRenewal is an immutable fenced lease extension.

func NewWorkLeaseRenewal 

func NewWorkLeaseRenewal(spec WorkLeaseRenewalSpec) (WorkLeaseRenewal, error)

NewWorkLeaseRenewal validates one bounded fenced extension.

func (WorkLeaseRenewal) ExtendBy 

func (renewal WorkLeaseRenewal) ExtendBy() time.Duration

ExtendBy returns the bounded new lease interval.

func (WorkLeaseRenewal) Now 

func (renewal WorkLeaseRenewal) Now() time.Time

Now returns the deterministic renewal time.

func (WorkLeaseRenewal) Owner 

func (renewal WorkLeaseRenewal) Owner() string

Owner returns the expected current owner.

func (WorkLeaseRenewal) Token 

func (renewal WorkLeaseRenewal) Token() uint64

Token returns the expected current fencing token.

func (WorkLeaseRenewal) Valid 

func (renewal WorkLeaseRenewal) Valid() bool

Valid reports whether the renewal is bounded and internally coherent.

func (WorkLeaseRenewal) WorkID 

func (renewal WorkLeaseRenewal) WorkID() string

WorkID returns the durable work identity.

type WorkLeaseRenewalSpec 

type WorkLeaseRenewalSpec struct {
	WorkID   string
	Owner    string
	Token    uint64
	Now      time.Time
	ExtendBy time.Duration
}

WorkLeaseRenewalSpec supplies one fenced lease extension.

type WorkLeaseSpec 

type WorkLeaseSpec struct {
	Work      PendingWork
	Owner     string
	Token     uint64
	Attempt   uint32
	ClaimedAt time.Time
	ExpiresAt time.Time
}

WorkLeaseSpec supplies one durable claimed-work record.

type WorkProcessor 

type WorkProcessor interface {
	Process(context.Context, WorkLease) (WorkDecision, error)
}

WorkProcessor executes one leased unit. It must honor context cancellation, return only after all owned goroutines stop, and persist any workflow transition before returning WorkComplete. An error leaves the lease for fenced recovery; it must not hide an unknown external side effect.

type WorkStore 

type WorkStore interface {
	Claim(context.Context, WorkClaimRequest) ([]WorkLease, error)
	Renew(context.Context, WorkLeaseRenewal) (WorkLease, error)
	Complete(context.Context, WorkCompletion) error
	Fail(context.Context, WorkFailure) error
}

WorkStore atomically claims due work and rejects every renewal or terminal mutation whose owner or fencing token is stale.

type Worker 

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

Worker owns bounded claim, processing, renewal, and graceful shutdown goroutines. Process supervision remains the caller's responsibility.

func NewWorker 

func NewWorker(config WorkerConfig) (*Worker, error)

NewWorker validates explicit dependencies and resource limits.

func (*Worker) Handle 

func (worker *Worker) Handle(ctx context.Context, lease WorkLease) error

Handle processes one valid current lease, renews it on a bounded cadence, and applies its explicit fenced disposition. It is exposed for embedding in caller-owned admission loops.

func (*Worker) Run 

func (worker *Worker) Run(ctx context.Context) error

Run claims work immediately, owns at most MaxConcurrent handlers, and stops claiming on cancellation. It waits for every processor to honor cancellation and exit before returning. Recoverable store and processor failures leave fenced leases for expiry recovery and do not stop the process loop.

type WorkerConfig 

type WorkerConfig struct {
	Store           WorkStore
	Processor       WorkProcessor
	Clock           Clock
	Owner           string
	MaxConcurrent   uint32
	ClaimLimit      uint32
	LeaseDuration   time.Duration
	RenewEvery      time.Duration
	PollInterval    time.Duration
	FinalizeTimeout time.Duration
	Hooks           WorkerHooks
}

WorkerConfig supplies explicit bounded worker ownership and timing policy.

type WorkerEvent 

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

WorkerEvent is one synchronous lifecycle notification. WorkID is data and must not be used as an unbounded metric label.

func (WorkerEvent) At 

func (event WorkerEvent) At() time.Time

At returns the deterministic observation time.

func (WorkerEvent) Attempt 

func (event WorkerEvent) Attempt() uint32

Attempt returns the durable claim attempt, or zero for a claim failure.

func (WorkerEvent) Cause 

func (event WorkerEvent) Cause() error

Cause returns the underlying error for caller-owned logs and traces.

func (WorkerEvent) Kind 

func (event WorkerEvent) Kind() WorkerEventKind

Kind returns the lifecycle classification.

func (WorkerEvent) WorkID 

func (event WorkerEvent) WorkID() string

WorkID returns the affected work identity, or empty for claim failures.

func (WorkerEvent) WorkKind 

func (event WorkerEvent) WorkKind() WorkKind

WorkKind returns the bounded durable-work classification, or zero for a claim operation that failed before any work was admitted.

type WorkerEventKind 

type WorkerEventKind uint8

WorkerEventKind classifies bounded worker lifecycle hooks. 

const (
	// WorkerClaimFailed reports a recoverable durable admission failure.
	WorkerClaimFailed WorkerEventKind = 1
	// WorkerProcessingFailed reports work left leased for expiry recovery.
	WorkerProcessingFailed WorkerEventKind = 2
	// WorkerLeaseLost reports cancellation caused by a stale ownership fence.
	WorkerLeaseLost WorkerEventKind = 3
	// WorkerWorkClaimed reports first admission of one durable work item.
	WorkerWorkClaimed WorkerEventKind = 4
	// WorkerWorkReadmitted reports admission after any prior durable claim.
	// Attempt metadata distinguishes it from first admission; the hook does not
	// guess whether the cause was an explicit retry or lease-expiry recovery.
	WorkerWorkReadmitted WorkerEventKind = 5
	// WorkerProcessingStarted reports bounded processor invocation.
	WorkerProcessingStarted WorkerEventKind = 6
	// WorkerLeaseRenewed reports successful ownership extension or heartbeat.
	WorkerLeaseRenewed WorkerEventKind = 7
	// WorkerCompleted reports durable successful work finalization.
	WorkerCompleted WorkerEventKind = 8
	// WorkerRetryScheduled reports durable retry admission.
	WorkerRetryScheduled WorkerEventKind = 9
	// WorkerDeadLettered reports durable poison-work isolation.
	WorkerDeadLettered WorkerEventKind = 10
)

type WorkerHooks 

type WorkerHooks interface {
	OnWorkerEvent(WorkerEvent)
}

WorkerHooks receives synchronous bounded lifecycle events. Calls may occur concurrently from the bounded worker handlers. Implementations must provide their own synchronization, return promptly, and must not panic.

Source Files

View all Source files

Directories

Path Synopsis
Package postgres provides durable PostgreSQL workflow persistence.
 Package postgres provides durable PostgreSQL workflow persistence.

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.