1import { assert, describe, expect, it } from "@effect/vitest"
2import { Array, Context, Data, Deferred, Equal, Fiber, Hash, Schema } from "effect"
3import * as Cause from "effect/Cause"
4import * as Effect from "effect/Effect"
5import * as Exit from "effect/Exit"
6import { flow, pipe } from "effect/Function"
7import { Persistable, Persistence } from "effect/persistence"
8import * as Request from "effect/Request"
9import * as Resolver from "effect/RequestResolver"
10
11class Counter extends Context.Service<Counter, { count: number }>()("Counter") {}
12class Requests extends Context.Service<Requests, { count: number }>()("Requests") {}
13class Interrupts extends Context.Reference("Interrupts", {
14 defaultValue: () => ({ interrupts: 0 })
15}) {}
16class RequestService extends Context.Reference("RequestService", {
17 defaultValue: () => ({ value: "default" })
18}) {}
19const delay = <A, E, R>(self: Effect.Effect<A, E, R>) =>
20 Effect.andThen(
21 Effect.promise(() => new Promise((r) => setTimeout(() => r(0), 0))),
22 self
23 )
24
25const userIds: ReadonlyArray<number> = Array.range(1, 26)
26
27const userNames: ReadonlyMap<number, string> = new Map(
28 Array.zipWith(
29 userIds,
30 Array.map(Array.range(97, 122), (a) => String.fromCharCode(a)),
31 (a, b) => [a, b] as const
32 )
33)
34
35type UserRequest = GetAllIds | GetNameById
36
37interface GetAllIds extends Request.Request<ReadonlyArray<number>> {
38 readonly _tag: "GetAllIds"
39}
40const GetAllIds = Request.tagged<GetAllIds>("GetAllIds")
41
42interface GetRequestService extends Request.Request<string> {
43 readonly _tag: "GetRequestService"
44}
45const GetRequestService = Request.tagged<GetRequestService>("GetRequestService")
46
47class GetNameById extends Request.TaggedClass("GetNameById")<
48 {
49 readonly id: number
50 },
51 string,
52 string
53> {}
54
55const makeUserResolver = Effect.gen(function*() {
56 const counter = yield* Counter
57 const requests_ = yield* Requests
58
59 const resolver = Resolver.make<UserRequest>(Effect.fnUntraced(function*(entries) {
60 counter.count++
61 requests_.count += entries.length
62 for (const entry of entries) {
63 yield* delay(processRequest(entry))
64 }
65 })).pipe(Resolver.batchN(15))
66
67 const getIds = Effect.request(GetAllIds(), resolver)
68 const getNameById = (id: number) => Effect.request(new GetNameById({ id }), resolver)
69 const getNameByIdPiped = (id: number) => pipe(new GetNameById({ id }), Effect.request(resolver))
70 const getNames = getIds.pipe(
71 Effect.flatMap(Effect.forEach(getNameById, { concurrency: "unbounded" })),
72 Effect.onInterrupt(() =>
73 Effect.tap(Interrupts, (i) => {
74 i.interrupts++
75 return Effect.void
76 })
77 )
78 )
79
80 return { getNames, getIds, getNameById, getNameByIdPiped } as const
81})
82
83const makeUserResolverTagged = Effect.gen(function*() {
84 const counter = yield* Counter
85 const requests = yield* Requests
86
87 const resolver = Resolver.fromEffectTagged<UserRequest>()({
88 GetAllIds: Effect.fnUntraced(function*(reqs) {
89 counter.count++
90 requests.count += reqs.length
91 return reqs.map(() => userIds)
92 }),
93 GetNameById: Effect.fnUntraced(function*(reqs) {
94 counter.count++
95 requests.count += reqs.length
96
97 const names: Array<string> = []
98 for (let i = 0; i < reqs.length; i++) {
99 const req = reqs[i]
100 if (!userNames.has(req.request.id)) return yield* Effect.fail("Not Found")
101 names.push(userNames.get(req.request.id)!)
102 }
103 return names
104 })
105 }).pipe(Resolver.batchN(15))
106
107 const getIds = Effect.request(GetAllIds(), resolver)
108 const getNameById = (id: number) => Effect.request(new GetNameById({ id }), resolver)
109 const allNames = getIds.pipe(
110 Effect.flatMap(Effect.forEach(getNameById, { concurrency: "unbounded" }))
111 )
112
113 return { allNames, getIds, getNameById } as const
114})
115
116const processRequest = (entry: Request.Entry<UserRequest>): Effect.Effect<void> => {
117 switch (entry.request._tag) {
118 case "GetAllIds": {
119 return Request.complete(entry, Exit.succeed(userIds))
120 }
121 case "GetNameById": {
122 if (userNames.has(entry.request.id)) {
123 const userName = userNames.get(entry.request.id)!
124 return Request.complete(entry, Exit.succeed(userName))
125 }
126 return Request.completeEffect(entry, Exit.fail("Not Found"))
127 }
128 }
129}
130
131const provideEnv = flow(
132 Effect.provideServiceEffect(Counter, Effect.sync(() => ({ count: 0 }))),
133 Effect.provideServiceEffect(Requests, Effect.sync(() => ({ count: 0 })))
134)
135
136describe("Request", { concurrent: false }, () => {
137 it("preserves __proto__ as an own constructor property", () => {
138 interface ProtoRequest extends Request.Request<void> {
139 readonly "__proto__": { readonly polluted: boolean }
140 }
141 const make = Request.of<ProtoRequest>()
142 const value = { polluted: true }
143 const request = make({ ["__proto__"]: value })
144
145 assert.isTrue(Request.isRequest(request))
146 assert.isTrue(Object.hasOwn(request, "__proto__"))
147 assert.strictEqual(request["__proto__"], value)
148 })
149
150 it("copies enumerable symbol properties in Class", () => {
151 const key = Symbol()
152 class SymbolRequest extends Request.Class<{ readonly [key]: string }, void> {}
153
154 assert.strictEqual(new SymbolRequest({ [key]: "value" })[key], "value")
155 })
156
157 it("compares StructuralProto values when hashes collide", () => {
158 class Req extends Request.Class<{ id: string; account: string }, string> {}
159
160 const a = new Req({ id: "id-8", account: "acct-1975" })
161 const b = new Req({ id: "id-18", account: "acct-433" })
162
163 assert.strictEqual(Hash.hash(a), Hash.hash(b))
164 assert.strictEqual(Equal.equals(a, b), false)
165 })
166
167 it.effect(
168 "requests are executed correctly",
169 Effect.fnUntraced(function*() {
170 const { getNames } = yield* makeUserResolver
171 const names = yield* getNames
172 const counter = yield* Counter
173 const requests = yield* Requests
174 assert.strictEqual(counter.count, 3)
175 assert.strictEqual(requests.count, userIds.length + 1)
176 assert.deepStrictEqual(names, userIds.map((id) => userNames.get(id)))
177 }, provideEnv)
178 )
179
180 it.effect(
181 "requests with dual syntax are executed correctly",
182 Effect.fnUntraced(function*() {
183 const names = yield* (yield* makeUserResolver).getNames
184 const counter = yield* Counter
185 const requests = yield* Requests
186 assert.strictEqual(counter.count, 3)
187 assert.strictEqual(requests.count, userIds.length + 1)
188 assert.deepStrictEqual(names, userIds.map((id) => userNames.get(id)))
189 }, provideEnv)
190 )
191
192 it.effect(
193 "requests are executed correctly with fromEffectTagged",
194 Effect.fnUntraced(function*() {
195 const { allNames } = yield* makeUserResolverTagged
196 const names = yield* allNames
197 const count = yield* Counter
198 expect(count.count).toEqual(3)
199 expect(names.length).toBeGreaterThan(2)
200 expect(names).toEqual(userIds.map((id) => userNames.get(id)))
201 }, provideEnv)
202 )
203
204 it.effect("fromEffectTagged preserves typed errors", () =>
205 Effect.gen(function*() {
206 const resolver = Resolver.fromEffectTagged<GetNameById>()({
207 GetNameById: () => Effect.fail("missing")
208 })
209 const error = yield* Effect.flip(Effect.request(new GetNameById({ id: 1 }), resolver))
210
211 assert.strictEqual(error, "missing")
212 }))
213
214 it.effect("fromEffectTagged preserves defects", () =>
215 Effect.gen(function*() {
216 const defect = new Error("boom")
217 const resolver = Resolver.fromEffectTagged<GetNameById>()({
218 GetNameById: () => Effect.die(defect)
219 })
220 const exit = yield* Effect.exit(Effect.request(new GetNameById({ id: 1 }), resolver))
221
222 assert.isTrue(Exit.isFailure(exit))
223 if (Exit.isFailure(exit)) {
224 assert.strictEqual(exit.cause.reasons.length, 1)
225 const reason = exit.cause.reasons[0]
226 assert.isTrue(Cause.isDieReason(reason))
227 if (Cause.isDieReason(reason)) {
228 assert.strictEqual(reason.defect, defect)
229 }
230 }
231 }))
232
233 it.effect("fromEffectTagged preserves interrupts", () =>
234 Effect.gen(function*() {
235 const resolver = Resolver.fromEffectTagged<GetNameById>()({
236 GetNameById: () => Effect.interrupt
237 })
238 const exit = yield* Effect.exit(Effect.request(new GetNameById({ id: 1 }), resolver))
239
240 assert.isTrue(Exit.hasInterrupts(exit))
241 }))
242
243 it.effect.each([
244 { name: "array", make: (values: Array<string>) => values },
245 { name: "array iterator", make: (values: Array<string>) => values.values() },
246 {
247 name: "generator",
248 make: (values: Array<string>) =>
249 (function*() {
250 yield* values
251 })()
252 }
253 ])("fromEffectTagged resolves requests from $name results", ({ make }) =>
254 Effect.gen(function*() {
255 const resolver = Resolver.fromEffectTagged<GetNameById>()({
256 GetNameById: (requests) => Effect.succeed(make(requests.map(({ request }) => String(request.id))))
257 })
258 const names = yield* Effect.forEach(
259 [1, 2, 3],
260 (id) => Effect.request(new GetNameById({ id }), resolver),
261 { concurrency: "unbounded" }
262 )
263
264 assert.deepStrictEqual(names, ["1", "2", "3"])
265 }))
266
267 it.effect(
268 "requests don't break interruption",
269 Effect.fnUntraced(
270 function*() {
271 const { getNames } = yield* makeUserResolver
272 const fiber = yield* Effect.forkChild(getNames)
273 yield* Effect.yieldNow
274 yield* Fiber.interrupt(fiber)
275 const exit = yield* Fiber.await(fiber)
276 expect(exit._tag).toEqual("Failure")
277 if (exit._tag === "Failure") {
278 expect(Cause.hasInterruptsOnly(exit.cause)).toEqual(true)
279 }
280 expect(yield* Counter).toEqual({ count: 0 })
281 expect(yield* Interrupts).toEqual({ interrupts: 1 })
282 },
283 provideEnv,
284 Effect.provideService(Interrupts, { interrupts: 0 })
285 )
286 )
287
288 it.effect(
289 "requests work with uninterruptible",
290 Effect.fnUntraced(
291 function*() {
292 const { getNames } = yield* makeUserResolver
293 const fiber = yield* Effect.forkChild(Effect.uninterruptible(getNames))
294 yield* Effect.yieldNow
295 yield* Fiber.interrupt(fiber)
296 const exit = yield* Fiber.await(fiber)
297 expect(exit._tag).toEqual("Failure")
298 if (exit._tag === "Failure") {
299 expect(Cause.hasInterruptsOnly(exit.cause)).toEqual(true)
300 }
301 expect(yield* Counter).toEqual({ count: 3 })
302 expect(yield* Interrupts).toEqual({ interrupts: 0 })
303 },
304 provideEnv,
305 Effect.provideService(Interrupts, { interrupts: 0 })
306 )
307 )
308
309 it.effect(
310 "grouped requests can be interrupted before execution",
311 Effect.fnUntraced(function*() {
312 let resolverExecuted = false
313
314 const resolver = Resolver.make<GetNameById>(Effect.fnUntraced(function*(entries) {
315 resolverExecuted = true
316 for (const entry of entries) {
317 entry.completeUnsafe(Exit.succeed(userNames.get(entry.request.id)!))
318 }
319 })).pipe(
320 Resolver.grouped(({ request }) => request.id),
321 Resolver.setDelayEffect(Effect.never)
322 )
323
324 const fiber = yield* Effect.forkChild(Effect.request(new GetNameById({ id: userIds[0] }), resolver))
325 yield* Effect.yieldNow
326 yield* Fiber.interrupt(fiber)
327 const exit = yield* Fiber.await(fiber)
328
329 assert.strictEqual(exit._tag, "Failure")
330 if (exit._tag === "Failure") {
331 assert.strictEqual(Cause.hasInterruptsOnly(exit.cause), true)
332 }
333 assert.strictEqual(resolverExecuted, false)
334 })
335 )
336
337 for (const cached of [false, true]) {
338 it.effect(`repro: cancelling a pending batch does not poison ${cached ? "cached" : "uncached"} requests`, () =>
339 Effect.gen(function*() {
340 const gate = yield* Deferred.make<void>()
341 const base = Resolver.make<GetNameById>((entries) =>
342 Effect.sync(() => {
343 for (const entry of entries) entry.completeUnsafe(Exit.succeed("Alice"))
344 })
345 ).pipe(Resolver.setDelayEffect(Deferred.await(gate)))
346 const resolver = cached ? yield* Resolver.withCache(base, { capacity: 10 }) : base
347 const request = new GetNameById({ id: 1 })
348
349 const first = yield* Effect.request(request, resolver).pipe(Effect.forkChild({ startImmediately: true }))
350 yield* Fiber.interrupt(first)
351 const next = yield* Effect.request(request, resolver).pipe(Effect.forkChild({ startImmediately: true }))
352
353 // Allow execution only after cancelling the first batch and retrying.
354 yield* Deferred.succeed(gate, undefined)
355 yield* Effect.yieldNow
356 const exit = next.pollUnsafe()
357 yield* Fiber.interrupt(next)
358
359 assert.deepStrictEqual(exit, Exit.succeed("Alice"))
360 }))
361 }
362
363 it.effect("withCache removes cancelled entries from a surviving batch", () =>
364 Effect.gen(function*() {
365 const gate = yield* Deferred.make<void>()
366 const batches: Array<Array<number>> = []
367 const resolver = yield* Resolver.make<GetNameById>((entries) =>
368 Effect.sync(() => {
369 batches.push(entries.map((entry) => entry.request.id))
370 for (const entry of entries) entry.completeUnsafe(Exit.succeed(String(entry.request.id)))
371 })
372 ).pipe(
373 Resolver.setDelayEffect(Effect.andThen(Effect.yieldNow, Deferred.await(gate))),
374 Resolver.withCache({ capacity: 10 })
375 )
376 const first = yield* Effect.request(new GetNameById({ id: 1 }), resolver).pipe(
377 Effect.forkChild({ startImmediately: true })
378 )
379 const other = yield* Effect.request(new GetNameById({ id: 2 }), resolver).pipe(
380 Effect.forkChild({ startImmediately: true })
381 )
382 yield* Fiber.interrupt(first)
383 yield* Deferred.succeed(gate, undefined)
384 yield* Effect.yieldNow
385 assert.deepStrictEqual(other.pollUnsafe(), Exit.succeed("2"))
386 assert.deepStrictEqual(batches, [[2]])
387
388 const retry = yield* Effect.request(new GetNameById({ id: 1 }), resolver).pipe(
389 Effect.forkChild({ startImmediately: true })
390 )
391 yield* Effect.yieldNow
392 assert.deepStrictEqual(retry.pollUnsafe(), Exit.succeed("1"))
393 assert.deepStrictEqual(batches, [[2], [1]])
394 }))
395
396 for (const cancelFirst of [false, true]) {
397 it.effect(`withCache preserves coalesced requests when the ${cancelFirst ? "first" : "second"} caller cancels`, () =>
398 Effect.gen(function*() {
399 const gate = yield* Deferred.make<void>()
400 const batches: Array<number> = []
401 const resolver = yield* Resolver.make<GetNameById>((entries) =>
402 Effect.sync(() => {
403 batches.push(entries.length)
404 for (const entry of entries) entry.completeUnsafe(Exit.succeed("Alice"))
405 })
406 ).pipe(Resolver.setDelayEffect(Deferred.await(gate)), Resolver.withCache({ capacity: 10 }))
407 const first = yield* Effect.request(new GetNameById({ id: 1 }), resolver).pipe(
408 Effect.forkChild({ startImmediately: true })
409 )
410 const second = yield* Effect.request(new GetNameById({ id: 1 }), resolver).pipe(
411 Effect.forkChild({ startImmediately: true })
412 )
413 const cancelled = cancelFirst ? first : second
414 const remaining = cancelFirst ? second : first
415 yield* Fiber.interrupt(cancelled)
416 assert.isTrue(Exit.hasInterrupts(yield* Fiber.await(cancelled)))
417 assert.deepStrictEqual(batches, [])
418 yield* Deferred.succeed(gate, undefined)
419 yield* Effect.yieldNow
420
421 assert.deepStrictEqual(remaining.pollUnsafe(), Exit.succeed("Alice"))
422 assert.strictEqual(yield* Effect.request(new GetNameById({ id: 1 }), resolver), "Alice")
423 assert.deepStrictEqual(batches, [1])
424 }))
425 }
426
427 it.effect("withCache still caches completed failures", () =>
428 Effect.gen(function*() {
429 let calls = 0
430 const resolver = yield* Resolver.make<GetNameById>((entries) =>
431 Effect.sync(() => {
432 calls++
433 for (const entry of entries) entry.completeUnsafe(Exit.fail("Not Found"))
434 })
435 ).pipe(Resolver.withCache({ capacity: 10 }))
436
437 for (let i = 0; i < 2; i++) {
438 assert.deepStrictEqual(
439 yield* Effect.exit(Effect.request(new GetNameById({ id: 1 }), resolver)),
440 Exit.fail("Not Found")
441 )
442 }
443 assert.strictEqual(calls, 1)
444 }))
445
446 it.effect("withCache does not cache interrupted resolver results", () =>
447 Effect.gen(function*() {
448 let calls = 0
449 const resolver = yield* Resolver.make<GetNameById>((entries) =>
450 Effect.sync(() => {
451 calls++
452 for (const entry of entries) {
453 entry.completeUnsafe(calls === 1 ? Exit.interrupt() : Exit.succeed("Alice"))
454 }
455 })
456 ).pipe(Resolver.withCache({ capacity: 10 }))
457
458 assert.isTrue(Exit.hasInterrupts(yield* Effect.exit(Effect.request(new GetNameById({ id: 1 }), resolver))))
459 assert.strictEqual(yield* Effect.request(new GetNameById({ id: 1 }), resolver), "Alice")
460 assert.strictEqual(calls, 2)
461 }))
462
463 it.effect("withCache cancellation does not evict a replacement entry", () =>
464 Effect.gen(function*() {
465 const gate = yield* Deferred.make<void>()
466 const batches: Array<Array<number>> = []
467 let delays = 0
468 const resolver = yield* Resolver.make<GetNameById>((entries) =>
469 Effect.sync(() => {
470 batches.push(entries.map((entry) => entry.request.id))
471 for (const entry of entries) entry.completeUnsafe(Exit.succeed("Alice"))
472 })
473 ).pipe(
474 Resolver.grouped(({ request }) => request.id),
475 Resolver.setDelayEffect(Effect.suspend(() => ++delays === 1 ? Deferred.await(gate) : Effect.yieldNow)),
476 Resolver.withCache({ capacity: 1 })
477 )
478 const first = yield* Effect.request(new GetNameById({ id: 1 }), resolver).pipe(
479 Effect.forkChild({ startImmediately: true })
480 )
481 yield* Effect.yieldNow
482 // Completing another group evicts the first entry while its batch is still pending.
483 yield* Effect.request(new GetNameById({ id: 2 }), resolver)
484 yield* Effect.yieldNow
485 const replacement = yield* Effect.request(new GetNameById({ id: 1 }), resolver).pipe(
486 Effect.forkChild({ startImmediately: true })
487 )
488 yield* Fiber.interrupt(first)
489 const coalesced = yield* Effect.request(new GetNameById({ id: 1 }), resolver).pipe(
490 Effect.forkChild({ startImmediately: true })
491 )
492 yield* Deferred.succeed(gate, undefined)
493 yield* Effect.yieldNow
494
495 assert.deepStrictEqual(replacement.pollUnsafe(), Exit.succeed("Alice"))
496 assert.deepStrictEqual(coalesced.pollUnsafe(), Exit.succeed("Alice"))
497 assert.deepStrictEqual(batches, [[2], [1]])
498 }))
499
500 it.effect("withCache allows an equal retry from the cancelled delay's finalizer", () =>
501 Effect.gen(function*() {
502 const gate = yield* Deferred.make<void>()
503 const context = yield* Effect.context<never>()
504 const results: Array<Request.Result<GetNameById>> = []
505 const batches: Array<number> = []
506 let cancelledCallbacks = 0
507 let finalizers = 0
508 const resolver = yield* Resolver.make<GetNameById>((entries) =>
509 Effect.sync(() => {
510 batches.push(entries.length)
511 for (const entry of entries) entry.completeUnsafe(Exit.succeed("Alice"))
512 })
513 ).pipe(
514 Resolver.setDelayEffect(
515 Deferred.await(gate).pipe(
516 Effect.onInterrupt(() =>
517 Effect.sync(() => {
518 finalizers++
519 Effect.requestUnsafe(new GetNameById({ id: 1 }), {
520 resolver,
521 context,
522 onExit: (exit) => results.push(exit)
523 })
524 })
525 )
526 )
527 ),
528 Resolver.withCache({ capacity: 10 })
529 )
530 const cancel = Effect.requestUnsafe(new GetNameById({ id: 1 }), {
531 resolver,
532 context,
533 onExit: () => cancelledCallbacks++
534 })
535 cancel()
536 assert.strictEqual(finalizers, 1)
537 assert.deepStrictEqual(results, [])
538 cancel()
539 yield* Deferred.succeed(gate, undefined)
540 yield* Effect.yieldNow
541
542 assert.strictEqual(cancelledCallbacks, 0)
543 assert.strictEqual(finalizers, 1)
544 assert.deepStrictEqual(results, [Exit.succeed("Alice")])
545 assert.deepStrictEqual(batches, [1])
546 }))
547
548 it.effect("withCache preserves the race winner when the losing resolver is interrupted", () =>
549 Effect.gen(function*() {
550 const finished = yield* Deferred.make<void>()
551 let calls = 0
552 let interruptions = 0
553 const fast = Resolver.fromEffect<GetNameById>(() =>
554 Effect.andThen(
555 Effect.yieldNow,
556 Effect.sync(() => {
557 calls++
558 return "Alice"
559 })
560 )
561 )
562 const slow = Resolver.fromEffect<GetNameById>(() =>
563 Effect.never.pipe(Effect.onInterrupt(() =>
564 Effect.sync(() => {
565 interruptions++
566 })
567 ))
568 )
569 const resolver = yield* Resolver.race(fast, slow).pipe(
570 Resolver.around(() => Effect.void, () => Deferred.succeed(finished, undefined)),
571 Resolver.withCache({ capacity: 10 })
572 )
573
574 assert.strictEqual(yield* Effect.request(new GetNameById({ id: 1 }), resolver), "Alice")
575 // Wait for the losing resolver's interruption before reading the cache again.
576 yield* Deferred.await(finished)
577 assert.strictEqual(interruptions, 1)
578 assert.deepStrictEqual(
579 yield* Effect.exit(Effect.request(new GetNameById({ id: 1 }), resolver)),
580 Exit.succeed("Alice")
581 )
582 assert.strictEqual(calls, 1)
583 }))
584
585 it.effect("withCache accepts refreshed results from stale-while-revalidate", () =>
586 Effect.gen(function*() {
587 class GetName extends Persistable.Class()("GetName", {
588 primaryKey: () => "name",
589 success: Schema.String
590 }) {}
591 const store = yield* (yield* Persistence.Persistence).make({ storeId: "names" })
592 yield* store.set(new GetName(), Exit.succeed("Alice"))
593 const finished = yield* Deferred.make<void>()
594 let calls = 0
595 const persisted = yield* Resolver.fromFunction<GetName>(() => {
596 calls++
597 return "Bob"
598 }).pipe(Resolver.persisted({ storeId: "names", staleWhileRevalidate: () => true }))
599 const resolver = yield* persisted.pipe(
600 Resolver.around(() => Effect.void, () => Deferred.succeed(finished, undefined)),
601 Resolver.withCache({ capacity: 10 })
602 )
603
604 assert.strictEqual(yield* Effect.request(new GetName(), resolver), "Alice")
605 yield* Deferred.await(finished)
606 assert.strictEqual(yield* Effect.request(new GetName(), resolver), "Bob")
607 assert.strictEqual(calls, 1)
608 }).pipe(Effect.provide(Persistence.layerMemory)))
609
610 it.effect("persisted propagates failures to unfinished requests", () =>
611 Effect.gen(function*() {
612 class GetValue extends Persistable.Class<{ payload: { id: number } }>()("GetValue", {
613 primaryKey: ({ id }) => String(id),
614 success: Schema.String,
615 error: Persistence.PersistenceError
616 }) {}
617 const request = new GetValue({ id: 1 })
618 const error = new Persistence.PersistenceError({ message: "backend failed" })
619 const resolver = Resolver.make<GetValue>((entries) => {
620 entries[0].completeUnsafe(Exit.succeed("first"))
621 return Effect.fail(error)
622 })
623 const persisted = yield* Resolver.persisted(resolver, { storeId: "resolver-failure" })
624
625 assert.deepStrictEqual(
626 yield* Effect.forEach(
627 [request, request, new GetValue({ id: 2 })],
628 (request) => Effect.exit(Effect.request(request, persisted)),
629 { concurrency: "unbounded" }
630 ),
631 [Exit.succeed("first"), Exit.fail(error), Exit.fail(error)]
632 )
633 }).pipe(Effect.provide(Persistence.layerMemory)))
634
635 it.effect("persisted stores synchronous resolver failures", () =>
636 Effect.gen(function*() {
637 class GetValue extends Persistable.Class()("ThrowingValue", {
638 primaryKey: () => "value",
639 success: Schema.String
640 }) {}
641 let calls = 0
642 const resolver = Resolver.make<GetValue>(() => {
643 calls++
644 throw "backend defect"
645 })
646 const persisted = yield* Resolver.persisted(resolver, { storeId: "resolver-throw" })
647 const request = new GetValue()
648
649 assert.deepStrictEqual(yield* Effect.exit(Effect.request(request, persisted)), Exit.die("backend defect"))
650 assert.deepStrictEqual(yield* Effect.exit(Effect.request(request, persisted)), Exit.die("backend defect"))
651 assert.strictEqual(calls, 1)
652 }).pipe(Effect.provide(Persistence.layerMemory)))
653
654 it.effect(
655 "batching preserves individual & identical requests",
656 Effect.fnUntraced(function*() {
657 const { getNameById } = yield* makeUserResolver
658 yield* Effect.all([getNameById(userIds[0]), getNameById(userIds[0])], {
659 concurrency: "unbounded",
660 discard: true
661 })
662 const requests = yield* Requests
663 const invocations = yield* Counter
664 expect(requests.count).toEqual(2)
665 expect(invocations.count).toEqual(1)
666 }, provideEnv)
667 )
668
669 it.effect(
670 "grouped requests + batchN",
671 Effect.fnUntraced(function*() {
672 let count = 0
673 let requestsCount = 0
674
675 class Key extends Data.Class<{ id: number }> {}
676
677 const resolver = Resolver.make<GetNameById>(Effect.fnUntraced(function*(entries) {
678 count++
679 requestsCount += entries.length
680 for (const entry of entries) {
681 entry.completeUnsafe(Exit.succeed(userNames.get(entry.request.id)!))
682 }
683 })).pipe(
684 Resolver.batchN(5),
685 Resolver.grouped(({ request }) => new Key({ id: request.id % 2 }))
686 )
687
688 yield* Effect.forEach(userIds, (id) => Effect.request(new GetNameById({ id }), resolver), {
689 concurrency: "unbounded"
690 })
691
692 expect(count).toEqual(6)
693 expect(requestsCount).toEqual(26)
694 })
695 )
696
697 it.effect("batchN normalizes the maximum batch size", () =>
698 Effect.gen(function*() {
699 const sizes = yield* Effect.forEach([Number.NaN, -1, 0, 0.5, 1.9, 2.9], (n) =>
700 Effect.gen(function*() {
701 const batchSizes: Array<number> = []
702 const resolver = Resolver.make<GetNameById>((entries) =>
703 Effect.sync(() => {
704 batchSizes.push(entries.length)
705 for (const entry of entries) {
706 entry.completeUnsafe(Exit.succeed(String(entry.request.id)))
707 }
708 })
709 ).pipe(Resolver.batchN(n))
710
711 yield* Effect.forEach([1, 2, 3], (id) => Effect.request(new GetNameById({ id }), resolver), {
712 concurrency: "unbounded"
713 })
714 return batchSizes
715 }))
716
717 assert.deepStrictEqual(sizes, [
718 [1, 1, 1],
719 [1, 1, 1],
720 [1, 1, 1],
721 [1, 1, 1],
722 [1, 1, 1],
723 [2, 1]
724 ])
725 }))
726
727 it.effect(
728 "batch fibers use request services for runAll",
729 Effect.fnUntraced(function*() {
730 const resolver = Resolver.make<GetRequestService>(Effect.fnUntraced(function*(entries) {
731 const value = (yield* RequestService).value
732 for (const entry of entries) {
733 entry.completeUnsafe(Exit.succeed(value))
734 }
735 })).pipe(Resolver.batchN(1))
736
737 const value = yield* Effect.request(GetRequestService(), resolver).pipe(
738 Effect.provideService(RequestService, { value: "provided" })
739 )
740
741 assert.strictEqual(value, "provided")
742 })
743 )
744
745 it.effect(
746 "batch fibers use request services for delay effects",
747 Effect.fnUntraced(function*() {
748 let delayServiceValue = ""
749
750 const resolver = Resolver.make<GetRequestService>((entries) =>
751 Effect.sync(() => {
752 for (const entry of entries) {
753 entry.completeUnsafe(Exit.succeed("ok"))
754 }
755 })
756 ).pipe(
757 Resolver.setDelayEffect(
758 Effect.andThen(
759 Effect.yieldNow,
760 Effect.gen(function*() {
761 delayServiceValue = (yield* RequestService).value
762 })
763 )
764 )
765 )
766
767 yield* Effect.request(GetRequestService(), resolver).pipe(
768 Effect.provideService(RequestService, { value: "provided" })
769 )
770
771 assert.strictEqual(delayServiceValue, "provided")
772 })
773 )
774})