fix(streams): don't lose bytes on drain (#9768)

* fix

* clear

* update

* test

* fix test

---------

Co-authored-by: Jarred Sumner <jarred@jarredsumner.com>

src/bun.js/webcore/streams.zig

@@ -3749,13 +3749,20 @@ pub const FileReader = struct {
 
             if (!bun.isSliceInBuffer(buf, this.buffered.allocatedSlice())) {
                 if (this.reader.isDone()) {
+                    if (bun.isSliceInBuffer(buf, this.reader.buffer().allocatedSlice())) {
+                        this.reader.buffer().* = std.ArrayList(u8).init(bun.default_allocator);
+                    }
                     this.pending.result = .{
                         .temporary_and_done = bun.ByteList.init(buf),
                     };
                 } else {
                     this.pending.result = .{
                         .temporary = bun.ByteList.init(buf),
                     };
+
+                    if (bun.isSliceInBuffer(buf, this.reader.buffer().allocatedSlice())) {
+                        this.reader.buffer().clearRetainingCapacity();
+                    }
                 }
 
                 this.pending_value.clear();
@@ -3780,6 +3787,9 @@ pub const FileReader = struct {
             return !was_done;
         } else if (!bun.isSliceInBuffer(buf, this.buffered.allocatedSlice())) {
             this.buffered.appendSlice(bun.default_allocator, buf) catch bun.outOfMemory();
+            if (bun.isSliceInBuffer(buf, this.reader.buffer().allocatedSlice())) {
+                this.reader.buffer().clearRetainingCapacity();
+            }
         }
 
         // For pipes, we have to keep pulling or the other process will block.
@@ -3886,14 +3896,17 @@ pub const FileReader = struct {
         if (this.buffered.items.len > 0) {
             const out = bun.ByteList.init(this.buffered.items);
             this.buffered = .{};
+            if (comptime Environment.allow_assert) {
+                std.debug.assert(this.reader.buffer().items.ptr != out.ptr);
+            }
             return out;
         }
 
         if (this.reader.hasPendingRead()) {
             return .{};
         }
 
-        const out = this.reader.buffer();
+        const out = this.reader.buffer().*;
         this.reader.buffer().* = std.ArrayList(u8).init(bun.default_allocator);
         return bun.ByteList.fromList(out);
     }

src/io/PipeReader.zig

@@ -339,7 +339,7 @@ pub fn WindowsPipeReader(
             var this = bun.cast(*This, stream.data);
 
             const nread_int = nread.int();
-            bun.sys.syslog("onStreamRead() = {d}", .{nread_int});
+            bun.sys.syslog("onStreamRead(0x{d}) = {d}", .{ @intFromPtr(this), nread_int });
 
             // NOTE: pipes/tty need to call stopReading on errors (yeah)
             switch (nread_int) {

src/js/builtins/ReadableStreamInternals.ts

@@ -1656,7 +1656,7 @@ export function lazyLoadStream(stream, autoAllocateChunkSize) {
       $assert(controller, "controller is missing");
 
       if (result && $isPromise(result)) {
-        return result.then(
+        return result.$then(
           handleNativeReadableStreamPromiseResult.bind({
             c: controller,
             v: view,

test/js/bun/spawn/spawn.test.ts

@@ -490,6 +490,20 @@ for (let [gcTick, label] of [
             });
           });
         }
+
+        it("should allow reading stdout after a few milliseconds", async () => {
+          for (let i = 0; i < 50; i++) {
+            const proc = Bun.spawn({
+              cmd: ["git", "--version"],
+              stdout: "pipe",
+              stderr: "ignore",
+              stdin: "ignore",
+            });
+            await Bun.sleep(1);
+            const out = await Bun.readableStreamToText(proc.stdout);
+            expect(out).not.toBe("");
+          }
+        });
       });
 
       it("throws errors for invalid arguments", async () => {