Sharding

Runs shard ownership and message routing for Effect Cluster.

Sharding decides which shard owns an entity id, tracks which shards belong to the local runner, and sends cluster messages to local handlers or remote runners. It also registers entities and singletons, creates clients for entity requests, polls stored messages, and tracks shutdown state. The main layer connects these responsibilities to runner communication, storage, health checks, configuration, and local resources.

2 exports Added in v4.0.0 Source

Layers

layer

Added in v4.0.0 Source

Layer that constructs the Sharding service from sharding configuration, runner communication, message storage, runner storage, runner health, the snowflake generator, and the entity reaper.

When to use

Use when you need to assemble a cluster sharding runtime from explicit sharding configuration, runner communication, message storage, runner storage, and runner health layers.

Details

The layer provides the Sharding service and installs its own snowflake generator and entity reaper. Callers still provide ShardingConfig, Runners, MessageStorage, RunnerStorage, and RunnerHealth.

Gotchas

Persisted messages require a non-no-op MessageStorage; if this layer is provided with MessageStorage.layerNoop, persisted sends defect.

See

Sharding for the service provided by this layer

Signature

declare const layer: Layer.Layer<
  Sharding,
  never,
  | ShardingConfig
  | Runners
  | MessageStorage.MessageStorage
  | RunnerStorage
  | RunnerHealth.RunnerHealth
>;

Services

Sharding

Added in v4.0.0 Source

Service that registers entities and singletons, routes messages to owned shards, generates runner-local snowflake ids, and polls storage for persisted work.

When to use

Use to access or provide cluster routing, shard ownership, entity registration, singleton registration, and persisted-work polling.

Signature

declare class Sharding extends Shape<
  "effect/cluster/Sharding",
  {
    readonly activeEntityCount: Effect<number>;
    readonly getRegistrationEvents: Stream<ShardingRegistrationEvent>;
    readonly getShardId: (
      entityId: string & Brand<"~effect/cluster/EntityId">,
      group: string,
    ) => ShardId;
    readonly getSnowflake: Effect<Snowflake>;
    readonly hasShardId: (shardId: ShardId) => boolean;
    readonly isShutdown: Effect<boolean>;
    readonly makeClient: <Type extends string, Rpcs extends Any>(
      entity: Entity<Type, Rpcs>,
    ) => Effect<
      (
        entityId: string,
      ) => RpcClient.RpcClient.From<Rpcs, MailboxFull | AlreadyProcessingMessage | PersistenceError>
    >;
    readonly notify: (
      message: Incoming<any>,
      options?: {
        readonly waitUntilRead?: boolean;
      },
    ) => Effect<void, EntityNotAssignedToRunner | AlreadyProcessingMessage>;
    readonly pollStorage: Effect<void>;
    readonly registerEntity: <
      Type extends string,
      Rpcs extends Any,
      Handlers extends HandlersFrom<Rpcs>,
      RX,
    >(
      entity: Entity<Type, Rpcs>,
      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>;
      },
    ) => Effect<
      void,
      never,
      | Scope
      | ServicesServer<Rpcs>
      | Middleware<Rpcs>
      | Exclude<RX, Scope | CurrentAddress | CurrentRunnerAddress>
    >;
    readonly registerSingleton: <E, R>(
      name: string,
      run: Effect<void, E, R>,
      options?: {
        readonly shardGroup?: string;
      },
    ) => Effect<void, never, Scope | R>;
    readonly reset: (requestId: Snowflake) => Effect<boolean>;
    readonly send: (
      message: Incoming<any>,
    ) => Effect<void, EntityNotAssignedToRunner | MailboxFull | AlreadyProcessingMessage>;
    readonly sendOutgoing: (
      message: Outgoing<any>,
      discard: boolean,
    ) => Effect<void, PersistenceError | MailboxFull | AlreadyProcessingMessage>;
  },
  this
> {
  constructor(_: never);
}