1import { assert, it } from "@effect/vitest";
2import * as Deferred from "effect/Deferred";
3import * as Effect from "effect/Effect";
4import * as Fiber from "effect/Fiber";
5import * as Queue from "effect/Queue";
6import * as Ref from "effect/Ref";
7import * as TestClock from "effect/testing/TestClock";
8
9import * as ServerActivation from "../serverActivation.ts";
10import * as Scheduler from "./Scheduler.ts";
11
12it.effect("keeps other sources running after a source defects", () =>
13 Effect.gen(function* () {
14 const scheduler = yield* Scheduler.Scheduler;
15 const receipts = yield* Queue.unbounded<string>();
16 yield* scheduler.register(
17 "broken",
18 Queue.offer(receipts, "broken").pipe(Effect.andThen(Effect.die("fixture failure"))),
19 );
20 yield* scheduler.register("healthy", Queue.offer(receipts, "healthy").pipe(Effect.asVoid));
21 for (let tick = 0; tick < 3; tick += 1) {
22 if (tick > 0) yield* TestClock.adjust("5 seconds");
23 assert.deepEqual([yield* Queue.take(receipts), yield* Queue.take(receipts)].sort(), [
24 "broken",
25 "healthy",
26 ]);
27 }
28 }).pipe(Effect.provide(Scheduler.layer)),
29);
30
31it.effect("does not overlap slow work or hold up another source", () =>
32 Effect.gen(function* () {
33 const scheduler = yield* Scheduler.Scheduler;
34 const started = yield* Deferred.make<void>();
35 const release = yield* Deferred.make<void>();
36 const runs = yield* Ref.make(0);
37 const receipts = yield* Queue.unbounded<void>();
38 yield* scheduler.register(
39 "slow",
40 Ref.update(runs, (n) => n + 1).pipe(
41 Effect.andThen(Deferred.succeed(started, undefined)),
42 Effect.andThen(Deferred.await(release)),
43 ),
44 );
45 yield* scheduler.register("healthy", Queue.offer(receipts, undefined).pipe(Effect.asVoid));
46 yield* Deferred.await(started);
47 yield* Queue.take(receipts);
48 yield* TestClock.adjust("10 seconds");
49 yield* Queue.take(receipts);
50 yield* Queue.take(receipts);
51 assert.equal(yield* Ref.get(runs), 1);
52 yield* Deferred.succeed(release, undefined);
53 yield* TestClock.adjust("5 seconds");
54 yield* Queue.take(receipts);
55 assert.equal(yield* Ref.get(runs), 2);
56 }).pipe(Effect.provide(Scheduler.layer)),
57);
58
59it.effect("unregisters closed sources and interrupts their in-flight work", () =>
60 Effect.gen(function* () {
61 const scheduler = yield* Scheduler.Scheduler;
62 const started = yield* Deferred.make<void>();
63 const stopped = yield* Deferred.make<void>();
64 const runs = yield* Ref.make(0);
65 const registration = yield* Effect.scoped(
66 scheduler
67 .register(
68 "scoped",
69 Ref.update(runs, (n) => n + 1).pipe(
70 Effect.andThen(Deferred.succeed(started, undefined)),
71 Effect.andThen(Effect.never),
72 Effect.ensuring(Deferred.succeed(stopped, undefined)),
73 ),
74 )
75 .pipe(Effect.andThen(Effect.never)),
76 ).pipe(Effect.forkChild);
77 yield* Deferred.await(started);
78 yield* Fiber.interrupt(registration);
79 yield* Deferred.await(stopped);
80 const receipts = yield* Queue.unbounded<void>();
81 yield* scheduler.register("remaining", Queue.offer(receipts, undefined).pipe(Effect.asVoid));
82 yield* Queue.take(receipts);
83 yield* TestClock.adjust("5 seconds");
84 yield* Queue.take(receipts);
85 assert.equal(yield* Ref.get(runs), 1);
86 }).pipe(Effect.provide(Scheduler.layer)),
87);
88
89it.effect("holds due work and the clock until startup recovery activates the server", () =>
90 Effect.gen(function* () {
91 const activation = yield* Deferred.make<void>();
92 const runs = yield* Ref.make(0);
93 const ran = yield* Deferred.make<void>();
94 yield* Effect.gen(function* () {
95 const scheduler = yield* Scheduler.Scheduler;
96 yield* scheduler.register(
97 "due",
98 Ref.update(runs, (n) => n + 1).pipe(Effect.andThen(Deferred.succeed(ran, undefined))),
99 );
100 yield* TestClock.adjust("10 seconds");
101 assert.equal(yield* Ref.get(runs), 0);
102 yield* Deferred.succeed(activation, undefined);
103 yield* Deferred.await(ran);
104 }).pipe(
105 Effect.provide(Scheduler.layer),
106 Effect.provideService(ServerActivation.ServerActivation, Deferred.await(activation)),
107 );
108 assert.isAtLeast(yield* Ref.get(runs), 1);
109 }),
110);