1import { NodeCrypto, NodeFileSystem } from "@effect/platform-node"
2import { SqliteClient } from "@effect/sql-sqlite-node"
3import { assert, describe, expect, it } from "@effect/vitest"
4import { Effect, Exit, Fiber, FileSystem, Latch, Layer, Option } from "effect"
5import {
6 Entity,
7 EntityAddress,
8 EntityId,
9 EntityType,
10 Envelope,
11 Message,
12 MessageStorage,
13 RunnerHealth,
14 Runners,
15 RunnerStorage,
16 ShardId,
17 Sharding,
18 ShardingConfig,
19 Snowflake,
20 SqlMessageStorage
21} from "effect/cluster"
22import { SqlClient } from "effect/sql"
23import { TestClock } from "effect/testing"
24import { MysqlContainer } from "../fixtures/mysql2-utils.ts"
25import { PgContainer } from "../fixtures/pg-utils.ts"
26import {
27 GetUserRpc,
28 LongKeyRpc,
29 makeAckChunk,
30 makeChunkReply,
31 makeReply,
32 makeRequest,
33 PrimaryKeyTest,
34 StreamRpc
35} from "./MessageStorageTest.ts"
36
37const TestEntity = Entity.make("test", [GetUserRpc])
38const TestEntityLayer = TestEntity.toLayer({
39 GetUser: () => Effect.void
40})
41
42const StorageLayer = SqlMessageStorage.layer.pipe(
43 Layer.provideMerge(Snowflake.layerGenerator),
44 Layer.provide([ShardingConfig.layerDefaults, NodeCrypto.layer])
45)
46
47const truncate = Effect.gen(function*() {
48 const sql = yield* SqlClient.SqlClient
49 yield* sql`DELETE FROM cluster_replies`
50 yield* sql`DELETE FROM cluster_messages`
51})
52
53describe("SqlMessageStorage", () => {
54 ;([
55 ["pg", Layer.orDie(PgContainer.layerClient)],
56 ["mysql", Layer.orDie(MysqlContainer.layerClient)],
57 ["sqlite", Layer.orDie(SqliteLayer)]
58 ] as const).forEach(([label, layer]) => {
59 // Tests truncate this backend's shared tables.
60 it.layer(StorageLayer.pipe(Layer.provideMerge(layer)), {
61 concurrent: false,
62 timeout: 120000
63 })(label, (it) => {
64 if (label === "pg") {
65 for (const reset of ["resetShards", "resetAddresses"] as const) {
66 it.effect(`${reset} does not deadlock with a concurrent message claim`, () =>
67 Effect.gen(function*() {
68 yield* truncate
69 const sql = yield* SqlClient.SqlClient
70 const storage = yield* SqlMessageStorage.makeEncoded().pipe(Effect.provide(NodeCrypto.layer))
71 // Heap order is the reverse of the claim query's rowid order.
72 yield* sql`
73 INSERT INTO cluster_messages
74 (id, rowid, shard_id, entity_type, entity_id, kind, tag, payload, headers, processed, request_id)
75 VALUES
76 (2, 2, 'default:1', 'test', '1', 0, 'GetUser', '{}', '{}', FALSE, 2),
77 (1, 1, 'default:1', 'test', '1', 0, 'GetUser', '{}', '{}', FALSE, 1)
78 `
79 // Pause after the reset locks row 2, so the claim can lock row 1.
80 yield* sql`
81 CREATE FUNCTION cluster_test_slow_reset() RETURNS trigger AS $$
82 BEGIN
83 IF NEW.last_read IS NULL AND NEW.rowid = 2 THEN
84 PERFORM pg_sleep(2);
85 END IF;
86 RETURN NEW;
87 END $$ LANGUAGE plpgsql
88 `
89 yield* sql`
90 CREATE TRIGGER cluster_test_slow_reset BEFORE UPDATE ON cluster_messages
91 FOR EACH ROW EXECUTE FUNCTION cluster_test_slow_reset()
92 `
93 yield* Effect.gen(function*() {
94 const resetFiber = yield* Effect.gen(function*() {
95 // Make the unordered UPDATE use heap order regardless of indexes.
96 yield* sql`SET LOCAL enable_indexscan = off`
97 yield* sql`SET LOCAL enable_bitmapscan = off`
98 yield* reset === "resetShards"
99 ? storage.resetShards(["default:1"])
100 : storage.resetAddresses([EntityAddress.make({
101 shardId: ShardId.make("default", 1),
102 entityType: EntityType.make("test"),
103 entityId: EntityId.make("1")
104 })])
105 }).pipe(sql.withTransaction, Effect.exit, Effect.forkChild)
106 yield* Effect.gen(function*() {
107 while (true) {
108 const waiting = yield* sql`
109 SELECT 1 FROM pg_stat_activity
110 WHERE pid <> pg_backend_pid() AND wait_event = 'PgSleep'
111 AND query LIKE '%cluster_messages%'
112 `
113 if (waiting.length > 0) return
114 yield* Effect.sleep("10 millis")
115 }
116 }).pipe(Effect.timeout("5 seconds"))
117 const claim = yield* Effect.exit(storage.unprocessedMessages(["default:1"], Date.now()))
118 const resetResult = yield* Fiber.join(resetFiber)
119 expect(Exit.isSuccess(resetResult)).toBe(true)
120 assert(Exit.isSuccess(claim), String(claim))
121 expect(claim.value.map((message) => message.envelope.requestId)).toEqual(["1", "2"])
122 expect(yield* storage.unprocessedMessages(["default:1"], Date.now())).toHaveLength(0)
123 }).pipe(Effect.ensuring(
124 Effect.gen(function*() {
125 yield* sql`DROP TRIGGER cluster_test_slow_reset ON cluster_messages`
126 yield* sql`DROP FUNCTION cluster_test_slow_reset()`
127 yield* truncate
128 }).pipe(Effect.orDie)
129 ))
130 }).pipe(TestClock.withLive))
131 }
132
133 it.effect("creates an index for insertion-ordered message reads", () =>
134 Effect.gen(function*() {
135 const sql = yield* SqlClient.SqlClient
136 const indexes = yield* sql<{ indexname: string }>`
137 SELECT indexname
138 FROM pg_indexes
139 WHERE tablename = 'cluster_messages'
140 AND indexname = 'cluster_messages_rowid_idx'
141 `
142 expect(indexes).toHaveLength(1)
143 }))
144 }
145
146 it.effect("resetRequests with no IDs leaves existing claims untouched", () =>
147 Effect.gen(function*() {
148 yield* truncate
149 const storage = yield* MessageStorage.MessageStorage
150 const request = yield* makeRequest()
151 yield* storage.saveRequest(request)
152 expect(yield* storage.unprocessedMessages([request.envelope.address.shardId])).toHaveLength(1)
153 yield* storage.resetRequests([])
154 expect(yield* storage.unprocessedMessages([request.envelope.address.shardId])).toHaveLength(0)
155 }))
156
157 it.effect("resetRequests releases only the selected request at a shared address", () =>
158 Effect.gen(function*() {
159 yield* truncate
160 const storage = yield* MessageStorage.MessageStorage
161 const selected = yield* makeRequest()
162 const unrelated = yield* makeRequest()
163 yield* storage.saveRequest(selected)
164 yield* storage.saveRequest(unrelated)
165 const shards = [selected.envelope.address.shardId]
166 expect(yield* storage.unprocessedMessages(shards)).toHaveLength(2)
167 yield* storage.resetRequests([selected.envelope.requestId])
168 const messages = yield* storage.unprocessedMessages(shards)
169 expect(messages.map((message) => message.envelope.requestId)).toEqual([selected.envelope.requestId])
170 expect(yield* storage.unprocessedMessages(shards)).toHaveLength(0)
171 }))
172
173 it.effect("resetRequests preserves chunk replies, exit replies, and completed state", () =>
174 Effect.gen(function*() {
175 yield* truncate
176 const storage = yield* MessageStorage.MessageStorage
177 const streaming = yield* makeRequest({ rpc: StreamRpc, payload: StreamRpc.payloadSchema.make({ id: 123 }) })
178 const completed = yield* makeRequest()
179 yield* storage.saveRequest(streaming)
180 yield* storage.saveRequest(completed)
181 const shards = [streaming.envelope.address.shardId]
182 expect(yield* storage.unprocessedMessages(shards)).toHaveLength(2)
183 yield* storage.saveReply(yield* makeChunkReply(streaming))
184 yield* storage.saveReply(yield* makeReply(completed))
185 const replies = yield* storage.repliesFor([streaming, completed])
186 expect(replies).toHaveLength(2)
187 expect(yield* storage.unprocessedMessages(shards)).toHaveLength(0)
188 const sql = yield* SqlClient.SqlClient
189 const processed = yield* sql`SELECT processed FROM cluster_messages ORDER BY rowid`
190 yield* storage.resetRequests([streaming.envelope.requestId, completed.envelope.requestId])
191 const messages = yield* storage.unprocessedMessages(shards)
192 // Replies keep both requests out of the SQL read loop.
193 expect(messages).toHaveLength(0)
194 expect(yield* storage.repliesFor([streaming, completed])).toEqual(replies)
195 expect(yield* sql`SELECT processed FROM cluster_messages ORDER BY rowid`).toEqual(processed)
196 yield* truncate
197 }))
198
199 it.effect("clearReplies requeues when the expected reply is current", () =>
200 Effect.gen(function*() {
201 yield* truncate
202 const storage = yield* MessageStorage.MessageStorage
203 const request = yield* makeRequest()
204 const reply = yield* makeReply(request)
205 yield* storage.saveRequest(request)
206 yield* storage.saveReply(reply)
207 yield* storage.clearReplies(request.envelope.requestId, { expectedReplyId: reply.reply.id })
208 expect(yield* storage.repliesFor([request])).toHaveLength(0)
209 expect(yield* storage.unprocessedMessages([request.envelope.address.shardId])).toHaveLength(1)
210 }))
211
212 it.effect("clearReplies with a stale expected reply preserves a newer completion and processed state", () =>
213 Effect.gen(function*() {
214 yield* truncate
215 const sql = yield* SqlClient.SqlClient
216 const storage = yield* MessageStorage.MessageStorage
217 const request = yield* makeRequest({ rpc: StreamRpc, payload: StreamRpc.payloadSchema.make({ id: 123 }) })
218 const oldReply = yield* makeChunkReply(request)
219 const completed = yield* makeReply(request)
220 yield* storage.saveRequest(request)
221 yield* storage.saveReply(oldReply)
222 yield* storage.saveReply(completed)
223 const before = yield* sql`SELECT processed, last_reply_id FROM cluster_messages WHERE id = ${
224 String(request.envelope.requestId)
225 }`.pipe(Effect.provideService(SqlClient.SafeIntegers, true))
226 yield* storage.clearReplies(request.envelope.requestId, { expectedReplyId: oldReply.reply.id })
227 expect((yield* storage.repliesFor([request])).map((r) => r.id)).toEqual([
228 oldReply.reply.id,
229 completed.reply.id
230 ])
231 expect(
232 yield* sql`SELECT processed, last_reply_id FROM cluster_messages WHERE id = ${
233 String(request.envelope.requestId)
234 }`.pipe(Effect.provideService(SqlClient.SafeIntegers, true))
235 ).toEqual(before)
236 expect(yield* storage.unprocessedMessages([request.envelope.address.shardId])).toHaveLength(0)
237 yield* truncate
238 }))
239
240 it.effect("saveRequest", () =>
241 Effect.gen(function*() {
242 const storage = yield* MessageStorage.MessageStorage
243 const request = yield* makeRequest({ payload: { id: 1 } })
244 const result = yield* storage.saveRequest(request)
245 expect(result._tag).toEqual("Success")
246
247 for (let i = 2; i <= 5; i++) {
248 yield* storage.saveRequest(yield* makeRequest({ payload: { id: i } }))
249 }
250
251 yield* storage.saveReply(yield* makeReply(request))
252
253 let messages = yield* storage.unprocessedMessages([request.envelope.address.shardId])
254 expect(messages).toHaveLength(4)
255 expect(messages.map((m: any) => m.envelope.payload.id)).toEqual([2, 3, 4, 5])
256
257 for (let i = 6; i <= 10; i++) {
258 yield* storage.saveRequest(yield* makeRequest({ payload: { id: i } }))
259 }
260 messages = yield* storage.unprocessedMessages([request.envelope.address.shardId])
261 expect(messages).toHaveLength(5)
262 expect(messages.map((m: any) => m.envelope.payload.id)).toEqual([6, 7, 8, 9, 10])
263 }))
264
265 it.effect("saveReply + saveRequest duplicate", () =>
266 Effect.gen(function*() {
267 const sql = yield* SqlClient.SqlClient
268 const storage = yield* MessageStorage.MessageStorage
269 const request = yield* makeRequest({
270 rpc: StreamRpc,
271 payload: StreamRpc.payloadSchema.make({ id: 123 })
272 })
273 let result = yield* storage.saveRequest(request)
274 expect(result._tag).toEqual("Success")
275
276 let chunk = yield* makeChunkReply(request, 0)
277 yield* storage.saveReply(chunk)
278 const ackChunk = yield* makeAckChunk(request, chunk)
279 yield* storage.saveEnvelope(ackChunk)
280
281 chunk = yield* makeChunkReply(request, 1)
282 yield* storage.saveReply(chunk)
283
284 result = yield* storage.saveRequest(
285 yield* makeRequest({
286 rpc: StreamRpc,
287 payload: StreamRpc.payloadSchema.make({ id: 123 })
288 })
289 )
290 assert(result._tag === "Duplicate" && Option.isSome(result.lastReceivedReply))
291 expect(result.lastReceivedReply.value._tag).toEqual("Chunk")
292
293 // get the un-acked chunk
294 const replies = yield* storage.repliesFor([request])
295 expect(replies).toHaveLength(1)
296
297 yield* storage.saveReply(yield* makeReply(request))
298
299 result = yield* storage.saveRequest(
300 yield* makeRequest({
301 rpc: StreamRpc,
302 payload: StreamRpc.payloadSchema.make({ id: 123 })
303 })
304 )
305 assert(result._tag === "Duplicate" && Option.isSome(result.lastReceivedReply))
306 expect(result.lastReceivedReply.value._tag).toEqual("WithExit")
307
308 // duplicate WithExit
309 const fiber = yield* storage.saveReply(yield* makeReply(request)).pipe(Effect.forkChild)
310 yield* TestClock.adjust(1)
311 while (!fiber.pollUnsafe()) {
312 yield* sql`SELECT 1`
313 yield* TestClock.adjust(1000)
314 }
315 const error = yield* Effect.flip(Fiber.join(fiber))
316 expect(error._tag).toEqual("PersistenceError")
317 }))
318
319 it.effect("detects duplicates", () =>
320 Effect.gen(function*() {
321 yield* truncate
322
323 const storage = yield* MessageStorage.MessageStorage
324 yield* storage.saveRequest(
325 yield* makeRequest({
326 rpc: PrimaryKeyTest,
327 payload: PrimaryKeyTest.payloadSchema.make({ id: 123 })
328 })
329 )
330 const result = yield* storage.saveRequest(
331 yield* makeRequest({
332 rpc: PrimaryKeyTest,
333 payload: PrimaryKeyTest.payloadSchema.make({ id: 123 })
334 })
335 )
336 expect(result._tag).toEqual("Duplicate")
337 }))
338
339 it.effect("hashes primary keys longer than the message_id column", () =>
340 Effect.gen(function*() {
341 yield* truncate
342
343 const storage = yield* MessageStorage.MessageStorage
344 const longId = "long-key-".repeat(50)
345 const request = yield* makeRequest({
346 rpc: LongKeyRpc,
347 payload: LongKeyRpc.payloadSchema.make({ id: longId })
348 })
349 const result = yield* storage.saveRequest(request)
350 expect(result._tag).toEqual("Success")
351
352 const duplicate = yield* storage.saveRequest(
353 yield* makeRequest({
354 rpc: LongKeyRpc,
355 payload: LongKeyRpc.payloadSchema.make({ id: longId })
356 })
357 )
358 expect(duplicate._tag).toEqual("Duplicate")
359
360 const requestId = yield* storage.requestIdForPrimaryKey({
361 address: request.envelope.address,
362 tag: request.envelope.tag,
363 id: longId
364 })
365 expect(requestId).toEqual(Option.some(request.envelope.requestId))
366
367 const sql = yield* SqlClient.SqlClient
368 const rows = yield* sql<{ message_id: string }>`SELECT message_id FROM cluster_messages`
369 expect(rows).toHaveLength(1)
370 expect(rows[0].message_id).toMatch(/^[0-9a-f]{64}$/)
371 }))
372
373 it.effect("keeps primary keys within the column width as plaintext", () =>
374 Effect.gen(function*() {
375 yield* truncate
376
377 const sql = yield* SqlClient.SqlClient
378 const storage = yield* MessageStorage.MessageStorage
379 const request = yield* makeRequest({
380 rpc: PrimaryKeyTest,
381 payload: PrimaryKeyTest.payloadSchema.make({ id: 456 })
382 })
383 yield* storage.saveRequest(request)
384
385 // rows written by previous versions store the plaintext composed
386 // key, so short keys must stay byte-identical to keep deduplicating
387 const plaintext = Envelope.primaryKey(request.envelope)!
388 const rows = yield* sql<{ message_id: string }>`SELECT message_id FROM cluster_messages`
389 expect(rows).toHaveLength(1)
390 expect(rows[0].message_id).toEqual(plaintext)
391
392 const result = yield* storage.saveRequest(
393 yield* makeRequest({
394 rpc: PrimaryKeyTest,
395 payload: PrimaryKeyTest.payloadSchema.make({ id: 456 })
396 })
397 )
398 assert(result._tag === "Duplicate")
399 expect(result.originalId).toEqual(request.envelope.requestId)
400
401 const requestId = yield* storage.requestIdForPrimaryKey({
402 address: request.envelope.address,
403 tag: request.envelope.tag,
404 id: "456"
405 })
406 expect(requestId).toEqual(Option.some(request.envelope.requestId))
407 }))
408
409 if (label === "sqlite") {
410 // sqlite's TEXT message_id column stored over-long plaintext keys
411 // before hashing, so the legacy fallback must also cover keys longer
412 // than the 255-character limit of the width-enforcing dialects
413 it.effect("detects duplicates for legacy long-key plaintext rows", () =>
414 Effect.gen(function*() {
415 yield* truncate
416
417 const sql = yield* SqlClient.SqlClient
418 const storage = yield* MessageStorage.MessageStorage
419 const longId = "legacy-long-key-".repeat(30)
420 const request = yield* makeRequest({
421 rpc: LongKeyRpc,
422 payload: LongKeyRpc.payloadSchema.make({ id: longId })
423 })
424 yield* storage.saveRequest(request)
425
426 // simulate a row written before message_id values were hashed
427 const plaintext = Envelope.primaryKey(request.envelope)!
428 expect(plaintext.length).toBeGreaterThan(255)
429 yield* sql`UPDATE cluster_messages SET message_id = ${plaintext} WHERE id = ${
430 String(request.envelope.requestId)
431 }`
432
433 const result = yield* storage.saveRequest(
434 yield* makeRequest({
435 rpc: LongKeyRpc,
436 payload: LongKeyRpc.payloadSchema.make({ id: longId })
437 })
438 )
439 assert(result._tag === "Duplicate")
440 expect(result.originalId).toEqual(request.envelope.requestId)
441
442 const requestId = yield* storage.requestIdForPrimaryKey({
443 address: request.envelope.address,
444 tag: request.envelope.tag,
445 id: longId
446 })
447 expect(requestId).toEqual(Option.some(request.envelope.requestId))
448 }))
449 }
450
451 it.effect("unprocessedMessages", () =>
452 Effect.gen(function*() {
453 yield* truncate
454
455 const storage = yield* MessageStorage.MessageStorage
456 const request = yield* makeRequest()
457 yield* storage.saveRequest(request)
458 let messages = yield* storage.unprocessedMessages([request.envelope.address.shardId])
459 expect(messages).toHaveLength(1)
460 messages = yield* storage.unprocessedMessages([request.envelope.address.shardId])
461 expect(messages).toHaveLength(0)
462 yield* storage.saveRequest(yield* makeRequest())
463 messages = yield* storage.unprocessedMessages([request.envelope.address.shardId])
464 expect(messages).toHaveLength(1)
465 }))
466
467 it.effect("unprocessedMessages honors the limit and claims only returned rows", () =>
468 Effect.gen(function*() {
469 yield* truncate
470
471 const storage = yield* MessageStorage.MessageStorage
472 const shardId = ShardId.make("default", 1)
473 for (let i = 1; i <= 5; i++) {
474 yield* storage.saveRequest(yield* makeRequest({ payload: { id: i }, entityId: String(i) }))
475 }
476 const limited = yield* storage.unprocessedMessages([shardId], { limit: 3 })
477 expect(limited).toHaveLength(3)
478 expect(limited.map((m: any) => m.envelope.payload.id)).toEqual([1, 2, 3])
479
480 // rows beyond the limit were not claimed and are returned by the
481 // next read
482 const rest = yield* storage.unprocessedMessages([shardId])
483 expect(rest.map((m: any) => m.envelope.payload.id)).toEqual([4, 5])
484 }))
485
486 it.effect("unprocessedMessages filters by address", () =>
487 Effect.gen(function*() {
488 yield* truncate
489
490 const storage = yield* MessageStorage.MessageStorage
491 const shardId = ShardId.make("default", 1)
492 const address = (entityId: string) =>
493 EntityAddress.make({
494 shardId,
495 entityType: EntityType.make("test"),
496 entityId: EntityId.make(entityId)
497 })
498 for (let i = 1; i <= 4; i++) {
499 yield* storage.saveRequest(yield* makeRequest({ payload: { id: i }, entityId: String(i) }))
500 }
501 const filtered = yield* storage.unprocessedMessages([shardId], {
502 addresses: [address("2"), address("4")]
503 })
504 expect(filtered.map((m: any) => m.envelope.payload.id)).toEqual([2, 4])
505
506 // the filtered read must not claim the other addresses
507 const rest = yield* storage.unprocessedMessages([shardId])
508 expect(rest.map((m: any) => m.envelope.payload.id)).toEqual([1, 3])
509 }))
510
511 it.effect("unprocessedMessages filters addresses by shard", () =>
512 Effect.gen(function*() {
513 yield* truncate
514
515 const storage = yield* MessageStorage.MessageStorage
516 const shardOne = ShardId.make("default", 1)
517 const shardTwo = ShardId.make("default", 2)
518 yield* storage.saveRequest(yield* makeRequest({ payload: { id: 1 }, shardId: shardOne }))
519 yield* storage.saveRequest(yield* makeRequest({ payload: { id: 2 }, shardId: shardTwo }))
520
521 const messages = yield* storage.unprocessedMessages([shardOne, shardTwo], {
522 addresses: [EntityAddress.make({
523 shardId: shardOne,
524 entityType: EntityType.make("test"),
525 entityId: EntityId.make("1")
526 })]
527 })
528 expect(messages.map((message: any) => message.envelope.payload.id)).toEqual([1])
529
530 const rest = yield* storage.unprocessedMessages([shardOne, shardTwo])
531 expect(rest.map((message: any) => message.envelope.payload.id)).toEqual([2])
532 }))
533
534 it.effect("encoded unprocessedMessages fails closed for empty addresses", () =>
535 Effect.gen(function*() {
536 yield* truncate
537
538 const storage = yield* MessageStorage.MessageStorage
539 const request = yield* makeRequest()
540 yield* storage.saveRequest(request)
541 const encoded = yield* SqlMessageStorage.makeEncoded().pipe(Effect.provide(NodeCrypto.layer))
542 const shardId = request.envelope.address.shardId.toString()
543 const messages = yield* encoded.unprocessedMessages([shardId], Date.now(), { addresses: [] })
544 expect(messages).toHaveLength(0)
545
546 const unfiltered = yield* encoded.unprocessedMessages([shardId], Date.now())
547 expect(unfiltered).toHaveLength(1)
548 }))
549
550 it.effect("encoded resetAddresses fails closed for an empty address list", () =>
551 Effect.gen(function*() {
552 yield* truncate
553
554 const storage = yield* MessageStorage.MessageStorage
555 const request = yield* makeRequest()
556 yield* storage.saveRequest(request)
557 const encoded = yield* SqlMessageStorage.makeEncoded().pipe(Effect.provide(NodeCrypto.layer))
558 const shardId = request.envelope.address.shardId.toString()
559 const claimed = yield* encoded.unprocessedMessages([shardId], Date.now())
560 expect(claimed).toHaveLength(1)
561
562 yield* encoded.resetAddresses([])
563 const messages = yield* encoded.unprocessedMessages([shardId], Date.now())
564 expect(messages).toHaveLength(0)
565 }))
566
567 it.effect("resetAddresses releases claims in one batch", () =>
568 Effect.gen(function*() {
569 yield* truncate
570
571 const storage = yield* MessageStorage.MessageStorage
572 const shardId = ShardId.make("default", 1)
573 const address = (entityId: string) =>
574 EntityAddress.make({
575 shardId,
576 entityType: EntityType.make("test"),
577 entityId: EntityId.make(entityId)
578 })
579 for (let i = 1; i <= 4; i++) {
580 yield* storage.saveRequest(yield* makeRequest({ payload: { id: i }, entityId: String(i) }))
581 }
582 yield* storage.unprocessedMessages([shardId], { limit: 3 })
583 yield* storage.resetAddresses([address("1"), address("3")])
584
585 const messages = yield* storage.unprocessedMessages([shardId])
586 expect(messages.map((m: any) => m.envelope.payload.id)).toEqual([1, 3, 4])
587 }))
588
589 it.effect("unprocessedMessages excludes complete requests", () =>
590 Effect.gen(function*() {
591 yield* truncate
592
593 const storage = yield* MessageStorage.MessageStorage
594 const request = yield* makeRequest()
595 yield* storage.saveRequest(request)
596 yield* storage.saveReply(yield* makeReply(request))
597 const messages = yield* storage.unprocessedMessages([request.envelope.address.shardId])
598 expect(messages).toHaveLength(0)
599 }))
600
601 it.effect("repliesFor", () =>
602 Effect.gen(function*() {
603 yield* truncate
604
605 const storage = yield* MessageStorage.MessageStorage
606 const request = yield* makeRequest()
607 yield* storage.saveRequest(request)
608 let replies = yield* storage.repliesFor([request])
609 expect(replies).toHaveLength(0)
610 yield* storage.saveReply(yield* makeReply(request))
611 replies = yield* storage.repliesFor([request])
612 expect(replies).toHaveLength(1)
613 expect(replies[0].requestId).toEqual(request.envelope.requestId)
614 }))
615
616 it.effect("registerReplyHandler", () =>
617 Effect.gen(function*() {
618 const storage = yield* MessageStorage.MessageStorage
619 const latch = yield* Latch.make()
620 const request = yield* makeRequest()
621 yield* storage.saveRequest(request)
622 const fiber = yield* storage.registerReplyHandler(
623 new Message.OutgoingRequest({
624 ...request,
625 respond: () => latch.open
626 })
627 ).pipe(Effect.forkChild)
628 yield* TestClock.adjust(1)
629 yield* storage.saveReply(yield* makeReply(request))
630 yield* latch.await
631 yield* Fiber.await(fiber)
632 }))
633
634 it.effect("unprocessedMessagesById", () =>
635 Effect.gen(function*() {
636 yield* truncate
637
638 const storage = yield* MessageStorage.MessageStorage
639 const request = yield* makeRequest({
640 rpc: StreamRpc,
641 payload: StreamRpc.payloadSchema.make({ id: 123 })
642 })
643 yield* storage.saveRequest(request)
644 let messages = yield* storage.unprocessedMessagesById([request.envelope.requestId])
645 expect(messages).toHaveLength(1)
646
647 const chunk = yield* makeChunkReply(request)
648 yield* storage.saveReply(chunk)
649 const ack = yield* makeAckChunk(request, chunk)
650 yield* storage.saveEnvelope(ack)
651
652 const encoded = yield* SqlMessageStorage.makeEncoded().pipe(Effect.provide(NodeCrypto.layer))
653 const acknowledgements = yield* encoded.unprocessedMessagesById([ack.envelope.id], 0)
654 expect(acknowledgements).toHaveLength(1)
655 assert(acknowledgements[0].envelope._tag === "AckChunk")
656 expect(acknowledgements[0].envelope.replyId).toEqual(String(chunk.reply.id))
657
658 yield* storage.saveReply(yield* makeReply(request))
659 messages = yield* storage.unprocessedMessagesById([request.envelope.requestId])
660 expect(messages).toHaveLength(0)
661 }))
662 })
663 })
664
665 it.effect("keeps held messages readable while entity layers build", () =>
666 Effect.gen(function*() {
667 const config = ShardingConfig.layer({
668 entityRegistrationTimeout: 6000,
669 entityTerminationTimeout: 0,
670 entityMessagePollInterval: 100,
671 refreshAssignmentsInterval: 0
672 })
673 const delayedEnv = TestEntityLayer.pipe(
674 Layer.provide(Layer.effectDiscard(Effect.sleep(10_000))),
675 Layer.provideMerge(Sharding.layer),
676 Layer.provide(RunnerStorage.layerMemory),
677 Layer.provide(RunnerHealth.layerNoop),
678 Layer.provide(Runners.layerNoop),
679 Layer.provide(config)
680 )
681 const storage = yield* MessageStorage.MessageStorage
682 const request = yield* makeRequest()
683 yield* storage.saveRequest(request)
684
685 const fiber = yield* Effect.never.pipe(
686 Effect.provide(delayedEnv),
687 Effect.scoped,
688 Effect.forkChild({ startImmediately: true })
689 )
690
691 yield* TestClock.adjust(7500)
692 expect(yield* storage.repliesFor([request])).toHaveLength(0)
693
694 yield* TestClock.adjust(2500)
695 yield* TestClock.adjust(100)
696 expect(yield* storage.repliesFor([request])).toHaveLength(1)
697 yield* Fiber.interrupt(fiber)
698 }).pipe(Effect.provide(StorageLayer.pipe(
699 Layer.provideMerge(SqliteLayer)
700 ))), { timeout: 15_000 })
701})
702
703const SqliteLayer = Effect.gen(function*() {
704 const fs = yield* FileSystem.FileSystem
705 const dir = yield* fs.makeTempDirectoryScoped()
706 return SqliteClient.layer({
707 filename: dir + "/test.db"
708 })
709}).pipe(Layer.unwrap, Layer.provide(NodeFileSystem.layer))