1import { NodeFileSystem } from "@effect/platform-node"
2import { SqliteClient } from "@effect/sql-sqlite-node"
3import { assert, describe, expect, it } from "@effect/vitest"
4import { Cause, Duration, Effect, Exit, Fiber, FileSystem, Latch, Layer, Schedule } from "effect"
5import {
6 ClusterError,
7 Runner,
8 RunnerAddress,
9 RunnerStorage,
10 ShardId,
11 ShardingConfig,
12 SqlRunnerStorage
13} from "effect/cluster"
14import { SqlClient, type SqlConnection, SqlError } from "effect/sql"
15import { TestClock } from "effect/testing"
16import { MysqlContainer } from "../fixtures/mysql2-utils.ts"
17import { PgContainer } from "../fixtures/pg-utils.ts"
18
19const StorageLayer = SqlRunnerStorage.layer
20
21// Allow for CI latency in healthy PostgreSQL queries.
22const lockOperationInterval = 1000
23const partitionConfig = {
24 // Keep expiration / 3 above the operation deadline.
25 shardLockExpiration: lockOperationInterval * 10,
26 shardLockRefreshInterval: lockOperationInterval
27}
28
29describe("SqlRunnerStorage", () => {
30 it.effect("bounds shard lock operations and rebuilds an unresponsive reserved connection", () => {
31 const partitioned = makePartitionState()
32 const layer = StorageLayer.pipe(
33 Layer.provideMerge(blackholeReservedConnection(partitioned, true)),
34 Layer.provide(ShardingConfig.layer(partitionConfig))
35 )
36
37 return Effect.gen(function*() {
38 const storage = yield* RunnerStorage.RunnerStorage
39 const runner = Runner.make({
40 address: runnerAddress1,
41 groups: ["default"],
42 weight: 1
43 })
44 const shards = [ShardId.make("default", 1)]
45
46 yield* storage.register(runner, true)
47 yield* storage.acquire(runnerAddress1, shards)
48 partitionConnection(partitioned)
49
50 const expectDeadline = Effect.fnUntraced(function*(operation: Effect.Effect<unknown, unknown>) {
51 const [elapsed, exit] = yield* operation.pipe(
52 Effect.exit,
53 Effect.timed
54 )
55 assert(Exit.isFailure(exit))
56 const error = Cause.squash(exit.cause)
57 assert(error instanceof ClusterError.PersistenceError)
58 assert.isBelow(Duration.toMillis(elapsed), lockOperationInterval * 5)
59 yield* Effect.sleep(20)
60 })
61
62 const expectRecovery = Effect.fnUntraced(function*() {
63 restoreConnection(partitioned)
64 expect(
65 yield* storage.refresh(runnerAddress1, shards).pipe(
66 Effect.retry({ times: 5, schedule: Schedule.spaced(20) })
67 )
68 ).toEqual(shards)
69 })
70
71 yield* expectDeadline(storage.refresh(runnerAddress1, shards))
72 yield* expectDeadline(storage.refresh(runnerAddress1, shards))
73 yield* expectDeadline(storage.refresh(runnerAddress1, shards))
74
75 assert.isAtLeast(partitioned.interruptedQueries, 1)
76 assert.isAtMost(partitioned.maxActiveQueries, 1)
77
78 yield* expectRecovery()
79
80 partitionConnection(partitioned)
81 yield* expectDeadline(storage.acquire(runnerAddress1, [ShardId.make("default", 2)]))
82 yield* expectRecovery()
83
84 partitionConnection(partitioned)
85 yield* expectDeadline(storage.release(runnerAddress1, shards[0]))
86 yield* expectRecovery()
87 yield* storage.release(runnerAddress1, shards[0])
88
89 assert.strictEqual(partitioned.activeQueries, 0)
90 }).pipe(
91 Effect.timeoutOrElse({
92 duration: 30_000,
93 orElse: () =>
94 Effect.die(
95 `timed out exercising shard lock rebuilds (${partitionDiagnostics(partitioned)})`
96 )
97 }),
98 // Ensure layer teardown cannot mask a body failure with Vitest's timeout.
99 Effect.ensuring(Effect.sync(() => {
100 restoreConnection(partitioned)
101 })),
102 Effect.provide(layer),
103 TestClock.withLive
104 )
105 }, 60_000)
106
107 it.effect("empty liveness probe bypasses a wedged reserved connection", () => {
108 const partitioned = makePartitionState()
109 const layer = StorageLayer.pipe(
110 Layer.provideMerge(blackholeReservedConnection(partitioned, true)),
111 Layer.provide(ShardingConfig.layer(partitionConfig))
112 )
113
114 return Effect.gen(function*() {
115 const storage = yield* RunnerStorage.RunnerStorage
116 const runner = Runner.make({
117 address: runnerAddress1,
118 groups: ["default"],
119 weight: 1
120 })
121 const shards = [ShardId.make("default", 1)]
122
123 yield* storage.register(runner, true)
124 yield* storage.acquire(runnerAddress1, shards)
125 partitionConnection(partitioned)
126
127 // shard lock operations are stuck behind the wedged reserved connection
128 const exit = yield* storage.refresh(runnerAddress1, shards).pipe(Effect.exit)
129 assert(Exit.isFailure(exit))
130
131 // the liveness probe runs on the shared pool, so it must succeed while
132 // the reserved connection stays wedged
133 expect(yield* storage.refresh(runnerAddress1, [])).toEqual([])
134
135 restoreConnection(partitioned)
136 expect(
137 yield* storage.refresh(runnerAddress1, shards).pipe(
138 Effect.retry({ times: 20, schedule: Schedule.spaced(20) })
139 )
140 ).toEqual(shards)
141 }).pipe(
142 Effect.ensuring(Effect.sync(() => {
143 restoreConnection(partitioned)
144 })),
145 Effect.provide(layer),
146 TestClock.withLive
147 )
148 }, 60_000)
149
150 it.effect("recovers when a blackholed query cannot resume after the partition clears", () => {
151 const partitioned = makePartitionState()
152 const layer = StorageLayer.pipe(
153 Layer.provideMerge(blackholeReservedConnection(partitioned, false)),
154 Layer.provide(ShardingConfig.layer({
155 shardLockDisableAdvisory: true,
156 ...partitionConfig
157 }))
158 )
159
160 return Effect.gen(function*() {
161 const storage = yield* RunnerStorage.RunnerStorage
162 const runner = Runner.make({
163 address: runnerAddress1,
164 groups: ["default"],
165 weight: 1
166 })
167 const shards = [ShardId.make("default", 1)]
168
169 yield* storage.register(runner, true)
170 yield* storage.acquire(runnerAddress1, shards)
171 partitionConnection(partitioned)
172 yield* storage.refresh(runnerAddress1, shards).pipe(Effect.exit)
173 yield* waitUntil(() => partitioned.activeQueries === 0)
174
175 restoreConnection(partitioned)
176 expect(
177 yield* storage.refresh(runnerAddress1, shards).pipe(
178 Effect.retry({ times: 20, schedule: Schedule.spaced(20) })
179 )
180 ).toEqual(shards)
181 assert.isAtLeast(partitioned.interruptedQueries, 1)
182 assert.isAtMost(partitioned.maxActiveQueries, 1)
183 }).pipe(
184 Effect.ensuring(Effect.sync(() => {
185 restoreConnection(partitioned)
186 })),
187 Effect.provide(layer),
188 TestClock.withLive
189 )
190 }, 60_000)
191
192 it.effect("rebuilds the reserved connection again when a rebuilt connection stops responding", () => {
193 const partitioned = makePartitionState()
194 const layer = StorageLayer.pipe(
195 Layer.provideMerge(blackholeReservedConnection(partitioned, false)),
196 Layer.provide(ShardingConfig.layer(partitionConfig))
197 )
198
199 return Effect.gen(function*() {
200 const storage = yield* RunnerStorage.RunnerStorage
201 const runner = Runner.make({
202 address: runnerAddress1,
203 groups: ["default"],
204 weight: 1
205 })
206 const shards = [ShardId.make("default", 1)]
207
208 yield* storage.register(runner, true)
209 yield* storage.acquire(runnerAddress1, shards)
210
211 // a failing lock operation rebuilds the reserved connection
212 partitioned.failNextQueries = 1
213 yield* storage.refresh(runnerAddress1, shards).pipe(Effect.exit)
214 yield* waitUntil(() => partitioned.usableConnections === 2)
215
216 // the rebuilt connection then wedges, without any lock operation
217 // succeeding in between - a further rebuild still has to be attempted
218 const reserved = partitioned.reservedConnections
219 partitionConnection(partitioned)
220 yield* storage.refresh(runnerAddress1, shards).pipe(Effect.exit)
221 yield* storage.refresh(runnerAddress1, shards).pipe(Effect.exit)
222 restoreConnection(partitioned)
223 yield* waitUntil(() => partitioned.reservedConnections > reserved)
224
225 expect(
226 yield* storage.refresh(runnerAddress1, shards).pipe(
227 Effect.retry({ times: 5, schedule: Schedule.spaced(20) })
228 )
229 ).toEqual(shards)
230 }).pipe(
231 Effect.ensuring(Effect.sync(() => {
232 restoreConnection(partitioned)
233 })),
234 Effect.provide(layer),
235 TestClock.withLive
236 )
237 }, 60_000)
238
239 it.effect(
240 "rebuilds the reserved connection when releasing the previous one hangs",
241 () =>
242 withStalledRelease((storage, shards, partitioned) =>
243 Effect.gen(function*() {
244 yield* Effect.sleep(lockOperationInterval * 1.5)
245
246 // the stalled release must not disable further rebuilds
247 restoreConnection(partitioned)
248 const reserved = partitioned.reservedConnections
249 yield* storage.refresh(runnerAddress1, shards).pipe(Effect.exit)
250 yield* waitUntil(() => partitioned.reservedConnections > reserved)
251
252 expect(
253 yield* storage.refresh(runnerAddress1, shards).pipe(
254 Effect.retry({ times: 20, schedule: Schedule.spaced(20) })
255 )
256 ).toEqual(shards)
257 })
258 ),
259 60_000
260 )
261
262 it.effect(
263 "replaces a pending reserved connection once when operations waiting on it fail",
264 () =>
265 withStalledRelease((storage, shards, partitioned) =>
266 Effect.gen(function*() {
267 restoreConnection(partitioned)
268 const reserved = partitioned.reservedConnections
269
270 // both operations wait on the pending connection and time out together,
271 // but only one of them may replace it
272 const exits = yield* Effect.all([
273 storage.refresh(runnerAddress1, shards).pipe(Effect.exit),
274 storage.refresh(runnerAddress1, shards).pipe(Effect.exit)
275 ], { concurrency: "unbounded" })
276 assert(exits.every(Exit.isFailure))
277 yield* waitUntil(() => partitioned.reservedConnections > reserved)
278 expect(
279 yield* storage.refresh(runnerAddress1, shards).pipe(
280 Effect.retry({ times: 20, schedule: Schedule.spaced(20) })
281 )
282 ).toEqual(shards)
283 yield* Effect.sleep(lockOperationInterval / 2)
284 expect(partitioned.reservedConnections, partitionDiagnostics(partitioned)).toEqual(reserved + 1)
285 })
286 ),
287 60_000
288 )
289
290 it.effect(
291 "replaces a connection that fails an operation which waited for it",
292 () =>
293 withStalledRelease((storage, shards, partitioned) =>
294 Effect.gen(function*() {
295 restoreConnection(partitioned)
296 const waiting = yield* storage.refresh(runnerAddress1, shards).pipe(Effect.exit, Effect.forkChild)
297 yield* Effect.sleep(lockOperationInterval / 10)
298
299 // the waiting operation runs on the replacement, which then fails it
300 const reserved = partitioned.reservedConnections
301 partitioned.failNextQueries = 1
302 partitioned.blockRelease = false
303 yield* waitUntil(() => partitioned.reservedConnections > reserved)
304 assert(Exit.isFailure(yield* Fiber.join(waiting)))
305
306 // the failing replacement is replaced without another operation failing on it
307 yield* waitUntil(() => partitioned.reservedConnections > reserved + 1)
308 expect(
309 yield* storage.refresh(runnerAddress1, shards).pipe(
310 Effect.retry({ times: 20, schedule: Schedule.spaced(20) })
311 )
312 ).toEqual(shards)
313 })
314 ),
315 60_000
316 )
317
318 it.effect("isolates advisory shard locks by prefix", () =>
319 Effect.gen(function*() {
320 const storageA = yield* SqlRunnerStorage.make({ prefix: "cluster" })
321 const storageB = yield* SqlRunnerStorage.make({ prefix: "other" })
322 const shard = ShardId.make("default", 1)
323
324 expect(yield* storageA.acquire(runnerAddress1, [shard])).toEqual([shard])
325 expect(yield* storageB.acquire(runnerAddress2, [shard])).toEqual([shard])
326 }).pipe(
327 // Release the advisory-lock connections before the PostgreSQL layer.
328 Effect.scoped,
329 Effect.provide(PgContainer.layerClient),
330 Effect.provide(ShardingConfig.layer())
331 ), 60_000)
332
333 it.effect("excludes other storages using the same prefix", () =>
334 Effect.gen(function*() {
335 const storageA = yield* SqlRunnerStorage.make({ prefix: "cluster" })
336 const storageB = yield* SqlRunnerStorage.make({ prefix: "cluster" })
337 const shard = ShardId.make("default", 1)
338
339 expect(yield* storageA.acquire(runnerAddress1, [shard])).toEqual([shard])
340 expect(yield* storageB.acquire(runnerAddress2, [shard])).toEqual([])
341 }).pipe(
342 // Release the advisory-lock connections before the PostgreSQL layer.
343 Effect.scoped,
344 Effect.provide(PgContainer.layerClient),
345 Effect.provide(ShardingConfig.layer())
346 ), 60_000)
347
348 it.effect(
349 "pg (no advisory) does not deadlock acquisition with refresh or bulk release",
350 () =>
351 Effect.gen(function*() {
352 const sql = yield* SqlClient.SqlClient
353 const storageA = yield* SqlRunnerStorage.make({ prefix: "cluster" })
354 const storageB = yield* SqlRunnerStorage.make({ prefix: "cluster" })
355 const shards = Array.from({ length: 12 }, (_, i) => ShardId.make("default", i + 1))
356 const reversed = [...shards].reverse()
357 // Rewrite each lease in reverse so the physical row order differs from
358 // the sorted lock order.
359 const acquireShuffled = Effect.gen(function*() {
360 yield* storageA.acquire(runnerAddress1, shards)
361 for (const shard of reversed) {
362 yield* storageA.refresh(runnerAddress1, [shard])
363 }
364 })
365
366 // Slow each row write so the concurrent statements overlap.
367 yield* sql`CREATE FUNCTION delay_row() RETURNS trigger LANGUAGE plpgsql AS $$
368 BEGIN PERFORM pg_sleep(0.003); RETURN COALESCE(NEW, OLD); END $$`
369 yield* sql`CREATE TRIGGER delay_row BEFORE UPDATE OR DELETE ON cluster_locks
370 FOR EACH ROW EXECUTE FUNCTION delay_row()`
371
372 for (let i = 0; i < 8; i++) {
373 yield* acquireShuffled
374 yield* sql`UPDATE cluster_locks SET acquired_at = NOW() - INTERVAL '1 hour'`
375 yield* Effect.all([
376 storageA.refresh(runnerAddress1, shards),
377 storageB.acquire(runnerAddress2, reversed)
378 ], { concurrency: 2 })
379 yield* storageA.releaseAll(runnerAddress1)
380 yield* storageB.releaseAll(runnerAddress2)
381
382 yield* acquireShuffled
383 yield* Effect.all([
384 storageA.releaseAll(runnerAddress1),
385 storageB.acquire(runnerAddress2, reversed)
386 ], { concurrency: 2 })
387 yield* storageB.releaseAll(runnerAddress2)
388 }
389 }).pipe(
390 Effect.scoped,
391 Effect.provide(PgContainer.layerClient),
392 Effect.provide(ShardingConfig.layer({ shardLockDisableAdvisory: true })),
393 TestClock.withLive
394 ),
395 60_000
396 )
397 ;([
398 ["pg", Layer.orDie(PgContainer.layerClient)],
399 ["mysql", Layer.orDie(MysqlContainer.layerClient)],
400 ["vitess", Layer.orDie(MysqlContainer.layerClientVitess)],
401 ["sqlite", Layer.orDie(SqliteLayer)]
402 ] as const).flatMap(([label, layer]) =>
403 [
404 [label, StorageLayer.pipe(Layer.provideMerge(layer), Layer.provide(ShardingConfig.layer()))],
405 [
406 label + " (no advisory)",
407 StorageLayer.pipe(
408 Layer.provideMerge(layer),
409 Layer.provide(ShardingConfig.layer({
410 shardLockDisableAdvisory: true
411 }))
412 )
413 ]
414 ] as const
415 ).forEach(([label, layer]) => {
416 // Both tests update the same runner rows.
417 it.layer(layer, {
418 concurrent: false,
419 timeout: 60000
420 })(label, (it) => {
421 it.effect("getRunners", () =>
422 Effect.gen(function*() {
423 const storage = yield* RunnerStorage.RunnerStorage
424
425 const runner = Runner.make({
426 address: runnerAddress1,
427 groups: ["default"],
428 weight: 1
429 })
430 const machineId = yield* storage.register(runner, true)
431 yield* storage.register(runner, true)
432 expect(machineId).toEqual(1)
433 expect(yield* storage.getRunners).toEqual([[runner, true]])
434
435 yield* storage.setRunnerHealth(runnerAddress1, false)
436 expect(yield* storage.getRunners).toEqual([[runner, false]])
437
438 yield* storage.unregister(runnerAddress1)
439 expect(yield* storage.getRunners).toEqual([])
440 }), { timeout: 30_000 })
441
442 it.effect("acquireShards", () =>
443 Effect.gen(function*() {
444 const storage = yield* RunnerStorage.RunnerStorage
445
446 let acquired = yield* storage.acquire(runnerAddress1, [
447 ShardId.make("default", 1),
448 ShardId.make("default", 2),
449 ShardId.make("default", 3)
450 ])
451 expect(acquired.map((_) => _.id)).toEqual([1, 2, 3])
452 acquired = yield* storage.acquire(runnerAddress1, [
453 ShardId.make("default", 1),
454 ShardId.make("default", 2),
455 ShardId.make("default", 3)
456 ])
457 expect(acquired.map((_) => _.id)).toEqual([1, 2, 3])
458 acquired = yield* storage.acquire(runnerAddress1, [ShardId.make("default", 4)])
459 expect(acquired.map((_) => _.id)).toEqual([4])
460
461 const refreshed = yield* storage.refresh(runnerAddress1, [
462 ShardId.make("default", 1),
463 ShardId.make("default", 2),
464 ShardId.make("default", 3)
465 ])
466 expect(refreshed.map((_) => _.id).sort((a, b) => a - b)).toEqual([1, 2, 3])
467
468 // smoke test release
469 yield* storage.release(runnerAddress1, ShardId.make("default", 2))
470 }))
471 })
472 })
473})
474
475const runnerAddress1 = RunnerAddress.make("localhost", 1234)
476const runnerAddress2 = RunnerAddress.make("localhost", 5678)
477
478interface PartitionState {
479 connectionAvailable: Latch.Latch
480 blockRelease: boolean
481 activeQueries: number
482 maxActiveQueries: number
483 interruptedQueries: number
484 failNextQueries: number
485 reservedConnections: number
486 usableConnections: number
487}
488
489// Registers a runner holding one shard, then wedges the reserved connection
490// and blocks its release, so the first rebuild stalls closing the previous
491// scope. `body` runs with the stall in place and may lift it through
492// `partitioned`; it is lifted afterwards so the layer can be torn down.
493const withStalledRelease = (
494 body: (
495 storage: RunnerStorage.RunnerStorage["Service"],
496 shards: ReadonlyArray<ShardId.ShardId>,
497 partitioned: PartitionState
498 ) => Effect.Effect<void, unknown>
499) => {
500 const partitioned = makePartitionState()
501 const layer = StorageLayer.pipe(
502 Layer.provideMerge(blackholeReservedConnection(partitioned, false)),
503 Layer.provide(ShardingConfig.layer({
504 shardLockDisableAdvisory: true,
505 ...partitionConfig
506 }))
507 )
508 return Effect.gen(function*() {
509 const storage = yield* RunnerStorage.RunnerStorage
510 const runner = Runner.make({
511 address: runnerAddress1,
512 groups: ["default"],
513 weight: 1
514 })
515 const shards = [ShardId.make("default", 1)]
516 yield* storage.register(runner, true)
517 yield* storage.acquire(runnerAddress1, shards)
518 partitionConnection(partitioned)
519 partitioned.blockRelease = true
520 yield* storage.refresh(runnerAddress1, shards).pipe(Effect.exit)
521 yield* body(storage, shards, partitioned)
522 }).pipe(
523 Effect.ensuring(Effect.sync(() => {
524 restoreConnection(partitioned)
525 partitioned.blockRelease = false
526 })),
527 Effect.provide(layer),
528 TestClock.withLive
529 )
530}
531
532const makePartitionState = (): PartitionState => ({
533 connectionAvailable: Latch.makeUnsafe(true),
534 blockRelease: false,
535 activeQueries: 0,
536 maxActiveQueries: 0,
537 interruptedQueries: 0,
538 failNextQueries: 0,
539 reservedConnections: 0,
540 usableConnections: 0
541})
542
543const partitionConnection = (partitioned: PartitionState) => {
544 Latch.closeUnsafe(partitioned.connectionAvailable)
545}
546
547const restoreConnection = (partitioned: PartitionState) => {
548 Latch.openUnsafe(partitioned.connectionAvailable)
549}
550
551const partitionDiagnostics = (partitioned: PartitionState) =>
552 [
553 `active=${partitioned.activeQueries}`,
554 `interrupted=${partitioned.interruptedQueries}`,
555 `reserved=${partitioned.reservedConnections}`,
556 `usable=${partitioned.usableConnections}`
557 ].join(", ")
558
559const waitUntil = Effect.fnUntraced(
560 function*(predicate: () => boolean) {
561 while (!predicate()) {
562 yield* Effect.sleep(20)
563 }
564 },
565 Effect.timeoutOrElse({
566 duration: 10_000,
567 orElse: () => Effect.die("timed out waiting for condition")
568 })
569)
570
571const blackholeReservedConnection = (partitioned: PartitionState, resumePending: boolean) =>
572 Layer.effect(
573 SqlClient.SqlClient,
574 Effect.gen(function*() {
575 const sql = yield* SqlClient.SqlClient
576 const wrapConnection = (connection: SqlConnection.Connection): SqlConnection.Connection => {
577 let usable = false
578 const execute = <A, E extends SqlError.SqlError, R>(effect: Effect.Effect<A, E, R>) =>
579 Effect.suspend((): Effect.Effect<A, E | SqlError.SqlError, R> => {
580 if (partitioned.failNextQueries > 0) {
581 partitioned.failNextQueries--
582 return Effect.fail(
583 new SqlError.SqlError({
584 reason: new SqlError.ConnectionError({ cause: new Error("connection lost") })
585 })
586 )
587 }
588 partitioned.activeQueries++
589 partitioned.maxActiveQueries = Math.max(partitioned.maxActiveQueries, partitioned.activeQueries)
590 return (!Latch.isOpen(partitioned.connectionAvailable)
591 ? resumePending
592 ? Effect.andThen(Latch.await(partitioned.connectionAvailable), effect)
593 : Effect.never
594 : effect).pipe(
595 Effect.onExit((exit) =>
596 Effect.sync(() => {
597 partitioned.activeQueries--
598 if (Exit.hasInterrupts(exit)) {
599 partitioned.interruptedQueries++
600 }
601 if (Exit.isSuccess(exit) && !usable) {
602 usable = true
603 partitioned.usableConnections++
604 }
605 })
606 )
607 )
608 })
609 return {
610 ...connection,
611 execute: (...args) => execute(connection.execute(...args)),
612 executeRaw: (...args) => execute(connection.executeRaw(...args)),
613 executeValues: (...args) => execute(connection.executeValues(...args)),
614 executeValuesUnprepared: (...args) => execute(connection.executeValuesUnprepared(...args)),
615 executeUnprepared: (...args) => execute(connection.executeUnprepared(...args))
616 }
617 }
618 let client: SqlClient.SqlClient
619 client = new Proxy(sql, {
620 get(target, property, receiver) {
621 if (property === "reserve") {
622 return Effect.andThen(
623 // simulates a driver that stalls uninterruptibly while tearing
624 // down a reserved connection, so closing its scope cannot be
625 // interrupted
626 Effect.addFinalizer(() =>
627 Effect.uninterruptible(Effect.suspend(function waitForRelease(): Effect.Effect<void> {
628 if (!partitioned.blockRelease) return Effect.void
629 return Effect.andThen(
630 Effect.promise(() => new Promise<void>((resolve) => setTimeout(resolve, 5))),
631 waitForRelease
632 )
633 }))
634 ),
635 Effect.map(target.reserve, (connection) => {
636 partitioned.reservedConnections++
637 return wrapConnection(connection)
638 })
639 )
640 }
641 if (property === "withoutTransforms") {
642 return () => client
643 }
644 return Reflect.get(target, property, receiver)
645 }
646 })
647 return client
648 })
649 ).pipe(Layer.provide(PgContainer.layerClient))
650
651const SqliteLayer = Effect.gen(function*() {
652 const fs = yield* FileSystem.FileSystem
653 const dir = yield* fs.makeTempDirectoryScoped()
654 return SqliteClient.layer({
655 filename: dir + "/test.db"
656 })
657}).pipe(Layer.unwrap, Layer.provide(NodeFileSystem.layer))