1import { assert, describe, it } from "@effect/vitest"
2import { Cause, Clock, DateTime, Effect, Exit, Fiber, Latch, Option, PrimaryKey, Schema, Stream } from "effect"
3import { ClusterSchema, DeliverAt, Entity, EntityId, ShardId } from "effect/cluster"
4import { Rpc, RpcSchema } from "effect/rpc"
5import { type Backend, make } from "./harness.ts"
6
7class KeyedPayload extends Schema.Class<KeyedPayload>("ClusterPersistenceKeyedPayload")({
8 id: Schema.String
9}) {
10 [PrimaryKey.symbol]() {
11 return this.id
12 }
13}
14
15class ScheduledPayload extends Schema.Class<ScheduledPayload>("ClusterPersistenceScheduledPayload")({
16 deliverAt: Schema.Number,
17 id: Schema.String
18}) {
19 [PrimaryKey.symbol]() {
20 return this.id
21 }
22
23 [DeliverAt.symbol]() {
24 return DateTime.makeUnsafe(this.deliverAt)
25 }
26}
27
28const PersistedRpc = Rpc.make("Persisted", {
29 payload: KeyedPayload,
30 success: Schema.String
31})
32
33const StoredReplyRpc = Rpc.make("StoredReply", {
34 payload: KeyedPayload,
35 success: Schema.String
36})
37
38const UninterruptibleRpc = Rpc.make("Uninterruptible", {
39 payload: KeyedPayload,
40 success: Schema.String
41}).annotate(ClusterSchema.Uninterruptible, true)
42
43const VolatileRpc = Rpc.make("Volatile", {
44 payload: KeyedPayload,
45 success: Schema.String
46}).annotate(ClusterSchema.Persisted, false)
47
48const TypedFailureRpc = Rpc.make("TypedFailure", {
49 error: Schema.String,
50 payload: KeyedPayload,
51 success: Schema.Never
52})
53
54const DefectRpc = Rpc.make("Defect", {
55 payload: KeyedPayload,
56 success: Schema.Never
57})
58
59const HealthyRpc = Rpc.make("Healthy", {
60 payload: KeyedPayload,
61 success: Schema.String
62})
63
64const UnencodableRpc = Rpc.make("Unencodable", {
65 payload: KeyedPayload,
66 success: Schema.Unknown
67})
68
69const NonJsonFailureRpc = Rpc.make("NonJsonFailure", {
70 error: Schema.Unknown,
71 payload: KeyedPayload,
72 success: Schema.Never
73})
74
75const StreamedRpc = Rpc.make("Streamed", {
76 payload: KeyedPayload,
77 success: RpcSchema.Stream(Schema.Number, Schema.Never)
78})
79
80const ScheduledRpc = Rpc.make("Scheduled", {
81 payload: ScheduledPayload,
82 success: Schema.Number
83})
84
85const PersistenceEntity = Entity.make("ClusterIntegrationPersistence", [
86 PersistedRpc,
87 StoredReplyRpc,
88 UninterruptibleRpc,
89 VolatileRpc,
90 TypedFailureRpc,
91 DefectRpc,
92 HealthyRpc,
93 UnencodableRpc,
94 NonJsonFailureRpc,
95 StreamedRpc,
96 ScheduledRpc
97]).annotateRpcs(ClusterSchema.Persisted, true)
98
99const freshState = () => ({
100 completedUninterruptible: 0,
101 completedVolatile: 0,
102 counts: new Map<string, number>(),
103 healthyEntered: Latch.makeUnsafe(),
104 healthyGate: Latch.makeUnsafe(),
105 scheduledDeliveries: [] as Array<number>,
106 streamTerminalEntered: Latch.makeUnsafe(),
107 streamTerminalGate: Latch.makeUnsafe(),
108 streamThirdEntered: Latch.makeUnsafe(),
109 streamThirdGate: Latch.makeUnsafe(),
110 uninterruptibleEntered: Latch.makeUnsafe(),
111 uninterruptibleGate: Latch.makeUnsafe(),
112 volatileEntered: Latch.makeUnsafe(),
113 volatileGate: Latch.makeUnsafe()
114})
115
116let state = freshState()
117
118const resetState = () => {
119 state = freshState()
120}
121
122const increment = (tag: string, id: string) => {
123 const key = `${tag}:${id}`
124 const next = (state.counts.get(key) ?? 0) + 1
125 state.counts.set(key, next)
126 return next
127}
128
129const count = (tag: string, id: string) => state.counts.get(`${tag}:${id}`) ?? 0
130
131const assertStoredAsDefect = Effect.fnUntraced(function*<A, E>(
132 request: Effect.Effect<A, E>,
133 tag: string,
134 id: string
135) {
136 for (let attempt = 0; attempt < 2; attempt++) {
137 const exit = yield* request.pipe(Effect.exit)
138 assert.isTrue(Exit.isFailure(exit))
139 if (Exit.isFailure(exit)) {
140 assert.include(Cause.pretty(exit.cause), "MalformedMessage")
141 }
142 }
143 assert.strictEqual(count(tag, id), 1)
144})
145
146const PersistenceEntityLayer = PersistenceEntity.toLayer({
147 Defect: ({ payload }) =>
148 Effect.sync(() => increment("Defect", payload.id)).pipe(
149 Effect.andThen(Effect.die(`defect:${payload.id}`))
150 ),
151 Healthy: Effect.fnUntraced(function*({ payload }) {
152 increment("Healthy", payload.id)
153 if (payload.id.endsWith("-blocked")) {
154 state.healthyEntered.openUnsafe()
155 yield* state.healthyGate.await
156 }
157 return `healthy:${payload.id}`
158 }),
159 NonJsonFailure: ({ payload }) =>
160 Effect.sync(() => increment("NonJsonFailure", payload.id)).pipe(
161 Effect.andThen(Effect.fail(new Error(`non-json:${payload.id}`)))
162 ),
163 Persisted: ({ payload }) =>
164 Effect.sync(() => {
165 increment("Persisted", payload.id)
166 return `persisted:${payload.id}`
167 }),
168 Scheduled: Effect.fnUntraced(function*({ payload }) {
169 increment("Scheduled", payload.id)
170 const deliveredAt = yield* Clock.currentTimeMillis
171 state.scheduledDeliveries.push(deliveredAt)
172 return deliveredAt
173 }),
174 StoredReply: ({ payload }) =>
175 Effect.sync(() => {
176 increment("StoredReply", payload.id)
177 return `stored:${payload.id}`
178 }),
179 Streamed: (request) => {
180 increment("Streamed", request.payload.id)
181 const start = Option.match(request.lastSentChunkValue, {
182 onNone: () => 0,
183 onSome: (value) => value + 1
184 })
185 const values = request.payload.id.endsWith("-terminal-race") ? [0] : [0, 1, 2, 3, 4]
186 const stream = Stream.fromIterable(values.slice(start)).pipe(
187 Stream.mapEffect((value) => {
188 if (request.payload.id.endsWith("-restart") && value === 2) {
189 state.streamThirdEntered.openUnsafe()
190 return Effect.as(state.streamThirdGate.await, value)
191 }
192 return Effect.succeed(value)
193 }),
194 Stream.rechunk(1)
195 )
196 if (!request.payload.id.endsWith("-terminal-race")) return stream
197 return Stream.concat(
198 stream,
199 Stream.fromEffect(
200 Effect.sync(() => state.streamTerminalEntered.openUnsafe()).pipe(
201 Effect.andThen(state.streamTerminalGate.await)
202 )
203 ).pipe(Stream.drain)
204 )
205 },
206 TypedFailure: ({ payload }) =>
207 Effect.sync(() => increment("TypedFailure", payload.id)).pipe(
208 Effect.andThen(Effect.fail(`typed:${payload.id}`))
209 ),
210 Uninterruptible: Effect.fnUntraced(function*({ payload }) {
211 increment("Uninterruptible", payload.id)
212 state.uninterruptibleEntered.openUnsafe()
213 yield* state.uninterruptibleGate.await
214 state.completedUninterruptible++
215 return `uninterruptible:${payload.id}`
216 }),
217 Unencodable: ({ payload }) =>
218 Effect.sync(() => {
219 increment("Unencodable", payload.id)
220 return new Error(`unencodable:${payload.id}`)
221 }),
222 Volatile: Effect.fnUntraced(function*({ payload }) {
223 increment("Volatile", payload.id)
224 state.volatileEntered.openUnsafe()
225 yield* state.volatileGate.await
226 state.completedVolatile++
227 return `volatile:${payload.id}`
228 })
229}, { disableFatalDefects: true })
230
231describe("cluster message persistence integration", () => {
232 for (const backend of ["pg", "mysql"] satisfies ReadonlyArray<Backend>) {
233 it.live(`${backend}: delivers a persisted request sent while its runner is down exactly once`, () =>
234 Effect.gen(function*() {
235 resetState()
236 const cluster = yield* make({ backend, entities: PersistenceEntityLayer })
237 yield* cluster.start(1)
238 yield* cluster.waitForStableAssignments()
239 const owner = yield* cluster.ownerOfEntity(PersistenceEntity, "restart")
240 yield* cluster.kill(owner!)
241
242 const client = yield* cluster.getClient(PersistenceEntity)
243 const replyFiber = yield* client("restart").Persisted(
244 new KeyedPayload({ id: `${backend}-restart` })
245 ).pipe(Effect.forkChild({ startImmediately: true }))
246 yield* cluster.waitUntil(
247 "The request was not persisted while the runner was down",
248 Effect.map(cluster.unprocessedMessageCount, (value) => value === 1)
249 )
250
251 yield* cluster.start(1)
252 yield* cluster.waitForStableAssignments()
253 assert.strictEqual(yield* Fiber.join(replyFiber), `persisted:${backend}-restart`)
254 yield* cluster.waitUntil(
255 "The persisted reply was not recorded",
256 Effect.map(cluster.repliedMessageCount, (value) => value === 1)
257 )
258 assert.strictEqual(count("Persisted", `${backend}-restart`), 1)
259 assert.deepStrictEqual(yield* cluster.messageCounts(), {
260 failed: 0,
261 replied: 1,
262 unprocessed: 0
263 })
264 }))
265
266 it.live(`${backend}: serves primary-key duplicates from the stored reply`, () =>
267 Effect.gen(function*() {
268 resetState()
269 const cluster = yield* make({ backend, entities: PersistenceEntityLayer })
270 yield* cluster.start(2)
271 yield* cluster.waitForStableAssignments()
272 const client = yield* cluster.getClient(PersistenceEntity)
273 const id = `${backend}-stored`
274 const payload = new KeyedPayload({ id })
275
276 assert.strictEqual(yield* client("stored").StoredReply(payload), `stored:${id}`)
277 const firstOwner = yield* cluster.ownerOfEntity(PersistenceEntity, "stored")
278 yield* cluster.kill(firstOwner!)
279 yield* cluster.waitForStableAssignments()
280
281 assert.strictEqual(yield* client("stored").StoredReply(payload), `stored:${id}`)
282 assert.strictEqual(count("StoredReply", id), 1)
283 assert.deepStrictEqual(yield* cluster.messageCounts(), {
284 failed: 0,
285 replied: 1,
286 unprocessed: 0
287 })
288 }))
289
290 it.live(`${backend}: does not lose an uninterruptible request during runner shutdown`, () =>
291 Effect.gen(function*() {
292 resetState()
293 const cluster = yield* make({
294 backend,
295 config: { entityTerminationTimeout: 100 },
296 entities: PersistenceEntityLayer
297 })
298 const [owner] = yield* cluster.start(1)
299 yield* cluster.waitForStableAssignments()
300 const client = yield* cluster.getClient(PersistenceEntity)
301 const id = `${backend}-uninterruptible`
302 const replyFiber = yield* client("uninterruptible").Uninterruptible(
303 new KeyedPayload({ id })
304 ).pipe(Effect.forkChild({ startImmediately: true }))
305 yield* cluster.waitUntil(
306 "The uninterruptible handler did not start",
307 Effect.as(state.uninterruptibleEntered.await, true)
308 )
309
310 const stopping = yield* cluster.stop(owner).pipe(Effect.forkChild({ startImmediately: true }))
311 yield* cluster.start(1)
312 yield* cluster.waitForStableAssignments()
313 yield* cluster.waitUntil(
314 "The uninterruptible request was not resumed by the replacement runner",
315 Effect.sync(() => count("Uninterruptible", id) >= 2)
316 )
317 state.uninterruptibleGate.openUnsafe()
318 yield* cluster.waitUntil(
319 "The resumed uninterruptible request did not complete",
320 Effect.sync(() => state.completedUninterruptible === 1)
321 )
322
323 assert.strictEqual(yield* Fiber.join(replyFiber), `uninterruptible:${id}`)
324 yield* Fiber.join(stopping)
325 assert.strictEqual(yield* cluster.repliedMessageCount, 1)
326 }))
327
328 it.live(`${backend}: does not store or redeliver a volatile request after runner failure`, () =>
329 Effect.gen(function*() {
330 resetState()
331 const cluster = yield* make({
332 backend,
333 config: { entityTerminationTimeout: 100 },
334 entities: PersistenceEntityLayer
335 })
336 const [owner] = yield* cluster.start(1)
337 yield* cluster.waitForStableAssignments()
338 const client = yield* cluster.getClient(PersistenceEntity)
339 const id = `${backend}-volatile`
340 const requestFiber = yield* client("volatile").Volatile(
341 new KeyedPayload({ id }),
342 { discard: true }
343 ).pipe(Effect.forkChild({ startImmediately: true }))
344 yield* cluster.waitUntil(
345 "The volatile handler did not start",
346 Effect.as(state.volatileEntered.await, true)
347 )
348
349 yield* cluster.kill(owner)
350 yield* Fiber.interrupt(requestFiber)
351 state.volatileGate.openUnsafe()
352 yield* cluster.start(1)
353 yield* cluster.waitForStableAssignments()
354 const observationDeadline = (yield* Clock.currentTimeMillis) + 1_000
355 yield* cluster.waitUntil(
356 "The volatile redelivery observation window did not elapse",
357 Effect.map(Clock.currentTimeMillis, (now) => now >= observationDeadline),
358 "2 seconds"
359 )
360
361 assert.strictEqual(count("Volatile", id), 1)
362 assert.strictEqual(state.completedVolatile, 0)
363 assert.deepStrictEqual(yield* cluster.messageCounts(), {
364 failed: 0,
365 replied: 0,
366 unprocessed: 0
367 })
368 }))
369
370 it.live(`${backend}: completes a discarded volatile caller while the handler is still blocked`, () =>
371 Effect.gen(function*() {
372 resetState()
373 const cluster = yield* make({
374 backend,
375 config: { entityTerminationTimeout: 100 },
376 entities: PersistenceEntityLayer
377 })
378 yield* cluster.start(1)
379 yield* cluster.waitForStableAssignments()
380 const client = yield* cluster.getClient(PersistenceEntity)
381 const id = `${backend}-discarded-volatile`
382 const caller = yield* client("discarded-volatile").Volatile(
383 new KeyedPayload({ id }),
384 { discard: true }
385 ).pipe(Effect.forkChild({ startImmediately: true }))
386 yield* cluster.waitUntil(
387 "The discarded volatile handler did not start",
388 Effect.as(state.volatileEntered.await, true)
389 )
390 yield* cluster.waitUntil(
391 "The discarded volatile caller waited for the blocked handler",
392 Effect.sync(() => caller.pollUnsafe() !== undefined)
393 )
394
395 yield* Fiber.join(caller)
396 assert.strictEqual(count("Volatile", id), 1)
397 }))
398
399 it.live(`${backend}: stores an unencodable persisted reply as a defect and does not hang the caller`, () =>
400 Effect.gen(function*() {
401 resetState()
402 const cluster = yield* make({ backend, entities: PersistenceEntityLayer })
403 yield* cluster.start(2)
404 yield* cluster.waitForStableAssignments()
405 const client = yield* cluster.getClient(PersistenceEntity)
406 const id = `${backend}-unencodable`
407 const payload = new KeyedPayload({ id })
408
409 yield* assertStoredAsDefect(client("unencodable").Unencodable(payload), "Unencodable", id)
410 assert.deepStrictEqual(yield* cluster.messageCounts(), {
411 failed: 1,
412 replied: 0,
413 unprocessed: 0
414 })
415 }))
416
417 it.live(`${backend}: isolates a reply-encoding failure so a sibling request on the same connection succeeds`, () =>
418 Effect.gen(function*() {
419 resetState()
420 const cluster = yield* make({ backend, entities: PersistenceEntityLayer })
421 yield* cluster.start(1)
422 yield* cluster.waitForStableAssignments()
423 const client = yield* cluster.getClient(PersistenceEntity)
424 const badEntityId = `${backend}-encoding-bad`
425 const siblingEntityId = `${backend}-encoding-sibling`
426
427 const healthyId = `${backend}-healthy-blocked`
428 const healthy = yield* client(siblingEntityId).Healthy(
429 new KeyedPayload({ id: healthyId })
430 ).pipe(Effect.forkChild({ startImmediately: true }))
431 yield* cluster.waitUntil(
432 "The sibling request did not start",
433 Effect.as(state.healthyEntered.await, true)
434 )
435
436 const badId = `${backend}-encoding-failure`
437 const bad = yield* client(badEntityId).Unencodable(
438 new KeyedPayload({ id: badId })
439 ).pipe(Effect.exit)
440 assert.isTrue(Exit.isFailure(bad))
441 if (Exit.isFailure(bad)) {
442 assert.include(Cause.pretty(bad.cause), "MalformedMessage")
443 }
444
445 state.healthyGate.openUnsafe()
446 assert.strictEqual(yield* Fiber.join(healthy), `healthy:${healthyId}`)
447 assert.strictEqual(count("Unencodable", badId), 1)
448 assert.strictEqual(count("Healthy", healthyId), 1)
449 assert.deepStrictEqual(yield* cluster.messageCounts(), {
450 failed: 1,
451 replied: 1,
452 unprocessed: 0
453 })
454 }))
455
456 it.live(`${backend}: persists a non-JSON Error as a serializable defect`, () =>
457 Effect.gen(function*() {
458 resetState()
459 const cluster = yield* make({ backend, entities: PersistenceEntityLayer })
460 yield* cluster.start(2)
461 yield* cluster.waitForStableAssignments()
462 const client = yield* cluster.getClient(PersistenceEntity)
463 const id = `${backend}-non-json-failure`
464 const payload = new KeyedPayload({ id })
465
466 yield* assertStoredAsDefect(
467 client("non-json-failure").NonJsonFailure(payload),
468 "NonJsonFailure",
469 id
470 )
471 assert.deepStrictEqual(yield* cluster.messageCounts(), {
472 failed: 1,
473 replied: 0,
474 unprocessed: 0
475 })
476 }))
477
478 it.live(`${backend}: persists typed failures and defects without wedging the mailbox`, () =>
479 Effect.gen(function*() {
480 resetState()
481 const cluster = yield* make({ backend, entities: PersistenceEntityLayer })
482 yield* cluster.start(2)
483 yield* cluster.waitForStableAssignments()
484 const client = yield* cluster.getClient(PersistenceEntity)
485 const typedId = `${backend}-typed`
486 const defectId = `${backend}-defect`
487
488 assert.strictEqual(
489 yield* client("failures").TypedFailure(new KeyedPayload({ id: typedId })).pipe(Effect.flip),
490 `typed:${typedId}`
491 )
492 assert.strictEqual(
493 yield* client("failures").TypedFailure(new KeyedPayload({ id: typedId })).pipe(Effect.flip),
494 `typed:${typedId}`
495 )
496 assert.strictEqual(count("TypedFailure", typedId), 1)
497
498 const firstDefect = yield* client("failures").Defect(
499 new KeyedPayload({ id: defectId })
500 ).pipe(Effect.exit)
501 assert.isTrue(Exit.isFailure(firstDefect))
502 if (Exit.isFailure(firstDefect)) {
503 assert.include(Cause.pretty(firstDefect.cause), `defect:${defectId}`)
504 }
505 const storedDefect = yield* client("failures").Defect(
506 new KeyedPayload({ id: defectId })
507 ).pipe(Effect.exit)
508 assert.isTrue(Exit.isFailure(storedDefect))
509 if (Exit.isFailure(storedDefect)) {
510 assert.include(Cause.pretty(storedDefect.cause), `defect:${defectId}`)
511 }
512 assert.strictEqual(count("Defect", defectId), 1)
513
514 assert.strictEqual(
515 yield* client("failures").Healthy(new KeyedPayload({ id: `${backend}-healthy` })),
516 `healthy:${backend}-healthy`
517 )
518 yield* cluster.waitUntil(
519 "The failure replies were not persisted",
520 Effect.map(cluster.messageCounts(), (counts) => counts.failed === 2 && counts.replied === 1)
521 )
522 assert.deepStrictEqual(yield* cluster.messageCounts(), {
523 failed: 2,
524 replied: 1,
525 unprocessed: 0
526 })
527 }))
528
529 it.live(`${backend}: defects a malformed persisted envelope without wedging the mailbox`, () =>
530 Effect.gen(function*() {
531 resetState()
532 const cluster = yield* make({ backend, entities: PersistenceEntityLayer })
533 const malformedId = `${backend}-malformed-stored`
534 const entityId = EntityId.make(malformedId)
535 const shardId = cluster.clientSharding.getShardId(entityId, PersistenceEntity.getShardGroup(entityId))
536 yield* cluster.insertMessage({
537 deliver_at: null,
538 entity_id: malformedId,
539 entity_type: PersistenceEntity.type,
540 headers: JSON.stringify({ invalid: 1 }),
541 id: "1",
542 kind: 0,
543 message_id: null,
544 payload: JSON.stringify({ id: malformedId }),
545 reply_id: null,
546 request_id: "1",
547 sampled: null,
548 shard_id: ShardId.toString(shardId),
549 span_id: null,
550 tag: "Healthy",
551 trace_id: null
552 })
553
554 yield* cluster.start(1)
555 yield* cluster.waitForStableAssignments()
556 yield* cluster.waitUntil(
557 "The malformed persisted envelope was not stored as a defect",
558 Effect.map(cluster.failedMessageCount, (value) => value === 1)
559 )
560 assert.strictEqual(count("Healthy", malformedId), 0)
561
562 const client = yield* cluster.getClient(PersistenceEntity)
563 const healthyId = `${backend}-malformed-stored-healthy`
564 assert.strictEqual(
565 yield* client("malformed-stored-healthy").Healthy(new KeyedPayload({ id: healthyId })),
566 `healthy:${healthyId}`
567 )
568 assert.strictEqual(count("Healthy", healthyId), 1)
569 assert.deepStrictEqual(yield* cluster.messageCounts(), {
570 failed: 1,
571 replied: 1,
572 unprocessed: 0
573 })
574 }))
575
576 it.live(`${backend}: abandons a stream chunk acknowledgement during shutdown without stranding either runner`, () =>
577 Effect.gen(function*() {
578 resetState()
579 const cluster = yield* make({ backend, entities: PersistenceEntityLayer })
580 const runners = yield* cluster.start(2)
581 yield* cluster.waitForStableAssignments()
582 const client = yield* cluster.getClient(PersistenceEntity)
583 const entityId = `${backend}-shutdown-stream`
584 const id = `${entityId}-restart`
585 const owner = yield* cluster.ownerOfEntity(PersistenceEntity, entityId)
586 assert.isDefined(owner)
587 const peer = runners.find((runner) => runner !== owner)
588 assert.isDefined(peer)
589 const values = yield* client(entityId).Streamed(new KeyedPayload({ id })).pipe(
590 Stream.runCollect,
591 Effect.forkChild({ startImmediately: true })
592 )
593 yield* cluster.waitUntil(
594 "The stream handler did not reach its third chunk",
595 Effect.as(state.streamThirdEntered.await, true)
596 )
597
598 const stopping = yield* cluster.stop(owner).pipe(Effect.forkChild({ startImmediately: true }))
599 state.streamThirdGate.openUnsafe()
600 yield* Fiber.join(stopping)
601 yield* cluster.waitForStableAssignments()
602
603 const healthyId = `${backend}-shutdown-stream-healthy`
604 assert.strictEqual(
605 yield* client(entityId).Healthy(new KeyedPayload({ id: healthyId })),
606 `healthy:${healthyId}`
607 )
608 assert.deepStrictEqual(Array.from(yield* Fiber.join(values)), [0, 1, 2, 3, 4])
609 assert.strictEqual(yield* cluster.ownerOfEntity(PersistenceEntity, entityId), peer)
610 }))
611
612 it.live(`${backend}: round-trips a chunked reply through storage`, () =>
613 Effect.gen(function*() {
614 resetState()
615 const cluster = yield* make({ backend, entities: PersistenceEntityLayer })
616 yield* cluster.start(2)
617 yield* cluster.waitForStableAssignments()
618 const client = yield* cluster.getClient(PersistenceEntity)
619 const id = `${backend}-stream`
620 const values = yield* client("stream").Streamed(new KeyedPayload({ id })).pipe(Stream.runCollect)
621 assert.deepStrictEqual(Array.from(values), [0, 1, 2, 3, 4])
622 yield* cluster.waitUntil(
623 "The terminal stream reply was not persisted",
624 Effect.map(cluster.repliedMessageCount, (value) => value === 1)
625 )
626 assert.strictEqual(count("Streamed", id), 1)
627 }))
628
629 it.live(`${backend}: resumes a persisted stream after its runner is killed`, () =>
630 Effect.gen(function*() {
631 resetState()
632 const cluster = yield* make({ backend, entities: PersistenceEntityLayer })
633 const [owner] = yield* cluster.start(1)
634 yield* cluster.waitForStableAssignments()
635 const client = yield* cluster.getClient(PersistenceEntity)
636 const id = `${backend}-stream-restart`
637 const received: Array<number> = []
638 const valuesFiber = yield* client("stream-restart").Streamed(new KeyedPayload({ id })).pipe(
639 Stream.tap((value) => Effect.sync(() => received.push(value))),
640 Stream.runCollect,
641 Effect.forkChild({ startImmediately: true })
642 )
643 yield* cluster.waitUntil(
644 "The stream did not deliver two chunks before blocking the third",
645 Effect.sync(() => received.length === 2 && received[0] === 0 && received[1] === 1)
646 )
647 yield* cluster.waitUntil(
648 "The stream handler did not block before delivering its third chunk",
649 Effect.as(state.streamThirdEntered.await, true)
650 )
651
652 yield* cluster.kill(owner)
653 yield* cluster.start(1)
654 yield* cluster.waitForStableAssignments()
655 yield* cluster.waitUntil(
656 "The replacement runner did not resume the persisted stream",
657 Effect.sync(() => count("Streamed", id) === 2)
658 )
659 state.streamThirdGate.openUnsafe()
660
661 assert.deepStrictEqual(Array.from(yield* Fiber.join(valuesFiber)), [0, 1, 2, 3, 4])
662 yield* cluster.waitUntil(
663 "The terminal stream reply was not persisted after recovery",
664 Effect.map(
665 cluster.messageCounts(),
666 (counts) => counts.replied === 1 && counts.unprocessed === 0
667 )
668 )
669 assert.deepStrictEqual(yield* cluster.messageCounts(), {
670 failed: 0,
671 replied: 1,
672 unprocessed: 0
673 })
674 }))
675
676 it.live(`${backend}: resumes a terminal stream on the replacement owner and persists exactly one reply`, () =>
677 Effect.gen(function*() {
678 resetState()
679 yield* Effect.addFinalizer(() => Effect.sync(() => state.streamTerminalGate.openUnsafe()))
680 const cluster = yield* make({ backend, entities: PersistenceEntityLayer })
681 const runners = yield* cluster.start(2)
682 yield* cluster.waitForStableAssignments()
683 const client = yield* cluster.getClient(PersistenceEntity)
684 const entityId = `${backend}-terminal-race`
685 const id = entityId
686 const owner = yield* cluster.ownerOfEntity(PersistenceEntity, entityId)
687 assert.isDefined(owner)
688 const peer = runners.find((runner) => runner !== owner)
689 assert.isDefined(peer)
690 const received: Array<number> = []
691 const payload = new KeyedPayload({ id })
692 const values = yield* client(entityId).Streamed(payload).pipe(
693 Stream.tap((value) => Effect.sync(() => received.push(value))),
694 Stream.runCollect,
695 Effect.forkChild({ startImmediately: true })
696 )
697 yield* cluster.waitUntil(
698 "The first owner did not reach terminal stream completion",
699 Effect.as(state.streamTerminalEntered.await, true)
700 )
701 yield* cluster.waitUntil(
702 "The client did not receive the final stream value before owner death",
703 Effect.sync(() => received.length === 1)
704 )
705 assert.deepStrictEqual(received, [0])
706
707 yield* cluster.kill(owner)
708 yield* cluster.waitForEntityOwner(PersistenceEntity, entityId, peer)
709 yield* cluster.waitUntil(
710 "The replacement owner did not resume the terminal stream",
711 Effect.sync(() => count("Streamed", id) === 2)
712 )
713 state.streamTerminalGate.openUnsafe()
714
715 assert.deepStrictEqual(Array.from(yield* Fiber.join(values)), [0])
716 yield* cluster.waitUntil(
717 "The surviving terminal stream reply was not persisted",
718 Effect.map(
719 cluster.messageCounts(),
720 (counts) => counts.replied === 1 && counts.unprocessed === 0
721 )
722 )
723 assert.strictEqual(count("Streamed", id), 2)
724 assert.strictEqual(yield* cluster.ownerOfEntity(PersistenceEntity, entityId), peer)
725 }))
726
727 it.live(`${backend}: delivers scheduled messages only after their deadline`, () =>
728 Effect.gen(function*() {
729 resetState()
730 const cluster = yield* make({ backend, entities: PersistenceEntityLayer })
731 yield* cluster.start(2)
732 yield* cluster.waitForStableAssignments()
733 const client = yield* cluster.getClient(PersistenceEntity)
734 const id = `${backend}-scheduled`
735 const deliverAt = (yield* Clock.currentTimeMillis) + 1_500
736 const replyFiber = yield* client("scheduled").Scheduled(
737 new ScheduledPayload({ deliverAt, id })
738 ).pipe(Effect.forkChild({ startImmediately: true }))
739
740 yield* cluster.waitUntil(
741 "The early-delivery observation point was not reached",
742 Effect.map(Clock.currentTimeMillis, (now) => now >= deliverAt - 500),
743 "2 seconds"
744 )
745 assert.strictEqual(count("Scheduled", id), 0)
746 assert.deepStrictEqual(state.scheduledDeliveries, [])
747
748 yield* cluster.waitUntil(
749 "The scheduled message was not delivered after its deadline",
750 Effect.sync(() => state.scheduledDeliveries.length === 1),
751 "5 seconds"
752 )
753 yield* cluster.waitUntil(
754 "The scheduled reply was not persisted",
755 Effect.map(cluster.repliedMessageCount, (value) => value === 1)
756 )
757 const deliveredAt = yield* Fiber.join(replyFiber)
758 assert.isAtLeast(deliveredAt, deliverAt)
759 assert.strictEqual(state.scheduledDeliveries[0], deliveredAt)
760 }))
761 }
762})