Documentation
¶
Index ¶
- Constants
- Variables
- func CanonicalJSON(value any) (json.RawMessage, string, error)
- func ExtractFactPaths(expression string) map[string]struct{}
- func IdempotencyKey(namespace, sagaID, effectID string, direction invocation.Direction) string
- func IsTerminalExecutionState(state ExecutionState) bool
- func MigrateSagaV2(ctx context.Context, db *sql.DB) error
- 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
- type BufBreakingConfig
- type BufBuildConfig
- type BufConfig
- type BufGenConfig
- type BufGoPackagePrefixConfig
- type BufIntegration
- 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
- type BufManagedConfig
- type BufPluginConfig
- type CheckedEnqueueRequest
- type ClaimOptions
- type CodeGenerationResult
- type Completion
- type CreateSagaRequest
- type Dispatch
- type DispatchAttempt
- type DispatchState
- type Dispatcher
- type DispatcherOptions
- type DurableAdmission
- type EnqueueStepRequest
- type ExecutionArtifact
- type ExecutionLease
- type ExecutionLedger
- type ExecutionPlanRecord
- type ExecutionRecord
- type ExecutionState
- type FactApplication
- type FactSchema
- type FactSchemaRegistry
- type FencingRequirement
- type 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
- 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 IndexDefinition
- type OutboxStore
- type PostgresOutboxStore
- 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 PrivacyRule
- type PruneOptions
- type PruneReport
- type PruneResult
- type RecoveryStats
- type RecoveryStatsReader
- type RetentionPolicy
- type SagaInstance
- type SagaState
- type SagaStep
- type SchemaValidationResult
- type StepState
- type UnknownOutcomeRetryPolicy
- type VerbSchema
- type VerbSchemaRegistry
Constants ¶
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 )
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 )
Variables ¶
var ( ErrArtifactNotFound = errors.New("execution artifact not found") ErrExecutionNotFound = errors.New("execution not found") ErrStaleExecutionLease = errors.New("stale execution recovery lease") )
var ( // ErrIdentityConflict is re-exported as runtime.ErrIdentityConflict. Keep // this leaf-package declaration so durable stores and runtime share one // sentinel without creating an import cycle. ErrIdentityConflict = errors.New("engine admission 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") )
Functions ¶
func CanonicalJSON ¶
func CanonicalJSON(value any) (json.RawMessage, string, error)
CanonicalJSON encodes JSON-compatible data and rejects lossy or unsupported values.
func ExtractFactPaths ¶
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 MigrateSagaV2 ¶
MigrateSagaV2 applies the versioned durable runtime migrations. Production deployments should use a separate DDL credential and validate at runtime.
func StableAdmissionID ¶
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 ¶
StableExecutionID derives an execution identity from admission identity.
func StableSagaID ¶
StableSagaID derives one plan saga identity from an execution.
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. Removal deadline: 2027-09-01.
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 ¶
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 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 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 (*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 (*InMemoryOutboxStore) SaveFencingGrants ¶
func (store *InMemoryOutboxStore) SaveFencingGrants(ctx context.Context, dispatchID string, attempt uint64, leaseToken string, grants []invocation.FencingGrant) error
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 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 (*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 (*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 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 ¶
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 RetentionPolicy ¶
type RetentionPolicy struct {
Duration string `json:"duration"`
Strategy string `json:"strategy"`
Conditions map[string]string `json:"conditions"`
}
RetentionPolicy defines data retention rules
type SagaInstance ¶
type SagaInstance = workflow.SagaInstance
type SagaState ¶
Durable workflow compatibility aliases. New contract consumers import schema/workflow.
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 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
¶
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 workflow defines durable saga and outbox contracts without storage implementations.
|
Package workflow defines durable saga and outbox contracts without storage implementations. |