Fix asyncPush bounded buffer termination

.changeset/tame-streams-end.md

@@ -0,0 +1,5 @@
+---
+"effect": patch
+---
+
+Reserve queue capacity for termination signals in `Stream.asyncPush` with bounded dropping or sliding buffers.

packages/effect/src/internal/stream.ts

@@ -603,9 +603,9 @@ const queueFromBufferOptionsPush = <A, E>(
   }
   switch (options?.strategy) {
     case "sliding":
-      return Queue.sliding(options.bufferSize ?? 16)
+      return Queue.sliding((options.bufferSize ?? 16) + 1)
     default:
-      return Queue.dropping(options?.bufferSize ?? 16)
+      return Queue.dropping((options?.bufferSize ?? 16) + 1)
   }
 }
 

packages/effect/test/Stream/async.test.ts

@@ -438,6 +438,21 @@ describe("Stream", () => {
       assertTrue(Chunk.isEmpty(result))
     }))
 
+  for (const strategy of ["dropping", "sliding"] as const) {
+    it.effect(`asyncPush - signals the end with a bounded ${strategy} buffer`, () =>
+      Effect.gen(function*() {
+        const result = yield* Stream.asyncPush<number>((emit) => {
+          emit.single(42)
+          emit.end()
+          return Effect.void
+        }, { bufferSize: 1, strategy }).pipe(
+          Stream.runCollect,
+          Effect.timeout("1 second")
+        )
+        deepStrictEqual(Array.from(result), [42])
+      }))
+  }
+
   it.effect("asyncPush - handles errors", () =>
     Effect.gen(function*() {
       const error = new Cause.RuntimeException("boom")