Visitar URL original
Fix cluster shutdown waiting on queued persisted workflow activities … · Effect-TS/effect@46dc30a · GitHub
Skip to content

Commit 46dc30a

Browse files
effect-boteffect-bot
authored
Fix cluster shutdown waiting on queued persisted workflow activities (#9817)
1 parent 5fbccf4 commit 46dc30a

6 files changed

Lines changed: 464 additions & 27 deletions

File tree

‎.changeset/cluster-queued-activity-shutdown.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
"effect": patch
3+
---
4+
5+
Fix cluster shutdown waiting for `entityTerminationTimeout` when a workflow is waiting on a persisted activity request that has not started. During runner shutdown the caller is now abandoned so the request replays under the next owner, and a durable caller abandoned this way is left for replay instead of being answered with an interrupt. With `preemptiveShutdown: false`, entities drain before the runner is marked as shutting down, so such a caller still waits for `entityTerminationTimeout`.

‎packages/effect/src/cluster/MessageStorage.ts‎

Lines changed: 19 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -133,8 +133,15 @@ export class MessageStorage extends Context.Service<MessageStorage, {
133133

134134
/**
135135
* Unregister the reply handler for the specified message.
136+
*
137+
* By default the waiters are resumed with `EntityNotAssignedToRunner`. With
138+
* `interrupt: true` they are interrupted with a cluster-abandon annotation
139+
* instead, for when the request will not be served by this runner.
136140
*/
137-
readonly unregisterReplyHandler: (requestId: Snowflake.Snowflake) => Effect.Effect<void>
141+
readonly unregisterReplyHandler: (
142+
requestId: Snowflake.Snowflake,
143+
options?: { readonly interrupt?: boolean | undefined }
144+
) => Effect.Effect<void>
138145

139146
/**
140147
* Unregister the reply handlers for the specified ShardId.
@@ -569,6 +576,14 @@ export const make = (
569576
}
570577
const replyHandlers = new Map<Snowflake.Snowflake, Array<ReplyHandler>>()
571578
const replyHandlersShard = new Map<string, Set<ReplyHandler>>()
579+
const resumeUnregistered = (handler: ReplyHandler, options?: { readonly interrupt?: boolean | undefined }) =>
580+
handler.resume(
581+
options?.interrupt ? ClusterAbandon.interrupt : Effect.fail(
582+
new EntityNotAssignedToRunner({
583+
address: handler.message.envelope.address
584+
})
585+
)
586+
)
572587
return MessageStorage.of({
573588
...storage,
574589
// Nested transactions defer notifications to the outermost commit.
@@ -614,19 +629,15 @@ export const make = (
614629
})
615630
})
616631
},
617-
unregisterReplyHandler: (requestId) =>
632+
unregisterReplyHandler: (requestId, options) =>
618633
Effect.sync(() => {
619634
const handlers = replyHandlers.get(requestId)
620635
if (!handlers) return
621636
replyHandlers.delete(requestId)
622637
for (let i = 0; i < handlers.length; i++) {
623638
const handler = handlers[i]
624639
handler.shardSet.delete(handler)
625-
handler.resume(Effect.fail(
626-
new EntityNotAssignedToRunner({
627-
address: handler.message.envelope.address
628-
})
629-
))
640+
resumeUnregistered(handler, options)
630641
}
631642
}),
632643
unregisterShardReplyHandlers: (shardId, options) =>
@@ -637,13 +648,7 @@ export const make = (
637648
replyHandlersShard.delete(id)
638649
shardSet.forEach((handler) => {
639650
replyHandlers.delete(handler.message.envelope.requestId)
640-
handler.resume(
641-
options?.interrupt ? ClusterAbandon.interrupt : Effect.fail(
642-
new EntityNotAssignedToRunner({
643-
address: handler.message.envelope.address
644-
})
645-
)
646-
)
651+
resumeUnregistered(handler, options)
647652
})
648653
}),
649654
saveReply(reply) {

‎packages/effect/src/cluster/Sharding.ts‎

Lines changed: 23 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1541,6 +1541,27 @@ const make = Effect.gen(function*() {
15411541
}
15421542
const clientRequests = new Map<Snowflake.Snowflake, ClientRequestEntry>()
15431543

1544+
// During shutdown, abandon local callers parked on persisted requests that a
1545+
// retiring entity never started. They are served by the next owner. Callers
1546+
// on a runner that keeps running are left waiting, so they stay durable.
1547+
const abandonUndelivered = (
1548+
address: EntityAddress,
1549+
isDelivered: (requestId: Snowflake.Snowflake) => boolean
1550+
): Effect.Effect<void> =>
1551+
Effect.suspend(() => {
1552+
if (!MutableRef.get(isShutdown)) return Effect.void
1553+
const abandon: Array<Effect.Effect<void>> = []
1554+
for (const [requestId, entry] of clientRequests) {
1555+
if (
1556+
!Context.get(entry.message.annotations, Persisted) ||
1557+
!Equal.equals(entry.message.envelope.address, address) ||
1558+
isDelivered(requestId)
1559+
) continue
1560+
abandon.push(storage.unregisterReplyHandler(requestId, { interrupt: true }))
1561+
}
1562+
return Effect.all(abandon, { discard: true })
1563+
})
1564+
15441565
const clients: ResourceMap<
15451566
Entity<any, any>,
15461567
(entityId: string) => RpcClient.RpcClient<
@@ -1895,7 +1916,8 @@ const make = Effect.gen(function*() {
18951916
storage,
18961917
runnerAddress,
18971918
residency,
1898-
sharding
1919+
sharding,
1920+
abandonUndelivered
18991921
}).pipe(
19001922
Effect.provideContext(services.pipe(
19011923
// Registration overrides construction fallbacks, except for runner-owned services.

‎packages/effect/src/cluster/internal/entityManager.ts‎

Lines changed: 28 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ import type { ShardId } from "../ShardId.ts"
3737
import type { Sharding } from "../Sharding.ts"
3838
import { ShardingConfig } from "../ShardingConfig.ts"
3939
import * as Snowflake from "../Snowflake.ts"
40+
import * as ClusterAbandon from "./clusterAbandon.ts"
4041
import { CurrentActivationScope } from "./entityActivation.ts"
4142
import { EntityReaper } from "./entityReaper.ts"
4243
import { acquireEntity, releaseEntity } from "./interruptors.ts"
@@ -144,6 +145,14 @@ export const make = Effect.fnUntraced(function*<
144145
readonly disableFatalDefects?: boolean | undefined
145146
readonly defectRetryPolicy?: Schedule.Schedule<any, unknown, never, never> | undefined
146147
readonly spanAttributes?: Record<string, string> | undefined
148+
/**
149+
* Called once a retiring entity stops accepting requests, so callers
150+
* waiting on requests it has not started can be released.
151+
*/
152+
readonly abandonUndelivered: (
153+
address: EntityAddress,
154+
isDelivered: (requestId: Snowflake.Snowflake) => boolean
155+
) => Effect.Effect<void>
147156
}
148157
) {
149158
const config = yield* ShardingConfig
@@ -326,25 +335,25 @@ export const make = Effect.fnUntraced(function*<
326335
return Effect.void
327336
}
328337

329-
// For durable messages, ignore interrupts during shutdown.
330-
// They will be retried when the entity is restarted.
331-
// Also, if the request is uninterruptible, we ignore the
332-
// interrupt.
333338
const persisted = storageEnabled && Context.get(request.message.annotations, Persisted)
334-
if (
335-
persisted &&
336-
Exit.hasInterrupts(response.exit) &&
337-
(isShuttingDown || request.terminating || isUninterruptibleForServer(request.message.annotations))
338-
) {
339-
if (!isShuttingDown && !request.terminating) {
339+
if (persisted && Exit.isFailure(response.exit) && Cause.hasInterrupts(response.exit.cause)) {
340+
// A durable request interrupted by shutdown, or abandoned
341+
// because this owner can no longer serve it, is left for
342+
// replay instead of being answered.
343+
const abandoned = ClusterAbandon.isCause(response.exit.cause) &&
344+
(!isActive() || !options.sharding.hasShardId(address.shardId))
345+
if (isShuttingDown || request.terminating || abandoned) {
346+
activeRequests.delete(Snowflake.Snowflake(response.requestId))
347+
return options.storage.unregisterReplyHandler(request.message.envelope.requestId)
348+
}
349+
// An uninterruptible request ignores the interrupt and runs again.
350+
if (isUninterruptibleForServer(request.message.annotations)) {
340351
request.sentExit = false
341352
return server.write(0, requestEnvelope(request), requestWriteOptions(request)).pipe(
342353
Effect.setContext(handlerContext),
343354
Effect.forkIn(handlerScope)
344355
)
345356
}
346-
activeRequests.delete(Snowflake.Snowflake(response.requestId))
347-
return options.storage.unregisterReplyHandler(request.message.envelope.requestId)
348357
}
349358
const save = retryRespond(
350359
4,
@@ -539,6 +548,13 @@ export const make = Effect.fnUntraced(function*<
539548
request.terminating = true
540549
yield* state.write(0, { _tag: "Interrupt", requestId: requestId as any, interruptors: [] })
541550
}
551+
// A handler this drain waits for may be parked on a request the
552+
// retired entity will never start, such as its own queued request.
553+
// Release callers before `Eof`, which waits for handler acquisition.
554+
yield* options.abandonUndelivered(
555+
address,
556+
(requestId) => activeRequests.get(requestId)?.delivered === true
557+
)
542558
yield* state.write(0, { _tag: "Eof" })
543559
}).pipe(
544560
Effect.andThen(endLatch.await),

‎packages/effect/test/cluster/ClusterWorkflowEngine.test.ts‎

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ import {
2525
EntityAddress,
2626
EntityId,
2727
EntityType,
28+
type Message,
2829
MessageStorage,
2930
RunnerHealth,
3031
Runners,
@@ -220,6 +221,127 @@ describe.concurrent("ClusterWorkflowEngine", () => {
220221
}).pipe(Effect.provide(layer()))
221222
}))
222223

224+
it.effect("shuts down without waiting for a queued activity and replays it after restart", () =>
225+
Effect.gen(function*() {
226+
const workflow = Workflow.make("QueuedActivityShutdown", {
227+
payload: {},
228+
idempotencyKey: () => "one"
229+
})
230+
let executions = 0
231+
const step = Activity.make({
232+
name: "step",
233+
execute: Effect.sync(() => {
234+
executions++
235+
})
236+
})
237+
const storage = yield* makeSharedStorage
238+
const queued = yield* Latch.make()
239+
const isActivity = (message: Message.Incoming<any>) =>
240+
message.envelope._tag === "Request" && message.envelope.tag === "activity"
241+
// The activity request is persisted, but the runner never reads it back.
242+
const queuedStorage = Layer.succeed(MessageStorage.MessageStorage, {
243+
...storage,
244+
saveRequest: (message) =>
245+
Effect.tap(storage.saveRequest(message), () => message.rpc._tag === "activity" ? queued.open : Effect.void),
246+
unprocessedMessages: (shardIds, options) =>
247+
Effect.map(
248+
storage.unprocessedMessages(shardIds, options),
249+
(messages) => messages.filter((m) => !isActivity(m))
250+
),
251+
unprocessedMessagesById: (ids) =>
252+
Effect.map(storage.unprocessedMessagesById(ids), (messages) => messages.filter((m) => !isActivity(m)))
253+
})
254+
const layer = (storageLayer: Layer.Layer<MessageStorage.MessageStorage>) =>
255+
workflow.toLayer(() => step).pipe(
256+
Layer.provideMerge(makeTestWorkflowEngine({
257+
storageLayer,
258+
config: { entityTerminationTimeout: Duration.minutes(1) }
259+
}))
260+
)
261+
262+
const scope = yield* Scope.make()
263+
const context = yield* Layer.buildWithScope(layer(queuedStorage), scope)
264+
const executionId = yield* workflow.execute({}, { discard: true }).pipe(Effect.provide(context))
265+
yield* queued.await
266+
const shutdown = yield* Effect.forkChild(Scope.close(scope, Exit.void), { startImmediately: true })
267+
for (let i = 0; i < 100 && shutdown.pollUnsafe() === undefined; i++) yield* TestClock.adjust(10)
268+
const closedBeforeTimeout = shutdown.pollUnsafe() !== undefined
269+
// Let a stalled shutdown finish, so the runner is released before asserting.
270+
yield* TestClock.adjust(Duration.minutes(1))
271+
yield* Fiber.join(shutdown)
272+
assert.isTrue(closedBeforeTimeout, "shutdown must not wait for entityTerminationTimeout")
273+
assert.strictEqual(executions, 0)
274+
275+
yield* Effect.gen(function*() {
276+
assert.deepStrictEqual(
277+
yield* pollUntil(workflow, executionId, "Complete"),
278+
new Workflow.Complete({ exit: Exit.void })
279+
)
280+
assert.strictEqual(executions, 1)
281+
}).pipe(Effect.provide(layer(Layer.succeed(MessageStorage.MessageStorage, storage))))
282+
}))
283+
284+
it.effect("shutdown waits for a running activity before finalizing its workflow", () =>
285+
Effect.gen(function*() {
286+
const workflow = Workflow.make("RunningActivityShutdown", {
287+
payload: {},
288+
idempotencyKey: () => "one"
289+
})
290+
const events: Array<string> = []
291+
const started = yield* Latch.make()
292+
const release = yield* Latch.make()
293+
const step = Activity.make({
294+
name: "step",
295+
execute: Effect.gen(function*() {
296+
events.push("activity:start")
297+
yield* started.open
298+
yield* release.await
299+
events.push("activity:end")
300+
})
301+
})
302+
const storage = yield* makeSharedStorage
303+
const layer = workflow.toLayer(() =>
304+
Effect.gen(function*() {
305+
yield* Effect.addFinalizer(() =>
306+
Effect.sync(() => {
307+
events.push("finalized")
308+
})
309+
).pipe(Workflow.provideScope)
310+
yield* step
311+
})
312+
).pipe(
313+
Layer.provideMerge(makeTestWorkflowEngine({
314+
storageLayer: Layer.succeed(MessageStorage.MessageStorage, storage),
315+
config: { entityTerminationTimeout: Duration.minutes(1) }
316+
}))
317+
)
318+
319+
const scope = yield* Scope.make()
320+
const context = yield* Layer.buildWithScope(layer, scope)
321+
const executionId = yield* workflow.execute({}, { discard: true }).pipe(Effect.provide(context))
322+
yield* advanceUntil(() => started.isOpen(), "activity must start")
323+
const shutdown = yield* Effect.forkChild(Scope.close(scope, Exit.void), { startImmediately: true })
324+
for (let i = 0; i < 100; i++) yield* TestClock.adjust(10)
325+
const pendingWhileRunning = shutdown.pollUnsafe() === undefined
326+
yield* release.open
327+
for (let i = 0; i < 100 && shutdown.pollUnsafe() === undefined; i++) yield* TestClock.adjust(10)
328+
const closedBeforeTimeout = shutdown.pollUnsafe() !== undefined
329+
// Let a stalled shutdown finish, so the runner is released before asserting.
330+
yield* TestClock.adjust(Duration.minutes(1))
331+
yield* Fiber.join(shutdown)
332+
assert.isTrue(pendingWhileRunning, "shutdown must wait for the running activity")
333+
assert.isTrue(closedBeforeTimeout, "shutdown must finish once the activity completes")
334+
assert.deepStrictEqual(events, ["activity:start", "activity:end", "finalized"])
335+
336+
yield* Effect.gen(function*() {
337+
assert.deepStrictEqual(
338+
yield* pollUntil(workflow, executionId, "Complete"),
339+
new Workflow.Complete({ exit: Exit.void })
340+
)
341+
}).pipe(Effect.provide(layer))
342+
assert.deepStrictEqual(events, ["activity:start", "activity:end", "finalized"])
343+
}))
344+
223345
it.effect("retries an early activity registered while its Suspended reply is still being persisted", () =>
224346
Effect.gen(function*() {
225347
const workflow = Workflow.make("EarlyActivity", {

0 commit comments

Comments
 (0)