engine

Documentation

Overview

Package engine provides a provider-neutral durable-delivery runtime core for actorlayer dispatch.

The engine contract is deliberately small. A product supplies a Source that yields deliveries, a Handler that executes one delivery, a Resolver that maps each delivery to a deterministic lane key, an optional EventSink for generic lifecycle events, and RetryPolicy hooks for retry classification, backoff, and exhaustion checks.

Engine events are generic execution facts: running, in_progress, acked, retrying, and deadlettered. The engine does not define product event subjects, queue names, task state transitions, DLQ storage, projection writes, provider selection, ADK sessions, Telegram delivery, MCP tools, model routing, or workspace policy. Products such as Balda own those decisions in their integration layer while using this package for deterministic lane execution and delivery settlement orchestration.

Index

• Variables
• type AddressResolver
• type Config
• type Delivery
• type DispatchRuntime
  • func NewDispatchRuntime(cfg RuntimeConfig) (*DispatchRuntime, error)
  • func (r *DispatchRuntime) Handle(ctx context.Context, delivery Delivery) error
  • func (r *DispatchRuntime) LaneStatus() LaneStatus
  • func (r *DispatchRuntime) Run(ctx context.Context, source Source) error
• type Event
• type EventSink
• type EventType
• type Handler
• type LaneKeyResolver
• type LaneStatus
• type ResolveError
  • func (e *ResolveError) Error() string
  • func (e *ResolveError) Unwrap() error
• type Resolver
• type RetryPolicy
• type Runtime
  • func New(cfg Config) (*Runtime, error)
  • func (r *Runtime) EmitInProgress(ctx context.Context, delivery Delivery)
  • func (r *Runtime) Handle(ctx context.Context, delivery Delivery, handler Handler) error
  • func (r *Runtime) LaneStatus() LaneStatus
  • func (r *Runtime) Run(ctx context.Context, src Source, handler Handler) error
• type RuntimeConfig
• type Source

Variables

var ErrActorNotFound = actorlayer.ErrActorNotFound

Types

type AddressResolver

type AddressResolver func(envelope any) (string, error)

type Config

type Config struct {
	Resolver    Resolver
	Retry       RetryPolicy
	Sink        EventSink
	LaneIdleTTL time.Duration
}

type Delivery

type Delivery interface {
	Envelope() any
	Attempt() int
	MaxAttempts() int
	InProgress(ctx context.Context) error
	Ack(ctx context.Context) error
	Retry(ctx context.Context, delay time.Duration, reason string) error
	DeadLetter(ctx context.Context, reason string) error
}

type DispatchRuntime

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

func NewDispatchRuntime

func NewDispatchRuntime(cfg RuntimeConfig) (*DispatchRuntime, error)

func (*DispatchRuntime) Handle

func (r *DispatchRuntime) Handle(ctx context.Context, delivery Delivery) error

func (*DispatchRuntime) LaneStatus

func (r *DispatchRuntime) LaneStatus() LaneStatus

func (*DispatchRuntime) Run

func (r *DispatchRuntime) Run(ctx context.Context, source Source) error

type Event

type Event struct {
	Type        EventType
	Reason      string
	RetryDelay  time.Duration
	Attempt     int
	MaxAttempts int
}

type EventSink

type EventSink interface {
	Publish(ctx context.Context, event Event)
}

type EventType

type EventType string

const (
	EventRunning      EventType = "running"
	EventInProgress   EventType = "in_progress"
	EventAcked        EventType = "acked"
	EventRetrying     EventType = "retrying"
	EventDeadLettered EventType = "deadlettered"
)

type Handler

type Handler func(ctx context.Context, delivery Delivery) error

type LaneKeyResolver

type LaneKeyResolver func(envelope any) string

type LaneStatus

type LaneStatus struct {
	Active int
	Keys   []string
}

type ResolveError

type ResolveError struct {
	Address string
}

ResolveError is returned when a dispatch address cannot be resolved. It wraps ErrActorNotFound to support errors.Is checks while preserving the concrete address in messages and logs.

func (*ResolveError) Error

func (e *ResolveError) Error() string

func (*ResolveError) Unwrap

func (e *ResolveError) Unwrap() error

type Resolver

type Resolver interface {
	LaneKey(delivery Delivery) string
}

type RetryPolicy

type RetryPolicy struct {
	IsRetryable    func(error) bool
	Backoff        func(attempt int) time.Duration
	RetryExhausted func(delivery Delivery) bool
}

type Runtime

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

func New

func New(cfg Config) (*Runtime, error)

func (*Runtime) EmitInProgress

func (r *Runtime) EmitInProgress(ctx context.Context, delivery Delivery)

func (*Runtime) Handle

func (r *Runtime) Handle(ctx context.Context, delivery Delivery, handler Handler) error

func (*Runtime) LaneStatus

func (r *Runtime) LaneStatus() LaneStatus

func (*Runtime) Run

func (r *Runtime) Run(ctx context.Context, src Source, handler Handler) error

type RuntimeConfig

type RuntimeConfig struct {
	Registry    dispatch.Registry
	AddressOf   AddressResolver
	LaneKey     LaneKeyResolver
	Retry       RetryPolicy
	Sink        EventSink
	LaneIdleTTL time.Duration
}

type Source

type Source interface {
	Run(ctx context.Context, handler Handler) error
}