Entities

An entity is a sharded, addressable actor. You give it a stable type name and an 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

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 for simple values, or an STM TxRef 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.

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

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.

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

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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

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.

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

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.

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.

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.

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.

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.

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.
• Durable mailbox storage (ClusterSchema.Persisted) is backed by SQL; see also the Persistence section.
• The entity protocol is ordinary RPC; the same Rpc.make definitions can back an HTTP API too.