From eadff61b760012150c1eef1bae4e022b441f2bea Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Wed, 12 Aug 2026 22:13:13 +0000 Subject: [PATCH 1/9] feat(cluster): bound runner entity residency and storage reads Adds admission control to cluster runners: - ShardingConfig.maxResidentEntities (default 10_000) caps how many entities can be resident on a runner at once. The spawn gate lives in front of the entity ResourceMap, so the storage read loop and volatile sends share one counter. At the cap, volatile sends to new addresses fail with MailboxFull, persisted sends still succeed, and messages for new addresses stay in storage until a slot frees up. - ShardingConfig.unprocessedMessageBatchSize (default 1024) bounds each storage read. The read loop delivers to already-resident addresses first, then admits new addresses while slots remain, reopening the read latch after a full batch and waking on entity removal at the cap. - MessageStorage.unprocessedMessages accepts { limit, addresses } and claims only the rows it returns (SQL, memory and noop drivers). Co-authored-by: Claude Fable 5 --- .changeset/wild-donuts-brake.md | 20 ++ .../src/unstable/cluster/ClusterError.ts | 4 +- .../src/unstable/cluster/MessageStorage.ts | 67 ++++++- .../effect/src/unstable/cluster/Sharding.ts | 141 ++++++++++++- .../src/unstable/cluster/ShardingConfig.ts | 35 ++++ .../src/unstable/cluster/SqlMessageStorage.ts | 66 +++++- .../cluster/internal/entityManager.ts | 30 ++- .../unstable/cluster/internal/entityReaper.ts | 6 +- .../unstable/cluster/internal/resourceMap.ts | 12 ++ .../test/cluster/MessageStorage.test.ts | 39 +++- packages/effect/test/cluster/Sharding.test.ts | 189 ++++++++++++++++++ .../test/cluster/ShardingConfig.test.ts | 27 +++ .../cluster-integration/Residency.test.ts | 76 +++++++ .../node/test/cluster/MessageStorageTest.ts | 3 +- .../SqlMessageStorage.integration.test.ts | 48 +++++ 15 files changed, 736 insertions(+), 27 deletions(-) create mode 100644 .changeset/wild-donuts-brake.md create mode 100644 packages/platform/node/test/cluster-integration/Residency.test.ts diff --git a/.changeset/wild-donuts-brake.md b/.changeset/wild-donuts-brake.md new file mode 100644 index 00000000000..b341ea8b2b5 --- /dev/null +++ b/.changeset/wild-donuts-brake.md @@ -0,0 +1,20 @@ +--- +"effect": patch +--- + +Bound cluster runner entity residency and storage reads. + +`ShardingConfig` gains two knobs: + +- `maxResidentEntities` (default `10_000`): the maximum number of entities + that can be resident on a runner at the same time. At the cap, the storage + read loop stops admitting messages for new entity addresses (they stay in + storage until a slot frees up) and volatile sends to new addresses fail with + `MailboxFull`. Persisted sends still succeed. `"unbounded"` restores the + previous behaviour and can only be set programmatically. +- `unprocessedMessageBatchSize` (default `1024`): the maximum number of + unprocessed messages read from storage in a single poll. + +`MessageStorage.unprocessedMessages` accepts an optional +`{ limit, addresses }` argument, and only claims the messages it actually +returns. diff --git a/packages/effect/src/unstable/cluster/ClusterError.ts b/packages/effect/src/unstable/cluster/ClusterError.ts index e8faae7fea3..6e74365ee7b 100644 --- a/packages/effect/src/unstable/cluster/ClusterError.ts +++ b/packages/effect/src/unstable/cluster/ClusterError.ts @@ -171,7 +171,9 @@ export class RunnerUnavailable extends Schema.Error(`${TypeId * * **Details** * - * Carries the `address` whose bounded mailbox is at capacity. + * Carries the `address` whose bounded mailbox is at capacity. It is also used + * when the runner has reached `ShardingConfig.maxResidentEntities` and cannot + * spawn a new entity for the `address`. * * **Gotchas** * diff --git a/packages/effect/src/unstable/cluster/MessageStorage.ts b/packages/effect/src/unstable/cluster/MessageStorage.ts index 8d33454a50d..ceebdf03f14 100644 --- a/packages/effect/src/unstable/cluster/MessageStorage.ts +++ b/packages/effect/src/unstable/cluster/MessageStorage.ts @@ -131,9 +131,15 @@ export class MessageStorage extends Context.Service + shardIds: Iterable, + options?: UnprocessedOptions | undefined ) => Effect.Effect>, PersistenceError> /** @@ -172,6 +178,23 @@ export class MessageStorage extends Context.Service Effect.Effect }>()("effect/cluster/MessageStorage") {} +/** + * Options for reading unprocessed messages. + * + * **Details** + * + * `limit` bounds the number of messages returned by a single read, and + * `addresses` restricts the read to the provided entity addresses. Storage + * implementations only claim the messages they actually return. + * + * @category models + * @since 4.0.0 + */ +export type UnprocessedOptions = { + readonly limit?: number | undefined + readonly addresses?: Iterable | undefined +} + /** * Result of saving a request or envelope into message storage. * @@ -349,10 +372,16 @@ export type Encoded = { * - 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 + now: number, + options?: EncodedUnprocessedMessagesOptions | undefined ) => Effect.Effect< Array<{ readonly envelope: Envelope.Encoded @@ -404,6 +433,17 @@ export type Encoded = { ) => Effect.Effect } +/** + * Options for reading encoded unprocessed messages from a storage driver. + * + * @category models + * @since 4.0.0 + */ +export type EncodedUnprocessedMessagesOptions = { + readonly limit?: number | undefined + readonly addresses?: ReadonlyArray | undefined +} + /** * Cursor options for reading encoded unprocessed messages across shard sets. * @@ -629,12 +669,18 @@ export const makeEncoded: (encoded: Encoded) => Effect.Effect< const primaryKey = Envelope.primaryKeyByAddress(options) return encoded.requestIdForPrimaryKey(primaryKey) }, - unprocessedMessages(shardIds) { + unprocessedMessages(shardIds, options) { const storage = this as MessageStorage["Service"] const shards = Array.from(shardIds, (id) => id.toString()) if (!Arr.isArrayNonEmpty(shards)) return Effect.succeed([]) + let encodedOptions: EncodedUnprocessedMessagesOptions | undefined + if (options) { + const addresses = options.addresses && Array.from(options.addresses) + if (addresses && addresses.length === 0) return Effect.succeed([]) + encodedOptions = { limit: options.limit, addresses } + } return Effect.flatMap( - Effect.suspend(() => encoded.unprocessedMessages(shards, clock.currentTimeMillisUnsafe())), + Effect.suspend(() => encoded.unprocessedMessages(shards, clock.currentTimeMillisUnsafe(), encodedOptions)), (messages) => decodeMessages(storage, messages) ) }, @@ -947,20 +993,31 @@ export class MemoryDriver extends Context.Service()("effect/cluste repliesFor: (requestIds) => Effect.sync(() => repliesFor(requestIds)), repliesForUnfiltered: (requestIds) => Effect.sync(() => requestIds.flatMap((id) => requests.get(String(id))?.replies ?? [])), - unprocessedMessages: (shardIds) => + unprocessedMessages: (shardIds, _now, options) => Effect.sync(() => { if (unprocessed.size === 0) return [] const now = clock.currentTimeMillisUnsafe() + const limit = options?.limit ?? Infinity + let addressFilter: Set | undefined + if (options?.addresses) { + addressFilter = new Set( + options.addresses.map((address) => `${address.entityType}/${address.entityId}`) + ) + } const messages = Arr.empty<{ envelope: Envelope.Encoded lastSentReply: Option.Option }>() for (let index = 0; index < journal.length; index++) { + if (messages.length >= limit) break const envelope = journal[index] const shardId = ShardId.make(envelope.address.shardId.group, envelope.address.shardId.id) if (!unprocessed.has(envelope as any) || !shardIds.includes(shardId.toString())) { continue } + if (addressFilter && !addressFilter.has(`${envelope.address.entityType}/${envelope.address.entityId}`)) { + continue + } if (envelope._tag === "Request") { const entry = requests.get(envelope.requestId)! if (entry.deliverAt && entry.deliverAt > now) { diff --git a/packages/effect/src/unstable/cluster/Sharding.ts b/packages/effect/src/unstable/cluster/Sharding.ts index ab40a7cfecb..643ae96e918 100644 --- a/packages/effect/src/unstable/cluster/Sharding.ts +++ b/packages/effect/src/unstable/cluster/Sharding.ts @@ -588,6 +588,33 @@ const make = Effect.gen(function*() { const storageReadLock = Semaphore.makeUnsafe(1) const withStorageReadLock = storageReadLock.withPermits(1) + // --- Entity residency --- + // + // A runner-wide counter that bounds how many entities can be resident at + // the same time. Entity managers reserve a slot before spawning an entity, + // so the sequential storage read loop and concurrent volatile sends share + // the same cap. + + const maxResidentEntities = config.maxResidentEntities + let residentEntityCount = 0 + const residencyAtCapacityUnsafe = () => + maxResidentEntities !== "unbounded" && residentEntityCount >= maxResidentEntities + const residency: EntityManager.Residency = { + admitUnsafe() { + if (residencyAtCapacityUnsafe()) return false + residentEntityCount++ + return true + }, + releaseUnsafe() { + residentEntityCount-- + // a slot has freed up, so the storage read loop can admit messages for + // entities it previously had to skip + if (maxResidentEntities !== "unbounded") { + storageReadLatch.openUnsafe() + } + } + } + if (storageEnabled && initialRunnerAddress) { const selfAddress = initialRunnerAddress const entityRegistrationTimeoutMillis = Duration.toMillis( @@ -600,8 +627,11 @@ const make = Effect.gen(function*() { let index = 0 let messages: Array> = [] + let cappedThisRead = false + let deliveredThisRead = false const removableNotifications = new Set() const resetAddresses = MutableHashSet.empty() + const cappedAddresses = MutableHashSet.empty() const processMessages = Effect.whileLoop({ while: () => index < messages.length, @@ -642,6 +672,7 @@ const make = Effect.gen(function*() { if (message._tag === "IncomingEnvelope" && isProcessing) { // If the message might affect a currently processing request, we // send it to the entity manager to be processed. + deliveredThisRead = true return state.manager.send(message) } else if (isProcessing || state.status === "closing") { // If the request is already processing, we skip it. @@ -654,6 +685,15 @@ const make = Effect.gen(function*() { entry.resume(Effect.void) } + // The runner is at entity capacity, so messages for entities that + // are not already resident are skipped. Their claims are released + // after the read, so they stay eligible for once a slot frees up. + if (residencyAtCapacityUnsafe() && !state.manager.isResidentUnsafe(address)) { + cappedThisRead = true + MutableHashSet.add(cappedAddresses, address) + return Effect.void + } + // If the entity was resuming in another fiber, we add the message // id to the unprocessed set. const resumptionState = MutableHashMap.get(entityResumptionState, address) @@ -664,6 +704,7 @@ const make = Effect.gen(function*() { } return Effect.void } + deliveredThisRead = true return state.manager.send(message) }), (cause) => { @@ -681,13 +722,26 @@ const make = Effect.gen(function*() { })) } if (error.success._tag === "MailboxFull") { + const address = message.envelope.address + const state = entityManagers.get(address.entityType) + // A resident entity has a full per-entity mailbox; its messages + // are resumed from storage once there is capacity again. // MailboxFull can only happen for requests, so this cast is safe - return resumeEntityFromStorage(message as Message.IncomingRequest) + if (message._tag === "IncomingRequest" && state?.manager.isResidentUnsafe(address)) { + return resumeEntityFromStorage(message as Message.IncomingRequest) + } + // Otherwise the runner ran out of entity slots while the message + // was in flight; leave it in storage. + cappedThisRead = true + MutableHashSet.add(cappedAddresses, address) + return Effect.void } return Effect.void } ) + const batchSize = config.unprocessedMessageBatchSize + while (true) { // wait for the next poll interval, or if we get notified of a change yield* storageReadLatch.await @@ -706,15 +760,66 @@ const make = Effect.gen(function*() { pendingNotifications.forEach((entry) => removableNotifications.add(entry)) } - messages = yield* storage.unprocessedMessages(acquiredShards) - index = 0 - yield* processMessages + cappedThisRead = false + deliveredThisRead = false + let readCount = 0 + let fullBatch = false + let exhaustive = false - if (removableNotifications.size > 0) { - removableNotifications.forEach(({ message, resume }) => { - pendingNotifications.delete(message.envelope.requestId) - resume(Effect.fail(new EntityNotAssignedToRunner({ address: message.envelope.address }))) + // First deliver messages for entities that are already resident, so + // they keep making progress even when the runner is at entity + // capacity. + const residentAddresses: Array = [] + for (const state of entityManagers.values()) { + if (state.status === "closed") continue + for (const address of state.manager.residentAddressesUnsafe()) { + residentAddresses.push(address) + } + } + if (residentAddresses.length > 0) { + messages = yield* storage.unprocessedMessages(acquiredShards, { + limit: batchSize, + addresses: residentAddresses }) + index = 0 + yield* processMessages + readCount = messages.length + fullBatch = messages.length >= batchSize + } + + // Then walk the remaining messages, spawning new entities while the + // runner has entity slots left. + if (!residencyAtCapacityUnsafe() && readCount < batchSize) { + const limit = batchSize - readCount + messages = yield* storage.unprocessedMessages(acquiredShards, { limit }) + index = 0 + yield* processMessages + if (messages.length >= limit) { + fullBatch = true + } else if (!cappedThisRead) { + exhaustive = true + } + } + + if (removableNotifications.size > 0) { + if (exhaustive) { + // The whole backlog was read, so a message that was not seen is + // no longer processable by this runner. + removableNotifications.forEach(({ message, resume }) => { + pendingNotifications.delete(message.envelope.requestId) + resume(Effect.fail(new EntityNotAssignedToRunner({ address: message.envelope.address }))) + }) + } else if (cappedThisRead || residencyAtCapacityUnsafe()) { + // The runner is at entity capacity. The messages are safely + // persisted and are delivered once a slot frees up, so persisted + // senders succeed instead of failing at the cap. + removableNotifications.forEach(({ message, resume }) => { + pendingNotifications.delete(message.envelope.requestId) + resume(Effect.void) + }) + } + // Otherwise the read was truncated by the batch size; the + // notifications stay registered for the immediately following read. removableNotifications.clear() } if (MutableHashSet.size(resetAddresses) > 0) { @@ -726,9 +831,28 @@ const make = Effect.gen(function*() { } MutableHashSet.clear(resetAddresses) } + // Release the claims of the messages that were skipped because of the + // entity cap, so they are re-read as soon as a slot frees up. + if (MutableHashSet.size(cappedAddresses) > 0) { + yield* Effect.forEach( + cappedAddresses, + (address) => Effect.ignore(storage.resetAddress(address)), + { discard: true } + ) + MutableHashSet.clear(cappedAddresses) + } // let the resuming entities check if they are done yield* storageReadLock.release(1) + + // A full batch means more messages could be waiting; start the next + // read immediately, as long as this read made progress (a full batch + // of skipped messages must not spin the loop). When the runner is at + // capacity, the next read waits for an entity to be removed or the + // next poll interval instead. + if (fullBatch && deliveredThisRead) { + storageReadLatch.openUnsafe() + } } }).pipe( Effect.scoped, @@ -1475,6 +1599,7 @@ const make = Effect.gen(function*() { ...options, storage, runnerAddress, + residency, sharding }).pipe( Effect.provideContext(services.pipe( diff --git a/packages/effect/src/unstable/cluster/ShardingConfig.ts b/packages/effect/src/unstable/cluster/ShardingConfig.ts index e9a465d6375..15745a2a873 100644 --- a/packages/effect/src/unstable/cluster/ShardingConfig.ts +++ b/packages/effect/src/unstable/cluster/ShardingConfig.ts @@ -95,6 +95,31 @@ export class ShardingConfig extends Context.Service = Config.all({ Config.withDefault(defaults.entityMailboxCapacity) // Config.withDescription("The default capacity of the mailbox for entities.") ), + maxResidentEntities: Config.int("maxResidentEntities").pipe( + Config.withDefault(defaults.maxResidentEntities) + // Config.withDescription("The maximum number of entities that can be resident on this runner at the same time.") + ), + unprocessedMessageBatchSize: Config.int("unprocessedMessageBatchSize").pipe( + Config.withDefault(defaults.unprocessedMessageBatchSize) + // Config.withDescription("The maximum number of unprocessed messages read from storage in a single poll.") + ), entityMaxIdleTime: Config.duration("entityMaxIdleTime").pipe( Config.withDefault(defaults.entityMaxIdleTime) // Config.withDescription( diff --git a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts index e8c2ed4c0d2..b7752e489d5 100644 --- a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts +++ b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts @@ -372,9 +372,48 @@ export const make: (options?: { orElse: () => sql.literal("FOR UPDATE") }) + const emptyFragment = sql.literal("") + // mssql only limits with OFFSET/FETCH, which requires the ORDER BY that + // both queries already have + const limitFragment = sql.onDialectOrElse({ + mssql: () => (limit: number) => sql.literal(`OFFSET 0 ROWS FETCH NEXT ${Math.floor(limit)} ROWS ONLY`), + orElse: () => (limit: number) => sql.literal(`LIMIT ${Math.floor(limit)}`) + }) + const unprocessedFilters = (options?: MessageStorage.EncodedUnprocessedMessagesOptions | undefined) => { + let addressFilter = emptyFragment + if (options?.addresses !== undefined) { + const idsByType = new Map>() + for (const address of options.addresses) { + let ids = idsByType.get(address.entityType) + if (ids === undefined) { + ids = [] + idsByType.set(address.entityType, ids) + } + ids.push(address.entityId) + } + addressFilter = sql`AND (${ + sql.or(Array.from(idsByType, ([entityType, entityIds]) => + sql.and([ + sql`m.entity_type = ${entityType}`, + sql.in("m.entity_id", entityIds) + ]))) + })` + } + return { + addressFilter, + limit: options?.limit !== undefined ? limitFragment(options.limit) : emptyFragment + } + } + const getUnprocessedMessages = sql.onDialectOrElse({ - pg: () => (shardIds: ReadonlyArray, now: number) => - sql` + pg: () => + ( + shardIds: ReadonlyArray, + now: number, + options?: MessageStorage.EncodedUnprocessedMessagesOptions | undefined + ) => { + const filters = unprocessedFilters(options) + return sql` WITH messages AS ( UPDATE ${messagesTableSql} m SET last_read = ${sqlNow} @@ -382,6 +421,7 @@ export const make: (options?: { SELECT m.* FROM ${messagesTableSql} m WHERE m.shard_id IN (${sql.literal(shardIds.map(wrapString).join(","))}) + ${filters.addressFilter} AND NOT EXISTS ( SELECT 1 FROM ${repliesTableSql} WHERE request_id = m.request_id @@ -390,6 +430,8 @@ export const make: (options?: { AND m.processed = ${sqlFalse} AND (m.last_read IS NULL OR m.last_read < ${tenMinutesAgo}) AND (m.deliver_at IS NULL OR m.deliver_at <= ${sql.literal(String(now))}) + ORDER BY m.rowid ASC + ${filters.limit} FOR UPDATE ) AS ids LEFT JOIN ${repliesTableSql} r ON r.id = ids.last_reply_id @@ -397,13 +439,21 @@ export const make: (options?: { RETURNING ids.*, r.id as reply_reply_id, r.kind as reply_kind, r.payload as reply_payload, r.sequence as reply_sequence ) SELECT * FROM messages ORDER BY rowid ASC - `, - orElse: () => (shardIds: ReadonlyArray, now: number) => - sql` + ` + }, + orElse: () => + ( + shardIds: ReadonlyArray, + now: number, + options?: MessageStorage.EncodedUnprocessedMessagesOptions | undefined + ) => { + const filters = unprocessedFilters(options) + return sql` SELECT m.*, r.id as reply_reply_id, r.kind as reply_kind, r.payload as reply_payload, r.sequence as reply_sequence FROM ${messagesTableSql} m LEFT JOIN ${repliesTableSql} r ON r.id = m.last_reply_id WHERE m.shard_id IN (${sql.literal(shardIds.map(wrapString).join(","))}) + ${filters.addressFilter} AND NOT EXISTS ( SELECT 1 FROM ${repliesTableSql} WHERE request_id = m.request_id @@ -413,6 +463,7 @@ export const make: (options?: { AND (m.last_read IS NULL OR m.last_read < ${tenMinutesAgo}) AND (m.deliver_at IS NULL OR m.deliver_at <= ${sql.literal(String(now))}) ORDER BY m.rowid ASC + ${filters.limit} ${forUpdate} `.unprepared.pipe( Effect.tap((rows) => { @@ -427,6 +478,7 @@ export const make: (options?: { }), sql.withTransaction ) + } }) return yield* MessageStorage.makeEncoded({ @@ -579,8 +631,8 @@ export const make: (options?: { ), unprocessedMessages: Effect.fnUntraced( - function*(shardIds, now) { - const rows = yield* getUnprocessedMessages(shardIds, now) + function*(shardIds, now, options) { + const rows = yield* getUnprocessedMessages(shardIds, now, options) if (rows.length === 0) { return [] } diff --git a/packages/effect/src/unstable/cluster/internal/entityManager.ts b/packages/effect/src/unstable/cluster/internal/entityManager.ts index e568d3d07e2..b0f62f937d4 100644 --- a/packages/effect/src/unstable/cluster/internal/entityManager.ts +++ b/packages/effect/src/unstable/cluster/internal/entityManager.ts @@ -56,6 +56,9 @@ export interface EntityManager { }) => boolean readonly clearProcessed: () => void + readonly isResidentUnsafe: (address: EntityAddress) => boolean + readonly residentAddressesUnsafe: () => Array + readonly interruptShard: (shardId: ShardId, options?: { readonly force?: boolean }) => Effect.Effect @@ -63,6 +66,19 @@ export interface EntityManager { readonly activeEntityCount: Effect.Effect } +// Tracks how many entities are resident on the runner across all entity +// managers, so the spawn of new entities can be gated by +// `ShardingConfig.maxResidentEntities`. +/** @internal */ +export interface Residency { + /** + * Reserve a slot for a new entity. Returns `false` when the runner is at + * capacity. + */ + readonly admitUnsafe: () => boolean + readonly releaseUnsafe: () => void +} + // Represents the entities managed by this entity manager /** @internal */ export type EntityState = { @@ -94,6 +110,7 @@ export const make = Effect.fnUntraced(function*< readonly sharding: Sharding["Service"] readonly storage: MessageStorage.MessageStorage["Service"] readonly runnerAddress: RunnerAddress + readonly residency: Residency readonly maxIdleTime?: Input | undefined readonly concurrency?: number | "unbounded" | undefined readonly mailboxCapacity?: number | "unbounded" | undefined @@ -128,13 +145,22 @@ export const make = Effect.fnUntraced(function*< const entities: ResourceMap< EntityAddress, EntityState, - EntityNotAssignedToRunner + EntityNotAssignedToRunner | MailboxFull > = yield* ResourceMap.make(Effect.fnUntraced(function*(address: EntityAddress) { if (!options.sharding.hasShardId(address.shardId)) { return yield* new EntityNotAssignedToRunner({ address }) } const scope = yield* Effect.scope + + // Gate the spawn on the runner-wide entity cap. Registering the release + // must be atomic with taking the slot, otherwise an interrupt in between + // would leak it. + yield* Effect.uninterruptible(Effect.suspend(() => + options.residency.admitUnsafe() + ? Scope.addFinalizer(scope, Effect.sync(options.residency.releaseUnsafe)) + : Effect.fail(new MailboxFull({ address })) + )) const endLatch = Latch.makeUnsafe() const keepAliveLatch = Latch.makeUnsafe() const closeLatches = { @@ -590,6 +616,8 @@ export const make = Effect.fnUntraced(function*< clearProcessed() { processedRequestIds.clear() }, + isResidentUnsafe: (address) => entities.hasUnsafe(address), + residentAddressesUnsafe: () => entities.keysUnsafe(), sendLocal, send: (message) => decodeMessage(message).pipe( diff --git a/packages/effect/src/unstable/cluster/internal/entityReaper.ts b/packages/effect/src/unstable/cluster/internal/entityReaper.ts index 0c3c974b813..5037b46c9ae 100644 --- a/packages/effect/src/unstable/cluster/internal/entityReaper.ts +++ b/packages/effect/src/unstable/cluster/internal/entityReaper.ts @@ -3,7 +3,7 @@ import * as Context from "../../../Context.ts" import * as Effect from "../../../Effect.ts" import * as Latch from "../../../Latch.ts" import * as Layer from "../../../Layer.ts" -import type { EntityNotAssignedToRunner } from "../ClusterError.ts" +import type { EntityNotAssignedToRunner, MailboxFull } from "../ClusterError.ts" import type { EntityAddress } from "../EntityAddress.ts" import type { EntityId } from "../EntityId.ts" import type { EntityState } from "./entityManager.ts" @@ -16,14 +16,14 @@ export class EntityReaper extends Context.Service()("effect/cluste const registered: Array<{ readonly maxIdleTime: number readonly servers: Map - readonly entities: ResourceMap + readonly entities: ResourceMap }> = [] const latch = yield* Latch.make() const register = (options: { readonly maxIdleTime: number readonly servers: Map - readonly entities: ResourceMap + readonly entities: ResourceMap }) => Effect.suspend(() => { currentResolution = Math.max(Math.min(currentResolution, options.maxIdleTime), 5000) diff --git a/packages/effect/src/unstable/cluster/internal/resourceMap.ts b/packages/effect/src/unstable/cluster/internal/resourceMap.ts index 13377199b60..938056653ee 100644 --- a/packages/effect/src/unstable/cluster/internal/resourceMap.ts +++ b/packages/effect/src/unstable/cluster/internal/resourceMap.ts @@ -56,6 +56,18 @@ export class ResourceMap { ) }) + hasUnsafe(key: K): boolean { + return backingGet(this.entries, key) !== undefined + } + + keysUnsafe(): Array { + const keys: Array = [] + for (const [key] of this.entries.map) { + keys.push(key) + } + return keys + } + get(key: K): Effect.Effect { return Effect.suspend(() => { if (MutableRef.get(this.isClosed)) { diff --git a/packages/effect/test/cluster/MessageStorage.test.ts b/packages/effect/test/cluster/MessageStorage.test.ts index b0fdb941eba..94a6eb5680a 100644 --- a/packages/effect/test/cluster/MessageStorage.test.ts +++ b/packages/effect/test/cluster/MessageStorage.test.ts @@ -87,6 +87,42 @@ describe("MessageStorage", () => { expect(messages).toHaveLength(0) }).pipe(Effect.provide(MemoryLive))) + it.effect("unprocessedMessages honors the limit option", () => + Effect.gen(function*() { + const storage = yield* MessageStorage.MessageStorage + for (let i = 1; i <= 5; i++) { + yield* storage.saveRequest(yield* makeRequest({ payload: { id: i }, entityId: String(i) })) + } + const shardId = ShardId.make("default", 1) + const messages = yield* storage.unprocessedMessages([shardId], { limit: 3 }) + expect(messages).toHaveLength(3) + expect(messages.map((m: any) => m.envelope.payload.id)).toEqual([1, 2, 3]) + const all = yield* storage.unprocessedMessages([shardId]) + expect(all).toHaveLength(5) + }).pipe(Effect.provide(MemoryLive))) + + it.effect("unprocessedMessages filters by address", () => + Effect.gen(function*() { + const storage = yield* MessageStorage.MessageStorage + for (let i = 1; i <= 4; i++) { + yield* storage.saveRequest(yield* makeRequest({ payload: { id: i }, entityId: String(i) })) + } + const shardId = ShardId.make("default", 1) + const address = (entityId: string) => + EntityAddress.make({ + shardId, + entityType: EntityType.make("test"), + entityId: EntityId.make(entityId) + }) + const messages = yield* storage.unprocessedMessages([shardId], { + addresses: [address("2"), address("4")] + }) + expect(messages.map((m: any) => m.envelope.payload.id)).toEqual([2, 4]) + // an empty address filter returns nothing + const none = yield* storage.unprocessedMessages([shardId], { addresses: [] }) + expect(none).toHaveLength(0) + }).pipe(Effect.provide(MemoryLive))) + it.effect("repliesFor", () => Effect.gen(function*() { const storage = yield* MessageStorage.MessageStorage @@ -127,6 +163,7 @@ export const GetUserRpc = Rpc.make("GetUser", { export const makeRequest = Effect.fnUntraced(function*(options?: { readonly rpc?: Rpc.AnyWithProps readonly payload?: any + readonly entityId?: string }) { const snowflake = yield* Snowflake.Generator const rpc = options?.rpc ?? GetUserRpc @@ -136,7 +173,7 @@ export const makeRequest = Effect.fnUntraced(function*(options?: { address: EntityAddress.make({ shardId: ShardId.make("default", 1), entityType: EntityType.make("test"), - entityId: EntityId.make("1") + entityId: EntityId.make(options?.entityId ?? "1") }), tag: rpc._tag, payload: options?.payload ?? { id: 123 }, diff --git a/packages/effect/test/cluster/Sharding.test.ts b/packages/effect/test/cluster/Sharding.test.ts index 651bee01225..3bf753bef9a 100644 --- a/packages/effect/test/cluster/Sharding.test.ts +++ b/packages/effect/test/cluster/Sharding.test.ts @@ -23,8 +23,12 @@ import { ClusterMetrics, ClusterSchema, Entity, + EntityAddress, EntityId, + EntityType, + Envelope, MachineId, + Message, MessageStorage, Runner, RunnerAddress, @@ -36,6 +40,7 @@ import { ShardingConfig, Snowflake } from "effect/unstable/cluster" +import { Headers } from "effect/unstable/http" import { Rpc } from "effect/unstable/rpc" import { CallerId, @@ -1070,6 +1075,163 @@ describe.concurrent("Sharding", () => { })) }) +describe.concurrent("Sharding residency cap", () => { + it.effect("bounds resident entities to maxResidentEntities", () => + Effect.gen(function*() { + yield* TestClock.adjust(1) + const sharding = yield* Sharding.Sharding + const makeClient = yield* TestEntity.client + const fibers: Array> = [] + for (let i = 1; i <= 6; i++) { + fibers.push( + yield* makeClient(String(i)).GetUser({ id: i }).pipe( + Effect.forkChild({ startImmediately: true }) + ) + ) + } + yield* TestClock.adjust(1) + assert.isAtMost(yield* sharding.activeEntityCount, 2) + + // idle entities are reaped over time, freeing slots for the backlog + for (let i = 0; i < 12; i++) { + yield* TestClock.adjust(5000) + assert.isAtMost(yield* sharding.activeEntityCount, 2) + } + + const users = yield* Fiber.joinAll(fibers) + assert.deepStrictEqual(users.map((user) => user.id), [1, 2, 3, 4, 5, 6]) + }).pipe(Effect.provide(CappedSharding({ maxResidentEntities: 2, entityMaxIdleTime: 1000 })))) + + it.effect("volatile sends to new entities fail with MailboxFull at the cap", () => + Effect.gen(function*() { + yield* TestClock.adjust(1) + const driver = yield* MessageStorage.MemoryDriver + const sharding = yield* Sharding.Sharding + const makeClient = yield* TestEntity.client + // occupy the only slot + yield* makeClient("1").NeverVolatile().pipe(Effect.forkChild({ startImmediately: true })) + yield* TestClock.adjust(1) + assert.strictEqual(yield* sharding.activeEntityCount, 1) + + const error = yield* makeClient("2").GetUserVolatile({ id: 2 }).pipe(Effect.flip) + assert.strictEqual(error._tag, "MailboxFull") + + // persisted sends still succeed and wait in storage + yield* makeClient("2").GetUser({ id: 2 }, { discard: true }) + yield* TestClock.adjust(5000) + assert.strictEqual(yield* sharding.activeEntityCount, 1) + assert.strictEqual(driver.unprocessed.size, 1) + assert.strictEqual(driver.replyIds.size, 0) + }).pipe(Effect.provide(CappedSharding({ maxResidentEntities: 1 })))) + + it.effect("keeps delivering to resident entities at the cap", () => + Effect.gen(function*() { + yield* TestClock.adjust(1) + const driver = yield* MessageStorage.MemoryDriver + const state = yield* TestEntityState + const sharding = yield* Sharding.Sharding + const makeClient = yield* TestEntity.client + // make entity "1" resident and keep it busy + yield* makeClient("1").NeverFork().pipe(Effect.forkChild({ startImmediately: true })) + yield* TestClock.adjust(1) + assert.strictEqual(Queue.sizeUnsafe(state.envelopes), 1) + + // a backlog of new entity ids in front of the resident's next message + for (let i = 2; i <= 4; i++) { + yield* makeClient(String(i)).GetUser({ id: i }, { discard: true }) + } + const fiber = yield* makeClient("1").GetUser({ id: 1 }).pipe( + Effect.forkChild({ startImmediately: true }) + ) + yield* TestClock.adjust(5000) + + // the resident entity received its message despite the backlog in front + assert.deepStrictEqual(yield* Fiber.join(fiber), new User({ id: 1, name: "User 1" })) + // the new ids were not admitted and their requests stay in storage + assert.strictEqual(yield* sharding.activeEntityCount, 1) + assert.strictEqual(driver.unprocessed.size, 4) + assert.strictEqual(driver.replyIds.size, 1) + }).pipe(Effect.provide(CappedSharding({ + maxResidentEntities: 1, + unprocessedMessageBatchSize: 2 + })))) + + it.effect("drains a full batch without waiting for the poll interval", () => + Effect.gen(function*() { + yield* TestClock.adjust(1) + const driver = yield* MessageStorage.MemoryDriver + const storage = yield* MessageStorage.MessageStorage + const sharding = yield* Sharding.Sharding + const rpc = TestEntity.protocol.requests.get("GetUser")! as any + + // save requests directly to storage, so the read loop only learns + // about them from the next poll + for (let i = 1; i <= 5; i++) { + const entityId = EntityId.make(String(i)) + yield* storage.saveRequest( + new Message.OutgoingRequest({ + envelope: Envelope.makeRequest({ + requestId: yield* sharding.getSnowflake, + address: EntityAddress.make({ + shardId: sharding.getShardId(entityId, "default"), + entityType: EntityType.make(TestEntity.type), + entityId + }), + tag: "GetUser", + payload: { id: i }, + headers: Headers.empty + }), + annotations: rpc.annotations, + context: Context.empty() as any, + rpc, + lastReceivedReply: Option.none(), + respond: () => Effect.void + }) + ) + } + assert.strictEqual(driver.replyIds.size, 0) + + // a single poll drains the whole backlog in batches of 2 + yield* TestClock.adjust(5000) + assert.strictEqual(driver.replyIds.size, 5) + }).pipe(Effect.provide(CappedSharding({ unprocessedMessageBatchSize: 2 })))) + + it.effect("does not busy-poll storage at the cap", () => + Effect.gen(function*() { + let reads = 0 + yield* Effect.gen(function*() { + yield* TestClock.adjust(1) + const makeClient = yield* TestEntity.client + yield* makeClient("1").NeverVolatile().pipe(Effect.forkChild({ startImmediately: true })) + yield* TestClock.adjust(1) + // a backlog for a new id that cannot be admitted + yield* makeClient("2").GetUser({ id: 2 }, { discard: true }) + yield* TestClock.adjust(1) + const before = reads + yield* TestClock.adjust(4000) // less than the poll interval + assert.isAtMost(reads - before, 1) + }).pipe(Effect.provide(CappedSharding({ maxResidentEntities: 1 }, (storage) => ({ + ...storage, + unprocessedMessages(shardIds, options) { + reads++ + return storage.unprocessedMessages(shardIds, options) + } + })))) + })) + + it.effect("unbounded maxResidentEntities does not limit spawning", () => + Effect.gen(function*() { + yield* TestClock.adjust(1) + const sharding = yield* Sharding.Sharding + const makeClient = yield* TestEntity.client + for (let i = 1; i <= 15; i++) { + yield* makeClient(String(i)).NeverVolatile().pipe(Effect.forkChild({ startImmediately: true })) + } + yield* TestClock.adjust(1) + assert.strictEqual(yield* sharding.activeEntityCount, 15) + }).pipe(Effect.provide(CappedSharding({ maxResidentEntities: "unbounded" })))) +}) + describe("Sharding shard lock failover", () => { it.effect("interrupts entities and reacquires shards after lock storage recovers", () => Effect.gen(function*() { @@ -1493,4 +1655,31 @@ const TestSharding = TestShardingWithoutStorage.pipe( Layer.provide(TestShardingConfig) ) +const CappedSharding = ( + config: Partial, + transformStorage?: ( + storage: MessageStorage.MessageStorage["Service"] + ) => MessageStorage.MessageStorage["Service"] +) => { + const configLayer = ShardingConfig.layer({ + entityMailboxCapacity: 10, + entityTerminationTimeout: 0, + entityMessagePollInterval: 5000, + sendRetryInterval: 100, + refreshAssignmentsInterval: 0, + ...config + }) + let layer = TestShardingWithoutRunners.pipe( + Layer.provide(Runners.layerNoop), + Layer.provide(configLayer) + ) + if (transformStorage) { + layer = layer.pipe(Layer.updateService(MessageStorage.MessageStorage, transformStorage)) + } + return layer.pipe( + Layer.provideMerge(MessageStorage.layerMemory), + Layer.provide(configLayer) + ) +} + const ContextBleedSharding = ContextBleedLayer.pipe(Layer.provideMerge(TestSharding)) diff --git a/packages/effect/test/cluster/ShardingConfig.test.ts b/packages/effect/test/cluster/ShardingConfig.test.ts index 44ef9a893f2..46be4bbaad9 100644 --- a/packages/effect/test/cluster/ShardingConfig.test.ts +++ b/packages/effect/test/cluster/ShardingConfig.test.ts @@ -3,6 +3,33 @@ import { ConfigProvider, Effect, Option } from "effect" import { RunnerAddress, ShardingConfig } from "effect/unstable/cluster" describe("ShardingConfig", () => { + it.effect("bounds runner residency and storage reads by default", () => + Effect.gen(function*() { + assert.strictEqual(ShardingConfig.defaults.maxResidentEntities, 10_000) + assert.strictEqual(ShardingConfig.defaults.unprocessedMessageBatchSize, 1024) + + const defaults = yield* ShardingConfig.config.parse(ConfigProvider.fromUnknown({})) + assert.strictEqual(defaults.maxResidentEntities, 10_000) + assert.strictEqual(defaults.unprocessedMessageBatchSize, 1024) + + // environment values are integers only + const fromEnv = yield* ShardingConfig.config.parse( + ConfigProvider.fromUnknown({ maxResidentEntities: "500", unprocessedMessageBatchSize: "128" }) + ) + assert.strictEqual(fromEnv.maxResidentEntities, 500) + assert.strictEqual(fromEnv.unprocessedMessageBatchSize, 128) + + const invalid = yield* ShardingConfig.config.parse( + ConfigProvider.fromUnknown({ maxResidentEntities: "unbounded" }) + ).pipe(Effect.flip) + assert.strictEqual(invalid._tag, "ConfigError") + + // "unbounded" is only available programmatically + const layer = ShardingConfig.layer({ maxResidentEntities: "unbounded" }) + const config = yield* Effect.provide(ShardingConfig.ShardingConfig, layer) + assert.strictEqual(config.maxResidentEntities, "unbounded") + })) + it.effect("treats the optional listen address as an atomic group", () => Effect.gen(function*() { const defaults = yield* ShardingConfig.config.parse(ConfigProvider.fromUnknown({})) diff --git a/packages/platform/node/test/cluster-integration/Residency.test.ts b/packages/platform/node/test/cluster-integration/Residency.test.ts new file mode 100644 index 00000000000..509c176b332 --- /dev/null +++ b/packages/platform/node/test/cluster-integration/Residency.test.ts @@ -0,0 +1,76 @@ +import { assert, describe, it } from "@effect/vitest" +import { Effect, Fiber, Latch, Schema } from "effect" +import { ClusterSchema, Entity } from "effect/unstable/cluster" +import { Rpc } from "effect/unstable/rpc" +import { type Backend, make } from "./harness.ts" + +const ResidencyEntity = Entity.make("ClusterIntegrationResidency", [ + Rpc.make("Ping", { + payload: { id: Schema.String }, + success: Schema.String + }), + Rpc.make("Hold", { + payload: { id: Schema.String } + }) +]).annotateRpcs(ClusterSchema.Persisted, true) + +let holdGate = Latch.makeUnsafe() + +const ResidencyEntityLayer = ResidencyEntity.toLayer({ + Ping: ({ payload }) => Effect.succeed(payload.id), + Hold: () => holdGate.await +}, { maxIdleTime: "1 second", concurrency: 2 }) + +describe("cluster residency integration", () => { + for (const backend of ["pg", "mysql"] satisfies ReadonlyArray) { + it.live(`${backend}: stays at the entity cap and keeps residents moving`, () => + Effect.gen(function*() { + holdGate = Latch.makeUnsafe() + const cluster = yield* make({ + backend, + entities: ResidencyEntityLayer, + config: { maxResidentEntities: 3 } + }) + const [runner] = yield* cluster.start(1) + yield* cluster.waitForStableAssignments() + const client = yield* cluster.getClient(ResidencyEntity) + + // fill every entity slot with a held request + const held = yield* Effect.forEach( + ["held-1", "held-2", "held-3"], + (id) => client(id).Hold({ id }).pipe(Effect.forkChild({ startImmediately: true })) + ) + yield* cluster.waitUntil( + "The held entities did not become resident", + Effect.map(runner.sharding.activeEntityCount, (count) => count === 3) + ) + + // a backlog of messages for new entity ids stays in storage + for (let i = 1; i <= 8; i++) { + yield* client(`backlog-${i}`).Ping({ id: `backlog-${i}` }, { discard: true }) + } + + // resident entities keep receiving messages while at the cap + assert.strictEqual(yield* client("held-1").Ping({ id: "first" }), "first") + assert.strictEqual(yield* client("held-2").Ping({ id: "second" }), "second") + + // the runner never spawned entities for the backlog + assert.strictEqual(yield* runner.sharding.activeEntityCount, 3) + assert.isAtLeast(yield* cluster.unprocessedMessageCount, 8) + + // release the held entities: as slots free up, the backlog drains in + // waves without ever exceeding the cap + holdGate.openUnsafe() + yield* Fiber.joinAll(held) + yield* cluster.waitUntil( + "The backlog did not drain after entity slots freed up", + Effect.gen(function*() { + assert.isAtMost(yield* runner.sharding.activeEntityCount, 3) + const counts = yield* cluster.messageCounts() + return counts.unprocessed === 0 && counts.failed === 0 + }), + "45 seconds" + ) + })) + } +}) diff --git a/packages/platform/node/test/cluster/MessageStorageTest.ts b/packages/platform/node/test/cluster/MessageStorageTest.ts index 972719fda0f..ce0aa697df8 100644 --- a/packages/platform/node/test/cluster/MessageStorageTest.ts +++ b/packages/platform/node/test/cluster/MessageStorageTest.ts @@ -101,6 +101,7 @@ export const GetUserRpc = Rpc.make("GetUser", { export const makeRequest = Effect.fnUntraced(function*(options?: { readonly rpc?: Rpc.AnyWithProps readonly payload?: any + readonly entityId?: string }) { const snowflake = yield* Snowflake.Generator const rpc = options?.rpc ?? GetUserRpc @@ -110,7 +111,7 @@ export const makeRequest = Effect.fnUntraced(function*(options?: { address: EntityAddress.make({ shardId: ShardId.make("default", 1), entityType: EntityType.make("test"), - entityId: EntityId.make("1") + entityId: EntityId.make(options?.entityId ?? "1") }), tag: rpc._tag, payload: options?.payload ?? { id: 123 }, diff --git a/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts b/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts index 6c88a13aaca..914f79f4292 100644 --- a/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts +++ b/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts @@ -5,12 +5,16 @@ import { Effect, Fiber, FileSystem, Latch, Layer, Option } from "effect" import { TestClock } from "effect/testing" import { Entity, + EntityAddress, + EntityId, + EntityType, Envelope, Message, MessageStorage, RunnerHealth, Runners, RunnerStorage, + ShardId, Sharding, ShardingConfig, Snowflake, @@ -282,6 +286,50 @@ describe("SqlMessageStorage", () => { expect(messages).toHaveLength(1) })) + it.effect("unprocessedMessages honors the limit and claims only returned rows", () => + Effect.gen(function*() { + yield* truncate + + const storage = yield* MessageStorage.MessageStorage + const shardId = ShardId.make("default", 1) + for (let i = 1; i <= 5; i++) { + yield* storage.saveRequest(yield* makeRequest({ payload: { id: i }, entityId: String(i) })) + } + const limited = yield* storage.unprocessedMessages([shardId], { limit: 3 }) + expect(limited).toHaveLength(3) + expect(limited.map((m: any) => m.envelope.payload.id)).toEqual([1, 2, 3]) + + // rows beyond the limit were not claimed and are returned by the + // next read + const rest = yield* storage.unprocessedMessages([shardId]) + expect(rest.map((m: any) => m.envelope.payload.id)).toEqual([4, 5]) + })) + + it.effect("unprocessedMessages filters by address", () => + Effect.gen(function*() { + yield* truncate + + const storage = yield* MessageStorage.MessageStorage + const shardId = ShardId.make("default", 1) + const address = (entityId: string) => + EntityAddress.make({ + shardId, + entityType: EntityType.make("test"), + entityId: EntityId.make(entityId) + }) + for (let i = 1; i <= 4; i++) { + yield* storage.saveRequest(yield* makeRequest({ payload: { id: i }, entityId: String(i) })) + } + const filtered = yield* storage.unprocessedMessages([shardId], { + addresses: [address("2"), address("4")] + }) + expect(filtered.map((m: any) => m.envelope.payload.id)).toEqual([2, 4]) + + // the filtered read must not claim the other addresses + const rest = yield* storage.unprocessedMessages([shardId]) + expect(rest.map((m: any) => m.envelope.payload.id)).toEqual([1, 3]) + })) + it.effect("unprocessedMessages excludes complete requests", () => Effect.gen(function*() { yield* truncate From 66b64dd38c37e9bce411d61d811aff9c45f8c8e2 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Wed, 12 Aug 2026 22:39:23 +0000 Subject: [PATCH 2/9] fix(cluster): address residency review findings --- .../src/unstable/cluster/MessageStorage.ts | 58 ++++++++++++++-- .../effect/src/unstable/cluster/Sharding.ts | 67 ++++++++++--------- .../src/unstable/cluster/ShardingConfig.ts | 10 ++- .../src/unstable/cluster/SqlMessageStorage.ts | 26 ++++++- .../test/cluster/MessageStorage.test.ts | 24 ++++++- packages/effect/test/cluster/Sharding.test.ts | 14 ++++ .../test/cluster/ShardingConfig.test.ts | 15 +++++ .../SqlMessageStorage.integration.test.ts | 22 ++++++ 8 files changed, 196 insertions(+), 40 deletions(-) diff --git a/packages/effect/src/unstable/cluster/MessageStorage.ts b/packages/effect/src/unstable/cluster/MessageStorage.ts index ceebdf03f14..6159a4636fa 100644 --- a/packages/effect/src/unstable/cluster/MessageStorage.ts +++ b/packages/effect/src/unstable/cluster/MessageStorage.ts @@ -163,6 +163,13 @@ export class MessageStorage extends Context.Service Effect.Effect + /** + * Reset the mailbox state for the provided addresses. + */ + readonly resetAddresses: ( + addresses: Iterable + ) => Effect.Effect + /** * Clear all messages and replies for the provided address. */ @@ -411,6 +418,13 @@ export type Encoded = { address: EntityAddress ) => 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. */ @@ -694,6 +708,10 @@ export const makeEncoded: (encoded: Encoded) => Effect.Effect< ) }, resetAddress: encoded.resetAddress, + resetAddresses: (addresses) => { + const array = Array.from(addresses) + return array.length === 0 ? Effect.void : encoded.resetAddresses(array) + }, clearAddress: encoded.clearAddress, resetShards: (shardIds) => { const shards = Array.from(shardIds, (id) => id.toString()) @@ -822,6 +840,7 @@ export const noop: MessageStorage["Service"] = Effect.runSync(make({ unprocessedMessages: () => Effect.succeed([]), unprocessedMessagesById: () => Effect.succeed([]), resetAddress: () => Effect.void, + resetAddresses: () => Effect.void, clearAddress: () => Effect.void, resetShards: () => Effect.void, withTransaction: identity @@ -874,9 +893,23 @@ export class MemoryDriver extends Context.Service()("effect/cluste const requestsByPrimaryKey = new Map() const unprocessed = new Set() const replyIds = new Set() + const lastRead = new Map() const journal: Array = [] + const addressKey = (address: Envelope.Encoded["address"] | EntityAddress) => + `${address.shardId.group}/${address.shardId.id}/${address.entityType}/${address.entityId}` + + const resetAddresses = (addresses: ReadonlyArray) => + Effect.sync(() => { + const keys = new Set(addresses.map(addressKey)) + for (const envelope of journal) { + if (keys.has(addressKey(envelope.address))) { + lastRead.delete(envelope) + } + } + }) + const cursors = new WeakMap<{}, number>() const unprocessedWith = (predicate: Predicate) => { @@ -972,6 +1005,7 @@ export class MemoryDriver extends Context.Service()("effect/cluste if (!entry || replyIds.has(reply.id)) return if (reply._tag === "WithExit") { unprocessed.delete(entry.envelope) + lastRead.delete(entry.envelope) } entry.replies.push(reply) replyIds.add(reply.id) @@ -984,6 +1018,7 @@ export class MemoryDriver extends Context.Service()("effect/cluste entry.replies = [] entry.lastReceivedChunk = undefined unprocessed.add(entry.envelope) + lastRead.delete(entry.envelope) }), requestIdForPrimaryKey: (primaryKey) => Effect.sync(() => { @@ -993,10 +1028,9 @@ export class MemoryDriver extends Context.Service()("effect/cluste repliesFor: (requestIds) => Effect.sync(() => repliesFor(requestIds)), repliesForUnfiltered: (requestIds) => Effect.sync(() => requestIds.flatMap((id) => requests.get(String(id))?.replies ?? [])), - unprocessedMessages: (shardIds, _now, options) => + unprocessedMessages: (shardIds, now, options) => Effect.sync(() => { if (unprocessed.size === 0) return [] - const now = clock.currentTimeMillisUnsafe() const limit = options?.limit ?? Infinity let addressFilter: Set | undefined if (options?.addresses) { @@ -1023,10 +1057,15 @@ export class MemoryDriver extends Context.Service()("effect/cluste if (entry.deliverAt && entry.deliverAt > now) { continue } + const claimedAt = lastRead.get(envelope) + if (claimedAt !== undefined && claimedAt > now - 10 * 60 * 1000) { + continue + } messages.push({ envelope, lastSentReply: Option.fromNullishOr(entry.replies[entry.replies.length - 1]) }) + lastRead.set(envelope, now) } else { messages.push({ envelope, @@ -1045,7 +1084,8 @@ export class MemoryDriver extends Context.Service()("effect/cluste } return unprocessedWith((envelope) => envelopeIds.has(envelope.requestId)) }), - resetAddress: () => Effect.void, + resetAddress: (address) => resetAddresses([address]), + resetAddresses, clearAddress: (address) => Effect.sync(() => { for (const [primaryKey, entry] of requestsByPrimaryKey) { @@ -1064,11 +1104,21 @@ export class MemoryDriver extends Context.Service()("effect/cluste continue } unprocessed.delete(envelope) + lastRead.delete(envelope) requests.delete(envelope.requestId) journal.splice(i, 1) } }), - resetShards: () => Effect.void, + resetShards: (shardIds) => + Effect.sync(() => { + const shards = new Set(shardIds) + for (const envelope of journal) { + const shardId = ShardId.make(envelope.address.shardId.group, envelope.address.shardId.id) + if (shards.has(shardId.toString())) { + lastRead.delete(envelope) + } + } + }), withTransaction: Effect.provideService(MemoryTransaction, true) } diff --git a/packages/effect/src/unstable/cluster/Sharding.ts b/packages/effect/src/unstable/cluster/Sharding.ts index 643ae96e918..b58d05cf8b2 100644 --- a/packages/effect/src/unstable/cluster/Sharding.ts +++ b/packages/effect/src/unstable/cluster/Sharding.ts @@ -672,8 +672,10 @@ const make = Effect.gen(function*() { if (message._tag === "IncomingEnvelope" && isProcessing) { // If the message might affect a currently processing request, we // send it to the entity manager to be processed. - deliveredThisRead = true - return state.manager.send(message) + return Effect.tap(state.manager.send(message), () => + Effect.sync(() => { + deliveredThisRead = true + })) } else if (isProcessing || state.status === "closing") { // If the request is already processing, we skip it. // Or if the entity is closing, we skip all incoming messages. @@ -704,8 +706,10 @@ const make = Effect.gen(function*() { } return Effect.void } - deliveredThisRead = true - return state.manager.send(message) + return Effect.tap(state.manager.send(message), () => + Effect.sync(() => { + deliveredThisRead = true + })) }), (cause) => { const message = messages[index] @@ -769,28 +773,33 @@ const make = Effect.gen(function*() { // First deliver messages for entities that are already resident, so // they keep making progress even when the runner is at entity // capacity. - const residentAddresses: Array = [] - for (const state of entityManagers.values()) { - if (state.status === "closed") continue - for (const address of state.manager.residentAddressesUnsafe()) { - residentAddresses.push(address) + if (residencyAtCapacityUnsafe()) { + const residentAddresses: Array = [] + for (const state of entityManagers.values()) { + if (state.status === "closed") continue + for (const address of state.manager.residentAddressesUnsafe()) { + residentAddresses.push(address) + } + } + if (residentAddresses.length > 0) { + messages = yield* storage.unprocessedMessages(acquiredShards, { + limit: batchSize, + addresses: residentAddresses + }) + index = 0 + yield* processMessages + readCount = messages.length + fullBatch = messages.length >= batchSize } - } - if (residentAddresses.length > 0) { - messages = yield* storage.unprocessedMessages(acquiredShards, { - limit: batchSize, - addresses: residentAddresses - }) - index = 0 - yield* processMessages - readCount = messages.length - fullBatch = messages.length >= batchSize } // Then walk the remaining messages, spawning new entities while the // runner has entity slots left. if (!residencyAtCapacityUnsafe() && readCount < batchSize) { - const limit = batchSize - readCount + const freeSlots = maxResidentEntities === "unbounded" + ? batchSize + : maxResidentEntities - residentEntityCount + const limit = Math.min(batchSize - readCount, freeSlots) messages = yield* storage.unprocessedMessages(acquiredShards, { limit }) index = 0 yield* processMessages @@ -831,20 +840,18 @@ const make = Effect.gen(function*() { } MutableHashSet.clear(resetAddresses) } - // Release the claims of the messages that were skipped because of the - // entity cap, so they are re-read as soon as a slot frees up. - if (MutableHashSet.size(cappedAddresses) > 0) { - yield* Effect.forEach( - cappedAddresses, - (address) => Effect.ignore(storage.resetAddress(address)), - { discard: true } - ) - MutableHashSet.clear(cappedAddresses) - } + // Capture claims skipped because of the entity cap. They are reset in + // one storage operation after releasing the read lock. + const cappedAddressesToReset = Arr.fromIterable(cappedAddresses) + MutableHashSet.clear(cappedAddresses) // let the resuming entities check if they are done yield* storageReadLock.release(1) + if (cappedAddressesToReset.length > 0) { + yield* Effect.ignore(storage.resetAddresses(cappedAddressesToReset)) + } + // A full batch means more messages could be waiting; start the next // read immediately, as long as this read made progress (a full batch // of skipped messages must not spin the loop). When the runner is at diff --git a/packages/effect/src/unstable/cluster/ShardingConfig.ts b/packages/effect/src/unstable/cluster/ShardingConfig.ts index 15745a2a873..1fab4e7ec7f 100644 --- a/packages/effect/src/unstable/cluster/ShardingConfig.ts +++ b/packages/effect/src/unstable/cluster/ShardingConfig.ts @@ -304,11 +304,17 @@ export const config: Config.Config = Config.all({ Config.withDefault(defaults.entityMailboxCapacity) // Config.withDescription("The default capacity of the mailbox for entities.") ), - maxResidentEntities: Config.int("maxResidentEntities").pipe( + maxResidentEntities: Config.schema( + Schema.Int.check(Schema.isGreaterThan(0)), + "maxResidentEntities" + ).pipe( Config.withDefault(defaults.maxResidentEntities) // Config.withDescription("The maximum number of entities that can be resident on this runner at the same time.") ), - unprocessedMessageBatchSize: Config.int("unprocessedMessageBatchSize").pipe( + unprocessedMessageBatchSize: Config.schema( + Schema.Int.check(Schema.isGreaterThan(0)), + "unprocessedMessageBatchSize" + ).pipe( Config.withDefault(defaults.unprocessedMessageBatchSize) // Config.withDescription("The maximum number of unprocessed messages read from storage in a single poll.") ), diff --git a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts index b7752e489d5..98c590c2c84 100644 --- a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts +++ b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts @@ -430,7 +430,7 @@ export const make: (options?: { AND m.processed = ${sqlFalse} AND (m.last_read IS NULL OR m.last_read < ${tenMinutesAgo}) AND (m.deliver_at IS NULL OR m.deliver_at <= ${sql.literal(String(now))}) - ORDER BY m.rowid ASC + ORDER BY m.id ASC ${filters.limit} FOR UPDATE ) AS ids @@ -438,7 +438,7 @@ export const make: (options?: { WHERE m.id = ids.id RETURNING ids.*, r.id as reply_reply_id, r.kind as reply_kind, r.payload as reply_payload, r.sequence as reply_sequence ) - SELECT * FROM messages ORDER BY rowid ASC + SELECT * FROM messages ORDER BY id ASC ` }, orElse: () => @@ -689,6 +689,28 @@ export const make: (options?: { withTracerDisabled ), + resetAddresses: (addresses) => { + if (addresses.length === 0) return Effect.void + return sql` + UPDATE ${messagesTableSql} + SET last_read = NULL + WHERE processed = ${sqlFalse} + AND (${ + sql.or(addresses.map((address) => + sql.and([ + sql`shard_id = ${address.shardId.toString()}`, + sql`entity_type = ${address.entityType}`, + sql`entity_id = ${address.entityId}` + ]) + )) + }) + `.pipe( + Effect.asVoid, + PersistenceError.refail, + withTracerDisabled + ) + }, + clearAddress: (address) => sql` DELETE FROM ${repliesTableSql} diff --git a/packages/effect/test/cluster/MessageStorage.test.ts b/packages/effect/test/cluster/MessageStorage.test.ts index 94a6eb5680a..43608c32c10 100644 --- a/packages/effect/test/cluster/MessageStorage.test.ts +++ b/packages/effect/test/cluster/MessageStorage.test.ts @@ -97,8 +97,8 @@ describe("MessageStorage", () => { const messages = yield* storage.unprocessedMessages([shardId], { limit: 3 }) expect(messages).toHaveLength(3) expect(messages.map((m: any) => m.envelope.payload.id)).toEqual([1, 2, 3]) - const all = yield* storage.unprocessedMessages([shardId]) - expect(all).toHaveLength(5) + const remaining = yield* storage.unprocessedMessages([shardId]) + expect(remaining.map((m: any) => m.envelope.payload.id)).toEqual([4, 5]) }).pipe(Effect.provide(MemoryLive))) it.effect("unprocessedMessages filters by address", () => @@ -123,6 +123,26 @@ describe("MessageStorage", () => { expect(none).toHaveLength(0) }).pipe(Effect.provide(MemoryLive))) + it.effect("resetAddresses makes claimed messages eligible again", () => + Effect.gen(function*() { + const storage = yield* MessageStorage.MessageStorage + for (let i = 1; i <= 4; i++) { + yield* storage.saveRequest(yield* makeRequest({ payload: { id: i }, entityId: String(i) })) + } + const shardId = ShardId.make("default", 1) + yield* storage.unprocessedMessages([shardId], { limit: 3 }) + yield* storage.resetAddresses(["1", "3"].map((entityId) => + EntityAddress.make({ + shardId, + entityType: EntityType.make("test"), + entityId: EntityId.make(entityId) + }) + )) + + const messages = yield* storage.unprocessedMessages([shardId]) + expect(messages.map((m: any) => m.envelope.payload.id)).toEqual([1, 3, 4]) + }).pipe(Effect.provide(MemoryLive))) + it.effect("repliesFor", () => Effect.gen(function*() { const storage = yield* MessageStorage.MessageStorage diff --git a/packages/effect/test/cluster/Sharding.test.ts b/packages/effect/test/cluster/Sharding.test.ts index 3bf753bef9a..12ab2b95969 100644 --- a/packages/effect/test/cluster/Sharding.test.ts +++ b/packages/effect/test/cluster/Sharding.test.ts @@ -1196,6 +1196,20 @@ describe.concurrent("Sharding residency cap", () => { assert.strictEqual(driver.replyIds.size, 5) }).pipe(Effect.provide(CappedSharding({ unprocessedMessageBatchSize: 2 })))) + it.effect("advances past in-flight memory requests when reading bounded batches", () => + Effect.gen(function*() { + yield* TestClock.adjust(1) + const makeClient = yield* TestEntity.client + yield* makeClient("1").Never().pipe(Effect.forkChild({ startImmediately: true })) + yield* makeClient("2").Never().pipe(Effect.forkChild({ startImmediately: true })) + const fiber = yield* makeClient("3").GetUser({ id: 3 }).pipe( + Effect.forkChild({ startImmediately: true }) + ) + + yield* TestClock.adjust(5000) + assert.deepStrictEqual(yield* Fiber.join(fiber), new User({ id: 3, name: "User 3" })) + }).pipe(Effect.provide(CappedSharding({ unprocessedMessageBatchSize: 2 })))) + it.effect("does not busy-poll storage at the cap", () => Effect.gen(function*() { let reads = 0 diff --git a/packages/effect/test/cluster/ShardingConfig.test.ts b/packages/effect/test/cluster/ShardingConfig.test.ts index 46be4bbaad9..bf8762f39b7 100644 --- a/packages/effect/test/cluster/ShardingConfig.test.ts +++ b/packages/effect/test/cluster/ShardingConfig.test.ts @@ -61,4 +61,19 @@ describe("ShardingConfig", () => { at ["listenPort"]` ) })) + + it.effect("rejects non-positive residency and batch limits", () => + Effect.gen(function*() { + for ( + const values of [ + { maxResidentEntities: "0" }, + { maxResidentEntities: "-1" }, + { unprocessedMessageBatchSize: "0" }, + { unprocessedMessageBatchSize: "-1" } + ] + ) { + const error = yield* ShardingConfig.config.parse(ConfigProvider.fromUnknown(values)).pipe(Effect.flip) + assert.strictEqual(error._tag, "ConfigError") + } + })) }) diff --git a/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts b/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts index 914f79f4292..75a9862e129 100644 --- a/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts +++ b/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts @@ -330,6 +330,28 @@ describe("SqlMessageStorage", () => { expect(rest.map((m: any) => m.envelope.payload.id)).toEqual([1, 3]) })) + it.effect("resetAddresses releases claims in one batch", () => + Effect.gen(function*() { + yield* truncate + + const storage = yield* MessageStorage.MessageStorage + const shardId = ShardId.make("default", 1) + const address = (entityId: string) => + EntityAddress.make({ + shardId, + entityType: EntityType.make("test"), + entityId: EntityId.make(entityId) + }) + for (let i = 1; i <= 4; i++) { + yield* storage.saveRequest(yield* makeRequest({ payload: { id: i }, entityId: String(i) })) + } + yield* storage.unprocessedMessages([shardId], { limit: 3 }) + yield* storage.resetAddresses([address("1"), address("3")]) + + const messages = yield* storage.unprocessedMessages([shardId]) + expect(messages.map((m: any) => m.envelope.payload.id)).toEqual([1, 3, 4]) + })) + it.effect("unprocessedMessages excludes complete requests", () => Effect.gen(function*() { yield* truncate From 42669a30161c0142b47e6963cc92392c67a0924b Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Wed, 12 Aug 2026 22:54:45 +0000 Subject: [PATCH 3/9] fix(cluster): preserve saturated read throughput --- .changeset/wild-donuts-brake.md | 4 +- .../effect/src/unstable/cluster/Sharding.ts | 7 +- .../src/unstable/cluster/SqlMessageStorage.ts | 18 ++++- packages/effect/test/cluster/Sharding.test.ts | 68 +++++++++++++++++++ .../SqlMessageStorage.integration.test.ts | 14 ++++ 5 files changed, 103 insertions(+), 8 deletions(-) diff --git a/.changeset/wild-donuts-brake.md b/.changeset/wild-donuts-brake.md index b341ea8b2b5..9abd31aab64 100644 --- a/.changeset/wild-donuts-brake.md +++ b/.changeset/wild-donuts-brake.md @@ -17,4 +17,6 @@ Bound cluster runner entity residency and storage reads. `MessageStorage.unprocessedMessages` accepts an optional `{ limit, addresses }` argument, and only claims the messages it actually -returns. +returns. The memory implementation now applies the same ten-minute claim +window as SQL, so bounded reads advance past in-flight requests; resetting an +address or shard makes its claimed messages immediately eligible again. diff --git a/packages/effect/src/unstable/cluster/Sharding.ts b/packages/effect/src/unstable/cluster/Sharding.ts index b58d05cf8b2..7fe93916168 100644 --- a/packages/effect/src/unstable/cluster/Sharding.ts +++ b/packages/effect/src/unstable/cluster/Sharding.ts @@ -744,7 +744,7 @@ const make = Effect.gen(function*() { } ) - const batchSize = config.unprocessedMessageBatchSize + const batchSize = Math.max(1, config.unprocessedMessageBatchSize) while (true) { // wait for the next poll interval, or if we get notified of a change @@ -796,10 +796,7 @@ const make = Effect.gen(function*() { // Then walk the remaining messages, spawning new entities while the // runner has entity slots left. if (!residencyAtCapacityUnsafe() && readCount < batchSize) { - const freeSlots = maxResidentEntities === "unbounded" - ? batchSize - : maxResidentEntities - residentEntityCount - const limit = Math.min(batchSize - readCount, freeSlots) + const limit = batchSize - readCount messages = yield* storage.unprocessedMessages(acquiredShards, { limit }) index = 0 yield* processMessages diff --git a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts index 98c590c2c84..0086403d177 100644 --- a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts +++ b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts @@ -430,7 +430,7 @@ export const make: (options?: { AND m.processed = ${sqlFalse} AND (m.last_read IS NULL OR m.last_read < ${tenMinutesAgo}) AND (m.deliver_at IS NULL OR m.deliver_at <= ${sql.literal(String(now))}) - ORDER BY m.id ASC + ORDER BY m.rowid ASC ${filters.limit} FOR UPDATE ) AS ids @@ -438,7 +438,7 @@ export const make: (options?: { WHERE m.id = ids.id RETURNING ids.*, r.id as reply_reply_id, r.kind as reply_kind, r.payload as reply_payload, r.sequence as reply_sequence ) - SELECT * FROM messages ORDER BY id ASC + SELECT * FROM messages ORDER BY rowid ASC ` }, orElse: () => @@ -1099,6 +1099,20 @@ const migrations = (options?: { // sqlite Effect.void }) + }), + "0003_pg_messages_rowid_index": Effect.gen(function*() { + const sql = (yield* SqlClient.SqlClient).withoutTransforms() + const messagesTableSql = sql(messagesTable) + const rowIdIndex = `${messagesTable}_rowid_idx` + + yield* sql.onDialectOrElse({ + pg: () => + sql` + CREATE INDEX IF NOT EXISTS ${sql(rowIdIndex)} + ON ${messagesTableSql} (rowid) + `, + orElse: () => Effect.void + }) }) }) } diff --git a/packages/effect/test/cluster/Sharding.test.ts b/packages/effect/test/cluster/Sharding.test.ts index 12ab2b95969..cd61630e126 100644 --- a/packages/effect/test/cluster/Sharding.test.ts +++ b/packages/effect/test/cluster/Sharding.test.ts @@ -1210,6 +1210,74 @@ describe.concurrent("Sharding residency cap", () => { assert.deepStrictEqual(yield* Fiber.join(fiber), new User({ id: 3, name: "User 3" })) }).pipe(Effect.provide(CappedSharding({ unprocessedMessageBatchSize: 2 })))) + it.effect("keeps full-sized storage reads when approaching the entity cap", () => + Effect.gen(function*() { + const limits: Array = [] + yield* Effect.gen(function*() { + yield* TestClock.adjust(1) + const sharding = yield* Sharding.Sharding + const makeClient = yield* TestEntity.client + for (let i = 1; i <= 4; i++) { + yield* makeClient(String(i)).NeverVolatile().pipe(Effect.forkChild({ startImmediately: true })) + } + yield* TestClock.adjust(1) + assert.strictEqual(yield* sharding.activeEntityCount, 4) + limits.length = 0 + + const storage = yield* MessageStorage.MessageStorage + const rpc = TestEntity.protocol.requests.get("GetUser")! as any + for (let i = 0; i < 10; i++) { + const entityId = EntityId.make(String(i % 4 + 1)) + yield* storage.saveRequest( + new Message.OutgoingRequest({ + envelope: Envelope.makeRequest({ + requestId: yield* sharding.getSnowflake, + address: EntityAddress.make({ + shardId: sharding.getShardId(entityId, "default"), + entityType: EntityType.make(TestEntity.type), + entityId + }), + tag: "GetUser", + payload: { id: i }, + headers: Headers.empty + }), + annotations: rpc.annotations, + context: Context.empty() as any, + rpc, + lastReceivedReply: Option.none(), + respond: () => Effect.void + }) + ) + } + yield* TestClock.adjust(5000) + assert.isNotEmpty(limits) + assert.deepStrictEqual([...new Set(limits)], [100]) + }).pipe(Effect.provide(CappedSharding({ + maxResidentEntities: 5, + unprocessedMessageBatchSize: 100 + }, (storage) => ({ + ...storage, + unprocessedMessages(shardIds, options) { + if (options?.limit !== undefined) limits.push(options.limit) + return storage.unprocessedMessages(shardIds, options) + } + })))) + })) + + it.effect("normalizes programmatic batch sizes below one", () => + Effect.forEach([0, -1], (unprocessedMessageBatchSize) => + Effect.gen(function*() { + yield* TestClock.adjust(1) + const makeClient = yield* TestEntity.client + const fiber = yield* makeClient("1").GetUser({ id: 1 }).pipe( + Effect.forkChild({ startImmediately: true }) + ) + + yield* TestClock.adjust(5000) + assert.isDefined(fiber.pollUnsafe()) + assert.deepStrictEqual(yield* Fiber.join(fiber), new User({ id: 1, name: "User 1" })) + }).pipe(Effect.provide(CappedSharding({ unprocessedMessageBatchSize }))), { discard: true })) + it.effect("does not busy-poll storage at the cap", () => Effect.gen(function*() { let reads = 0 diff --git a/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts b/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts index 75a9862e129..bfbf7c56b9d 100644 --- a/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts +++ b/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts @@ -59,6 +59,20 @@ describe("SqlMessageStorage", () => { it.layer(StorageLive.pipe(Layer.provideMerge(layer)), { timeout: 120000 })(label, (it) => { + if (label === "pg") { + it.effect("creates an index for insertion-ordered message reads", () => + Effect.gen(function*() { + const sql = yield* SqlClient.SqlClient + const indexes = yield* sql<{ indexname: string }>` + SELECT indexname + FROM pg_indexes + WHERE tablename = 'cluster_messages' + AND indexname = 'cluster_messages_rowid_idx' + ` + expect(indexes).toHaveLength(1) + })) + } + it.effect("saveRequest", () => Effect.gen(function*() { const storage = yield* MessageStorage.MessageStorage From 5b4fa58171bc8884e0bcf037d6da90706a0f2069 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Wed, 12 Aug 2026 23:27:23 +0000 Subject: [PATCH 4/9] refactor(cluster): simplify residency admission internals Simplicity pass over the admission-control change, no behavior changes: - Derive the capped state from cappedAddresses instead of a parallel boolean, and merge the notification sweep into one guarded block. - Read-and-process helper for the two storage read steps; hoisted markDelivered effect instead of per-message closures. - Drop resetAddress from the encoded storage driver contract; drivers implement only the batched resetAddresses and the singular form is derived in makeEncoded. Group the SQL bulk reset with IN lists. - One shared UnprocessedOptions type (ReadonlyArray) for the decoded and encoded storage APIs, removing the conversion glue in makeEncoded. - Hoist option handling out of the per-dialect query builders; reuse Arr.groupBy for the address filter. - Memory driver uses one addressKey helper and a named claim-window constant; entity reaper no longer depends on the spawn error union. - Shared test config defaults and a saveGetUserRequest helper replace duplicated envelope plumbing in Sharding.test.ts. Co-authored-by: Claude Fable 5 --- .../src/unstable/cluster/MessageStorage.ts | 55 +++------ .../effect/src/unstable/cluster/Sharding.ts | 79 ++++++------- .../src/unstable/cluster/SqlMessageStorage.ts | 108 +++++++----------- .../unstable/cluster/internal/entityReaper.ts | 5 +- .../unstable/cluster/internal/resourceMap.ts | 6 +- packages/effect/test/cluster/Sharding.test.ts | 95 ++++++--------- 6 files changed, 128 insertions(+), 220 deletions(-) diff --git a/packages/effect/src/unstable/cluster/MessageStorage.ts b/packages/effect/src/unstable/cluster/MessageStorage.ts index 6159a4636fa..89239329ef3 100644 --- a/packages/effect/src/unstable/cluster/MessageStorage.ts +++ b/packages/effect/src/unstable/cluster/MessageStorage.ts @@ -167,7 +167,7 @@ export class MessageStorage extends Context.Service + addresses: ReadonlyArray ) => Effect.Effect /** @@ -199,7 +199,7 @@ export class MessageStorage extends Context.Service | undefined + readonly addresses?: ReadonlyArray | undefined } /** @@ -388,7 +388,7 @@ export type Encoded = { readonly unprocessedMessages: ( shardIds: Arr.NonEmptyArray, now: number, - options?: EncodedUnprocessedMessagesOptions | undefined + options?: UnprocessedOptions | undefined ) => Effect.Effect< Array<{ readonly envelope: Envelope.Encoded @@ -411,13 +411,6 @@ export type Encoded = { PersistenceError > - /** - * Reset the mailbox state for the provided address. - */ - readonly resetAddress: ( - address: EntityAddress - ) => Effect.Effect - /** * Reset the mailbox state for the provided addresses. */ @@ -447,17 +440,6 @@ export type Encoded = { ) => Effect.Effect } -/** - * Options for reading encoded unprocessed messages from a storage driver. - * - * @category models - * @since 4.0.0 - */ -export type EncodedUnprocessedMessagesOptions = { - readonly limit?: number | undefined - readonly addresses?: ReadonlyArray | undefined -} - /** * Cursor options for reading encoded unprocessed messages across shard sets. * @@ -687,14 +669,9 @@ export const makeEncoded: (encoded: Encoded) => Effect.Effect< const storage = this as MessageStorage["Service"] const shards = Array.from(shardIds, (id) => id.toString()) if (!Arr.isArrayNonEmpty(shards)) return Effect.succeed([]) - let encodedOptions: EncodedUnprocessedMessagesOptions | undefined - if (options) { - const addresses = options.addresses && Array.from(options.addresses) - if (addresses && addresses.length === 0) return Effect.succeed([]) - encodedOptions = { limit: options.limit, addresses } - } + if (options?.addresses !== undefined && options.addresses.length === 0) return Effect.succeed([]) return Effect.flatMap( - Effect.suspend(() => encoded.unprocessedMessages(shards, clock.currentTimeMillisUnsafe(), encodedOptions)), + Effect.suspend(() => encoded.unprocessedMessages(shards, clock.currentTimeMillisUnsafe(), options)), (messages) => decodeMessages(storage, messages) ) }, @@ -707,11 +684,8 @@ export const makeEncoded: (encoded: Encoded) => Effect.Effect< (messages) => decodeMessages(storage, messages) ) }, - resetAddress: encoded.resetAddress, - resetAddresses: (addresses) => { - const array = Array.from(addresses) - return array.length === 0 ? Effect.void : encoded.resetAddresses(array) - }, + resetAddress: (address) => encoded.resetAddresses([address]), + resetAddresses: (addresses) => addresses.length === 0 ? Effect.void : encoded.resetAddresses(addresses), clearAddress: encoded.clearAddress, resetShards: (shardIds) => { const shards = Array.from(shardIds, (id) => id.toString()) @@ -874,6 +848,9 @@ export const MemoryTransaction = Context.Reference("effect/cluster/Mess defaultValue: constFalse }) +// the same claim window the SQL driver uses (`last_read < ten minutes ago`) +const claimExpirationMillis = 10 * 60 * 1000 + /** * Service that provides an in-memory message storage driver with inspectable backing state. * @@ -1032,12 +1009,7 @@ export class MemoryDriver extends Context.Service()("effect/cluste Effect.sync(() => { if (unprocessed.size === 0) return [] const limit = options?.limit ?? Infinity - let addressFilter: Set | undefined - if (options?.addresses) { - addressFilter = new Set( - options.addresses.map((address) => `${address.entityType}/${address.entityId}`) - ) - } + const addressFilter = options?.addresses && new Set(options.addresses.map(addressKey)) const messages = Arr.empty<{ envelope: Envelope.Encoded lastSentReply: Option.Option @@ -1049,7 +1021,7 @@ export class MemoryDriver extends Context.Service()("effect/cluste if (!unprocessed.has(envelope as any) || !shardIds.includes(shardId.toString())) { continue } - if (addressFilter && !addressFilter.has(`${envelope.address.entityType}/${envelope.address.entityId}`)) { + if (addressFilter && !addressFilter.has(addressKey(envelope.address))) { continue } if (envelope._tag === "Request") { @@ -1058,7 +1030,7 @@ export class MemoryDriver extends Context.Service()("effect/cluste continue } const claimedAt = lastRead.get(envelope) - if (claimedAt !== undefined && claimedAt > now - 10 * 60 * 1000) { + if (claimedAt !== undefined && claimedAt > now - claimExpirationMillis) { continue } messages.push({ @@ -1084,7 +1056,6 @@ export class MemoryDriver extends Context.Service()("effect/cluste } return unprocessedWith((envelope) => envelopeIds.has(envelope.requestId)) }), - resetAddress: (address) => resetAddresses([address]), resetAddresses, clearAddress: (address) => Effect.sync(() => { diff --git a/packages/effect/src/unstable/cluster/Sharding.ts b/packages/effect/src/unstable/cluster/Sharding.ts index 7fe93916168..47c12f61b59 100644 --- a/packages/effect/src/unstable/cluster/Sharding.ts +++ b/packages/effect/src/unstable/cluster/Sharding.ts @@ -627,18 +627,28 @@ const make = Effect.gen(function*() { let index = 0 let messages: Array> = [] - let cappedThisRead = false let deliveredThisRead = false const removableNotifications = new Set() const resetAddresses = MutableHashSet.empty() const cappedAddresses = MutableHashSet.empty() + const markDelivered = Effect.sync(() => { + deliveredThisRead = true + }) + const processMessages = Effect.whileLoop({ while: () => index < messages.length, step: () => index++, body: () => send }) + const readAndProcess = Effect.fnUntraced(function*(options?: MessageStorage.UnprocessedOptions) { + messages = yield* storage.unprocessedMessages(acquiredShards, options) + index = 0 + yield* processMessages + return messages.length + }) + const send = Effect.catchCause( Effect.suspend(() => { const message = messages[index] @@ -672,10 +682,7 @@ const make = Effect.gen(function*() { if (message._tag === "IncomingEnvelope" && isProcessing) { // If the message might affect a currently processing request, we // send it to the entity manager to be processed. - return Effect.tap(state.manager.send(message), () => - Effect.sync(() => { - deliveredThisRead = true - })) + return Effect.tap(state.manager.send(message), markDelivered) } else if (isProcessing || state.status === "closing") { // If the request is already processing, we skip it. // Or if the entity is closing, we skip all incoming messages. @@ -691,7 +698,6 @@ const make = Effect.gen(function*() { // are not already resident are skipped. Their claims are released // after the read, so they stay eligible for once a slot frees up. if (residencyAtCapacityUnsafe() && !state.manager.isResidentUnsafe(address)) { - cappedThisRead = true MutableHashSet.add(cappedAddresses, address) return Effect.void } @@ -706,10 +712,7 @@ const make = Effect.gen(function*() { } return Effect.void } - return Effect.tap(state.manager.send(message), () => - Effect.sync(() => { - deliveredThisRead = true - })) + return Effect.tap(state.manager.send(message), markDelivered) }), (cause) => { const message = messages[index] @@ -736,7 +739,6 @@ const make = Effect.gen(function*() { } // Otherwise the runner ran out of entity slots while the message // was in flight; leave it in storage. - cappedThisRead = true MutableHashSet.add(cappedAddresses, address) return Effect.void } @@ -764,7 +766,6 @@ const make = Effect.gen(function*() { pendingNotifications.forEach((entry) => removableNotifications.add(entry)) } - cappedThisRead = false deliveredThisRead = false let readCount = 0 let fullBatch = false @@ -782,14 +783,8 @@ const make = Effect.gen(function*() { } } if (residentAddresses.length > 0) { - messages = yield* storage.unprocessedMessages(acquiredShards, { - limit: batchSize, - addresses: residentAddresses - }) - index = 0 - yield* processMessages - readCount = messages.length - fullBatch = messages.length >= batchSize + readCount = yield* readAndProcess({ limit: batchSize, addresses: residentAddresses }) + fullBatch = readCount >= batchSize } } @@ -797,35 +792,32 @@ const make = Effect.gen(function*() { // runner has entity slots left. if (!residencyAtCapacityUnsafe() && readCount < batchSize) { const limit = batchSize - readCount - messages = yield* storage.unprocessedMessages(acquiredShards, { limit }) - index = 0 - yield* processMessages - if (messages.length >= limit) { + const read = yield* readAndProcess({ limit }) + if (read >= limit) { fullBatch = true - } else if (!cappedThisRead) { + } else if (MutableHashSet.size(cappedAddresses) === 0) { exhaustive = true } } if (removableNotifications.size > 0) { - if (exhaustive) { - // The whole backlog was read, so a message that was not seen is - // no longer processable by this runner. + // On an exhaustive read, a message that was not seen is no longer + // processable by this runner. At the entity cap, the messages are + // safely persisted and delivered once a slot frees up, so persisted + // senders succeed instead of failing. Otherwise the read was + // truncated by the batch size and the notifications stay registered + // for the immediately following read. + const capped = MutableHashSet.size(cappedAddresses) > 0 || residencyAtCapacityUnsafe() + if (exhaustive || capped) { removableNotifications.forEach(({ message, resume }) => { pendingNotifications.delete(message.envelope.requestId) - resume(Effect.fail(new EntityNotAssignedToRunner({ address: message.envelope.address }))) - }) - } else if (cappedThisRead || residencyAtCapacityUnsafe()) { - // The runner is at entity capacity. The messages are safely - // persisted and are delivered once a slot frees up, so persisted - // senders succeed instead of failing at the cap. - removableNotifications.forEach(({ message, resume }) => { - pendingNotifications.delete(message.envelope.requestId) - resume(Effect.void) + resume( + exhaustive + ? Effect.fail(new EntityNotAssignedToRunner({ address: message.envelope.address })) + : Effect.void + ) }) } - // Otherwise the read was truncated by the batch size; the - // notifications stay registered for the immediately following read. removableNotifications.clear() } if (MutableHashSet.size(resetAddresses) > 0) { @@ -839,13 +831,16 @@ const make = Effect.gen(function*() { } // Capture claims skipped because of the entity cap. They are reset in // one storage operation after releasing the read lock. - const cappedAddressesToReset = Arr.fromIterable(cappedAddresses) - MutableHashSet.clear(cappedAddresses) + let cappedAddressesToReset: Array | undefined + if (MutableHashSet.size(cappedAddresses) > 0) { + cappedAddressesToReset = Arr.fromIterable(cappedAddresses) + MutableHashSet.clear(cappedAddresses) + } // let the resuming entities check if they are done yield* storageReadLock.release(1) - if (cappedAddressesToReset.length > 0) { + if (cappedAddressesToReset !== undefined) { yield* Effect.ignore(storage.resetAddresses(cappedAddressesToReset)) } diff --git a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts index 0086403d177..a13fbe08fce 100644 --- a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts +++ b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts @@ -379,41 +379,27 @@ export const make: (options?: { mssql: () => (limit: number) => sql.literal(`OFFSET 0 ROWS FETCH NEXT ${Math.floor(limit)} ROWS ONLY`), orElse: () => (limit: number) => sql.literal(`LIMIT ${Math.floor(limit)}`) }) - const unprocessedFilters = (options?: MessageStorage.EncodedUnprocessedMessagesOptions | undefined) => { - let addressFilter = emptyFragment - if (options?.addresses !== undefined) { - const idsByType = new Map>() - for (const address of options.addresses) { - let ids = idsByType.get(address.entityType) - if (ids === undefined) { - ids = [] - idsByType.set(address.entityType, ids) - } - ids.push(address.entityId) - } - addressFilter = sql`AND (${ - sql.or(Array.from(idsByType, ([entityType, entityIds]) => - sql.and([ - sql`m.entity_type = ${entityType}`, - sql.in("m.entity_id", entityIds) - ]))) + const unprocessedFilters = (options?: MessageStorage.UnprocessedOptions | undefined) => ({ + addressFilter: options?.addresses !== undefined + ? sql`AND (${ + sql.or( + Object.entries(Arr.groupBy(options.addresses, (address) => address.entityType)).map( + ([entityType, group]) => + sql.and([ + sql`m.entity_type = ${entityType}`, + sql.in("m.entity_id", group.map((address) => address.entityId)) + ]) + ) + ) })` - } - return { - addressFilter, - limit: options?.limit !== undefined ? limitFragment(options.limit) : emptyFragment - } - } + : emptyFragment, + limit: options?.limit !== undefined ? limitFragment(options.limit) : emptyFragment + }) + type UnprocessedFilters = ReturnType - const getUnprocessedMessages = sql.onDialectOrElse({ - pg: () => - ( - shardIds: ReadonlyArray, - now: number, - options?: MessageStorage.EncodedUnprocessedMessagesOptions | undefined - ) => { - const filters = unprocessedFilters(options) - return sql` + const getUnprocessedMessagesForDialect = sql.onDialectOrElse({ + pg: () => (shardIds: ReadonlyArray, now: number, filters: UnprocessedFilters) => + sql` WITH messages AS ( UPDATE ${messagesTableSql} m SET last_read = ${sqlNow} @@ -439,16 +425,9 @@ export const make: (options?: { RETURNING ids.*, r.id as reply_reply_id, r.kind as reply_kind, r.payload as reply_payload, r.sequence as reply_sequence ) SELECT * FROM messages ORDER BY rowid ASC - ` - }, - orElse: () => - ( - shardIds: ReadonlyArray, - now: number, - options?: MessageStorage.EncodedUnprocessedMessagesOptions | undefined - ) => { - const filters = unprocessedFilters(options) - return sql` + `, + orElse: () => (shardIds: ReadonlyArray, now: number, filters: UnprocessedFilters) => + sql` SELECT m.*, r.id as reply_reply_id, r.kind as reply_kind, r.payload as reply_payload, r.sequence as reply_sequence FROM ${messagesTableSql} m LEFT JOIN ${repliesTableSql} r ON r.id = m.last_reply_id @@ -478,8 +457,12 @@ export const make: (options?: { }), sql.withTransaction ) - } }) + const getUnprocessedMessages = ( + shardIds: ReadonlyArray, + now: number, + options?: MessageStorage.UnprocessedOptions | undefined + ) => getUnprocessedMessagesForDialect(shardIds, now, unprocessedFilters(options)) return yield* MessageStorage.makeEncoded({ saveEnvelope: ({ deliverAt, envelope, primaryKey }) => @@ -675,41 +658,28 @@ export const make: (options?: { ) }, - resetAddress: (address) => + resetAddresses: (addresses) => sql` - UPDATE ${messagesTableSql} - SET last_read = NULL - WHERE processed = ${sqlFalse} - AND shard_id = ${address.shardId.toString()} - AND entity_type = ${address.entityType} - AND entity_id = ${address.entityId} - `.pipe( - Effect.asVoid, - PersistenceError.refail, - withTracerDisabled - ), - - resetAddresses: (addresses) => { - if (addresses.length === 0) return Effect.void - return sql` UPDATE ${messagesTableSql} SET last_read = NULL WHERE processed = ${sqlFalse} AND (${ - sql.or(addresses.map((address) => - sql.and([ - sql`shard_id = ${address.shardId.toString()}`, - sql`entity_type = ${address.entityType}`, - sql`entity_id = ${address.entityId}` - ]) - )) + sql.or( + Object.values(Arr.groupBy(addresses, (address) => `${address.shardId}/${address.entityType}`)).map( + (group) => + sql.and([ + sql`shard_id = ${group[0].shardId.toString()}`, + sql`entity_type = ${group[0].entityType}`, + sql.in("entity_id", group.map((address) => address.entityId)) + ]) + ) + ) }) `.pipe( Effect.asVoid, PersistenceError.refail, withTracerDisabled - ) - }, + ), clearAddress: (address) => sql` diff --git a/packages/effect/src/unstable/cluster/internal/entityReaper.ts b/packages/effect/src/unstable/cluster/internal/entityReaper.ts index 5037b46c9ae..d50c602fc24 100644 --- a/packages/effect/src/unstable/cluster/internal/entityReaper.ts +++ b/packages/effect/src/unstable/cluster/internal/entityReaper.ts @@ -3,7 +3,6 @@ import * as Context from "../../../Context.ts" import * as Effect from "../../../Effect.ts" import * as Latch from "../../../Latch.ts" import * as Layer from "../../../Layer.ts" -import type { EntityNotAssignedToRunner, MailboxFull } from "../ClusterError.ts" import type { EntityAddress } from "../EntityAddress.ts" import type { EntityId } from "../EntityId.ts" import type { EntityState } from "./entityManager.ts" @@ -16,14 +15,14 @@ export class EntityReaper extends Context.Service()("effect/cluste const registered: Array<{ readonly maxIdleTime: number readonly servers: Map - readonly entities: ResourceMap + readonly entities: ResourceMap }> = [] const latch = yield* Latch.make() const register = (options: { readonly maxIdleTime: number readonly servers: Map - readonly entities: ResourceMap + readonly entities: ResourceMap }) => Effect.suspend(() => { currentResolution = Math.max(Math.min(currentResolution, options.maxIdleTime), 5000) diff --git a/packages/effect/src/unstable/cluster/internal/resourceMap.ts b/packages/effect/src/unstable/cluster/internal/resourceMap.ts index 938056653ee..e09e1f67b41 100644 --- a/packages/effect/src/unstable/cluster/internal/resourceMap.ts +++ b/packages/effect/src/unstable/cluster/internal/resourceMap.ts @@ -61,11 +61,7 @@ export class ResourceMap { } keysUnsafe(): Array { - const keys: Array = [] - for (const [key] of this.entries.map) { - keys.push(key) - } - return keys + return Array.from(this.entries.map, ([key]) => key) } get(key: K): Effect.Effect { diff --git a/packages/effect/test/cluster/Sharding.test.ts b/packages/effect/test/cluster/Sharding.test.ts index cd61630e126..f7ed7ec6eb5 100644 --- a/packages/effect/test/cluster/Sharding.test.ts +++ b/packages/effect/test/cluster/Sharding.test.ts @@ -1160,34 +1160,9 @@ describe.concurrent("Sharding residency cap", () => { Effect.gen(function*() { yield* TestClock.adjust(1) const driver = yield* MessageStorage.MemoryDriver - const storage = yield* MessageStorage.MessageStorage - const sharding = yield* Sharding.Sharding - const rpc = TestEntity.protocol.requests.get("GetUser")! as any - // save requests directly to storage, so the read loop only learns - // about them from the next poll for (let i = 1; i <= 5; i++) { - const entityId = EntityId.make(String(i)) - yield* storage.saveRequest( - new Message.OutgoingRequest({ - envelope: Envelope.makeRequest({ - requestId: yield* sharding.getSnowflake, - address: EntityAddress.make({ - shardId: sharding.getShardId(entityId, "default"), - entityType: EntityType.make(TestEntity.type), - entityId - }), - tag: "GetUser", - payload: { id: i }, - headers: Headers.empty - }), - annotations: rpc.annotations, - context: Context.empty() as any, - rpc, - lastReceivedReply: Option.none(), - respond: () => Effect.void - }) - ) + yield* saveGetUserRequest(String(i), i) } assert.strictEqual(driver.replyIds.size, 0) @@ -1224,30 +1199,8 @@ describe.concurrent("Sharding residency cap", () => { assert.strictEqual(yield* sharding.activeEntityCount, 4) limits.length = 0 - const storage = yield* MessageStorage.MessageStorage - const rpc = TestEntity.protocol.requests.get("GetUser")! as any for (let i = 0; i < 10; i++) { - const entityId = EntityId.make(String(i % 4 + 1)) - yield* storage.saveRequest( - new Message.OutgoingRequest({ - envelope: Envelope.makeRequest({ - requestId: yield* sharding.getSnowflake, - address: EntityAddress.make({ - shardId: sharding.getShardId(entityId, "default"), - entityType: EntityType.make(TestEntity.type), - entityId - }), - tag: "GetUser", - payload: { id: i }, - headers: Headers.empty - }), - annotations: rpc.annotations, - context: Context.empty() as any, - rpc, - lastReceivedReply: Option.none(), - respond: () => Effect.void - }) - ) + yield* saveGetUserRequest(String(i % 4 + 1), i) } yield* TestClock.adjust(5000) assert.isNotEmpty(limits) @@ -1707,13 +1660,15 @@ const otherRunner = Runner.make({ weight: 1 }) -const TestShardingConfig = ShardingConfig.layer({ +const testConfigDefaults: Partial = { entityMailboxCapacity: 10, entityTerminationTimeout: 0, entityMessagePollInterval: 5000, sendRetryInterval: 100, refreshAssignmentsInterval: 0 -}) +} + +const TestShardingConfig = ShardingConfig.layer(testConfigDefaults) const TestShardingWithoutState = TestEntityNoState.pipe( Layer.provideMerge(Sharding.layer), @@ -1743,14 +1698,7 @@ const CappedSharding = ( storage: MessageStorage.MessageStorage["Service"] ) => MessageStorage.MessageStorage["Service"] ) => { - const configLayer = ShardingConfig.layer({ - entityMailboxCapacity: 10, - entityTerminationTimeout: 0, - entityMessagePollInterval: 5000, - sendRetryInterval: 100, - refreshAssignmentsInterval: 0, - ...config - }) + const configLayer = ShardingConfig.layer({ ...testConfigDefaults, ...config }) let layer = TestShardingWithoutRunners.pipe( Layer.provide(Runners.layerNoop), Layer.provide(configLayer) @@ -1765,3 +1713,32 @@ const CappedSharding = ( } const ContextBleedSharding = ContextBleedLayer.pipe(Layer.provideMerge(TestSharding)) + +// saves a persisted GetUser request directly to storage, bypassing the client, +// so the storage read loop only learns about it from the next poll +const saveGetUserRequest = Effect.fnUntraced(function*(entityId: string, id: number) { + const storage = yield* MessageStorage.MessageStorage + const sharding = yield* Sharding.Sharding + const rpc = TestEntity.protocol.requests.get("GetUser")! as any + const entity = EntityId.make(entityId) + yield* storage.saveRequest( + new Message.OutgoingRequest({ + envelope: Envelope.makeRequest({ + requestId: yield* sharding.getSnowflake, + address: EntityAddress.make({ + shardId: sharding.getShardId(entity, "default"), + entityType: EntityType.make(TestEntity.type), + entityId: entity + }), + tag: "GetUser", + payload: { id }, + headers: Headers.empty + }), + annotations: rpc.annotations, + context: Context.empty() as any, + rpc, + lastReceivedReply: Option.none(), + respond: () => Effect.void + }) + ) +}) From 7a049e00725877adbf2d132f375ef10edc73e2bb Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Wed, 12 Aug 2026 23:47:51 +0000 Subject: [PATCH 5/9] Fix empty encoded address filters --- .changeset/wild-donuts-brake.md | 3 ++ .../src/unstable/cluster/MessageStorage.ts | 18 ++++--- .../src/unstable/cluster/SqlMessageStorage.ts | 39 ++++++++------ .../test/cluster/MessageStorage.test.ts | 53 +++++++++++++++---- 4 files changed, 79 insertions(+), 34 deletions(-) diff --git a/.changeset/wild-donuts-brake.md b/.changeset/wild-donuts-brake.md index 9abd31aab64..755cc716a6a 100644 --- a/.changeset/wild-donuts-brake.md +++ b/.changeset/wild-donuts-brake.md @@ -20,3 +20,6 @@ Bound cluster runner entity residency and storage reads. returns. The memory implementation now applies the same ten-minute claim window as SQL, so bounded reads advance past in-flight requests; resetting an address or shard makes its claimed messages immediately eligible again. + +The encoded driver contract replaces `Encoded.resetAddress` with the batched +`Encoded.resetAddresses` operation. diff --git a/packages/effect/src/unstable/cluster/MessageStorage.ts b/packages/effect/src/unstable/cluster/MessageStorage.ts index 89239329ef3..5f49730f3dc 100644 --- a/packages/effect/src/unstable/cluster/MessageStorage.ts +++ b/packages/effect/src/unstable/cluster/MessageStorage.ts @@ -878,14 +878,16 @@ export class MemoryDriver extends Context.Service()("effect/cluste `${address.shardId.group}/${address.shardId.id}/${address.entityType}/${address.entityId}` const resetAddresses = (addresses: ReadonlyArray) => - Effect.sync(() => { - const keys = new Set(addresses.map(addressKey)) - for (const envelope of journal) { - if (keys.has(addressKey(envelope.address))) { - lastRead.delete(envelope) + addresses.length === 0 + ? Effect.void + : Effect.sync(() => { + const keys = new Set(addresses.map(addressKey)) + for (const envelope of journal) { + if (keys.has(addressKey(envelope.address))) { + lastRead.delete(envelope) + } } - } - }) + }) const cursors = new WeakMap<{}, number>() @@ -1006,7 +1008,7 @@ export class MemoryDriver extends Context.Service()("effect/cluste repliesForUnfiltered: (requestIds) => Effect.sync(() => requestIds.flatMap((id) => requests.get(String(id))?.replies ?? [])), unprocessedMessages: (shardIds, now, options) => - Effect.sync(() => { + options?.addresses?.length === 0 ? Effect.succeed([]) : Effect.sync(() => { if (unprocessed.size === 0) return [] const limit = options?.limit ?? Infinity const addressFilter = options?.addresses && new Set(options.addresses.map(addressKey)) diff --git a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts index a13fbe08fce..d3870ea9c4f 100644 --- a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts +++ b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts @@ -462,7 +462,10 @@ export const make: (options?: { shardIds: ReadonlyArray, now: number, options?: MessageStorage.UnprocessedOptions | undefined - ) => getUnprocessedMessagesForDialect(shardIds, now, unprocessedFilters(options)) + ) => + options?.addresses?.length === 0 + ? Effect.succeed([]) + : getUnprocessedMessagesForDialect(shardIds, now, unprocessedFilters(options)) return yield* MessageStorage.makeEncoded({ saveEnvelope: ({ deliverAt, envelope, primaryKey }) => @@ -659,27 +662,29 @@ export const make: (options?: { }, resetAddresses: (addresses) => - sql` + addresses.length === 0 + ? Effect.void + : sql` UPDATE ${messagesTableSql} SET last_read = NULL WHERE processed = ${sqlFalse} AND (${ - sql.or( - Object.values(Arr.groupBy(addresses, (address) => `${address.shardId}/${address.entityType}`)).map( - (group) => - sql.and([ - sql`shard_id = ${group[0].shardId.toString()}`, - sql`entity_type = ${group[0].entityType}`, - sql.in("entity_id", group.map((address) => address.entityId)) - ]) + sql.or( + Object.values(Arr.groupBy(addresses, (address) => `${address.shardId}/${address.entityType}`)).map( + (group) => + sql.and([ + sql`shard_id = ${group[0].shardId.toString()}`, + sql`entity_type = ${group[0].entityType}`, + sql.in("entity_id", group.map((address) => address.entityId)) + ]) + ) ) - ) - }) - `.pipe( - Effect.asVoid, - PersistenceError.refail, - withTracerDisabled - ), + }) + `.pipe( + Effect.asVoid, + PersistenceError.refail, + withTracerDisabled + ), clearAddress: (address) => sql` diff --git a/packages/effect/test/cluster/MessageStorage.test.ts b/packages/effect/test/cluster/MessageStorage.test.ts index 43608c32c10..b1305b3ef4e 100644 --- a/packages/effect/test/cluster/MessageStorage.test.ts +++ b/packages/effect/test/cluster/MessageStorage.test.ts @@ -16,7 +16,7 @@ import { import { Headers } from "effect/unstable/http" import { Rpc, RpcSchema } from "effect/unstable/rpc" -const MemoryLive = MessageStorage.layerMemory.pipe( +const MemoryLayer = MessageStorage.layerMemory.pipe( Layer.provideMerge(Snowflake.layerGenerator), Layer.provide(ShardingConfig.layerDefaults) ) @@ -49,6 +49,41 @@ describe("MessageStorage", () => { expect(result._tag).toEqual("Success") }).pipe(Effect.provide(MessageStorage.MemoryDriver.layer))) + it.effect("encoded unprocessedMessages fails closed for an empty address filter", () => + Effect.gen(function*() { + const driver = yield* MessageStorage.MemoryDriver + const envelope: Envelope.PartialRequestEncoded = { + _tag: "Request", + requestId: "1", + address: { shardId: { group: "default", id: 1 }, entityType: "Repro", entityId: "one" }, + tag: "Repro", + payload: {}, + headers: {} + } + yield* driver.encoded.saveEnvelope({ envelope, primaryKey: null, deliverAt: null }) + const messages = yield* driver.encoded.unprocessedMessages(["default:1"], 0, { addresses: [] }) + expect(messages).toHaveLength(0) + }).pipe(Effect.provide(MessageStorage.MemoryDriver.layer))) + + it.effect("encoded resetAddresses fails closed for an empty address list", () => + Effect.gen(function*() { + const driver = yield* MessageStorage.MemoryDriver + const envelope: Envelope.PartialRequestEncoded = { + _tag: "Request", + requestId: "1", + address: { shardId: { group: "default", id: 1 }, entityType: "Repro", entityId: "one" }, + tag: "Repro", + payload: {}, + headers: {} + } + yield* driver.encoded.saveEnvelope({ envelope, primaryKey: null, deliverAt: null }) + const claimed = yield* driver.encoded.unprocessedMessages(["default:1"], 0) + expect(claimed).toHaveLength(1) + yield* driver.encoded.resetAddresses([]) + const messages = yield* driver.encoded.unprocessedMessages(["default:1"], 1) + expect(messages).toHaveLength(0) + }).pipe(Effect.provide(MessageStorage.MemoryDriver.layer))) + it.effect("saves a request", () => Effect.gen(function*() { const storage = yield* MessageStorage.MessageStorage @@ -57,7 +92,7 @@ describe("MessageStorage", () => { expect(result._tag).toEqual("Success") const messages = yield* storage.unprocessedMessages([request.envelope.address.shardId]) expect(messages).toHaveLength(1) - }).pipe(Effect.provide(MemoryLive))) + }).pipe(Effect.provide(MemoryLayer))) it.effect("detects duplicates", () => Effect.gen(function*() { @@ -75,7 +110,7 @@ describe("MessageStorage", () => { }) ) expect(result._tag).toEqual("Duplicate") - }).pipe(Effect.provide(MemoryLive))) + }).pipe(Effect.provide(MemoryLayer))) it.effect("unprocessedMessages excludes complete requests", () => Effect.gen(function*() { @@ -85,7 +120,7 @@ describe("MessageStorage", () => { yield* storage.saveReply(yield* makeReply(request)) const messages = yield* storage.unprocessedMessages([request.envelope.address.shardId]) expect(messages).toHaveLength(0) - }).pipe(Effect.provide(MemoryLive))) + }).pipe(Effect.provide(MemoryLayer))) it.effect("unprocessedMessages honors the limit option", () => Effect.gen(function*() { @@ -99,7 +134,7 @@ describe("MessageStorage", () => { expect(messages.map((m: any) => m.envelope.payload.id)).toEqual([1, 2, 3]) const remaining = yield* storage.unprocessedMessages([shardId]) expect(remaining.map((m: any) => m.envelope.payload.id)).toEqual([4, 5]) - }).pipe(Effect.provide(MemoryLive))) + }).pipe(Effect.provide(MemoryLayer))) it.effect("unprocessedMessages filters by address", () => Effect.gen(function*() { @@ -121,7 +156,7 @@ describe("MessageStorage", () => { // an empty address filter returns nothing const none = yield* storage.unprocessedMessages([shardId], { addresses: [] }) expect(none).toHaveLength(0) - }).pipe(Effect.provide(MemoryLive))) + }).pipe(Effect.provide(MemoryLayer))) it.effect("resetAddresses makes claimed messages eligible again", () => Effect.gen(function*() { @@ -141,7 +176,7 @@ describe("MessageStorage", () => { const messages = yield* storage.unprocessedMessages([shardId]) expect(messages.map((m: any) => m.envelope.payload.id)).toEqual([1, 3, 4]) - }).pipe(Effect.provide(MemoryLive))) + }).pipe(Effect.provide(MemoryLayer))) it.effect("repliesFor", () => Effect.gen(function*() { @@ -154,7 +189,7 @@ describe("MessageStorage", () => { replies = yield* storage.repliesFor([request]) expect(replies).toHaveLength(1) expect(replies[0].requestId).toEqual(request.envelope.requestId) - }).pipe(Effect.provide(MemoryLive))) + }).pipe(Effect.provide(MemoryLayer))) it.effect("registerReplyHandler", () => Effect.gen(function*() { @@ -172,7 +207,7 @@ describe("MessageStorage", () => { yield* storage.saveReply(yield* makeReply(request)) yield* latch.await yield* Fiber.await(fiber) - }).pipe(Effect.provide(MemoryLive))) + }).pipe(Effect.provide(MemoryLayer))) }) }) From eec4650de84e33e26bdf58fd4dd6db721b768fcf Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Thu, 13 Aug 2026 00:07:35 +0000 Subject: [PATCH 6/9] Fix SQL message address filtering --- .changeset/wild-donuts-brake.md | 3 +- .../src/unstable/cluster/SqlMessageStorage.ts | 77 ++++++++++++++----- .../node/test/cluster/MessageStorageTest.ts | 3 +- .../SqlMessageStorage.integration.test.ts | 56 ++++++++++++++ 4 files changed, 117 insertions(+), 22 deletions(-) diff --git a/.changeset/wild-donuts-brake.md b/.changeset/wild-donuts-brake.md index 755cc716a6a..2e1f609d80d 100644 --- a/.changeset/wild-donuts-brake.md +++ b/.changeset/wild-donuts-brake.md @@ -22,4 +22,5 @@ window as SQL, so bounded reads advance past in-flight requests; resetting an address or shard makes its claimed messages immediately eligible again. The encoded driver contract replaces `Encoded.resetAddress` with the batched -`Encoded.resetAddresses` operation. +`Encoded.resetAddresses` operation. `SqlMessageStorage.makeEncoded` constructs +the SQL encoded driver directly for custom storage composition. diff --git a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts index d3870ea9c4f..75f4d3ea354 100644 --- a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts +++ b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts @@ -29,6 +29,7 @@ import * as SqlClient from "../sql/SqlClient.ts" import type { Row } from "../sql/SqlConnection.ts" import { isSqlError, type SqlError } from "../sql/SqlError.ts" import { PersistenceError } from "./ClusterError.ts" +import type * as EntityAddress from "./EntityAddress.ts" import type * as Envelope from "./Envelope.ts" import * as MessageStorage from "./MessageStorage.ts" import { SaveResultEncoded } from "./MessageStorage.ts" @@ -40,13 +41,13 @@ import * as Snowflake from "./Snowflake.ts" const withTracerDisabled = Effect.withTracerEnabled(false) /** - * Creates a SQL-backed `MessageStorage` implementation, running its migrations + * Creates a SQL-backed encoded message storage driver, running its migrations * and using the optional table prefix. * * **When to use** * - * Use when you need the SQL-backed `MessageStorage` service directly, such as - * when composing a custom layer or providing your own `Snowflake.Generator`. + * Use when you need the SQL-backed encoded driver directly, such as when + * composing a custom message storage adapter. * * **Details** * @@ -58,18 +59,17 @@ const withTracerDisabled = Effect.withTracerEnabled(false) * Changing `prefix` after deployment points the runtime at a different set of * tables, including the migration history table. * - * @see {@link layer} for a ready-made layer using the default prefix and generator - * @see {@link layerWith} for a ready-made layer with a custom table prefix + * @see {@link make} for the decoded `MessageStorage` constructor * * @category constructors * @since 4.0.0 */ -export const make: (options?: { +export const makeEncoded: (options?: { readonly prefix?: string | undefined }) => Effect.Effect< - MessageStorage.MessageStorage["Service"], + MessageStorage.Encoded, never, - SqlClient.SqlClient | Snowflake.Generator | Crypto.Crypto + SqlClient.SqlClient | Crypto.Crypto > = Effect.fnUntraced(function*(options) { const sql = (yield* SqlClient.SqlClient).withoutTransforms() const crypto = yield* Crypto.Crypto @@ -379,16 +379,37 @@ export const make: (options?: { mssql: () => (limit: number) => sql.literal(`OFFSET 0 ROWS FETCH NEXT ${Math.floor(limit)} ROWS ONLY`), orElse: () => (limit: number) => sql.literal(`LIMIT ${Math.floor(limit)}`) }) + const groupAddresses = (addresses: ReadonlyArray) => { + const byShard = new Map>>() + for (const address of addresses) { + const shardId = address.shardId.toString() + let byEntityType = byShard.get(shardId) + if (byEntityType === undefined) { + byShard.set(shardId, byEntityType = new Map()) + } + const group = byEntityType.get(address.entityType) + if (group === undefined) { + byEntityType.set(address.entityType, [address]) + } else { + group.push(address) + } + } + return Array.from( + byShard, + ([shardId, byEntityType]) => + Array.from(byEntityType, ([entityType, addresses]) => ({ shardId, entityType, addresses })) + ).flat() + } const unprocessedFilters = (options?: MessageStorage.UnprocessedOptions | undefined) => ({ addressFilter: options?.addresses !== undefined ? sql`AND (${ sql.or( - Object.entries(Arr.groupBy(options.addresses, (address) => address.entityType)).map( - ([entityType, group]) => - sql.and([ - sql`m.entity_type = ${entityType}`, - sql.in("m.entity_id", group.map((address) => address.entityId)) - ]) + groupAddresses(options.addresses).map((group) => + sql.and([ + sql`m.shard_id = ${group.shardId}`, + sql`m.entity_type = ${group.entityType}`, + sql.in("m.entity_id", group.addresses.map((address) => address.entityId)) + ]) ) ) })` @@ -467,7 +488,7 @@ export const make: (options?: { ? Effect.succeed([]) : getUnprocessedMessagesForDialect(shardIds, now, unprocessedFilters(options)) - return yield* MessageStorage.makeEncoded({ + const encoded: MessageStorage.Encoded = { saveEnvelope: ({ deliverAt, envelope, primaryKey }) => Effect.suspend(() => { let insert: Effect.Effect, SqlError | PlatformError.PlatformError> @@ -670,12 +691,12 @@ export const make: (options?: { WHERE processed = ${sqlFalse} AND (${ sql.or( - Object.values(Arr.groupBy(addresses, (address) => `${address.shardId}/${address.entityType}`)).map( + groupAddresses(addresses).map( (group) => sql.and([ - sql`shard_id = ${group[0].shardId.toString()}`, - sql`entity_type = ${group[0].entityType}`, - sql.in("entity_id", group.map((address) => address.entityId)) + sql`shard_id = ${group.shardId}`, + sql`entity_type = ${group.entityType}`, + sql.in("entity_id", group.addresses.map((address) => address.entityId)) ]) ) ) @@ -724,9 +745,25 @@ export const make: (options?: { sql.withTransaction(effect).pipe( Effect.catchIf(isSqlError, Effect.die) ) - }) + } + return encoded }, withTracerDisabled) +/** + * Creates a SQL-backed `MessageStorage` implementation, running its migrations + * and using the optional table prefix. + * + * @category constructors + * @since 4.0.0 + */ +export const make: (options?: { + readonly prefix?: string | undefined +}) => Effect.Effect< + MessageStorage.MessageStorage["Service"], + never, + SqlClient.SqlClient | Snowflake.Generator | Crypto.Crypto +> = (options) => Effect.flatMap(makeEncoded(options), MessageStorage.makeEncoded) + /** * Layer that provides SQL-backed `MessageStorage` using the default table prefix * and the default snowflake generator. diff --git a/packages/platform/node/test/cluster/MessageStorageTest.ts b/packages/platform/node/test/cluster/MessageStorageTest.ts index ec56d5e308c..33c415e482e 100644 --- a/packages/platform/node/test/cluster/MessageStorageTest.ts +++ b/packages/platform/node/test/cluster/MessageStorageTest.ts @@ -102,6 +102,7 @@ export const makeRequest = Effect.fnUntraced(function*(options?: { readonly rpc?: Rpc.AnyWithProps readonly payload?: any readonly entityId?: string + readonly shardId?: ShardId.ShardId }) { const snowflake = yield* Snowflake.Generator const rpc = options?.rpc ?? GetUserRpc @@ -109,7 +110,7 @@ export const makeRequest = Effect.fnUntraced(function*(options?: { envelope: Envelope.makeRequest({ requestId: snowflake.nextUnsafe(), address: EntityAddress.make({ - shardId: ShardId.make("default", 1), + shardId: options?.shardId ?? ShardId.make("default", 1), entityType: EntityType.make("test"), entityId: EntityId.make(options?.entityId ?? "1") }), diff --git a/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts b/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts index b2745ce1695..762ad924d08 100644 --- a/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts +++ b/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts @@ -344,6 +344,62 @@ describe("SqlMessageStorage", () => { expect(rest.map((m: any) => m.envelope.payload.id)).toEqual([1, 3]) })) + it.effect("unprocessedMessages filters addresses by shard", () => + Effect.gen(function*() { + yield* truncate + + const storage = yield* MessageStorage.MessageStorage + const shardOne = ShardId.make("default", 1) + const shardTwo = ShardId.make("default", 2) + yield* storage.saveRequest(yield* makeRequest({ payload: { id: 1 }, shardId: shardOne })) + yield* storage.saveRequest(yield* makeRequest({ payload: { id: 2 }, shardId: shardTwo })) + + const messages = yield* storage.unprocessedMessages([shardOne, shardTwo], { + addresses: [EntityAddress.make({ + shardId: shardOne, + entityType: EntityType.make("test"), + entityId: EntityId.make("1") + })] + }) + expect(messages.map((message: any) => message.envelope.payload.id)).toEqual([1]) + + const rest = yield* storage.unprocessedMessages([shardOne, shardTwo]) + expect(rest.map((message: any) => message.envelope.payload.id)).toEqual([2]) + })) + + it.effect("encoded unprocessedMessages fails closed for empty addresses", () => + Effect.gen(function*() { + yield* truncate + + const storage = yield* MessageStorage.MessageStorage + const request = yield* makeRequest() + yield* storage.saveRequest(request) + const encoded = yield* SqlMessageStorage.makeEncoded().pipe(Effect.provide(NodeCrypto.layer)) + const shardId = request.envelope.address.shardId.toString() + const messages = yield* encoded.unprocessedMessages([shardId], Date.now(), { addresses: [] }) + expect(messages).toHaveLength(0) + + const unfiltered = yield* encoded.unprocessedMessages([shardId], Date.now()) + expect(unfiltered).toHaveLength(1) + })) + + it.effect("encoded resetAddresses fails closed for an empty address list", () => + Effect.gen(function*() { + yield* truncate + + const storage = yield* MessageStorage.MessageStorage + const request = yield* makeRequest() + yield* storage.saveRequest(request) + const encoded = yield* SqlMessageStorage.makeEncoded().pipe(Effect.provide(NodeCrypto.layer)) + const shardId = request.envelope.address.shardId.toString() + const claimed = yield* encoded.unprocessedMessages([shardId], Date.now()) + expect(claimed).toHaveLength(1) + + yield* encoded.resetAddresses([]) + const messages = yield* encoded.unprocessedMessages([shardId], Date.now()) + expect(messages).toHaveLength(0) + })) + it.effect("resetAddresses releases claims in one batch", () => Effect.gen(function*() { yield* truncate From 0fa6335f15e21d8feb3159af344503c83d59dde5 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Thu, 13 Aug 2026 00:41:03 +0000 Subject: [PATCH 7/9] feat(cluster): release idle workflow entities quickly Workflow and durable clock entities now use a ten-second idle time instead of the one-minute entityMaxIdleTime default. Their state is durable, so a completed or suspended execution can be evicted safely and is rebuilt from storage when its next message arrives. This keeps finished executions from holding maxResidentEntities slots. Co-authored-by: Claude Fable 5 --- .changeset/wild-donuts-brake.md | 6 +++ .../unstable/cluster/ClusterWorkflowEngine.ts | 41 +++++++++++-------- .../cluster/ClusterWorkflowEngine.test.ts | 23 +++++++++++ 3 files changed, 54 insertions(+), 16 deletions(-) diff --git a/.changeset/wild-donuts-brake.md b/.changeset/wild-donuts-brake.md index 2e1f609d80d..b16ee1bd983 100644 --- a/.changeset/wild-donuts-brake.md +++ b/.changeset/wild-donuts-brake.md @@ -24,3 +24,9 @@ address or shard makes its claimed messages immediately eligible again. The encoded driver contract replaces `Encoded.resetAddress` with the batched `Encoded.resetAddresses` operation. `SqlMessageStorage.makeEncoded` constructs the SQL encoded driver directly for custom storage composition. + +`ClusterWorkflowEngine` entities (workflows and the durable clock) now use a +ten-second idle time instead of the one-minute `entityMaxIdleTime` default, so +completed and suspended executions release their entity slots quickly. Their +state is durable, so an evicted execution is rebuilt from storage when its +next message arrives. diff --git a/packages/effect/src/unstable/cluster/ClusterWorkflowEngine.ts b/packages/effect/src/unstable/cluster/ClusterWorkflowEngine.ts index 797a996506a..db1c5f95dbe 100644 --- a/packages/effect/src/unstable/cluster/ClusterWorkflowEngine.ts +++ b/packages/effect/src/unstable/cluster/ClusterWorkflowEngine.ts @@ -444,7 +444,7 @@ export const make = Effect.gen(function*() { } }), // Reserve a slot for deferred completions to wake the active run. - { concurrency: 2 } + { concurrency: 2, maxIdleTime: entityMaxIdleTime } ) as Effect.Effect ), @@ -702,6 +702,12 @@ const ResumeRpc = Rpc.make("resume", { const payloadParentKey = "~effect/cluster/ClusterWorkflowEngine/payloadParentKey" +// Workflow state is durable, so an idle entity (completed or suspended) can +// be released quickly and is rebuilt from storage when the next message +// arrives. This keeps completed and suspended executions from holding +// `maxResidentEntities` slots for the full `entityMaxIdleTime`. +const entityMaxIdleTime = Duration.seconds(10) + const makeWorkflowEntity = (workflow: Workflow.Any) => Entity.make(`Workflow/${workflow._tag}`, [ Rpc.make("run", { @@ -756,22 +762,25 @@ const ClockEntity = Entity.make("Workflow/-/DurableClock", [ ClockRpc ]) -const ClockEntityLayer = ClockEntity.toLayer(Effect.gen(function*() { - const engine = yield* WorkflowEngine.WorkflowEngine - const address = yield* Entity.CurrentAddress - const executionId = address.entityId - return { - run(request) { - const deferred = DurableClock.make({ name: request.payload.name, duration: Duration.zero }).deferred - return ensureSuccess(engine.deferredDone(deferred, { - workflowName: request.payload.workflowName, - executionId, - deferredName: deferred.name, - exit: Exit.void - })) +const ClockEntityLayer = ClockEntity.toLayer( + Effect.gen(function*() { + const engine = yield* WorkflowEngine.WorkflowEngine + const address = yield* Entity.CurrentAddress + const executionId = address.entityId + return { + run(request) { + const deferred = DurableClock.make({ name: request.payload.name, duration: Duration.zero }).deferred + return ensureSuccess(engine.deferredDone(deferred, { + workflowName: request.payload.workflowName, + executionId, + deferredName: deferred.name, + exit: Exit.void + })) + } } - } -})) + }), + { maxIdleTime: entityMaxIdleTime } +) const InterruptSignal = DurableDeferred.make("Workflow/InterruptSignal") diff --git a/packages/effect/test/cluster/ClusterWorkflowEngine.test.ts b/packages/effect/test/cluster/ClusterWorkflowEngine.test.ts index 150e6c0434e..4ddf3a8aeb0 100644 --- a/packages/effect/test/cluster/ClusterWorkflowEngine.test.ts +++ b/packages/effect/test/cluster/ClusterWorkflowEngine.test.ts @@ -84,6 +84,29 @@ describe.concurrent("ClusterWorkflowEngine", () => { expect(yield* EmailWorkflow.poll(executionId)).toEqual(Option.some(new Workflow.Complete({ exit: Exit.void }))) }).pipe(Effect.provide(TestWorkflowLayer))) + it.effect("releases idle workflow entities quickly", () => + Effect.gen(function*() { + const sharding = yield* Sharding.Sharding + yield* TestClock.adjust(1) + + yield* EmailWorkflow.execute({ + id: "test-email-reap", + to: "bob@example.com" + }).pipe(Effect.forkChild({ startImmediately: true })) + + yield* TestClock.adjust("10 seconds") + yield* sharding.pollStorage + yield* TestClock.adjust(5000) + + // the workflow is suspended, so its entities are idle but resident + assert.isAtLeast(yield* sharding.activeEntityCount, 1) + + // idle workflow entities are released well before the default + // one-minute entityMaxIdleTime + yield* TestClock.adjust("30 seconds") + assert.strictEqual(yield* sharding.activeEntityCount, 0) + }).pipe(Effect.provide(TestWorkflowLayer))) + it.effect("interrupts a suspended workflow and runs compensation", () => Effect.gen(function*() { const sharding = yield* Sharding.Sharding From 44be45a72ae8da068b52ab751adbb569329bcda6 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Thu, 13 Aug 2026 01:04:25 +0000 Subject: [PATCH 8/9] fix(cluster): cap workflow idle time without lengthening configured values The ten-second workflow entity idle time is now a cap applied with Duration.min over the configured entityMaxIdleTime, so operators who tune the global idle time below ten seconds keep their value. The config is read optionally from the layer context, keeping ShardingConfig out of ClusterWorkflowEngine.layer's public requirements; without it the cap applies as-is. The eviction test now resumes the workflow after its entities were released and asserts it completes, and a new test pins the sub-cap config behaviour. Co-authored-by: Claude Fable 5 --- .changeset/wild-donuts-brake.md | 10 +-- .../unstable/cluster/ClusterWorkflowEngine.ts | 58 +++++++++----- .../cluster/ClusterWorkflowEngine.test.ts | 80 ++++++++++++++----- 3 files changed, 103 insertions(+), 45 deletions(-) diff --git a/.changeset/wild-donuts-brake.md b/.changeset/wild-donuts-brake.md index b16ee1bd983..0e343491726 100644 --- a/.changeset/wild-donuts-brake.md +++ b/.changeset/wild-donuts-brake.md @@ -25,8 +25,8 @@ The encoded driver contract replaces `Encoded.resetAddress` with the batched `Encoded.resetAddresses` operation. `SqlMessageStorage.makeEncoded` constructs the SQL encoded driver directly for custom storage composition. -`ClusterWorkflowEngine` entities (workflows and the durable clock) now use a -ten-second idle time instead of the one-minute `entityMaxIdleTime` default, so -completed and suspended executions release their entity slots quickly. Their -state is durable, so an evicted execution is rebuilt from storage when its -next message arrives. +`ClusterWorkflowEngine` entities (workflows and the durable clock) now cap +their idle time at ten seconds, so completed and suspended executions release +their entity slots quickly. A configured `entityMaxIdleTime` below the cap +still applies. Their state is durable, so an evicted execution is rebuilt +from storage when its next message arrives. diff --git a/packages/effect/src/unstable/cluster/ClusterWorkflowEngine.ts b/packages/effect/src/unstable/cluster/ClusterWorkflowEngine.ts index db1c5f95dbe..a19ee3a1a9d 100644 --- a/packages/effect/src/unstable/cluster/ClusterWorkflowEngine.ts +++ b/packages/effect/src/unstable/cluster/ClusterWorkflowEngine.ts @@ -41,6 +41,7 @@ import { MessageStorage } from "./MessageStorage.ts" import type { WithExitEncoded } from "./Reply.ts" import * as Reply from "./Reply.ts" import * as Sharding from "./Sharding.ts" +import { ShardingConfig } from "./ShardingConfig.ts" import * as Snowflake from "./Snowflake.ts" /** @@ -58,6 +59,7 @@ import * as Snowflake from "./Snowflake.ts" export const make = Effect.gen(function*() { const sharding = yield* Sharding.Sharding const storage = yield* MessageStorage + const maxIdleTime = yield* workflowMaxIdleTime const workflows = new Map() const entities = new Map< @@ -444,7 +446,7 @@ export const make = Effect.gen(function*() { } }), // Reserve a slot for deferred completions to wake the active run. - { concurrency: 2, maxIdleTime: entityMaxIdleTime } + { concurrency: 2, maxIdleTime } ) as Effect.Effect ), @@ -704,9 +706,20 @@ const payloadParentKey = "~effect/cluster/ClusterWorkflowEngine/payloadParentKey // Workflow state is durable, so an idle entity (completed or suspended) can // be released quickly and is rebuilt from storage when the next message -// arrives. This keeps completed and suspended executions from holding -// `maxResidentEntities` slots for the full `entityMaxIdleTime`. -const entityMaxIdleTime = Duration.seconds(10) +// arrives. Cap the idle time so completed and suspended executions do not +// hold `maxResidentEntities` slots for the full `entityMaxIdleTime`. +const workflowIdleTimeCap = Duration.seconds(10) + +// The cap only ever shortens the configured `entityMaxIdleTime`, so operators +// tuning it below the cap keep their value. The config is read optionally to +// keep it out of the public layer requirements; without it the cap applies. +const workflowMaxIdleTime: Effect.Effect = Effect.map( + Effect.serviceOption(ShardingConfig), + Option.match({ + onNone: () => workflowIdleTimeCap, + onSome: (config) => Duration.min(workflowIdleTimeCap, Duration.fromInputUnsafe(config.entityMaxIdleTime)) + }) +) const makeWorkflowEntity = (workflow: Workflow.Any) => Entity.make(`Workflow/${workflow._tag}`, [ @@ -762,25 +775,26 @@ const ClockEntity = Entity.make("Workflow/-/DurableClock", [ ClockRpc ]) -const ClockEntityLayer = ClockEntity.toLayer( - Effect.gen(function*() { - const engine = yield* WorkflowEngine.WorkflowEngine - const address = yield* Entity.CurrentAddress - const executionId = address.entityId - return { - run(request) { - const deferred = DurableClock.make({ name: request.payload.name, duration: Duration.zero }).deferred - return ensureSuccess(engine.deferredDone(deferred, { - workflowName: request.payload.workflowName, - executionId, - deferredName: deferred.name, - exit: Exit.void - })) +const ClockEntityLayer = Layer.unwrap(Effect.map(workflowMaxIdleTime, (maxIdleTime) => + ClockEntity.toLayer( + Effect.gen(function*() { + const engine = yield* WorkflowEngine.WorkflowEngine + const address = yield* Entity.CurrentAddress + const executionId = address.entityId + return { + run(request) { + const deferred = DurableClock.make({ name: request.payload.name, duration: Duration.zero }).deferred + return ensureSuccess(engine.deferredDone(deferred, { + workflowName: request.payload.workflowName, + executionId, + deferredName: deferred.name, + exit: Exit.void + })) + } } - } - }), - { maxIdleTime: entityMaxIdleTime } -) + }), + { maxIdleTime } + ))) const InterruptSignal = DurableDeferred.make("Workflow/InterruptSignal") diff --git a/packages/effect/test/cluster/ClusterWorkflowEngine.test.ts b/packages/effect/test/cluster/ClusterWorkflowEngine.test.ts index 4ddf3a8aeb0..561ac2fdf22 100644 --- a/packages/effect/test/cluster/ClusterWorkflowEngine.test.ts +++ b/packages/effect/test/cluster/ClusterWorkflowEngine.test.ts @@ -87,9 +87,10 @@ describe.concurrent("ClusterWorkflowEngine", () => { it.effect("releases idle workflow entities quickly", () => Effect.gen(function*() { const sharding = yield* Sharding.Sharding + const driver = yield* MessageStorage.MemoryDriver yield* TestClock.adjust(1) - yield* EmailWorkflow.execute({ + const fiber = yield* EmailWorkflow.execute({ id: "test-email-reap", to: "bob@example.com" }).pipe(Effect.forkChild({ startImmediately: true })) @@ -105,8 +106,49 @@ describe.concurrent("ClusterWorkflowEngine", () => { // one-minute entityMaxIdleTime yield* TestClock.adjust("30 seconds") assert.strictEqual(yield* sharding.activeEntityCount, 0) + + // eviction is safe: the evicted execution is rebuilt from storage and + // runs to completion when it is resumed + const executionId = driver.journal[0].address.entityId + const token = yield* DurableDeferred.token(EmailTrigger).pipe( + Effect.provideService(WorkflowInstance, WorkflowInstance.initial(EmailWorkflow, executionId)) + ) + yield* DurableDeferred.done(EmailTrigger, { + token, + exit: Exit.succeed("done") + }) + yield* sharding.pollStorage + yield* TestClock.adjust(10000) + expect(yield* Fiber.join(fiber)).toBeUndefined() + expect(yield* EmailWorkflow.poll(executionId)).toEqual(Option.some(new Workflow.Complete({ exit: Exit.void }))) }).pipe(Effect.provide(TestWorkflowLayer))) + it.effect("never lengthens a configured idle time below the workflow cap", () => + Effect.gen(function*() { + const sharding = yield* Sharding.Sharding + yield* TestClock.adjust(1) + + yield* EmailWorkflow.execute({ + id: "test-email-short-idle", + to: "bob@example.com" + }).pipe(Effect.forkChild({ startImmediately: true })) + + yield* TestClock.adjust("10 seconds") + yield* sharding.pollStorage + yield* TestClock.adjust(5000) + + // the workflow is suspended, so its entities are idle but resident + assert.isAtLeast(yield* sharding.activeEntityCount, 1) + + // with entityMaxIdleTime below the ten-second workflow cap, entities + // are released at the configured five seconds + yield* TestClock.adjust(8000) + assert.strictEqual(yield* sharding.activeEntityCount, 0) + }).pipe(Effect.provide(EmailWorkflowLayer.pipe( + Layer.provideMerge(Flags.layer), + Layer.provideMerge(makeTestWorkflowEngine({ entityMaxIdleTime: 5000 })) + )))) + it.effect("interrupts a suspended workflow and runs compensation", () => Effect.gen(function*() { const sharding = yield* Sharding.Sharding @@ -759,24 +801,26 @@ describe.concurrent("ClusterWorkflowEngine", () => { }).pipe(Effect.provide(TestWorkflowLayer))) }) -const TestShardingConfig = ShardingConfig.layer({ - shardsPerGroup: 300, - availableShardGroups: ["default", "workflow"], - assignedShardGroups: ["default", "workflow"], - entityMailboxCapacity: 10, - entityTerminationTimeout: 0, - entityMessagePollInterval: 5000, - sendRetryInterval: 100 -}) +const makeTestWorkflowEngine = (config?: Partial) => + ClusterWorkflowEngine.layer.pipe( + Layer.provideMerge(Sharding.layer), + Layer.provide(Runners.layerNoop), + Layer.provideMerge(MessageStorage.layerMemory), + Layer.provide(RunnerStorage.layerMemory), + Layer.provide(RunnerHealth.layerNoop), + Layer.provide(ShardingConfig.layer({ + shardsPerGroup: 300, + availableShardGroups: ["default", "workflow"], + assignedShardGroups: ["default", "workflow"], + entityMailboxCapacity: 10, + entityTerminationTimeout: 0, + entityMessagePollInterval: 5000, + sendRetryInterval: 100, + ...config + })) + ) -const TestWorkflowEngine = ClusterWorkflowEngine.layer.pipe( - Layer.provideMerge(Sharding.layer), - Layer.provide(Runners.layerNoop), - Layer.provideMerge(MessageStorage.layerMemory), - Layer.provide(RunnerStorage.layerMemory), - Layer.provide(RunnerHealth.layerNoop), - Layer.provide(TestShardingConfig) -) +const TestWorkflowEngine = makeTestWorkflowEngine() class SendEmailError extends Schema.Error("SendEmailError")({ _tag: Schema.tag("SendEmailError"), From d8061b5bc92757dcd0802e9ab1b4852162d155bf Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Thu, 13 Aug 2026 01:44:18 +0000 Subject: [PATCH 9/9] fix(cluster): address final review feedback --- .changeset/wild-donuts-brake.md | 9 ++- .../unstable/cluster/ClusterWorkflowEngine.ts | 58 +++++++---------- .../src/unstable/cluster/MessageStorage.ts | 27 +++----- .../effect/src/unstable/cluster/Sharding.ts | 5 +- .../src/unstable/cluster/ShardingConfig.ts | 3 - .../src/unstable/cluster/SqlMessageStorage.ts | 8 ++- .../cluster/ClusterWorkflowEngine.test.ts | 65 ------------------- 7 files changed, 44 insertions(+), 131 deletions(-) diff --git a/.changeset/wild-donuts-brake.md b/.changeset/wild-donuts-brake.md index 0e343491726..4bfeecea7f9 100644 --- a/.changeset/wild-donuts-brake.md +++ b/.changeset/wild-donuts-brake.md @@ -25,8 +25,7 @@ The encoded driver contract replaces `Encoded.resetAddress` with the batched `Encoded.resetAddresses` operation. `SqlMessageStorage.makeEncoded` constructs the SQL encoded driver directly for custom storage composition. -`ClusterWorkflowEngine` entities (workflows and the durable clock) now cap -their idle time at ten seconds, so completed and suspended executions release -their entity slots quickly. A configured `entityMaxIdleTime` below the cap -still applies. Their state is durable, so an evicted execution is rebuilt -from storage when its next message arrives. +`ClusterWorkflowEngine` entities (workflows and the durable clock) now use a +fixed ten-second idle time, so completed and suspended executions release their +entity slots quickly. Their state is durable, so an evicted execution is +rebuilt from storage when its next message arrives. diff --git a/packages/effect/src/unstable/cluster/ClusterWorkflowEngine.ts b/packages/effect/src/unstable/cluster/ClusterWorkflowEngine.ts index a19ee3a1a9d..db1c5f95dbe 100644 --- a/packages/effect/src/unstable/cluster/ClusterWorkflowEngine.ts +++ b/packages/effect/src/unstable/cluster/ClusterWorkflowEngine.ts @@ -41,7 +41,6 @@ import { MessageStorage } from "./MessageStorage.ts" import type { WithExitEncoded } from "./Reply.ts" import * as Reply from "./Reply.ts" import * as Sharding from "./Sharding.ts" -import { ShardingConfig } from "./ShardingConfig.ts" import * as Snowflake from "./Snowflake.ts" /** @@ -59,7 +58,6 @@ import * as Snowflake from "./Snowflake.ts" export const make = Effect.gen(function*() { const sharding = yield* Sharding.Sharding const storage = yield* MessageStorage - const maxIdleTime = yield* workflowMaxIdleTime const workflows = new Map() const entities = new Map< @@ -446,7 +444,7 @@ export const make = Effect.gen(function*() { } }), // Reserve a slot for deferred completions to wake the active run. - { concurrency: 2, maxIdleTime } + { concurrency: 2, maxIdleTime: entityMaxIdleTime } ) as Effect.Effect ), @@ -706,20 +704,9 @@ const payloadParentKey = "~effect/cluster/ClusterWorkflowEngine/payloadParentKey // Workflow state is durable, so an idle entity (completed or suspended) can // be released quickly and is rebuilt from storage when the next message -// arrives. Cap the idle time so completed and suspended executions do not -// hold `maxResidentEntities` slots for the full `entityMaxIdleTime`. -const workflowIdleTimeCap = Duration.seconds(10) - -// The cap only ever shortens the configured `entityMaxIdleTime`, so operators -// tuning it below the cap keep their value. The config is read optionally to -// keep it out of the public layer requirements; without it the cap applies. -const workflowMaxIdleTime: Effect.Effect = Effect.map( - Effect.serviceOption(ShardingConfig), - Option.match({ - onNone: () => workflowIdleTimeCap, - onSome: (config) => Duration.min(workflowIdleTimeCap, Duration.fromInputUnsafe(config.entityMaxIdleTime)) - }) -) +// arrives. This keeps completed and suspended executions from holding +// `maxResidentEntities` slots for the full `entityMaxIdleTime`. +const entityMaxIdleTime = Duration.seconds(10) const makeWorkflowEntity = (workflow: Workflow.Any) => Entity.make(`Workflow/${workflow._tag}`, [ @@ -775,26 +762,25 @@ const ClockEntity = Entity.make("Workflow/-/DurableClock", [ ClockRpc ]) -const ClockEntityLayer = Layer.unwrap(Effect.map(workflowMaxIdleTime, (maxIdleTime) => - ClockEntity.toLayer( - Effect.gen(function*() { - const engine = yield* WorkflowEngine.WorkflowEngine - const address = yield* Entity.CurrentAddress - const executionId = address.entityId - return { - run(request) { - const deferred = DurableClock.make({ name: request.payload.name, duration: Duration.zero }).deferred - return ensureSuccess(engine.deferredDone(deferred, { - workflowName: request.payload.workflowName, - executionId, - deferredName: deferred.name, - exit: Exit.void - })) - } +const ClockEntityLayer = ClockEntity.toLayer( + Effect.gen(function*() { + const engine = yield* WorkflowEngine.WorkflowEngine + const address = yield* Entity.CurrentAddress + const executionId = address.entityId + return { + run(request) { + const deferred = DurableClock.make({ name: request.payload.name, duration: Duration.zero }).deferred + return ensureSuccess(engine.deferredDone(deferred, { + workflowName: request.payload.workflowName, + executionId, + deferredName: deferred.name, + exit: Exit.void + })) } - }), - { maxIdleTime } - ))) + } + }), + { maxIdleTime: entityMaxIdleTime } +) const InterruptSignal = DurableDeferred.make("Workflow/InterruptSignal") diff --git a/packages/effect/src/unstable/cluster/MessageStorage.ts b/packages/effect/src/unstable/cluster/MessageStorage.ts index 5f49730f3dc..79490b90316 100644 --- a/packages/effect/src/unstable/cluster/MessageStorage.ts +++ b/packages/effect/src/unstable/cluster/MessageStorage.ts @@ -139,7 +139,10 @@ export class MessageStorage extends Context.Service, - options?: UnprocessedOptions | undefined + options?: { + readonly limit?: number | undefined + readonly addresses?: ReadonlyArray | undefined + } | undefined ) => Effect.Effect>, PersistenceError> /** @@ -185,23 +188,6 @@ export class MessageStorage extends Context.Service Effect.Effect }>()("effect/cluster/MessageStorage") {} -/** - * Options for reading unprocessed messages. - * - * **Details** - * - * `limit` bounds the number of messages returned by a single read, and - * `addresses` restricts the read to the provided entity addresses. Storage - * implementations only claim the messages they actually return. - * - * @category models - * @since 4.0.0 - */ -export type UnprocessedOptions = { - readonly limit?: number | undefined - readonly addresses?: ReadonlyArray | undefined -} - /** * Result of saving a request or envelope into message storage. * @@ -388,7 +374,10 @@ export type Encoded = { readonly unprocessedMessages: ( shardIds: Arr.NonEmptyArray, now: number, - options?: UnprocessedOptions | undefined + options?: { + readonly limit?: number | undefined + readonly addresses?: ReadonlyArray | undefined + } | undefined ) => Effect.Effect< Array<{ readonly envelope: Envelope.Encoded diff --git a/packages/effect/src/unstable/cluster/Sharding.ts b/packages/effect/src/unstable/cluster/Sharding.ts index 47c12f61b59..33486152398 100644 --- a/packages/effect/src/unstable/cluster/Sharding.ts +++ b/packages/effect/src/unstable/cluster/Sharding.ts @@ -642,7 +642,10 @@ const make = Effect.gen(function*() { body: () => send }) - const readAndProcess = Effect.fnUntraced(function*(options?: MessageStorage.UnprocessedOptions) { + const readAndProcess = Effect.fnUntraced(function*(options?: { + readonly limit?: number | undefined + readonly addresses?: ReadonlyArray | undefined + }) { messages = yield* storage.unprocessedMessages(acquiredShards, options) index = 0 yield* processMessages diff --git a/packages/effect/src/unstable/cluster/ShardingConfig.ts b/packages/effect/src/unstable/cluster/ShardingConfig.ts index 1fab4e7ec7f..1c35f200b33 100644 --- a/packages/effect/src/unstable/cluster/ShardingConfig.ts +++ b/packages/effect/src/unstable/cluster/ShardingConfig.ts @@ -104,9 +104,6 @@ export class ShardingConfig extends Context.Service ({ shardId, entityType, addresses })) ).flat() } - const unprocessedFilters = (options?: MessageStorage.UnprocessedOptions | undefined) => ({ + type UnprocessedOptions = { + readonly limit?: number | undefined + readonly addresses?: ReadonlyArray | undefined + } + const unprocessedFilters = (options?: UnprocessedOptions | undefined) => ({ addressFilter: options?.addresses !== undefined ? sql`AND (${ sql.or( @@ -482,7 +486,7 @@ export const makeEncoded: (options?: { const getUnprocessedMessages = ( shardIds: ReadonlyArray, now: number, - options?: MessageStorage.UnprocessedOptions | undefined + options?: UnprocessedOptions | undefined ) => options?.addresses?.length === 0 ? Effect.succeed([]) diff --git a/packages/effect/test/cluster/ClusterWorkflowEngine.test.ts b/packages/effect/test/cluster/ClusterWorkflowEngine.test.ts index 561ac2fdf22..1232a5aad17 100644 --- a/packages/effect/test/cluster/ClusterWorkflowEngine.test.ts +++ b/packages/effect/test/cluster/ClusterWorkflowEngine.test.ts @@ -84,71 +84,6 @@ describe.concurrent("ClusterWorkflowEngine", () => { expect(yield* EmailWorkflow.poll(executionId)).toEqual(Option.some(new Workflow.Complete({ exit: Exit.void }))) }).pipe(Effect.provide(TestWorkflowLayer))) - it.effect("releases idle workflow entities quickly", () => - Effect.gen(function*() { - const sharding = yield* Sharding.Sharding - const driver = yield* MessageStorage.MemoryDriver - yield* TestClock.adjust(1) - - const fiber = yield* EmailWorkflow.execute({ - id: "test-email-reap", - to: "bob@example.com" - }).pipe(Effect.forkChild({ startImmediately: true })) - - yield* TestClock.adjust("10 seconds") - yield* sharding.pollStorage - yield* TestClock.adjust(5000) - - // the workflow is suspended, so its entities are idle but resident - assert.isAtLeast(yield* sharding.activeEntityCount, 1) - - // idle workflow entities are released well before the default - // one-minute entityMaxIdleTime - yield* TestClock.adjust("30 seconds") - assert.strictEqual(yield* sharding.activeEntityCount, 0) - - // eviction is safe: the evicted execution is rebuilt from storage and - // runs to completion when it is resumed - const executionId = driver.journal[0].address.entityId - const token = yield* DurableDeferred.token(EmailTrigger).pipe( - Effect.provideService(WorkflowInstance, WorkflowInstance.initial(EmailWorkflow, executionId)) - ) - yield* DurableDeferred.done(EmailTrigger, { - token, - exit: Exit.succeed("done") - }) - yield* sharding.pollStorage - yield* TestClock.adjust(10000) - expect(yield* Fiber.join(fiber)).toBeUndefined() - expect(yield* EmailWorkflow.poll(executionId)).toEqual(Option.some(new Workflow.Complete({ exit: Exit.void }))) - }).pipe(Effect.provide(TestWorkflowLayer))) - - it.effect("never lengthens a configured idle time below the workflow cap", () => - Effect.gen(function*() { - const sharding = yield* Sharding.Sharding - yield* TestClock.adjust(1) - - yield* EmailWorkflow.execute({ - id: "test-email-short-idle", - to: "bob@example.com" - }).pipe(Effect.forkChild({ startImmediately: true })) - - yield* TestClock.adjust("10 seconds") - yield* sharding.pollStorage - yield* TestClock.adjust(5000) - - // the workflow is suspended, so its entities are idle but resident - assert.isAtLeast(yield* sharding.activeEntityCount, 1) - - // with entityMaxIdleTime below the ten-second workflow cap, entities - // are released at the configured five seconds - yield* TestClock.adjust(8000) - assert.strictEqual(yield* sharding.activeEntityCount, 0) - }).pipe(Effect.provide(EmailWorkflowLayer.pipe( - Layer.provideMerge(Flags.layer), - Layer.provideMerge(makeTestWorkflowEngine({ entityMaxIdleTime: 5000 })) - )))) - it.effect("interrupts a suspended workflow and runs compensation", () => Effect.gen(function*() { const sharding = yield* Sharding.Sharding