Versions in this module Expand all Collapse all v0 v0.4.0 Sep 2, 2026 v0.3.0 Sep 1, 2026 Changes in this version + const DispatchBlockedFence + const DispatchBlockedUnknown + const DispatchFailedPermanent + const DispatchInFlight + const DispatchQueued + const DispatchRetryWait + const DispatchSucceeded + const ExecutionAccepted + const ExecutionAdmitting + const ExecutionBlockedCompensation + const ExecutionBlockedDependency + const ExecutionBlockedFence + const ExecutionBlockedUnknown + const ExecutionCompleted + const ExecutionFailed + const ExecutionRunning + const SagaBlockedCompensation + const SagaBlockedDependency + const SagaBlockedFence + const SagaBlockedUnknown + const SagaCompensated + const SagaCompensating + const SagaCompleted + const SagaEffectCompensated + const SagaEffectFailed + const SagaEffectPending + const SagaEffectSuccess + const SagaFailed + const SagaRunning + const StepCompensated + const StepFailed + const StepPending + const StepSucceeded + var ErrArtifactNotFound = errors.New("execution artifact not found") + var ErrExecutionNotFound = errors.New("execution not found") + var ErrIdentityConflict = errors.New("durable identity conflict") + var ErrInvalidTransition = errors.New("invalid saga state transition") + var ErrNoDispatch = errors.New("no eligible dispatch") + var ErrOptimisticConflict = errors.New("optimistic persistence conflict") + var ErrSagaStoreRequired = errors.New("saga execution requires a saga store") + var ErrStaleExecutionLease = errors.New("stale execution recovery lease") + var ErrStaleLease = errors.New("stale dispatch lease") + var ErrTerminalSaga = errors.New("terminal saga cannot be reopened") + func CanonicalJSON(value any) (json.RawMessage, string, error) + func CreateExtensionManagerWithDefaults(extensionDirs ...string) (*loader.ExtensionManager, error) + func EvaluatePredicatesWithFacts(predicates []*Predicate, facts effectus.Facts) bool + func EvaluatePredicatesWithFactsE(predicates []*Predicate, facts effectus.Facts) (bool, error) + func ExtractFactPaths(expression string) map[string]struct + func IdempotencyKey(namespace, sagaID, effectID string, direction invocation.Direction) string + func IsTerminalExecutionState(state ExecutionState) bool + func LoadExtensionsIntoRegistries(em *loader.ExtensionManager, registry *Registry, verbRegistry *verb.Registry) error + func MigrateRedisOutboxV2(ctx context.Context, options RedisOutboxStoreOptions) error + func MigrateSagaV2(ctx context.Context, db *sql.DB) error + func ResetDefaultClock() + func SagaEffectID(sequence int) string + func SetDefaultClock(clock func() time.Time) + func SetFixedTime(now time.Time) + func StableAdmissionID(namespace, deliveryID, ruleset, version string) string + func StableExecutionID(namespace, deliveryID, ruleset, version string) string + func StableSagaID(executionID, planID string) string + func ValidateSagaV2(ctx context.Context, db *sql.DB) error + type AtomicAdmissionStore = ledger.AtomicAdmissionStore + type BufBreakingConfig struct + Use []string + type BufBuildConfig struct + Excludes []string + type BufConfig struct + Breaking BufBreakingConfig + Build BufBuildConfig + Deps []string + Lint BufLintConfig + Name string + Version string + type BufGenConfig struct + Managed BufManagedConfig + Plugins []BufPluginConfig + Version string + type BufGoPackagePrefixConfig struct + Default string + Except []string + type BufIntegration struct + func NewBufIntegration(workspaceRoot string) (*BufIntegration, error) + func (b *BufIntegration) GenerateCode(ctx context.Context) (*CodeGenerationResult, error) + func (b *BufIntegration) GetFactSchema(name string) (*FactSchema, bool) + func (b *BufIntegration) GetVerbSchema(name string) (*VerbSchema, bool) + func (b *BufIntegration) ListFactSchemas() map[string]*FactSchema + func (b *BufIntegration) ListVerbSchemas() map[string]*VerbSchema + func (b *BufIntegration) RegisterFactSchema(ctx context.Context, schema *FactSchema) error + func (b *BufIntegration) RegisterVerbSchema(ctx context.Context, schema *VerbSchema) error + func (b *BufIntegration) ValidateSchemas(ctx context.Context) (*SchemaValidationResult, error) + type BufLintConfig struct + AllowCommentIgnores bool + Except []string + Use []string + type BufManagedConfig struct + Enabled bool + GoPackagePrefix BufGoPackagePrefixConfig + type BufPluginConfig struct + Opt []string + Out string + Plugin string + type CheckedEnqueueRequest struct + Arguments map[string]any + CompensationArguments map[string]any + CompensationContract string + CompensationVerb string + EffectID string + Facts map[string]any + Fencing []FencingRequirement + PlanID string + ResultSlots []any + SagaID string + type ClaimOptions = workflow.ClaimOptions + type CodeGenerationResult struct + Duration time.Duration + Errors []string + GeneratedFiles []string + Metadata map[string]interface{} + Success bool + Warnings []string + type Completion = workflow.Completion + type CreateSagaRequest = workflow.CreateSagaRequest + type Dispatch = workflow.Dispatch + func EnqueueCheckedStep(ctx context.Context, store OutboxStore, checked *ir.Checked, ...) (*Dispatch, error) + type DispatchAttempt = workflow.DispatchAttempt + type DispatchState = workflow.DispatchState + type Dispatcher struct + func NewDispatcher(store OutboxStore, provider fencing.Provider, executor invocation.Executor, ...) (*Dispatcher, error) + func (dispatcher *Dispatcher) Dispatch(ctx context.Context, targetDispatchID string) (*Dispatch, error) + func (dispatcher *Dispatcher) DispatchOne(ctx context.Context) (*Dispatch, error) + type DispatcherOptions struct + InitialBackoff time.Duration + InvocationTimeout time.Duration + LeaseDuration time.Duration + MaxAttempts uint64 + MaxBackoff time.Duration + Owner string + RequestID string + RequireDurableFencing bool + type DurableAdmission = ledger.DurableAdmission + type EnqueueStepRequest = workflow.EnqueueStepRequest + type ExecutionArtifact = ledger.ExecutionArtifact + type ExecutionLease = ledger.ExecutionLease + type ExecutionLedger = ledger.ExecutionLedger + type ExecutionPlan struct + Steps []*ExecutionStep + type ExecutionPlanRecord = ledger.ExecutionPlanRecord + type ExecutionRecord = ledger.ExecutionRecord + type ExecutionState = ledger.ExecutionState + type ExecutionStep struct + CanRunConcurrently bool + Effects []effectus.Effect + Sequence int + type FactApplication = ledger.FactApplication + type FactSchema struct + BufCommit string + BufModule string + CreatedAt time.Time + Description string + Indexes []IndexDefinition + Name string + PrivacyRules []PrivacyRule + RetentionPolicy *RetentionPolicy + Schema map[string]interface{} + UpdatedAt time.Time + Version string + type FactSchemaRegistry struct + type FencingRequirement = workflow.FencingRequirement + type InMemoryExecutionLedger struct + func NewInMemoryExecutionLedger() *InMemoryExecutionLedger + func (store *InMemoryExecutionLedger) AdmitExecution(ctx context.Context, admission DurableAdmission) (ExecutionRecord, bool, error) + func (store *InMemoryExecutionLedger) FinishExecutionLease(ctx context.Context, lease ExecutionLease, state ExecutionState, ...) error + func (store *InMemoryExecutionLedger) GetArtifact(ctx context.Context, digest string) (ExecutionArtifact, error) + func (store *InMemoryExecutionLedger) GetExecution(ctx context.Context, id string) (ExecutionRecord, error) + func (store *InMemoryExecutionLedger) GetExecutionByAdmission(ctx context.Context, identity string) (ExecutionRecord, error) + func (store *InMemoryExecutionLedger) LeaseExecutions(ctx context.Context, owner string, limit int, duration time.Duration) ([]ExecutionLease, error) + func (store *InMemoryExecutionLedger) PutArtifact(ctx context.Context, artifact ExecutionArtifact) error + func (store *InMemoryExecutionLedger) SetExecutionState(ctx context.Context, id string, revision uint64, state ExecutionState, ...) (ExecutionRecord, error) + type InMemoryOutboxStore struct + func NewInMemoryOutboxStore() *InMemoryOutboxStore + func (store *InMemoryOutboxStore) ClaimDispatch(ctx context.Context, options ClaimOptions) (*Dispatch, error) + func (store *InMemoryOutboxStore) CompleteDispatch(ctx context.Context, completion Completion) error + func (store *InMemoryOutboxStore) CompleteSaga(ctx context.Context, sagaID string) error + func (store *InMemoryOutboxStore) CreateSaga(ctx context.Context, request CreateSagaRequest) (*SagaInstance, error) + func (store *InMemoryOutboxStore) EnqueueStep(ctx context.Context, request EnqueueStepRequest) (*Dispatch, error) + func (store *InMemoryOutboxStore) GetDispatch(ctx context.Context, dispatchID string) (*Dispatch, error) + func (store *InMemoryOutboxStore) GetSaga(ctx context.Context, sagaID string) (*SagaInstance, error) + func (store *InMemoryOutboxStore) ListAttempts(ctx context.Context, dispatchID string) ([]DispatchAttempt, error) + func (store *InMemoryOutboxStore) ListDispatches(ctx context.Context, sagaID string) ([]*Dispatch, error) + func (store *InMemoryOutboxStore) SaveFencingGrants(ctx context.Context, dispatchID string, attempt uint64, leaseToken string, ...) error + type InMemorySagaStore struct + func NewInMemorySagaStore() *InMemorySagaStore + func (ims *InMemorySagaStore) CompleteSaga(sagaID string) error + func (ims *InMemorySagaStore) GetActiveSagas() ([]string, error) + func (ims *InMemorySagaStore) GetTransactionEffects(sagaID string) ([]*SagaEffect, error) + func (ims *InMemorySagaStore) MarkCompensated(sagaID, effectID string) error + func (ims *InMemorySagaStore) MarkFailed(sagaID, effectID string, reason error) error + func (ims *InMemorySagaStore) MarkSuccess(sagaID, effectID string, result interface{}) error + func (ims *InMemorySagaStore) RecordEffect(sagaID, effectID string, sequence int, verb string, ...) error + func (ims *InMemorySagaStore) StartTransaction(sagaID, ruleName string) error + type IndexDefinition struct + Fields []string + Name string + Options map[string]string + Sparse bool + Type string + Unique bool + type LoaderAdapter struct + func NewLoaderAdapter(registry *Registry, verbRegistry *verb.Registry) *LoaderAdapter + func (la *LoaderAdapter) LoadData(path string, value interface{}) error + func (la *LoaderAdapter) RegisterFunction(name string, fn interface{}) error + func (la *LoaderAdapter) RegisterType(name string, typeDef loader.TypeDefinition) error + func (la *LoaderAdapter) RegisterVerb(spec loader.VerbSpec, executor loader.VerbExecutor) error + func (la *LoaderAdapter) RegisterVerbDescriptor(spec loader.VerbSpec, descriptor loader.ExecutorDescriptor) error + type OutboxStore = workflow.OutboxStore + type PostgresOutboxStore struct + func NewPostgresOutboxStore(db *sql.DB) (*PostgresOutboxStore, error) + func (store *PostgresOutboxStore) AdmitExecution(ctx context.Context, admission DurableAdmission) (ExecutionRecord, bool, error) + func (store *PostgresOutboxStore) AdmitExecutionAtomic(ctx context.Context, admission DurableAdmission) (ExecutionRecord, bool, error) + func (store *PostgresOutboxStore) ClaimDispatch(ctx context.Context, options ClaimOptions) (*Dispatch, error) + func (store *PostgresOutboxStore) CompleteDispatch(ctx context.Context, completion Completion) error + func (store *PostgresOutboxStore) CompleteSaga(ctx context.Context, sagaID string) error + func (store *PostgresOutboxStore) CreateSaga(ctx context.Context, request CreateSagaRequest) (*SagaInstance, error) + func (store *PostgresOutboxStore) EnqueueStep(ctx context.Context, request EnqueueStepRequest) (*Dispatch, error) + func (store *PostgresOutboxStore) FinishExecutionLease(ctx context.Context, lease ExecutionLease, state ExecutionState, ...) error + func (store *PostgresOutboxStore) GetArtifact(ctx context.Context, digest string) (ExecutionArtifact, error) + func (store *PostgresOutboxStore) GetDispatch(ctx context.Context, dispatchID string) (*Dispatch, error) + func (store *PostgresOutboxStore) GetExecution(ctx context.Context, id string) (ExecutionRecord, error) + func (store *PostgresOutboxStore) GetExecutionByAdmission(ctx context.Context, identity string) (ExecutionRecord, error) + func (store *PostgresOutboxStore) GetSaga(ctx context.Context, sagaID string) (*SagaInstance, error) + func (store *PostgresOutboxStore) LeaseExecution(ctx context.Context, executionID, owner string, duration time.Duration) (ExecutionLease, error) + func (store *PostgresOutboxStore) LeaseExecutions(ctx context.Context, owner string, limit int, duration time.Duration) ([]ExecutionLease, error) + func (store *PostgresOutboxStore) ListAttempts(ctx context.Context, dispatchID string) ([]DispatchAttempt, error) + func (store *PostgresOutboxStore) ListDispatches(ctx context.Context, sagaID string) ([]*Dispatch, error) + func (store *PostgresOutboxStore) PutArtifact(ctx context.Context, artifact ExecutionArtifact) error + func (store *PostgresOutboxStore) RecoveryStats(ctx context.Context) (RecoveryStats, error) + func (store *PostgresOutboxStore) SaveFencingGrants(ctx context.Context, dispatchID string, attempt uint64, leaseToken string, ...) error + func (store *PostgresOutboxStore) SetExecutionState(ctx context.Context, id string, revision uint64, state ExecutionState, ...) (ExecutionRecord, error) + type PostgresSagaStore struct + func NewPostgresSagaStore(dsn string) (*PostgresSagaStore, error) + func (ps *PostgresSagaStore) CompleteSaga(sagaID string) error + func (ps *PostgresSagaStore) GetActiveSagas() ([]string, error) + func (ps *PostgresSagaStore) GetTransactionEffects(sagaID string) ([]*SagaEffect, error) + func (ps *PostgresSagaStore) MarkCompensated(sagaID, effectID string) error + func (ps *PostgresSagaStore) MarkFailed(sagaID, effectID string, reason error) error + func (ps *PostgresSagaStore) MarkSuccess(sagaID, effectID string, result interface{}) error + func (ps *PostgresSagaStore) RecordEffect(sagaID, effectID string, sequence int, verb string, ...) error + func (ps *PostgresSagaStore) StartTransaction(sagaID, ruleName string) error + type Predicate struct + Expression string + func (p *Predicate) Evaluate() (bool, error) + func (p *Predicate) EvaluateWithRegistry(registry *Registry) (bool, error) + type PrivacyRule struct + Action string + AllowedRoles []string + Conditions map[string]string + FieldPath string + MaskPattern string + type PruneOptions struct + BatchSize int + Before time.Time + DryRun bool + Retention time.Duration + type PruneReport struct + Artifacts int64 + ExecutionPlans int64 + Executions int64 + FactApplications int64 + FactSnapshots int64 + KafkaDeliveries int64 + RuleGenerations int64 + SagaAttempts int64 + SagaInstances int64 + SagaOutbox int64 + SagaSteps int64 + Sagas int64 + func PruneTerminalRecords(ctx context.Context, db *sql.DB, options PruneOptions) (PruneReport, error) + type PruneResult = PruneReport + type RecoveryStats = ledger.RecoveryStats + type RecoveryStatsReader = ledger.RecoveryStatsReader + type RedisOutboxStore struct + func NewRedisOutboxStore(options RedisOutboxStoreOptions) (*RedisOutboxStore, error) + func (store *RedisOutboxStore) ClaimDispatch(ctx context.Context, options ClaimOptions) (*Dispatch, error) + func (store *RedisOutboxStore) Close() error + func (store *RedisOutboxStore) CompleteDispatch(ctx context.Context, completion Completion) error + func (store *RedisOutboxStore) CompleteSaga(ctx context.Context, sagaID string) error + func (store *RedisOutboxStore) CreateSaga(ctx context.Context, request CreateSagaRequest) (*SagaInstance, error) + func (store *RedisOutboxStore) EnqueueStep(ctx context.Context, request EnqueueStepRequest) (*Dispatch, error) + func (store *RedisOutboxStore) GetDispatch(ctx context.Context, id string) (*Dispatch, error) + func (store *RedisOutboxStore) GetSaga(ctx context.Context, sagaID string) (*SagaInstance, error) + func (store *RedisOutboxStore) ListAttempts(ctx context.Context, id string) ([]DispatchAttempt, error) + func (store *RedisOutboxStore) ListDispatches(ctx context.Context, sagaID string) ([]*Dispatch, error) + func (store *RedisOutboxStore) OptimisticConflictRetries() uint64 + func (store *RedisOutboxStore) SaveFencingGrants(ctx context.Context, dispatchID string, attempt uint64, token string, ...) error + type RedisOutboxStoreOptions struct + Addr string + DB int + MaxLegacyBytes int + MaxRetries int + MaxSagaBytes int + Password string + Prefix string + TTL time.Duration + type RedisSagaStore struct + func NewRedisSagaStore(opts RedisSagaStoreOptions) (*RedisSagaStore, error) + func (rs *RedisSagaStore) Close() error + func (rs *RedisSagaStore) CompleteSaga(sagaID string) error + func (rs *RedisSagaStore) GetActiveSagas() ([]string, error) + func (rs *RedisSagaStore) GetTransactionEffects(sagaID string) ([]*SagaEffect, error) + func (rs *RedisSagaStore) MarkCompensated(sagaID, effectID string) error + func (rs *RedisSagaStore) MarkFailed(sagaID, effectID string, reason error) error + func (rs *RedisSagaStore) MarkSuccess(sagaID, effectID string, result interface{}) error + func (rs *RedisSagaStore) RecordEffect(sagaID, effectID string, sequence int, verb string, ...) error + func (rs *RedisSagaStore) StartTransaction(sagaID, ruleName string) error + type RedisSagaStoreOptions struct + Addr string + DB int + Password string + Prefix string + TTL time.Duration + type Registry struct + func NewRegistry() *Registry + func (r *Registry) Clear() + func (r *Registry) ClearAll() + func (r *Registry) CompileExpression(expression string) error + func (r *Registry) CompileLogicalExpression(expression string, schemaInfo effectus.SchemaInfo) ([]*Predicate, map[string]struct{}, error) + func (r *Registry) EvaluateBoolean(expression string) (bool, error) + func (r *Registry) EvaluateCompiled(expression string) (interface{}, error) + func (r *Registry) EvaluateExpression(expression string) (interface{}, error) + func (r *Registry) EvaluatePredicates(predicates []*Predicate, facts effectus.Facts) bool + func (r *Registry) Get(path string) (interface{}, bool) + func (r *Registry) GetPathsWithPrefix(prefix string) []string + func (r *Registry) GetType(path string) (interface{}, bool) + func (r *Registry) LoadFromFacts(facts effectus.Facts) + func (r *Registry) LoadFromJSON(jsonData []byte) error + func (r *Registry) LoadFromMap(data map[string]interface{}) + func (r *Registry) Merge(other *Registry) + func (r *Registry) NewPredicate(expression string) (*Predicate, error) + func (r *Registry) RegisterFunction(name string, fn interface{}) + func (r *Registry) Set(path string, value interface{}) + func (r *Registry) SetClock(clock func() time.Time) + func (r *Registry) SetNow(now time.Time) + func (r *Registry) TypeCheckExpression(expression string) error + type RetentionPolicy struct + Conditions map[string]string + Duration string + Strategy string + type SagaEffect struct + Args map[string]interface{} + Error string + ID string + Result interface{} + Sequence int + Status string + Timestamp time.Time + Verb string + func GetSagaEffect(store SagaStore, sagaID, effectID string) (*SagaEffect, error) + type SagaExecutor struct + func NewSagaExecutor(executor effectus.Executor, sagaStore SagaStore, ...) *SagaExecutor + func (se *SagaExecutor) ExecuteWithSaga(ctx context.Context, sagaID string, ruleName string, effects []effectus.Effect) ([]interface{}, error) + type SagaInstance = workflow.SagaInstance + type SagaState = workflow.SagaState + type SagaStep = workflow.SagaStep + type SagaStore interface + CompleteSaga func(sagaID string) error + GetActiveSagas func() ([]string, error) + GetTransactionEffects func(sagaID string) ([]*SagaEffect, error) + MarkCompensated func(sagaID, effectID string) error + MarkFailed func(sagaID, effectID string, reason error) error + MarkSuccess func(sagaID, effectID string, result interface{}) error + RecordEffect func(sagaID, effectID string, sequence int, verb string, ...) error + StartTransaction func(sagaID, ruleName string) error + type SagaVerbRegistry interface + GetVerb func(name string) (*verb.Spec, bool) + type SchemaValidationResult struct + BreakingChanges []string + Errors []string + Suggestions []string + Valid bool + Warnings []string + type StepState = workflow.StepState + type UnknownOutcomeRetryPolicy = workflow.UnknownOutcomeRetryPolicy + type VerbSchema struct + BufCommit string + BufModule string + Compensatable bool + CreatedAt time.Time + Description string + ExecutionType string + Idempotent bool + InputSchema map[string]interface{} + Name string + OutputSchema map[string]interface{} + RequiredCapabilities []string + UpdatedAt time.Time + Version string + type VerbSchemaRegistry struct
Go to main page