#Pipelines and Resilience

Every primitive with a handler in Alepha is a pipeline: a handler with a list of middleware wrapped around it. $action, $job and $page all extend the same base, which is why they all take a use array, and $pipeline is that base exposed on its own for the functions that are not any of those.

1import { $pipeline } from "alepha";

That shared shape is what makes retries, timeouts, throttling and circuit breaking one topic instead of four. Each is a middleware, each works anywhere a use array does.

#$pipeline

Wrap a plain function and call it like a function:

 1import { $pipeline } from "alepha"; 2import { $retry } from "alepha/retry"; 3import { $timeout } from "alepha/datetime"; 4  5class OrderService { 6  processOrder = $pipeline({ 7    use: [$retry({ max: 3 }), $timeout([30, "seconds"])], 8    handler: async (orderId: string) => { 9      return { orderId, status: "paid" };10    },11  });12 13  async run(id: string) {14    return this.processOrder(id);15  }16}

Reach for it when the work is not a route, a job or a page: an internal service method called from several places that should carry its retry policy with it rather than have each caller remember one.

#Order in use is not cosmetic

The first middleware in the array is the outermost. use: [A, B] composes as A(B(handler)), so A sees the call first and the result last.

This changes what the pair above means:

Both are reasonable. Only one is what you meant.

#$scope

Host primitives run their handler inside an AsyncLocalStorage scope, which is what makes alepha.context.get() and .set() work per request. A standalone $pipeline has no such scope, so add one when the handler needs it:

 1import { $pipeline, $scope } from "alepha"; 2  3class OrderService { 4  processOrder = $pipeline({ 5    use: [$scope()], 6    handler: async (orderId: string) => { 7      return orderId; 8    }, 9  });10}

Adding $scope() to an $action, $job or $page throws, on purpose: you are already inside a scope, and nesting one would give you a second, empty context that silently loses everything the outer one held.

#The resilience middlewares

Six middlewares, all usable in any use array, all process-local unless the table says otherwise.

#$retry

 1import { $action } from "alepha/server"; 2import { $retry } from "alepha/retry"; 3  4class Payments { 5  charge = $action({ 6    use: [ 7      $retry({ 8        max: 3, 9        backoff: { initial: 500, factor: 2, jitter: true },10        maxDuration: [10, "seconds"],11        when: (error) => !error.message.includes("card_declined"),12      }),13    ],14    handler: async () => "charged",15  });16}

max counts attempts, not extra attempts: max: 3 runs the handler up to three times. backoff takes a fixed number of milliseconds or an exponential configuration, and defaults to { initial: 200, factor: 2, jitter: true }. maxDuration caps the total elapsed time across every attempt.

when is the important one. Retrying a declined card is not resilience, it is three declined cards: return false for errors that will never succeed.

Retries abort on application shutdown, so a stopping process does not sit in a backoff sleep it will never wake from.

#$timeout

1import { $pipeline } from "alepha";2import { $timeout } from "alepha/datetime";3 4class Orders {5  process = $pipeline({6    use: [$timeout([30, "seconds"])],7    handler: async (orderId: string) => orderId,8  });9}

The deadline rejects the promise. It does not cancel the underlying work, which keeps running until whatever it is waiting on gives up, so pair it with an AbortSignal where the dependency supports one.

It uses managed timeouts from DateTimeProvider, which means travel() moves it in tests instead of forcing a real 30 second wait.

#$throttle

1import { $action } from "alepha/server";2import { $throttle } from "alepha/datetime";3 4class Payments {5  charge = $action({6    use: [$throttle({ rate: 80, per: [1, "second"] })],7    handler: async () => "charged",8  });9}

A token bucket that delays excess calls rather than rejecting them, which is what separates it from $rateLimit: throttling shapes your outbound traffic, rate limiting refuses somebody else's inbound traffic.

Two limits worth knowing before you rely on it. It is process-local, so four instances at rate: 80 produce up to 320 calls per second at the API. And the refill is re-checked only when a waiter wakes, so a burst of concurrent calls can wake inside the same window and briefly exceed rate. Treat it as smoothing, not as a quota you can promise a vendor.

#$debounce

 1import { $action } from "alepha/server"; 2import { $debounce } from "alepha/datetime"; 3  4class Search { 5  query = $action({ 6    path: "/search", 7    use: [ 8      $debounce({ 9        delay: [200, "ms"],10        key: (req: { query: { q: string } }) => req.query.q,11      }),12    ],13    handler: async ({ query }) => query.q,14  });15}

Concurrent calls sharing a key are coalesced into one execution, and every caller receives that one result. The classic use is a cache expiring under load: a hundred requests arrive for the same key, and one rebuild serves all of them.

There is no storage behind it. Once the handler settles the next call starts fresh, and the key defaults to JSON.stringify(args), which is rarely what you want for a request object. Pass key.

#$circuit

1import { $action } from "alepha/server";2import { $circuit } from "alepha/server";3 4class Payments {5  charge = $action({6    use: [$circuit({ threshold: 5, reset: [30, "seconds"] })],7    handler: async () => "charged",8  });9}

Three states. Closed passes calls through and counts consecutive failures. At threshold it opens and rejects every call without touching the handler. After reset it goes half-open and lets one call through: success closes it, failure opens it again.

The point is not to protect you, it is to protect the thing you are calling. A dependency that is failing under load recovers faster when the traffic stops.

#$memoize

1import { $memoize } from "alepha";2import { $action } from "alepha/server";3 4class Stats {5  summary = $action({6    use: [$memoize({ max: 100 })],7    handler: async () => "42",8  });9}

A plain Map, FIFO eviction at max (default 1000), no TTL, no invalidation and no sharing between processes. It stores the promise immediately, so concurrent calls for one key deduplicate, and it deletes the entry when the handler throws, so failures are never cached.

Entries live until they are evicted. That is the whole design, and it is why this is for values that do not go stale in a way that matters. Anything needing a TTL, explicit invalidation or Redis wants $cache instead.

#$batch

$batch is the odd one out: a primitive rather than a middleware, and it changes the shape of the call rather than wrapping it. It exists for the case where one call per item is wasteful and one call per hundred items is not.

 1import { z } from "alepha"; 2import { $batch } from "alepha/batch"; 3  4class Indexer { 5  documents = $batch({ 6    schema: z.object({ id: z.uuid(), body: z.text() }), 7    maxSize: 100, 8    maxDuration: [2, "seconds"], 9    concurrency: 2,10    handler: async (items) => {11      return items.map((item) => ({ id: item.id, indexed: true }));12    },13  });14 15  async index(id: string, body: string) {16    const ticket = await this.documents.push({ id, body });17    return this.documents.wait(ticket);18  }19}

push() validates the item against schema, queues it and returns a ticket immediately. The handler runs when either maxSize items have accumulated or maxDuration has elapsed, whichever comes first. wait(ticket) resolves with that item's result once the batch it landed in has been processed.

partitionBy is what keeps a batch honest when items are not interchangeable: partition by tenant and one tenant's flush never carries another tenant's rows.

flush() forces a partition (or all of them) without waiting, status(ticket) reads an item's state without blocking, and clearCompleted() drops finished items from memory. Call that last one periodically in a long-running process: completed results are retained so that wait() can still answer, and nothing evicts them for you.

#See also

• Middlewares for the server-side middleware that ships with AlephaServer
• Caching for $cache, which is what $memoize is not
• Background Jobs for work that has to survive the process