fix semaphore race condition where permits could be leaked

.changeset/smooth-states-kick.md

@@ -0,0 +1,5 @@
+---
+"effect": patch
+---
+
+fix semaphore race condition where permits could be leaked

packages/effect/src/internal/effect/circular.ts

@@ -47,20 +47,24 @@ class Semaphore {
     core.asyncInterrupt<number>((resume) => {
       if (this.free < n) {
         const observer = () => {
-          if (this.free < n) {
-            return
-          }
+          if (this.free < n) return
           this.waiters.delete(observer)
-          this.taken += n
-          resume(core.succeed(n))
+          resume(core.suspend(() => {
+            if (this.free < n) return this.take(n)
+            this.taken += n
+            return core.succeed(n)
+          }))
         }
         this.waiters.add(observer)
         return core.sync(() => {
           this.waiters.delete(observer)
         })
       }
-      this.taken += n
-      return resume(core.succeed(n))
+      resume(core.suspend(() => {
+        if (this.free < n) return this.take(n)
+        this.taken += n
+        return core.succeed(n)
+      }))
     })
 
   updateTakenUnsafe(fiber: Fiber.RuntimeFiber<any, any>, f: (n: number) => number): Effect.Effect<number> {

packages/effect/test/Effect/semaphore.test.ts

@@ -2,6 +2,9 @@ import { assert, describe, it } from "@effect/vitest"
 import { strictEqual } from "@effect/vitest/utils"
 import * as D from "effect/Duration"
 import * as Effect from "effect/Effect"
+import * as FiberId from "effect/FiberId"
+import * as Option from "effect/Option"
+import * as Scheduler from "effect/Scheduler"
 import * as TestClock from "effect/TestClock"
 
 describe("Effect", () => {
@@ -56,4 +59,27 @@ describe("Effect", () => {
       yield* TestClock.adjust(1)
       assert.isTrue(fiber.unsafePoll() !== null)
     }))
+
+  it.effect("take interruption does not leak permits", () =>
+    Effect.gen(function*() {
+      const scheduler = new Scheduler.ControlledScheduler()
+      const sem = yield* Effect.makeSemaphore(0)
+      const waiter = yield* sem.take(1).pipe(
+        Effect.withScheduler(scheduler),
+        Effect.fork
+      )
+
+      yield* Effect.yieldNow()
+      yield* sem.release(1).pipe(Effect.withScheduler(scheduler))
+      assert.isNull(waiter.unsafePoll())
+
+      scheduler.step()
+      assert.isNull(waiter.unsafePoll())
+
+      waiter.unsafeInterruptAsFork(FiberId.none)
+      scheduler.step()
+
+      const result = yield* sem.withPermitsIfAvailable(1)(Effect.void)
+      assert.isTrue(Option.isSome(result))
+    }))
 })