/** * Stores Effect Cluster messages and replies behind a pluggable backend. * * `MessageStorage` is the boundary between cluster runner logic and the storage * system that keeps mailbox state recoverable. It saves requests, control * envelopes, and replies; finds unprocessed messages for assigned shards; * tracks duplicate requests; and manages reply handlers waiting for responses. * This module also includes the encoded storage-driver contract and no-op or * in-memory implementations for local use and tests. * * @stability unstable * @since 4.0.0 */ import * as Arr from "../Array.ts"; import * as Context from "../Context.ts"; import * as Data from "../Data.ts"; import * as Effect from "../Effect.ts"; import * as Layer from "../Layer.ts"; import * as Option from "../Option.ts"; import type * as Rpc from "../rpc/Rpc.ts"; import { EntityNotAssignedToRunner, MalformedMessage, type PersistenceError } from "./ClusterError.ts"; import type { EntityAddress } from "./EntityAddress.ts"; import * as Envelope from "./Envelope.ts"; import * as Message from "./Message.ts"; import * as Reply from "./Reply.ts"; import * as ShardId from "./ShardId.ts"; import type { ShardingConfig } from "./ShardingConfig.ts"; import * as Snowflake from "./Snowflake.ts"; declare const MessageStorage_base: Context.ServiceClass(envelope: Message.OutgoingRequest) => Effect.Effect, PersistenceError | MalformedMessage>; /** * Save the provided message and its associated metadata. */ readonly saveEnvelope: (envelope: Message.OutgoingEnvelope) => Effect.Effect; /** * Save the provided `Reply` and its associated metadata. */ readonly saveReply: (reply: Reply.ReplyWithContext) => Effect.Effect; /** * Clear replies for the given request id. When options.expectedReplyId is * provided, clear only if it is still the latest reply at the storage boundary. * Without it, clear unconditionally. * Custom storage implementations must honor this condition to protect * workflows from stale concurrent resumes. */ readonly clearReplies: (requestId: Snowflake.Snowflake, options?: { readonly expectedReplyId?: Snowflake.Snowflake | undefined; }) => Effect.Effect; /** * Retrieves the replies for the specified requests. * * **Details** * * This returns: * * - Un-acknowledged chunk replies * - `WithExit` replies */ readonly repliesFor: (requests: Iterable>) => Effect.Effect>, PersistenceError | MalformedMessage>; /** * Retrieves the encoded replies for the specified request ids. */ readonly repliesForUnfiltered: (requestIds: Iterable) => Effect.Effect, PersistenceError | MalformedMessage>; /** * Retrieves the request id for the specified primary key. */ readonly requestIdForPrimaryKey: (options: { readonly address: EntityAddress; readonly tag: string; readonly id: string; }) => Effect.Effect, PersistenceError>; /** * For locally sent messages, register a handler to process the replies. * * **Details** * * The handler must be registered as soon as the effect starts, before it * first suspends. Callers fork it with `startImmediately` before saving or * notifying the request, and rely on it to see every reply saved after that * point. The effect completes when a `WithExit` reply arrives. */ readonly registerReplyHandler: (message: Message.OutgoingRequest | Message.IncomingRequest) => Effect.Effect; /** * Unregister the reply handler for the specified message. */ readonly unregisterReplyHandler: (requestId: Snowflake.Snowflake) => Effect.Effect; /** * Unregister the reply handlers for the specified ShardId. * * By default the waiters are resumed with `EntityNotAssignedToRunner`, so * they can re-route the wait. With `interrupt: true` the waiters are * interrupted with a cluster-abandon annotation instead, for when the runner * is shutting down and the reply can never be observed here. */ readonly unregisterShardReplyHandlers: (shardId: ShardId.ShardId, options?: { readonly interrupt?: boolean | undefined; }) => Effect.Effect; /** * Retrieves the unprocessed messages for the specified shards. * * **Details** * * A message is unprocessed when: * * - Requests that have no `WithExit` replies or no unacknowledged chunk replies * - The latest `AckChunk` envelope * - All `Interrupt` envelopes for unprocessed requests * * The `limit` option bounds the number of messages returned in a single * read, and the `addresses` option restricts the read to the provided * entity addresses. Only the returned messages are claimed; other messages * stay eligible for later reads. */ readonly unprocessedMessages: (shardIds: Iterable, options?: { readonly limit?: number | undefined; readonly addresses?: ReadonlyArray | undefined; } | undefined) => Effect.Effect>, PersistenceError>; /** * Retrieves the unprocessed messages by id. */ readonly unprocessedMessagesById: (messageIds: Iterable) => Effect.Effect>, PersistenceError>; /** * Reset the mailbox state for the provided shards. */ readonly resetShards: (shardIds: Iterable) => Effect.Effect; /** * Reset the mailbox state for the provided address. */ readonly resetAddress: (address: EntityAddress) => Effect.Effect; /** Release request claims without changing replies or processed state. */ readonly resetRequests: (requestIds: ReadonlyArray) => Effect.Effect; /** * Reset the mailbox state for the provided addresses. */ readonly resetAddresses: (addresses: ReadonlyArray) => Effect.Effect; /** * Clear all messages and replies for the provided address. */ readonly clearAddress: (address: EntityAddress) => Effect.Effect; /** * Wraps requests in storage transactions. * * `WithExit` replies saved in a transaction notify callers after the outermost * commit; stream chunks notify immediately. Replies saved in a failed * transaction must be discarded. The wrapper may fail only with the wrapped * effect's own failure or a defect. */ readonly withTransaction: (effect: Effect.Effect) => Effect.Effect; }>; /** * Service for cluster mailbox persistence and reply delivery. * * **Details** * * It stores outgoing requests, control envelopes, and replies; reads unprocessed * messages; manages reply handlers; and provides transaction wrapping for storage * operations. * * @stability unstable * @category services * @since 4.0.0 */ export declare class MessageStorage extends MessageStorage_base { } /** * Result of saving a request or envelope into message storage. * * **Details** * * A duplicate result carries the original request ID and the last reply already * received for the duplicated request. * * @stability unstable * @category models * @since 4.0.0 */ export type SaveResult = SaveResult.Success | SaveResult.Duplicate; /** * Constructors and matchers for decoded save results. * * @stability unstable * @category constructors * @since 4.0.0 */ export declare const SaveResult: { readonly $is: (tag: Tag) => { >(u: T): u is T & { readonly _tag: Tag; }; (u: unknown): u is Extract, { readonly _tag: Tag; }> | Extract; }; readonly $match: { , { readonly _tag: Tag; }>) => any; }>(cases: Cases): (self: SaveResult) => import("../Unify.ts").Unify>; , { readonly _tag: Tag; }>) => any; }>(self: SaveResult, cases: Cases): import("../Unify.ts").Unify>; }; readonly Duplicate: (args: { readonly originalId: Snowflake.Snowflake; readonly lastReceivedReply: Option.Option>; }) => SaveResult.Duplicate; readonly Success: (args: void) => SaveResult.Success; }; /** * Constructors and matchers for encoded save results returned by storage * drivers. * * @stability unstable * @category constructors * @since 4.0.0 */ export declare const SaveResultEncoded: { readonly $is: (tag: Tag) => (u: unknown) => u is Extract | Extract; readonly $match: { ) => any; }>(cases: Cases): (value: SaveResult.Encoded) => import("../Unify.ts").Unify>; ) => any; }>(value: SaveResult.Encoded, cases: Cases): import("../Unify.ts").Unify>; }; readonly Duplicate: Data.TaggedEnum.ConstructorFrom; readonly Success: Data.TaggedEnum.ConstructorFrom; }; /** * Variants and helper types for `SaveResult`. * * @stability unstable * @since 4.0.0 */ export declare namespace SaveResult { /** * Encoded storage-driver form of `SaveResult`. * * **Details** * * Duplicate results contain an encoded last received reply instead of a decoded * reply. * * @stability unstable * @category models * @since 4.0.0 */ type Encoded = SaveResult.Success | SaveResult.DuplicateEncoded; /** * Variant indicating that the message was saved as a new storage entry. * * @stability unstable * @category models * @since 4.0.0 */ interface Success { readonly _tag: "Success"; } /** * Variant indicating that the request duplicates an existing stored request. * * **Details** * * It carries the original request ID and the latest decoded reply, when one is * available. * * @stability unstable * @category models * @since 4.0.0 */ interface Duplicate { readonly _tag: "Duplicate"; readonly originalId: Snowflake.Snowflake; readonly lastReceivedReply: Option.Option>; } /** * Encoded duplicate-save variant returned by lower-level storage drivers. * * **Details** * * It carries the original request ID and the latest encoded reply, when one is * available. * * @stability unstable * @category models * @since 4.0.0 */ interface DuplicateEncoded { readonly _tag: "Duplicate"; readonly originalId: Snowflake.Snowflake; readonly lastReceivedReply: Option.Option; } /** * Generic tagged enum constructor type for `SaveResult`. * * @stability unstable * @category utility types * @since 4.0.0 */ interface Constructor extends Data.TaggedEnum.WithGenerics<1> { readonly taggedEnum: SaveResult; } } /** * Low-level storage-driver contract for encoded envelopes and replies. * * **Details** * * Implementations persist encoded messages, track primary keys and delayed * delivery, read unprocessed messages, and provide transaction wrapping. * * @stability unstable * @category services * @since 4.0.0 */ export type Encoded = { /** * Save the provided message and its associated metadata. */ readonly saveEnvelope: (options: { readonly envelope: Envelope.Encoded; readonly primaryKey: string | null; readonly deliverAt: number | null; }) => Effect.Effect; /** * Save the provided `Reply` and its associated metadata. */ readonly saveReply: (reply: Reply.Encoded) => Effect.Effect; /** * Remove the replies for the specified request. If options.expectedReplyId * is provided, compare it with the latest reply atomically before clearing. * Without it, clear unconditionally. * Custom storage implementations must honor this condition to protect * workflows from stale concurrent resumes. */ readonly clearReplies: (requestId: Snowflake.Snowflake, options?: { readonly expectedReplyId?: Snowflake.Snowflake | undefined; }) => Effect.Effect; /** * Retrieves the request id for the specified primary key. */ readonly requestIdForPrimaryKey: (primaryKey: string) => Effect.Effect, PersistenceError>; /** * Retrieves the replies for the specified requests. * * **Details** * * This returns: * * - Un-acknowledged chunk replies * - `WithExit` replies */ readonly repliesFor: (requestIds: Arr.NonEmptyArray) => Effect.Effect, PersistenceError>; /** * Retrieves the replies for the specified request ids. */ readonly repliesForUnfiltered: (requestIds: Arr.NonEmptyArray) => Effect.Effect, PersistenceError>; /** * Retrieves the unprocessed messages for the given shards. * * **Details** * * A message is unprocessed when: * * - Requests that have no `WithExit` replies or no unacknowledged chunk replies * - The latest `AckChunk` envelope * - All `Interrupt` envelopes for unprocessed requests * * The `limit` option bounds the number of rows returned by a single read, * and the `addresses` option restricts the read to the provided entity * addresses. Only the returned rows are claimed; other rows stay eligible * for later reads. */ readonly unprocessedMessages: (shardIds: Arr.NonEmptyArray, now: number, options?: { readonly limit?: number | undefined; readonly addresses?: ReadonlyArray | undefined; } | undefined) => Effect.Effect; }>, PersistenceError>; /** * Retrieves the unprocessed messages by id. */ readonly unprocessedMessagesById: (messageIds: Arr.NonEmptyArray, now: number) => Effect.Effect; }>, PersistenceError>; /** Release request claims without changing replies or processed state. */ readonly resetRequests: (requestIds: ReadonlyArray) => Effect.Effect; /** * Reset the mailbox state for the provided addresses. */ readonly resetAddresses: (addresses: ReadonlyArray) => Effect.Effect; /** * Clear all messages and replies for the provided address. */ readonly clearAddress: (address: EntityAddress) => Effect.Effect; /** * Reset the mailbox state for the provided shards. */ readonly resetShards: (shardIds: Arr.NonEmptyArray) => Effect.Effect; /** * Used to wrap requests with transactions. */ readonly withTransaction: (effect: Effect.Effect) => Effect.Effect; }; /** * Cursor options for reading encoded unprocessed messages across shard sets. * * **Details** * * The fields distinguish existing shards from newly assigned shards and carry the * driver-specific pagination cursor. * * @stability unstable * @category models * @since 4.0.0 */ export type EncodedUnprocessedOptions = { readonly existingShards: Array; readonly newShards: Array; readonly cursor: Option.Option; }; /** * Cursor options for reading encoded replies across request sets. * * **Details** * * The fields distinguish existing requests from new requests and carry the * driver-specific pagination cursor. * * @stability unstable * @category models * @since 4.0.0 */ export type EncodedRepliesOptions = { readonly existingRequests: Array; readonly newRequests: Array; readonly cursor: Option.Option; }; /** * Wraps a concrete message storage implementation with reply-handler management. * * **Details** * * The returned service can register waiting reply handlers, notify them when * replies are saved, and fail them when a request or shard is unregistered. * * @stability unstable * @category constructors * @since 4.0.0 */ export declare const make: (storage: Omit & { /** * Save the provided `Reply`, returning the reply as it was persisted. * * Implementations that persist a different representation, such as a * defect reply when the original cannot be encoded, return that fallback * so reply handlers deliver exactly what storage recorded. */ readonly saveReply: (reply: Reply.ReplyWithContext) => Effect.Effect, PersistenceError | MalformedMessage>; }) => Effect.Effect; /** * Builds a `MessageStorage` service from an encoded storage driver. * * **Details** * * The adapter handles envelope and reply encoding and decoding, primary-key * generation, delayed delivery checks, duplicate decoding, and malformed-message * defect replies. * * @stability unstable * @category constructors * @since 4.0.0 */ export declare const makeEncoded: (encoded: Encoded) => Effect.Effect; /** * No-op `MessageStorage` service that does not persist messages or replies. * * @stability unstable * @category constructors * @since 4.0.0 */ export declare const noop: MessageStorage["Service"]; /** * In-memory storage entry for a request envelope. * * **Details** * * It stores the encoded envelope, last acknowledged chunk, accumulated replies, * and optional delivery time. * * @stability unstable * @category models * @since 4.0.0 */ export type MemoryEntry = { readonly envelope: Envelope.Encoded; lastReceivedChunk: Reply.ChunkEncoded | undefined; replies: Array; deliverAt: number | null; }; /** * Provides a context reference used in tests to simulate a transaction. * * @stability unstable * @category services * @since 4.0.0 */ export declare const MemoryTransaction: Context.Reference; declare const MemoryDriver_base: Context.ServiceClass(envelope: Message.OutgoingRequest) => Effect.Effect, PersistenceError | MalformedMessage>; /** * Save the provided message and its associated metadata. */ readonly saveEnvelope: (envelope: Message.OutgoingEnvelope) => Effect.Effect; /** * Save the provided `Reply` and its associated metadata. */ readonly saveReply: (reply: Reply.ReplyWithContext) => Effect.Effect; /** * Clear replies for the given request id. When options.expectedReplyId is * provided, clear only if it is still the latest reply at the storage boundary. * Without it, clear unconditionally. * Custom storage implementations must honor this condition to protect * workflows from stale concurrent resumes. */ readonly clearReplies: (requestId: Snowflake.Snowflake, options?: { readonly expectedReplyId?: Snowflake.Snowflake | undefined; }) => Effect.Effect; /** * Retrieves the replies for the specified requests. * * **Details** * * This returns: * * - Un-acknowledged chunk replies * - `WithExit` replies */ readonly repliesFor: (requests: Iterable>) => Effect.Effect>, PersistenceError | MalformedMessage>; /** * Retrieves the encoded replies for the specified request ids. */ readonly repliesForUnfiltered: (requestIds: Iterable) => Effect.Effect, PersistenceError | MalformedMessage>; /** * Retrieves the request id for the specified primary key. */ readonly requestIdForPrimaryKey: (options: { readonly address: EntityAddress; readonly tag: string; readonly id: string; }) => Effect.Effect, PersistenceError>; /** * For locally sent messages, register a handler to process the replies. * * **Details** * * The handler must be registered as soon as the effect starts, before it * first suspends. Callers fork it with `startImmediately` before saving or * notifying the request, and rely on it to see every reply saved after that * point. The effect completes when a `WithExit` reply arrives. */ readonly registerReplyHandler: (message: Message.OutgoingRequest | Message.IncomingRequest) => Effect.Effect; /** * Unregister the reply handler for the specified message. */ readonly unregisterReplyHandler: (requestId: Snowflake.Snowflake) => Effect.Effect; /** * Unregister the reply handlers for the specified ShardId. * * By default the waiters are resumed with `EntityNotAssignedToRunner`, so * they can re-route the wait. With `interrupt: true` the waiters are * interrupted with a cluster-abandon annotation instead, for when the runner * is shutting down and the reply can never be observed here. */ readonly unregisterShardReplyHandlers: (shardId: ShardId.ShardId, options?: { readonly interrupt?: boolean | undefined; }) => Effect.Effect; /** * Retrieves the unprocessed messages for the specified shards. * * **Details** * * A message is unprocessed when: * * - Requests that have no `WithExit` replies or no unacknowledged chunk replies * - The latest `AckChunk` envelope * - All `Interrupt` envelopes for unprocessed requests * * The `limit` option bounds the number of messages returned in a single * read, and the `addresses` option restricts the read to the provided * entity addresses. Only the returned messages are claimed; other messages * stay eligible for later reads. */ readonly unprocessedMessages: (shardIds: Iterable, options?: { readonly limit?: number | undefined; readonly addresses?: ReadonlyArray | undefined; } | undefined) => Effect.Effect>, PersistenceError>; /** * Retrieves the unprocessed messages by id. */ readonly unprocessedMessagesById: (messageIds: Iterable) => Effect.Effect>, PersistenceError>; /** * Reset the mailbox state for the provided shards. */ readonly resetShards: (shardIds: Iterable) => Effect.Effect; /** * Reset the mailbox state for the provided address. */ readonly resetAddress: (address: EntityAddress) => Effect.Effect; /** Release request claims without changing replies or processed state. */ readonly resetRequests: (requestIds: ReadonlyArray) => Effect.Effect; /** * Reset the mailbox state for the provided addresses. */ readonly resetAddresses: (addresses: ReadonlyArray) => Effect.Effect; /** * Clear all messages and replies for the provided address. */ readonly clearAddress: (address: EntityAddress) => Effect.Effect; /** * Wraps requests in storage transactions. * * `WithExit` replies saved in a transaction notify callers after the outermost * commit; stream chunks notify immediately. Replies saved in a failed * transaction must be discarded. The wrapper may fail only with the wrapped * effect's own failure or a defect. */ readonly withTransaction: (effect: Effect.Effect) => Effect.Effect; }; readonly encoded: Encoded; readonly requests: Map; readonly requestsByPrimaryKey: Map; readonly unprocessed: Set; readonly replyIds: Set; readonly journal: Envelope.Encoded[]; readonly cursors: WeakMap<{}, number>; }> & { readonly make: Effect.Effect<{ readonly storage: { /** * Save the provided message and its associated metadata. */ readonly saveRequest: (envelope: Message.OutgoingRequest) => Effect.Effect, PersistenceError | MalformedMessage>; /** * Save the provided message and its associated metadata. */ readonly saveEnvelope: (envelope: Message.OutgoingEnvelope) => Effect.Effect; /** * Save the provided `Reply` and its associated metadata. */ readonly saveReply: (reply: Reply.ReplyWithContext) => Effect.Effect; /** * Clear replies for the given request id. When options.expectedReplyId is * provided, clear only if it is still the latest reply at the storage boundary. * Without it, clear unconditionally. * Custom storage implementations must honor this condition to protect * workflows from stale concurrent resumes. */ readonly clearReplies: (requestId: Snowflake.Snowflake, options?: { readonly expectedReplyId?: Snowflake.Snowflake | undefined; }) => Effect.Effect; /** * Retrieves the replies for the specified requests. * * **Details** * * This returns: * * - Un-acknowledged chunk replies * - `WithExit` replies */ readonly repliesFor: (requests: Iterable>) => Effect.Effect>, PersistenceError | MalformedMessage>; /** * Retrieves the encoded replies for the specified request ids. */ readonly repliesForUnfiltered: (requestIds: Iterable) => Effect.Effect, PersistenceError | MalformedMessage>; /** * Retrieves the request id for the specified primary key. */ readonly requestIdForPrimaryKey: (options: { readonly address: EntityAddress; readonly tag: string; readonly id: string; }) => Effect.Effect, PersistenceError>; /** * For locally sent messages, register a handler to process the replies. * * **Details** * * The handler must be registered as soon as the effect starts, before it * first suspends. Callers fork it with `startImmediately` before saving or * notifying the request, and rely on it to see every reply saved after that * point. The effect completes when a `WithExit` reply arrives. */ readonly registerReplyHandler: (message: Message.OutgoingRequest | Message.IncomingRequest) => Effect.Effect; /** * Unregister the reply handler for the specified message. */ readonly unregisterReplyHandler: (requestId: Snowflake.Snowflake) => Effect.Effect; /** * Unregister the reply handlers for the specified ShardId. * * By default the waiters are resumed with `EntityNotAssignedToRunner`, so * they can re-route the wait. With `interrupt: true` the waiters are * interrupted with a cluster-abandon annotation instead, for when the runner * is shutting down and the reply can never be observed here. */ readonly unregisterShardReplyHandlers: (shardId: ShardId.ShardId, options?: { readonly interrupt?: boolean | undefined; }) => Effect.Effect; /** * Retrieves the unprocessed messages for the specified shards. * * **Details** * * A message is unprocessed when: * * - Requests that have no `WithExit` replies or no unacknowledged chunk replies * - The latest `AckChunk` envelope * - All `Interrupt` envelopes for unprocessed requests * * The `limit` option bounds the number of messages returned in a single * read, and the `addresses` option restricts the read to the provided * entity addresses. Only the returned messages are claimed; other messages * stay eligible for later reads. */ readonly unprocessedMessages: (shardIds: Iterable, options?: { readonly limit?: number | undefined; readonly addresses?: ReadonlyArray | undefined; } | undefined) => Effect.Effect>, PersistenceError>; /** * Retrieves the unprocessed messages by id. */ readonly unprocessedMessagesById: (messageIds: Iterable) => Effect.Effect>, PersistenceError>; /** * Reset the mailbox state for the provided shards. */ readonly resetShards: (shardIds: Iterable) => Effect.Effect; /** * Reset the mailbox state for the provided address. */ readonly resetAddress: (address: EntityAddress) => Effect.Effect; /** Release request claims without changing replies or processed state. */ readonly resetRequests: (requestIds: ReadonlyArray) => Effect.Effect; /** * Reset the mailbox state for the provided addresses. */ readonly resetAddresses: (addresses: ReadonlyArray) => Effect.Effect; /** * Clear all messages and replies for the provided address. */ readonly clearAddress: (address: EntityAddress) => Effect.Effect; /** * Wraps requests in storage transactions. * * `WithExit` replies saved in a transaction notify callers after the outermost * commit; stream chunks notify immediately. Replies saved in a failed * transaction must be discarded. The wrapper may fail only with the wrapped * effect's own failure or a defect. */ readonly withTransaction: (effect: Effect.Effect) => Effect.Effect; }; readonly encoded: Encoded; readonly requests: Map; readonly requestsByPrimaryKey: Map; readonly unprocessed: Set; readonly replyIds: Set; readonly journal: Envelope.Encoded[]; readonly cursors: WeakMap<{}, number>; }, never, Snowflake.Generator>; }; /** * Service that provides an in-memory message storage driver with inspectable backing state. * * **Details** * * It provides a `MessageStorage` service, the encoded driver implementation, and * maps used to track requests, primary keys, unprocessed envelopes, reply IDs, * and the journal. * * @stability unstable * @category services * @since 4.0.0 */ export declare class MemoryDriver extends MemoryDriver_base { /** * Layer that provides the in-memory message storage driver. * * @stability unstable * @since 4.0.0 */ static readonly layer: Layer.Layer; } /** * Layer that provides the no-op `MessageStorage` service. * * @stability unstable * @category layers * @since 4.0.0 */ export declare const layerNoop: Layer.Layer; /** * Layer that provides in-memory message storage and its backing `MemoryDriver`. * * @stability unstable * @category layers * @since 4.0.0 */ export declare const layerMemory: Layer.Layer; export {}; //# sourceMappingURL=MessageStorage.d.ts.map