The orders of the order served over HTTP by a runner of Effect Cluster, as a process of its
own on a file of SQLite. The boundary takes the request id from the header Idempotency-Key and the sender from
Authorization. The world the orders act on is a server of its own, which charges once per idempotency key, as a
payment provider does. The tests start the runners as processes and kill them: an order whose runner is killed
during its charge is paid once, by the runner started again on the same file. With two runners, the one that
survives finishes the orders of the one killed, and those of one stopped by SIGTERM, a test run off Windows.
Launched with DRAIN_MILLIS, a runner stopped by SIGTERM first sends the emails under way, each once, and the
other runner finds them sent, a test run off Windows too.
server.ts: the runner's layer and the boundary's routes.main.ts: the process, configured by its environment.server.test.ts: the processes started, killed and started again.The guide's page on Effect Cluster shows this server.
/**
* Runs the server of the orders as a process (decision 0178):
*
* DATABASE=orders.db WORLD_URL=http://localhost:4000 HOST=localhost PORT=3000 bun examples/server/main.ts
*
* `HOST` and `PORT` are the runner's address, which the other runners reach it at and its boundary listens on.
* Cluster's `ShardingConfig` reads them, and the rest of its settings, from the environment, such as
* `SHARD_LOCK_EXPIRATION`. `RUNNER_HEALTH_CHECK_INTERVAL`, a minute by default, is how long the steps of an older
* version die and wait once a newer one launches, and a deployment that rolls out often shortens it
* (decision 0159). The process stops on a signal, and the runtime stops with the layer's scope. `DRAIN_MILLIS`, if
* set, is how long that stop waits for the emails under way (decision 0119).
*/
import { BunClusterHttp, BunRuntime } from '@effect/platform-bun'
import { ShardingConfig } from 'effect/cluster'
import * as Config from 'effect/Config'
import * as Effect from 'effect/Effect'
import * as Layer from 'effect/Layer'
import * as Option from 'effect/Option'
import { server } from './server.ts'
const settings = Config.all({
filename: Config.String('DATABASE'),
world: Config.String('WORLD_URL'),
drain: Config.option(Config.Int('DRAIN_MILLIS')),
})
const main = Layer.unwrap(Effect.map(settings, ({ filename, world, drain }) =>
server(Option.match(drain, {
onNone: () => ({ filename, world }),
onSome: (millis) => ({ filename, world, drain: { millis } }),
})))).pipe(
Layer.provide(BunClusterHttp.layerHttpServer),
Layer.provide(ShardingConfig.layerFromEnv()),
)
Layer.launch(main).pipe(BunRuntime.runMain)
/**
* The orders of `examples/order` served over HTTP by a runner of Effect Cluster, on a file of SQLite
* (decision 0178). The boundary takes the request id from `Idempotency-Key` and the sender from `Authorization`,
* and calls the door of the orders through the deployment. The world the orders act on is a server of its own,
* which the runner reaches over HTTP.
*/
import { SqliteClient } from '@effect/sql-sqlite-bun'
import {
HttpRunner,
RunnerHealth,
Runners,
RunnerServer,
ShardingConfig,
SqlMessageStorage,
SqlRunnerStorage,
} from 'effect/cluster'
import * as Context from 'effect/Context'
import * as Data from 'effect/Data'
import { Digest } from 'effect/crypto'
import * as Effect from 'effect/Effect'
import {
FetchHttpClient,
HttpBody,
HttpClient,
HttpClientRequest,
HttpClientResponse,
HttpIncomingMessage,
HttpRouter,
HttpServerRequest,
HttpServerRespondable,
HttpServerResponse,
HttpStatus,
} from 'effect/http'
import * as Layer from 'effect/Layer'
import * as Option from 'effect/Option'
import * as Schedule from 'effect/Schedule'
import * as Schema from 'effect/Schema'
import { RpcSerialization, RpcServer } from 'effect/rpc'
import {
type Answer,
type Creation,
type Handle,
type Operated,
type TypesOf,
} from '@fgaudo/effect-tea-persistent'
import { launch, sharding } from '@fgaudo/effect-tea-persistent/cluster'
import { Mailer, order, Payments, type Sender, Warehouse } from '../order/order.ts'
/**
* Where the runner's file and the world are, and how long a planned stop drains the at-most-once operations, if it
* drains them (decision 0119). The runner's address, which its boundary listens on too, and the lease's expiration
* are Cluster's `ShardingConfig`, read from the environment.
*/
export type Settings = Readonly<{ filename: string; world: string; drain?: Readonly<{ millis: number }> }>
/**
* The world over HTTP, through a client of the world's address, which tells the world the runner that calls in
* `x-runner`. A transport's failure is a defect, which suspends the order: the world's `402` is the only failure an
* operation of the orders declares. A call that carries an idempotency key is retried for a while first, since the
* world acts on it once whatever the number of tries. The email carries none and runs at most once
* (decision 0011), so it is sent once: a retry after a lost answer would send the email twice.
*/
const world = (base: string) =>
Layer.effectContext(Effect.gen(function*() {
const { runnerAddress } = yield* ShardingConfig.ShardingConfig
const runner = Option.match(runnerAddress, { onNone: () => '', onSome: ({ host, port }) => `${host}:${port}` })
const once = (yield* HttpClient.HttpClient).pipe(
HttpClient.mapRequest(HttpClientRequest.prependUrl(base)),
HttpClient.mapRequest(HttpClientRequest.setHeader('x-runner', runner)),
)
const keyed = once.pipe(HttpClient.retryTransient({ schedule: Schedule.spaced('200 millis'), times: 50 }))
const post = (client: HttpClient.HttpClient, path: string, body: unknown) =>
client.post(path, { body: HttpBody.jsonUnsafe(body) })
const told = (client: HttpClient.HttpClient, path: string, body: unknown) =>
post(client, path, body).pipe(Effect.flatMap(HttpClientResponse.filterStatusOk), Effect.orDie, Effect.asVoid)
const Charged = Schema.Struct({ charge: Schema.String })
const payments = Payments.of({
charge: (request) =>
post(keyed, '/charge', request).pipe(
Effect.orDie,
Effect.flatMap(HttpClientResponse.matchStatus({
402: () => Effect.fail('Declined' as const),
orElse: (response) =>
HttpClientResponse.filterStatusOk(response).pipe(
Effect.flatMap(HttpIncomingMessage.schemaBodyJson(Charged)),
Effect.map(({ charge }) => charge),
Effect.orDie,
),
})),
),
refund: (request) => told(keyed, '/refund', request),
})
return Context.make(Payments, payments).pipe(
Context.add(Warehouse, Warehouse.of({ prepare: (request) => told(keyed, '/prepare', request) })),
Context.add(Mailer, Mailer.of({ send: (email) => told(once, '/mail', email) })),
)
})).pipe(Layer.provide(FetchHttpClient.layer))
/** The deployment of the orders, which the routes call. */
type OrderTypes = TypesOf<typeof order>
export class Orders extends Context.Service<Orders, Handle<OrderTypes>>()('examples/server/Orders') {}
/**
* The runner: the layers `HttpRunner.layerHttp` builds, with `sharding` in place of `Sharding.layer`, so that the
* versions are recorded before Cluster registers the runner (decision 0174). The runners send to each other over
* HTTP at `/runner`, with the serialization `BunClusterHttp.layer` chooses. A runner that dies stops its heartbeat,
* and `SqlRunnerStorage` stops reading it once its lease expires, so that its shards move with no check of health.
* The ping of `RunnerHealth.layerPing`, `BunClusterHttp.layer`'s default, also takes the shards of a runner that
* still beats but does not answer. `sharding` answers through it whether a runner runs the newest version, so the
* check of health is also what moves the shards of an older version once a newer one launches (decision 0159). The
* runtime is launched and the orders deployed in the layer's scope, so that
* they stop when the server stops.
*/
const orders = (settings: Settings) =>
Layer.effect(
Orders,
Effect.flatMap(
launch(settings.drain === undefined ? {} : { drain: settings.drain }),
(runtime) => runtime.deploy(order),
),
).pipe(
Layer.provide(world(settings.world)),
Layer.provideMerge(
sharding([order]).pipe(
Layer.provide(RunnerHealth.layerPing),
Layer.provideMerge(Runners.layerRpc),
Layer.provide(HttpRunner.layerClientProtocolHttp({ path: runnerPath }).pipe(Layer.provide(FetchHttpClient.layer))),
Layer.provideMerge(SqlMessageStorage.layer),
Layer.provide(SqlRunnerStorage.layer),
Layer.provideMerge(RpcSerialization.layerSchemaBinary()),
Layer.provideMerge(SqliteClient.layer({ filename: settings.filename })),
Layer.provide(Digest.layerWebCrypto),
Layer.orDie,
),
),
)
/** The path the runners send to each other at, beside the boundary's routes. */
const runnerPath = '/runner'
/**
* A request the boundary refuses, answered with its status. It is `Respondable`, so the router answers it as it
* answers a `SchemaError`, with no handler of its own.
*/
class Refused extends Data.TaggedError('Refused')<{ readonly status: number; readonly reason: string }>
implements HttpServerRespondable.Respondable
{
[HttpServerRespondable.symbol]() {
return HttpServerResponse.json({ _tag: 'Refused', reason: this.reason }, { status: this.status })
}
}
/**
* Who calls, from `Authorization: Bearer <who>`: `customer:<id>`, `warehouse` or `operator`. A real boundary checks
* a token here, and the example only reads the name the token carries (decision 0124).
*/
const Caller = Schema.Struct({
authorization: Schema.Union([
Schema.Literals(['Bearer warehouse', 'Bearer operator']),
Schema.TemplateLiteralParser(['Bearer customer:', Schema.NonEmptyString]),
]),
})
const caller = HttpServerRequest.schemaHeaders(Caller).pipe(
Effect.mapError(() => new Refused({ status: 401, reason: 'no caller' })),
Effect.map(({ authorization }): Sender | 'Operator' =>
authorization === 'Bearer warehouse'
? { _tag: 'Warehouse' }
: authorization === 'Bearer operator'
? 'Operator'
: { _tag: 'Customer', id: authorization[1] }
),
)
/** The request id of a create or a send, which a caller gives so that its retry is answered once. */
const RequestId = Schema.Struct({ 'idempotency-key': Schema.NonEmptyString })
/** A request of a caller: its sender and its request id. */
const request = Effect.gen(function*() {
const sender = yield* caller
if (sender === 'Operator') return yield* new Refused({ status: 403, reason: 'an operator sends nothing' })
const headers = yield* HttpServerRequest.schemaHeaders(RequestId).pipe(
Effect.mapError(() => new Refused({ status: 400, reason: 'no Idempotency-Key' })),
)
return { sender, requestId: headers['idempotency-key'] }
})
/**
* How long a reply to a send is kept: the day this server's clients retry a request for. A retry within it is
* answered from the stored reply, and the next step of the order or an operator's `expire` deletes the reply after
* it, so the replies do not add up for ever (decision 0125).
*/
const keep = { millis: 24 * 60 * 60 * 1_000 }
const operator = Effect.flatMap(
caller,
(who) => who === 'Operator' ? Effect.void : Effect.fail(new Refused({ status: 403, reason: 'no operator' })),
)
const id = Effect.map(HttpRouter.schemaPathParams(Schema.Struct({ id: Schema.NonEmptyString })), ({ id }) => id)
/** Every answer the routes give: of a create, of a send, and of an operator's resume or discard. */
type Answered = Creation<unknown> | Answer<unknown> | Operated
/**
* The status of each answer, one for each tag, so that a tag the library adds fails to compile here
* (decision 0190). A suspended order is answered `409` and not `503`, since an HTTP client retries a `503` as
* transient, and a suspended order waits for an operator. A full entity is answered `503`, since its caller retries
* later under the same request id. On Cluster a caller is never answered `Full` (decision 0165), and the entry
* serves a boundary on one process that copies the map. A request id reused for another body is answered `422`, as
* the header `Idempotency-Key` of HTTP answers it, since the body is the client's to fix (decision 0269).
*/
const status = {
Created: 'Created',
AlreadyExists: 'Conflict',
Accepted: 'OK',
Rejected: 'Conflict',
NotFound: 'NotFound',
Suspended: 'Conflict',
Full: 'ServiceUnavailable',
Reused: 'UnprocessableEntity',
Resumed: 'OK',
Discarded: 'OK',
NotDiscarded: 'Conflict',
StillSuspended: 'Conflict',
NotSuspended: 'Conflict',
} as const satisfies Readonly<Record<Answered['_tag'], HttpStatus.Literal>>
const answer = (body: Answered) =>
HttpServerResponse.json(body, {
status: HttpStatus.fromLiteral(status[body._tag]),
headers: body._tag === 'Full' ? { 'retry-after': '1' } : undefined,
})
/** The body of a create: the order's total. */
const NewOrder = Schema.Struct({ totalInCents: Schema.Int })
const routes = Layer.mergeAll(
HttpRouter.add('GET', '/health', HttpServerResponse.text('ok')),
HttpRouter.add(
'PUT',
'/orders/:id',
Effect.gen(function*() {
const { sender, requestId } = yield* request
if (sender._tag !== 'Customer') return yield* new Refused({ status: 403, reason: 'only a customer orders' })
const { totalInCents } = yield* HttpServerRequest.schemaBodyJson(NewOrder)
const orders = yield* Orders
const flags = { order: { customer: sender.id, totalInCents } }
return yield* answer(yield* orders.create(yield* id, { flags, sender, requestId }))
}),
),
HttpRouter.add(
'POST',
'/orders/:id/events',
Effect.gen(function*() {
const caller = yield* request
const message = yield* HttpServerRequest.schemaBodyJson(order.schemas.inbound)
const orders = yield* Orders
return yield* answer(yield* orders.send(yield* id, message, { ...caller, keep }))
}),
),
HttpRouter.add(
'GET',
'/orders/:id',
Effect.gen(function*() {
const orders = yield* Orders
const read = yield* orders.read(yield* id)
return yield* read === null ? answer({ _tag: 'NotFound' }) : HttpServerResponse.json(read)
}),
),
HttpRouter.add(
'POST',
'/orders/:id/resume',
Effect.gen(function*() {
yield* operator
const orders = yield* Orders
return yield* answer(yield* orders.resume(yield* id))
}),
),
HttpRouter.add(
'POST',
'/orders/:id/discard',
Effect.gen(function*() {
yield* operator
const orders = yield* Orders
return yield* answer(yield* orders.discard(yield* id))
}),
),
)
/**
* The server: the boundary's routes and the runner's, given the orders and the runner from outside `serve`, so that
* both share one deployment and one `Sharding`.
*/
export const server = (settings: Settings) =>
HttpRouter.serve(
Layer.mergeAll(routes, RunnerServer.layer.pipe(Layer.provide(RpcServer.layerProtocolHttp({ path: runnerPath })))),
{ disableListenLog: true, disableLogger: true },
).pipe(Layer.provide(orders(settings)))
/**
* The server of the orders as a process (decision 0178). The world is a server of the test, which charges once per
* idempotency key, as a payment provider does. It holds the first charge of each key open until the test releases
* it, and records the runners that asked for each one. It records each email too, and holds it open if asked.
*/
import { BunHttpServer, BunServices } from '@effect/platform-bun'
import { SqliteClient } from '@effect/sql-sqlite-bun'
import { expect, test } from 'bun:test'
import * as Context from 'effect/Context'
import * as Deferred from 'effect/Deferred'
import * as Effect from 'effect/Effect'
import * as Exit from 'effect/Exit'
import * as Fiber from 'effect/Fiber'
import * as FileSystem from 'effect/FileSystem'
import {
FetchHttpClient,
HttpBody,
HttpClient,
HttpClientRequest,
HttpClientResponse,
HttpRouter,
HttpServer,
HttpServerRequest,
HttpServerResponse,
} from 'effect/http'
import * as Layer from 'effect/Layer'
import * as Option from 'effect/Option'
import * as Path from 'effect/Path'
import { ChildProcess, ChildProcessSpawner } from 'effect/process'
import * as Ref from 'effect/Ref'
import * as Schedule from 'effect/Schedule'
import * as Schema from 'effect/Schema'
import * as Scope from 'effect/Scope'
import { SqlClient } from 'effect/sql'
/**
* A test that waits over two seconds on a clock that is real, for a runner's lease: `bun test` skips it, and
* `bun run test:release` and `bun run test:slow` run it, under `SCHEDULES=all`.
*/
const slow = Bun.env['SCHEDULES'] === 'all' ? test : test.skip
/** The port of a server built by `layer`, which listens until `scope` closes. */
const portOf = <E, R>(layer: Layer.Layer<HttpServer.HttpServer, E, R>, scope: Scope.Scope) =>
Effect.map(Layer.buildWithScope(layer, scope), (context) => {
const address = Context.get(context, HttpServer.HttpServer).address
if (address._tag === 'UnixPathAddress') throw new Error('the server listens on a path')
return address.port
})
/** A port no server listens on now: the port of a server that listened on any port, and stopped. */
const freePort = Effect.scoped(Effect.flatMap(Effect.scope, (scope) => portOf(BunHttpServer.layerServer({ port: 0 }), scope)))
/** What the world does besides charging once per key: fail the first calls, or hold each email open. */
type Behaviour = Readonly<{ failing?: boolean; holdingMails?: boolean }>
/**
* The world. `charges` holds the charge of each idempotency key, and `asked` the runners that asked for it, in
* order. A charge is taken as it arrives, so a charge whose answer the process never got is still taken. The first
* charge of each key waits for `release`. A world `failing` answers `503` to the first charge, before it takes or
* records it, and to the first email. `mailed` holds each email asked for, with the runner that asked, and `mails`
* counts them. A world `holdingMails` answers each email only once `releaseMails` runs.
*/
const world = ({ failing = false, holdingMails = false }: Behaviour) => Effect.gen(function*() {
const charges = yield* Ref.make(new Map<string, string>())
const asked = yield* Ref.make(new Map<string, ReadonlyArray<string>>())
const released = yield* Deferred.make<void>()
const failed = yield* Ref.make(new Set<string>())
const mailed = yield* Ref.make<ReadonlyArray<Readonly<{ runner: string; subject: string }>>>([])
const mailsReleased = yield* Deferred.make<void>()
/** Whether the world answers `503` to the call of `path`: only the first, and only of a world `failing`. */
const fails = (path: string) => Ref.modify(failed, (paths) => [failing && !paths.has(path), new Set(paths).add(path)])
const unavailable = HttpServerResponse.empty({ status: 503 })
const Charge = Schema.Struct({ idempotencyKey: Schema.String })
const Mail = Schema.Struct({ subject: Schema.String })
const Runner = Schema.Struct({ 'x-runner': Schema.String })
const routes = Layer.mergeAll(
HttpRouter.add(
'POST',
'/charge',
Effect.gen(function*() {
if (yield* fails('/charge')) return unavailable
const { idempotencyKey } = yield* HttpServerRequest.schemaBodyJson(Charge)
const { 'x-runner': runner } = yield* HttpServerRequest.schemaHeaders(Runner)
const before = yield* Ref.getAndUpdate(asked, (all) =>
new Map(all).set(idempotencyKey, [...(all.get(idempotencyKey) ?? []), runner]))
const charge = yield* Ref.modify(charges, (taken) => {
const charge = taken.get(idempotencyKey) ?? `ch_${taken.size + 1}`
return [charge, new Map(taken).set(idempotencyKey, charge)]
})
if (!before.has(idempotencyKey)) yield* Deferred.await(released)
return yield* HttpServerResponse.json({ charge })
}),
),
HttpRouter.add(
'POST',
'/mail',
Effect.gen(function*() {
const { subject } = yield* HttpServerRequest.schemaBodyJson(Mail)
const { 'x-runner': runner } = yield* HttpServerRequest.schemaHeaders(Runner)
yield* Ref.update(mailed, (all) => [...all, { runner, subject }])
if (yield* fails('/mail')) return unavailable
if (holdingMails) yield* Deferred.await(mailsReleased)
return HttpServerResponse.empty({ status: 204 })
}),
),
HttpRouter.add('POST', '/*', HttpServerResponse.empty({ status: 204 })),
)
const port = yield* portOf(
HttpRouter.serve(routes, { disableLogger: true, disableListenLog: true }).pipe(
Layer.provideMerge(BunHttpServer.layer({ port: 0 })),
),
yield* Effect.scope,
)
return {
url: `http://localhost:${port}`,
charges: Ref.get(charges),
asked: Ref.get(asked),
mailed: Ref.get(mailed),
mails: Effect.map(Ref.get(mailed), (all) => all.length),
/** Waits until the world holds the first charge of `count` keys. */
holding: (count: number) =>
Ref.get(asked).pipe(
Effect.filterOrFail((all) => all.size >= count),
Effect.retry({ schedule: Schedule.spaced('50 millis'), times: 600 }),
),
release: Deferred.succeed(released, undefined),
/** Waits until the world holds `count` emails. */
holdingMails: (count: number) =>
Ref.get(mailed).pipe(
Effect.filterOrFail((all) => all.length >= count),
Effect.retry({ schedule: Schedule.spaced('50 millis'), times: 600 }),
),
releaseMails: Deferred.succeed(mailsReleased, undefined),
}
})
type Settings = Readonly<{ filename: string; world: string }>
/**
* The process of a runner at `localhost:<port>` in the current scope, once its boundary answers. Closing the scope
* kills it, and the client returned calls its boundary. The lease expires after three seconds, so that a runner that
* dies frees its shards within the test, unless `env` sets Cluster's settings otherwise.
*/
const start = Effect.fn('start')(function*(settings: Settings, port: number, env: Record<string, string> = {}) {
const spawner = yield* ChildProcessSpawner.ChildProcessSpawner
const path = yield* Path.Path
const process = yield* spawner.spawn(ChildProcess.make('bun', [path.join(import.meta.dir, 'main.ts')], {
env: {
DATABASE: settings.filename,
WORLD_URL: settings.world,
HOST: 'localhost',
PORT: String(port),
SHARD_LOCK_EXPIRATION: '3 seconds',
...env,
},
extendEnv: true,
stdout: 'ignore',
stderr: 'inherit',
}))
const client = (yield* HttpClient.HttpClient).pipe(
HttpClient.mapRequest(HttpClientRequest.prependUrl(`http://localhost:${port}`)),
)
yield* client.get('/health').pipe(
Effect.flatMap(HttpClientResponse.filterStatusOk),
Effect.retry({ schedule: Schedule.spaced('100 millis'), times: 200 }),
)
return { process, client, address: `localhost:${port}` }
})
const as = (who: string, requestId: string) => ({
authorization: `Bearer ${who}`,
'idempotency-key': requestId,
})
const Tagged = Schema.Struct({ _tag: Schema.String })
const Read = Schema.Struct({ status: Schema.String })
/** Creates the order `id` of ada through `client`, and asks for its payment. */
const order = (client: HttpClient.HttpClient, id: string) =>
Effect.gen(function*() {
const created = yield* client.put(`/orders/${id}`, {
headers: as('customer:ada', `c-${id}`),
body: HttpBody.jsonUnsafe({ totalInCents: 1_000 }),
})
expect(created.status).toBe(201)
const paid = yield* client.post(`/orders/${id}/events`, {
headers: as('customer:ada', `p-${id}`),
body: HttpBody.jsonUnsafe({ _tag: 'PayRequested', payment: 'card-1' }),
}).pipe(Effect.flatMap(HttpClientResponse.schemaBodyJson(Tagged)))
expect(paid).toEqual({ _tag: 'Accepted' })
})
/** Waits until the order `id` reads `Paid` through `client`. */
const paid = (client: HttpClient.HttpClient, id: string) =>
client.get(`/orders/${id}`).pipe(
Effect.flatMap(HttpClientResponse.schemaBodyJson(Read)),
Effect.filterOrFail(({ status }) => status === 'Paid'),
Effect.retry({ schedule: Schedule.spaced('100 millis'), times: 400 }),
)
/**
* Sends `message` to the order `id` as `who` until it is answered `tag`, each try under a request id of its own,
* since a retry under the same request id is answered as the first try.
*/
const answered = <M extends { readonly _tag: string }>(
client: HttpClient.HttpClient,
id: string,
who: string,
message: M,
tag: string,
) => {
let tries = 0
return Effect.suspend(() =>
client.post(`/orders/${id}/events`, {
headers: as(who, `${message._tag}-${id}-${tries++}`),
body: HttpBody.jsonUnsafe(message),
})
).pipe(
Effect.flatMap(HttpClientResponse.schemaBodyJson(Tagged)),
Effect.filterOrFail(({ _tag }) => _tag === tag),
Effect.retry({ schedule: Schedule.spaced('100 millis'), times: 200 }),
)
}
/** The email of each order, read from the state the runtime stored in the file of `settings`. */
const emails = (settings: Settings) =>
Effect.gen(function*() {
const sql = yield* SqlClient.SqlClient
const rows = yield* sql<{ id: string; state: string }>`SELECT id, state FROM effecttea_entities`
return new Map(rows.map(({ id, state }) => [id, (JSON.parse(state) as { email?: string }).email]))
}).pipe(Effect.provide(SqliteClient.layer({ filename: settings.filename })))
/**
* Waits until each runner of `addresses` holds a lock of `cluster_locks`, so that the orders created next are divided
* between them. A runner takes its shards at its next reading of the runners, which `refreshAssignmentsInterval`
* spaces, and a shard another runner lets go may stay free for two of them, as the JSDoc of `runnerJoinDebounce`
* says.
*/
const holdingShards = (settings: Settings, addresses: ReadonlyArray<string>) =>
Effect.gen(function*() {
const sql = yield* SqlClient.SqlClient
return yield* sql<{ address: string }>`SELECT DISTINCT address FROM cluster_locks`.pipe(
Effect.filterOrFail((rows) => addresses.every((address) => rows.some((row) => row.address === address))),
Effect.retry({ schedule: Schedule.spaced('100 millis'), times: 300 }),
)
}).pipe(Effect.provide(SqliteClient.layer({ filename: settings.filename })))
/** Runs `body` with a world of `behaviour`, and a file of SQLite in a directory removed afterwards. */
const run = <A, E>(
body: (settings: Settings, theWorld: Effect.Success<ReturnType<typeof world>>) => Effect.Effect<A, E, any>,
behaviour: Behaviour = {},
) =>
Effect.runPromise(Effect.scoped(Effect.gen(function*() {
const fs = yield* FileSystem.FileSystem
const path = yield* Path.Path
const directory = yield* fs.makeTempDirectoryScoped({ prefix: 'effect-tea-persistent-server-' })
const theWorld = yield* world(behaviour)
return yield* body({ filename: path.join(directory, 'orders.db'), world: theWorld.url }, theWorld)
})).pipe(Effect.provide([BunServices.layer, FetchHttpClient.layer])) as Effect.Effect<A, unknown>)
test('an order whose process is killed during its charge is paid once by the process started again', async () => {
await run((settings, theWorld) =>
Effect.gen(function*() {
const port = yield* freePort
yield* Effect.scoped(Effect.gen(function*() {
const { process, client } = yield* start(settings, port)
const body = HttpBody.jsonUnsafe({ totalInCents: 1_000 })
expect((yield* client.put('/orders/o-1', { body })).status).toBe(401)
expect((yield* client.put('/orders/o-1', { headers: { authorization: 'Bearer customer:ada' }, body })).status)
.toBe(400)
yield* order(client, 'o-1')
const again = yield* client.put('/orders/o-1', { headers: as('customer:ada', 'c-o-1-again'), body })
expect(again.status).toBe(409)
expect(yield* HttpClientResponse.schemaBodyJson(Tagged)(again)).toEqual({ _tag: 'AlreadyExists' })
yield* theWorld.holding(1)
yield* process.kill({ killSignal: 'SIGKILL' })
}))
yield* theWorld.release
const { client } = yield* start(settings, port)
yield* paid(client, 'o-1')
expect([...(yield* theWorld.asked).values()].map((runners) => runners.length >= 2)).toEqual([true])
expect([...(yield* theWorld.charges).values()]).toEqual(['ch_1'])
})
)
}, 60_000)
slow('the orders of a runner killed during their charges are paid once by the other runner', async () => {
await run((settings, theWorld) =>
Effect.gen(function*() {
const ids = Array.from({ length: 16 }, (_, n) => `o-${n}`)
const doomed = yield* Scope.make()
const killed = yield* start(settings, yield* freePort).pipe(Scope.provide(doomed))
const survivor = yield* start(settings, yield* freePort)
yield* holdingShards(settings, [killed.address, survivor.address])
for (const id of ids) yield* order(survivor.client, id)
yield* theWorld.holding(ids.length)
const askedOfKilled = [...(yield* theWorld.asked)].filter(([, runners]) => runners[0] === killed.address)
expect(askedOfKilled.length).toBeGreaterThan(0)
yield* killed.process.kill({ killSignal: 'SIGKILL' })
yield* Scope.close(doomed, Exit.void)
yield* theWorld.release
for (const id of ids) yield* paid(survivor.client, id)
const asked = yield* theWorld.asked
for (const [key] of askedOfKilled) expect(asked.get(key)?.at(-1)).toBe(survivor.address)
expect((yield* theWorld.charges).size).toBe(ids.length)
})
)
}, 90_000)
// The process stops on the signal as its layer's scope closes, and Cluster lets the runner's shards go as the runner
// unregisters. Without `DRAIN_MILLIS` the server launches its runtime without `drain`, so its charges stop as in a
// crash (decision 0119).
// Under a lease and a health check longer than the test, only the unregistration frees the shards. On Windows a
// child process is ended at once by any signal, so the stop cannot be tested there. It waits as `slow` says.
test.skipIf(process.platform === 'win32' || Bun.env['SCHEDULES'] !== 'all')(
'the orders of a runner stopped by SIGTERM in their charges are paid once by the other runner, within its lease',
async () => {
await run((settings, theWorld) =>
Effect.gen(function*() {
const lease = { SHARD_LOCK_EXPIRATION: '10 minutes', RUNNER_HEALTH_CHECK_INTERVAL: '10 minutes' }
const ids = Array.from({ length: 16 }, (_, n) => `o-${n}`)
const doomed = yield* Scope.make()
const stopped = yield* start(settings, yield* freePort, lease).pipe(Scope.provide(doomed))
const survivor = yield* start(settings, yield* freePort, lease)
yield* holdingShards(settings, [stopped.address, survivor.address])
for (const id of ids) yield* order(survivor.client, id)
yield* theWorld.holding(ids.length)
const askedOfStopped = [...(yield* theWorld.asked)].filter(([, runners]) => runners[0] === stopped.address)
expect(askedOfStopped.length).toBeGreaterThan(0)
yield* stopped.process.kill({ killSignal: 'SIGTERM' })
yield* stopped.process.exitCode
yield* Scope.close(doomed, Exit.void)
yield* theWorld.release
for (const id of ids) yield* paid(survivor.client, id)
const asked = yield* theWorld.asked
for (const [key] of askedOfStopped) expect(asked.get(key)?.at(-1)).toBe(survivor.address)
expect((yield* theWorld.charges).size).toBe(ids.length)
})
)
},
90_000,
)
// With `DRAIN_MILLIS` the stop waits for the emails under way, and commits their completions before the runner lets
// its shards go. The other runner handles each completion, and finds each email sent and not unsure (decisions 0119,
// 0219). The world holds the emails, so the drain still holds the process two seconds after the signal. The emails are
// released only then, so that each ends in the drain and not before the stop. `kill` returns once the process has
// ended, so it runs in a fiber of its own.
test.skipIf(process.platform === 'win32')(
'a runner stopped by SIGTERM with a drain sends its emails under way once, and the other runner finishes the orders',
async () => {
await run((settings, theWorld) =>
Effect.gen(function*() {
const env = {
SHARD_LOCK_EXPIRATION: '10 minutes',
RUNNER_HEALTH_CHECK_INTERVAL: '10 minutes',
DRAIN_MILLIS: '30000',
}
const ids = Array.from({ length: 16 }, (_, n) => `o-${n}`)
yield* theWorld.release
const doomed = yield* Scope.make()
const stopped = yield* start(settings, yield* freePort, env).pipe(Scope.provide(doomed))
const survivor = yield* start(settings, yield* freePort, env)
yield* holdingShards(settings, [stopped.address, survivor.address])
for (const id of ids) yield* order(survivor.client, id)
for (const id of ids) yield* paid(survivor.client, id)
for (const id of ids) {
yield* answered(survivor.client, id, 'warehouse', { _tag: 'ShipmentReported', tracking: id }, 'Accepted')
}
yield* theWorld.holdingMails(ids.length)
const mailedByStopped = (yield* theWorld.mailed).filter(({ runner }) => runner === stopped.address)
expect(mailedByStopped.length).toBeGreaterThan(0)
const killing = yield* Effect.forkChild(stopped.process.kill({ killSignal: 'SIGTERM' }))
expect(Option.isNone(yield* Effect.timeoutOption(stopped.process.exitCode, '2 seconds'))).toBe(true)
yield* theWorld.releaseMails
yield* Fiber.join(killing)
yield* Scope.close(doomed, Exit.void)
for (const id of ids) {
yield* answered(survivor.client, id, 'customer:ada', { _tag: 'CancelRequested' }, 'Rejected')
}
const mailed = yield* theWorld.mailed
expect(mailed.length).toBe(ids.length)
expect(new Set(mailed.map(({ subject }) => subject)).size).toBe(ids.length)
const stored = yield* emails(settings)
expect(ids.map((id) => stored.get(id))).toEqual(ids.map(() => 'Sent'))
}), { holdingMails: true })
},
90_000,
)
// A charge carries an idempotency key, so its client tries it again. An email carries none and runs at most once,
// so the failure of its call suspends the order, which is then answered `409` and not `503` (decisions 0178, 0190).
test('a charge the world fails is tried again, and an email it fails suspends its order, answered 409', async () => {
await run((settings, theWorld) =>
Effect.gen(function*() {
yield* theWorld.release
const { client } = yield* start(settings, yield* freePort)
yield* order(client, 'o-1')
yield* paid(client, 'o-1')
expect([...(yield* theWorld.charges).values()]).toEqual(['ch_1'])
// A retry under the same request id is answered as the first time, so each try is a request of its own. The
// warehouse reports the shipment once it was told, and an earlier report is rejected and sent again.
let tries = 0
yield* Effect.suspend(() => client.post('/orders/o-1/events', {
headers: as('warehouse', `s-${tries++}`),
body: HttpBody.jsonUnsafe({ _tag: 'ShipmentReported', tracking: 'tr-1' }),
})).pipe(
Effect.flatMap(HttpClientResponse.schemaBodyJson(Tagged)),
Effect.filterOrFail(({ _tag }) => _tag === 'Accepted'),
Effect.retry({ schedule: Schedule.spaced('100 millis'), times: 200 }),
)
const suspended = yield* Effect.suspend(() => client.post('/orders/o-1/events', {
headers: as('customer:ada', `x-${tries++}`),
body: HttpBody.jsonUnsafe({ _tag: 'CancelRequested' }),
})).pipe(
Effect.flatMap((response) =>
Effect.map(HttpClientResponse.schemaBodyJson(Tagged)(response), (body) => ({ status: response.status, body }))
),
Effect.filterOrFail(({ body }) => body._tag === 'Suspended'),
Effect.retry({ schedule: Schedule.spaced('100 millis'), times: 200 }),
)
expect(suspended).toEqual({ status: 409, body: { _tag: 'Suspended' } })
expect(yield* theWorld.mails).toBe(1)
}), { failing: true })
}, 60_000)