schema

package
v0.3.0 Latest Latest
Warning

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

 Go to latest Published: Sep 1, 2026 License: MIT Imports: 43 Imported by: 0

Documentation

Overview

Package schema provides saga transaction management with capability-based locking

Index

Constants

View Source 
const (
	ExecutionAdmitting           = ledger.ExecutionAdmitting
	ExecutionAccepted            = ledger.ExecutionAccepted
	ExecutionRunning             = ledger.ExecutionRunning
	ExecutionCompleted           = ledger.ExecutionCompleted
	ExecutionFailed              = ledger.ExecutionFailed
	ExecutionBlockedUnknown      = ledger.ExecutionBlockedUnknown
	ExecutionBlockedFence        = ledger.ExecutionBlockedFence
	ExecutionBlockedDependency   = ledger.ExecutionBlockedDependency
	ExecutionBlockedCompensation = ledger.ExecutionBlockedCompensation
)
View Source 
const (
	SagaRunning             = workflow.SagaRunning
	SagaCompleted           = workflow.SagaCompleted
	SagaCompensating        = workflow.SagaCompensating
	SagaCompensated         = workflow.SagaCompensated
	SagaFailed              = workflow.SagaFailed
	SagaBlockedUnknown      = workflow.SagaBlockedUnknown
	SagaBlockedDependency   = workflow.SagaBlockedDependency
	SagaBlockedFence        = workflow.SagaBlockedFence
	SagaBlockedCompensation = workflow.SagaBlockedCompensation
	DispatchQueued          = workflow.DispatchQueued
	DispatchInFlight        = workflow.DispatchInFlight
	DispatchSucceeded       = workflow.DispatchSucceeded
	DispatchRetryWait       = workflow.DispatchRetryWait
	DispatchFailedPermanent = workflow.DispatchFailedPermanent
	DispatchBlockedUnknown  = workflow.DispatchBlockedUnknown
	DispatchBlockedFence    = workflow.DispatchBlockedFence
	StepPending             = workflow.StepPending
	StepSucceeded           = workflow.StepSucceeded
	StepFailed              = workflow.StepFailed
	StepCompensated         = workflow.StepCompensated
)
View Source 
const (
	SagaEffectPending     = "pending"
	SagaEffectSuccess     = "success"
	SagaEffectFailed      = "failed"
	SagaEffectCompensated = "compensated"
)

Variables

View Source 
var (
	ErrArtifactNotFound    = errors.New("execution artifact not found")
	ErrExecutionNotFound   = errors.New("execution not found")
	ErrStaleExecutionLease = errors.New("stale execution recovery lease")
)
View Source 
var (
	ErrIdentityConflict   = errors.New("durable identity conflict")
	ErrNoDispatch         = errors.New("no eligible dispatch")
	ErrStaleLease         = errors.New("stale dispatch lease")
	ErrTerminalSaga       = errors.New("terminal saga cannot be reopened")
	ErrInvalidTransition  = errors.New("invalid saga state transition")
	ErrOptimisticConflict = errors.New("optimistic persistence conflict")
)
View Source 
var ErrSagaStoreRequired = errors.New("saga execution requires a saga store")

ErrSagaStoreRequired reports an invalid saga execution configuration.

Functions

func CanonicalJSON 

func CanonicalJSON(value any) (json.RawMessage, string, error)

CanonicalJSON encodes JSON-compatible data and rejects lossy or unsupported values.

func CreateExtensionManagerWithDefaults 

func CreateExtensionManagerWithDefaults(extensionDirs ...string) (*loader.ExtensionManager, error)

CreateExtensionManagerWithDefaults creates an ExtensionManager with directory scanning

func EvaluatePredicatesWithFacts 

func EvaluatePredicatesWithFacts(predicates []*Predicate, facts effectus.Facts) bool

EvaluatePredicatesWithFacts evaluates predicates without mutating compiled state.

func EvaluatePredicatesWithFactsE 

func EvaluatePredicatesWithFactsE(predicates []*Predicate, facts effectus.Facts) (bool, error)

EvaluatePredicatesWithFactsE distinguishes a false predicate from an evaluation error.

func ExtractFactPaths 

func ExtractFactPaths(expression string) map[string]struct{}

ExtractFactPaths returns fact paths referenced by an expression.

func IdempotencyKey 

func IdempotencyKey(namespace, sagaID, effectID string, direction invocation.Direction) string

IdempotencyKey returns a stable key. An attempt is deliberately not an input.

func IsTerminalExecutionState 

func IsTerminalExecutionState(state ExecutionState) bool

func LoadExtensionsIntoRegistries 

func LoadExtensionsIntoRegistries(em *loader.ExtensionManager, registry *Registry, verbRegistry *verb.Registry) error

LoadExtensionsIntoRegistries loads extensions from an ExtensionManager into registries

func MigrateRedisOutboxV2 

func MigrateRedisOutboxV2(ctx context.Context, options RedisOutboxStoreOptions) error

MigrateRedisOutboxV2 imports the former global document idempotently. The legacy key is retained as an operator-controlled backup.

func MigrateSagaV2 

func MigrateSagaV2(ctx context.Context, db *sql.DB) error

MigrateSagaV2 applies the versioned durable runtime migrations. Production deployments should use a separate DDL credential and validate at runtime.

func ResetDefaultClock 

func ResetDefaultClock()

ResetDefaultClock restores the default clock to time.Now.

func SagaEffectID 

func SagaEffectID(sequence int) string

SagaEffectID returns the stable identity for an effect's source-order position.

func SetDefaultClock 

func SetDefaultClock(clock func() time.Time)

SetDefaultClock overrides the default clock used by new registries.

func SetFixedTime 

func SetFixedTime(now time.Time)

SetFixedTime pins registry time functions to a fixed timestamp.

func StableAdmissionID 

func StableAdmissionID(namespace, deliveryID, ruleset, version string) string

Durable workflow contracts live in schema/workflow. Compatibility aliases remain in contracts.go during the package-boundary migration. StableAdmissionID scopes a transport idempotency key to its checked ruleset.

func StableExecutionID 

func StableExecutionID(namespace, deliveryID, ruleset, version string) string

StableExecutionID derives an execution identity from admission identity.

func StableSagaID 

func StableSagaID(executionID, planID string) string

StableSagaID derives one plan saga identity from an execution.

func ValidateSagaV2 

func ValidateSagaV2(ctx context.Context, db *sql.DB) error

ValidateSagaV2 verifies that every embedded durable runtime migration is applied. It executes SELECT statements only and is safe for a DML role.

Types

type AtomicAdmissionStore 

type AtomicAdmissionStore = ledger.AtomicAdmissionStore

type BufBreakingConfig 

type BufBreakingConfig struct {
	Use []string `yaml:"use"`
}

BufBreakingConfig configures breaking change detection

type BufBuildConfig 

type BufBuildConfig struct {
	Excludes []string `yaml:"excludes"`
}

BufBuildConfig configures build settings

type BufConfig 

type BufConfig struct {
	Version  string            `yaml:"version"`
	Name     string            `yaml:"name"`
	Deps     []string          `yaml:"deps"`
	Breaking BufBreakingConfig `yaml:"breaking"`
	Lint     BufLintConfig     `yaml:"lint"`
	Build    BufBuildConfig    `yaml:"build"`
}

BufConfig represents the buf.yaml configuration

type BufGenConfig 

type BufGenConfig struct {
	Version string            `yaml:"version"`
	Managed BufManagedConfig  `yaml:"managed"`
	Plugins []BufPluginConfig `yaml:"plugins"`
}

BufGenConfig represents the buf.gen.yaml configuration

type BufGoPackagePrefixConfig 

type BufGoPackagePrefixConfig struct {
	Default string   `yaml:"default"`
	Except  []string `yaml:"except"`
}

BufGoPackagePrefixConfig configures Go package prefixes

type BufIntegration 

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

BufIntegration manages protobuf schemas and code generation using Buf

func NewBufIntegration 

func NewBufIntegration(workspaceRoot string) (*BufIntegration, error)

NewBufIntegration creates a new Buf integration service

func (*BufIntegration) GenerateCode 

func (b *BufIntegration) GenerateCode(ctx context.Context) (*CodeGenerationResult, error)

GenerateCode generates code for all registered schemas

func (*BufIntegration) GetFactSchema 

func (b *BufIntegration) GetFactSchema(name string) (*FactSchema, bool)

GetFactSchema retrieves a fact schema by name

func (*BufIntegration) GetVerbSchema 

func (b *BufIntegration) GetVerbSchema(name string) (*VerbSchema, bool)

GetVerbSchema retrieves a verb schema by name

func (*BufIntegration) ListFactSchemas 

func (b *BufIntegration) ListFactSchemas() map[string]*FactSchema

ListFactSchemas returns all registered fact schemas

func (*BufIntegration) ListVerbSchemas 

func (b *BufIntegration) ListVerbSchemas() map[string]*VerbSchema

ListVerbSchemas returns all registered verb schemas

func (*BufIntegration) RegisterFactSchema 

func (b *BufIntegration) RegisterFactSchema(ctx context.Context, schema *FactSchema) error

RegisterFactSchema registers a new fact schema

func (*BufIntegration) RegisterVerbSchema 

func (b *BufIntegration) RegisterVerbSchema(ctx context.Context, schema *VerbSchema) error

RegisterVerbSchema registers a new verb interface schema

func (*BufIntegration) ValidateSchemas 

func (b *BufIntegration) ValidateSchemas(ctx context.Context) (*SchemaValidationResult, error)

ValidateSchemas validates all registered schemas for compatibility

type BufLintConfig 

type BufLintConfig struct {
	Use                 []string `yaml:"use"`
	Except              []string `yaml:"except"`
	AllowCommentIgnores bool     `yaml:"allow_comment_ignores"`
}

BufLintConfig configures linting

type BufManagedConfig 

type BufManagedConfig struct {
	Enabled         bool                     `yaml:"enabled"`
	GoPackagePrefix BufGoPackagePrefixConfig `yaml:"go_package_prefix"`
}

BufManagedConfig configures managed mode

type BufPluginConfig 

type BufPluginConfig struct {
	Plugin string   `yaml:"plugin"`
	Out    string   `yaml:"out"`
	Opt    []string `yaml:"opt"`
}

BufPluginConfig configures a code generation plugin

type CheckedEnqueueRequest 

type CheckedEnqueueRequest struct {
	SagaID                string
	PlanID                string
	EffectID              string
	Facts                 map[string]any
	ResultSlots           []any
	Arguments             map[string]any // Deprecated: accepted only when identical to resolved checked arguments.
	CompensationVerb      string
	CompensationContract  string
	CompensationArguments map[string]any
	Fencing               []FencingRequirement
}

CheckedEnqueueRequest supplies runtime facts, prior result slots, and deployment metadata. The checked artifact supplies effect identity, order, verb, contract hash, and argument expressions.

type ClaimOptions 

type ClaimOptions = workflow.ClaimOptions

type CodeGenerationResult 

type CodeGenerationResult struct {
	Success        bool                   `json:"success"`
	GeneratedFiles []string               `json:"generated_files"`
	Errors         []string               `json:"errors"`
	Warnings       []string               `json:"warnings"`
	Duration       time.Duration          `json:"duration"`
	Metadata       map[string]interface{} `json:"metadata"`
}

CodeGenerationResult represents the result of code generation

type Completion 

type Completion = workflow.Completion

type CreateSagaRequest 

type CreateSagaRequest = workflow.CreateSagaRequest

type Dispatch 

type Dispatch = workflow.Dispatch

func EnqueueCheckedStep 

func EnqueueCheckedStep(ctx context.Context, store OutboxStore, checked *ir.Checked, request CheckedEnqueueRequest) (*Dispatch, error)

EnqueueCheckedStep resolves arguments from the exact checked plan before it creates durable intent. It does not invoke an executor.

type DispatchAttempt 

type DispatchAttempt = workflow.DispatchAttempt

type DispatchState 

type DispatchState = workflow.DispatchState

type Dispatcher 

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

Dispatcher claims committed intents, persists fencing grants, then invokes.

func NewDispatcher 

func NewDispatcher(store OutboxStore, provider fencing.Provider, executor invocation.Executor, options DispatcherOptions) (*Dispatcher, error)

func (*Dispatcher) Dispatch 

func (dispatcher *Dispatcher) Dispatch(ctx context.Context, targetDispatchID string) (*Dispatch, error)

Dispatch claims only targetDispatchID when it is non-empty.

func (*Dispatcher) DispatchOne 

func (dispatcher *Dispatcher) DispatchOne(ctx context.Context) (*Dispatch, error)

DispatchOne handles at most one eligible intent. External invocation occurs only after ClaimDispatch and SaveFencingGrants commit.

type DispatcherOptions 

type DispatcherOptions struct {
	Owner                 string
	RequestID             string
	LeaseDuration         time.Duration
	InvocationTimeout     time.Duration
	MaxAttempts           uint64
	InitialBackoff        time.Duration
	MaxBackoff            time.Duration
	RequireDurableFencing bool
}

DispatcherOptions control one durable dispatch worker.

type DurableAdmission 

type DurableAdmission = ledger.DurableAdmission

type EnqueueStepRequest 

type EnqueueStepRequest = workflow.EnqueueStepRequest

type ExecutionArtifact 

type ExecutionArtifact = ledger.ExecutionArtifact

type ExecutionLease 

type ExecutionLease = ledger.ExecutionLease

type ExecutionLedger 

type ExecutionLedger = ledger.ExecutionLedger

type ExecutionPlan 

type ExecutionPlan struct {
	Steps []*ExecutionStep
}

ExecutionPlan represents a plan for executing effects with proper ordering

type ExecutionPlanRecord 

type ExecutionPlanRecord = ledger.ExecutionPlanRecord

type ExecutionRecord 

type ExecutionRecord = ledger.ExecutionRecord

type ExecutionState 

type ExecutionState = ledger.ExecutionState

Execution-ledger compatibility aliases. New runtime code imports schema/ledger.

type ExecutionStep 

type ExecutionStep struct {
	Sequence           int
	Effects            []effectus.Effect
	CanRunConcurrently bool
}

ExecutionStep represents a step in the execution plan

type FactApplication 

type FactApplication = ledger.FactApplication

type FactSchema 

type FactSchema struct {
	Name            string                 `json:"name"`
	Version         string                 `json:"version"`
	Description     string                 `json:"description"`
	Schema          map[string]interface{} `json:"schema"`
	Indexes         []IndexDefinition      `json:"indexes"`
	RetentionPolicy *RetentionPolicy       `json:"retention_policy"`
	PrivacyRules    []PrivacyRule          `json:"privacy_rules"`
	BufModule       string                 `json:"buf_module"`
	BufCommit       string                 `json:"buf_commit"`
	CreatedAt       time.Time              `json:"created_at"`
	UpdatedAt       time.Time              `json:"updated_at"`
}

FactSchema represents a versioned fact schema

type FactSchemaRegistry 

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

FactSchemaRegistry manages fact schemas

type FencingRequirement 

type FencingRequirement = workflow.FencingRequirement

type InMemoryExecutionLedger 

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

InMemoryExecutionLedger provides deterministic development semantics. It is not durable and cannot make admission atomic with a separate OutboxStore.

func NewInMemoryExecutionLedger 

func NewInMemoryExecutionLedger() *InMemoryExecutionLedger

func (*InMemoryExecutionLedger) AdmitExecution 

func (store *InMemoryExecutionLedger) AdmitExecution(ctx context.Context, admission DurableAdmission) (ExecutionRecord, bool, error)

func (*InMemoryExecutionLedger) FinishExecutionLease 

func (store *InMemoryExecutionLedger) FinishExecutionLease(ctx context.Context, lease ExecutionLease, state ExecutionState, message string) error

func (*InMemoryExecutionLedger) GetArtifact 

func (store *InMemoryExecutionLedger) GetArtifact(ctx context.Context, digest string) (ExecutionArtifact, error)

func (*InMemoryExecutionLedger) GetExecution 

func (store *InMemoryExecutionLedger) GetExecution(ctx context.Context, id string) (ExecutionRecord, error)

func (*InMemoryExecutionLedger) GetExecutionByAdmission 

func (store *InMemoryExecutionLedger) GetExecutionByAdmission(ctx context.Context, identity string) (ExecutionRecord, error)

func (*InMemoryExecutionLedger) LeaseExecutions 

func (store *InMemoryExecutionLedger) LeaseExecutions(ctx context.Context, owner string, limit int, duration time.Duration) ([]ExecutionLease, error)

func (*InMemoryExecutionLedger) PutArtifact 

func (store *InMemoryExecutionLedger) PutArtifact(ctx context.Context, artifact ExecutionArtifact) error

func (*InMemoryExecutionLedger) SetExecutionState 

func (store *InMemoryExecutionLedger) SetExecutionState(ctx context.Context, id string, revision uint64, state ExecutionState, message string) (ExecutionRecord, error)

type InMemoryOutboxStore 

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

InMemoryOutboxStore implements the V2 protocol for tests and single-process development. Its data and fencing counters do not survive a process restart.

func NewInMemoryOutboxStore 

func NewInMemoryOutboxStore() *InMemoryOutboxStore

func (*InMemoryOutboxStore) ClaimDispatch 

func (store *InMemoryOutboxStore) ClaimDispatch(ctx context.Context, options ClaimOptions) (*Dispatch, error)

func (*InMemoryOutboxStore) CompleteDispatch 

func (store *InMemoryOutboxStore) CompleteDispatch(ctx context.Context, completion Completion) error

func (*InMemoryOutboxStore) CompleteSaga 

func (store *InMemoryOutboxStore) CompleteSaga(ctx context.Context, sagaID string) error

func (*InMemoryOutboxStore) CreateSaga 

func (store *InMemoryOutboxStore) CreateSaga(ctx context.Context, request CreateSagaRequest) (*SagaInstance, error)

func (*InMemoryOutboxStore) EnqueueStep 

func (store *InMemoryOutboxStore) EnqueueStep(ctx context.Context, request EnqueueStepRequest) (*Dispatch, error)

func (*InMemoryOutboxStore) GetDispatch 

func (store *InMemoryOutboxStore) GetDispatch(ctx context.Context, dispatchID string) (*Dispatch, error)

func (*InMemoryOutboxStore) GetSaga 

func (store *InMemoryOutboxStore) GetSaga(ctx context.Context, sagaID string) (*SagaInstance, error)

func (*InMemoryOutboxStore) ListAttempts 

func (store *InMemoryOutboxStore) ListAttempts(ctx context.Context, dispatchID string) ([]DispatchAttempt, error)

func (*InMemoryOutboxStore) ListDispatches 

func (store *InMemoryOutboxStore) ListDispatches(ctx context.Context, sagaID string) ([]*Dispatch, error)

func (*InMemoryOutboxStore) SaveFencingGrants 

func (store *InMemoryOutboxStore) SaveFencingGrants(ctx context.Context, dispatchID string, attempt uint64, leaseToken string, grants []invocation.FencingGrant) error

type InMemorySagaStore 

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

InMemorySagaStore provides an in-memory implementation of SagaStore for testing.

func NewInMemorySagaStore 

func NewInMemorySagaStore() *InMemorySagaStore

NewInMemorySagaStore creates a new in-memory saga store.

func (*InMemorySagaStore) CompleteSaga 

func (ims *InMemorySagaStore) CompleteSaga(sagaID string) error

CompleteSaga implements SagaStore.

func (*InMemorySagaStore) GetActiveSagas 

func (ims *InMemorySagaStore) GetActiveSagas() ([]string, error)

GetActiveSagas implements SagaStore.

func (*InMemorySagaStore) GetTransactionEffects 

func (ims *InMemorySagaStore) GetTransactionEffects(sagaID string) ([]*SagaEffect, error)

GetTransactionEffects implements SagaStore.

func (*InMemorySagaStore) MarkCompensated 

func (ims *InMemorySagaStore) MarkCompensated(sagaID, effectID string) error

MarkCompensated implements SagaStore.

func (*InMemorySagaStore) MarkFailed 

func (ims *InMemorySagaStore) MarkFailed(sagaID, effectID string, reason error) error

MarkFailed implements SagaStore.

func (*InMemorySagaStore) MarkSuccess 

func (ims *InMemorySagaStore) MarkSuccess(sagaID, effectID string, result interface{}) error

MarkSuccess implements SagaStore.

func (*InMemorySagaStore) RecordEffect 

func (ims *InMemorySagaStore) RecordEffect(sagaID, effectID string, sequence int, verb string, args map[string]interface{}) error

RecordEffect implements SagaStore.

func (*InMemorySagaStore) StartTransaction 

func (ims *InMemorySagaStore) StartTransaction(sagaID, ruleName string) error

StartTransaction implements SagaStore. Reopening a saga preserves its effect log.

type IndexDefinition 

type IndexDefinition struct {
	Name    string            `json:"name"`
	Fields  []string          `json:"fields"`
	Type    string            `json:"type"`
	Unique  bool              `json:"unique"`
	Sparse  bool              `json:"sparse"`
	Options map[string]string `json:"options"`
}

IndexDefinition defines an index on fact data

type LoaderAdapter 

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

LoaderAdapter adapts the ExtensionManager to work with existing registries

func NewLoaderAdapter 

func NewLoaderAdapter(registry *Registry, verbRegistry *verb.Registry) *LoaderAdapter

NewLoaderAdapter creates a new adapter for the extension system

func (*LoaderAdapter) LoadData 

func (la *LoaderAdapter) LoadData(path string, value interface{}) error

LoadData implements loader.LoadTarget for data loading

func (*LoaderAdapter) RegisterFunction 

func (la *LoaderAdapter) RegisterFunction(name string, fn interface{}) error

RegisterFunction implements loader.LoadTarget for function registration

func (*LoaderAdapter) RegisterType 

func (la *LoaderAdapter) RegisterType(name string, typeDef loader.TypeDefinition) error

RegisterType implements loader.LoadTarget for type registration

func (*LoaderAdapter) RegisterVerb 

func (la *LoaderAdapter) RegisterVerb(spec loader.VerbSpec, executor loader.VerbExecutor) error

RegisterVerb implements loader.LoadTarget for verb registration

func (*LoaderAdapter) RegisterVerbDescriptor 

func (la *LoaderAdapter) RegisterVerbDescriptor(spec loader.VerbSpec, descriptor loader.ExecutorDescriptor) error

RegisterVerbDescriptor records a transport contract without constructing a client. The checked runtime materializes the descriptor after compilation.

type OutboxStore 

type OutboxStore = workflow.OutboxStore

type PostgresOutboxStore 

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

PostgresOutboxStore implements the V2 saga protocol. It never creates or alters tables at startup. Apply schema/migrations before construction.

func NewPostgresOutboxStore 

func NewPostgresOutboxStore(db *sql.DB) (*PostgresOutboxStore, error)

func (*PostgresOutboxStore) AdmitExecution 

func (store *PostgresOutboxStore) AdmitExecution(ctx context.Context, admission DurableAdmission) (ExecutionRecord, bool, error)

func (*PostgresOutboxStore) AdmitExecutionAtomic 

func (store *PostgresOutboxStore) AdmitExecutionAtomic(ctx context.Context, admission DurableAdmission) (ExecutionRecord, bool, error)

func (*PostgresOutboxStore) ClaimDispatch 

func (store *PostgresOutboxStore) ClaimDispatch(ctx context.Context, options ClaimOptions) (*Dispatch, error)

func (*PostgresOutboxStore) CompleteDispatch 

func (store *PostgresOutboxStore) CompleteDispatch(ctx context.Context, completion Completion) error

func (*PostgresOutboxStore) CompleteSaga 

func (store *PostgresOutboxStore) CompleteSaga(ctx context.Context, sagaID string) error

func (*PostgresOutboxStore) CreateSaga 

func (store *PostgresOutboxStore) CreateSaga(ctx context.Context, request CreateSagaRequest) (*SagaInstance, error)

func (*PostgresOutboxStore) EnqueueStep 

func (store *PostgresOutboxStore) EnqueueStep(ctx context.Context, request EnqueueStepRequest) (*Dispatch, error)

func (*PostgresOutboxStore) FinishExecutionLease 

func (store *PostgresOutboxStore) FinishExecutionLease(ctx context.Context, lease ExecutionLease, state ExecutionState, message string) error

func (*PostgresOutboxStore) GetArtifact 

func (store *PostgresOutboxStore) GetArtifact(ctx context.Context, digest string) (ExecutionArtifact, error)

func (*PostgresOutboxStore) GetDispatch 

func (store *PostgresOutboxStore) GetDispatch(ctx context.Context, dispatchID string) (*Dispatch, error)

func (*PostgresOutboxStore) GetExecution 

func (store *PostgresOutboxStore) GetExecution(ctx context.Context, id string) (ExecutionRecord, error)

func (*PostgresOutboxStore) GetExecutionByAdmission 

func (store *PostgresOutboxStore) GetExecutionByAdmission(ctx context.Context, identity string) (ExecutionRecord, error)

func (*PostgresOutboxStore) GetSaga 

func (store *PostgresOutboxStore) GetSaga(ctx context.Context, sagaID string) (*SagaInstance, error)

func (*PostgresOutboxStore) LeaseExecution 

func (store *PostgresOutboxStore) LeaseExecution(ctx context.Context, executionID, owner string, duration time.Duration) (ExecutionLease, error)

LeaseExecution leases one known execution without contending with unrelated recovery work. It is useful for targeted operator recovery and deterministic integration checks.

func (*PostgresOutboxStore) LeaseExecutions 

func (store *PostgresOutboxStore) LeaseExecutions(ctx context.Context, owner string, limit int, duration time.Duration) ([]ExecutionLease, error)

func (*PostgresOutboxStore) ListAttempts 

func (store *PostgresOutboxStore) ListAttempts(ctx context.Context, dispatchID string) ([]DispatchAttempt, error)

func (*PostgresOutboxStore) ListDispatches 

func (store *PostgresOutboxStore) ListDispatches(ctx context.Context, sagaID string) ([]*Dispatch, error)

func (*PostgresOutboxStore) PutArtifact 

func (store *PostgresOutboxStore) PutArtifact(ctx context.Context, artifact ExecutionArtifact) error

func (*PostgresOutboxStore) RecoveryStats 

func (store *PostgresOutboxStore) RecoveryStats(ctx context.Context) (RecoveryStats, error)

func (*PostgresOutboxStore) SaveFencingGrants 

func (store *PostgresOutboxStore) SaveFencingGrants(ctx context.Context, dispatchID string, attempt uint64, leaseToken string, grants []invocation.FencingGrant) error

func (*PostgresOutboxStore) SetExecutionState 

func (store *PostgresOutboxStore) SetExecutionState(ctx context.Context, id string, revision uint64, state ExecutionState, message string) (ExecutionRecord, error)

type PostgresSagaStore 

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

PostgresSagaStore persists saga effects in Postgres.

func NewPostgresSagaStore 

func NewPostgresSagaStore(dsn string) (*PostgresSagaStore, error)

NewPostgresSagaStore creates a Postgres-backed saga store using the provided DSN.

func (*PostgresSagaStore) CompleteSaga 

func (ps *PostgresSagaStore) CompleteSaga(sagaID string) error

func (*PostgresSagaStore) GetActiveSagas 

func (ps *PostgresSagaStore) GetActiveSagas() ([]string, error)

func (*PostgresSagaStore) GetTransactionEffects 

func (ps *PostgresSagaStore) GetTransactionEffects(sagaID string) ([]*SagaEffect, error)

func (*PostgresSagaStore) MarkCompensated 

func (ps *PostgresSagaStore) MarkCompensated(sagaID, effectID string) error

func (*PostgresSagaStore) MarkFailed 

func (ps *PostgresSagaStore) MarkFailed(sagaID, effectID string, reason error) error

func (*PostgresSagaStore) MarkSuccess 

func (ps *PostgresSagaStore) MarkSuccess(sagaID, effectID string, result interface{}) error

func (*PostgresSagaStore) RecordEffect 

func (ps *PostgresSagaStore) RecordEffect(sagaID, effectID string, sequence int, verb string, args map[string]interface{}) error

func (*PostgresSagaStore) StartTransaction 

func (ps *PostgresSagaStore) StartTransaction(sagaID, ruleName string) error

type Predicate 

type Predicate struct {
	Expression string
	// contains filtered or unexported fields
}

Predicate represents a compiled predicate expression

func (*Predicate) Evaluate 

func (p *Predicate) Evaluate() (bool, error)

Evaluate evaluates the predicate against the registry that compiled it.

func (*Predicate) EvaluateWithRegistry 

func (p *Predicate) EvaluateWithRegistry(registry *Registry) (bool, error)

EvaluateWithRegistry evaluates a compiled predicate without mutating it.

type PrivacyRule 

type PrivacyRule struct {
	FieldPath    string            `json:"field_path"`
	Action       string            `json:"action"`
	MaskPattern  string            `json:"mask_pattern"`
	AllowedRoles []string          `json:"allowed_roles"`
	Conditions   map[string]string `json:"conditions"`
}

PrivacyRule defines privacy and masking rules

type PruneOptions 

type PruneOptions struct {
	Before    time.Time
	Retention time.Duration
	BatchSize int
	DryRun    bool
}

PruneOptions bounds one durable-record retention pass.

type PruneReport 

type PruneReport struct {
	Executions       int64 `json:"executions"`
	ExecutionPlans   int64 `json:"execution_plans"`
	FactApplications int64 `json:"fact_applications"`
	FactSnapshots    int64 `json:"fact_snapshots"`
	SagaInstances    int64 `json:"saga_instances"`
	// Sagas is a compatibility alias for SagaInstances.
	Sagas           int64 `json:"-"`
	SagaSteps       int64 `json:"saga_steps"`
	SagaOutbox      int64 `json:"saga_outbox"`
	SagaAttempts    int64 `json:"saga_attempts"`
	RuleGenerations int64 `json:"rule_generations"`
	Artifacts       int64 `json:"artifacts"`
	KafkaDeliveries int64 `json:"kafka_deliveries"`
}

PruneReport reports candidate or deleted rows by durable table.

func PruneTerminalRecords 

func PruneTerminalRecords(ctx context.Context, db *sql.DB, options PruneOptions) (PruneReport, error)

PruneTerminalRecords removes only old terminal execution graphs. Blocked, admitting, running, leased, retrying, and unacknowledged poison state is never selected. Deletions run in FK order in one bounded transaction.

type PruneResult 

type PruneResult = PruneReport

PruneResult is retained for compatibility with the retention-based API.

type RecoveryStats 

type RecoveryStats = ledger.RecoveryStats

type RecoveryStatsReader 

type RecoveryStatsReader = ledger.RecoveryStatsReader

type RedisOutboxStore 

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

RedisOutboxStore keeps one bounded document per saga. Documents, the ready index, and dispatch lookup keys share one Redis Cluster hash tag. Every mutation is a Lua compare-and-swap and never rewrites unrelated sagas.

func NewRedisOutboxStore 

func NewRedisOutboxStore(options RedisOutboxStoreOptions) (*RedisOutboxStore, error)

func (*RedisOutboxStore) ClaimDispatch 

func (store *RedisOutboxStore) ClaimDispatch(ctx context.Context, options ClaimOptions) (*Dispatch, error)

func (*RedisOutboxStore) Close 

func (store *RedisOutboxStore) Close() error

func (*RedisOutboxStore) CompleteDispatch 

func (store *RedisOutboxStore) CompleteDispatch(ctx context.Context, completion Completion) error

func (*RedisOutboxStore) CompleteSaga 

func (store *RedisOutboxStore) CompleteSaga(ctx context.Context, sagaID string) error

func (*RedisOutboxStore) CreateSaga 

func (store *RedisOutboxStore) CreateSaga(ctx context.Context, request CreateSagaRequest) (*SagaInstance, error)

func (*RedisOutboxStore) EnqueueStep 

func (store *RedisOutboxStore) EnqueueStep(ctx context.Context, request EnqueueStepRequest) (*Dispatch, error)

func (*RedisOutboxStore) GetDispatch 

func (store *RedisOutboxStore) GetDispatch(ctx context.Context, id string) (*Dispatch, error)

func (*RedisOutboxStore) GetSaga 

func (store *RedisOutboxStore) GetSaga(ctx context.Context, sagaID string) (*SagaInstance, error)

func (*RedisOutboxStore) ListAttempts 

func (store *RedisOutboxStore) ListAttempts(ctx context.Context, id string) ([]DispatchAttempt, error)

func (*RedisOutboxStore) ListDispatches 

func (store *RedisOutboxStore) ListDispatches(ctx context.Context, sagaID string) ([]*Dispatch, error)

func (*RedisOutboxStore) OptimisticConflictRetries 

func (store *RedisOutboxStore) OptimisticConflictRetries() uint64

func (*RedisOutboxStore) SaveFencingGrants 

func (store *RedisOutboxStore) SaveFencingGrants(ctx context.Context, dispatchID string, attempt uint64, token string, grants []invocation.FencingGrant) error

type RedisOutboxStoreOptions 

type RedisOutboxStoreOptions struct {
	Addr           string
	Password       string
	DB             int
	Prefix         string
	TTL            time.Duration
	MaxRetries     int
	MaxSagaBytes   int
	MaxLegacyBytes int
}

type RedisSagaStore 

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

RedisSagaStore persists saga effects in Redis.

func NewRedisSagaStore 

func NewRedisSagaStore(opts RedisSagaStoreOptions) (*RedisSagaStore, error)

NewRedisSagaStore creates a Redis-backed saga store.

func (*RedisSagaStore) Close 

func (rs *RedisSagaStore) Close() error

Close releases the Redis client resources.

func (*RedisSagaStore) CompleteSaga 

func (rs *RedisSagaStore) CompleteSaga(sagaID string) error

func (*RedisSagaStore) GetActiveSagas 

func (rs *RedisSagaStore) GetActiveSagas() ([]string, error)

func (*RedisSagaStore) GetTransactionEffects 

func (rs *RedisSagaStore) GetTransactionEffects(sagaID string) ([]*SagaEffect, error)

func (*RedisSagaStore) MarkCompensated 

func (rs *RedisSagaStore) MarkCompensated(sagaID, effectID string) error

func (*RedisSagaStore) MarkFailed 

func (rs *RedisSagaStore) MarkFailed(sagaID, effectID string, reason error) error

func (*RedisSagaStore) MarkSuccess 

func (rs *RedisSagaStore) MarkSuccess(sagaID, effectID string, result interface{}) error

func (*RedisSagaStore) RecordEffect 

func (rs *RedisSagaStore) RecordEffect(sagaID, effectID string, sequence int, verb string, args map[string]interface{}) error

func (*RedisSagaStore) StartTransaction 

func (rs *RedisSagaStore) StartTransaction(sagaID, ruleName string) error

type RedisSagaStoreOptions 

type RedisSagaStoreOptions struct {
	Addr     string
	Password string
	DB       int
	Prefix   string
	TTL      time.Duration
}

RedisSagaStoreOptions configures the Redis saga store.

type Registry 

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

Registry provides expression evaluation with extensible data and functions

func NewRegistry 

func NewRegistry() *Registry

NewRegistry creates a new empty registry

func (*Registry) Clear 

func (r *Registry) Clear()

Clear removes all data and compiled programs (keeps functions)

func (*Registry) ClearAll 

func (r *Registry) ClearAll()

ClearAll removes everything including functions

func (*Registry) CompileExpression 

func (r *Registry) CompileExpression(expression string) error

CompileExpression compiles an expression for faster repeated evaluation

func (*Registry) CompileLogicalExpression 

func (r *Registry) CompileLogicalExpression(expression string, schemaInfo effectus.SchemaInfo) ([]*Predicate, map[string]struct{}, error)

CompileLogicalExpression compiles a logical expression and extracts fact paths Note: expr handles path resolution automatically, so we don't need custom parsing

func (*Registry) EvaluateBoolean 

func (r *Registry) EvaluateBoolean(expression string) (bool, error)

EvaluateBoolean evaluates an expression expecting a boolean result

func (*Registry) EvaluateCompiled 

func (r *Registry) EvaluateCompiled(expression string) (interface{}, error)

EvaluateCompiled evaluates a pre-compiled expression

func (*Registry) EvaluateExpression 

func (r *Registry) EvaluateExpression(expression string) (interface{}, error)

EvaluateExpression evaluates an expression and returns the result

func (*Registry) EvaluatePredicates 

func (r *Registry) EvaluatePredicates(predicates []*Predicate, facts effectus.Facts) bool

EvaluatePredicates evaluates multiple predicates against a request-local registry.

func (*Registry) Get 

func (r *Registry) Get(path string) (interface{}, bool)

Get retrieves a value by path

func (*Registry) GetPathsWithPrefix 

func (r *Registry) GetPathsWithPrefix(prefix string) []string

GetPathsWithPrefix returns all data paths that start with the given prefix

func (*Registry) GetType 

func (r *Registry) GetType(path string) (interface{}, bool)

GetType returns type information for a path (basic reflection)

func (*Registry) LoadFromFacts 

func (r *Registry) LoadFromFacts(facts effectus.Facts)

LoadFromFacts loads facts from effectus.Facts into the registry

func (*Registry) LoadFromJSON 

func (r *Registry) LoadFromJSON(jsonData []byte) error

LoadFromJSON loads data from JSON bytes

func (*Registry) LoadFromMap 

func (r *Registry) LoadFromMap(data map[string]interface{})

LoadFromMap loads data from a map, flattening nested structures

func (*Registry) Merge 

func (r *Registry) Merge(other *Registry)

Merge combines another registry's data and functions into this one

func (*Registry) NewPredicate 

func (r *Registry) NewPredicate(expression string) (*Predicate, error)

NewPredicate creates a new predicate using the registry

func (*Registry) RegisterFunction 

func (r *Registry) RegisterFunction(name string, fn interface{})

RegisterFunction registers a function for use in expressions

func (*Registry) Set 

func (r *Registry) Set(path string, value interface{})

Set stores a value at the given path

func (*Registry) SetClock 

func (r *Registry) SetClock(clock func() time.Time)

SetClock overrides the time source for temporal functions like now().

func (*Registry) SetNow 

func (r *Registry) SetNow(now time.Time)

SetNow sets a fixed timestamp for temporal functions.

func (*Registry) TypeCheckExpression 

func (r *Registry) TypeCheckExpression(expression string) error

TypeCheckExpression validates an expression without evaluating it

type RetentionPolicy 

type RetentionPolicy struct {
	Duration   string            `json:"duration"`
	Strategy   string            `json:"strategy"`
	Conditions map[string]string `json:"conditions"`
}

RetentionPolicy defines data retention rules

type SagaEffect 

type SagaEffect struct {
	ID        string                 `json:"id"`
	Sequence  int                    `json:"sequence"`
	Verb      string                 `json:"verb"`
	Args      map[string]interface{} `json:"args"`
	Result    interface{}            `json:"result,omitempty"`
	Status    string                 `json:"status"`
	Timestamp time.Time              `json:"timestamp"`
	Error     string                 `json:"error,omitempty"`
}

SagaEffect represents an effect recorded in a saga transaction.

func GetSagaEffect 

func GetSagaEffect(store SagaStore, sagaID, effectID string) (*SagaEffect, error)

GetSagaEffect returns one effect occurrence by its stable identity.

type SagaExecutor 

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

SagaExecutor wraps an executor with saga and capability management

func NewSagaExecutor 

func NewSagaExecutor(executor effectus.Executor, sagaStore SagaStore, capSystem *capability.CapabilitySystem, verbRegistry SagaVerbRegistry, holderID string) *SagaExecutor

NewSagaExecutor creates a new saga-aware executor

func (*SagaExecutor) ExecuteWithSaga 

func (se *SagaExecutor) ExecuteWithSaga(ctx context.Context, sagaID string, ruleName string, effects []effectus.Effect) ([]interface{}, error)

ExecuteWithSaga executes a series of effects within a saga transaction

type SagaInstance 

type SagaInstance = workflow.SagaInstance

type SagaState 

type SagaState = workflow.SagaState

Durable workflow compatibility aliases. New contract consumers import schema/workflow.

type SagaStep 

type SagaStep = workflow.SagaStep

type SagaStore 

type SagaStore interface {
	StartTransaction(sagaID, ruleName string) error
	RecordEffect(sagaID, effectID string, sequence int, verb string, args map[string]interface{}) error
	MarkSuccess(sagaID, effectID string, result interface{}) error
	MarkFailed(sagaID, effectID string, reason error) error
	MarkCompensated(sagaID, effectID string) error
	GetTransactionEffects(sagaID string) ([]*SagaEffect, error)
	GetActiveSagas() ([]string, error)
	CompleteSaga(sagaID string) error
}

SagaStore defines the interface for persisting saga transactions

type SagaVerbRegistry 

type SagaVerbRegistry interface {
	GetVerb(name string) (*verb.Spec, bool)
}

SagaVerbRegistry defines the interface for accessing verb specifications in sagas.

type SchemaValidationResult 

type SchemaValidationResult struct {
	Valid           bool     `json:"valid"`
	Errors          []string `json:"errors"`
	Warnings        []string `json:"warnings"`
	BreakingChanges []string `json:"breaking_changes"`
	Suggestions     []string `json:"suggestions"`
}

SchemaValidationResult represents the result of schema validation

type StepState 

type StepState = workflow.StepState

type UnknownOutcomeRetryPolicy 

type UnknownOutcomeRetryPolicy = workflow.UnknownOutcomeRetryPolicy

type VerbSchema 

type VerbSchema struct {
	Name                 string                 `json:"name"`
	Version              string                 `json:"version"`
	Description          string                 `json:"description"`
	InputSchema          map[string]interface{} `json:"input_schema"`
	OutputSchema         map[string]interface{} `json:"output_schema"`
	RequiredCapabilities []string               `json:"required_capabilities"`
	ExecutionType        string                 `json:"execution_type"`
	Idempotent           bool                   `json:"idempotent"`
	Compensatable        bool                   `json:"compensatable"`
	BufModule            string                 `json:"buf_module"`
	BufCommit            string                 `json:"buf_commit"`
	CreatedAt            time.Time              `json:"created_at"`
	UpdatedAt            time.Time              `json:"updated_at"`
}

VerbSchema represents a versioned verb interface schema

type VerbSchemaRegistry 

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

VerbSchemaRegistry manages verb interface schemas

Source Files

View all Source files

Directories

Path Synopsis
Package expression defines the narrow contracts used by expression clients.
 Package expression defines the narrow contracts used by expression clients.
Package fencing defines resource leases that issue fencing tokens.
 Package fencing defines resource leases that issue fencing tokens.
Package ledger defines durable execution admission and recovery contracts without database or queue implementations.
 Package ledger defines durable execution admission and recovery contracts without database or queue implementations.
Package types provides the unified type system for Effectus
 Package types provides the unified type system for Effectus
Package verb provides definitions and utilities for effect verbs
 Package verb provides definitions and utilities for effect verbs
Package workflow defines durable saga and outbox contracts without storage implementations.
 Package workflow defines durable saga and outbox contracts without storage implementations.

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.