diff --git a/packages/core/src/job.ts b/packages/core/src/job.ts index 8976447386b..08faea74e1e 100644 --- a/packages/core/src/job.ts +++ b/packages/core/src/job.ts @@ -55,7 +55,6 @@ type Active = { done: Deferred.Deferred backgrounded: Deferred.Deferred scope: Scope.Closeable - token: object blockingSessions: Map isBackgrounded: boolean recovery?: Recovery @@ -77,7 +76,7 @@ type BackgroundResult = { backgrounded?: Deferred.Deferred } -type StartResult = { info: Info } | { info: Info; scope: Scope.Closeable; token: object } +type StartResult = { info: Info } | { info: Info; scope: Scope.Closeable } type BlockWait = { done: Deferred.Deferred @@ -184,14 +183,14 @@ export const make = Effect.gen(function* () { }) }) - const settle = Effect.fnUntraced(function* (id: string, token: object, exit: Exit.Exit) { + const settle = Effect.fnUntraced(function* (id: string, scope: Scope.Closeable, exit: Exit.Exit) { const completed_at = yield* Clock.currentTimeMillis const result = yield* SynchronizedRef.modifyEffect( state.jobs, Effect.fnUntraced(function* (jobs): Effect.fn.Return]> { const job = jobs.get(id) if (!job) return [{}, jobs] - if (job.token !== token) return [{}, jobs] + if (job.scope !== scope) return [{}, jobs] if (job.info.status !== "running") return [{ info: snapshot(job) }, jobs] const status: Exclude = Exit.isSuccess(exit) ? "completed" @@ -241,7 +240,6 @@ export const make = Effect.gen(function* () { return [{ info: snapshot(existing) }, jobs] } const scope = yield* Scope.fork(state.scope, "parallel") - const token = {} const job = { info: { id, @@ -255,18 +253,17 @@ export const make = Effect.gen(function* () { done, backgrounded, scope, - token, blockingSessions: new Map(), isBackgrounded: false, recovery: input.recovery, } - return [{ info: snapshot(job), scope, token }, new Map(jobs).set(id, job)] + return [{ info: snapshot(job), scope }, new Map(jobs).set(id, job)] }), ) if ("scope" in result) yield* restore(input.run).pipe( Effect.exit, - Effect.flatMap((exit) => settle(id, result.token, exit)), + Effect.flatMap((exit) => settle(id, result.scope, exit)), Effect.asVoid, Effect.forkIn(result.scope, { startImmediately: true }), ) diff --git a/packages/core/test/job.test.ts b/packages/core/test/job.test.ts index e08159dfc5a..62302facd2e 100644 --- a/packages/core/test/job.test.ts +++ b/packages/core/test/job.test.ts @@ -64,6 +64,71 @@ describe("Job", () => { }), ) + it.live("reuses running work when started again with the same ID", () => + Effect.gen(function* () { + const jobs = yield* Job.Service + const output = yield* Deferred.make() + const job = yield* jobs.start({ id: "job_reused", type: "test", run: Deferred.await(output) }) + + expect( + yield* jobs.start({ id: job.id, type: "duplicate", run: Effect.die("Duplicate work must not run") }), + ).toEqual(job) + + yield* Deferred.succeed(output, "original output") + expect((yield* jobs.wait({ id: job.id })).info).toMatchObject({ + type: "test", + status: "completed", + output: "original output", + }) + }), + ) + + it.live("ignores an obsolete callback after a cancellation waiter starts a same-ID replacement", () => + Effect.gen(function* () { + const jobs = yield* Job.Service + const callback = yield* Deferred.make<() => void>() + const output = yield* Deferred.make() + const finalized = yield* Deferred.make() + const job = yield* jobs.start({ + id: "job_replaced", + type: "test", + run: Effect.callback((resume) => { + Deferred.doneUnsafe( + callback, + Effect.succeed(() => resume(Effect.succeed("obsolete output"))), + ) + }), + }) + const complete = yield* Deferred.await(callback) + // Cancellation wakes waiters before closing the old scope, allowing the old callback to race replacement. + const replacement = yield* jobs.wait({ id: job.id }).pipe( + Effect.tap((result) => Effect.sync(() => expect(result.info?.status).toBe("cancelled"))), + Effect.andThen( + jobs.start({ + id: job.id, + type: "replacement", + run: Deferred.await(output).pipe(Effect.ensuring(Deferred.succeed(finalized, undefined))), + }), + ), + Effect.andThen(Effect.sync(complete)), + Effect.forkChild({ startImmediately: true }), + ) + + yield* jobs.cancel(job.id) + yield* Fiber.join(replacement) + expect(yield* jobs.get(job.id)).toMatchObject({ type: "replacement", status: "running" }) + expect(yield* Deferred.isDone(finalized)).toBe(false) + + yield* Deferred.succeed(output, "replacement output") + expect((yield* jobs.wait({ id: job.id })).info).toMatchObject({ + type: "replacement", + status: "completed", + output: "replacement output", + }) + expect(yield* Deferred.isDone(finalized)).toBe(true) + }), + ) + it.live("returns finished from a blocking wait when completion wins", () => Effect.gen(function* () { const jobs = yield* Job.Service