action

package
v0.19.0 Latest Latest
Warning

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

 Go to latest Published: Sep 22, 2026 License: Apache-2.0 Imports: 24 Imported by: 0

Documentation

Overview

Copyright 2018-2026 Marcin Polak. All rights reserved. Use of this source code is governed by an Apache-2.0 license that can be found in the LICENSE file.

Package action provides typed, composable application actions.

Build an action with New, configure it with fluent policies and middleware, then reuse the resulting BuiltAction concurrently. The typed Do method is the preferred execution path; decoded execution is intended for adapters.

Index

Constants

View Source 
const (
	DefaultIdempotencyTTL      = 24 * time.Hour
	DefaultIdempotencyLeaseTTL = 2 * time.Minute
)
View Source 
const (
	DefaultRateLimiterCapacity = 65536
	DefaultRateLimiterTTL      = 3 * time.Minute
)
View Source 
const DefaultMaxTreeDepth = 64

Variables

View Source 
var DefaultTextFields = []string{"content", "text", "output", "result", "message"}
View Source 
var ErrConcurrencyLimit = errors.New("concurrency limit exceeded")
View Source 
var ErrLocked = xerr.Conflict("resource is currently locked by another instance")

ErrLocked is returned when an action is already locked by another instance.

View Source 
var ErrTypeAssertion = errors.New("critical type assertion failure")

Functions

func AlwaysRetryPredicate added in v0.3.0 

func AlwaysRetryPredicate(err error) bool

AlwaysRetryPredicate retries any non-nil error.

func ApplyTreeHook added in v0.6.0 

func ApplyTreeHook(act AnyAction, hooks ...AnyHook) error

func ApplyTreeHookOpts added in v0.6.0 

func ApplyTreeHookOpts(act AnyAction, opts TreeHookOptions, hooks ...AnyHook) error

func Assign added in v0.9.0 

func Assign(target, source any) error

Assign writes 'source' into the pointer 'target' using zero-allocation fast paths before falling back to JSON serialization. 'target' MUST be a non-nil pointer.

func Async 

func Async[Req, Res any](ctx context.Context, act *BuiltAction[Req, Res], req Req) <-chan AsyncResult[Res]

Async executes the action in a goroutine and returns a result channel.

func Coerce added in v0.4.0 

func Coerce[T any](input any) (T, error)

Coerce converts input into T.

func CollectStream 

func CollectStream[Req, T any](ctx context.Context, a *StreamAction[Req, T], req Req) (out []T, err error)

func ConstantBackoff 

func ConstantBackoff(d time.Duration) func(attempt int) time.Duration

ConstantBackoff waits the same duration between every attempt.

func DefaultRetryPredicate added in v0.3.0 

func DefaultRetryPredicate(err error) bool

DefaultRetryPredicate retries only errors flagged as transient by xerr.

func ExecutionIDFrom 

func ExecutionIDFrom(ctx context.Context) string

func ExponentialBackoff 

func ExponentialBackoff(base, maxDuration time.Duration) func(attempt int) time.Duration

ExponentialBackoff returns a backoff that doubles each attempt, capped at max. 100ms → 200ms → 400ms → 800ms … → max

Usage: .Retry(3, action.ExponentialBackoff(100*time.Millisecond, 5*time.Second))

func ExponentialJitter 

func ExponentialJitter(base, maxDuration time.Duration) func(attempt int) time.Duration

ExponentialJitter adds ±30% random jitter to ExponentialBackoff. Prevents thundering-herd on simultaneous retries.

Usage: .Retry(3, action.ExponentialJitter(100*time.Millisecond, 5*time.Second))

func InvokeAny added in v0.4.0 

func InvokeAny(ctx context.Context, act, req any) (any, error)

InvokeAny executes an action with an in-memory input payload.

func LinearBackoff 

func LinearBackoff(step time.Duration) func(attempt int) time.Duration

LinearBackoff increments by step each attempt. 100ms → 200ms → 300ms …

func Race 

func Race[Req, Res any](
	ctx context.Context,
	act *BuiltAction[Req, Res],
	reqs []Req,
) (Res, error)

Race executes the action concurrently for each request and returns the first success.

func SpanIDFrom 

func SpanIDFrom(ctx context.Context) string

func StreamFromFunc added in v0.18.0 

func StreamFromFunc[T any](next func() (T, error)) iter.Seq2[T, error]

StreamFromFunc adapts a pull-based source into an iter.Seq2.

The next callback is expected to return: 

(item, nil)             → emit item and continue
(zero, io.EOF)          → stop cleanly, no error emitted downstream
(zero, other error)     → stop with error emitted downstream

Typical usage — wrapping bufio.Scanner or a database cursor: 

scanner := bufio.NewScanner(f)
seq := action.StreamFromFunc(func() (string, error) {
    if !scanner.Scan() {
        if err := scanner.Err(); err != nil {
            return "", err
        }
        return "", io.EOF
    }
    return scanner.Text(), nil
})

io.EOF is treated as the normal termination signal and is never propagated downstream. Use a different error type for real failures.

func StreamFromSlice added in v0.18.0 

func StreamFromSlice[T any](items []T) iter.Seq2[T, error]

StreamFromSlice returns an iter.Seq2 that yields items from the slice.

This is the canonical way to turn an in-memory collection into a stream source. The iterator is single-use (standard iter.Seq2 semantics): consuming it twice, or concurrently, is undefined behavior.

func TraceIDFrom 

func TraceIDFrom(ctx context.Context) string

func ValidateOperatorDeclaration added in v0.19.0 

func ValidateOperatorDeclaration(op NamedOperator) error

ValidateOperatorDeclaration checks the NamedOperator's own shape.

func ValidateOperatorParams added in v0.19.0 

func ValidateOperatorParams(op NamedOperator, params map[string]any) (map[string]any, error)

ValidateOperatorParams checks DSL parameters against ParamSpec.

func WithExecutionID 

func WithExecutionID(ctx context.Context, id string) context.Context

func WithTraceContext 

func WithTraceContext(ctx context.Context, traceID, spanID string) context.Context

Types

type ActionProvider 

type ActionProvider interface {
	Actions() []AnyAction
}

type ActionScope 

type ActionScope string

ActionScope describes the contract audience for an action. It is not an authorization rule: authentication and authorization remain enforced by the action's transport middleware and guards. 

const (
	// ScopePublic is the browser/client business contract. It is the zero-value
	// behavior so normal client actions remain concise.
	ScopePublic ActionScope = ""
	// ScopeInternal is a trusted service-to-service or runner contract.
	ScopeInternal ActionScope = "internal"
	// ScopeSystem is a framework or operational contract.
	ScopeSystem ActionScope = "system"
)

type AdaptiveConfig 

type AdaptiveConfig struct {
	FailureThreshold int
	ResetTimeout     time.Duration
	InitialTimeout   time.Duration
}

AdaptiveConfig tunes circuit breaker thresholds and timeout bounds.

func (*AdaptiveConfig) SetDefaults 

func (c *AdaptiveConfig) SetDefaults()

type Admission 

type Admission interface {
	Acquire(context.Context) error
	Release()
}

Admission controls whether an execution may enter a protected section. Implementations may be local or distributed. Acquire must return an error without retaining the request when admission is denied.

type Alias added in v0.12.0 

type Alias struct {
	Canonical string
	Short     []string
}

Alias maps a short name to a canonical name. Canonical must already resolve to a registered action, source, or operator; Short names are registered as aliases pointing at the same entry.

type AnyAction 

type AnyAction interface {
	Executable
	AnyDoer
	Describable
	GetBindings() []Binding
	GetAnyHooks() []AnyHook
	AddAnyHook(h ...AnyHook)
	CloneWithHooks(hooks ...AnyHook) AnyAction
}

AnyAction is the type-erased interface for the App and Transports to handle actions.

func ApplyHooks added in v0.15.0 

func ApplyHooks(actions []AnyAction, hooks ...AnyHook) []AnyAction

ApplyHooks returns a new slice of actions where each action is an independent clone with the given hooks appended. The originals are never modified and remain safe to reuse in multiple registries concurrently.

func FilterByNode added in v0.7.0 

func FilterByNode(actions []AnyAction, policy NodeFilterPolicy) []AnyAction

type AnyDoer added in v0.4.0 

type AnyDoer interface {
	DoAny(ctx context.Context, req any) (any, error)
}

AnyDoer is the single-method interface for in-memory untyped execution.

type AnyHook 

type AnyHook struct {
	OnBuild func(meta *Meta, reqType, resType reflect.Type) bool

	Before func(ctx context.Context, req any, meta *Meta) (context.Context, error)
	After  func(ctx context.Context, req any, res any, err error, meta *Meta)

	OnSuccess func(ctx context.Context, req any, res any, meta *Meta)
	OnTimeout func(ctx context.Context, req any, meta *Meta)
	OnError   func(ctx context.Context, req any, err error, meta *Meta)

	OnRetry        func(ctx context.Context, req any, attempt int, err error, meta *Meta)
	OnCacheHit     func(ctx context.Context, req any, res any, meta *Meta)
	OnCacheMiss    func(ctx context.Context, req any, meta *Meta)
	OnCoalesced    func(ctx context.Context, req any, meta *Meta)
	OnDeduplicated func(ctx context.Context, req any, meta *Meta)
	OnCancel       func(ctx context.Context, req any, meta *Meta)

	OnPanic func(ctx context.Context, req any, recovered any, meta *Meta)
}

AnyHook is the type-erased equivalent of Hook[Req, Res]. OnPanic is kept here because recovered panic values are not representable by typed hooks.

func Adapt 

func Adapt[Req, Res any](h Hook[Req, Res]) AnyHook

Adapt converts a typed hook to its type-erased representation.

type AnyStream added in v0.18.0 

type AnyStream func(yield func(item any, err error) bool)

AnyStream is the type-erased stream returned at integration boundaries. Each yielded value is an item; a non-nil item error terminates the stream's current consumer according to the hosting Flow/transport policy.

type AnyStreamAction added in v0.18.0 

type AnyStreamAction interface {
	Describable
	TypedPayload
	GetBindings() []Binding
	GetAnyHooks() []AnyHook
	AddAnyHook(h ...AnyHook)
	CloneWithHooks(h ...AnyHook) AnyStreamAction
	DoStreamAny(ctx context.Context, req any) (AnyStream, error)
}

AnyStreamAction is the cold-path contract for stream actions. It is intentionally separate from AnyAction: a stream is lazy, can fail during iteration, and must not be coerced into a unary response.

Hook lifecycle: Before fires when DoStreamAny is called; After and OnError fire when the stream is exhausted, errored, or the consumer stops early via yield(false). OnRetry/OnCacheHit/OnCacheMiss are reserved for future middleware and are not yet fired by StreamAction.

type AsyncResult 

type AsyncResult[Res any] struct {
	Value Res
	Err   error
}

AsyncResult carries the outcome of an asynchronous action execution.

func FanOut 

func FanOut[Req, Res any](
	ctx context.Context,
	act *BuiltAction[Req, Res],
	reqs []Req,
	maxConcurrency int,
) []AsyncResult[Res]

FanOut executes the action concurrently for each request. maxConcurrency limits how many run simultaneously; 0 = unbounded.

type AuditLogger 

type AuditLogger interface {
	Log(ctx context.Context, category string, actionName string, details string)
}

AuditLogger defines the minimal contract required by the builder to log audit events. This prevents circular import dependencies between action and audit adapters.

type Binding 

type Binding any

type Builder 

type Builder[Req, Res any] struct {
	// contains filtered or unexported fields
}

Builder[Req, Res] is the fluent construction API for an action. All methods return the same builder for chaining. Call Build() once — the result is immutable and safe for concurrent use.

func Branch 

func Branch[Req, Res any](
	name string,
	routes map[string]*Builder[Req, Res],
	router func(context.Context, Req) (string, error),
) *Builder[Req, Res]

Branch routes a request to one of several named actions based on a router function.

func BranchAny added in v0.4.0 

func BranchAny(
	name string,
	routes map[string]AnyAction,
	router func(context.Context, any) (string, error),
) *Builder[any, any]

BranchAny routes dynamic execution to one of several named actions based on a router function.

func Chain 

func Chain[T any](
	name string,
	builders ...*Builder[T, T],
) *Builder[T, T]

Chain runs same-typed actions sequentially; each receives the previous output.

func Dynamic added in v0.4.0 

func Dynamic(act AnyAction) *Builder[any, any]

Dynamic lifts any AnyAction into a *Builder[any, any]. It preserves all metadata, bindings, and hooks from the wrapped action.

func FirstSuccess 

func FirstSuccess[Req, Res any](
	name string,
	builders ...*Builder[Req, Res],
) *Builder[Req, Res]

FirstSuccess executes actions in order, returning the first non-error result.

func New 

func New[Req, Res any](name string, exec Fn[Req, Res]) *Builder[Req, Res]

New creates an action builder.

func Parallel 

func Parallel[Req, Res any](
	name string,
	builders ...*Builder[Req, Res],
) *Builder[Req, []Res]

Parallel executes all actions concurrently with the same request. Concurrency pattern: Scatter-gather using sync.WaitGroup with independent result slots.

func ParallelMap added in v0.4.0 

func ParallelMap[Req any](
	name string,
	actions ...AnyAction,
) *Builder[Req, map[string]any]

ParallelMap runs heterogeneous actions concurrently, keying the output map by each action's declared Name().

func ParallelNamed added in v0.4.0 

func ParallelNamed[Req any](
	name string,
	routes map[string]AnyAction,
) *Builder[Req, map[string]any]

ParallelNamed runs heterogeneous actions concurrently and returns results in a map keyed by branch name. Concurrency pattern: Scatter-gather using sync.WaitGroup with panic isolation on each branch.

func Pipe 

func Pipe[Req, Mid, Res any](
	name string,
	first *BuiltAction[Req, Mid],
	second *BuiltAction[Mid, Res],
) *Builder[Req, Res]

Pipe connects two actions: output of first feeds input of second. Usage: 

pipe := action.Pipe[Req, Middle, Res]("order.pipe",
    buildCheck, buildProcess)
result := pipe.Do(ctx, req)

func PipeWith added in v0.4.0 

func PipeWith[Req, Mid1, Mid2, Res any](
	name string,
	first *BuiltAction[Req, Mid1],
	transform func(context.Context, Mid1) (Mid2, error),
	second *BuiltAction[Mid2, Res],
) *Builder[Req, Res]

PipeWith connects two actions whose types don't match 1:1 by providing an inline transformation function from Mid1 to Mid2. Eliminates throwaway adapter actions and runs with 0 allocations on field extraction.

func (*Builder[Req, Res]) Add 

func (b *Builder[Req, Res]) Add(others ...AnyAction) *Builder[Req, Res]

Add composes AnyHooks and Bindings from other AnyActions (plugins) into this builder. Typed hooks are NOT composed — use Hook() directly for those. Safe to call multiple times; idempotent per unique plugin instance.

func (*Builder[Req, Res]) AnyHook 

func (b *Builder[Req, Res]) AnyHook(h ...AnyHook) *Builder[Req, Res]

AnyHook registers a type-erased hook (used by plugins: monitor, telemetry, tracing). Prefer typed Hook[Req,Res] when the action types are known.

func (*Builder[Req, Res]) Audited 

func (b *Builder[Req, Res]) Audited(logger AuditLogger, category string, detailsFn func(req Req, res Res) string) *Builder[Req, Res]

Audited attaches a generic audit hook invoked after successful action execution.

func (*Builder[Req, Res]) Build 

func (b *Builder[Req, Res]) Build() *BuiltAction[Req, Res]

func (*Builder[Req, Res]) Cache 

func (b *Builder[Req, Res]) Cache(ttl time.Duration, keyFn func(Req) string, layers ...CacheLayer[Res]) *Builder[Req, Res]

Cache adds a multi-layer cache to the action. Layers are checked in order: L1 (fast, local) → L2 (shared, e.g. Redis). On a miss, the handler runs and the result is written to all layers. On an L2 hit, L1 is back-filled automatically.

The keyFn derives a stable string cache key from the request. Keep keys short and deterministic (UUID, int, composite "tenant:id", etc.).

Example — single in-memory layer: 

action.New("catalog.list", fetchCatalog).
    Cache(30*time.Minute,
        func(_ CatalogReq) string { return "catalog:all" },
        cache.NewInMemory[[]Product](30*time.Minute),
    ).Build()

Example — L1 memory + L2 Redis: 

action.New("product.get", fetchProduct).
    Cache(10*time.Minute,
        func(r ProductReq) string { return r.ProductID },
        cache.NewInMemory[Product](5*time.Minute),
        cache.NewRedis[Product](redisClient),
    ).Build()

func (*Builder[Req, Res]) Coalesce 

func (b *Builder[Req, Res]) Coalesce(c *Coalescer, keyFn func(Req) string) *Builder[Req, Res]

Coalesce deduplicates concurrent requests using request coalescing. Unlike Dedup, Coalesce allows a caller whose context is canceled to bail out early without killing the underlying in-flight request — the in-flight request continues for other waiters.

A single *Coalescer can be shared across multiple actions: 

c := action.NewCoalescer()

productAct := action.New("product.get", fetchProduct).
    Coalesce(c, func(r ProductReq) string { return r.ProductID }).
    Build()

inventoryAct := action.New("inventory.check", checkStock).
    Coalesce(c, func(r InventoryReq) string { return r.SKU }).
    Build()

func (*Builder[Req, Res]) Compose 

func (b *Builder[Req, Res]) Compose(other *BuiltAction[Req, Res]) *Builder[Req, Res]

Compose copies runtime hooks (typed + AnyHook), bindings, and transport hints (Idempotency, SuccessStatus) from a compiled action into this builder.

Middleware-backed policies are intentionally NOT inherited. Timeout, RetryMax, ConcurrencyLimit, CacheTTL, RequiresAuth, and the Required* guards are already compiled into other.exec as closures and cannot be extracted from a BuiltAction.

func (*Builder[Req, Res]) ConcurrencyLimit 

func (b *Builder[Req, Res]) ConcurrencyLimit(limit int32) *Builder[Req, Res]

func (*Builder[Req, Res]) Dedup 

func (b *Builder[Req, Res]) Dedup(keyFn func(Req) string) *Builder[Req, Res]

Dedup prevents concurrent identical requests from executing multiple times. All callers with the same key block until the first one completes, then share its result — exactly one handler invocation per unique key at any point in time.

keyFn returns the dedup key. An empty string disables dedup for that request.

Use case — prevent N concurrent callers from all hitting the DB for the same product: 

productAct := action.New("product.price", fetchPrice).
    Dedup(func(r PriceReq) string { return r.ProductID }).
    Build()

func (*Builder[Req, Res]) Describe 

func (b *Builder[Req, Res]) Describe() *Meta

func (*Builder[Req, Res]) Description 

func (b *Builder[Req, Res]) Description(d string) *Builder[Req, Res]

func (*Builder[Req, Res]) Emits 

func (b *Builder[Req, Res]) Emits(subject string, mapper func(res Res) any) *Builder[Req, Res]

Emits registers automatic event emission upon successful action execution. Payload mapping is evaluated lazily only when an active EventPublisher is in context.

func (*Builder[Req, Res]) Example added in v0.11.0 

func (b *Builder[Req, Res]) Example(v Req) *Builder[Req, Res]

Example attaches a real, runnable request payload to the action. testkit, the flow inspector, and LLM tool specs use it to build smoke tests, CLI hints, and function-call examples without guessing.

func (*Builder[Req, Res]) Exclusive 

func (b *Builder[Req, Res]) Exclusive(m Mutex, ttl time.Duration, keyFn func(Req) string) *Builder[Req, Res]

Exclusive wraps an action with a standard distributed lock. If the lock cannot be acquired, ErrLocked is returned immediately.

func (*Builder[Req, Res]) ExclusiveFenced 

func (b *Builder[Req, Res]) ExclusiveFenced(m FencedMutex, ttl time.Duration, keyFn func(Req) string) *Builder[Req, Res]

ExclusiveFenced runs an action under an ownership-checked lease. The lease is renewed while the action is running; loss of the lease cancels the action context and returns an unavailable error rather than claiming success.

func (*Builder[Req, Res]) Hook 

func (b *Builder[Req, Res]) Hook(h ...Hook[Req, Res]) *Builder[Req, Res]

Hook registers a full typed hook struct. Use when you need more than one hook event in a single declaration.

func (*Builder[Req, Res]) HookAfter 

func (b *Builder[Req, Res]) HookAfter(fn func(ctx context.Context, req Req, res Res, err error, meta *Meta)) *Builder[Req, Res]

HookAfter runs fn after execution regardless of success or failure. Use for audit logging, always-on cleanup, unconditional metrics.

func (*Builder[Req, Res]) HookBefore 

func (b *Builder[Req, Res]) HookBefore(fn func(ctx context.Context, req Req, meta *Meta) (context.Context, error)) *Builder[Req, Res]

HookBefore runs fn before the handler. Can enrich ctx or abort with an error. Aborting returns the error immediately — the handler never runs.

func (*Builder[Req, Res]) HookBuild added in v0.13.0 

func (b *Builder[Req, Res]) HookBuild(
	fn func(meta *Meta, reqType, resType reflect.Type) bool,
) *Builder[Req, Res]

HookBuild runs fn once per action at build time. Must be pure and O(1): no I/O, no network. Return true to attach the hook, false to drop it — a dropped hook pays zero cost in Do().

reqType and resType are reflect.TypeFor[Req]() and reflect.TypeFor[Res](). Use them to filter the hook to a subset of actions.

For per-request decisions (feature flags, tenant policy, quota), use the runtime Before callback instead.

Example — attach a cost ledger hook only to actions whose response reports cost: 

.HookBuild(func(meta *action.Meta, _, resType reflect.Type) bool {
    if !resType.Implements(reflect.TypeFor[CostReporter]()) {
        return false
    }
    ledger.Register(meta.Name)
    return true
})

func (*Builder[Req, Res]) HookCacheHit 

func (b *Builder[Req, Res]) HookCacheHit(fn func(ctx context.Context, req Req, res Res, meta *Meta)) *Builder[Req, Res]

HookCacheHit runs fn when the CacheMiddleware serves a response from cache. The handler is NOT called in this case. Use for cache-hit metrics, hit-rate logging.

func (*Builder[Req, Res]) HookCacheHitEvent added in v0.16.0 

func (b *Builder[Req, Res]) HookCacheHitEvent(fn func()) *Builder[Req, Res]

HookCacheHitEvent runs fn when the cache serves a response.

func (*Builder[Req, Res]) HookCacheMiss 

func (b *Builder[Req, Res]) HookCacheMiss(fn func(ctx context.Context, req Req, meta *Meta)) *Builder[Req, Res]

HookCacheMiss runs fn when the CacheMiddleware finds no cached result. The handler will be called immediately after. Use for cache-miss metrics, warming triggers.

func (*Builder[Req, Res]) HookCacheMissEvent added in v0.16.0 

func (b *Builder[Req, Res]) HookCacheMissEvent(fn func()) *Builder[Req, Res]

HookCacheMissEvent runs fn when the cache has no response for a request.

func (*Builder[Req, Res]) HookCancel 

func (b *Builder[Req, Res]) HookCancel(fn func(ctx context.Context, req Req, meta *Meta)) *Builder[Req, Res]

HookCancel runs fn when the request context is canceled (client disconnect, upstream timeout, explicit cancel). Use for cleanup, releasing reserved resources, cancellation metrics.

func (*Builder[Req, Res]) HookCancelEvent added in v0.16.0 

func (b *Builder[Req, Res]) HookCancelEvent(fn func()) *Builder[Req, Res]

HookCancelEvent runs fn when the request is canceled.

func (*Builder[Req, Res]) HookCoalescedEvent added in v0.16.0 

func (b *Builder[Req, Res]) HookCoalescedEvent(fn func()) *Builder[Req, Res]

HookCoalescedEvent runs fn when this call joins an in-flight coalesced call.

func (*Builder[Req, Res]) HookDeduplicatedEvent added in v0.16.0 

func (b *Builder[Req, Res]) HookDeduplicatedEvent(fn func()) *Builder[Req, Res]

HookDeduplicatedEvent runs fn when a duplicate call is suppressed.

func (*Builder[Req, Res]) HookError 

func (b *Builder[Req, Res]) HookError(fn func(ctx context.Context, req Req, err error, meta *Meta)) *Builder[Req, Res]

HookError runs fn only when the handler returns a non-nil error. Use for alerting, dead-letter queues, structured error logging.

func (*Builder[Req, Res]) HookErrorEvent added in v0.16.0 

func (b *Builder[Req, Res]) HookErrorEvent(fn func(error)) *Builder[Req, Res]

HookErrorEvent runs fn when execution returns an error.

func (*Builder[Req, Res]) HookRetry 

func (b *Builder[Req, Res]) HookRetry(fn func(ctx context.Context, req Req, attempt int, err error, meta *Meta)) *Builder[Req, Res]

HookRetry runs fn before each retry attempt made by the Retry middleware. attempt starts at 1 for the first retry. Use for retry-specific logging, jitter metrics, backoff tracing.

func (*Builder[Req, Res]) HookRetryEvent added in v0.16.0 

func (b *Builder[Req, Res]) HookRetryEvent(fn func(attempt int, err error)) *Builder[Req, Res]

HookRetryEvent runs fn before each retry attempt.

func (*Builder[Req, Res]) HookSuccess added in v0.17.0 

func (b *Builder[Req, Res]) HookSuccess(fn func(ctx context.Context, req Req, res Res, meta *Meta)) *Builder[Req, Res]

HookSuccess runs fn only after a successful real execution. Use for domain-event publishing, analytics, cache warming.

func (*Builder[Req, Res]) HookSuccessEvent added in v0.17.0 

func (b *Builder[Req, Res]) HookSuccessEvent(fn func()) *Builder[Req, Res]

HookSuccessEvent runs fn after a successful real execution, except when the result is served directly from cache.

func (*Builder[Req, Res]) HookTimeout added in v0.17.0 

func (b *Builder[Req, Res]) HookTimeout(fn func(ctx context.Context, req Req, meta *Meta)) *Builder[Req, Res]

HookTimeout runs fn when execution returns context.DeadlineExceeded.

func (*Builder[Req, Res]) HookTimeoutEvent added in v0.17.0 

func (b *Builder[Req, Res]) HookTimeoutEvent(fn func()) *Builder[Req, Res]

HookTimeoutEvent runs fn when execution returns context.DeadlineExceeded.

func (*Builder[Req, Res]) Idempotent 

func (b *Builder[Req, Res]) Idempotent() *Builder[Req, Res]

func (*Builder[Req, Res]) IdempotentWithConfig 

func (b *Builder[Req, Res]) IdempotentWithConfig(cfg IdempotencyConfig) *Builder[Req, Res]

func (*Builder[Req, Res]) ImmutableWhen 

func (b *Builder[Req, Res]) ImmutableWhen(guard func(ctx context.Context, req Req) (bool, error), reason string) *Builder[Req, Res]

ImmutableWhen aborts execution with 403 Forbidden if the guard condition returns true. Useful for locking entities in terminal or processed states (e.g. settled invoices).

func (*Builder[Req, Res]) InferredResilient 

func (b *Builder[Req, Res]) InferredResilient() *Builder[Req, Res]

func (*Builder[Req, Res]) Instrument 

func (b *Builder[Req, Res]) Instrument(
	incCall func(actionName string),
	incError func(actionName string),
	recordLatency func(actionName string, ms float64),
) *Builder[Req, Res]

Instrument wires a minimal set of Prometheus-style counters via AnyHook. Counts are tracked by calling the provided inc functions — no Prometheus import required, works with any counter abstraction.

Example with Prometheus: 

calls   := prometheus.NewCounterVec(...)
errors  := prometheus.NewCounterVec(...)
latency := prometheus.NewHistogramVec(...)

act := action.New("order.create", handler).
    Instrument(
        func(name string) { calls.WithLabelValues(name).Inc() },
        func(name string) { errors.WithLabelValues(name).Inc() },
        func(name string, ms float64) { latency.WithLabelValues(name).Observe(ms / 1000) },
    ).Build()

func (*Builder[Req, Res]) Internal 

func (b *Builder[Req, Res]) Internal() *Builder[Req, Res]

Internal exposes this action only through an explicitly requested trusted service-to-service or runner contract.

func (*Builder[Req, Res]) LeaderOnly 

func (b *Builder[Req, Res]) LeaderOnly(m Mutex, ttl time.Duration) *Builder[Req, Res]

LeaderOnly restricts action execution to a single instance using a global leader key.

func (*Builder[Req, Res]) LeaderOnlyFenced 

func (b *Builder[Req, Res]) LeaderOnlyFenced(m FencedMutex, ttl time.Duration) *Builder[Req, Res]

LeaderOnlyFenced is the safe singleton-action form. It provides a renewable ownership lease, not merely a best-effort process-local convention.

func (*Builder[Req, Res]) LogCalls 

func (b *Builder[Req, Res]) LogCalls(log *slog.Logger) *Builder[Req, Res]

LogCalls injects a structured log entry for every call using the provided logger. Logs before the call (level=Debug) and after (level=Info on success, Error on failure).

This is a lightweight alternative to telemetry.New() for simple deployments. 

debugAct := action.New("order.create", handler).
    LogCalls(slog.Default()).
    Build()

func (*Builder[Req, Res]) LogSlowWhen 

func (b *Builder[Req, Res]) LogSlowWhen(d time.Duration) *Builder[Req, Res]

func (*Builder[Req, Res]) Name 

func (b *Builder[Req, Res]) Name(name string) *Builder[Req, Res]

func (*Builder[Req, Res]) Node 

func (b *Builder[Req, Res]) Node(nodeName string) *Builder[Req, Res]

func (*Builder[Req, Res]) Once 

func (b *Builder[Req, Res]) Once() *Builder[Req, Res]

Once caches the result of the first successful execution and returns it for all subsequent calls. Errors are also cached – the action will not retry. Use only for idempotent, read‑only actions (e.g., static config generation).

The first caller determines the cached result; later callers receive the same value or error without re‑executing the handler. Context cancellation is ignored after the first execution.

Example: 

action.New("config.generate", generator).
    Once().
    Route(thttp.GET("/config.json")).
    Build()

func (*Builder[Req, Res]) Public 

func (b *Builder[Req, Res]) Public() *Builder[Req, Res]

Public exposes this action through the browser/client business contract. It does not weaken authentication or authorization requirements.

func (*Builder[Req, Res]) RateLimit 

func (b *Builder[Req, Res]) RateLimit(requestsPerSecond float64, burst int) *Builder[Req, Res]

func (*Builder[Req, Res]) RateLimitDistributed 

func (b *Builder[Req, Res]) RateLimitDistributed(limiter RateLimiter, keyFn func(context.Context) string) *Builder[Req, Res]

func (*Builder[Req, Res]) RateLimitWithKey 

func (b *Builder[Req, Res]) RateLimitWithKey(rps float64, burst int, keyFn func(context.Context) string) *Builder[Req, Res]

func (*Builder[Req, Res]) RecordHistory 

func (b *Builder[Req, Res]) RecordHistory(capacity int) *Builder[Req, Res]

WithHistory attaches a ring-buffer execution history to the action AND returns the *History[Req, Res] handle for inspection. Replaces the two-return-value .WithHistory pattern with a single method that stores the handle on the builder for later retrieval.

The built action keeps the last cap records (newest-first via Snapshot).

Usage: 

paymentAct := action.New("payment.charge", chargeCard).
    RecordHistory(200).
    Build()

hist := paymentAct.History() // nil if RecordHistory was not called

func (*Builder[Req, Res]) RequireAnyFeature 

func (b *Builder[Req, Res]) RequireAnyFeature(keys ...string) *Builder[Req, Res]

RequireAnyFeature aborts with 403 Forbidden if NONE of the listed feature flags are enabled in ctx (logical OR — at least one must be on). 

betaOrPremiumAct := action.New("widget.beta", handler).
    RequireAnyFeature("beta-access", "premium-plan").
    Build()

func (*Builder[Req, Res]) RequireAnyRole 

func (b *Builder[Req, Res]) RequireAnyRole(roles ...string) *Builder[Req, Res]

RequireAnyRole aborts with 403 Forbidden if ctx contains none of the given roles. First matching role passes — order does not matter. 

approveRefundAct := action.New("refund.approve", handler).
    RequireAnyRole("admin", "finance", "support").
    Build()

func (*Builder[Req, Res]) RequireAuth 

func (b *Builder[Req, Res]) RequireAuth() *Builder[Req, Res]

RequireAuth aborts with 401 Unauthorized if ctx has no authenticated user (UserID is empty). Use when an endpoint requires login but any role is fine. 

myProfileAct := action.New("user.me", handler).
    RequireAuth().
    Build()

func (*Builder[Req, Res]) RequireCreationLimit 

func (b *Builder[Req, Res]) RequireCreationLimit(resourceName string, checkFn QuotaCheckFunc) *Builder[Req, Res]

RequireCreationLimit enforces plan quotas ONLY during new entity creation (ID == 0). Updates to existing resources bypass this check.

func (*Builder[Req, Res]) RequireFeature 

func (b *Builder[Req, Res]) RequireFeature(keys ...string) *Builder[Req, Res]

RequireFeature aborts with 403 Forbidden if ALL listed feature flags are not enabled in ctx (logical AND — every flag must be on). Feature flags are injected from the "features" JWT claim.

Single flag: 

aiCheckoutAct := action.New("checkout.ai", handler).
    RequireFeature("ai-checkout").
    Build()

Multiple flags (all must be enabled): 

premiumExportAct := action.New("export.premium", handler).
    RequireFeature("premium-plan", "data-export").
    Build()

func (*Builder[Req, Res]) RequirePermission 

func (b *Builder[Req, Res]) RequirePermission(perm string) *Builder[Req, Res]

RequirePermission aborts with 403 Forbidden if ctx does not contain the given fine-grained permission string. Permissions are injected from the "perms" JWT claim. 

exportAct := action.New("data.export", handler).
    RequirePermission("data:export").
    Build()

func (*Builder[Req, Res]) RequireRole 

func (b *Builder[Req, Res]) RequireRole(role string) *Builder[Req, Res]

RequireRole aborts with 403 Forbidden if ctx does not contain the given role. Roles are injected by the JWT middleware from the "roles" claim. 

deleteOrderAct := action.New("order.delete", handler).
    RequireRole("admin").
    Route(thttp.DELETE("/api/v1/orders/{id}")).
    Build()

func (*Builder[Req, Res]) RequireTenant 

func (b *Builder[Req, Res]) RequireTenant() *Builder[Req, Res]

RequireTenant aborts with 401 Unauthorized if ctx has no tenant ID. Use to guard multi-tenant endpoints from unauthenticated callers. 

tenantOrderAct := action.New("order.list", handler).
    RequireTenant().
    Build()

func (*Builder[Req, Res]) Resilient 

func (b *Builder[Req, Res]) Resilient(cfg ResilienceConfig) *Builder[Req, Res]

func (*Builder[Req, Res]) Retry 

func (b *Builder[Req, Res]) Retry(
	maxRetry int,
	backoff func(attempt int) time.Duration,
) *Builder[Req, Res]

Retry retries only transient errors detected by xerr.IsTransient.

func (*Builder[Req, Res]) RetryAll added in v0.3.0 

func (b *Builder[Req, Res]) RetryAll(
	maxRetry int,
	backoff func(attempt int) time.Duration,
) *Builder[Req, Res]

RetryAll retries on ANY non-nil error. Use only for idempotent jobs, scripts, and safe batch operations.

func (*Builder[Req, Res]) RetryIf added in v0.3.0 

func (b *Builder[Req, Res]) RetryIf(
	maxRetry int,
	backoff func(attempt int) time.Duration,
	predicate RetryPredicate,
) *Builder[Req, Res]

RetryIf retries when the supplied predicate returns true.

func (*Builder[Req, Res]) Route 

func (b *Builder[Req, Res]) Route(bs ...Binding) *Builder[Req, Res]

func (*Builder[Req, Res]) SuccessStatus 

func (b *Builder[Req, Res]) SuccessStatus(code int) *Builder[Req, Res]

func (*Builder[Req, Res]) System 

func (b *Builder[Req, Res]) System() *Builder[Req, Res]

System marks a framework or operations-plane action. It is excluded from public and trusted business contracts.

func (*Builder[Req, Res]) Tag 

func (b *Builder[Req, Res]) Tag(tags ...string) *Builder[Req, Res]

func (*Builder[Req, Res]) Timeout 

func (b *Builder[Req, Res]) Timeout(d time.Duration) *Builder[Req, Res]

func (*Builder[Req, Res]) TrackPIIAccess 

func (b *Builder[Req, Res]) TrackPIIAccess(tracker PIITracker, purpose string) *Builder[Req, Res]

TrackPIIAccess attaches a non-blocking hook recording the purpose of PII data access.

func (*Builder[Req, Res]) Transactional 

func (b *Builder[Req, Res]) Transactional(runner TxRunner) *Builder[Req, Res]

func (*Builder[Req, Res]) Use 

func (b *Builder[Req, Res]) Use(m Middleware[Req, Res]) *Builder[Req, Res]

func (*Builder[Req, Res]) UseFirst added in v0.14.1 

func (b *Builder[Req, Res]) UseFirst(m Middleware[Req, Res]) *Builder[Req, Res]

UseFirst inserts a middleware at the outermost position of the chain. Middlewares added with Use wrap earlier ones from the inside, so a wrapper that must observe every inner failure — timeouts, exhausted retries, cancellations — has to be installed with UseFirst. Use for recovery, circuit-breaking, or measurement.

func (*Builder[Req, Res]) UseWithDispatcher 

func (b *Builder[Req, Res]) UseWithDispatcher(m DispatcherMiddleware[Req, Res]) *Builder[Req, Res]

func (*Builder[Req, Res]) Validate 

func (b *Builder[Req, Res]) Validate(fn func(ctx context.Context, req Req) error) *Builder[Req, Res]

Validate adds a request validation middleware to the action construction pipeline.

func (*Builder[Req, Res]) WithDSL added in v0.7.0 

func (b *Builder[Req, Res]) WithDSL(m DSLModifiers) *Builder[Req, Res]

WithDSL applies every non-zero field of m to the builder. Designed for the declarative overlay pattern: parse a .flow / YAML manifest once at boot, then for each registered action call WithDSL with its matching modifiers.

Ordering matters — scope is applied first so later auth/permission wrappers see the correct default scope; hooks are applied last so they wrap the fully configured action.

func (*Builder[Req, Res]) WithHistory 

func (b *Builder[Req, Res]) WithHistory(capacity int) (*Builder[Req, Res], *History[Req, Res])

func (*Builder[Req, Res]) WithProfile 

func (b *Builder[Req, Res]) WithProfile(profile Profile) *Builder[Req, Res]

WithProfile applies a policy bundle before optional action-specific overrides. The returned builder remains fully fluent, so exceptional actions can explicitly refine timeout, route, or idempotency configuration.

type BuiltAction 

type BuiltAction[Req, Res any] struct {
	// contains filtered or unexported fields
}

func (*BuiltAction[Req, Res]) AddAnyHook 

func (a *BuiltAction[Req, Res]) AddAnyHook(h ...AnyHook)

func (*BuiltAction[Req, Res]) ApplyDSL added in v0.7.0 

func (a *BuiltAction[Req, Res]) ApplyDSL(m DSLModifiers) AnyAction

ApplyDSL bridges the generic builder to the non-generic DSLApplier interface.

func (*BuiltAction[Req, Res]) CloneWithHooks added in v0.15.0 

func (a *BuiltAction[Req, Res]) CloneWithHooks(hooks ...AnyHook) AnyAction

CloneWithHooks returns a fully independent action: metadata, executor, bindings, and existing hooks are copied, and the new hooks are appended.

This is the type-erased clone primitive: it is what registries and plugin systems use, because they hold AnyAction and cannot call the generic ToBuilder. Preserving generic Req/Res types keeps the hot path zero-allocation.

func (*BuiltAction[Req, Res]) Describe 

func (a *BuiltAction[Req, Res]) Describe() *Meta

func (*BuiltAction[Req, Res]) Do 

func (a *BuiltAction[Req, Res]) Do(ctx context.Context, req Req) (res Res, err error)

func (*BuiltAction[Req, Res]) DoAny added in v0.4.0 

func (a *BuiltAction[Req, Res]) DoAny(ctx context.Context, req any) (any, error)

func (*BuiltAction[Req, Res]) ExecuteDecoded 

func (a *BuiltAction[Req, Res]) ExecuteDecoded(ctx context.Context, decode DecodeFunc) (any, error)

func (*BuiltAction[Req, Res]) GetAnyHooks 

func (a *BuiltAction[Req, Res]) GetAnyHooks() []AnyHook

func (*BuiltAction[Req, Res]) GetBindings 

func (a *BuiltAction[Req, Res]) GetBindings() []Binding

func (*BuiltAction[Req, Res]) GetMeta 

func (a *BuiltAction[Req, Res]) GetMeta() *Meta

func (*BuiltAction[Req, Res]) History 

func (a *BuiltAction[Req, Res]) History() *History[Req, Res]

func (*BuiltAction[Req, Res]) OnCacheHit 

func (a *BuiltAction[Req, Res]) OnCacheHit(ctx context.Context, req Req, res Res)

func (*BuiltAction[Req, Res]) OnCacheMiss 

func (a *BuiltAction[Req, Res]) OnCacheMiss(ctx context.Context, req Req)

func (*BuiltAction[Req, Res]) OnCoalesced 

func (a *BuiltAction[Req, Res]) OnCoalesced(ctx context.Context, req Req)

func (*BuiltAction[Req, Res]) OnDeduplicated 

func (a *BuiltAction[Req, Res]) OnDeduplicated(ctx context.Context, req Req)

func (*BuiltAction[Req, Res]) OnRetry 

func (a *BuiltAction[Req, Res]) OnRetry(ctx context.Context, req Req, attempt int, err error)

func (*BuiltAction[Req, Res]) ReqPayload 

func (a *BuiltAction[Req, Res]) ReqPayload() any

func (*BuiltAction[Req, Res]) ResPayload 

func (a *BuiltAction[Req, Res]) ResPayload() any

func (*BuiltAction[Req, Res]) String 

func (a *BuiltAction[Req, Res]) String() string

func (*BuiltAction[Req, Res]) ToBuilder added in v0.4.0 

func (b *BuiltAction[Req, Res]) ToBuilder() *Builder[Req, Res]

ToBuilder creates a mutable configuration state from an existing action.

type BuiltSaga 

type BuiltSaga[Req, Res any] struct {
	// contains filtered or unexported fields
}

BuiltSaga is the immutable, executable Saga.

func (*BuiltSaga[Req, Res]) AsAction added in v0.4.0 

func (s *BuiltSaga[Req, Res]) AsAction() *BuiltAction[Req, SagaResult[Res]]

AsAction converts the compiled Saga into a standard *BuiltAction, allowing it to be routed over HTTP, CLI, MCP, or composed via action.Pipe / action.Parallel.

func (*BuiltSaga[Req, Res]) Do 

func (s *BuiltSaga[Req, Res]) Do(ctx context.Context, req Req) (SagaResult[Res], error)

Do executes the Saga. If a mandatory step fails, it automatically runs Undo functions in reverse order.

type CacheConfig 

type CacheConfig[Req, Res any] struct {
	KeyFunc func(Req) string
	Layers  []CacheLayer[Res]
	TTL     time.Duration
	Timeout time.Duration
}

type CacheLayer 

type CacheLayer[V any] interface {
	Get(ctx context.Context, key string) (val V, hit bool, err error)
	Set(ctx context.Context, key string, val V, ttl time.Duration) error
}

type Coalescer 

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

Coalescer coordinates in-flight request sharing across different actions or callers.

func NewCoalescer 

func NewCoalescer() *Coalescer

NewCoalescer creates a new request coalescing coordinator.

func (*Coalescer) Do 

func (c *Coalescer) Do(ctx context.Context, key string, fn func(context.Context) (any, error)) (val any, shared bool, err error)

type Composite added in v0.6.0 

type Composite interface {
	Children() []AnyAction
}

type DSLApplier added in v0.7.0 

type DSLApplier interface {
	ApplyDSL(m DSLModifiers) AnyAction
}

DSLApplier is implemented by every BuiltAction. It lets a non-generic registry ([]AnyAction) delegate modifier application to the generic builder machinery without knowing Req/Res.

type DSLModifiers added in v0.7.0 

type DSLModifiers struct {
	// Scope — "public" | "internal" | "system". Empty = leave unchanged.
	Scope string

	// Auth
	RequiresAuth bool
	Roles        []string
	Permissions  []string
	Features     []string

	// Resilience
	RateLimit     float64
	Burst         int
	Concurrency   int32
	Timeout       time.Duration
	RetryMax      int
	RetryBackoff  func(attempt int) time.Duration
	RetryIf       RetryPredicate
	Idempotent    bool
	Idempotency   *IdempotencyConfig
	CacheTTL      time.Duration
	CacheKeyFnRaw any // func(Req) string — type asserted inside WithDSL

	// Metadata
	Tags          []string
	SuccessStatus int

	// Extra hooks (HITL, budget guard, audit, custom middleware).
	// Passed straight through to Builder.AnyHook.
	Hooks []AnyHook
}

DSLModifiers is the runtime subset of declarative metadata that a declarative overlay (a .flow file, a YAML manifest, etc.) can apply to a BuiltAction after construction. Fields map 1:1 to Builder methods that already exist; the struct exists only to bridge generic-typed Builders with non-generic action registries.

Zero value is "apply nothing" — leave every field at its default to skip.

type DecodeFunc 

type DecodeFunc func(v any) error

DecodeFunc allows transports to inject data directly into the concrete type.

type Describable 

type Describable interface {
	Describe() *Meta
}

Describable handles the COLD path — boot, discovery, routing, and CLI help.

type DispatcherMiddleware 

type DispatcherMiddleware[Req, Res any] func(next Fn[Req, Res], hooks HookDispatcher[Req, Res]) Fn[Req, Res]

func AdmissionMiddleware 

func AdmissionMiddleware[Req, Res any](admission Admission) DispatcherMiddleware[Req, Res]

AdmissionMiddleware applies an external admission policy around execution. It keeps the policy outside Builder while allowing typed middleware composition and automatic request/response inference.

func CacheMiddleware 

func CacheMiddleware[Req, Res any](cfg CacheConfig[Req, Res]) DispatcherMiddleware[Req, Res]

CacheMiddleware implements the Read-Through / Write-Behind pattern with isolated singleflight execution.

func CoalesceMiddleware 

func CoalesceMiddleware[Req, Res any](coalescer *Coalescer, actionName string, keyFn func(Req) string) DispatcherMiddleware[Req, Res]

CoalesceMiddleware wraps an action with shared request coalescing.

func Deduplicate 

func Deduplicate[Req, Res any](keyFn func(Req) string) DispatcherMiddleware[Req, Res]

Deduplicate returns a middleware that collapses concurrent identical requests into a single execution, sharing the result with all waiting callers.

func RetryMiddleware 

func RetryMiddleware[Req, Res any](
	maxRetry int,
	backoff func(attempt int) time.Duration,
) DispatcherMiddleware[Req, Res]

RetryMiddleware is the backward-compatible, safe default retry middleware. It retries only transient errors detected by xerr.IsTransient.

func RetryWithPredicateMiddleware added in v0.3.0 

func RetryWithPredicateMiddleware[Req, Res any](
	maxRetry int,
	backoff func(attempt int) time.Duration,
	predicate RetryPredicate,
) DispatcherMiddleware[Req, Res]

RetryWithPredicateMiddleware retries errors when predicate returns true. If predicate is nil, it falls back to DefaultRetryPredicate. If backoff is nil, it defaults to ConstantBackoff(0). If maxRetry is negative, it is clamped to 0.

func SmartResilience 

func SmartResilience[Req, Res any](name string) DispatcherMiddleware[Req, Res]

SmartResilience automatically applies intelligent backoff and circuit breaking based entirely on the xerr.Kind of the returned error. It delegates to the universal RetryIf middleware to ensure 100% hook/telemetry fidelity.

type Executable 

type Executable interface {
	ExecuteDecoded(ctx context.Context, decode DecodeFunc) (any, error)
}

Executable handles the HOT path — execution only.

type FencedMutex 

type FencedMutex interface {
	Acquire(ctx context.Context, key string, ttl time.Duration) (lease LockLease, acquired bool, err error)
	Renew(ctx context.Context, lease LockLease, ttl time.Duration) (renewed bool, err error)
	Release(ctx context.Context, lease LockLease) (released bool, err error)
}

FencedMutex is the production-safe distributed coordination contract. A lease belongs to exactly one owner, can be renewed only by that owner, and can be released only by that owner.

type Fn 

type Fn[Req, Res any] func(context.Context, Req) (Res, error)

type History 

type History[Req, Res any] struct {
	// contains filtered or unexported fields
}

func NewHistory 

func NewHistory[Req, Res any](capacity int) *History[Req, Res]

func (*History[Req, Res]) Push 

func (h *History[Req, Res]) Push(rec Record[Req, Res])

func (*History[Req, Res]) Snapshot 

func (h *History[Req, Res]) Snapshot() []Record[Req, Res]

type Hook 

type Hook[Req, Res any] struct {
	OnBuild func(meta *Meta, reqType, resType reflect.Type) bool

	Before func(ctx context.Context, req Req, meta *Meta) (context.Context, error)
	After  func(ctx context.Context, req Req, res Res, err error, meta *Meta)

	// OnSuccess fires after a successful real execution. It does not fire for
	// a result returned directly from cache.
	OnSuccess func(ctx context.Context, req Req, res Res, meta *Meta)

	// OnTimeout fires before OnError when the returned error matches
	// context.DeadlineExceeded.
	OnTimeout func(ctx context.Context, req Req, meta *Meta)
	OnError   func(ctx context.Context, req Req, err error, meta *Meta)

	OnRetry        func(ctx context.Context, req Req, attempt int, err error, meta *Meta)
	OnCacheHit     func(ctx context.Context, req Req, res Res, meta *Meta)
	OnCacheMiss    func(ctx context.Context, req Req, meta *Meta)
	OnCoalesced    func(ctx context.Context, req Req, meta *Meta)
	OnDeduplicated func(ctx context.Context, req Req, meta *Meta)
	OnCancel       func(ctx context.Context, req Req, meta *Meta)
}

Hook is a strongly typed hook set for one request/response pair. All fields are optional. Hook panics are isolated by the action runtime.

type HookDispatcher 

type HookDispatcher[Req, Res any] interface {
	OnCacheHit(ctx context.Context, req Req, res Res)
	OnCacheMiss(ctx context.Context, req Req)
	OnRetry(ctx context.Context, req Req, attempt int, err error)
	OnCoalesced(ctx context.Context, req Req)
	OnDeduplicated(ctx context.Context, req Req)
}

type HookProvider 

type HookProvider interface {
	GetAnyHooks() []AnyHook
}

type IdempotencyClaim 

type IdempotencyClaim struct {
	State IdempotencyClaimState
	Token string
	Entry IdempotencyEntry
}

type IdempotencyClaimState 

type IdempotencyClaimState uint8
const (
	IdempotencyClaimAcquired IdempotencyClaimState = iota + 1
	IdempotencyClaimCompleted
	IdempotencyClaimInProgress
	IdempotencyClaimConflict
)

type IdempotencyConfig 

type IdempotencyConfig struct {
	Enabled   bool
	TTL       time.Duration
	KeyHeader string
	KeyFunc   func(body []byte) string
	LeaseTTL  time.Duration
	Store     IdempotencyStore
}

func (IdempotencyConfig) EffectiveLeaseTTL 

func (c IdempotencyConfig) EffectiveLeaseTTL() time.Duration

func (IdempotencyConfig) Header 

func (c IdempotencyConfig) Header() string

type IdempotencyCoordinator 

type IdempotencyCoordinator interface {
	IdempotencyStore
	Claim(ctx context.Context, key, requestHash string, leaseTTL time.Duration) (IdempotencyClaim, error)
	Complete(ctx context.Context, key, token string, entry IdempotencyEntry, ttl time.Duration) error
	Release(ctx context.Context, key, token string) error
}

type IdempotencyEntry 

type IdempotencyEntry struct {
	Status      int               `json:"status"`
	Body        []byte            `json:"body"`
	Headers     map[string]string `json:"headers,omitempty"`
	StoredAt    time.Time         `json:"stored_at"`
	RequestHash string            `json:"request_hash"`
}

type IdempotencyStore 

type IdempotencyStore interface {
	Get(ctx context.Context, key string) (IdempotencyEntry, bool)
	Set(ctx context.Context, key string, entry IdempotencyEntry, ttl time.Duration)
}

type Library added in v0.12.0 

type Library struct {
	Name        string
	Description string
	Actions     []AnyAction
	Sources     []AnyStreamAction
	Operators   []NamedOperator
	Hooks       []AnyHook
	Aliases     []Alias
	Overrides   []string // Used by test harnesses to shadow existing actions
}

Library is a named group of actions, sources, and operators.

Hooks declared here are applied to every Action and every Source at Register time via CloneWithHooks. The originals are never mutated; the registry holds private clones. Operators have no lifecycle, so Library-level hooks do not apply to them.

func Of added in v0.12.0 

func Of(actions ...AnyAction) Library

Of is the terse constructor used by tests and inline libraries.

type LoadShedConfig 

type LoadShedConfig struct {
	MaxCPU        float64
	MaxGoroutines int
}

type LockLease 

type LockLease struct {
	Key   string
	Owner string
	Fence int64
}

LockLease proves ownership of a distributed lock. Fence is monotonically increasing for a key and must be carried to any downstream system that can reject stale writers.

func LeaseFromContext added in v0.2.0 

func LeaseFromContext(ctx context.Context) (LockLease, bool)

LeaseFromContext retrieves the active LockLease from the execution context.

type MemoryIdempotencyStore 

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

func NewMemoryIdempotencyStore 

func NewMemoryIdempotencyStore(defTTL time.Duration) *MemoryIdempotencyStore

func (*MemoryIdempotencyStore) Claim added in v0.2.0 

func (s *MemoryIdempotencyStore) Claim(ctx context.Context, key, requestHash string, leaseTTL time.Duration) (IdempotencyClaim, error)

func (*MemoryIdempotencyStore) Complete added in v0.2.0 

func (s *MemoryIdempotencyStore) Complete(_ context.Context, key, token string, entry IdempotencyEntry, ttl time.Duration) error

func (*MemoryIdempotencyStore) Get 

func (s *MemoryIdempotencyStore) Get(_ context.Context, key string) (IdempotencyEntry, bool)

func (*MemoryIdempotencyStore) Release added in v0.2.0 

func (s *MemoryIdempotencyStore) Release(_ context.Context, key, token string) error

func (*MemoryIdempotencyStore) Set 

func (s *MemoryIdempotencyStore) Set(_ context.Context, key string, entry IdempotencyEntry, ttl time.Duration)

type MemoryRateLimiterTestHelper added in v0.15.0 

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

func NewMemoryRateLimiterForTest added in v0.15.0 

func NewMemoryRateLimiterForTest(rps float64, burst int, ttl time.Duration, maxCapacity int) *MemoryRateLimiterTestHelper

func (*MemoryRateLimiterTestHelper) Allow added in v0.15.0 

func (h *MemoryRateLimiterTestHelper) Allow(ctx context.Context, key string) (bool, error)

func (*MemoryRateLimiterTestHelper) Has added in v0.15.0 

func (h *MemoryRateLimiterTestHelper) Has(key string) bool

func (*MemoryRateLimiterTestHelper) Len added in v0.15.0 

func (h *MemoryRateLimiterTestHelper) Len() int

type MessageRes 

type MessageRes struct {
	Message string `json:"message"`
}

MessageRes is a standard DTO for actions that only need to return a text message. Using a strongly-typed struct instead of map[string]string ensures precise SDK generation and clean OpenAPI documentation.

type Meta 

type Meta struct {
	Name             string            `json:"name"`
	Description      string            `json:"description,omitempty"`
	Node             string            `json:"node,omitempty"`
	Tags             []string          `json:"tags,omitempty"`
	Scope            ActionScope       `json:"scope,omitempty"`
	Idempotency      IdempotencyConfig `json:"idempotency"`
	SuccessStatus    int               `json:"success_status,omitempty"`
	LogSlowThreshold time.Duration     `json:"log_slow_threshold,omitempty"`
	Example          any               `json:"example,omitempty"`

	RequiredRoles       []string `json:"required_roles,omitempty"`
	RequiredPermissions []string `json:"required_permissions,omitempty"`
	RequiredFeatures    []string `json:"required_features,omitempty"`
	RequiresAuth        bool     `json:"requires_auth,omitempty"`

	RetryMax         int           `json:"retry_max,omitempty"`
	Timeout          time.Duration `json:"timeout,omitempty"`
	ConcurrencyLimit int32         `json:"concurrency_limit,omitempty"`
	RateLimit        string        `json:"rate_limit,omitempty"`
	CacheTTL         time.Duration `json:"cache_ttl,omitempty"`
	Deduplicated     bool          `json:"deduplicated,omitempty"`
	Coalesced        bool          `json:"coalesced,omitempty"`
}

func (*Meta) IsInternal 

func (m *Meta) IsInternal() bool

IsInternal reports whether the action belongs only to trusted callers.

func (*Meta) IsPublic 

func (m *Meta) IsPublic() bool

IsPublic reports whether the action is part of the public business contract. The zero value is public for concise ordinary client actions.

func (*Meta) IsSystem 

func (m *Meta) IsSystem() bool

IsSystem reports whether the action belongs to the framework/operations plane.

func (*Meta) MatchesNode added in v0.7.0 

func (m *Meta) MatchesNode(policy NodeFilterPolicy) bool

func (Meta) String 

func (m Meta) String() string

type Middleware 

type Middleware[Req, Res any] func(Fn[Req, Res]) Fn[Req, Res]

func Adaptive 

func Adaptive[Req, Res any](name string, cfg AdaptiveConfig) Middleware[Req, Res]

Adaptive wraps an action with a dynamic timeout and a stateful circuit breaker.

func AdaptiveLoadShedding 

func AdaptiveLoadShedding[Req, Res any](stats SystemStats, cfg LoadShedConfig, p Priority) Middleware[Req, Res]

AdaptiveLoadShedding drops traffic instantly if the server is physically choking.

func ConcurrencyLimitMiddleware 

func ConcurrencyLimitMiddleware[Req, Res any](limit int32) Middleware[Req, Res]

func HistoryMiddleware 

func HistoryMiddleware[Req, Res any](hist *History[Req, Res]) Middleware[Req, Res]

func IdempotencyMiddleware added in v0.16.0 

func IdempotencyMiddleware[Req, Res any](store IdempotencyStore, cfg IdempotencyConfig) Middleware[Req, Res]

IdempotencyMiddleware collapses concurrent identical requests into a single handler execution and replays the stored response for later calls.

The request hash is computed before every store lookup so a reused key with a different payload returns Conflict instead of silently replaying the old response.

func SlowLogMiddleware added in v0.2.0 

func SlowLogMiddleware[Req, Res any](threshold time.Duration, actionName string) Middleware[Req, Res]

func TimeoutMiddleware 

func TimeoutMiddleware[Req, Res any](d time.Duration) Middleware[Req, Res]

TimeoutMiddleware uses standard Middleware.

type Mutex 

type Mutex interface {
	TryLock(ctx context.Context, key string, ttl time.Duration) (bool, error)
	Unlock(ctx context.Context, key string) error
}

Mutex defines a standard distributed lock contract for critical sections. For systems requiring protection against stale writers during process pauses, prefer FencedMutex via ExclusiveFenced.

type NamedOperator added in v0.19.0 

type NamedOperator struct {
	Name        string
	Description string
	InType      reflect.Type
	OutType     reflect.Type
	Params      []ParamSpec
	Build       func(params map[string]any) (StreamOperator, error)
}

NamedOperator binds metadata to a StreamOperator builder.

func (NamedOperator) Clone added in v0.19.0 

func (n NamedOperator) Clone() NamedOperator

type NodeFilterPolicy added in v0.7.0 

type NodeFilterPolicy struct {
	ActiveNode              string
	DefaultNode             string
	AllowUntaggedEverywhere bool
}

type PIITracker 

type PIITracker interface {
	TrackPIIAccess(ctx context.Context, purpose string, actionName string)
}

PIITracker defines the minimal contract for personal data access tracking.

type ParamSpec added in v0.19.0 

type ParamSpec struct {
	Name    string
	Type    string
	Default any
	Usage   string
	Enum    []string // meaningful only when Type == "enum"
}

ParamSpec describes one external parameter of a NamedOperator.

type Priority 

type Priority uint8
const (
	PriorityCritical Priority = 0 // Payments, Logins
	PriorityNormal   Priority = 1 // Standard CRUD
	PriorityLow      Priority = 2 // Background syncs, Exports
)

type Profile 

type Profile struct {
	Tags                []string
	Scope               ActionScope
	Timeout             time.Duration
	ConcurrencyLimit    int32
	SuccessStatus       int
	RequireAuth         bool
	RequiredPermissions []string
	Idempotency         *IdempotencyConfig
}

Profile is a named bundle of action metadata and middleware defaults. It deliberately excludes a route and description: those are action-specific contract details and should remain visible at registration sites.

func AuthenticatedReadProfile 

func AuthenticatedReadProfile(permission string, tags ...string) Profile

AuthenticatedReadProfile is the conservative default for an authenticated, read-only query. It intentionally does not enable idempotency because no side effect exists to replay.

func IdempotentCommandProfile 

func IdempotentCommandProfile(permission string, tags ...string) Profile

IdempotentCommandProfile is the default for a mutation that can safely replay the same request key. Business code remains responsible for durable transactions and external-side-effect coordination.

func InternalEventProfile 

func InternalEventProfile(tags ...string) Profile

InternalEventProfile is for service-to-service ingestion routes. It requires an authenticated transport identity and idempotency, but makes no claim about the application-specific identity verifier installed by the service.

type Proxy added in v0.6.0 

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

Proxy wraps an AnyAction and allows it to be atomically hot-swapped at runtime with zero downtime and zero locks on the hot path.

func NewProxy added in v0.6.0 

func NewProxy(initial AnyAction) *Proxy

NewProxy creates a new hot-swappable proxy initialized with a starting action.

func (*Proxy) AddAnyHook added in v0.6.0 

func (p *Proxy) AddAnyHook(hooks ...AnyHook)

AddAnyHook applies variadic hooks to the active action.

func (*Proxy) Children added in v0.6.0 

func (p *Proxy) Children() []AnyAction

Children satisfies the Composite interface by unwrapping the active action.

func (*Proxy) CloneWithHooks added in v0.15.0 

func (p *Proxy) CloneWithHooks(hooks ...AnyHook) AnyAction

func (*Proxy) Current added in v0.6.0 

func (p *Proxy) Current() AnyAction

Current returns the active action or nil if uninitialized.

func (*Proxy) Describe added in v0.6.0 

func (p *Proxy) Describe() *Meta

Describe delegates metadata description to the active action.

func (*Proxy) DoAny added in v0.6.0 

func (p *Proxy) DoAny(ctx context.Context, req any) (any, error)

DoAny delegates execution to the currently active action.

func (*Proxy) ExecuteDecoded added in v0.6.0 

func (p *Proxy) ExecuteDecoded(ctx context.Context, decodeFn DecodeFunc) (any, error)

ExecuteDecoded delegates execution using the kernel's DecodeFunc type.

func (*Proxy) GetAnyHooks added in v0.6.0 

func (p *Proxy) GetAnyHooks() []AnyHook

GetAnyHooks returns the hooks from the active action.

func (*Proxy) GetBindings added in v0.6.0 

func (p *Proxy) GetBindings() []Binding

GetBindings returns typed bindings from the active action.

func (*Proxy) Swap added in v0.6.0 

func (p *Proxy) Swap(newAction AnyAction)

Swap atomically replaces the underlying action with a new one.

type QuotaCheckFunc 

type QuotaCheckFunc func(ctx context.Context, tenantID string) (current int64, limit int64, err error)

type RateLimiter 

type RateLimiter interface {
	Allow(ctx context.Context, key string) (bool, error)
}

type Record 

type Record[Req, Res any] struct {
	Time     time.Time
	Duration time.Duration
	Req      Req
	Res      Res
	Err      error
}

type Registry added in v0.12.0 

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

Registry is an immutable lookup table assembled from one or more Library values. It never mutates actions: hooks and bindings must be attached before Build(), and library-level hooks are applied via CloneWithHooks so originals stay clean.

func MustNewRegistry added in v0.12.1 

func MustNewRegistry(libs ...Library) *Registry

func NewRegistry added in v0.12.0 

func NewRegistry(libs ...Library) (*Registry, error)

NewRegistry assembles the given libraries into a single Registry.

func (*Registry) Actions added in v0.12.0 

func (r *Registry) Actions() []AnyAction

func (*Registry) Get added in v0.12.0 

func (r *Registry) Get(name string) (AnyAction, bool)

func (*Registry) GetOperator added in v0.19.0 

func (r *Registry) GetOperator(name string) (NamedOperator, bool)

func (*Registry) GetStream added in v0.19.0 

func (r *Registry) GetStream(name string) (AnyStreamAction, bool)

func (*Registry) Len added in v0.12.0 

func (r *Registry) Len() int

func (*Registry) Names added in v0.19.0 

func (r *Registry) Names() []string

type ResilienceConfig 

type ResilienceConfig struct {
	MaxRetries    int
	Backoff       func(attempt int) time.Duration
	Predicate     RetryPredicate
	Timeout       time.Duration
	MaxConcurrent int32
	Adaptive      *AdaptiveConfig
}

type RetryPredicate added in v0.3.0 

type RetryPredicate func(err error) bool

RetryPredicate determines whether a given error warrants an execution retry.

type SagaBuilder 

type SagaBuilder[Req, Res any] struct {
	// contains filtered or unexported fields
}

SagaBuilder constructs a distributed transaction pipeline.

func NewSaga 

func NewSaga[Req, Res any](name string) *SagaBuilder[Req, Res]

NewSaga initializes an in-memory Saga pipeline.

func (*SagaBuilder[Req, Res]) AddOptionalStep 

func (s *SagaBuilder[Req, Res]) AddOptionalStep(
	name string,
	do func(context.Context, Req) (Res, error),
	undo func(context.Context, Req) error,
) *SagaBuilder[Req, Res]

AddOptionalStep appends a step that skips on failure without triggering a Saga rollback.

func (*SagaBuilder[Req, Res]) AddStep 

func (s *SagaBuilder[Req, Res]) AddStep(
	name string,
	do func(context.Context, Req) (Res, error),
	undo func(context.Context, Req) error,
) *SagaBuilder[Req, Res]

AddStep appends a mandatory (Do, Undo) pair.

func (*SagaBuilder[Req, Res]) Build 

func (s *SagaBuilder[Req, Res]) Build() *BuiltSaga[Req, Res]

Build compiles the Saga.

type SagaResult 

type SagaResult[Res any] struct {
	Saga       string       `json:"saga"`
	Success    bool         `json:"success"`
	Output     Res          `json:"output,omitempty"`
	Steps      []StepResult `json:"steps"`
	Error      string       `json:"error,omitempty"`
	RolledBack bool         `json:"rolled_back,omitempty"`
	DurationMs int64        `json:"duration_ms"`
}

SagaResult captures the full execution audit and output of the Saga.

type SagaStep 

type SagaStep[Req, Res any] struct {
	Name     string
	Do       func(context.Context, Req) (Res, error)
	Undo     func(context.Context, Req) error
	Optional bool
}

SagaStep represents a single operation and its compensating rollback.

type StateEntity 

type StateEntity interface {
	GetState() string
	SetState(string)
}

StateEntity defines an interface for structs that possess a lifecycle state.

type StateMachineBuilder 

type StateMachineBuilder[Req StateEntity, Res any] struct {
	// contains filtered or unexported fields
}

StateMachineBuilder provides a fluent DSL for building state-guarded actions.

func NewStateMachine 

func NewStateMachine[Req StateEntity, Res any](name string, exec Fn[Req, Res]) *StateMachineBuilder[Req, Res]

NewStateMachine creates an action that strictly guards state transitions.

func (*StateMachineBuilder[Req, Res]) Allow 

func (sm *StateMachineBuilder[Req, Res]) Allow(from string, to ...string) *StateMachineBuilder[Req, Res]

Allow maps a valid state transition. Example: .Allow("pending", "paid", "canceled")

func (*StateMachineBuilder[Req, Res]) Build 

func (sm *StateMachineBuilder[Req, Res]) Build() *BuiltAction[Req, Res]

Build compiles the State Machine into a standard Nexss Builder, injecting the validation middleware automatically.

type StepResult 

type StepResult struct {
	Step       string `json:"step"`
	Success    bool   `json:"success"`
	Skipped    bool   `json:"skipped,omitempty"`
	Error      string `json:"error,omitempty"`
	DurationMs int64  `json:"duration_ms"`
}

StepResult captures execution metadata for a single Saga step.

type StreamAction 

type StreamAction[Req, T any] struct {
	// contains filtered or unexported fields
}

func NewStream 

func NewStream[Req, T any](name string, h StreamHandler[Req, T]) *StreamAction[Req, T]

func (*StreamAction[Req, T]) AddAnyHook added in v0.19.0 

func (a *StreamAction[Req, T]) AddAnyHook(h ...AnyHook)

AddAnyHook appends erased hooks. They fire with the stream's input request and the resulting iterator as the "response".

func (*StreamAction[Req, T]) AnyHook added in v0.19.0 

func (a *StreamAction[Req, T]) AnyHook(h ...AnyHook) *StreamAction[Req, T]

AnyHook attaches a type-erased hook directly to the stream.

func (*StreamAction[Req, T]) CloneWithHooks added in v0.19.0 

func (a *StreamAction[Req, T]) CloneWithHooks(hooks ...AnyHook) AnyStreamAction

CloneWithHooks returns a new StreamAction carrying the same handler, bindings, and existing hooks, plus the ones passed in. The receiver is never modified. Used by registries to apply library-level hooks without mutating originals.

func (*StreamAction[Req, T]) Describe added in v0.18.0 

func (a *StreamAction[Req, T]) Describe() *Meta

func (*StreamAction[Req, T]) Do 

func (a *StreamAction[Req, T]) Do(ctx context.Context, req Req) (seq iter.Seq2[T, error], err error)

func (*StreamAction[Req, T]) DoStreamAny added in v0.18.0 

func (a *StreamAction[Req, T]) DoStreamAny(ctx context.Context, req any) (AnyStream, error)

DoStreamAny is the dynamic integration boundary used by Flow and transports. A mismatched request is converted to the zero value rather than panicking, matching the kernel's other dynamic boundaries.

func (*StreamAction[Req, T]) GetAnyHooks added in v0.19.0 

func (a *StreamAction[Req, T]) GetAnyHooks() []AnyHook

GetAnyHooks returns a snapshot of both typed and erased hooks, with typed hooks adapted through Adapt so the caller sees a uniform slice.

func (*StreamAction[Req, T]) GetBindings added in v0.18.0 

func (a *StreamAction[Req, T]) GetBindings() []Binding

func (*StreamAction[Req, T]) HookCancelEvent added in v0.19.0 

func (a *StreamAction[Req, T]) HookCancelEvent(fn func()) *StreamAction[Req, T]

HookCancelEvent runs fn when the stream context is canceled.

func (*StreamAction[Req, T]) HookErrorEvent added in v0.19.0 

func (a *StreamAction[Req, T]) HookErrorEvent(fn func(error)) *StreamAction[Req, T]

HookErrorEvent runs fn when the stream terminates with an error.

func (*StreamAction[Req, T]) HookSuccessEvent added in v0.19.0 

func (a *StreamAction[Req, T]) HookSuccessEvent(fn func()) *StreamAction[Req, T]

HookSuccessEvent runs fn when the stream completes successfully.

func (*StreamAction[Req, T]) ReqPayload added in v0.18.0 

func (a *StreamAction[Req, T]) ReqPayload() any

func (*StreamAction[Req, T]) ResPayload added in v0.18.0 

func (a *StreamAction[Req, T]) ResPayload() any

func (*StreamAction[Req, T]) Route added in v0.18.0 

func (a *StreamAction[Req, T]) Route(bindings ...Binding) *StreamAction[Req, T]

func (*StreamAction[Req, T]) Use 

func (a *StreamAction[Req, T]) Use(h ...Hook[Req, iter.Seq2[T, error]]) *StreamAction[Req, T]

Use attaches typed hooks. The typed hook signature differs from AnyAction: the "response" side is the iterator itself, not a single value. Lifecycle events fire when the stream is exhausted, errored, or the consumer stops early via yield(false).

type StreamHandler 

type StreamHandler[Req, T any] func(context.Context, Req) (iter.Seq2[T, error], error)

type StreamOp added in v0.18.0 

type StreamOp[In, Out any] func(iter.Seq2[In, error]) iter.Seq2[Out, error]

StreamOp is a generic, typed stream operator.

Takes a stream of elements In, returns a stream of elements Out. Contract:

  • lazy: does not consume input until output is iterated
  • backpressure: yield(false) upstream stops the chain
  • an item error propagates without breaking the stream (unless the operator decides otherwise)

StreamOp is not AnyAction nor AnyStreamAction — it is the "middle" in the pipeline. Concrete implementations (Filter, Collect, Map) live in stream_ops.go.

func StreamCollect added in v0.18.0 

func StreamCollect[T any]() StreamOp[T, []T]

StreamCollect buffers the entire upstream into a slice and yields it as a single item.

UNBOUNDED. The caller is responsible for ensuring the input is bounded. Use StreamCollectN when the input size is not guaranteed.

The buffer is grown dynamically. On upstream error, the error is yielded and no partial slice is produced.

func StreamCollectN added in v0.18.0 

func StreamCollectN[T any](maxItems int) StreamOp[T, []T]

StreamCollectN is the bounded variant of StreamCollect.

maxItems must be > 0. The function panics if maxItems <= 0 — this is a programmer error, not a runtime condition. Configuration validation in the Flow layer is expected to reject invalid limits before construction.

If the upstream yields more than maxItems items, a xerr.Forbidden error is emitted and the stream terminates. Partial results are never returned. If the upstream yields exactly maxItems items, they are returned without error.

func StreamFilter added in v0.18.0 

func StreamFilter[T any](pred func(T) bool) StreamOp[T, T]

StreamFilter filters items through a pure predicate.

Contract:

  • Upstream errors propagate unchanged; the predicate does not see them.
  • Returning false from the downstream yield stops the upstream.
  • Zero allocations per item (predicate is typed, no boxing).

The predicate must be pure: no I/O, no mutation of shared state. For side-effecting transforms that may fail, use a StreamMapAction (planned for F1) instead.

type StreamOperator added in v0.19.0 

type StreamOperator interface {
	Name() string
	Apply(up AnyStream) (AnyStream, error)
}

StreamOperator is the runtime contract of a stream operator.

func ComposeOperators added in v0.19.0 

func ComposeOperators(ops ...StreamOperator) (StreamOperator, error)

ComposeOperators chains multiple operators into one.

func NewTypedStreamOperator added in v0.19.0 

func NewTypedStreamOperator[In, Out any](name string, op StreamOp[In, Out]) StreamOperator

NewTypedStreamOperator wraps a typed StreamOp as an erased StreamOperator.

type SystemStats 

type SystemStats interface {
	CPUPercent() float64
	Goroutines() int
}

SystemStats is implemented lock-free by nexss/monitor.

type Testable 

type Testable[Req, Res any] struct {
	// contains filtered or unexported fields
}

Testable wraps a BuiltAction and provides helpers for unit testing.

func TestFrom 

func TestFrom[Req, Res any](b *Builder[Req, Res]) *Testable[Req, Res]

func TestFromAction 

func TestFromAction[Req, Res any](act *BuiltAction[Req, Res]) *Testable[Req, Res]

FromAction returns a Testable wrapper for a BuiltAction.

func (*Testable[Req, Res]) CaptureReq 

func (t *Testable[Req, Res]) CaptureReq(ctx context.Context, input Req) (captured Req, res Res, err error)

CaptureReq executes the action and captures the request as seen by the base handler. The captured value is the one after all middleware have run, right before calling exec.

func (*Testable[Req, Res]) Do 

func (t *Testable[Req, Res]) Do(ctx context.Context, req Req) (Res, error)

Do executes the full action (with all middleware/hooks).

func (*Testable[Req, Res]) DoRaw 

func (t *Testable[Req, Res]) DoRaw(ctx context.Context, req Req) (Res, error)

DoRaw executes only the base handler — no validation, no cache, no middleware.

func (*Testable[Req, Res]) ExpectErr 

func (t *Testable[Req, Res]) ExpectErr(ctx context.Context, req Req, expected string) error

ExpectErr checks if action returns expected error kind.

type TreeHookOptions added in v0.6.0 

type TreeHookOptions struct {
	MaxDepth int
}

type TxRunner 

type TxRunner interface {
	RunInTx(ctx context.Context, fn func(txCtx context.Context) error) error
}

type TypedPayload 

type TypedPayload interface {
	ReqPayload() any
	ResPayload() any
}

TypedPayload allows plugins (like OpenAPI) to discover the underlying Request and Response types at boot time without storing them in metadata or using reflection during execution.

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.