don't drop queue elements when takeBetween is interrupted

.changeset/tame-ligers-learn.md

@@ -0,0 +1,5 @@
+---
+"@effect/io": patch
+---
+
+don't drop queue elements when takeBetween is interrupted

src/internal/queue.ts

@@ -279,7 +279,7 @@ class QueueImpl<A> implements Queue.Queue<A> {
 
 /** @internal */
 const takeRemainderLoop = <A>(
-  self: Queue.Dequeue<A>,
+  self: Queue.Queue<A>,
   min: number,
   max: number,
   acc: Chunk.Chunk<A>
@@ -291,10 +291,12 @@ const takeRemainderLoop = <A>(
     takeUpTo(self, max),
     core.flatMap((bs) => {
       const remaining = min - bs.length
+      acc = Chunk.appendAll(acc, bs)
       if (remaining === 1) {
         return pipe(
           take(self),
-          core.map((b) => pipe(acc, Chunk.appendAll(bs), Chunk.append(b)))
+          core.map((b) => Chunk.append(acc, b)),
+          core.onInterrupt(() => offerAll(self, acc))
         )
       }
       if (remaining > 1) {
@@ -305,12 +307,13 @@ const takeRemainderLoop = <A>(
               self,
               remaining - 1,
               max - bs.length - 1,
-              pipe(acc, Chunk.appendAll(bs), Chunk.append(b))
+              Chunk.append(acc, b)
             )
-          )
+          ),
+          core.onInterrupt(() => offerAll(self, acc))
         )
       }
-      return core.succeed(pipe(acc, Chunk.appendAll(bs)))
+      return core.succeed(acc)
     })
   )
 }

test/Queue.ts

@@ -618,6 +618,26 @@ describe.concurrent("Queue", () => {
       yield* $(Fiber.interrupt(fiber))
       assert.deepStrictEqual(Array.from(result), values)
     }))
+  it.effect("takeBetween doesn't drop elements if interrupted", () =>
+    Effect.gen(function*($) {
+      const queue = yield* $(Queue.unbounded<number>())
+      yield* $(Queue.offer(queue, 1))
+      yield* $(Queue.offer(queue, 2))
+      yield* $(Queue.offer(queue, 3))
+
+      let fiber = yield* $(Queue.takeBetween(queue, 4, 4), Effect.fork)
+      yield* $(Effect.yieldNow())
+      yield* $(Fiber.interrupt(fiber))
+      let size = yield* $(Queue.size(queue))
+      assert.deepStrictEqual(size, 3)
+
+      // test for when remaining > 1
+      fiber = yield* $(Queue.takeBetween(queue, 5, 5), Effect.fork)
+      yield* $(Effect.yieldNow())
+      yield* $(Fiber.interrupt(fiber))
+      size = yield* $(Queue.size(queue))
+      assert.deepStrictEqual(size, 3)
+    }))
   it.effect("takeN returns immediately if there is enough elements", () =>
     Effect.gen(function*($) {
       const queue = yield* $(Queue.bounded<number>(100))