effect

package
v1.4.0 Latest Latest
Warning

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

 Go to latest Published: Oct 20, 2026 License: Apache-2.0 Imports: 9 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrEffectNotFound indicates that the requested effect record does not exist.
	ErrEffectNotFound = errors.New("effect record not found")

	// ErrEffectFenced indicates that the attempt is stale or superseded and cannot execute effects.
	ErrEffectFenced = errors.New("effect rejected by fencing: attempt is stale or superseded")

	// ErrEffectUnknown indicates that the effect outcome is ambiguous and cannot be confirmed.
	// Automatic replay of an effect in UNKNOWN state is strictly prohibited.
	ErrEffectUnknown = errors.New("effect is in UNKNOWN state: outcome is ambiguous, automatic replay is strictly prohibited")

	// ErrEffectExecuting indicates that the effect is currently executing by another attempt or worker.
	ErrEffectExecuting = errors.New("effect is currently executing by another active attempt")

	// ErrPayloadMismatch indicates an idempotency key collision with a different payload hash.
	ErrPayloadMismatch = errors.New("idempotency key conflict: payload hash does not match previous request")

	// ErrProviderNotFound indicates that the requested external provider is not registered.
	ErrProviderNotFound = errors.New("effect provider not registered")

	// ErrProviderFailed indicates that the external provider execution failed with a definitive error.
	ErrProviderFailed = errors.New("effect provider execution failed")

	// ErrEffectExpired indicates that the effect request deadline has passed.
	ErrEffectExpired = errors.New("effect request deadline expired")

	// ErrInvalidRequest indicates that the effect request parameters failed validation.
	ErrInvalidRequest = errors.New("invalid effect request")

	// ErrEffectAlreadyCommitted indicates that the effect has already completed successfully.
	ErrEffectAlreadyCommitted = errors.New("effect has already been committed")

	// ErrInvalidStatusTransition indicates an invalid state transition on an effect record.
	ErrInvalidStatusTransition = errors.New("invalid effect status transition")
)

Functions

func ComputePayloadHash 

func ComputePayloadHash(payload []byte) string

ComputePayloadHash calculates a standard SHA-256 hex digest of the payload bytes.

func IsAmbiguous 

func IsAmbiguous(err error) bool

IsAmbiguous tests whether an error represents an ambiguous execution outcome.

func MarkAmbiguous 

func MarkAmbiguous(err error) error

MarkAmbiguous wraps an error into an AmbiguousError.

Types

type AmbiguousError 

type AmbiguousError struct {
	Err error
}

AmbiguousError indicates that the outcome of an external operation cannot be determined. For instance, a network connection timed out or dropped after sending the HTTP payload, but before the server returned an acknowledgement.

func (*AmbiguousError) Error 

func (e *AmbiguousError) Error() string

func (*AmbiguousError) Unwrap 

func (e *AmbiguousError) Unwrap() error

type EffectReceipt 

type EffectReceipt struct {
	EffectID        string       `json:"effect_id"`
	TenantID        string       `json:"tenant_id"`
	Status          EffectStatus `json:"status"`
	ProviderReceipt []byte       `json:"provider_receipt,omitempty"`
	StartedAt       time.Time    `json:"started_at"`
	CompletedAt     time.Time    `json:"completed_at"`
	AttemptID       string       `json:"attempt_id,omitempty"`
	FencingToken    int64        `json:"fencing_token"`
	ResultHash      string       `json:"result_hash,omitempty"`
	ErrorMessage    string       `json:"error_message,omitempty"`
}

EffectReceipt contains the durable execution outcome and proof of an external side effect.

type EffectRecord 

type EffectRecord struct {
	Request   EffectRequest  `json:"request"`
	Receipt   *EffectReceipt `json:"receipt,omitempty"`
	Status    EffectStatus   `json:"status"`
	CreatedAt time.Time      `json:"created_at"`
	UpdatedAt time.Time      `json:"updated_at"`
	Version   int64          `json:"version"`
}

EffectRecord represents the persistent entity stored in the EffectStore.

type EffectRequest 

type EffectRequest struct {
	ID             string     `json:"effect_id"`
	TenantID       string     `json:"tenant_id"`
	AgentID        string     `json:"agent_id"`
	RunID          string     `json:"run_id,omitempty"`
	AttemptID      string     `json:"attempt_id,omitempty"`
	FencingToken   int64      `json:"fencing_token"`
	Provider       string     `json:"provider"`
	Operation      string     `json:"operation"`
	IdempotencyKey string     `json:"idempotency_key"`
	PayloadHash    string     `json:"payload_hash"`
	Payload        []byte     `json:"payload,omitempty"`
	Deadline       *time.Time `json:"deadline,omitempty"`
	TraceID        string     `json:"trace_id,omitempty"`
	CreatedAt      time.Time  `json:"created_at"`
}

EffectRequest defines a kernel-level request to produce an external side effect.

func (*EffectRequest) Validate 

func (r *EffectRequest) Validate() error

Validate checks essential constraints on the EffectRequest and populates ID and PayloadHash if missing.

type EffectStatus 

type EffectStatus string

EffectStatus represents the lifecycle state of an external side effect. 

const (
	// EffectStatusPending indicates the effect request has been registered but execution has not begun.
	EffectStatusPending EffectStatus = "PENDING"

	// EffectStatusExecuting indicates the effect is currently being dispatched to the external provider.
	EffectStatusExecuting EffectStatus = "EXECUTING"

	// EffectStatusCommitted indicates the effect was successfully executed and acknowledged by the external provider.
	// It is terminal and safe for idempotent replay (logical effect = 1).
	EffectStatusCommitted EffectStatus = "COMMITTED"

	// EffectStatusFailed indicates the external provider definitively rejected or failed the request.
	EffectStatusFailed EffectStatus = "FAILED"

	// EffectStatusUnknown indicates the request may or may not have reached or completed at the external provider
	// (e.g. timeout, connection reset, ambiguous ack).
	// CRITICAL: UNKNOWN must NEVER be automatically replayed or retried blindly.
	EffectStatusUnknown EffectStatus = "UNKNOWN"
)

func (EffectStatus) IsTerminal 

func (s EffectStatus) IsTerminal() bool

IsTerminal returns true if the status is a completed terminal state.

type Fencer 

type Fencer interface {
	ValidateFencingToken(ctx context.Context, tenantID, agentID, runID, attemptID string, fencingToken int64) error
}

Fencer validates whether an agent attempt's fencing token is current or stale.

type MemoryFencer 

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

MemoryFencer provides an in-memory fencing registry for testing and development.

func NewMemoryFencer 

func NewMemoryFencer() *MemoryFencer

NewMemoryFencer creates a new in-memory fencer.

func (*MemoryFencer) SetActiveToken 

func (f *MemoryFencer) SetActiveToken(tenantID, agentID, runID string, token int64)

SetActiveToken sets the currently valid active fencing token for a tenant/agent/run tuple.

func (*MemoryFencer) ValidateFencingToken 

func (f *MemoryFencer) ValidateFencingToken(ctx context.Context, tenantID, agentID, runID, attemptID string, fencingToken int64) error

ValidateFencingToken checks if fencingToken is at least the active token. If active token is set and fencingToken < activeToken, ErrEffectFenced is returned.

type MemoryMetrics 

type MemoryMetrics struct {
	ExecutedCount  int64
	CommittedCount int64
	FailedCount    int64
	UnknownCount   int64
	FencedCount    int64
	IdempotentHits int64
	// contains filtered or unexported fields
}

MemoryMetrics tracks counts in memory for test verification and monitoring.

func NewMemoryMetrics 

func NewMemoryMetrics() *MemoryMetrics

NewMemoryMetrics creates a new MemoryMetrics instance.

func (*MemoryMetrics) RecordCommitted 

func (m *MemoryMetrics) RecordCommitted(ctx context.Context, tenantID, provider, operation string)

func (*MemoryMetrics) RecordExecuted 

func (m *MemoryMetrics) RecordExecuted(ctx context.Context, tenantID, provider, operation string)

func (*MemoryMetrics) RecordFailed 

func (m *MemoryMetrics) RecordFailed(ctx context.Context, tenantID, provider, operation, reason string)

func (*MemoryMetrics) RecordFenced 

func (m *MemoryMetrics) RecordFenced(ctx context.Context, tenantID, provider string)

func (*MemoryMetrics) RecordIdempotentHit 

func (m *MemoryMetrics) RecordIdempotentHit(ctx context.Context, tenantID, provider, operation string)

func (*MemoryMetrics) RecordUnknown 

func (m *MemoryMetrics) RecordUnknown(ctx context.Context, tenantID, provider, operation string)

type MemoryStore 

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

MemoryStore provides a concurrent-safe, in-memory implementation of Store.

func NewMemoryStore 

func NewMemoryStore() *MemoryStore

NewMemoryStore creates a new MemoryStore.

func (*MemoryStore) CreateRecord 

func (s *MemoryStore) CreateRecord(ctx context.Context, record *EffectRecord) error

CreateRecord persists a new effect record in PENDING status.

func (*MemoryStore) ForceStatus 

func (s *MemoryStore) ForceStatus(ctx context.Context, tenantID, effectID string, newStatus EffectStatus, receipt *EffectReceipt) (*EffectRecord, error)

ForceStatus updates the status directly, used for administrative resolution of UNKNOWN state.

func (*MemoryStore) GetByIdempotencyKey 

func (s *MemoryStore) GetByIdempotencyKey(ctx context.Context, tenantID, agentID, key string) (*EffectRecord, error)

GetByIdempotencyKey retrieves an effect record by its idempotency key.

func (*MemoryStore) GetEffect 

func (s *MemoryStore) GetEffect(ctx context.Context, tenantID, effectID string) (*EffectRecord, error)

GetEffect retrieves an effect record by tenant and effect ID.

func (*MemoryStore) TransitionStatus 

func (s *MemoryStore) TransitionStatus(ctx context.Context, tenantID, effectID string, expectedStatus EffectStatus, newStatus EffectStatus, receipt *EffectReceipt) (*EffectRecord, error)

TransitionStatus performs an atomic Compare-And-Swap (CAS) on the effect status.

type Metrics 

type Metrics interface {
	RecordExecuted(ctx context.Context, tenantID, provider, operation string)
	RecordCommitted(ctx context.Context, tenantID, provider, operation string)
	RecordFailed(ctx context.Context, tenantID, provider, operation, reason string)
	RecordUnknown(ctx context.Context, tenantID, provider, operation string)
	RecordFenced(ctx context.Context, tenantID, provider string)
	RecordIdempotentHit(ctx context.Context, tenantID, provider, operation string)
}

Metrics records telemetry and counters for effect lifecycle events.

type NoopMetrics 

type NoopMetrics struct{}

NoopMetrics is a no-op implementation of Metrics.

func (NoopMetrics) RecordCommitted 

func (NoopMetrics) RecordCommitted(context.Context, string, string, string)

func (NoopMetrics) RecordExecuted 

func (NoopMetrics) RecordExecuted(context.Context, string, string, string)

func (NoopMetrics) RecordFailed 

func (NoopMetrics) RecordFailed(context.Context, string, string, string, string)

func (NoopMetrics) RecordFenced 

func (NoopMetrics) RecordFenced(context.Context, string, string)

func (NoopMetrics) RecordIdempotentHit 

func (NoopMetrics) RecordIdempotentHit(context.Context, string, string, string)

func (NoopMetrics) RecordUnknown 

func (NoopMetrics) RecordUnknown(context.Context, string, string, string)

type Provider 

type Provider interface {
	Name() string
	Execute(ctx context.Context, req *EffectRequest) ([]byte, error)
}

Provider defines the execution interface for an external effect provider.

type Service 

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

Service coordinates external effect execution, fencing, deduplication, and durable receipts.

func NewService 

func NewService(store Store, fencer Fencer, metrics Metrics) *Service

NewService constructs an Effect Service.

func (*Service) ExecuteEffect 

func (s *Service) ExecuteEffect(ctx context.Context, req *EffectRequest) (*EffectReceipt, error)

ExecuteEffect executes an external side effect with strict fencing, idempotency, and receipt recording.

func (*Service) GetEffect 

func (s *Service) GetEffect(ctx context.Context, tenantID, effectID string) (*EffectRecord, error)

GetEffect retrieves an effect record by tenant and effect ID.

func (*Service) GetEffectByIdempotencyKey 

func (s *Service) GetEffectByIdempotencyKey(ctx context.Context, tenantID, agentID, idempotencyKey string) (*EffectRecord, error)

GetEffectByIdempotencyKey retrieves an effect record by idempotency key.

func (*Service) RegisterProvider 

func (s *Service) RegisterProvider(provider Provider) error

RegisterProvider registers an external effect provider.

func (*Service) ResolveUnknownEffect 

func (s *Service) ResolveUnknownEffect(ctx context.Context, tenantID, effectID string, forceStatus EffectStatus, providerReceipt []byte, reason string) (*EffectReceipt, error)

ResolveUnknownEffect allows manual or administrative resolution of an effect in UNKNOWN status. This is used when external verification (e.g. contacting the external payment/webhook provider) has determined the actual outcome.

type Store 

type Store interface {
	GetEffect(ctx context.Context, tenantID, effectID string) (*EffectRecord, error)
	GetByIdempotencyKey(ctx context.Context, tenantID, agentID, idempotencyKey string) (*EffectRecord, error)
	CreateRecord(ctx context.Context, record *EffectRecord) error
	TransitionStatus(ctx context.Context, tenantID, effectID string, expectedStatus EffectStatus, newStatus EffectStatus, receipt *EffectReceipt) (*EffectRecord, error)
	ForceStatus(ctx context.Context, tenantID, effectID string, newStatus EffectStatus, receipt *EffectReceipt) (*EffectRecord, error)
}

Store defines persistent storage and atomic state transitions for Effect records.

Source Files

View all Source files

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.