Skip to content

Entity

Defines addressable entity types for Effect Cluster.

An entity gives a stable name and RPC protocol to a group of values that are addressed by id. The cluster uses that information to choose a shard and route each request to the runner responsible for that id. This module includes constructors for entity definitions, helpers for creating sharded clients, layer builders for registering handlers, and services that expose the current entity address while a request is being handled.

15 exports Added in v4.0.0 Source

Constructors

fromRpcGroup

Added in v4.0.0 Source

Creates a new Entity of the specified type which will accept messages that adhere to the provided RpcGroup.

Signature

declare function fromRpcGroup<Type extends string, Rpcs extends Any>(
  type: Type,
  protocol: RpcGroup<Rpcs>,
): Entity<Type, Rpcs>;

make

Added in v4.0.0 Source

Creates a new Entity of the specified type which will accept messages that adhere to the provided schemas.

When to use

Use to define a cluster entity from individual Rpc definitions, giving the cluster runtime a typed protocol for handlers and per-entity clients.

Details

The type argument is stored as the entity EntityType, and the RPC array is grouped into the entity's protocol.

Gotchas

RPC tags should be unique within the array. If multiple definitions use the same tag, the resulting protocol keeps the later definition for that tag.

See

  • fromRpcGroup for creating an entity from an existing RpcGroup

Signature

declare function make<Type extends string, Rpcs extends readonly Array<Any>>(type: Type, protocol: Rpcs): Entity<Type, Rpcs[number]>

Guards

isEntity

Added in v4.0.0 Source

Returns true when the supplied value is a cluster Entity.

Details

The check is based on the internal entity type identifier.

Signature

declare function isEntity(u: unknown): u is Any;

Keep Alive

keepAlive

Added in v4.0.0 Source

Enables or disables keep-alive for the current entity.

Details

When enabled it sends the internal keep-alive RPC for the current address; when disabled it releases the keep-alive latch if one is present.

Signature

declare const keepAlive: (
  enabled: boolean,
) => Effect.Effect<void, never, Sharding | CurrentAddress>;

KeepAliveRpc

Added in v4.0.0 Source

RPC used internally to keep an entity active while a resource is held.

Details

The RPC is marked as persisted and uninterruptible so the keep-alive signal survives normal entity restarts.

Signature

declare const KeepAliveRpc: Rpc<"Cluster/Entity/keepAlive", Void, Void, Never, never, never>;

Models

Any type

Added in v4.0.0 Source

Type alias for any cluster Entity, regardless of entity type or RPC protocol.

Signature

type Any = Entity<string, Rpc.Any>;

Entity interface

Added in v4.0.0 Source

Represents a cluster entity type and the RPC protocol it can handle.

Details

An entity defines how ids map to shard groups, exposes a sharded client, and can be registered as a layer using RPC handlers or a mailbox queue.

Signature

interface Entity<in out Type extends string, in out Rpcs extends Rpc.Any> extends Equal {
  readonly "~effect/cluster/Entity": "~effect/cluster/Entity";
  readonly client: Effect<
    (
      entityId: string,
    ) => RpcClient.RpcClient.From<Rpcs, MailboxFull | AlreadyProcessingMessage | PersistenceError>,
    never,
    Sharding
  >;
  readonly protocol: RpcGroup<Rpcs>;
  readonly type: string & Brand<"~effect/cluster/EntityType">;
  annotate<I, S>(key: Key<I, S>, value: S): Entity<Type, Rpcs>;
  annotateMerge<S>(annotation: Context<S>): Entity<Type, Rpcs>;
  annotateRpcs<I, S>(key: Key<I, S>, value: S): Entity<Type, Rpcs>;
  annotateRpcsMerge<S>(context: Context<S>): Entity<Type, Rpcs>;
  getShardGroup(entityId: string & Brand<"~effect/cluster/EntityId">): string;
  getShardId(
    entityId: string & Brand<"~effect/cluster/EntityId">,
  ): Effect<ShardId, never, Sharding>;
  of<Handlers extends HandlersFrom<Rpcs>>(handlers: Handlers): Handlers;
  toLayer<Handlers extends HandlersFrom<Rpcs>, RX = never>(
    build: Handlers | Effect<Handlers, never, RX>,
    options?: {
      readonly concurrency?: number | "unbounded";
      readonly defectRetryPolicy?: Schedule<any, unknown, never, never>;
      readonly disableFatalDefects?: boolean;
      readonly mailboxCapacity?: number | "unbounded";
      readonly maxIdleTime?: Input;
      readonly spanAttributes?: Record<string, string>;
    },
  ): Layer<
    never,
    never,
    | Sharding
    | Exclude<RX, Scope | CurrentAddress | CurrentRunnerAddress>
    | HandlersServices<Rpcs, Handlers>
    | ServicesClient<Rpcs>
    | ServicesServer<Rpcs>
    | Middleware<Rpcs>
  >;
  toLayerQueue<R, RX = never>(
    build: (
      queue: Dequeue<Request<Rpcs>>,
      replier: Replier<Rpcs>,
    ) =>
      | Effect<never, never, R>
      | Effect<
          (queue: Dequeue<Request<Rpcs>>, replier: Replier<Rpcs>) => Effect<never, never, R>,
          never,
          RX
        >,
    options?: {
      readonly defectRetryPolicy?: Schedule<any, unknown, never, never>;
      readonly disableFatalDefects?: boolean;
      readonly mailboxCapacity?: number | "unbounded";
      readonly maxIdleTime?: Input;
      readonly spanAttributes?: Record<string, string>;
    },
  ): Layer<
    never,
    never,
    | Sharding
    | R
    | ServicesClient<Rpcs>
    | ServicesServer<Rpcs>
    | Middleware<Rpcs>
    | Exclude<RX, Scope | CurrentAddress | CurrentRunnerAddress>
  >;
}

Replier interface

Added in v4.0.0 Source

Reply API passed to queue-based entity handlers.

When to use

Use when you use it to complete an entity request by succeeding, failing, failing with a cause, or supplying an explicit Exit.

Signature

interface Replier<Rpcs extends Rpc.Any> {
  readonly complete: <R extends Any>(
    request: Request<R>,
    exit: Exit<Success<R>, Type<ErrorSchema<R>>>,
  ) => Effect<void>;
  readonly fail: <R extends Any>(request: Request<R>, error: Type<ErrorSchema<R>>) => Effect<void>;
  readonly failCause: <R extends Any>(
    request: Request<R>,
    cause: Cause<Type<ErrorSchema<R>>>,
  ) => Effect<void>;
  readonly succeed: <R extends Any>(request: Request<R>, value: Success<R>) => Effect<void>;
}

Request

Added in v4.0.0 Source

Represents an entity request envelope delivered to entity handlers.

Details

It includes the underlying request envelope plus the last stream reply chunk that was sent, allowing handlers to resume chunk sequencing after a restart.

Signature

declare class Request<Rpc extends Rpc.Any> extends Class<Envelope.Request<Rpc> & {
  readonly lastSentChunk: Option.Option<Reply.Chunk<Rpc>>;
}> {
  constructor<Rpc extends Any>(args: {
    readonly _tag: "Request";
    readonly "~effect/cluster/Envelope": "~effect/cluster/Envelope";
    readonly address: EntityAddress;
    readonly headers: Headers;
    readonly lastSentChunk: Option<Chunk<Rpc>>;
    readonly payload: Payload<Rpc>;
    readonly requestId: Snowflake;
    readonly sampled?: boolean;
    readonly spanId?: string;
    readonly tag: Tag<Rpc>;
    readonly traceId?: string;
  });
  lastSentChunkValue: Option<SuccessChunk<Rpc>>;
  nextSequence: number;
}

Other

Replier

Added in v4.0.0 Source

Helper types used by the Replier API.

Services

CurrentAddress

Added in v4.0.0 Source

Service tag for the entity address currently being processed.

When to use

Use to read the current entity identity and shard address from entity handlers and keep-alive logic.

Signature

declare class CurrentAddress extends Shape<
  "effect/cluster/Entity/EntityAddress",
  EntityAddress,
  this
> {
  constructor(_: never);
}

CurrentRunnerAddress

Added in v4.0.0 Source

Service tag for the runner address currently registering entity handlers.

When to use

Use to read the runner address associated with the current entity handler registration.

Signature

declare class CurrentRunnerAddress extends Shape<
  "effect/cluster/Entity/RunnerAddress",
  RunnerAddress,
  this
> {
  constructor(_: never);
}

KeepAliveLatch

Added in v4.0.0 Source

Service tag for the latch that coordinates entity keep-alive state.

Details

keepAlive closes the latch when keep-alive is active and opens it again when the resource no longer needs to keep the entity alive.

Signature

declare class KeepAliveLatch extends Shape<"effect/cluster/Entity/KeepAliveLatch", Latch, this> {
  constructor(_: never);
}

Testing

makeTestClient

Added in v4.0.0 Source

Builds an in-memory test client for an entity layer.

Details

The returned function creates a no-serialization RPC client for each entity ID, using a test sharding service instead of the cluster transport.

Signature

declare const makeTestClient: <Type extends string, Rpcs extends Rpc.Any, LA, LE, LR>(
  entity: Entity<Type, Rpcs>,
  layer: Layer.Layer<LA, LE, LR>,
) => Effect.Effect<
  (entityId: string) => Effect.Effect<RpcClient.RpcClient<Rpcs>>,
  LE,
  Scope | ShardingConfig | Exclude<LR, Sharding> | Rpc.MiddlewareClient<Rpcs>
>;

Utility Types

HandlersFrom type

Added in v4.0.0 Source

Maps each RPC in an entity protocol to the handler function expected by Entity.toLayer.

Details

Each handler receives the entity request envelope for that RPC and returns the RPC result or a supported RPC wrapper.

Signature

type HandlersFrom<Rpc extends Rpc.Any> = {
  [Current in Rpc]: (envelope: Request<Current>) => Rpc.WrapperOr<Rpc.ResultFrom<Current, any>>;
};