1import { assert, describe, it } from "@effect/vitest"
2import { Clock, Context, Cron, DateTime, Duration, Effect, Latch, Layer } from "effect"
3import { ClusterCron, ClusterSchema, Entity } from "effect/cluster"
4import { type Backend, make } from "./harness.ts"
5
6interface Tick {
7 readonly at: number
8 readonly runner: string
9 readonly scheduled: string
10}
11
12const everySecond = Cron.parseUnsafe("* * * * * *", "UTC")
13const testConfig = { shardsPerGroup: 12 } as const
14
15const addressString = (address: { readonly host: string; readonly port: number }) => `${address.host}:${address.port}`
16
17const recordTick = (ticks: Array<Tick>) =>
18 Effect.contextWith((context: Context.Context<never>) =>
19 Effect.gen(function*() {
20 const address = Context.getUnsafe(context, Entity.CurrentAddress)
21 const runner = Context.getUnsafe(context, Entity.CurrentRunnerAddress)
22 const at = yield* Clock.currentTimeMillis
23 const tick = {
24 at,
25 runner: addressString(runner),
26 scheduled: String(address.entityId)
27 }
28 const isFirst = ticks.length === 0
29 ticks.push(tick)
30 return isFirst
31 })
32 )
33
34const cronProbe = (name: string, shardGroup = "default") =>
35 Entity.make(`ClusterCron/${name}`, []).annotate(ClusterSchema.ShardGroup, () => shardGroup)
36
37const nextScheduled = (cron: Cron.Cron, after: DateTime.DateTime.Input) =>
38 DateTime.formatIso(DateTime.fromDateUnsafe(Cron.next(cron, after)))
39
40const assertScheduledFromExecutionTime = (
41 cron: Cron.Cron,
42 ticks: ReadonlyArray<Tick>,
43 startIndex = 1
44) => {
45 for (let index = startIndex; index < ticks.length; index++) {
46 assert.strictEqual(ticks[index].scheduled, nextScheduled(cron, ticks[index - 1].at))
47 }
48}
49
50const assertScheduledFromPrevious = (cron: Cron.Cron, ticks: ReadonlyArray<Tick>) => {
51 assert.strictEqual(ticks[0].scheduled, "initial")
52 for (let index = 2; index < ticks.length; index++) {
53 assert.strictEqual(ticks[index].scheduled, nextScheduled(cron, ticks[index - 1].scheduled))
54 }
55}
56
57describe("cluster cron integration", () => {
58 for (const backend of ["pg", "mysql"] satisfies ReadonlyArray<Backend>) {
59 it.live(`${backend}: runs once per scheduled instant and continues after failure`, () =>
60 Effect.gen(function*() {
61 const ticks: Array<Tick> = []
62 let failingAttempts = 0
63 let successfulAttempts = 0
64 const cron = ClusterCron.make({
65 name: `basic-${backend}`,
66 cron: everySecond,
67 execute: recordTick(ticks)
68 })
69 const failingCron = ClusterCron.make({
70 name: `failure-${backend}`,
71 cron: everySecond,
72 execute: Effect.suspend(() => {
73 failingAttempts++
74 if (failingAttempts === 1) return Effect.fail("expected cron failure")
75 successfulAttempts++
76 return Effect.void
77 })
78 })
79 const cluster = yield* make({
80 backend,
81 config: testConfig,
82 entities: Layer.merge(cron, failingCron)
83 })
84 yield* cluster.start(3)
85 yield* cluster.waitForStableAssignments()
86 yield* cluster.waitUntil(
87 "The cron jobs did not continue through four scheduled instants",
88 Effect.sync(() => ticks.length >= 4 && failingAttempts >= 4)
89 )
90
91 const firstFour = ticks.slice(0, 4)
92 assert.strictEqual(new Set(firstFour.map((tick) => tick.scheduled)).size, firstFour.length)
93 assertScheduledFromExecutionTime(everySecond, firstFour)
94 assert.strictEqual(successfulAttempts, failingAttempts - 1)
95 }))
96
97 it.live(`${backend}: calculates the next run from the previous instant or the current time`, () =>
98 Effect.gen(function*() {
99 const previousTicks: Array<Tick> = []
100 const currentTicks: Array<Tick> = []
101 const gate = Latch.makeUnsafe()
102 let entered = 0
103 const blockedExecution = Effect.fnUntraced(
104 function*(ticks: Array<Tick>) {
105 if (yield* recordTick(ticks)) {
106 entered++
107 yield* gate.await
108 }
109 },
110 Effect.uninterruptible
111 )
112 const previousCron = ClusterCron.make({
113 name: `previous-${backend}`,
114 cron: everySecond,
115 calculateNextRunFromPrevious: true,
116 execute: blockedExecution(previousTicks)
117 })
118 const currentCron = ClusterCron.make({
119 name: `current-${backend}`,
120 cron: everySecond,
121 execute: blockedExecution(currentTicks)
122 })
123 const cluster = yield* make({
124 backend,
125 config: testConfig,
126 entities: Layer.merge(previousCron, currentCron)
127 })
128 yield* cluster.start(3)
129 yield* cluster.waitForStableAssignments()
130 yield* cluster.waitUntil(
131 "Both cron executions did not enter their first run",
132 Effect.sync(() => entered === 2)
133 )
134 const blockedAt = yield* Clock.currentTimeMillis
135 yield* cluster.waitUntil(
136 "The cron executions were not held across several scheduled instants",
137 Effect.map(Clock.currentTimeMillis, (now) => now >= blockedAt + 3_200),
138 "5 seconds"
139 )
140 gate.openUnsafe()
141 yield* cluster.waitUntil(
142 "The cron jobs did not resume after their first execution",
143 Effect.sync(() => previousTicks.length >= 4 && currentTicks.length >= 4),
144 "12 seconds"
145 )
146
147 const previousSecond = DateTime.toEpochMillis(DateTime.makeUnsafe(previousTicks[1].scheduled))
148 const currentSecond = DateTime.toEpochMillis(DateTime.makeUnsafe(currentTicks[1].scheduled))
149 assertScheduledFromPrevious(everySecond, previousTicks)
150 assertScheduledFromExecutionTime(everySecond, currentTicks, 2)
151 assert.isAtLeast(previousTicks[1].at - previousSecond, 2_000)
152 assert.isAtMost(currentTicks[1].at - currentSecond, 1_000)
153 }))
154
155 it.live(`${backend}: catches up or skips stale runs and preserves the schedule across restart`, () =>
156 Effect.gen(function*() {
157 const catchUpTicks: Array<Tick> = []
158 const skipTicks: Array<Tick> = []
159 const catchUpCron = ClusterCron.make({
160 name: `catch-up-${backend}`,
161 cron: everySecond,
162 calculateNextRunFromPrevious: true,
163 execute: recordTick(catchUpTicks)
164 })
165 const skipCron = ClusterCron.make({
166 name: `skip-stale-${backend}`,
167 cron: everySecond,
168 calculateNextRunFromPrevious: true,
169 skipIfOlderThan: Duration.millis(500),
170 execute: recordTick(skipTicks)
171 })
172 const cluster = yield* make({
173 backend,
174 config: testConfig,
175 entities: Layer.merge(catchUpCron, skipCron)
176 })
177 const runners = yield* cluster.start(3)
178 yield* cluster.waitForStableAssignments()
179 yield* cluster.waitUntil(
180 "The cron schedules did not start",
181 Effect.sync(() => catchUpTicks.length >= 2 && skipTicks.length >= 2)
182 )
183 for (const runner of runners) {
184 yield* cluster.kill(runner)
185 }
186 const catchUpBefore = catchUpTicks.length
187 const skipBefore = skipTicks.length
188 const lastScheduledBeforeRestart = catchUpTicks[catchUpBefore - 1].scheduled
189 const stoppedAt = yield* Clock.currentTimeMillis
190 yield* cluster.waitUntil(
191 "The cluster downtime window did not elapse",
192 Effect.map(Clock.currentTimeMillis, (now) => now >= stoppedAt + 3_200),
193 "5 seconds"
194 )
195 assert.strictEqual(catchUpTicks.length, catchUpBefore)
196 assert.strictEqual(skipTicks.length, skipBefore)
197
198 yield* cluster.start(3)
199 yield* cluster.waitForStableAssignments()
200 yield* cluster.waitUntil(
201 "The restarted cluster did not expose the catch-up and stale-skip difference",
202 Effect.sync(() => {
203 const caughtUp = catchUpTicks.length - catchUpBefore
204 const skipped = skipTicks.length - skipBefore
205 return caughtUp >= 3 && caughtUp >= skipped + 2
206 }),
207 "15 seconds"
208 )
209 yield* cluster.waitUntil(
210 "The stale-skipping cron did not resume",
211 Effect.sync(() => skipTicks.length > skipBefore)
212 )
213
214 const firstCatchUp = catchUpTicks[catchUpBefore]
215 const firstAfterSkip = skipTicks[skipBefore]
216 const firstAfterSkipScheduledAt = DateTime.toEpochMillis(DateTime.makeUnsafe(firstAfterSkip.scheduled))
217 assert.strictEqual(
218 firstCatchUp.scheduled,
219 nextScheduled(everySecond, lastScheduledBeforeRestart)
220 )
221 assert.isAtLeast(firstCatchUp.at - DateTime.toEpochMillis(DateTime.makeUnsafe(firstCatchUp.scheduled)), 2_000)
222 assert.isAtMost(firstAfterSkip.at - firstAfterSkipScheduledAt, 750)
223 assertScheduledFromPrevious(everySecond, catchUpTicks)
224 }))
225
226 it.live(`${backend}: resumes without duplicate or missing ticks after the singleton owner dies`, () =>
227 Effect.gen(function*() {
228 const name = `owner-failover-${backend}`
229 const ticks: Array<Tick> = []
230 const cron = ClusterCron.make({
231 name,
232 cron: everySecond,
233 calculateNextRunFromPrevious: true,
234 execute: recordTick(ticks)
235 })
236 const cluster = yield* make({ backend, config: testConfig, entities: cron })
237 yield* cluster.start(3)
238 yield* cluster.waitForStableAssignments()
239 yield* cluster.waitUntil(
240 "The cron did not execute before failover",
241 Effect.sync(() => ticks.length >= 2)
242 )
243 const owner = yield* cluster.ownerOfEntity(cronProbe(name), name)
244 assert.isDefined(owner)
245 yield* cluster.kill(owner!)
246 const afterKill = ticks.length
247 yield* cluster.waitUntil(
248 "The cron did not resume after its singleton owner died",
249 Effect.sync(() => ticks.length >= afterKill + 3),
250 "12 seconds"
251 )
252 yield* cluster.waitForStableAssignments()
253
254 assertScheduledFromPrevious(everySecond, ticks)
255 assert.strictEqual(new Set(ticks.map((tick) => tick.scheduled)).size, ticks.length)
256 assert.isTrue(ticks.slice(afterKill).every((tick) => tick.runner !== addressString(owner!.address)))
257 }))
258
259 it.live(`${backend}: assigns cron singletons and executions to their shard groups`, () =>
260 Effect.gen(function*() {
261 const defaultName = `default-group-${backend}`
262 const specialName = `special-group-${backend}`
263 const defaultTicks: Array<Tick> = []
264 const specialTicks: Array<Tick> = []
265 const defaultCron = ClusterCron.make({
266 name: defaultName,
267 cron: everySecond,
268 execute: recordTick(defaultTicks)
269 })
270 const specialCron = ClusterCron.make({
271 name: specialName,
272 cron: everySecond,
273 shardGroup: "special",
274 execute: recordTick(specialTicks)
275 })
276 const cluster = yield* make({
277 backend,
278 config: {
279 availableShardGroups: ["default", "special"],
280 shardsPerGroup: 12
281 },
282 entities: Layer.merge(defaultCron, specialCron)
283 })
284 const [defaultRunner] = yield* cluster.start(1, { assignedShardGroups: ["default"] })
285 const [specialRunner] = yield* cluster.start(1, { assignedShardGroups: ["special"] })
286 yield* cluster.waitForStableAssignments()
287 yield* cluster.waitUntil(
288 "The shard-group cron jobs did not execute",
289 Effect.sync(() => defaultTicks.length >= 2 && specialTicks.length >= 2)
290 )
291
292 assert.strictEqual(yield* cluster.ownerOfEntity(cronProbe(defaultName), defaultName), defaultRunner)
293 assert.strictEqual(
294 yield* cluster.ownerOfEntity(cronProbe(specialName, "special"), specialName),
295 specialRunner
296 )
297 assert.isTrue(defaultTicks.every((tick) => tick.runner === addressString(defaultRunner.address)))
298 assert.isTrue(specialTicks.every((tick) => tick.runner === addressString(specialRunner.address)))
299 }))
300 }
301})