// @ozaco/server

Policies

Composable resilience policies — retry, cache, circuit-breaker, coalescing, bulkhead, timeout, fallback and metrics — layered around every dispatch.

import from@ozaco/server/policy/retry

A policy is a std plugin, built with definePolicy, that wraps every action dispatch to add resilience or observability. You install a policy once with install(SomePolicy, options) to set global defaults; from then on it participates in the dispatch of every action.

Installed policies are applied as an onion sorted ascending by priority — the LOWEST number is the OUTERMOST layer. The built-in defaults live in PolicyPriority: Cache (0) is outermost, then Fallback (5), Bucket (10), Bulk (20), Retry (30), CircuitBreaker (40), Metrics (50), and Timeout (60) innermost. Timeout sits innermost on purpose: it must surface as a normal thrown CoreErrors.Timeout that the outer stateful policies observe through their catch blocks — if it sat outside them it would halt their generators, and a halt runs neither catch nor finally.

Every policy also exposes .actions.config(options?) and .actions.disable(). Both return a lazy PolicyDef.Setting you attach to a single action's settings array (via defineAction) to override options or turn the policy off for just that action. bypass(PolicyA, PolicyB, …) from @ozaco/server/core disables several policies in one entry, and describePolicyChain() returns the resolved onion order at runtime.

Most policies skip streaming dispatches by default (each exposes an opt-in flag such as retryStreams / timeoutStreams), since buffering, racing or replaying a live byte stream is rarely what you want.

// additional entry points

@ozaco/server/policy/retryRe-runs a failed dispatch with configurable attempts, fixed or exponential-backoff delay, and an optional `when` predicate to retry only certain failures.
@ozaco/server/policy/cacheMemoises non-streaming dispatch results in an in-memory TTL + max-size cache, keyed per caller identity (`vary`) with an optional `shouldCache` predicate.
@ozaco/server/policy/circuit-breakerTrips a per `service.action` circuit open after a failure threshold, short-circuiting with `CoreErrors.CircuitOpen`, then probes recovery in a half-open state.
@ozaco/server/policy/bucketSingle-flight request coalescing: identical concurrent calls within a batch window join one in-flight dispatch and share its outcome, collapsing duplicate load.
@ozaco/server/policy/bulkBulkhead concurrency limiter: caps in-flight dispatches at `maxConcurrent`, parks the overflow in a bounded queue, and fails fast with `CoreErrors.BulkQueueFull` / `BulkQueueTimeout`.
@ozaco/server/policy/timeoutBounds a dispatch by racing it against a timer, cancelling the loser and failing with `CoreErrors.Timeout`. Innermost by default.
@ozaco/server/policy/fallbackOn failure, returns a static `value` or runs a `handler(failure, dispatch)` instead of propagating, gated by an optional `when` predicate.
@ozaco/server/policy/metricsFires `onCall` / `onSuccess` / `onFailure` observability hooks around each dispatch (with duration), isolated so a throwing exporter never mislabels the outcome.

// on this page

Retry Cache Circuit breaker Bucket (coalescing)Bulkhead Timeout Fallback Metrics

Retry

Re-run a failed dispatch with backoff.

const

RetryPolicy

Retry a failed dispatch with configurable attempts, delay, backoff and `when` filter.

signature

// from '@ozaco/server/policy/retry'
const RetryPolicy: Policy<Retry.Options>

install(RetryPolicy, options?: Retry.Options) // set global defaults

RetryPolicy.actions.config(
  options?: Partial<Retry.Options>,
): Future<PolicyDef.Setting<Retry.Options>>
RetryPolicy.actions.disable(): Future<PolicyDef.Setting<Retry.Options>>

Wraps the inner dispatch with std:effect's retry. Defaults: attempts 3, delay 0 ms, backoff 1 (multiplier), maxDelay 30_000 ms. Supply when to retry only matching failures, and retryStreams: true to also retry streaming dispatches (skipped by default). Default onion priority is 30.

example

import { install } from '@ozaco/std/plugin'
import { RetryPolicy } from '@ozaco/server/policy/retry'
import { defineAction } from '@ozaco/server/core'

yield* install(RetryPolicy, { attempts: 5, delay: 100, backoff: 2 })

// override for one action only:
const flaky = defineAction(
  { settings: [RetryPolicy.actions.config({ attempts: 2 })] },
  function* () {
    /* ... */
  },
)

Retry

Install options and resolved context for the retry policy.

signature

namespace Retry {
  interface Options extends PolicyDef.Options {
    attempts?: number // default 3
    delay?: number // ms, default 0
    backoff?: number // multiplier, default 1
    maxDelay?: number // ms, default 30_000
    when?: (failure: Result.Failure<unknown>) => boolean
    retryStreams?: boolean // default false
  }
  interface Context extends PolicyDef.Context {
    attempts: number
    delay: number
    backoff: number
    maxDelay: number
    when?: (failure: Result.Failure<unknown>) => boolean
    retryStreams: boolean
  }
}

const

RetryPolicyKey

The per-action setting key used to resolve retry overrides (`'retry'`).

signature

const RetryPolicyKey: 'retry'

Cache

Memoise dispatch results in memory.

const

CachePolicy

Cache non-streaming dispatch results with a TTL, max size and per-identity keying.

signature

// from '@ozaco/server/policy/cache'
const CachePolicy: Policy<Cache.Options>

install(CachePolicy, options?: Cache.Options)

CachePolicy.actions.config(
  options?: Partial<Cache.Options>,
): Future<PolicyDef.Setting<Cache.Options>>
CachePolicy.actions.disable(): Future<PolicyDef.Setting<Cache.Options>>

An in-memory LRU-by-age cache. Defaults: ttl 30_000 ms, max 1000 entries, vary 'principal'. Keys combine the dispatch identity with the caller principal so different callers never share an entry; set vary: 'none' to share one entry across all callers. shouldCache(dispatch) opts individual dispatches out. Expired entries are evicted by timer and the oldest is evicted when max is reached; streaming dispatches bypass the cache entirely. Outermost by default (priority 0).

example

import { install } from '@ozaco/std/plugin'
import { CachePolicy } from '@ozaco/server/policy/cache'
import { bypass, defineAction } from '@ozaco/server/core'

yield* install(CachePolicy, { ttl: 60_000, max: 500 })

// a "fresh read" action that skips the cache (and coalescing):
const fresh = defineAction({ settings: bypass(CachePolicy) }, function* () {
  /* ... */
})

Cache

Options, cache entry shape and resolved context for the cache policy.

signature

namespace Cache {
  interface Options extends PolicyDef.Options {
    ttl?: number // ms, default 30_000
    max?: number // default 1000
    vary?: 'principal' | 'none' // default 'principal'
    shouldCache?: (dispatch: PolicyDef.DispatchContext) => boolean
  }
  interface Entry {
    value: unknown
    expiresAt: number
    timer: ReturnType<typeof setTimeout>
  }
  interface Context extends PolicyDef.Context {
    ttl: number
    max: number
    vary: 'principal' | 'none'
    shouldCache?: (dispatch: PolicyDef.DispatchContext) => boolean
    entries: Map<string, Entry>
  }
}

const

CachePolicyKey

The per-action setting key used to resolve cache overrides (`'cache'`).

signature

const CachePolicyKey: 'cache'

Circuit breaker

Trip open on repeated failures, then probe recovery.

const

CircuitBreakerPolicy

Per `service.action` circuit breaker with closed / open / half-open states.

signature

// from '@ozaco/server/policy/circuit-breaker'
const CircuitBreakerPolicy: Policy<CircuitBreaker.Options>

install(CircuitBreakerPolicy, options?: CircuitBreaker.Options)

CircuitBreakerPolicy.actions.config(
  options?: Partial<CircuitBreaker.Options>,
): Future<PolicyDef.Setting<CircuitBreaker.Options>>
CircuitBreakerPolicy.actions.disable(): Future<PolicyDef.Setting<CircuitBreaker.Options>>

Tracks failures per service\0action key. After threshold failures (default 5) the circuit opens and every call short-circuits with CoreErrors.CircuitOpen ('server:core.circuit-open', HTTP 503) until resetTimeout (default 30_000 ms) elapses; it then enters half-open and admits up to halfOpenMax (default 1) probe calls — a success closes it, a failure re-opens it. isFailure(failure) narrows which failures count. Default priority 40.

example

import { install } from '@ozaco/std/plugin'
import { CircuitBreakerPolicy } from '@ozaco/server/policy/circuit-breaker'

yield* install(CircuitBreakerPolicy, {
  threshold: 10,
  resetTimeout: 15_000,
})

CircuitBreaker

State, options, per-key entry and resolved context for the circuit breaker.

signature

namespace CircuitBreaker {
  type State = 'closed' | 'open' | 'half-open'
  interface Options extends PolicyDef.Options {
    threshold?: number // default 5
    resetTimeout?: number // ms, default 30_000
    halfOpenMax?: number // default 1
    isFailure?: (failure: Result.Failure<unknown>) => boolean
  }
  interface Entry {
    state: State
    failures: number
    halfOpenInflight: number
    openedAt: number
  }
  interface Context extends PolicyDef.Context {
    threshold: number
    resetTimeout: number
    halfOpenMax: number
    isFailure?: (failure: Result.Failure<unknown>) => boolean
    entries: Map<string, Entry>
  }
}

const

CircuitBreakerPolicyKey

The per-action setting key for circuit-breaker overrides (`'circuitBreaker'`).

signature

const CircuitBreakerPolicyKey: 'circuitBreaker'

Bucket (coalescing)

Coalesce identical concurrent dispatches into one.

const

BucketPolicy

Single-flight request coalescing over a short batch window.

signature

// from '@ozaco/server/policy/bucket'
const BucketPolicy: Policy<Bucket.Options>

install(BucketPolicy, options?: Bucket.Options)

BucketPolicy.actions.config(
  options?: Partial<Bucket.Options>,
): Future<PolicyDef.Setting<Bucket.Options>>
BucketPolicy.actions.disable(): Future<PolicyDef.Setting<Bucket.Options>>

Collapses duplicate concurrent work: the first caller of an identity key runs the dispatch inline, and up to max (default 100) further identical callers arriving during the batch join that in-flight dispatch and share its resolved outcome (each re-throwing a shared failure inside its own coroutine so its own outer policies still see it). After the dispatch settles the batch entry is cleared interval ms later (default 20). Like the cache, vary ('principal' default, or 'none') controls whether callers of different identity coalesce. Streaming dispatches are never coalesced. Default priority 10.

example

import { install } from '@ozaco/std/plugin'
import { BucketPolicy } from '@ozaco/server/policy/bucket'

// coalesce identical reads that land within 50ms of each other
yield* install(BucketPolicy, { interval: 50, max: 200 })

Bucket

Options, shared outcome, batch entry and resolved context for coalescing.

signature

namespace Bucket {
  interface Options extends PolicyDef.Options {
    interval?: number // ms, default 20
    max?: number // default 100
    vary?: 'principal' | 'none' // default 'principal'
  }
  type Outcome =
    | { ok: true; value: unknown }
    | { ok: false; failure: Result.Failure<unknown> }
  interface Entry {
    count: number
    resolvers: Helpers.WithResolvers<Outcome>
    timer?: ReturnType<typeof setTimeout>
  }
  interface Context extends PolicyDef.Context {
    interval: number
    max: number
    vary: 'principal' | 'none'
    entries: Map<string, Entry>
  }
}

const

BucketPolicyKey

The per-action setting key for bucket overrides (`'bucket'`).

signature

const BucketPolicyKey: 'bucket'

Bulkhead

Limit concurrency with a bounded queue.

const

BulkPolicy

Cap concurrent in-flight dispatches, queueing or rejecting the overflow.

signature

// from '@ozaco/server/policy/bulk'
const BulkPolicy: Policy<Bulk.Options>

install(BulkPolicy, options?: Bulk.Options)

BulkPolicy.actions.config(
  options?: Partial<Bulk.Options>,
): Future<PolicyDef.Setting<Bulk.Options>>
BulkPolicy.actions.disable(): Future<PolicyDef.Setting<Bulk.Options>>

A bulkhead / semaphore. Up to maxConcurrent (default 20) dispatches run at once; additional callers park in a queue of at most maxQueue (default 0 — no queue, so the overflow fails immediately with CoreErrors.BulkQueueFull, HTTP 503). When queueTimeout (default 0 = wait forever) is positive, a caller that waits longer fails with CoreErrors.BulkQueueTimeout (HTTP 504). A halted parked caller removes itself from the queue so a freed slot is never handed to a dead waiter. Default priority 20.

example

import { install } from '@ozaco/std/plugin'
import { BulkPolicy } from '@ozaco/server/policy/bulk'

yield* install(BulkPolicy, {
  maxConcurrent: 50,
  maxQueue: 100,
  queueTimeout: 2_000,
})

Bulk

Options, parked-waiter shape and resolved context for the bulkhead.

signature

namespace Bulk {
  interface Options extends PolicyDef.Options {
    maxConcurrent?: number // default 20
    maxQueue?: number // default 0 (no queue)
    queueTimeout?: number // ms, default 0 (wait forever)
  }
  interface Waiter {
    resolvers: Helpers.WithResolvers<void>
    timer?: ReturnType<typeof setTimeout>
    cancelled?: boolean
  }
  interface Context extends PolicyDef.Context {
    maxConcurrent: number
    maxQueue: number
    queueTimeout: number
    inflight: number
    queue: Waiter[]
  }
}

const

BulkPolicyKey

The per-action setting key for bulkhead overrides (`'bulk'`).

signature

const BulkPolicyKey: 'bulk'

Timeout

Bound how long a dispatch may run.

const

TimeoutPolicy

Race a dispatch against a timer, failing with `CoreErrors.Timeout` if it loses.

signature

// from '@ozaco/server/policy/timeout'
const TimeoutPolicy: Policy<Timeout.Options>

install(TimeoutPolicy, options?: Timeout.Options)

TimeoutPolicy.actions.config(
  options?: Partial<Timeout.Options>,
): Future<PolicyDef.Setting<Timeout.Options>>
TimeoutPolicy.actions.disable(): Future<PolicyDef.Setting<Timeout.Options>>

Races the dispatch against a sleep(timeoutMs) (default 30_000 ms); the loser is halted automatically, so on timeout the in-flight dispatch is cancelled and on success the timer is. A timeout fails with CoreErrors.Timeout ('server:core.timeout', HTTP 504). Installed innermost (priority 60) so this failure surfaces as an ordinary thrown value that outer stateful policies observe. Streaming dispatches are not bounded unless timeoutStreams: true.

example

import { install } from '@ozaco/std/plugin'
import { TimeoutPolicy } from '@ozaco/server/policy/timeout'

yield* install(TimeoutPolicy, { timeoutMs: 5_000 })

Timeout

Options and resolved context for the timeout policy.

signature

namespace Timeout {
  interface Options extends PolicyDef.Options {
    timeoutMs?: number // default 30_000
    timeoutStreams?: boolean // default false
  }
  interface Context extends PolicyDef.Context {
    timeoutMs: number
    timeoutStreams: boolean
  }
}

const

TimeoutPolicyKey

The per-action setting key for timeout overrides (`'timeout'`).

signature

const TimeoutPolicyKey: 'timeout'

Fallback

Substitute a value or handler on failure.

const

FallbackPolicy

Return a static value or run a handler instead of propagating a failure.

signature

// from '@ozaco/server/policy/fallback'
const FallbackPolicy: Policy<Fallback.Options>

install(FallbackPolicy, options?: Fallback.Options)

FallbackPolicy.actions.config(
  options?: Partial<Fallback.Options>,
): Future<PolicyDef.Setting<Fallback.Options>>
FallbackPolicy.actions.disable(): Future<PolicyDef.Setting<Fallback.Options>>

Catches any failure from the inner dispatch and, if when(failure) is absent or truthy, recovers: it runs handler(failure, dispatch) when provided, otherwise returns the configured value (presence-checked, so an explicit undefined still counts). If neither is configured the failure is re-thrown. Fallback is usually configured per action rather than globally, and streaming dispatches are passed through untouched. Default priority 5.

example

import { install } from '@ozaco/std/plugin'
import { FallbackPolicy } from '@ozaco/server/policy/fallback'
import { defineAction } from '@ozaco/server/core'

yield* install(FallbackPolicy)

const rates = defineAction(
  { settings: [FallbackPolicy.actions.config({ value: { usd: 1 } })] },
  function* () {
    /* ...may fail... */
  },
)

Fallback

Handler type, options and resolved context for the fallback policy.

signature

namespace Fallback {
  type Handler = (
    failure: Result.Failure<unknown>,
    ctx: PolicyDef.DispatchContext,
  ) => Operation<unknown>
  interface Options extends PolicyDef.Options {
    value?: unknown
    handler?: Handler
    when?: (failure: Result.Failure<unknown>) => boolean
  }
  interface Context extends PolicyDef.Context {
    value?: unknown
    handler?: Handler
    when?: (failure: Result.Failure<unknown>) => boolean
  }
}

const

FallbackPolicyKey

The per-action setting key for fallback overrides (`'fallback'`).

signature

const FallbackPolicyKey: 'fallback'

Metrics

Observe dispatch lifecycle events.

const

MetricsPolicy

Fire call / success / failure hooks (with duration) around each dispatch.

signature

// from '@ozaco/server/policy/metrics'
const MetricsPolicy: Policy<Metrics.Options>

install(MetricsPolicy, options?: Metrics.Options)

MetricsPolicy.actions.config(
  options?: Partial<Metrics.Options>,
): Future<PolicyDef.Setting<Metrics.Options>>
MetricsPolicy.actions.disable(): Future<PolicyDef.Setting<Metrics.Options>>

Invokes onCall before the dispatch and exactly one of onSuccess / onFailure after, each carrying the serviceName, actionKey, startedAt and a computed durationMs. The hooks run through an isolated runHook so a throwing exporter can neither re-enter the catch (mislabelling a success as a failure) nor break the dispatch. Default priority 50.

example

import { install } from '@ozaco/std/plugin'
import { MetricsPolicy } from '@ozaco/server/policy/metrics'

yield* install(MetricsPolicy, {
  onSuccess: e => console.log(e.serviceName, e.actionKey, e.durationMs),
  onFailure: e => console.error(e.actionKey, e.failure.message),
})

Metrics

Event shapes, options and resolved context for the metrics policy.

signature

namespace Metrics {
  interface Event {
    serviceName: string
    actionKey: string
    startedAt: number
    durationMs: number
  }
  interface SuccessEvent extends Event {
    value: unknown
  }
  interface FailureEvent extends Event {
    failure: Result.Failure<unknown>
  }
  interface Options extends PolicyDef.Options {
    onCall?: (event: Pick<Event, 'serviceName' | 'actionKey' | 'startedAt'>) => void
    onSuccess?: (event: SuccessEvent) => void
    onFailure?: (event: FailureEvent) => void
  }
  interface Context extends PolicyDef.Context {
    onCall?: (event: Pick<Event, 'serviceName' | 'actionKey' | 'startedAt'>) => void
    onSuccess?: (event: SuccessEvent) => void
    onFailure?: (event: FailureEvent) => void
  }
}

const

MetricsPolicyKey

The per-action setting key for metrics overrides (`'metrics'`).

signature

const MetricsPolicyKey: 'metrics'