1import { NodeRedis } from "@effect/platform-node"
2import { assert, it } from "@effect/vitest"
3import { RedisContainer } from "@testcontainers/redis"
4import { Clock, Duration, Effect, Layer, Queue, Schema } from "effect"
5import * as PersistedCacheTest from "effect-test/persistence/PersistedCacheTest"
6import * as PersistedQueueTest from "effect-test/persistence/PersistedQueueTest"
7import * as RateLimiterTest from "effect-test/persistence/RateLimiterTest"
8import { PersistedQueue, Persistence, RateLimiter, Redis } from "effect/persistence"
9import { TestClock } from "effect/testing"
10import { createServer } from "node:net"
11
12const RedisLayer = Layer.unwrap(
13 Effect.gen(function*() {
14 const container = yield* Effect.acquireRelease(
15 Effect.promise(() => new RedisContainer("redis:alpine").start()),
16 (container) => Effect.promise(() => container.stop())
17 )
18 return NodeRedis.layer({
19 socket: {
20 host: container.getHost(),
21 port: container.getMappedPort(6379)
22 }
23 })
24 }).pipe(
25 Effect.catchCause(() => Effect.fail(new PersistedCacheTest.TransientError()))
26 )
27)
28
29PersistedCacheTest.suite(
30 "NodeRedis",
31 Persistence.layerRedis.pipe(Layer.provide(RedisLayer))
32)
33
34PersistedQueueTest.suite(
35 "NodeRedis",
36 // short intervals so the periodic reset runs while the suite's takes are
37 // in flight
38 PersistedQueue.layerStoreRedis({
39 pollInterval: "50 millis",
40 lockRefreshInterval: "100 millis"
41 }).pipe(Layer.provide(RedisLayer))
42)
43
44RateLimiterTest.suite(
45 "NodeRedis",
46 RateLimiter.layerStoreRedis().pipe(Layer.provide(RedisLayer))
47)
48
49it.layer(RateLimiter.layerStoreRedis().pipe(Layer.provideMerge(RedisLayer)), {
50 timeout: "30 seconds",
51 concurrent: false
52})(
53 "RateLimiter token-bucket storage (NodeRedis)",
54 (it) => {
55 it.effect("returns the persisted balance after accumulating fractional costs", () =>
56 Effect.gen(function*() {
57 const redis = yield* Redis.Redis
58 const key = "fractional-persisted-balance"
59 const results = yield* RateLimiterTest.consumeFractionalCosts(key)
60 const stored = Number(yield* redis.send<string>("GET", `ratelimiter:${key}`))
61
62 assert.strictEqual(stored, 3.9999999999999996)
63 assert.strictEqual(results[results.length - 1].remaining, stored)
64 }))
65
66 it.effect(
67 "does not restore capacity early after a fractional token cost",
68 () =>
69 Effect.gen(function*() {
70 const redis = yield* Redis.Redis
71 const store = yield* RateLimiter.RateLimiterStore
72 const memory = yield* RateLimiter.RateLimiterStore.pipe(Effect.provide(RateLimiter.layerStoreMemory))
73 const opts = {
74 key: "fractional-cost-expiry",
75 limit: 5,
76 refillRate: Duration.seconds(4),
77 allowOverflow: false
78 }
79 const key = `ratelimiter:${opts.key}`
80 const refillKey = `${key}:refill`
81 yield* memory.tokenBucket({ ...opts, tokens: 0.5 })
82 yield* store.tokenBucket({ ...opts, tokens: 0.5 })
83 const refillAt = Number(yield* redis.send<string>("GET", refillKey))
84 assert.isAbove(yield* redis.send<number>("PTTL", key), 0)
85
86 // Use wall time: the reported bug expires these keys after 2s, before the 4s refill.
87 yield* Effect.sleep("2500 millis")
88 const beforeRefill = {
89 keys: yield* redis.send<number>("EXISTS", key, refillKey),
90 tokens: yield* redis.send<string | null>("GET", key),
91 memoryRemaining: (yield* memory.tokenBucket({ ...opts, tokens: 0 }))[0],
92 redisRemaining: (yield* store.tokenBucket({ ...opts, tokens: 0 }))[0]
93 }
94 const elapsed = (yield* Clock.currentTimeMillis) - refillAt
95 assert.isAtLeast(elapsed, 2_500)
96 assert.isBelow(elapsed, 4_000)
97
98 yield* Effect.sleep(4_100 - elapsed)
99 assert.strictEqual((yield* memory.tokenBucket({ ...opts, tokens: 0 }))[0], 5)
100 assert.strictEqual((yield* store.tokenBucket({ ...opts, tokens: 0 }))[0], 5)
101 assert.deepStrictEqual(beforeRefill, { keys: 2, tokens: "4.5", memoryRemaining: 4.5, redisRemaining: 4.5 })
102 }).pipe(TestClock.withLive),
103 10_000
104 )
105
106 for (const onExceeded of ["fail", "delay"] as const) {
107 it.effect(`${onExceeded}: restarts the refill interval after both Redis keys expire`, () => {
108 const key = `timing-expired-${onExceeded}`
109 const redisKey = `ratelimiter:${key}`
110 const refillKey = `${redisKey}:refill`
111 // TestClock does not advance Redis's expiry clock, so expire the keys by hand.
112 const idle = Effect.gen(function*() {
113 const redis = yield* Redis.Redis
114 yield* TestClock.adjust("359 seconds")
115 assert.strictEqual(yield* redis.send<number>("EXISTS", redisKey, refillKey), 2)
116 assert.strictEqual(yield* redis.send<number>("PEXPIRE", redisKey, "0"), 1)
117 assert.strictEqual(yield* redis.send<number>("PEXPIRE", refillKey, "0"), 1)
118 assert.strictEqual(yield* redis.send<number>("EXISTS", redisKey, refillKey), 0)
119 })
120 return RateLimiterTest.restartsInterval(key, onExceeded, 5, idle)
121 })
122 }
123 }
124)
125
126const PersistedQueueRedisLayer = Layer.mergeAll(
127 RedisLayer,
128 PersistedQueue.layer.pipe(
129 Layer.provideMerge(
130 PersistedQueue.layerStoreRedis().pipe(Layer.provide(RedisLayer))
131 )
132 )
133)
134
135it.effect("fails the initial connection by default", () =>
136 Effect.gen(function*() {
137 const port = yield* closedPort
138 const error = yield* Layer.build(NodeRedis.layer({
139 socket: {
140 host: "127.0.0.1",
141 port
142 }
143 })).pipe(Effect.flip)
144
145 assert.instanceOf(error, Redis.RedisError)
146 }))
147
148it.layer(PersistedQueueRedisLayer, { timeout: "30 seconds" })(
149 "PersistedQueue (NodeRedis)",
150 (it) => {
151 it.effect("uses the node-redis protocol default", () =>
152 Effect.gen(function*() {
153 const redis = yield* NodeRedis.NodeRedis
154 assert.strictEqual(redis.client.options.RESP, undefined)
155 }))
156
157 it.effect("receives published messages", () =>
158 Effect.gen(function*() {
159 const redis = yield* Redis.Redis
160 const subscription = yield* redis.subscribe("effect-test")
161
162 yield* redis.send("PUBLISH", "effect-test", "hello")
163
164 assert.deepStrictEqual(yield* Queue.take(subscription), {
165 channel: "effect-test",
166 message: "hello"
167 })
168 }))
169
170 // The shared PersistedQueue suite can only assert that exhausted elements
171 // are no longer delivered, which is also true if they are silently
172 // dropped. There is no public API for reading failed elements, so
173 // verifying they are preserved in the dead-letter list requires
174 // inspecting Redis directly.
175 it.effect("moves exhausted elements to the failed list", () =>
176 Effect.gen(function*() {
177 const redis = yield* NodeRedis.NodeRedis
178 const queueName = "test-redis-failed"
179
180 const queue = yield* PersistedQueue.make({
181 name: queueName,
182 schema: RedisItem,
183 maxAttempts: 1
184 })
185 const id = yield* queue.offer({ n: 42 })
186 const error = yield* queue.take(() => Effect.fail("boom")).pipe(Effect.flip)
187 assert.strictEqual(error, "boom")
188
189 const failed = yield* redis.use((client) => client.lRange(`effectq:${queueName}:failed`, 0, -1))
190 assert.strictEqual(failed.length, 1)
191 const failedItem = JSON.parse(failed[0])
192 assert.strictEqual(failedItem.id, id)
193 assert.deepStrictEqual(failedItem.element, { n: 42 })
194 assert.strictEqual(failedItem.attempts, 1)
195
196 const pending = yield* redis.use((client) => client.hLen(`effectq:${queueName}:pending`))
197 assert.strictEqual(pending, 0)
198 }))
199
200 it.effect("recovers elements from crashed workers", () =>
201 Effect.gen(function*() {
202 const prefix = "effectq-crash:"
203 const store = yield* PersistedQueue.makeStoreRedis({
204 prefix,
205 pollInterval: "50 millis",
206 lockRefreshInterval: "100 millis",
207 lockExpiration: "1 second"
208 })
209 const factory = yield* PersistedQueue.makeFactory.pipe(
210 Effect.provideService(PersistedQueue.PersistedQueueStore, store)
211 )
212 const queue = yield* factory.make({ name: "crash-recovery", schema: RedisItem })
213 const redis = yield* Redis.Redis
214
215 // simulate a worker that claimed the element and then crashed: the
216 // element sits in the pending hash with a consumed attempt and no lock
217 yield* redis.send(
218 "HSET",
219 `${prefix}crash-recovery:pending`,
220 "crashed",
221 JSON.stringify({ id: "crashed", element: { n: 1 } })
222 )
223 yield* redis.send("HSET", `${prefix}crash-recovery:attempts`, "crashed", "1")
224
225 const result = yield* queue.take((value, metadata) => Effect.succeed([value.n, metadata.attempts]))
226 assert.deepStrictEqual(result, [1, 2])
227 }).pipe(TestClock.withLive), { timeout: 20000 })
228
229 it.effect("dead-letters elements from workers that crashed on the final attempt", () =>
230 Effect.gen(function*() {
231 const prefix = "effectq-crash-exhausted:"
232 const store = yield* PersistedQueue.makeStoreRedis({
233 prefix,
234 pollInterval: "50 millis",
235 lockRefreshInterval: "100 millis",
236 lockExpiration: "1 second"
237 })
238 const factory = yield* PersistedQueue.makeFactory.pipe(
239 Effect.provideService(PersistedQueue.PersistedQueueStore, store)
240 )
241 const queue = yield* factory.make({ name: "crash-exhausted", schema: RedisItem, maxAttempts: 1 })
242 const redis = yield* Redis.Redis
243
244 // the final attempt was claimed by a worker that crashed, so no
245 // finalizer will ever settle this element
246 yield* redis.send(
247 "HSET",
248 `${prefix}crash-exhausted:pending`,
249 "crashed",
250 JSON.stringify({ id: "crashed", element: { n: 1 } })
251 )
252 yield* redis.send("HSET", `${prefix}crash-exhausted:attempts`, "crashed", "1")
253
254 // an active taker runs the periodic reset that dead-letters such
255 // elements instead of redelivering them
256 const fiber = yield* queue.take(Effect.succeed).pipe(Effect.forkScoped)
257 yield* Effect.sleep(1000)
258
259 const failed = yield* redis.send<Array<string>>("LRANGE", `${prefix}crash-exhausted:failed`, "0", "-1")
260 assert.strictEqual(failed.length, 1)
261 const failedItem = JSON.parse(failed[0])
262 assert.strictEqual(failedItem.id, "crashed")
263 assert.deepStrictEqual(failedItem.element, { n: 1 })
264 assert.strictEqual(failedItem.attempts, 1)
265 assert.include(failedItem.lastFailure, "Lock expired after final attempt")
266
267 const pending = yield* redis.send<number>("HLEN", `${prefix}crash-exhausted:pending`)
268 assert.strictEqual(Number(pending), 0)
269 assert.isUndefined(fiber.pollUnsafe())
270 }).pipe(TestClock.withLive), { timeout: 20000 })
271 }
272)
273
274const RedisItem = Schema.Struct({
275 n: Schema.Number
276})
277
278const closedPort = Effect.promise(
279 () =>
280 new Promise<number>((resolve, reject) => {
281 const server = createServer()
282 server.once("error", reject)
283 server.listen(0, "127.0.0.1", () => {
284 const address = server.address()
285 if (address === null || typeof address === "string") {
286 server.close()
287 reject(new Error("Could not allocate a TCP port"))
288 return
289 }
290 server.close((error) => {
291 if (error) {
292 reject(error)
293 } else {
294 resolve(address.port)
295 }
296 })
297 })
298 })
299)