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

// 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:

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

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:

// 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:

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.

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 }.

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:

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.

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":

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.

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

Combining groups

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:

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:

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:

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:

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:

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:

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:

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:

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>):

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'):

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>, ...>:

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:

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:

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.

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:

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:

// 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):

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:

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:

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

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:

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:

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.

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):

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:

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.

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:

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:

// 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

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>:

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.

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)