fix(fs): WriteStream pending write fastpath

src/io/PipeWriter.zig

@@ -41,35 +41,35 @@ pub fn PosixPipeWriter(
             };
         }
 
-        fn _tryWriteWithWriteFn(this: *This, buf_: []const u8, comptime write_fn: *const fn (bun.FileDescriptor, []const u8) JSC.Maybe(usize)) WriteResult {
+        fn _tryWriteWithWriteFn(this: *This, buf: []const u8, comptime write_fn: *const fn (bun.FileDescriptor, []const u8) JSC.Maybe(usize)) WriteResult {
             const fd = getFd(this);
-            var buf = buf_;
+            var rest = buf;
 
-            while (buf.len > 0) {
-                switch (write_fn(fd, buf)) {
+            while (rest.len > 0) {
+                switch (write_fn(fd, rest)) {
                     .err => |err| {
                         if (err.isRetry()) {
-                            return .{ .pending = buf_.len - buf.len };
+                            return .{ .pending = buf.len - rest.len };
                         }
 
                         if (err.getErrno() == .PIPE) {
-                            return .{ .done = buf_.len - buf.len };
+                            return .{ .done = buf.len - rest.len };
                         }
 
                         return .{ .err = err };
                     },
 
                     .result => |wrote| {
                         if (wrote == 0) {
-                            return .{ .done = buf_.len - buf.len };
+                            return .{ .done = buf.len - rest.len };
                         }
 
-                        buf = buf[wrote..];
+                        rest = rest[wrote..];
                     },
                 }
             }
 
-            return .{ .wrote = buf_.len - buf.len };
+            return .{ .wrote = buf.len - rest.len };
         }
 
         fn writeToFileType(comptime file_type: FileType) *const (fn (bun.FileDescriptor, []const u8) JSC.Maybe(usize)) {
@@ -394,8 +394,8 @@ pub fn PosixStreamingWriter(
         is_done: bool = false,
         closed_without_reporting: bool = false,
 
-        // TODO:
-        chunk_size: usize = 0,
+        // TODO: configurable?
+        const chunk_size: usize = 1024 * 4;
 
         pub fn memoryCost(this: *const @This()) usize {
             return @sizeOf(@This()) + this.buffer.capacity;
@@ -586,24 +586,51 @@ pub fn PosixStreamingWriter(
                 return .{ .done = 0 };
             }
 
-            if (this.buffer.items.len + buf.len < this.chunk_size) {
+            if (this.buffer.items.len > 0 or buf.len < chunk_size) {
                 this.buffer.appendSlice(buf) catch {
                     return .{ .err = bun.sys.Error.oom };
                 };
+                return .{ .pending = buf.len };
+            }
 
-                return .{ .pending = 0 };
+            this.head = 0;
+
+            if (this.buffer.items.len > 0) {
+                this.buffer.appendSlice(buf) catch {
+                    return .{ .err = bun.sys.Error.oom };
+                };
+                const rc = @This()._tryWrite(this, this.buffer.items);
+                switch (rc) {
+                    .pending => |amt| {
+                        onWrite(this.parent, amt, .pending);
+                        registerPoll(this);
+                    },
+                    .wrote => |amt| {
+                        if (amt < this.buffer.items.len) {
+                            onWrite(this.parent, amt, .pending);
+                        } else {
+                            this.buffer.clearRetainingCapacity();
+                            onWrite(this.parent, amt, .drained);
+                        }
+                    },
+                    .done => |amt| {
+                        this.buffer.clearRetainingCapacity();
+                        onWrite(this.parent, amt, .end_of_file);
+                        return .{ .done = amt };
+                    },
+                    else => {},
+                }
+                return rc;
             }
 
             const rc = @This()._tryWrite(this, buf);
-            this.head = 0;
+
             switch (rc) {
                 .pending => |amt| {
                     this.buffer.appendSlice(buf[amt..]) catch {
                         return .{ .err = bun.sys.Error.oom };
                     };
-
                     onWrite(this.parent, amt, .pending);
-
                     registerPoll(this);
                 },
                 .wrote => |amt| {