From ec72271c369958e056c93d649619c47cb25933ad Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Sun, 20 Sep 2026 01:31:32 +0000 Subject: [PATCH 01/12] test(cluster): reproduce reset request lost during storage read --- packages/effect/test/cluster/Sharding.test.ts | 64 +++++++++++++++++++ 1 file changed, 64 insertions(+) diff --git a/packages/effect/test/cluster/Sharding.test.ts b/packages/effect/test/cluster/Sharding.test.ts index 2d4d4b3f778..bf5f93d5364 100644 --- a/packages/effect/test/cluster/Sharding.test.ts +++ b/packages/effect/test/cluster/Sharding.test.ts @@ -54,6 +54,70 @@ import { } from "./TestEntity.ts" describe.concurrent("Sharding", () => { + it.effect("redelivers a request reset while an asynchronous storage read is pending", () => + Effect.gen(function*() { + const readStarted = yield* Deferred.make() + const releaseRead = yield* Deferred.make() + let pauseNextRead = false + const claimed: Array = [] + + yield* Effect.gen(function*() { + yield* TestClock.adjust(1) + const sharding = yield* Sharding.Sharding + const state = yield* TestEntityState + const client = (yield* TestEntity.client)("reset-race") + const firstRun = yield* client.RequestWithKey({ key: "run" }).pipe( + Effect.forkChild({ startImmediately: true }) + ) + const request = yield* Queue.take(state.envelopes) + + // The read loop has cleared processedRequestIds, but the asynchronous + // storage query has not selected or claimed any rows yet. + pauseNextRead = true + yield* sharding.pollStorage + yield* Deferred.await(readStarted) + + // Finish the first attempt during the pending read. Waiting for all + // fibers to settle also lets the manager record the processed ID. + yield* Queue.offer(state.messages, void 0) + yield* Fiber.join(firstRun) + yield* TestClock.adjust(1) + assert.isTrue(yield* sharding.reset(request.requestId)) + yield* sharding.pollStorage + claimed.length = 0 + + // The pending query now claims the reset request. The next poll must + // not lose it to the completed attempt's in-memory deduplication. + yield* Deferred.succeed(releaseRead, void 0) + yield* TestClock.adjust(5000) + assert.include(claimed, request.requestId) + const deliveriesBeforeClaimExpiry = Queue.sizeUnsafe(state.envelopes) + + // The memory driver uses the same ten-minute claim expiry as SQL. + // Prove the request is still recoverable, rather than a dead handler. + yield* TestClock.adjust("10 minutes") + assert.strictEqual(Queue.sizeUnsafe(state.envelopes), 1) + assert.strictEqual( + deliveriesBeforeClaimExpiry, + 1, + "reset request was claimed but not redelivered before claim expiry" + ) + }).pipe(Effect.provide(CappedSharding({}, (storage) => ({ + ...storage, + unprocessedMessages: (shardIds, options) => + Effect.gen(function*() { + if (pauseNextRead) { + pauseNextRead = false + yield* Deferred.succeed(readStarted, void 0) + yield* Deferred.await(releaseRead) + } + const messages = yield* storage.unprocessedMessages(shardIds, options) + for (const message of messages) claimed.push(message.envelope.requestId) + return messages + }) + })))) + })) + it.effect("delivers volatile requests directly to the entity", () => Effect.gen(function*() { yield* TestClock.adjust(1) From d5352ce6744094a7e4fb08ac34cb06175b7f0ef6 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Sun, 20 Sep 2026 01:37:47 +0000 Subject: [PATCH 02/12] fix(cluster): release claims suppressed by completed requests --- .changeset/cluster-reset-claims.md | 5 ++++ .../src/unstable/cluster/MessageStorage.ts | 25 +++++++++++++++++++ .../effect/src/unstable/cluster/Sharding.ts | 25 +++++++++++++++++++ .../src/unstable/cluster/SqlMessageStorage.ts | 14 +++++++++++ 4 files changed, 69 insertions(+) create mode 100644 .changeset/cluster-reset-claims.md diff --git a/.changeset/cluster-reset-claims.md b/.changeset/cluster-reset-claims.md new file mode 100644 index 00000000000..94fabe11ce9 --- /dev/null +++ b/.changeset/cluster-reset-claims.md @@ -0,0 +1,5 @@ +--- +"effect": patch +--- + +Release storage claims for deduplicated cluster requests so concurrent workflow resets can resume on the next poll instead of waiting for the ten-minute claim timeout. diff --git a/packages/effect/src/unstable/cluster/MessageStorage.ts b/packages/effect/src/unstable/cluster/MessageStorage.ts index 3a4575c3f16..c88d6e38e0a 100644 --- a/packages/effect/src/unstable/cluster/MessageStorage.ts +++ b/packages/effect/src/unstable/cluster/MessageStorage.ts @@ -178,6 +178,14 @@ export class MessageStorage extends Context.Service Effect.Effect + /** + * Release claims for the provided requests without changing their replies or + * processed state. + */ + readonly resetRequests: ( + requestIds: ReadonlyArray + ) => Effect.Effect + /** * Reset the mailbox state for the provided addresses. */ @@ -412,6 +420,14 @@ export type Encoded = { PersistenceError > + /** + * Release claims for the provided requests without changing their replies or + * processed state. + */ + readonly resetRequests: ( + requestIds: ReadonlyArray + ) => Effect.Effect + /** * Reset the mailbox state for the provided addresses. */ @@ -716,6 +732,7 @@ export const makeEncoded: (encoded: Encoded) => Effect.Effect< ) }, resetAddress: (address) => encoded.resetAddresses([address]), + resetRequests: (requestIds) => requestIds.length === 0 ? Effect.void : encoded.resetRequests(requestIds), resetAddresses: (addresses) => addresses.length === 0 ? Effect.void : encoded.resetAddresses(addresses), clearAddress: encoded.clearAddress, resetShards: (shardIds) => { @@ -846,6 +863,7 @@ export const noop: MessageStorage["Service"] = Effect.runSync(make({ unprocessedMessages: () => Effect.succeed([]), unprocessedMessagesById: () => Effect.succeed([]), resetAddress: () => Effect.void, + resetRequests: () => Effect.void, resetAddresses: () => Effect.void, clearAddress: () => Effect.void, resetShards: () => Effect.void, @@ -1091,6 +1109,13 @@ export class MemoryDriver extends Context.Service()("effect/cluste return unprocessedWith((envelope) => envelopeIds.has(envelope.requestId)) }), resetAddresses, + resetRequests: (requestIds) => + Effect.sync(() => { + for (const id of requestIds) { + const entry = requests.get(String(id)) + if (entry) lastRead.delete(entry.envelope) + } + }), clearAddress: (address) => Effect.sync(() => { for (const [primaryKey, entry] of requestsByPrimaryKey) { diff --git a/packages/effect/src/unstable/cluster/Sharding.ts b/packages/effect/src/unstable/cluster/Sharding.ts index 65f787c4003..a8133a9f77d 100644 --- a/packages/effect/src/unstable/cluster/Sharding.ts +++ b/packages/effect/src/unstable/cluster/Sharding.ts @@ -637,6 +637,7 @@ const make = Effect.gen(function*() { let deliveredThisRead = false const removableNotifications = new Set() const resetAddresses = MutableHashSet.empty() + const resetRequestIds: Array = [] const cappedAddresses = MutableHashSet.empty() const markDelivered = Effect.sync(() => { @@ -694,6 +695,17 @@ const make = Effect.gen(function*() { // send it to the entity manager to be processed. return Effect.tap(state.manager.send(message), markDelivered) } else if (isProcessing || state.status === "closing") { + if ( + isProcessing && message._tag === "IncomingRequest" && + !state.manager.isProcessingFor(message, { excludeReplies: true }) + ) { + // A reset (possibly on another runner) can race with this read + // and reply publication. Keep deduplicating the request, but do + // not leave its new claim stranded until the claim timeout. + // This also covers replies published before the active request + // has been moved to the processed-ID cache. + resetRequestIds.push(message.envelope.requestId) + } // If the request is already processing, we skip it. // Or if the entity is closing, we skip all incoming messages. return Effect.void @@ -738,6 +750,14 @@ const make = Effect.gen(function*() { defect: Cause.squash(cause) })) } + if (error.success._tag === "AlreadyProcessingMessage" && message._tag === "IncomingRequest") { + // Decoding or entity acquisition can yield after isProcessingFor. + const state = entityManagers.get(message.envelope.address.entityType) + if (!state?.manager.isProcessingFor(message, { excludeReplies: true })) { + resetRequestIds.push(message.envelope.requestId) + } + return Effect.void + } if (error.success._tag === "MailboxFull") { const address = message.envelope.address const state = entityManagers.get(address.entityType) @@ -850,6 +870,11 @@ const make = Effect.gen(function*() { // let the resuming entities check if they are done yield* storageReadLock.release(1) + if (resetRequestIds.length > 0) { + yield* storage.resetRequests(resetRequestIds) + resetRequestIds.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 75f87723cb2..4cc2bbc98ec 100644 --- a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts +++ b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts @@ -686,6 +686,20 @@ export const makeEncoded: (options?: { ) }, + resetRequests: (requestIds) => + requestIds.length === 0 + ? Effect.void + : sql` + UPDATE ${messagesTableSql} + SET last_read = NULL + WHERE processed = ${sqlFalse} + AND ${sql.in("id", requestIds.map(String))} + `.pipe( + Effect.asVoid, + PersistenceError.refail, + withTracerDisabled + ), + resetAddresses: (addresses) => addresses.length === 0 ? Effect.void From 710a1d51aa92b1b27f1d756342679d098cbf9bdf Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Sun, 20 Sep 2026 01:43:04 +0000 Subject: [PATCH 03/12] refactor(cluster): share completed-claim release in storage read loop --- .../effect/src/unstable/cluster/Sharding.ts | 33 +++++++++---------- 1 file changed, 16 insertions(+), 17 deletions(-) diff --git a/packages/effect/src/unstable/cluster/Sharding.ts b/packages/effect/src/unstable/cluster/Sharding.ts index a8133a9f77d..940921b0382 100644 --- a/packages/effect/src/unstable/cluster/Sharding.ts +++ b/packages/effect/src/unstable/cluster/Sharding.ts @@ -644,6 +644,18 @@ const make = Effect.gen(function*() { deliveredThisRead = true }) + // A request is deduplicated while it is running, but also while its + // reply is being published and, until the next read, after it has + // completed. A reset (possibly from another runner) can race with that + // window, so release the claim this read just took on a completed + // request. Otherwise the reset is only delivered once the claim expires. + const releaseCompletedClaim = (message: Message.Incoming) => { + const state = entityManagers.get(message.envelope.address.entityType) + if (!state?.manager.isProcessingFor(message, { excludeReplies: true })) { + resetRequestIds.push(message.envelope.requestId) + } + } + const processMessages = Effect.whileLoop({ while: () => index < messages.length, step: () => index++, @@ -695,19 +707,9 @@ const make = Effect.gen(function*() { // send it to the entity manager to be processed. return Effect.tap(state.manager.send(message), markDelivered) } else if (isProcessing || state.status === "closing") { - if ( - isProcessing && message._tag === "IncomingRequest" && - !state.manager.isProcessingFor(message, { excludeReplies: true }) - ) { - // A reset (possibly on another runner) can race with this read - // and reply publication. Keep deduplicating the request, but do - // not leave its new claim stranded until the claim timeout. - // This also covers replies published before the active request - // has been moved to the processed-ID cache. - resetRequestIds.push(message.envelope.requestId) - } // If the request is already processing, we skip it. // Or if the entity is closing, we skip all incoming messages. + if (isProcessing) releaseCompletedClaim(message) return Effect.void } else if (message._tag === "IncomingRequest" && pendingNotifications.has(message.envelope.requestId)) { const entry = pendingNotifications.get(message.envelope.requestId)! @@ -750,12 +752,9 @@ const make = Effect.gen(function*() { defect: Cause.squash(cause) })) } - if (error.success._tag === "AlreadyProcessingMessage" && message._tag === "IncomingRequest") { - // Decoding or entity acquisition can yield after isProcessingFor. - const state = entityManagers.get(message.envelope.address.entityType) - if (!state?.manager.isProcessingFor(message, { excludeReplies: true })) { - resetRequestIds.push(message.envelope.requestId) - } + if (error.success._tag === "AlreadyProcessingMessage") { + // The request completed while this message was being decoded. + releaseCompletedClaim(message) return Effect.void } if (error.success._tag === "MailboxFull") { From 83c2581ddf8d09d4d991157274546bb850eaa23a Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Sun, 20 Sep 2026 01:48:20 +0000 Subject: [PATCH 04/12] test(cluster): cover reset claim contracts and reply publication races --- .../test/cluster/MessageStorage.test.ts | 46 ++++++++ packages/effect/test/cluster/Sharding.test.ts | 102 ++++++++++++++++++ .../SqlMessageStorage.integration.test.ts | 54 ++++++++++ 3 files changed, 202 insertions(+) diff --git a/packages/effect/test/cluster/MessageStorage.test.ts b/packages/effect/test/cluster/MessageStorage.test.ts index fffb2841772..720fc0ae347 100644 --- a/packages/effect/test/cluster/MessageStorage.test.ts +++ b/packages/effect/test/cluster/MessageStorage.test.ts @@ -104,6 +104,52 @@ describe("MessageStorage", () => { expect(messages).toHaveLength(0) }).pipe(Effect.provide(MessageStorage.MemoryDriver.layer))) + it.effect("resetRequests with no IDs leaves existing claims untouched", () => + Effect.gen(function*() { + const storage = yield* MessageStorage.MessageStorage + const request = yield* makeRequest() + yield* storage.saveRequest(request) + expect(yield* storage.unprocessedMessages([request.envelope.address.shardId])).toHaveLength(1) + yield* storage.resetRequests([]) + const driver = yield* MessageStorage.MemoryDriver + yield* driver.encoded.resetRequests([]) + expect(yield* storage.unprocessedMessages([request.envelope.address.shardId])).toHaveLength(0) + }).pipe(Effect.provide(MemoryLayer))) + + it.effect("resetRequests releases only the selected request at a shared address", () => + Effect.gen(function*() { + const storage = yield* MessageStorage.MessageStorage + const selected = yield* makeRequest() + const unrelated = yield* makeRequest() + yield* storage.saveRequest(selected) + yield* storage.saveRequest(unrelated) + const shards = [selected.envelope.address.shardId] + expect(yield* storage.unprocessedMessages(shards)).toHaveLength(2) + yield* storage.resetRequests([selected.envelope.requestId]) + const messages = yield* storage.unprocessedMessages(shards) + expect(messages.map((message) => message.envelope.requestId)).toEqual([selected.envelope.requestId]) + expect(yield* storage.unprocessedMessages(shards)).toHaveLength(0) + }).pipe(Effect.provide(MemoryLayer))) + + it.effect("resetRequests preserves chunk replies, exit replies, and completed state", () => + Effect.gen(function*() { + const storage = yield* MessageStorage.MessageStorage + const streaming = yield* makeRequest({ rpc: StreamRpc, payload: StreamRpc.payloadSchema.make({ id: 123 }) }) + const completed = yield* makeRequest() + yield* storage.saveRequest(streaming) + yield* storage.saveRequest(completed) + yield* storage.saveReply(yield* makeChunkReply(streaming)) + yield* storage.saveReply(yield* makeReply(completed)) + const replies = yield* storage.repliesFor([streaming, completed]) + expect(replies).toHaveLength(2) + const shards = [streaming.envelope.address.shardId] + expect(yield* storage.unprocessedMessages(shards)).toHaveLength(1) + yield* storage.resetRequests([streaming.envelope.requestId, completed.envelope.requestId]) + const messages = yield* storage.unprocessedMessages(shards) + expect(messages.map((message) => message.envelope.requestId)).toEqual([streaming.envelope.requestId]) + expect(yield* storage.repliesFor([streaming, completed])).toEqual(replies) + }).pipe(Effect.provide(MemoryLayer))) + it.effect("saves a request", () => 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 bf5f93d5364..4d4b5a89461 100644 --- a/packages/effect/test/cluster/Sharding.test.ts +++ b/packages/effect/test/cluster/Sharding.test.ts @@ -118,6 +118,108 @@ describe.concurrent("Sharding", () => { })))) })) + for (const pauseReply of [false, true]) { + it.effect( + pauseReply + ? "redelivers a remote reset published before active-request cleanup" + : "redelivers a remote reset during a pending storage read", + () => + Effect.gen(function*() { + const readStarted = yield* Deferred.make() + const releaseRead = yield* Deferred.make() + const replyPublished = yield* Deferred.make() + const releaseReply = yield* Deferred.make() + let pauseNextRead = false + let pauseNextReply = pauseReply + const released: Array = [] + + yield* Effect.gen(function*() { + yield* TestClock.adjust(1) + const sharding = yield* Sharding.Sharding + const driver = yield* MessageStorage.MemoryDriver + // A separate adapter shares persistence, but has no access to the + // receiving runner's reply handlers or entity-manager state. + const remoteStorage = yield* MessageStorage.makeEncoded(driver.encoded).pipe( + Effect.provide(Snowflake.layerGenerator.pipe(Layer.provide(ShardingConfig.layerDefaults))) + ) + const state = yield* TestEntityState + const client = (yield* TestEntity.client)("remote-reset-race") + const firstRun = yield* client.RequestWithKey({ key: "run" }).pipe( + Effect.forkChild({ startImmediately: true }) + ) + const request = yield* Queue.take(state.envelopes) + + // Reclaim a genuinely active request: it must neither be delivered + // twice nor have its claim released by completed-request recovery. + yield* remoteStorage.resetRequests([request.requestId]) + yield* sharding.pollStorage + yield* TestClock.adjust(1) + expect(Queue.sizeUnsafe(state.envelopes)).toEqual(0) + expect(released).toEqual([]) + + pauseNextRead = true + yield* sharding.pollStorage + yield* Deferred.await(readStarted) + yield* Queue.offer(state.messages, void 0) + yield* Fiber.join(firstRun) + if (pauseReply) yield* Deferred.await(replyPublished) + yield* TestClock.adjust(1) + + // This is the persistence operation performed by Sharding.reset on + // another runner. Do not call the receiving runner's reset method. + yield* remoteStorage.clearReplies(request.requestId) + yield* Deferred.succeed(releaseRead, void 0) + yield* TestClock.adjust(1) + expect(released).toContain(request.requestId) + if (pauseReply) { + expect(Queue.sizeUnsafe(state.envelopes)).toEqual(0) + yield* Deferred.succeed(releaseReply, void 0) + yield* TestClock.adjust(1) + } + yield* sharding.pollStorage + yield* TestClock.adjust(5000) + expect(Queue.sizeUnsafe(state.envelopes)).toEqual(1) + expect((yield* Queue.take(state.envelopes)).requestId).toEqual(request.requestId) + yield* sharding.pollStorage + yield* TestClock.adjust(5000) + expect(Queue.sizeUnsafe(state.envelopes)).toEqual(0) + }).pipe(Effect.provide(CappedSharding({}, (storage) => ({ + ...storage, + resetRequests: (ids) => + Effect.gen(function*() { + released.push(...ids) + yield* storage.resetRequests(ids) + }), + unprocessedMessages: (shards, options) => + Effect.gen(function*() { + if (pauseNextRead) { + pauseNextRead = false + yield* Deferred.succeed(readStarted, void 0) + yield* Deferred.await(releaseRead) + } + const messages = yield* storage.unprocessedMessages(shards, options) + return messages.map((message) => + message._tag !== "IncomingRequest" ? + message : + new Message.IncomingRequest({ + ...message, + respond: (reply) => + Effect.gen(function*() { + yield* message.respond(reply) + if (pauseNextReply) { + pauseNextReply = false + yield* Deferred.succeed(replyPublished, void 0) + yield* Deferred.await(releaseReply) + } + }) + }) + ) + }) + })))) + }) + ) + } + it.effect("delivers volatile requests directly to the entity", () => Effect.gen(function*() { yield* TestClock.adjust(1) diff --git a/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts b/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts index ab498048e11..207615fceec 100644 --- a/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts +++ b/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts @@ -75,6 +75,60 @@ describe("SqlMessageStorage", () => { })) } + it.effect("resetRequests with no IDs leaves existing claims untouched", () => + Effect.gen(function*() { + yield* truncate + const storage = yield* MessageStorage.MessageStorage + const request = yield* makeRequest() + yield* storage.saveRequest(request) + expect(yield* storage.unprocessedMessages([request.envelope.address.shardId])).toHaveLength(1) + yield* storage.resetRequests([]) + expect(yield* storage.unprocessedMessages([request.envelope.address.shardId])).toHaveLength(0) + })) + + it.effect("resetRequests releases only the selected request at a shared address", () => + Effect.gen(function*() { + yield* truncate + const storage = yield* MessageStorage.MessageStorage + const selected = yield* makeRequest() + const unrelated = yield* makeRequest() + yield* storage.saveRequest(selected) + yield* storage.saveRequest(unrelated) + const shards = [selected.envelope.address.shardId] + expect(yield* storage.unprocessedMessages(shards)).toHaveLength(2) + yield* storage.resetRequests([selected.envelope.requestId]) + const messages = yield* storage.unprocessedMessages(shards) + expect(messages.map((message) => message.envelope.requestId)).toEqual([selected.envelope.requestId]) + expect(yield* storage.unprocessedMessages(shards)).toHaveLength(0) + })) + + it.effect("resetRequests preserves chunk replies, exit replies, and completed state", () => + Effect.gen(function*() { + yield* truncate + const storage = yield* MessageStorage.MessageStorage + const streaming = yield* makeRequest({ rpc: StreamRpc, payload: StreamRpc.payloadSchema.make({ id: 123 }) }) + const completed = yield* makeRequest() + yield* storage.saveRequest(streaming) + yield* storage.saveRequest(completed) + const shards = [streaming.envelope.address.shardId] + expect(yield* storage.unprocessedMessages(shards)).toHaveLength(2) + yield* storage.saveReply(yield* makeChunkReply(streaming)) + yield* storage.saveReply(yield* makeReply(completed)) + const replies = yield* storage.repliesFor([streaming, completed]) + expect(replies).toHaveLength(2) + expect(yield* storage.unprocessedMessages(shards)).toHaveLength(0) + const sql = yield* SqlClient.SqlClient + const processed = yield* sql`SELECT processed FROM cluster_messages ORDER BY rowid` + yield* storage.resetRequests([streaming.envelope.requestId, completed.envelope.requestId]) + const messages = yield* storage.unprocessedMessages(shards) + // SQL keeps unacknowledged chunks out of the read loop too. Claim + // release must not erase either kind of reply to force redelivery. + expect(messages).toHaveLength(0) + expect(yield* storage.repliesFor([streaming, completed])).toEqual(replies) + expect(yield* sql`SELECT processed FROM cluster_messages ORDER BY rowid`).toEqual(processed) + yield* truncate + })) + it.effect("saveRequest", () => Effect.gen(function*() { const storage = yield* MessageStorage.MessageStorage From c0b7205330c1875240225b910dc49e7a639a4ae8 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Sun, 20 Sep 2026 02:05:22 +0000 Subject: [PATCH 05/12] test(cluster): reproduce stream claim churn and release failure coupling --- packages/effect/test/cluster/Sharding.test.ts | 131 ++++++++++++++++++ 1 file changed, 131 insertions(+) diff --git a/packages/effect/test/cluster/Sharding.test.ts b/packages/effect/test/cluster/Sharding.test.ts index 4d4b5a89461..6946eddb57b 100644 --- a/packages/effect/test/cluster/Sharding.test.ts +++ b/packages/effect/test/cluster/Sharding.test.ts @@ -53,6 +53,137 @@ import { User } from "./TestEntity.ts" +// Keep the long-lived stream separate from concurrent suites that assert global shard metrics. +describe("Sharding claim release regressions", { concurrent: false }, () => { + it.effect("keeps the claim of an active stream after an acknowledged chunk", () => + Effect.gen(function*() { + const acked = yield* Deferred.make() + const released: Array = [] + const claimed: Array = [] + yield* Effect.gen(function*() { + yield* TestClock.adjust(1) + const sharding = yield* Sharding.Sharding + const state = yield* TestEntityState + const client = (yield* TestEntity.client)("active-stream") + const values: Array = [] + const stream = yield* client.StreamWithKey({ key: "run" }).pipe( + Stream.runForEach((value) => Effect.sync(() => values.push(value))), + Effect.forkChild({ startImmediately: true }) + ) + yield* Queue.offer(state.streamMessages, void 0) + yield* TestClock.adjust(1000) + yield* Deferred.await(acked) + expect(values).toEqual([0]) + expect(stream.pollUnsafe()).toBeUndefined() + claimed.length = 0 + released.length = 0 + + // An acked, still-open stream is eligible for SQL selection once its + // original claim expires. Memory storage models the claim timeout, + // but not SQL's unacked-reply exclusion; await the ack explicitly. + yield* TestClock.adjust("10 minutes") + for (let i = 0; i < 3; i++) { + yield* sharding.pollStorage + yield* TestClock.adjust(1) + } + expect(claimed.length).toBeGreaterThan(0) + expect(stream.pollUnsafe()).toBeUndefined() + expect(values).toEqual([0]) + assert.strictEqual(released.length, 0, "active stream claim must not be released after a chunk") + assert.strictEqual(claimed.length, 1, "active stream must not be reclaimed on every poll") + }).pipe(Effect.provide(CappedSharding({}, (storage) => ({ + ...storage, + saveEnvelope: (message) => + storage.saveEnvelope(message).pipe( + Effect.tap(() => message.envelope._tag === "AckChunk" ? Deferred.succeed(acked, void 0) : Effect.void) + ), + resetRequests: (ids) => + Effect.gen(function*() { + released.push(...ids) + yield* storage.resetRequests(ids) + }), + unprocessedMessages: (shards, options) => + storage.unprocessedMessages(shards, options).pipe( + Effect.tap((messages) => + Effect.sync(() => { + for (const message of messages) { + if (message._tag === "IncomingRequest") claimed.push(message.envelope.requestId) + } + }) + ) + ) + })))) + })) + + it.effect("releases capped addresses even when targeted claim release fails", () => + Effect.gen(function*() { + const readStarted = yield* Deferred.make() + const releaseRead = yield* Deferred.make() + let pauseNextRead = false + const failedReleases: Array = [] + const cappedReleases: Array = [] + const claimed: Array = [] + yield* Effect.gen(function*() { + yield* TestClock.adjust(1) + const sharding = yield* Sharding.Sharding + const state = yield* TestEntityState + const client = yield* TestEntity.client + const firstRun = yield* client("completed").RequestWithKey({ key: "run" }).pipe( + Effect.forkChild({ startImmediately: true }) + ) + const request = yield* Queue.take(state.envelopes) + pauseNextRead = true + yield* sharding.pollStorage + yield* Deferred.await(readStarted) + yield* Queue.offer(state.messages, void 0) + yield* Fiber.join(firstRun) + yield* TestClock.adjust(1) + yield* sharding.reset(request.requestId) + yield* saveGetUserRequest("capped", 42) + + // The unrestricted query started below capacity. Fill the second + // resident slot while it is paused, so its batch contains both a + // completed request to release and a newly capped address. + yield* client("resident").NeverVolatile().pipe(Effect.forkChild({ startImmediately: true })) + yield* Queue.take(state.envelopes) + expect(yield* sharding.activeEntityCount).toEqual(2) + yield* Deferred.succeed(releaseRead, void 0) + yield* TestClock.adjust(1) + expect(claimed).toContain("completed") + expect(claimed).toContain("capped") + expect(failedReleases).toEqual([request.requestId]) + assert.deepStrictEqual( + cappedReleases.map((address) => address.entityId), + ["capped"], + "failed targeted release must not skip capped-address release from the same batch" + ) + }).pipe(Effect.provide(CappedSharding({ maxResidentEntities: 2 }, (storage) => ({ + ...storage, + resetRequests: (ids) => + Effect.suspend(() => { + failedReleases.push(...ids) + return Effect.fail(new ClusterError.PersistenceError({ cause: "injected targeted release failure" })) + }), + resetAddresses: (addresses) => + Effect.gen(function*() { + cappedReleases.push(...addresses) + yield* storage.resetAddresses(addresses) + }), + unprocessedMessages: (shards, options) => + Effect.gen(function*() { + if (pauseNextRead) { + pauseNextRead = false + yield* Deferred.succeed(readStarted, void 0) + yield* Deferred.await(releaseRead) + } + const messages = yield* storage.unprocessedMessages(shards, options) + for (const message of messages) claimed.push(message.envelope.address.entityId) + return messages + }) + })))) + })) +}) + describe.concurrent("Sharding", () => { it.effect("redelivers a request reset while an asynchronous storage read is pending", () => Effect.gen(function*() { From ce1d7284d9eb444c3896d79450a0e0b4620e1e17 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Sun, 20 Sep 2026 02:09:29 +0000 Subject: [PATCH 06/12] fix(cluster): retain active stream claims and isolate release errors --- .changeset/cluster-reset-claims.md | 2 ++ .../effect/src/unstable/cluster/MessageStorage.ts | 3 +++ packages/effect/src/unstable/cluster/Sharding.ts | 6 +++--- .../src/unstable/cluster/internal/entityManager.ts | 11 ++++++++++- 4 files changed, 18 insertions(+), 4 deletions(-) diff --git a/.changeset/cluster-reset-claims.md b/.changeset/cluster-reset-claims.md index 94fabe11ce9..bd00768466a 100644 --- a/.changeset/cluster-reset-claims.md +++ b/.changeset/cluster-reset-claims.md @@ -3,3 +3,5 @@ --- Release storage claims for deduplicated cluster requests so concurrent workflow resets can resume on the next poll instead of waiting for the ten-minute claim timeout. + +Custom message storage drivers must implement the new required `resetRequests` operation on `MessageStorage.Encoded`. It releases only the specified request claims, preserving replies, processed state, and unrelated requests. This adds a required member to the exported storage interfaces despite the patch release classification on the rc line. diff --git a/packages/effect/src/unstable/cluster/MessageStorage.ts b/packages/effect/src/unstable/cluster/MessageStorage.ts index c88d6e38e0a..8f46f7074d4 100644 --- a/packages/effect/src/unstable/cluster/MessageStorage.ts +++ b/packages/effect/src/unstable/cluster/MessageStorage.ts @@ -1085,6 +1085,9 @@ export class MemoryDriver extends Context.Service()("effect/cluste if (claimedAt !== undefined && claimedAt > now - claimExpirationMillis) { continue } + // Unlike SQL storage, the memory driver does not exclude requests + // with unacknowledged chunk replies. It models claim expiry, not + // the SQL reply-exclusion predicate or concurrent transactions. messages.push({ envelope, lastSentReply: Option.fromNullishOr(entry.replies[entry.replies.length - 1]) diff --git a/packages/effect/src/unstable/cluster/Sharding.ts b/packages/effect/src/unstable/cluster/Sharding.ts index 940921b0382..adc73b90577 100644 --- a/packages/effect/src/unstable/cluster/Sharding.ts +++ b/packages/effect/src/unstable/cluster/Sharding.ts @@ -645,13 +645,13 @@ const make = Effect.gen(function*() { }) // A request is deduplicated while it is running, but also while its - // reply is being published and, until the next read, after it has + // exit reply is being published and, until the next read, after it has // completed. A reset (possibly from another runner) can race with that // window, so release the claim this read just took on a completed // request. Otherwise the reset is only delivered once the claim expires. const releaseCompletedClaim = (message: Message.Incoming) => { const state = entityManagers.get(message.envelope.address.entityType) - if (!state?.manager.isProcessingFor(message, { excludeReplies: true })) { + if (!state?.manager.isProcessingFor(message, { excludeCompleted: true })) { resetRequestIds.push(message.envelope.requestId) } } @@ -870,7 +870,7 @@ const make = Effect.gen(function*() { yield* storageReadLock.release(1) if (resetRequestIds.length > 0) { - yield* storage.resetRequests(resetRequestIds) + yield* Effect.ignore(storage.resetRequests(resetRequestIds)) resetRequestIds.length = 0 } diff --git a/packages/effect/src/unstable/cluster/internal/entityManager.ts b/packages/effect/src/unstable/cluster/internal/entityManager.ts index 3bcbf1bb22f..65b0a0f5dfe 100644 --- a/packages/effect/src/unstable/cluster/internal/entityManager.ts +++ b/packages/effect/src/unstable/cluster/internal/entityManager.ts @@ -55,6 +55,7 @@ export interface EntityManager { readonly isProcessingFor: (message: Message.Incoming, options?: { readonly excludeReplies?: boolean + readonly excludeCompleted?: boolean }) => boolean readonly clearProcessed: () => void @@ -90,6 +91,7 @@ export type EntityState = { readonly rpc: Rpc.AnyWithProps readonly message: Message.IncomingRequestLocal sentReply: boolean + sentExit: boolean lastSentChunk: Option.Option> sequence: number /** Set when the request should not outlive its caller. */ @@ -233,6 +235,7 @@ export const make = Effect.fnUntraced(function*< if (!request) return Effect.void request.sentReply = true + request.sentExit = true if ( isShuttingDown && @@ -530,6 +533,7 @@ export const make = Effect.fnUntraced(function*< rpc, message, sentReply: false, + sentExit: false, lastSentChunk: Option.filter( message.lastSentReply, (reply): reply is Reply.Chunk => reply._tag === "Chunk" @@ -645,7 +649,10 @@ export const make = Effect.fnUntraced(function*< return Effect.flatMap(Fiber.joinAll(fibers), loop) }), isProcessingFor(message, options) { - if (options?.excludeReplies !== true && processedRequestIds.has(message.envelope.requestId)) { + if ( + options?.excludeReplies !== true && options?.excludeCompleted !== true && + processedRequestIds.has(message.envelope.requestId) + ) { return true } const state = activeServers.get(message.envelope.address.entityId) @@ -655,6 +662,8 @@ export const make = Effect.fnUntraced(function*< return false } else if (options?.excludeReplies && request.sentReply) { return false + } else if (options?.excludeCompleted && request.sentExit) { + return false } return true }, From e55a3a0f6e88c32108b73ccaf9ee86fcece70cbb Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Sun, 20 Sep 2026 02:18:45 +0000 Subject: [PATCH 07/12] test(cluster): reproduce claim churn after uninterruptible restart --- packages/effect/test/cluster/Sharding.test.ts | 89 +++++++++++++++++++ 1 file changed, 89 insertions(+) diff --git a/packages/effect/test/cluster/Sharding.test.ts b/packages/effect/test/cluster/Sharding.test.ts index 6946eddb57b..4d1abedb6b5 100644 --- a/packages/effect/test/cluster/Sharding.test.ts +++ b/packages/effect/test/cluster/Sharding.test.ts @@ -55,6 +55,95 @@ import { // Keep the long-lived stream separate from concurrent suites that assert global shard metrics. describe("Sharding claim release regressions", { concurrent: false }, () => { + it.effect("keeps the claim of an active uninterruptible request after restarting", () => + Effect.gen(function*() { + const started = yield* Queue.make() + let starts = 0 + let active = 0 + const released: Array = [] + const claimed: Array = [] + const Restarting = Entity.make("ClaimRestart", [ + Rpc.make("Run") + .annotate(ClusterSchema.Persisted, true) + .annotate(ClusterSchema.Uninterruptible, true) + ]) + const layer = Restarting.toLayer({ + Run: () => + Effect.gen(function*() { + starts++ + active++ + yield* Queue.offer(started, starts) + return yield* Effect.never + }).pipe(Effect.ensuring(Effect.sync(() => active--))) + }).pipe(Layer.provideMerge(CappedSharding({}, (storage) => ({ + ...storage, + resetRequests: (ids) => + Effect.gen(function*() { + released.push(...ids) + yield* storage.resetRequests(ids) + }), + unprocessedMessages: (shards, options) => + storage.unprocessedMessages(shards, options).pipe( + Effect.tap((messages) => + Effect.sync(() => { + for (const message of messages) { + if (message._tag === "IncomingRequest") claimed.push(message.envelope.requestId) + } + }) + ) + ) + })))) + + yield* Effect.gen(function*() { + yield* TestClock.adjust(1) + const sharding = yield* Sharding.Sharding + const driver = yield* MessageStorage.MemoryDriver + const client = (yield* Restarting.client)("restart") + const running = yield* client.Run().pipe(Effect.forkChild({ startImmediately: true })) + assert.strictEqual(yield* Queue.take(started), 1) + const request = driver.journal[0] + assert.strictEqual(request._tag, "Request") + + // Deliver a persisted interrupt, not a local caller cancellation. The + // uninterruptible annotation makes the manager restart this same entry. + yield* driver.encoded.saveEnvelope({ + envelope: { + _tag: "Interrupt", + id: String(yield* sharding.getSnowflake), + requestId: request.requestId, + address: request.address + }, + primaryKey: null, + deliverAt: null + }) + yield* sharding.pollStorage + yield* TestClock.adjust(1) + assert.strictEqual(yield* Queue.take(started), 2, "handler must restart after the stored interrupt") + assert.strictEqual(active, 1) + expect(running.pollUnsafe()).toBeUndefined() + claimed.length = 0 + released.length = 0 + + // Make the running request eligible for storage selection, then check + // that repeated polls deduplicate it without repeatedly releasing it. + yield* TestClock.adjust("10 minutes") + for (let i = 0; i < 3; i++) { + yield* sharding.pollStorage + yield* TestClock.adjust(1) + } + expect(claimed).toContain(Snowflake.Snowflake(request.requestId)) + assert.strictEqual(starts, 2, "polling must not deliver the restarted request again") + assert.strictEqual(active, 1, "restarted handler must remain active") + expect(running.pollUnsafe()).toBeUndefined() + assert.strictEqual(released.length, 0, "restarted active request claim must not be released") + assert.deepStrictEqual( + claimed, + [Snowflake.Snowflake(request.requestId)], + "restarted request must not be reclaimed on every poll" + ) + }).pipe(Effect.provide(layer)) + })) + it.effect("keeps the claim of an active stream after an acknowledged chunk", () => Effect.gen(function*() { const acked = yield* Deferred.make() From 2f382371ecae8f7318c15692c474b9426c6c7b42 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Sun, 20 Sep 2026 02:21:05 +0000 Subject: [PATCH 08/12] fix(cluster): clear exit flag before uninterruptible restart --- packages/effect/src/unstable/cluster/internal/entityManager.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/effect/src/unstable/cluster/internal/entityManager.ts b/packages/effect/src/unstable/cluster/internal/entityManager.ts index 65b0a0f5dfe..f7b640b430d 100644 --- a/packages/effect/src/unstable/cluster/internal/entityManager.ts +++ b/packages/effect/src/unstable/cluster/internal/entityManager.ts @@ -256,6 +256,7 @@ export const make = Effect.fnUntraced(function*< (isShuttingDown || isUninterruptibleForServer(request.message.annotations)) ) { if (!isShuttingDown) { + request.sentExit = false return server.write( 0, { From 9b1c1cf9bef924d0da156abee29c73247f6cd0a5 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Sun, 20 Sep 2026 02:29:56 +0000 Subject: [PATCH 09/12] test(cluster): cover active request claims after defect replay --- packages/effect/test/cluster/Sharding.test.ts | 97 +++++++++++++++++++ 1 file changed, 97 insertions(+) diff --git a/packages/effect/test/cluster/Sharding.test.ts b/packages/effect/test/cluster/Sharding.test.ts index 4d1abedb6b5..a212fd3b52a 100644 --- a/packages/effect/test/cluster/Sharding.test.ts +++ b/packages/effect/test/cluster/Sharding.test.ts @@ -14,6 +14,7 @@ import { MutableRef, Option, Queue, + Schedule, Schema, Stream } from "effect" @@ -55,6 +56,102 @@ import { // Keep the long-lived stream separate from concurrent suites that assert global shard metrics. describe("Sharding claim release regressions", { concurrent: false }, () => { + it.effect("keeps the claim of an active request replayed after an entity defect", () => + Effect.gen(function*() { + const started = yield* Queue.make() + let starts = 0 + let active = 0 + let layerBuilds = 0 + let defectAttempts = 0 + const released: Array = [] + const claimed: Array = [] + const Replaying = Entity.make("ClaimDefectReplay", [ + Rpc.make("Run"), + Rpc.make("Defect") + ]).annotateRpcs(ClusterSchema.Persisted, true) + const layer = Replaying.toLayer( + Effect.sync(() => { + layerBuilds++ + return Replaying.of({ + Run: () => + Rpc.fork( + Effect.gen(function*() { + starts++ + active++ + yield* Queue.offer(started, starts) + return yield* Effect.never + }).pipe(Effect.ensuring(Effect.sync(() => active--))) + ), + Defect: () => + Effect.suspend(() => { + defectAttempts++ + return defectAttempts === 1 ? Effect.die("restart entity") : Effect.void + }) + }) + }), + { defectRetryPolicy: Schedule.forever } + ).pipe(Layer.provideMerge(CappedSharding({}, (storage) => ({ + ...storage, + resetRequests: (ids) => + Effect.gen(function*() { + released.push(...ids) + yield* storage.resetRequests(ids) + }), + unprocessedMessages: (shards, options) => + storage.unprocessedMessages(shards, options).pipe( + Effect.tap((messages) => + Effect.sync(() => { + for (const message of messages) { + if (message._tag === "IncomingRequest") claimed.push(message.envelope.requestId) + } + }) + ) + ) + })))) + + yield* Effect.gen(function*() { + yield* TestClock.adjust(1) + const sharding = yield* Sharding.Sharding + const driver = yield* MessageStorage.MemoryDriver + const client = (yield* Replaying.client)("replay") + const running = yield* client.Run().pipe(Effect.forkChild({ startImmediately: true })) + yield* TestClock.adjust(1) + assert.strictEqual(yield* Queue.take(started), 1) + const request = driver.journal[0] + assert.strictEqual(request._tag, "Request") + + // A different handler defects, interrupting Run and rebuilding the + // entity. Run must be replayed using the same active-request entry. + yield* client.Defect() + assert.strictEqual(yield* Queue.take(started), 2, "in-flight handler must replay after the defect") + assert.strictEqual(layerBuilds, 2) + assert.strictEqual(defectAttempts, 2) + assert.strictEqual(active, 1) + expect(running.pollUnsafe()).toBeUndefined() + claimed.length = 0 + released.length = 0 + + // Expire the original claim so storage selects the still-running + // request. Later polls must neither release it nor deliver it again. + yield* TestClock.adjust("10 minutes") + for (let i = 0; i < 3; i++) { + yield* sharding.pollStorage + yield* TestClock.adjust(1) + } + expect(claimed).toContain(Snowflake.Snowflake(request.requestId)) + assert.strictEqual(starts, 2, "polling must not deliver the replayed request again") + assert.strictEqual(layerBuilds, 2, "polling must not restart the entity again") + assert.strictEqual(active, 1, "replayed handler must remain active") + expect(running.pollUnsafe()).toBeUndefined() + assert.strictEqual(released.length, 0, "replayed active request claim must not be released") + assert.deepStrictEqual( + claimed, + [Snowflake.Snowflake(request.requestId)], + "replayed request must not be reclaimed on every poll" + ) + }).pipe(Effect.provide(layer)) + })) + it.effect("keeps the claim of an active uninterruptible request after restarting", () => Effect.gen(function*() { const started = yield* Queue.make() From ce95b200eb601cee22d6a1b514824cf8e72e184f Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Sun, 20 Sep 2026 02:32:34 +0000 Subject: [PATCH 10/12] fix(cluster): clear exit state before defect replay --- packages/effect/src/unstable/cluster/internal/entityManager.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/effect/src/unstable/cluster/internal/entityManager.ts b/packages/effect/src/unstable/cluster/internal/entityManager.ts index f7b640b430d..3b752a4122b 100644 --- a/packages/effect/src/unstable/cluster/internal/entityManager.ts +++ b/packages/effect/src/unstable/cluster/internal/entityManager.ts @@ -350,6 +350,7 @@ export const make = Effect.fnUntraced(function*< const request = activeRequests.get(id) if (!request) continue const { lastSentChunk, message } = request + request.sentExit = false yield* server.write( 0, { From 625ea250dd18e1268ddb6f6786a46a6fff40fffc Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Sun, 20 Sep 2026 08:16:56 +0000 Subject: [PATCH 11/12] docs(changeset): shorten cluster reset note --- .changeset/cluster-reset-claims.md | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/.changeset/cluster-reset-claims.md b/.changeset/cluster-reset-claims.md index bd00768466a..0dee0e39ea3 100644 --- a/.changeset/cluster-reset-claims.md +++ b/.changeset/cluster-reset-claims.md @@ -2,6 +2,4 @@ "effect": patch --- -Release storage claims for deduplicated cluster requests so concurrent workflow resets can resume on the next poll instead of waiting for the ten-minute claim timeout. - -Custom message storage drivers must implement the new required `resetRequests` operation on `MessageStorage.Encoded`. It releases only the specified request claims, preserving replies, processed state, and unrelated requests. This adds a required member to the exported storage interfaces despite the patch release classification on the rc line. +Fix cluster workflows stalling for ten minutes after a request reset. From dfa11c2c54798136f6195df8bdd49e0fbbf7443e Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Sun, 20 Sep 2026 09:00:34 +0000 Subject: [PATCH 12/12] docs(cluster): simplify reset comments --- .changeset/cluster-reset-claims.md | 2 +- .../src/unstable/cluster/MessageStorage.ts | 14 ++----- .../effect/src/unstable/cluster/Sharding.ts | 9 ++--- packages/effect/test/cluster/Sharding.test.ts | 40 +++++-------------- .../SqlMessageStorage.integration.test.ts | 3 +- 5 files changed, 19 insertions(+), 49 deletions(-) diff --git a/.changeset/cluster-reset-claims.md b/.changeset/cluster-reset-claims.md index 0dee0e39ea3..1f8143ff589 100644 --- a/.changeset/cluster-reset-claims.md +++ b/.changeset/cluster-reset-claims.md @@ -2,4 +2,4 @@ "effect": patch --- -Fix cluster workflows stalling for ten minutes after a request reset. +Fix cluster workflows stalling after request resets. diff --git a/packages/effect/src/unstable/cluster/MessageStorage.ts b/packages/effect/src/unstable/cluster/MessageStorage.ts index 8f46f7074d4..be9200afc19 100644 --- a/packages/effect/src/unstable/cluster/MessageStorage.ts +++ b/packages/effect/src/unstable/cluster/MessageStorage.ts @@ -178,10 +178,7 @@ export class MessageStorage extends Context.Service Effect.Effect - /** - * Release claims for the provided requests without changing their replies or - * processed state. - */ + /** Release request claims without changing replies or processed state. */ readonly resetRequests: ( requestIds: ReadonlyArray ) => Effect.Effect @@ -420,10 +417,7 @@ export type Encoded = { PersistenceError > - /** - * Release claims for the provided requests without changing their replies or - * processed state. - */ + /** Release request claims without changing replies or processed state. */ readonly resetRequests: ( requestIds: ReadonlyArray ) => Effect.Effect @@ -1085,9 +1079,7 @@ export class MemoryDriver extends Context.Service()("effect/cluste if (claimedAt !== undefined && claimedAt > now - claimExpirationMillis) { continue } - // Unlike SQL storage, the memory driver does not exclude requests - // with unacknowledged chunk replies. It models claim expiry, not - // the SQL reply-exclusion predicate or concurrent transactions. + // Memory storage models claim expiry, but not SQL reply filtering or transactions. messages.push({ envelope, lastSentReply: Option.fromNullishOr(entry.replies[entry.replies.length - 1]) diff --git a/packages/effect/src/unstable/cluster/Sharding.ts b/packages/effect/src/unstable/cluster/Sharding.ts index adc73b90577..64d56ed5425 100644 --- a/packages/effect/src/unstable/cluster/Sharding.ts +++ b/packages/effect/src/unstable/cluster/Sharding.ts @@ -644,11 +644,8 @@ const make = Effect.gen(function*() { deliveredThisRead = true }) - // A request is deduplicated while it is running, but also while its - // exit reply is being published and, until the next read, after it has - // completed. A reset (possibly from another runner) can race with that - // window, so release the claim this read just took on a completed - // request. Otherwise the reset is only delivered once the claim expires. + // A completed request may remain deduplicated after storage claims its reset. + // Release the claim so the reset can be redelivered immediately. const releaseCompletedClaim = (message: Message.Incoming) => { const state = entityManagers.get(message.envelope.address.entityType) if (!state?.manager.isProcessingFor(message, { excludeCompleted: true })) { @@ -753,7 +750,7 @@ const make = Effect.gen(function*() { })) } if (error.success._tag === "AlreadyProcessingMessage") { - // The request completed while this message was being decoded. + // Completion raced with decoding. releaseCompletedClaim(message) return Effect.void } diff --git a/packages/effect/test/cluster/Sharding.test.ts b/packages/effect/test/cluster/Sharding.test.ts index a212fd3b52a..ca3c06d5b41 100644 --- a/packages/effect/test/cluster/Sharding.test.ts +++ b/packages/effect/test/cluster/Sharding.test.ts @@ -54,7 +54,7 @@ import { User } from "./TestEntity.ts" -// Keep the long-lived stream separate from concurrent suites that assert global shard metrics. +// Isolate the long-lived stream from concurrent shard-metric tests. describe("Sharding claim release regressions", { concurrent: false }, () => { it.effect("keeps the claim of an active request replayed after an entity defect", () => Effect.gen(function*() { @@ -120,8 +120,7 @@ describe("Sharding claim release regressions", { concurrent: false }, () => { const request = driver.journal[0] assert.strictEqual(request._tag, "Request") - // A different handler defects, interrupting Run and rebuilding the - // entity. Run must be replayed using the same active-request entry. + // Defecting another handler rebuilds the entity and replays Run. yield* client.Defect() assert.strictEqual(yield* Queue.take(started), 2, "in-flight handler must replay after the defect") assert.strictEqual(layerBuilds, 2) @@ -131,8 +130,6 @@ describe("Sharding claim release regressions", { concurrent: false }, () => { claimed.length = 0 released.length = 0 - // Expire the original claim so storage selects the still-running - // request. Later polls must neither release it nor deliver it again. yield* TestClock.adjust("10 minutes") for (let i = 0; i < 3; i++) { yield* sharding.pollStorage @@ -201,8 +198,7 @@ describe("Sharding claim release regressions", { concurrent: false }, () => { const request = driver.journal[0] assert.strictEqual(request._tag, "Request") - // Deliver a persisted interrupt, not a local caller cancellation. The - // uninterruptible annotation makes the manager restart this same entry. + // A persisted interrupt restarts the uninterruptible request. yield* driver.encoded.saveEnvelope({ envelope: { _tag: "Interrupt", @@ -221,8 +217,6 @@ describe("Sharding claim release regressions", { concurrent: false }, () => { claimed.length = 0 released.length = 0 - // Make the running request eligible for storage selection, then check - // that repeated polls deduplicate it without repeatedly releasing it. yield* TestClock.adjust("10 minutes") for (let i = 0; i < 3; i++) { yield* sharding.pollStorage @@ -264,9 +258,7 @@ describe("Sharding claim release regressions", { concurrent: false }, () => { claimed.length = 0 released.length = 0 - // An acked, still-open stream is eligible for SQL selection once its - // original claim expires. Memory storage models the claim timeout, - // but not SQL's unacked-reply exclusion; await the ack explicitly. + // Await the ack because memory storage does not model SQL reply filtering. yield* TestClock.adjust("10 minutes") for (let i = 0; i < 3; i++) { yield* sharding.pollStorage @@ -327,9 +319,7 @@ describe("Sharding claim release regressions", { concurrent: false }, () => { yield* sharding.reset(request.requestId) yield* saveGetUserRequest("capped", 42) - // The unrestricted query started below capacity. Fill the second - // resident slot while it is paused, so its batch contains both a - // completed request to release and a newly capped address. + // Fill capacity during the read so its batch also contains a capped address. yield* client("resident").NeverVolatile().pipe(Effect.forkChild({ startImmediately: true })) yield* Queue.take(state.envelopes) expect(yield* sharding.activeEntityCount).toEqual(2) @@ -388,14 +378,12 @@ describe.concurrent("Sharding", () => { ) const request = yield* Queue.take(state.envelopes) - // The read loop has cleared processedRequestIds, but the asynchronous - // storage query has not selected or claimed any rows yet. + // Pause after clearing processed IDs but before storage claims rows. pauseNextRead = true yield* sharding.pollStorage yield* Deferred.await(readStarted) - // Finish the first attempt during the pending read. Waiting for all - // fibers to settle also lets the manager record the processed ID. + // Complete the request while the read is paused. yield* Queue.offer(state.messages, void 0) yield* Fiber.join(firstRun) yield* TestClock.adjust(1) @@ -403,15 +391,12 @@ describe.concurrent("Sharding", () => { yield* sharding.pollStorage claimed.length = 0 - // The pending query now claims the reset request. The next poll must - // not lose it to the completed attempt's in-memory deduplication. + // The reset must be redelivered before its new claim expires. yield* Deferred.succeed(releaseRead, void 0) yield* TestClock.adjust(5000) assert.include(claimed, request.requestId) const deliveriesBeforeClaimExpiry = Queue.sizeUnsafe(state.envelopes) - // The memory driver uses the same ten-minute claim expiry as SQL. - // Prove the request is still recoverable, rather than a dead handler. yield* TestClock.adjust("10 minutes") assert.strictEqual(Queue.sizeUnsafe(state.envelopes), 1) assert.strictEqual( @@ -454,8 +439,7 @@ describe.concurrent("Sharding", () => { yield* TestClock.adjust(1) const sharding = yield* Sharding.Sharding const driver = yield* MessageStorage.MemoryDriver - // A separate adapter shares persistence, but has no access to the - // receiving runner's reply handlers or entity-manager state. + // Share persistence without sharing runner state. const remoteStorage = yield* MessageStorage.makeEncoded(driver.encoded).pipe( Effect.provide(Snowflake.layerGenerator.pipe(Layer.provide(ShardingConfig.layerDefaults))) ) @@ -466,8 +450,7 @@ describe.concurrent("Sharding", () => { ) const request = yield* Queue.take(state.envelopes) - // Reclaim a genuinely active request: it must neither be delivered - // twice nor have its claim released by completed-request recovery. + // Reclaiming an active request must not redeliver it or release its claim. yield* remoteStorage.resetRequests([request.requestId]) yield* sharding.pollStorage yield* TestClock.adjust(1) @@ -482,8 +465,7 @@ describe.concurrent("Sharding", () => { if (pauseReply) yield* Deferred.await(replyPublished) yield* TestClock.adjust(1) - // This is the persistence operation performed by Sharding.reset on - // another runner. Do not call the receiving runner's reset method. + // Simulate Sharding.reset on another runner. yield* remoteStorage.clearReplies(request.requestId) yield* Deferred.succeed(releaseRead, void 0) yield* TestClock.adjust(1) diff --git a/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts b/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts index 207615fceec..a7fed18b707 100644 --- a/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts +++ b/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts @@ -121,8 +121,7 @@ describe("SqlMessageStorage", () => { const processed = yield* sql`SELECT processed FROM cluster_messages ORDER BY rowid` yield* storage.resetRequests([streaming.envelope.requestId, completed.envelope.requestId]) const messages = yield* storage.unprocessedMessages(shards) - // SQL keeps unacknowledged chunks out of the read loop too. Claim - // release must not erase either kind of reply to force redelivery. + // Replies keep both requests out of the SQL read loop. expect(messages).toHaveLength(0) expect(yield* storage.repliesFor([streaming, completed])).toEqual(replies) expect(yield* sql`SELECT processed FROM cluster_messages ORDER BY rowid`).toEqual(processed)