|
|
@@ -1,4 +1,4 @@
|
|
|
-import { describe, expect, test } from "bun:test"
|
|
|
+import { describe, expect } from "bun:test"
|
|
|
import { Deferred, Effect, Exit, Fiber, Ref, Scope } from "effect"
|
|
|
import { Runner } from "@/effect/runner"
|
|
|
import { it } from "../lib/effect"
|
|
|
@@ -198,58 +198,52 @@ describe("Runner", () => {
|
|
|
}),
|
|
|
)
|
|
|
|
|
|
- test("cancel does not deadlock when replacement work starts before interrupted run exits", async () => {
|
|
|
- function defer() {
|
|
|
- let resolve!: () => void
|
|
|
- const promise = new Promise<void>((done) => {
|
|
|
- resolve = done
|
|
|
- })
|
|
|
- return { promise, resolve }
|
|
|
- }
|
|
|
+ it.live(
|
|
|
+ "cancel does not deadlock when replacement work starts before interrupted run exits",
|
|
|
+ Effect.gen(function* () {
|
|
|
+ const s = yield* Scope.Scope
|
|
|
+ const hit = yield* Deferred.make<void>()
|
|
|
+ const hold = yield* Deferred.make<void>()
|
|
|
+ const done = yield* Deferred.make<void>()
|
|
|
+
|
|
|
+ yield* Effect.gen(function* () {
|
|
|
+ const runner = Runner.make<string>(s)
|
|
|
+ const first = Effect.never.pipe(
|
|
|
+ Effect.onInterrupt(() => Deferred.succeed(hit, undefined)),
|
|
|
+ Effect.ensuring(Deferred.await(hold)),
|
|
|
+ Effect.as("first"),
|
|
|
+ )
|
|
|
|
|
|
- function fail(ms: number, msg: string) {
|
|
|
- return new Promise<never>((_, reject) => {
|
|
|
- setTimeout(() => reject(new Error(msg)), ms)
|
|
|
- })
|
|
|
- }
|
|
|
+ const a = yield* runner.ensureRunning(first).pipe(Effect.exit, Effect.forkChild)
|
|
|
+ yield* Effect.sleep("10 millis")
|
|
|
|
|
|
- const s = await Effect.runPromise(Scope.make())
|
|
|
- const hit = defer()
|
|
|
- const hold = defer()
|
|
|
- const done = defer()
|
|
|
- try {
|
|
|
- const runner = Runner.make<string>(s)
|
|
|
- const first = Effect.never.pipe(
|
|
|
- Effect.onInterrupt(() => Effect.sync(() => hit.resolve())),
|
|
|
- Effect.ensuring(Effect.promise(() => hold.promise)),
|
|
|
- Effect.as("first"),
|
|
|
+ const stop = yield* runner.cancel.pipe(Effect.forkChild)
|
|
|
+ yield* Deferred.await(hit).pipe(Effect.timeout("250 millis"))
|
|
|
+
|
|
|
+ const b = yield* runner.ensureRunning(Deferred.await(done).pipe(Effect.as("second"))).pipe(Effect.forkChild)
|
|
|
+ yield* Effect.yieldNow
|
|
|
+ expect(runner.busy).toBe(true)
|
|
|
+
|
|
|
+ yield* Deferred.succeed(hold, undefined)
|
|
|
+ const stopExit = yield* Fiber.await(stop).pipe(Effect.timeout("250 millis"))
|
|
|
+ expect(Exit.isSuccess(stopExit)).toBe(true)
|
|
|
+
|
|
|
+ expect(runner.busy).toBe(true)
|
|
|
+ yield* Deferred.succeed(done, undefined)
|
|
|
+ expect(yield* Fiber.join(b).pipe(Effect.timeout("250 millis"))).toBe("second")
|
|
|
+ expect(runner.busy).toBe(false)
|
|
|
+
|
|
|
+ const exit = yield* Fiber.join(a)
|
|
|
+ expect(Exit.isFailure(exit)).toBe(true)
|
|
|
+ }).pipe(
|
|
|
+ Effect.ensuring(
|
|
|
+ Effect.all([Deferred.succeed(hold, undefined), Deferred.succeed(done, undefined)], { discard: true }).pipe(
|
|
|
+ Effect.ignore,
|
|
|
+ ),
|
|
|
+ ),
|
|
|
)
|
|
|
-
|
|
|
- const a = Effect.runPromiseExit(runner.ensureRunning(first))
|
|
|
- await Bun.sleep(10)
|
|
|
-
|
|
|
- const stop = Effect.runPromise(runner.cancel)
|
|
|
- await Promise.race([hit.promise, fail(250, "cancel did not interrupt running work")])
|
|
|
-
|
|
|
- const b = Effect.runPromise(runner.ensureRunning(Effect.promise(() => done.promise).pipe(Effect.as("second"))))
|
|
|
- expect(runner.busy).toBe(true)
|
|
|
-
|
|
|
- hold.resolve()
|
|
|
- await Promise.race([stop, fail(250, "cancel deadlocked while replacement run was active")])
|
|
|
-
|
|
|
- expect(runner.busy).toBe(true)
|
|
|
- done.resolve()
|
|
|
- expect(await b).toBe("second")
|
|
|
- expect(runner.busy).toBe(false)
|
|
|
-
|
|
|
- const exit = await a
|
|
|
- expect(Exit.isFailure(exit)).toBe(true)
|
|
|
- } finally {
|
|
|
- hold.resolve()
|
|
|
- done.resolve()
|
|
|
- await Promise.race([Effect.runPromise(Scope.close(s, Exit.void)), fail(1000, "runner scope did not close")])
|
|
|
- }
|
|
|
- })
|
|
|
+ }),
|
|
|
+ )
|
|
|
|
|
|
// --- shell semantics ---
|
|
|
|