# Effect Rpc Cluster

> Build typed RPC endpoints and cluster-distributed entities, singletons, cron jobs, and durable workflows with Effect's RPC and Cluster modules (Rpc/RpcGroup/RpcServer/RpcClient, Entity/Sharding/Singleton, Node/Bun bundles). Use when building RPC services or distributed/clustered Effect systems.

- Skill: `mpsuesser/effect-rpc-cluster` (Agent Skill)
- Install (CLI): `npx skillmds@latest add mpsuesser/effect-rpc-cluster`
- Raw SKILL.md: https://api.skillmd.com/api/skills/mpsuesser/effect-rpc-cluster/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: mpsuesser (https://skillmd.com/u/mpsuesser)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/mpsuesser/effect-rpc-cluster

---


You are an Effect TypeScript expert specializing in `effect/unstable/rpc` and `effect/unstable/cluster`.

These modules live under `effect/unstable/*`. There are no `@effect/rpc` or `@effect/cluster` packages in v4 — everything ships from the `effect` package. APIs may move between betas.

## Effect Source Reference

The Effect v4 source is at `~/.local/share/opencode/repos/github.com/Effect-TS/effect@main/`. Read it directly when in doubt — the shape of these modules changes more often than the website docs.

Key files:

- `packages/effect/src/unstable/rpc/Rpc.ts` — `Rpc.make`, custom constructors, `Wrapper`, `ServerClient`, `exitSchema`
- `packages/effect/src/unstable/rpc/RpcGroup.ts` — group construction, handler wiring (`toLayer` / `toHandlers` / `toLayerHandler` / `accessHandler`), prefixing, omit/merge, annotations
- `packages/effect/src/unstable/rpc/RpcServer.ts` — `make`, `layer`, `layerHttp`, every `layerProtocol*` and `toHttpEffect*`
- `packages/effect/src/unstable/rpc/RpcClient.ts` — `make`, `Protocol`, every `layerProtocol*`, `withHeaders`, `CurrentHeaders`, `ConnectionHooks`
- `packages/effect/src/unstable/rpc/RpcMiddleware.ts` — `Service` constructor, `layerClient`
- `packages/effect/src/unstable/rpc/RpcSerialization.ts` — json/ndjson/jsonRpc/ndJsonRpc/msgPack codecs and their layers
- `packages/effect/src/unstable/rpc/RpcTest.ts` — in-process test client
- `packages/effect/src/unstable/rpc/RpcWorker.ts` — `InitialMessage` for worker transports
- `packages/effect/src/unstable/rpc/RpcSchema.ts` — `Stream` schema marker, `ClientAbort` cause annotation
- `packages/effect/src/unstable/rpc/RpcClientError.ts` — client-side error union
- `packages/effect/src/unstable/cluster/Entity.ts` — `Entity.make` / `fromRpcGroup`, handler envelopes, `Replier`, `CurrentAddress`, `keepAlive`, `makeTestClient`
- `packages/effect/src/unstable/cluster/ClusterSchema.ts` — `Persisted`, `Uninterruptible`, `WithTransaction`, `ShardGroup`, `ClientTracingEnabled`, `Dynamic`
- `packages/effect/src/unstable/cluster/ClusterError.ts` — `MailboxFull`, `AlreadyProcessingMessage`, `PersistenceError`, `EntityNotAssignedToRunner`, `MalformedMessage`, `RunnerUnavailable`, `RunnerNotRegistered`
- `packages/effect/src/unstable/cluster/Sharding.ts` — the `Sharding` service surface
- `packages/effect/src/unstable/cluster/ShardingConfig.ts` — config schema + env loader
- `packages/effect/src/unstable/cluster/Singleton.ts` — singleton-per-cluster effects
- `packages/effect/src/unstable/cluster/ClusterCron.ts` — cron-driven singletons
- `packages/effect/src/unstable/cluster/SingleRunner.ts` — single-node sql-backed bundle
- `packages/effect/src/unstable/cluster/TestRunner.ts` — in-memory testing bundle
- `packages/effect/src/unstable/cluster/EntityProxy.ts` + `EntityProxyServer.ts` — entity ↔ RPC/HTTP bridge
- `packages/effect/src/unstable/workflow/WorkflowProxy.ts` + `WorkflowProxyServer.ts` — workflow ↔ RPC/HTTP bridge
- `packages/effect/src/unstable/cluster/ClusterWorkflowEngine.ts` — production workflow engine backed by sharding + storage
- `packages/effect/src/unstable/reactivity/AtomRpc.ts` — reactive RPC client for Atom UIs (see also `effect-atom-rpc` skill)
- `packages/platform/node/src/NodeClusterHttp.ts` / `NodeClusterSocket.ts` — Node "all-in-one" cluster layers
- `packages/platform/bun/src/BunClusterHttp.ts` / `BunClusterSocket.ts` — Bun equivalents
- `packages/platform/node/test/RpcServer.test.ts` + `test/fixtures/rpc-{schemas,e2e}.ts` — best end-to-end reference for real RPC wiring
- `packages/effect/test/cluster/TestEntity.ts` + `test/cluster/Entity.test.ts` — best reference for Entity + makeTestClient

## Imports

```ts
// RPC
import {
	Rpc,
	RpcClient,
	RpcGroup,
	RpcMiddleware,
	RpcSchema,
	RpcSerialization,
	RpcServer,
	RpcTest,
	RpcWorker
} from 'effect/unstable/rpc';
import { RpcClientError } from 'effect/unstable/rpc/RpcClientError';

// Cluster
import {
	ClusterCron,
	ClusterError,
	ClusterSchema,
	Entity,
	EntityProxy,
	EntityProxyServer,
	MessageStorage,
	RunnerHealth,
	Runners,
	RunnerStorage,
	Sharding,
	ShardingConfig,
	SingleRunner,
	Singleton,
	SqlMessageStorage,
	SqlRunnerStorage,
	TestRunner
} from 'effect/unstable/cluster';

// Workflow (see effect-workflow skill for the full surface)
import {
	Activity,
	DurableClock,
	DurableDeferred,
	Workflow,
	WorkflowProxy,
	WorkflowProxyServer
} from 'effect/unstable/workflow';
import { ClusterWorkflowEngine } from 'effect/unstable/cluster';

// Platform "all-in-one" cluster bundles
import { NodeClusterHttp, NodeClusterSocket } from '@effect/platform-node';
// or
import { BunClusterHttp, BunClusterSocket } from '@effect/platform-bun';
```

## Architecture at a Glance

```
                   wire format (json | ndjson | msgpack | jsonRpc | ndJsonRpc)
                                          │
       ┌──────────────┐        Protocol   │   Protocol         ┌──────────────┐
       │  RpcClient   │  ───────────────► │ ◄───────────────── │  RpcServer   │
       │  (make)      │   http/ws/socket/ │   http/ws/socket/  │  (layer)     │
       └──────┬───────┘   stdio/worker    │   stdio/worker     └──────┬───────┘
              │                                                        │
        client middlewares                                       server middlewares
              │                                                        │
              ▼                                                        ▼
     RpcGroup.make(...rpcs)  ◄── shared definition ──►   RpcGroup.toLayer(handlers)


   For distributed actor-style state:

       ┌──────────────────────────┐         entity rpcs travel through
       │  Entity.make(type,rpcs)  │  ───►   MessageStorage (durable) and
       │  ─ toLayer(handlers)     │         routed by Sharding to the
       │  ─ toLayerQueue(...)     │         runner that owns the entityId's shard
       │  ─ client                │
       └──────────────────────────┘
```

Two big invariants:

1. **An `Rpc` is a definition.** The same `Rpc` value can be served by an `RpcServer`, called by an `RpcClient`, mounted in an `Entity`, exposed via `EntityProxy.toRpcGroup`/`toHttpApiGroup`, or driven from `AtomRpc.query`/`mutation`. Define rpcs in a shared module so all sides share types.
2. **`RpcGroup` handlers and `Entity` handlers have different signatures.** RpcGroup handlers take `(payload, options)`; Entity handlers take `(envelope)`. Mixing them up is the most common mistake.

## Defining RPCs

`Rpc.make(tag, options?)` returns an `Rpc` value. It is *both* a value and a constructor — you can use it either way:

```ts
import { Schema } from 'effect';
import { Rpc } from 'effect/unstable/rpc';

// Style A — const value. Compact, fine for ad-hoc rpcs.
export const Ping = Rpc.make('Ping', { success: Schema.String });

// Style B — class extends. Gives the rpc a nominal class identity, useful
// when you want to import it as a type and pattern-match on it.
export class GetUser extends Rpc.make('GetUser', {
	success: User,
	payload: { id: Schema.String }
}) {}
```

Both styles are official. The platform-node test fixtures and the cluster test fixtures use both deliberately. Pick by feel:

- `class extends` when the rpc is shared across many modules and the nominal type helps documentation/imports
- `const` when you're listing a dozen rpcs in one file and the boilerplate hurts more than the nominal type helps

> Note: this is **not** the same as `Workflow.make`, `Activity.make`, `Entity.make`, or `RpcGroup.make` — those all return plain values you assign with `const`. The class-extends pattern is unique to `Rpc.make` (and to `Schema.Class`-style constructors) because `Rpc` declares `new (_: never): {}` in its interface.

### `Rpc.make` options

```ts
Rpc.make(tag, {
	payload?:    Schema.Top | Schema.Struct.Fields,  // struct fields or a Schema
	success?:    Schema.Top,                          // default Schema.Void
	error?:      Schema.Top,                          // default Schema.Never
	defect?:     Schema.Top,                          // default Schema.Defect()
	stream?:     boolean,                             // default false
	primaryKey?: (payload) => string                  // for cluster dedup / persistence
})
```

#### Payload as struct fields vs Schema

Passing a `Schema.Struct.Fields` literal lets `Rpc.make` build the struct for you. Passing a `Schema.Class` (or any `Schema.Top`) lets you reuse a named type:

```ts
// inline fields
Rpc.make('CreateUser', {
	payload: { name: Schema.String, email: Schema.String },
	success: User
});

// named class — preferred when the payload is reused
class CreateUserInput extends Schema.Class<CreateUserInput>('CreateUserInput')({
	name: Schema.String,
	email: Schema.String
}) {}
Rpc.make('CreateUser', { payload: CreateUserInput, success: User });
```

#### `defect` — custom defect schema (round-trip preservation)

By default `Rpc.make` uses `Schema.Defect()`, which round-trips defects as `unknown`. To keep stack traces, custom error names, or other defect properties intact across the wire, set an explicit defect schema:

```ts
import { Schema } from 'effect';

const DiagnosticDefect = Schema.Struct({
	name: Schema.String,
	message: Schema.String,
	stack: Schema.OptionFromNullishOr(Schema.String)
});

const Risky = Rpc.make('Risky', {
	success: Schema.Void,
	defect: Schema.Defect({ includeStack: true })
});
```

The cluster test fixture uses this: a handler does `Effect.die({ message, stack, name: 'CustomDefect' })` and the client receives the full object with stack intact.

#### `primaryKey` — deterministic envelope identity

`primaryKey` is required for cluster persistence to dedupe a request: the same payload that produces the same key will be treated as the same envelope, so retried sends are safe. It also makes `Rpc.make` build a `Schema.Class` for the payload (with `PrimaryKey.symbol` implemented) so `instanceof` works.

```ts
const Charge = Rpc.make('Charge', {
	payload: { invoiceId: Schema.String, amountCents: Schema.Int },
	success: ChargeReceipt,
	error: ChargeError,
	primaryKey: ({ invoiceId }) => invoiceId
});
```

#### `stream: true`

When true, `success` becomes the *element* schema, not the Effect's success. The actual return type the handler must produce is `Stream<success, error, R>` (or an `Effect<Queue.Dequeue<success, error | Cause.Done>, ...>` if the handler wants to control the queue itself). The client sees `Stream<success, error, R>` (default) or `Queue.Dequeue<success, error | Cause.Done>` if you pass `{ asQueue: true }`.

```ts
const Subscribe = Rpc.make('Subscribe', {
	payload: { topic: Schema.String },
	success: EventMessage, // element type
	error: SubscriptionError,
	stream: true
});
```

### Pipeable rpc combinators

An `Rpc` is `Pipeable`. The instance methods you'll actually use:

```ts
Rpc.make('GetUser', { ... })
	.middleware(AuthMiddleware)               // attach middleware
	.annotate(ClusterSchema.Persisted, true)  // single annotation
	.annotateMerge(otherContext)              // merge a Context.Context<I>
	.prefix('users.')                         // becomes 'users.GetUser'
	.setSuccess(NewSuccessSchema)             // swap the success schema
	.setError(NewErrorSchema)                 // swap the error schema
	.setPayload({ id: Schema.String })        // swap the payload schema
```

`prefix` is the right way to namespace rpcs when merging groups. `annotate` puts data on the rpc itself; `RpcGroup` has a separate `annotateRpcs` for marking *every rpc currently in the group*.

### `Rpc.fork` and `Rpc.uninterruptible`

These are **wrappers**, not options. They wrap a handler's *return value* (Effect or Stream) and tell the server to:

- `Rpc.fork(value)` — bypass the server instance's shared concurrency semaphore. Use for read-only or idempotent handlers that should not back up behind sequential ones.
- `Rpc.uninterruptible(value)` — run the handler in `Effect.uninterruptible`. Use for handlers that must complete (cleanup, finalize-then-return) regardless of client cancellation.
- `Rpc.wrap({ fork?, uninterruptible? })(value)` — apply both at once.

```ts
GetCount: () => Ref.get(count).pipe(Rpc.fork);
Charge: (payload) => chargeIdempotent(payload).pipe(Rpc.uninterruptible);
```

If you ever need to introspect: `Rpc.isWrapper(value)`, `Rpc.unwrap(value)`, `Rpc.wrapMap(value, f)`.

### `Rpc.exitSchema(rpc)`

Returns a `Schema.Exit<Success, Error, Defect>` for the rpc that includes any middleware-added errors. Useful for testing serialization or building generic envelope inspectors.

### `Rpc.custom` — higher-order rpc constructors

Rare but powerful: build a constructor that transforms every rpc's success/error schemas. Lets you encode a convention like "every list endpoint returns a paginated wrapper":

```ts
import { Rpc } from 'effect/unstable/rpc';
import { Schema } from 'effect';

interface PaginatedRpc extends Rpc.Custom {
	readonly out: Rpc.Custom.Out<
		Schema.Struct<{
			offset: typeof Schema.Number;
			total: typeof Schema.Number;
			results: Schema.$Array<this['success']>;
		}>,
		this['error']
	>;
}

const paginatedRpc = Rpc.custom<PaginatedRpc>((schemas) => ({
	...schemas,
	success: Schema.Struct({
		offset: Schema.Number,
		total: Schema.Number,
		results: Schema.Array(schemas.success)
	})
}));

// then use exactly like Rpc.make
const ListUsers = paginatedRpc('listUsers', { success: User });
```

## RpcGroup

`RpcGroup.make(...rpcs)` collects rpcs. Variadic — not named.

```ts
const UsersGroup = RpcGroup.make(GetUser, CreateUser, DeleteUser);
```

### Combining groups

```ts
UsersGroup
	.add(ListUsers, UpdateUser) // append rpcs
	.merge(OrdersGroup, PaymentsGroup) // union of groups (later annotations win)
	.omit('DeleteUser') // remove by tag
	.prefix('v2.'); // namespace every rpc
```

`merge` is shallow on both rpcs and group annotations — the *latest* value for any annotation key wins. If you need deeper composition, build the group from scratch.

### Group-level annotations

There are two flavors and both have a `Merge` variant:

```ts
group.annotate(SomeKey, value); // attach to the group itself
group.annotateMerge(context); // merge a Context.Context<I> into the group
group.annotateRpcs(SomeKey, value); // attach to every rpc currently in the group
group.annotateRpcsMerge(context); // merge a Context.Context<I> into every rpc
```

`annotateRpcs*` is the canonical way to mark a whole group `Persisted`, `Uninterruptible`, etc., without touching each rpc:

```ts
const PersistedUsers = UsersGroup.annotateRpcs(ClusterSchema.Persisted, true);
```

### Adding middleware to a group

`group.middleware(M)` appends `M` to *every rpc currently in the group* and returns a new group:

```ts
const AuthedUsers = UsersGroup.middleware(AuthMiddleware);
```

Rpcs added afterward via `.add(...)` won't have the middleware automatically — apply `.middleware(...)` again, or call it on the rpc directly before adding.

## Server-side handlers

A handler for an `RpcGroup` rpc has this shape:

```ts
type Handler<R extends Rpc.Any> = (
	payload: Rpc.Payload<R>,
	options: {
		readonly client: Rpc.ServerClient; // per-connection identity + annotations
		readonly requestId: RequestId;
		readonly headers: Headers;
		readonly rpc: R;
	}
) => Effect<Result, Error, Services> | Stream<Result, Error, Services>;
```

Note: the option is `client: ServerClient`, not `clientId: number`. `ServerClient` exposes `client.id: number` and a mutable `annotations: Context.Context<never>` you can extend with `client.annotate(key, value)` from middleware.

Entity handlers have a *different* signature — see the Entity section.

### Deferred responses

A non-stream handler may return an `Effect` that succeeds with a `Deferred<Success, Error>` instead of the success value directly. The server acknowledges the request but **does not send the final `Exit`** until that `Deferred` completes — useful when the result depends on a later external event and you don't want to hold a streaming connection open:

```ts
import { Deferred, Effect } from 'effect';

GetUserDeferred: () => {
	const deferred = Deferred.makeUnsafe<User>();
	// complete it later — e.g. from a webhook, another fiber, or a queue worker
	Deferred.doneUnsafe(deferred, Effect.succeed(new User({ id: '1', name: 'John' })));
	return Effect.succeed(deferred);
};
```

The client still sees a plain `Effect<Success, Error>`; the deferred round-trip is invisible on the wire.

### `group.toLayer(handlers | Effect<handlers>)`

The 80% case. Build all handlers and turn the result into a Layer that the server picks up:

```ts
const UsersLive = UsersGroup.toLayer(
	Effect.gen(function*() {
		const db = yield* Database;
		return UsersGroup.of({
			GetUser: (payload) => db.findUser(payload.id),
			CreateUser: (payload, { client, headers }) =>
				db
					.createUser(payload)
					.pipe(Effect.tap(() => Effect.logInfo('user created by', client.id))),
			DeleteUser: (payload) => db.deleteUser(payload.id)
		});
	})
);
```

`group.of(handlers)` is a no-op identity helper that *typechecks* the handler shape against the group. Always use it inside `toLayer` so type errors point at the wrong handler.

### `group.toLayerHandler(tag, handler | Effect<handler>)`

Implement *one* handler at a time. Useful when handlers have wildly different dependencies and you want to keep them in separate files:

```ts
const GetUserLive = UsersGroup.toLayerHandler(
	'GetUser',
	Effect.gen(function*() {
		const db = yield* Database;
		return (payload) => db.findUser(payload.id);
	})
);

// Compose them:
const UsersLive = Layer.mergeAll(GetUserLive, CreateUserLive, DeleteUserLive);
```

Each `toLayerHandler` produces `Layer<Rpc.Handler<Tag>, ...>`. The server requires the union `Rpc.ToHandler<Rpcs>` so leaving any tag unimplemented is a compile-time error.

### `group.toHandlers(handlers)`

Returns an `Effect<Context.Context<Rpc.ToHandler<R>>>` — the unprovided form of `toLayer`. Use it when composing manually inside `RpcServer.make` or `RpcTest.makeClient`.

### `group.accessHandler(tag)`

Returns an Effect that resolves to a single handler function with `services` already provided. The handler is callable as `(payload, options)` directly. This is the easiest way to unit-test one rpc handler in isolation:

```ts
import { Headers } from 'effect/unstable/http';
import { RequestId } from 'effect/unstable/rpc/RpcMessage';

const result =
	yield*
	UsersGroup.accessHandler('GetUser').pipe(
		Effect.flatMap((handler) =>
			handler({ id: 'u1' }, {
				client: new Rpc.ServerClient(0),
				requestId: RequestId(1),
				headers: Headers.empty,
				rpc: GetUser
			})
		),
		Effect.provide(UsersLive)
	);
```

## Running an RPC server

Two layers of API: the **transport-agnostic** server and the **transport-specific** glue.

### `RpcServer.layer(group, options?)` — transport-agnostic

Requires a `Protocol` in context (one of the `RpcServer.layerProtocol*`), the handlers (`Rpc.ToHandler<Rpcs>`), and any middleware (`Rpc.Middleware<Rpcs>`):

```ts
import { Layer } from 'effect';
import { HttpRouter } from 'effect/unstable/http';

const ServerLayer = RpcServer.layer(UsersGroup, {
	concurrency: 'unbounded', // default; set a number to backpressure handlers
	disableTracing: false,
	disableFatalDefects: false, // see below
	spanPrefix: 'RpcServer', // default; controls span naming
	spanAttributes: { service: 'users' }
}).pipe(
	Layer.provide(UsersLive), // handlers
	Layer.provide(RpcServer.layerProtocolHttp({ path: '/rpc' })),
	Layer.provide(RpcSerialization.layerNdjson),
	Layer.provide(HttpRouter.layer)
);
```

Server options:

- **`concurrency: number | 'unbounded'`** (default `'unbounded'`) — one semaphore around handler execution for the whole server instance, shared by all clients. `Rpc.fork(...)` opts a single handler out of this limit.
- **`disableFatalDefects: boolean`** (default `false`) — by default, a `die` inside a handler is treated as a *connection-level* defect and crashes the whole connection's response stream. With `true`, defects come back to the client as a normal `Cause.Die` in the request's exit. Production servers usually want `true`; the cluster fixture uses it.
- **`disableTracing: boolean`** + **`spanPrefix`** + **`spanAttributes`** — span control. Each rpc gets a span named `${spanPrefix}.${rpc._tag}`.

### `RpcServer.layerHttp({ group, path, protocol })` — convenience

One-call HTTP+server setup. Picks `layerProtocolHttp` or `layerProtocolWebsocket` for you (default `'websocket'`):

```ts
const ServerLayer = RpcServer.layerHttp({
	group: UsersGroup,
	path: '/api/rpc',
	protocol: 'http', // or 'websocket' (default)
	disableFatalDefects: true,
	concurrency: 'unbounded',
	streamBufferSize: 16 // framed HTTP response queue; default 16
}).pipe(
	Layer.provide(UsersLive),
	Layer.provide(RpcSerialization.layerNdjson),
	Layer.provide(HttpRouter.layer)
);
```

### `RpcServer.toHttpEffect(group, options?)` and `toHttpEffectWebsocket`

For when you want to mount the RPC handler as a single `HttpServerResponse` Effect on a router you control (Hono adapter, custom routes, etc.) rather than registering a route on `HttpRouter`. Returns `Effect<Effect<HttpServerResponse, never, Scope | HttpServerRequest>, ...>`:

```ts
const makeHttpApp = RpcServer.toHttpEffect(UsersGroup).pipe(
	Effect.provide(UsersLive),
	Effect.provide(RpcSerialization.layerNdjson)
);
```

Run `makeHttpApp` in the scope that owns the router or framework adapter, then mount the returned request/response effect there.

### Protocol layers (server side)

Pick one and `Layer.provide` it to `RpcServer.layer`/`layerHttp`:

| Layer | Requires | Notes |
|---|---|---|
| `RpcServer.layerProtocolHttp({ path, streamBufferSize? })` | `RpcSerialization`, `HttpRouter` | request/response, **no streaming acks** (`supportsAck: false`), no transferables, no span propagation |
| `RpcServer.layerProtocolWebsocket({ path })` | `RpcSerialization`, `HttpRouter` | full duplex, supports acks, supports span propagation |
| `RpcServer.layerProtocolSocketServer` | `RpcSerialization`, `SocketServer` | raw TCP socket server |
| `RpcServer.layerProtocolStdio` | `RpcSerialization`, `Stdio` | process stdin/stdout — for CLI subprocess RPC |
| `RpcServer.layerProtocolWorkerRunner` | `WorkerRunner.WorkerRunnerPlatform` | run inside a web/node worker; supports `RpcWorker.InitialMessage` |

Each layer also has a `make*` Effect counterpart (`makeProtocolHttp`, `makeProtocolWebsocket`, etc.) when you need to compose it inline. There are also `makeProtocolWithHttpEffect({ streamBufferSize? })` / `makeProtocolWithHttpEffectWebsocket` for "give me both the protocol and the http handler Effect" use cases. `makeProtocolWithHttpEffect` is a function, so call it as `yield* RpcServer.makeProtocolWithHttpEffect()` when using defaults.

Framed HTTP response queues are bounded to `16` messages by default. Configure `streamBufferSize` on `layerHttp`, `layerProtocolHttp`, `makeProtocolHttp`, `makeProtocolWithHttpEffect`, or `toHttpEffect`; pass `'unbounded'` only when unbounded buffering is intentional.

### `RpcServer.Protocol` service

The `Protocol` service exposes runtime capabilities tests and middleware can inspect:

```ts
const {
	supportsAck,
	supportsTransferables,
	supportsSpanPropagation,
	supportsNotifications,
	clientIds,
	initialMessage
} =
	yield* RpcServer.Protocol;
```

E2E tests use this to skip backpressure assertions on transports that don't support acks. `supportsNotifications` is true for sockets, stdio, workers, and framed HTTP; unframed buffered HTTP drops server notifications.

`RpcMessage.FromServerEncoded` now includes `RequestEncoded` for server-originated requests and notifications. Notifications set `isNotification: true`; JSON-RPC then omits the id. Custom server protocols must declare `supportsNotifications`.

## RPC clients

`RpcClient.make(group, options?)` returns an Effect producing a typed client object. Default error channel is `RpcClientError`.

```ts
const client = yield* RpcClient.make(UsersGroup, {
	spanPrefix: 'UsersClient',
	disableTracing: false,
	flatten: false,
	generateRequestId: undefined,
	spanAttributes: { service: 'users' }
}).pipe(
	Effect.provide(RpcClient.layerProtocolHttp({ url: '/api/rpc' })),
	Effect.provide(RpcSerialization.layerNdjson),
	Effect.provide(FetchHttpClient.layer)
);

const user = yield* client.GetUser({ id: 'u1' });
```

You will almost always wrap this in a `Context.Service` so consumers get the client by name instead of plumbing the Effect:

```ts
class UsersClient extends Context.Service<
	UsersClient,
	RpcClient.RpcClient<RpcGroup.Rpcs<typeof UsersGroup>, RpcClientError>
>()('UsersClient') {
	static readonly layer = Layer.effect(UsersClient)(
		RpcClient.make(UsersGroup)
	).pipe(Layer.provide(AuthClient));
}
```

### Per-call options

Each generated method is `(payload, options?) => Effect | Stream`. The option shape differs by stream-vs-non-stream:

```ts
// Non-stream rpc
client.GetUser({ id: 'u1' }, {
	headers?: Headers.Input,    // per-call headers
	context?: Context<never>,   // per-call context (rare)
	discard?: true              // returns Effect<void, transport | middleware errors>; no response decoding
});

// Stream rpc
client.Subscribe({ topic: 't' }, {
	headers?: Headers.Input,
	context?: Context<never>,
	asQueue?: true,             // returns Effect<Queue.Dequeue<A, E | Cause.Done>>
	streamBufferSize?: number   // default 16
});
```

`discard: true` skips response decoding and removes response-side failures;
transport and required client-middleware failures can still occur. A successful
send is not proof that the server completed the operation. Cluster persistent
messages additionally have their documented storage/delivery outcomes.

`asQueue: true` is useful when you need finer control than a `Stream` gives you — e.g., you want to take only one chunk, then drop it. The end-of-stream signal is `Cause.Done` in the queue's error channel.

### Headers

For one-off headers, use the per-call `headers` option. For region-scoped headers, use `RpcClient.withHeaders` (which updates the `RpcClient.CurrentHeaders` Reference):

```ts
import { RpcClient } from 'effect/unstable/rpc';

yield* program.pipe(
	RpcClient.withHeaders({ authorization: `Bearer ${token}`, userid: '123' })
);
```

`RpcClient.CurrentHeaders` is a `Context.Reference<Headers.Headers>` you can also set directly with `Effect.updateService`. Headers from `withHeaders` and the per-call option are merged; per-call wins on conflict.

### `flatten: true` mode

When set, the client becomes a single function `(tag, payload, options?)` instead of a property-per-tag object. `AtomRpc` uses this internally; you'll want it when proxying generically:

```ts
const client =
	yield*
	RpcClient.make(UsersGroup, { flatten: true });
const user = yield* client('GetUser', { id: 'u1' });
```

### Client error channel

Every method has the error channel:

```
Rpc.Error<R>                    // your declared rpc error
| MiddlewareError                  // any middleware errors
| MiddlewareClientError            // any client-side middleware errors
| RpcClientError                   // transport-level
```

`RpcClientError` is a tagged union itself:

```ts
class RpcClientError extends Schema.Error(...)({
	_tag: 'RpcClientError',
	reason: Schema.Union([
		WorkerErrorReason,
		SocketErrorReason,
		HttpClientErrorSchema,
		RpcClientDefect
	])
})
```

Pattern-match on `error.reason._tag` to handle transport faults (network down, malformed response, worker crash). The `RpcClientDefect` case wraps non-error throws and protocol bugs.

### Client protocol layers

| Layer | Requires | Notes |
|---|---|---|
| `RpcClient.layerProtocolHttp({ url, transformClient? })` | `RpcSerialization`, `HttpClient` | request/response. `transformClient` lets you rewrite the underlying `HttpClient` (e.g., add auth headers, prepend URL paths) |
| `RpcClient.layerProtocolSocket({ retryTransientErrors?, onTransientError? })` | `RpcSerialization`, `Socket.Socket` | full duplex. Auto-pings every 5s; reconnects on transient socket errors; reports retried open failures through `onTransientError` |
| `RpcClient.layerProtocolWorker(options)` | `Worker.WorkerPlatform`, `Worker.Spawner` | pool of worker-backed clients. Options: either `{ size, concurrency?, targetUtilization? }` or `{ minSize, maxSize, timeToLive, concurrency?, targetUtilization? }` |

For each there's a corresponding `make*` Effect (`makeProtocolHttp`, `makeProtocolSocket`, `makeProtocolWorker`) when you need finer control over context.

### `RpcClient.ConnectionHooks`

A `Context.Service` you can provide to get `onConnect` / `onDisconnect` callbacks for socket and worker transports. Use it to (re-)hydrate auth state on reconnect:

```ts
const ConnectionHooksLayer = Layer.succeed(RpcClient.ConnectionHooks, {
	onConnect: refreshAuthToken,
	onDisconnect: Effect.logWarning('rpc disconnected')
});
```

### `RpcSchema.ClientAbort`

When a client interrupts a subscription, inspect interrupt reasons in the exit
cause for the `ClientAbort` annotation. `onInterrupt` receives interruptor IDs,
not a Cause; use `onExit` to distinguish client cancel from server shutdown:

<!-- typecheck -->
```ts
import { RpcSchema } from 'effect/unstable/rpc';
import { Cause, Effect, Exit, Stream } from 'effect';
import * as Arr from 'effect/Array';

declare const stream: Stream.Stream<string>;
const subscribeHandler = stream.pipe(Stream.runDrain,
	Effect.onExit((exit) => {
		const isClientAbort = Exit.isFailure(exit) && Arr.some(exit.cause.reasons, (reason) =>
			Cause.isInterruptReason(reason) && reason.annotations.has(RpcSchema.ClientAbort.key));
		return Effect.logInfo('subscribe ended', { isClientAbort });
	})
);
```

## Middleware

`RpcMiddleware.Service<Self, Config>()(name, options)` defines a middleware service. The config positionally encodes what the middleware *provides*, *requires*, and what *client-only* error type it can throw. The options carry the wire-error schema and the `requiredForClient` enforcement flag.

```ts
import { RpcMiddleware } from 'effect/unstable/rpc';
import { Context, Schema } from 'effect';

class CurrentUser extends Context.Service<CurrentUser, User>()('CurrentUser') {}

class Unauthorized extends Schema.Error<Unauthorized>('Unauthorized')({
	_tag: Schema.tag('Unauthorized')
}) {}

class AuthMiddleware extends RpcMiddleware.Service<AuthMiddleware, {
	provides: CurrentUser; // injected into the wrapped handler
	requires: never; // services this middleware needs from outer context
	clientError: never; // errors only the client side can produce
}>()('AuthMiddleware', {
	error: Unauthorized, // wire-form error this middleware can produce
	requiredForClient: true // clients must supply layerClient or fail to compile
}) {}
```

The full config bag is `{ requires?, provides?, clientError? }` — all optional, all default `never`.

`requiredForClient: true` is what makes auth client-side enforcement a *compile-time* error rather than a runtime surprise: clients must `Layer.provide(RpcMiddleware.layerClient(AuthMiddleware, ...))` or `RpcClient.make` won't compile.

### Server-side middleware implementation

Implement the middleware as a Layer producing the service. The function receives `(effect, options)`:

```ts
import { Layer } from 'effect';

const AuthLive = Layer.succeed(AuthMiddleware)(
	AuthMiddleware.of((effect, { client, requestId, rpc, payload, headers }) =>
		Effect.flatMap(verifyToken(headers.authorization), (user) =>
			Effect.provideService(effect, CurrentUser, user)
		)
	)
);
```

Options shape: `{ client: ServerClient, requestId, rpc, payload, headers }`. The middleware can:

- Provide services to the inner effect (matching the `provides` config)
- Fail with the wire-error schema (`Unauthorized` here)
- Annotate `client` via `client.annotate(...)` so subsequent middlewares see the per-connection state
- Read `headers` directly (they're already parsed)

Middleware can chain `requires` and `provides` — `DbMiddleware extends RpcMiddleware.Service<…, { provides: DbConnection, requires: CurrentUser }>` will compile only when paired with an `AuthMiddleware` upstream that provides `CurrentUser`.

### Client-side middleware (`layerClient`)

For middleware that needs to *also* run client-side (most commonly: attach an auth header), provide a `layerClient`:

```ts
import { Headers } from 'effect/unstable/http';

export const AuthClient = RpcMiddleware.layerClient(
	AuthMiddleware,
	({ rpc, request, next }) =>
		next({
			...request,
			headers: Headers.set(request.headers, 'authorization', `Bearer ${currentToken}`)
		})
);
```

Important details:

- `request.headers` is `Headers.Headers` (already parsed). Use the helpers from `effect/unstable/http/Headers` (`Headers.set`, `Headers.merge`, `Headers.fromInput`).
- You **must** call `next(request)` (with the modified or original request) — the middleware's job is to wrap the send, not replace it.
- The Layer signature is `Layer<ForClient<AuthMiddleware>>` — it's a distinct service from the server-side middleware, and providing both is the norm for client packages.

## Serialization

The choice of serialization is load-bearing because of *framing*. Some transports (raw HTTP request/response) deliver one logical message at a time; others (sockets, ndjson over HTTP streams) deliver an unbounded stream of bytes that must be split into messages.

| Layer | Content-Type | Framed? | Use for | Notes |
|---|---|---|---|---|
| `RpcSerialization.layerJson` | `application/json` | no | `layerProtocolHttp` | Default JSON over request/response |
| `RpcSerialization.layerNdjson` | `application/ndjson` | yes (newline) | `layerProtocolWebsocket`, sockets, http+stream | Newline-delimited JSON; one of several streaming formats |
| `RpcSerialization.layerJsonRpc()` | `application/json` (configurable) | no | JSON-RPC 2.0 interop | Maps `_tag` to `method`; preserves batched arrays |
| `RpcSerialization.layerNdJsonRpc()` | `application/json-rpc` (configurable) | yes (newline) | JSON-RPC 2.0 over sockets | |
| `RpcSerialization.layerMsgPack` | `application/msgpack` | yes (msgpack frames) | binary transports | JSON-compatible schema codecs; uses `useRecords: true` |
| `RpcSerialization.layerSchemaBinary(options?)` | `application/vnd.effect.rpc+schema-binary` | yes | binary transports, framed HTTP | schema-aware binary payloads and envelopes |

In rc.112 serializations and both protocol services require `codecFor`.
Custom protocols forward it from the serialization; custom JSON-compatible
protocols can use `Schema.toCodecJson`. Cluster network codecs now follow the
transport while persistent message storage remains JSON. See `effect-rpc-server`
for the complete contract, binary options, and compatibility constraints.

`RpcSerialization.makeMsgPack(options?)` lets you customize msgpackr (`useRecords`, `useFloat32`, etc.).

Picking the wrong one is a real bug:

- `layerJson` over a websocket → no framing → the first chunk past the first message is misinterpreted
- `layerMsgPack` against a JSON-only HTTP client → garbled responses
- `layerNdjson` against `layerProtocolHttp` → streams incrementally through a bounded response queue (default 16), without RPC acks

## Testing — `RpcTest.makeClient`

In-process server+client wired together, no network. The simplest possible RPC test:

```ts
import { Effect, Layer } from 'effect';
import { RpcTest } from 'effect/unstable/rpc';
import { it } from '@effect/vitest';

const TestClient = Layer.effect(UsersClient)(
	RpcTest.makeClient(UsersGroup)
).pipe(Layer.provide([UsersLive, AuthLive, AuthClient]));

it.effect('GetUser', () =>
	Effect.gen(function*() {
		const client = yield* UsersClient;
		const user = yield* client.GetUser({ id: 'u1' });
		expect(user.id).toBe('u1');
	}).pipe(Effect.provide(TestClient)));
```

`makeClient` accepts `{ flatten?: boolean }` mirroring `RpcClient.make`. Required context is `Scope | Rpc.ToHandler<Rpcs> | Rpc.Middleware<Rpcs> | Rpc.MiddlewareClient<Rpcs>` — i.e. handler layers **and** any client-side middleware layers. Forgetting the latter is a common type error.

## Worker transports & `RpcWorker.InitialMessage`

For worker-backed clients, you can pass typed initial config at spawn time without a separate rpc round-trip:

```ts
// On the worker (server side):
import { RpcWorker } from 'effect/unstable/rpc';

class WorkerConfig extends Schema.Class<WorkerConfig>('WorkerConfig')({
	apiUrl: Schema.String,
	tenantId: Schema.String
}) {}

// Inside the worker, before serving:
const config = yield* RpcWorker.initialMessage(WorkerConfig);

// On the client (parent side):
const InitialMessageLayer = RpcWorker.layerInitialMessage(
	WorkerConfig,
	Effect.succeed(new WorkerConfig({ apiUrl: '/api', tenantId: 't1' }))
);

const ClientLayer = UsersClient.layer.pipe(
	Layer.provide(RpcClient.layerProtocolWorker({ size: 4 })),
	Layer.provide(InitialMessageLayer)
);
```

---

# Cluster

Cluster turns rpcs into **addressable, distributed actors** (`Entity`). Messages are routed to whichever runner currently owns the entity's shard, optionally persisted to durable storage, and replayed on restart.

## Entity

```ts
import { Schema } from 'effect';
import { ClusterSchema, Entity } from 'effect/unstable/cluster';
import { Rpc } from 'effect/unstable/rpc';

export class Increment extends Rpc.make('Increment', {
	payload: { amount: Schema.Number },
	success: Schema.Number,
	primaryKey: ({ amount }) => `inc-${amount}` // for dedup across retries
}) {}

export const GetCount = Rpc.make('GetCount', { success: Schema.Number });

export const Counter = Entity.make('Counter', [Increment, GetCount])
	.annotateRpcs(ClusterSchema.Persisted, true); // persist all messages
```

Two constructors:

- **`Entity.make(type, [rpcs])`** — variadic-array form
- **`Entity.fromRpcGroup(type, rpcGroup)`** — when the protocol already exists as an `RpcGroup` (e.g., shared with a non-clustered RPC server)

### Entity handler signature is **different from RpcGroup**

Entity handlers receive a single `envelope: Envelope.Request<R>`:

```ts
type EntityHandler<R extends Rpc.Any> = (
	envelope: {
		readonly _tag: 'Request';
		readonly requestId: Snowflake;
		readonly address: EntityAddress; // { entityType, entityId, shardId }
		readonly tag: Rpc.Tag<R>;
		readonly payload: Rpc.Payload<R>;
		readonly headers: Headers;
		readonly traceId?: string;
		readonly spanId?: string;
		readonly sampled?: boolean;
		// for stream rpcs:
		readonly lastSentChunk: Option<Reply.Chunk<R>>;
		readonly lastSentChunkValue: Option<SuccessChunk<R>>;
		readonly nextSequence: number;
	}
) => Effect | Stream;
```

You destructure `{ payload }` (and sometimes `{ payload, address }`) inside the handler. The full envelope is also useful for streaming rpcs that need to resume after a reconnect — `lastSentChunkValue` and `nextSequence` let you replay from the right offset.

```ts
export const CounterLive = Counter.toLayer(
	Effect.gen(function*() {
		const count = yield* Ref.make(0);

		return Counter.of({
			Increment: (envelope) =>
				Ref.updateAndGet(count, (n) => n + envelope.payload.amount),

			GetCount: () => Ref.get(count).pipe(Rpc.fork) // concurrent reads
		});
	}),
	{
		maxIdleTime: Duration.minutes(5),
		concurrency: 1, // sequential by default
		mailboxCapacity: 1000,
		disableFatalDefects: false,
		defectRetryPolicy: Schedule.exponential('200 millis'),
		spanAttributes: { entity: 'Counter' }
	}
);
```

`toLayer` options:

- **`maxIdleTime`** — passivation timeout. After this idle time the entity is stopped and recreated on the next message.
- **`concurrency`** (default `1`) — handlers run sequentially per entity. Set `'unbounded'` for concurrent handlers, or use `Rpc.fork(...)` per-handler.
- **`mailboxCapacity`** (default from `ShardingConfig.entityMailboxCapacity`, usually 4096) — backpressure threshold; sends fail with `MailboxFull` past this point.
- **`disableFatalDefects`** — by default a defect inside an entity hand

…(truncated)