diff --git a/.changeset/cluster-reset-claims.md b/.changeset/cluster-reset-claims.md new file mode 100644 index 00000000000..1f8143ff589 --- /dev/null +++ b/.changeset/cluster-reset-claims.md @@ -0,0 +1,5 @@ +--- +"effect": patch +--- + +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 3a4575c3f16..be9200afc19 100644 --- a/packages/effect/src/unstable/cluster/MessageStorage.ts +++ b/packages/effect/src/unstable/cluster/MessageStorage.ts @@ -178,6 +178,11 @@ export class MessageStorage extends Context.Service Effect.Effect + /** Release request claims without changing replies or processed state. */ + readonly resetRequests: ( + requestIds: ReadonlyArray + ) => Effect.Effect + /** * Reset the mailbox state for the provided addresses. */ @@ -412,6 +417,11 @@ export type Encoded = { PersistenceError > + /** Release request claims without changing replies or processed state. */ + readonly resetRequests: ( + requestIds: ReadonlyArray + ) => Effect.Effect + /** * Reset the mailbox state for the provided addresses. */ @@ -716,6 +726,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 +857,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, @@ -1067,6 +1079,7 @@ export class MemoryDriver extends Context.Service()("effect/cluste if (claimedAt !== undefined && claimedAt > now - claimExpirationMillis) { continue } + // Memory storage models claim expiry, but not SQL reply filtering or transactions. messages.push({ envelope, lastSentReply: Option.fromNullishOr(entry.replies[entry.replies.length - 1]) @@ -1091,6 +1104,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..64d56ed5425 100644 --- a/packages/effect/src/unstable/cluster/Sharding.ts +++ b/packages/effect/src/unstable/cluster/Sharding.ts @@ -637,12 +637,22 @@ 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(() => { deliveredThisRead = true }) + // 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 })) { + resetRequestIds.push(message.envelope.requestId) + } + } + const processMessages = Effect.whileLoop({ while: () => index < messages.length, step: () => index++, @@ -696,6 +706,7 @@ const make = Effect.gen(function*() { } 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. + if (isProcessing) releaseCompletedClaim(message) return Effect.void } else if (message._tag === "IncomingRequest" && pendingNotifications.has(message.envelope.requestId)) { const entry = pendingNotifications.get(message.envelope.requestId)! @@ -738,6 +749,11 @@ const make = Effect.gen(function*() { defect: Cause.squash(cause) })) } + if (error.success._tag === "AlreadyProcessingMessage") { + // Completion raced with decoding. + releaseCompletedClaim(message) + return Effect.void + } if (error.success._tag === "MailboxFull") { const address = message.envelope.address const state = entityManagers.get(address.entityType) @@ -850,6 +866,11 @@ const make = Effect.gen(function*() { // let the resuming entities check if they are done yield* storageReadLock.release(1) + if (resetRequestIds.length > 0) { + yield* Effect.ignore(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 diff --git a/packages/effect/src/unstable/cluster/internal/entityManager.ts b/packages/effect/src/unstable/cluster/internal/entityManager.ts index 3bcbf1bb22f..3b752a4122b 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 && @@ -253,6 +256,7 @@ export const make = Effect.fnUntraced(function*< (isShuttingDown || isUninterruptibleForServer(request.message.annotations)) ) { if (!isShuttingDown) { + request.sentExit = false return server.write( 0, { @@ -346,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, { @@ -530,6 +535,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 +651,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 +664,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 }, 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 2d4d4b3f778..ca3c06d5b41 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" @@ -53,7 +54,471 @@ import { User } from "./TestEntity.ts" +// 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*() { + 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") + + // 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) + assert.strictEqual(defectAttempts, 2) + assert.strictEqual(active, 1) + expect(running.pollUnsafe()).toBeUndefined() + claimed.length = 0 + released.length = 0 + + 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() + 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") + + // A persisted interrupt restarts the uninterruptible request. + 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 + + 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() + 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 + + // 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 + 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) + + // 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) + 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*() { + 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) + + // Pause after clearing processed IDs but before storage claims rows. + pauseNextRead = true + yield* sharding.pollStorage + yield* Deferred.await(readStarted) + + // Complete the request while the read is paused. + 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 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) + + 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 + }) + })))) + })) + + 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 + // Share persistence without sharing runner 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) + + // Reclaiming an active request must not redeliver it or release its claim. + 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) + + // Simulate Sharding.reset on another runner. + 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..a7fed18b707 100644 --- a/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts +++ b/packages/platform/node/test/cluster/SqlMessageStorage.integration.test.ts @@ -75,6 +75,59 @@ 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) + // 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) + yield* truncate + })) + it.effect("saveRequest", () => Effect.gen(function*() { const storage = yield* MessageStorage.MessageStorage