Flows
Fan out one parent Job into typed child Jobs and collect their results.
Use a Flow when one durable parent Job owns several child Jobs: for example, an order that must reserve each line and charge the order before fulfillment can finish. The parent remains visible as the order-level operation, while the Flow store records child progress and lets a Worker collect a typed result. That gives operators one lifecycle to inspect, retry, or cancel instead of a set of unrelated jobs whose relationship exists only in logs.
A complete parent-and-child flow
This example creates one inventory child per order line plus a payment child, runs them through one Layer-owned Worker, and completes the parent with a useful fulfillment summary. The Zod 4 provider schemas make each persisted payload boundary explicit.
import * as z from 'zod'
import { Effect, Layer, Runtime } from 'better-effect'
import { ClockLive } from 'better-effect/standard-services'
import { Result } from 'better-result'
import {
Codec,
Flow,
FlowStore,
JobEncodeFailure,
JobStore,
MemoryFlowStore,
MemoryJobStore,
Queue,
Worker
} from 'better-effect-mq'
import { Schema as CoreSchema } from 'better-effect-schema'
import { Schema } from 'better-effect-schema/zod'
const OrderItem = z.object({ sku: z.string().min(1), quantity: z.int().positive() })
const OrderFailure = z.object({ code: z.string().min(1), message: z.string().min(1) })
class FulfillOrderPayload extends Schema.Class<FulfillOrderPayload>('app/FulfillOrderPayload')({
orderId: z.string().min(1),
currency: z.string().length(3),
totalCents: z.int().positive(),
items: z.array(OrderItem).min(1)
}) {}
const FulfillOrderPayloadCodec = Codec.standardSchema({
schema: FulfillOrderPayload,
encode: (value) =>
CoreSchema.encode(FulfillOrderPayload, value).mapError(
(error) => new JobEncodeFailure({ message: error.message, code: 'schema-encode' })
)
})
class ReserveInventoryPayload extends Schema.Class<ReserveInventoryPayload>(
'app/ReserveInventoryPayload'
)({
orderId: z.string().min(1),
sku: z.string().min(1),
quantity: z.int().positive()
}) {}
const ReserveInventoryPayloadCodec = Codec.standardSchema({
schema: ReserveInventoryPayload,
encode: (value) =>
CoreSchema.encode(ReserveInventoryPayload, value).mapError(
(error) => new JobEncodeFailure({ message: error.message, code: 'schema-encode' })
)
})
const ReserveInventoryResult = z.object({
kind: z.literal('inventory'),
sku: z.string().min(1),
reservedQuantity: z.int().nonnegative()
})
class ChargeOrderPayload extends Schema.Class<ChargeOrderPayload>('app/ChargeOrderPayload')({
orderId: z.string().min(1),
currency: z.string().length(3),
amountCents: z.int().positive()
}) {}
const ChargeOrderPayloadCodec = Codec.standardSchema({
schema: ChargeOrderPayload,
encode: (value) =>
CoreSchema.encode(ChargeOrderPayload, value).mapError(
(error) => new JobEncodeFailure({ message: error.message, code: 'schema-encode' })
)
})
const ChargeOrderResult = z.object({
kind: z.literal('payment'),
chargeId: z.string().min(1),
amountCents: z.int().positive()
})
const FulfillOrderResult = z.object({
orderId: z.string().min(1),
requestedItems: z.int().nonnegative(),
reservedItems: z.int().nonnegative(),
payment: z.enum(['charged', 'not-charged']),
failedChildren: z.array(z.string())
})
const Orders = Queue.define('orders')
const FulfillOrder = Orders.job('fulfill-order', {
version: 1,
payload: FulfillOrderPayloadCodec,
result: Codec.standardSchema({ schema: FulfillOrderResult }),
failure: Codec.standardSchema({ schema: OrderFailure })
})
const ReserveInventory = Orders.job('reserve-inventory', {
version: 1,
payload: ReserveInventoryPayloadCodec,
result: Codec.standardSchema({ schema: ReserveInventoryResult }),
failure: Codec.standardSchema({ schema: OrderFailure })
})
const ChargeOrder = Orders.job('charge-order', {
version: 1,
payload: ChargeOrderPayloadCodec,
result: Codec.standardSchema({ schema: ChargeOrderResult }),
failure: Codec.standardSchema({ schema: OrderFailure })
})
const Fulfillment = Flow.define('order-fulfillment', {
parent: FulfillOrder,
children: [ReserveInventory, ChargeOrder] as const,
onChildFailure: 'continue'
})
const FulfillmentRoute = Flow.handle(Fulfillment, {
fanOut: (payload) =>
Effect.fn(async function* () {
return Result.ok([
Flow.children(
ReserveInventory,
payload.items.map((item) => ({
key: `reserve:${item.sku}`,
payload: { orderId: payload.orderId, sku: item.sku, quantity: item.quantity }
}))
),
Flow.children(ChargeOrder, [
{
key: 'charge',
payload: {
orderId: payload.orderId,
currency: payload.currency,
amountCents: payload.totalCents
}
}
])
] as const)
}),
collect: (payload, results) =>
Effect.fn(async function* () {
const children = yield* Result.await(results.all()())
const reservedItems = children.filter(
(child) => child.outcome === 'completed' && child.result?.kind === 'inventory'
).length
const charged = children.some(
(child) => child.outcome === 'completed' && child.result?.kind === 'payment'
)
const payment = charged ? ('charged' as const) : ('not-charged' as const)
return Result.ok({
orderId: payload.orderId,
requestedItems: payload.items.length,
reservedItems,
payment,
failedChildren: children
.filter((child) => child.outcome !== 'completed')
.map((child) => child.childKey)
})
})
})
const handlers = [
Worker.handle(ReserveInventory, (payload) =>
Effect.fn(async function* () {
// Reserve payload.sku in inventory; key the write by order ID and SKU for retries.
return Result.ok({
kind: 'inventory' as const,
sku: payload.sku,
reservedQuantity: payload.quantity
})
})
),
Worker.handle(ChargeOrder, (payload) =>
Effect.fn(async function* () {
// Pass payload.orderId as the payment provider's idempotency key.
return Result.ok({
kind: 'payment' as const,
chargeId: `charge:${payload.orderId}`,
amountCents: payload.amountCents
})
})
)
] as const
const FulfillmentWorker = Worker.service('@app/FulfillmentWorker')
const FulfillmentWorkerLive = FulfillmentWorker.layer(() => ({
handlers,
flows: [FulfillmentRoute] as const,
concurrency: 4,
pollIntervalMs: 250,
flowSweepIntervalMs: 1_000,
flowBatchSize: 50
}))
const StorageLive = Layer.merge(
Layer.succeed(JobStore, JobStore.of(MemoryJobStore.make())),
Layer.succeed(FlowStore, FlowStore.of(MemoryFlowStore.make()))
)
const AppLive = Layer.complete(
Layer.merge(StorageLive, Layer.merge(ClockLive, FulfillmentWorkerLive))
)
const runtime = await Runtime.make(AppLive)
try {
await runtime.warmup()
const started = await runtime.run(() =>
Effect.gen(async function* () {
const parentId = yield* FulfillOrder.enqueue({
orderId: 'order-123',
currency: 'USD',
totalCents: 12_500,
items: [
{ sku: 'coffee-beans', quantity: 2 },
{ sku: 'pour-over-kit', quantity: 1 }
]
})
return Result.ok(parentId)
})
)
if (Result.isError(started)) throw started.error
const completed = await runtime.run(() => FulfillOrder.awaitResult(started.value))
if (Result.isError(completed)) throw completed.error
console.log(completed.value)
} finally {
await runtime.dispose()
}The parent enqueue is the only producer operation. The Worker claims it, runs fanOut once, and
persists the child requests with stable keys. Child Workers settle each reservation and payment
independently; the Flow records those outcomes and invokes collect when all children are terminal.
With onChildFailure: 'continue', the parent can finish with a partial summary and list failed
children. Use 'fail' when any failed child should fail the parent and cancel the remaining work.
The Flow descriptors use Zod 4 schema-backed classes from
better-effect-schema through Codec.standardSchema, the same provider-backed
boundary recommended for ordinary Jobs. The trusted JSON escape hatch
remains useful for a trusted, already-JSON-safe internal value, but it is
intentionally not the default for persisted boundary data.
When to use a Flow
Use a Flow for durable fan-out/fan-in, bounded child progress, and parent cancellation or retry. Use
ordinary independent Jobs when work does not share a parent lifecycle. A Flow is at-least-once: child
handlers and the parent side effect should be safe to replay, and the adapter must provide the
JobStore and FlowStore Layers with the same persistence and recovery guarantees.
The in-memory Layers in this example are for a local example. Replace them with one adapter's durable Layers; the Queue, Job descriptors, handlers, Worker service, Layer composition, and Runtime boundary stay the same.