MessageRetention unstable

Deletes a persisted cluster entity's completed requests from message storage once they are older than a retention.

Persisted entities keep every request and its replies in MessageStorage to answer duplicates and resume interrupted work, so without a retention their storage grows without bound.

1 exports Since v4.1.3 Source

Constructors

make

unstable

Since v4.1.3

Creates a layer that registers a singleton deleting the entity's completed requests once they are older than retention.

When to use

Use when a long-lived persisted entity would otherwise keep one stored request and reply for every message it has ever handled.

Details

Every interval (default 1 minute), the singleton deletes in batches each request of the entity type whose final reply is older than retention, with its acknowledgements, interrupts and replies. Running requests and requests reset with MessageStorage.clearReplies are kept. Failures are logged and retried on the next interval.

The singleton runs in shardGroup (default "default"), not the entity's shard group.

Gotchas

A request sent again with the same primary key after deletion runs the handler again.
Choose a retention longer than any client retry window or reader of the stored replies.
Do not use it for workflow or ClusterCron entities, which rely on stored replies to know what already ran.
Include the layer on every runner that can be assigned shardGroup, so it keeps running when shards move.
SQL storage on MySQL needs MySQL 8.0.1+ or MariaDB 10.6+, and binlog_format ROW or MIXED. Otherwise every cleanup fails and is logged.

Signature

declare function make<Type extends string, Rpcs extends Any>(options: {
  readonly entity: Entity<Type, Rpcs>;
  readonly interval?: Input;
  readonly retention: Input;
  readonly shardGroup?: string;
}): Layer<never, never, Sharding | MessageStorage>

Example Forgetting a week-old request

import { Effect, Layer, Schema } from "effect"
import { ClusterSchema, Entity, MessageRetention } from "effect/cluster"
import { Rpc } from "effect/rpc"

const Counter = Entity.make("Counter", [
  Rpc.make("Hit", {
    payload: { id: Schema.String },
    primaryKey: ({ id }) => id
  })
]).annotateRpcs(ClusterSchema.Persisted, true)

export const CounterLive = Layer.mergeAll(
  Counter.toLayer({ Hit: () => Effect.void }),
  MessageRetention.make({ entity: Counter, retention: "7 days" })
)