# Entities

An **entity** is a sharded, addressable actor. You give it a stable type name and an
[RPC](https://effect.plants.sh/rpc/) protocol; the cluster then routes every message for a given *entity id*
to a single live instance. While that instance is active it can hold in-memory state
(a `Ref`, an STM `TxRef`, an open connection); when it sits idle long enough it is
*passivated* — stopped and recreated on demand the next time a message arrives.

This page builds a small counter entity end to end: define its messages, implement
its handlers, call it from a client, and assemble both a production and a test
cluster. Then it walks the entire `Entity` module surface — every constructor,
combinator, per-instance service, and behavior annotation.

## A complete entity

```ts
import { NodeClusterSocket, NodeRuntime } from "@effect/platform-node"
import { Effect, Layer, Ref, Schema } from "effect"
import { ClusterSchema, Entity, TestRunner } from "effect/unstable/cluster"
import { Rpc } from "effect/unstable/rpc"
import type { SqlClient } from "effect/unstable/sql"

// --- 1. The protocol -------------------------------------------------------
// Each message is an Rpc with a typed payload and success schema.
const Increment = Rpc.make("Increment", {
  payload: { amount: Schema.Number },
  success: Schema.Number
})

const GetCount = Rpc.make("GetCount", {
  success: Schema.Number
})
  // By default messages are volatile: they are sent over the network and lost if
  // the runner dies before processing. Annotating the Rpc with
  // `ClusterSchema.Persisted` saves it to durable mailbox storage first, so it
  // survives a crash and is retried.
  .annotate(ClusterSchema.Persisted, true)

// --- 2. The entity ---------------------------------------------------------
// `Entity.make` pairs a stable type name with the array of Rpcs it handles.
const Counter = Entity.make("Counter", [Increment, GetCount])

// --- 3. The handlers -------------------------------------------------------
// `toLayer` registers the entity with Sharding. The build effect runs once per
// active entity instance, so anything created here (the Ref) is that instance's
// private, in-memory state.
const CounterLayer = Counter.toLayer(
  Effect.gen(function*() {
    const count = yield* Ref.make(0)

    return Counter.of({
      Increment: ({ payload }) => Ref.updateAndGet(count, (n) => n + payload.amount),
      GetCount: () =>
        Ref.get(count).pipe(
          // Handlers for one entity run sequentially by default, which is what
          // gives you lock-free access to `count`. `Rpc.fork` opts a read-only
          // handler out of that ordering so it can run concurrently.
          Rpc.fork
        )
    })
  }),
  // If the entity receives no messages for this long, it is passivated. The next
  // message recreates it (with a fresh count of 0).
  { maxIdleTime: "5 minutes" }
)

// --- 4. Calling the entity -------------------------------------------------
// `Counter.client` yields a function from entity id to a typed RPC client.
const useCounter = Effect.gen(function*() {
  const clientFor = yield* Counter.client
  const counter = clientFor("counter-123")

  const afterIncrement = yield* counter.Increment({ amount: 1 })
  const current = yield* counter.GetCount()
  yield* Effect.log(`after increment: ${afterIncrement}, current: ${current}`)
})

// --- 5. Wiring a cluster ---------------------------------------------------
// In production, `NodeClusterSocket.layer` provides the socket transport. By
// default it persists messages to SQL, so it needs a SqlClient.
declare const SqlClientLayer: Layer.Layer<SqlClient.SqlClient>

const ClusterLayer = NodeClusterSocket.layer().pipe(
  Layer.provide(SqlClientLayer)
)

// Merge every entity layer, then provide the cluster transport + storage.
const EntitiesLayer = Layer.mergeAll(CounterLayer)

const ProductionLayer = EntitiesLayer.pipe(
  Layer.provide(ClusterLayer)
)

// `Layer.launch` keeps the runner alive to serve messages until interrupted.
Layer.launch(ProductionLayer).pipe(NodeRuntime.runMain)
```

## How it fits together

1. **Define the protocol with RPCs.** Each `Rpc.make` declares a message name, its
   `payload` schema, and its `success` schema (and optionally an `error` schema).
   Because the protocol is plain Schema, messages serialize for the wire and for
   storage automatically.

2. **Create the entity** with `Entity.make("Type", [...rpcs])`. The type name must be
   **stable and unique** across your deployment — it participates in routing, so
   renaming it moves work to different shards.

3. **Implement handlers** with `entity.toLayer(build, options)`. The `build` effect
   produces the handler record via `entity.of({ ... })`. It runs **once per active
   instance**, which is where you allocate per-id state.

4. **Send messages** through `entity.client`. You get a function `(id) => client`;
   the returned client exposes one method per Rpc, returning an `Effect`. The cluster
   routes the call to the instance that owns `id`.

5. **Provide a cluster layer** that supplies the transport and storage, then
   `Layer.launch` it.

### State and concurrency

The build effect runs per active entity instance, so state declared inside it — the
`Ref` above — belongs to a single id and a single owner. By default, **all handlers
for one entity instance run sequentially**, which is what makes lock-free state safe.
Reach for [`Ref`](https://effect.plants.sh/state-management/) for simple values, or an STM
[`TxRef`](https://effect.plants.sh/transactions/) when a handler needs to coordinate several pieces of state
atomically.

Wrap a read-only handler with `Rpc.fork` (as `GetCount` does) to let it run
concurrently with the sequential ones. You can also tune ordering and back pressure
with `toLayer` options such as `concurrency` and `mailboxCapacity`.

### Volatile vs. persisted messages

By default messages are **volatile**: sent directly over the network, and lost if the
owning runner dies before processing. Annotating an Rpc with `ClusterSchema.Persisted`
writes the message to durable mailbox storage first, so it survives crashes and is
retried after failover.
**Caution:** Persisted messages require message storage. The default `NodeClusterSocket.layer`
persists to SQL and therefore needs a `SqlClient`. Pass `{ storage: "local" }` to use
a no-op store (messages are not durable) or `{ storage: "byo" }` to supply your own
`MessageStorage`.

## Running entities in tests

You don't need sockets or a database to exercise an entity. `TestRunner.layer`
assembles a single in-process cluster with in-memory message and runner storage — the
same entity runtime model, no network.

```ts
import { Layer } from "effect"
import { TestRunner } from "effect/unstable/cluster"

// `EntitiesLayer` is the merged entity layers from the example above.
const TestLayer = EntitiesLayer.pipe(
  // `provideMerge` keeps the cluster services (Sharding, MessageStorage, ...) in
  // the result so a test can also inspect storage directly.
  Layer.provideMerge(TestRunner.layer)
)
```

Provide `TestLayer` to a test that resolves `Counter.client` and asserts on the
results. Because the in-memory storage is scoped to the layer, each test gets a fresh,
isolated cluster. See [Testing](https://effect.plants.sh/testing/) for the `it.effect` harness used to run
effects in tests.

## Reading the current address

Inside a handler you sometimes need the entity's own identity — for logging, metrics,
or routing decisions. Yield `Entity.CurrentAddress` to read the
`EntityAddress` (its type, id, and shard) currently being processed.

```ts
import { Effect } from "effect"
import { Entity } from "effect/unstable/cluster"

const handler = Effect.gen(function*() {
  const address = yield* Entity.CurrentAddress
  yield* Effect.log(`handling ${address.entityType}/${address.entityId}`)
})
```

## Constructors and combinators

Everything below is the public surface of `effect/unstable/cluster`'s `Entity`
module. Each entry is a runnable fragment that assumes the `Counter` entity from the
top of the page.

### Entity.make(type, rpcs)

Pairs a stable `type` name with an array of `Rpc` definitions, grouping them into the
entity's protocol. This is the constructor you reach for most of the time.

```ts
import { Schema } from "effect"
import { Entity } from "effect/unstable/cluster"
import { Rpc } from "effect/unstable/rpc"

const Increment = Rpc.make("Increment", { payload: { amount: Schema.Number }, success: Schema.Number })
const GetCount = Rpc.make("GetCount", { success: Schema.Number })

const Counter = Entity.make("Counter", [Increment, GetCount])
// => Entity<"Counter", Increment | GetCount>
Counter.type // => "Counter"
```
**Note:** If two Rpcs in the array share a tag, the later definition wins for that tag.

### Entity.fromRpcGroup(type, rpcGroup)

Builds an entity from an existing [`RpcGroup`](https://effect.plants.sh/rpc/). Use this when the protocol is
already defined as a group and shared with, say, an HTTP API. `Entity.make` is just
`fromRpcGroup` over `RpcGroup.make(...rpcs)`.

```ts
import { Schema } from "effect"
import { Entity } from "effect/unstable/cluster"
import { Rpc, RpcGroup } from "effect/unstable/rpc"

const CounterRpcs = RpcGroup.make(
  Rpc.make("Increment", { payload: { amount: Schema.Number }, success: Schema.Number }),
  Rpc.make("GetCount", { success: Schema.Number })
)

const Counter = Entity.fromRpcGroup("Counter", CounterRpcs)
Counter.protocol // => RpcGroup of Increment | GetCount
```

### entity.toLayer(build, options)

Registers typed RPC handlers with `Sharding`, returning a `Layer<never>`. The `build`
argument is either a handler record or an `Effect` that yields one (use the latter to
allocate per-instance state). `build` runs **once per active entity instance**.

```ts
import { Effect, Ref } from "effect"

const CounterLayer = Counter.toLayer(
  Effect.gen(function*() {
    const count = yield* Ref.make(0)
    return Counter.of({
      Increment: ({ payload }) => Ref.updateAndGet(count, (n) => n + payload.amount),
      GetCount: () => Ref.get(count)
    })
  }),
  { maxIdleTime: "5 minutes", concurrency: 1 }
)
// => Layer<never, never, Sharding | ...handler services>
```

The `options` object (all optional):

- `maxIdleTime` — `Duration.Input`; passivate the instance after this much inactivity.
- `concurrency` — `number | "unbounded"`; how many handlers run at once (default `1`,
  i.e. sequential). `Rpc.fork` opts an individual handler out of the sequential order.
- `mailboxCapacity` — `number | "unbounded"`; bounded mailbox size for back pressure.
- `disableFatalDefects` — `boolean`; when `true`, defects do not crash the instance.
- `defectRetryPolicy` — `Schedule.Schedule<any, unknown>`; retry schedule applied when
  a handler dies with a defect.
- `spanAttributes` — `Record<string, string>`; attributes added to handler tracing spans.

### entity.toLayerQueue(build, options)

Low-level handler form: instead of a per-Rpc record, you receive a
`Queue.Dequeue<Envelope.Request>` of incoming envelopes and a `Replier` to complete
them. The build forks a long-running processor into the layer scope. Prefer this when
you need to batch envelopes, reorder them, or drive a custom event loop rather than
respond one message at a time. It runs with `concurrency: "unbounded"` internally.

```ts
import { Effect, Queue } from "effect"

const CounterLayer = Counter.toLayerQueue(
  Effect.gen(function*() {
    let count = 0
    // The returned function is forked into the layer scope.
    return (mailbox, replier) =>
      Effect.gen(function*() {
        while (true) {
          const envelope = yield* Queue.take(mailbox)
          if (envelope.tag === "Increment") {
            count += (envelope.payload as { amount: number }).amount
            yield* replier.succeed(envelope, count)
          } else {
            yield* replier.succeed(envelope, count)
          }
        }
      })
  }),
  { maxIdleTime: "5 minutes" }
)
```

`toLayerQueue` options are the same as `toLayer` minus `concurrency` (it is fixed to
`"unbounded"`).

### entity.client

An `Effect` that yields a function `(entityId) => RpcClient`. The returned client has
one method per Rpc; calling it returns an `Effect` whose error channel includes the
cluster failures `MailboxFull | AlreadyProcessingMessage | PersistenceError` on top of
each Rpc's own errors. Requires `Sharding`.

```ts
import { Effect } from "effect"

const program = Effect.gen(function*() {
  const clientFor = yield* Counter.client
  const counter = clientFor("counter-123")
  const total = yield* counter.Increment({ amount: 5 }) // => 5
  return total
})
```

### entity.of(handlers)

The identity function, used purely to pin the handler record to the entity's expected
`HandlersFrom` shape so TypeScript infers and checks each handler against its Rpc.

```ts
import { Effect, Ref } from "effect"

const handlers = Effect.gen(function*() {
  const count = yield* Ref.make(0)
  return Counter.of({
    Increment: ({ payload }) => Ref.updateAndGet(count, (n) => n + payload.amount),
    GetCount: () => Ref.get(count)
  })
})
// => the same record, now typed as HandlersFrom<Increment | GetCount>
```

### entity.getShardId(entityId)

Resolves an entity id to its `ShardId` (shard group + numeric shard). Needs the
`Sharding` service because the shard count comes from runtime config.

```ts
import { Effect } from "effect"

const program = Effect.gen(function*() {
  const shardId = yield* Counter.getShardId("counter-123")
  return shardId.toString() // => e.g. "default:7"
})
```

### entity.getShardGroup(entityId)

Pure synchronous lookup of the shard *group* name for an id (default `"default"`, or
whatever `ClusterSchema.ShardGroup` returns). No `Sharding` required.

```ts
Counter.getShardGroup("counter-123") // => "default"
```

### entity.annotate / annotateRpcs / annotateMerge / annotateRpcsMerge

Attach `ClusterSchema` references to change cluster behavior. `annotate` applies the
key/value to the **whole entity** (all Rpcs); `annotateRpcs` applies it only to the
**Rpcs declared above the call**. The `*Merge` variants take a whole
`Context.Context` instead of a single key/value pair. Each returns a new entity.

```ts
import { ClusterSchema, Entity } from "effect/unstable/cluster"

// Persist every message for the entity.
const Durable = Counter.annotate(ClusterSchema.Persisted, true)

// Persist only the Rpcs defined before this point.
const PartlyDurable = Counter.annotateRpcs(ClusterSchema.Persisted, true)

// Merge several annotations at once via a Context.
import { Context } from "effect"
const ctx = Context.empty().pipe(
  Context.add(ClusterSchema.Persisted, true)
)
const Merged = Counter.annotateMerge(ctx)
```

### Entity.isEntity(u)

Type guard that returns `true` when a value is a cluster `Entity`.

```ts
import { Entity } from "effect/unstable/cluster"

Entity.isEntity(Counter) // => true
Entity.isEntity({}) // => false
```

### Entity.makeTestClient(entity, layer)

Builds an in-memory test client for an entity layer: it registers `layer` against a
test `Sharding`, then returns a function `(entityId) => Effect<RpcClient>` that talks
to the handlers directly with no serialization or network. Use it under a
`TestRunner`-style config to unit-test an entity in isolation.

```ts
import { Effect } from "effect"
import { Entity, ShardingConfig } from "effect/unstable/cluster"

const test = Effect.gen(function*() {
  const makeClient = yield* Entity.makeTestClient(Counter, CounterLayer)
  const counter = yield* makeClient("counter-123")
  const total = yield* counter.Increment({ amount: 2 }) // => 2
  return total
}).pipe(
  Effect.scoped,
  Effect.provide(ShardingConfig.layer())
)
```
**Note:** `makeTestClient` requires `ShardingConfig` (and a `Scope`); it dies if the supplied
layer never registered the entity. It is independent of `TestRunner.layer` — use it
when you want to drive one entity's handlers, and `TestRunner.layer` when you want a
full in-process cluster around `entity.client`.

## Per-instance services

These are available inside handlers (or `build`) via `yield*`.

### Entity.CurrentAddress

Yields the `EntityAddress` for the message being processed: `entityType`, `entityId`,
and `shardId`.

```ts
import { Effect } from "effect"
import { Entity } from "effect/unstable/cluster"

const handler = Effect.gen(function*() {
  const { entityType, entityId, shardId } = yield* Entity.CurrentAddress
  yield* Effect.log(`${entityType}/${entityId} on ${shardId}`)
})
```

### Entity.CurrentRunnerAddress

Yields the `RunnerAddress` of the runner that owns the current entity registration —
useful for logging which node is hosting an instance.

```ts
import { Effect } from "effect"
import { Entity } from "effect/unstable/cluster"

const build = Effect.gen(function*() {
  const runner = yield* Entity.CurrentRunnerAddress
  yield* Effect.log(`hosted on ${runner.host}:${runner.port}`)
  return Counter.of({ /* ... */ } as never)
})
```

### Entity.Request\<Rpc>

The envelope class each handler receives. It extends the request envelope with the
typed `payload`, message `address`, `requestId`, and — for streaming Rpcs — a
`lastSentChunk` plus the `lastSentChunkValue` / `nextSequence` accessors used to resume
chunk sequencing after a restart.

```ts
import { Ref } from "effect"

const handler = Counter.of({
  Increment: (request) => {
    request.payload.amount // => number
    request.address.entityId // => the entity id
    return Ref.updateAndGet(/* ... */ as never, (n: number) => n + request.payload.amount)
  },
  GetCount: () => Ref.get(/* ... */ as never)
})
```

### Entity.Replier (toLayerQueue)

The reply API handed to a `toLayerQueue` processor. Complete a request with `succeed`,
`fail`, `failCause`, or `complete` (an explicit `Exit`).

```ts
import { Effect, Exit } from "effect"

const processor = (mailbox: never, replier: Entity.Replier<never>) =>
  Effect.gen(function*() {
    const request = yield* (mailbox as never) // a typed Envelope.Request
    yield* replier.succeed(request, 42) // resolve the caller's Effect with 42
    // or: replier.fail(request, error)
    // or: replier.complete(request, Exit.succeed(42))
  })
```

### Entity.keepAlive / Entity.KeepAliveLatch / Entity.KeepAliveRpc

`Entity.keepAlive(enabled)` keeps an otherwise idle entity from being passivated past
its `maxIdleTime`. Call `keepAlive(true)` while you hold a long-lived resource and
`keepAlive(false)` to release it. It works by closing/opening the `KeepAliveLatch`
service and sending the internal, persisted-and-uninterruptible `KeepAliveRpc`.

```ts
import { Effect } from "effect"
import { Entity } from "effect/unstable/cluster"

const withResource = Effect.gen(function*() {
  yield* Entity.keepAlive(true) // entity stays active past maxIdleTime
  // ... do long-lived work, hold a connection, etc.
  yield* Entity.keepAlive(false) // allow normal passivation again
})
// `keepAlive` requires Sharding | Entity.CurrentAddress
```
**Tip:** For acquiring a resource tied to an entity address and keeping the entity alive while
it is held, see `EntityResource.make` in `effect/unstable/cluster`, which wraps
`keepAlive` and an `RcRef` for you.

## Entity annotations (ClusterSchema)

`ClusterSchema` exposes `Context.Reference` annotations that change how the cluster
runtime handles an entity's messages — without touching the request or response
schemas. Attach them with `entity.annotate` (whole entity), `entity.annotateRpcs`
(Rpcs above the call), or directly on an `Rpc` with `.annotate`.

### ClusterSchema.Persisted

`boolean` (default `false`). Saves messages to durable mailbox storage before
handling, so they survive a crash and are retried. Requires a `MessageStorage` that
can persist mailboxes (the SQL store does).

```ts
const Durable = Counter.annotate(ClusterSchema.Persisted, true)
```

### ClusterSchema.Uninterruptible

`boolean | "client" | "server"` (default `false`). `true` makes both sides
uninterruptible; `"client"` and `"server"` restrict it to one side. Use it for
messages whose handling must not be cut short.

```ts
const Safe = Counter.annotate(ClusterSchema.Uninterruptible, "server")
```

### ClusterSchema.WithTransaction

`boolean` (default `false`). Wraps server-side handling and its storage writes in the
configured storage transaction — effective only when the `MessageStorage` implements
transactions.

```ts
const Transactional = Counter.annotate(ClusterSchema.WithTransaction, true)
```

### ClusterSchema.ShardGroup

`(entityId: EntityId) => string` (default `() => "default"`). Routes ids into named
shard groups; the function must be deterministic across every runner in the cluster.

```ts
const Grouped = Counter.annotate(
  ClusterSchema.ShardGroup,
  (entityId) => (entityId.startsWith("vip-") ? "priority" : "default")
)
```

### ClusterSchema.ClientTracingEnabled

`boolean` (default `true`). Disable to suppress client-side request tracing on
internal or high-volume protocols.

```ts
const Quiet = Counter.annotate(ClusterSchema.ClientTracingEnabled, false)
```

### ClusterSchema.Dynamic

`(annotations, request) => annotations` (default identity). Derives per-request,
**server-side** annotations from the decoded request value — for example, persisting
only some payloads. It does not affect the generated client.

```ts
import { Context } from "effect"

const Conditional = Counter.annotate(
  ClusterSchema.Dynamic,
  (annotations, request) =>
    request.tag === "Increment"
      ? Context.add(annotations, ClusterSchema.Persisted, true)
      : annotations
)
```

## Next steps

- For one cluster-wide background process rather than many addressable ids, use
  `Singleton` from `effect/unstable/cluster` — see the [Cluster overview](https://effect.plants.sh/cluster/).
- Durable mailbox storage (`ClusterSchema.Persisted`) is backed by [SQL](https://effect.plants.sh/sql/); see
  also the [Persistence](https://effect.plants.sh/persistence/) section.
- The entity protocol is ordinary [RPC](https://effect.plants.sh/rpc/); the same `Rpc.make` definitions can
  back an [HTTP API](https://effect.plants.sh/http-api/) too.