refactor(core): reuse job scope as generation identity (#47121)

packages/core/src/job.ts

@@ -55,7 +55,6 @@ type Active = {
   done: Deferred.Deferred<Info>
   backgrounded: Deferred.Deferred<Info>
   scope: Scope.Closeable
-  token: object
   blockingSessions: Map<SessionSchema.ID, number>
   isBackgrounded: boolean
   recovery?: Recovery
@@ -77,7 +76,7 @@ type BackgroundResult = {
   backgrounded?: Deferred.Deferred<Info>
 }
 
-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<Info>
@@ -184,14 +183,14 @@ export const make = Effect.gen(function* () {
     })
   })
 
-  const settle = Effect.fnUntraced(function* (id: string, token: object, exit: Exit.Exit<string, unknown>) {
+  const settle = Effect.fnUntraced(function* (id: string, scope: Scope.Closeable, exit: Exit.Exit<string, unknown>) {
     const completed_at = yield* Clock.currentTimeMillis
     const result = yield* SynchronizedRef.modifyEffect(
       state.jobs,
       Effect.fnUntraced(function* (jobs): Effect.fn.Return<readonly [FinishResult, Map<string, Active>]> {
         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<Status, "running"> = 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<SessionSchema.ID, number>(),
               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 }),
           )

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<string>()
+      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<string>()
+      const finalized = yield* Deferred.make<void>()
+      const job = yield* jobs.start({
+        id: "job_replaced",
+        type: "test",
+        run: Effect.callback<string>((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