Fix several bugs with Request body streaming + store small response bodies without an extra memory allocation

10 files changed, 94 insertions, 12 deletions

diff --git a/src/bun.js/bindings/ZigGlobalObject.cpp b/src/bun.js/bindings/ZigGlobalObject.cpp
index 33828f7ee..f14352230 100644
--- a/src/bun.js/bindings/ZigGlobalObject.cpp
+++ b/src/bun.js/bindings/ZigGlobalObject.cpp
@@ -1,4 +1,3 @@
-
 #include "root.h"
 #include "ZigGlobalObject.h"
 
@@ -115,7 +114,8 @@
 #include "JavaScriptCore/RemoteInspectorServer.h"
 #endif
 
-using JSGlobalObject = JSC::JSGlobalObject;
+using JSGlobalObject
+    = JSC::JSGlobalObject;
 using Exception = JSC::Exception;
 using JSValue = JSC::JSValue;
 using JSString = JSC::JSString;
@@ -979,6 +979,30 @@ JSC_DEFINE_CUSTOM_SETTER(noop_setter,
 static NeverDestroyed<const String> pathToFileURLString(MAKE_STATIC_STRING_IMPL("pathToFileURL"));
 static NeverDestroyed<const String> fileURLToPathString(MAKE_STATIC_STRING_IMPL("fileURLToPath"));
 
+enum ReadableStreamTag : int32_t {
+    Invalid = -1,
+
+    /// ReadableStreamDefaultController or ReadableByteStreamController
+    JavaScript = 0,
+
+    /// ReadableByteStreamController
+    /// but with a BlobLoader
+    /// we can skip the BlobLoader and just use the underlying Blob
+    Blob = 1,
+
+    /// ReadableByteStreamController
+    /// but with a FileLoader
+    /// we can skip the FileLoader and just use the underlying File
+    File = 2,
+
+    /// This is a direct readable stream
+    /// That means we can turn it into whatever we want
+    Direct = 3,
+
+    // This is an ambiguous stream of bytes
+    Bytes = 4,
+};
+
 // we're trying out a new way to do this lazy loading
 static JSC_DECLARE_HOST_FUNCTION(functionLazyLoad);
 static JSC_DEFINE_HOST_FUNCTION(functionLazyLoad,
@@ -1010,12 +1034,16 @@ JSC:
                 return JSC::JSValue::encode(JSC::JSValue {});
             }
 
-            case 1: {
+            case ReadableStreamTag::Blob: {
                 return ByteBlob__JSReadableStreamSource__load(globalObject);
             }
-            case 2: {
+            case ReadableStreamTag::File: {
                 return FileBlobLoader__JSReadableStreamSource__load(globalObject);
             }
+            case ReadableStreamTag::Bytes: {
+                return ByteStream__JSReadableStreamSource__load(globalObject);
+            }
+
             default: {
                 auto scope = DECLARE_THROW_SCOPE(globalObject->vm());
                 JSC::throwTypeError(globalObject, scope, "lazyLoad expects a string"_s);
@@ -1352,6 +1380,7 @@ extern "C" void ReadableStream__cancel(JSC__JSValue possibleReadableStream, Zig:
     WebCore::Exception exception { AbortError };
     ReadableStream(*globalObject, *readableStream).cancel(exception);
 }
+
 extern "C" void ReadableStream__detach(JSC__JSValue possibleReadableStream, Zig::GlobalObject* globalObject);
 extern "C" void ReadableStream__detach(JSC__JSValue possibleReadableStream, Zig::GlobalObject* globalObject)
 {
diff --git a/src/bun.js/bindings/bindings.cpp b/src/bun.js/bindings/bindings.cpp
index 4b508591a..41337fd44 100644
--- a/src/bun.js/bindings/bindings.cpp
+++ b/src/bun.js/bindings/bindings.cpp
@@ -265,7 +265,7 @@ WebCore::FetchHeaders* WebCore__FetchHeaders__createFromUWS(JSC__JSGlobalObject*
         uint32_t hash = nameView.hash();
         nameHashes[i++] = hash;
         size_t name_len = nameView.length();
-        auto value = WTF::StringView(reinterpret_cast<const LChar*>(header.second.data()), header.second.length()).toStringWithoutCopying().isolatedCopy();
+        auto value = WTF::StringView(reinterpret_cast<const LChar*>(header.second.data()), header.second.length()).toString();
 
         if (name_len < 255) {
             if (seenHeaderSizes[name_len]) {
@@ -642,6 +642,17 @@ void JSC__JSGlobalObject__deleteModuleRegistryEntry(JSC__JSGlobalObject* global,
 
     map->remove(global, val);
 }
+
+void JSC__VM__collectAsync(JSC__VM* vm)
+{
+    vm->heap.collectAsync();
+}
+
+size_t JSC__VM__heapSize(JSC__VM* arg0)
+{
+    return arg0->heap.size();
+}
+
 // This is very naive!
 JSC__JSInternalPromise* JSC__VM__reloadModule(JSC__VM* vm, JSC__JSGlobalObject* arg1,
     ZigString arg2)
diff --git a/src/bun.js/bindings/bindings.zig b/src/bun.js/bindings/bindings.zig
index 0386fd991..54c574562 100644
--- a/src/bun.js/bindings/bindings.zig
+++ b/src/bun.js/bindings/bindings.zig
@@ -118,6 +118,18 @@ pub const ZigString = extern struct {
         return out;
     }
 
+    pub fn maxUTF8ByteLength(this: ZigString) usize {
+        if (this.isUTF8())
+            return this.len;
+
+        if (this.is16Bit()) {
+            return this.utf16SliceAligned().len * 3;
+        }
+
+        // latin1
+        return this.len * 2;
+    }
+
     pub fn utf8ByteLength(this: ZigString) usize {
         if (this.isUTF8()) {
             return this.len;
@@ -3581,6 +3593,18 @@ pub const VM = extern struct {
         });
     }
 
+    pub fn heapSize(vm: *VM) usize {
+        return cppFn("heapSize", .{
+            vm,
+        });
+    }
+
+    pub fn collectAsync(vm: *VM) void {
+        return cppFn("collectAsync", .{
+            vm,
+        });
+    }
+
     pub fn setExecutionForbidden(vm: *VM, forbidden: bool) void {
         cppFn("setExecutionForbidden", .{ vm, forbidden });
     }
@@ -3632,7 +3656,7 @@ pub const VM = extern struct {
             vm,
         });
     }
-    pub const Extern = [_][]const u8{ "releaseWeakRefs", "throwError", "doWork", "deferGC", "holdAPILock", "runGC", "generateHeapSnapshot", "isJITEnabled", "deleteAllCode", "create", "deinit", "setExecutionForbidden", "executionForbidden", "isEntered", "throwError", "drainMicrotasks", "whenIdle", "shrinkFootprint", "setExecutionTimeLimit", "clearExecutionTimeLimit" };
+    pub const Extern = [_][]const u8{ "collectAsync", "heapSize", "releaseWeakRefs", "throwError", "doWork", "deferGC", "holdAPILock", "runGC", "generateHeapSnapshot", "isJITEnabled", "deleteAllCode", "create", "deinit", "setExecutionForbidden", "executionForbidden", "isEntered", "throwError", "drainMicrotasks", "whenIdle", "shrinkFootprint", "setExecutionTimeLimit", "clearExecutionTimeLimit" };
 };
 
 pub const ThrowScope = extern struct {
diff --git a/src/bun.js/bindings/exports.zig b/src/bun.js/bindings/exports.zig
index c5ab610b2..1d9eddf49 100644
--- a/src/bun.js/bindings/exports.zig
+++ b/src/bun.js/bindings/exports.zig
@@ -2783,6 +2783,7 @@ comptime {
         JSHTTPSResponseSink.shim.ref();
         JSFileSink.shim.ref();
 
+        JSReadableStreamBytes.shim.ref();
         JSReadableStreamFile.shim.ref();
         _ = ZigString__free;
         _ = ZigString__free_global;
diff --git a/src/bun.js/bindings/headers-cpp.h b/src/bun.js/bindings/headers-cpp.h
index 5f97769c8..73446d12e 100644
--- a/src/bun.js/bindings/headers-cpp.h
+++ b/src/bun.js/bindings/headers-cpp.h
@@ -1,4 +1,4 @@
-//-- AUTOGENERATED FILE -- 1663581060
+//-- AUTOGENERATED FILE -- 1664343178
 // clang-format off
 #pragma once
 
diff --git a/src/bun.js/bindings/headers.h b/src/bun.js/bindings/headers.h
index 49b26d394..91652804a 100644
--- a/src/bun.js/bindings/headers.h
+++ b/src/bun.js/bindings/headers.h
@@ -1,5 +1,5 @@
 // clang-format off
-//-- AUTOGENERATED FILE -- 1663581060
+//-- AUTOGENERATED FILE -- 1664343178
 #pragma once
 
 #include <stddef.h>
@@ -567,6 +567,7 @@ CPP_DECL JSC__JSValue JSC__Exception__value(JSC__Exception* arg0);
 #pragma mark - JSC::VM
 
 CPP_DECL void JSC__VM__clearExecutionTimeLimit(JSC__VM* arg0);
+CPP_DECL void JSC__VM__collectAsync(JSC__VM* arg0);
 CPP_DECL JSC__VM* JSC__VM__create(unsigned char HeapType0);
 CPP_DECL void JSC__VM__deferGC(JSC__VM* arg0, void* arg1, void (* ArgFn2)(void* arg0));
 CPP_DECL void JSC__VM__deinit(JSC__VM* arg0, JSC__JSGlobalObject* arg1);
@@ -574,6 +575,7 @@ CPP_DECL void JSC__VM__deleteAllCode(JSC__VM* arg0, JSC__JSGlobalObject* arg1);
 CPP_DECL void JSC__VM__doWork(JSC__VM* arg0);
 CPP_DECL void JSC__VM__drainMicrotasks(JSC__VM* arg0);
 CPP_DECL bool JSC__VM__executionForbidden(JSC__VM* arg0);
+CPP_DECL size_t JSC__VM__heapSize(JSC__VM* arg0);
 CPP_DECL void JSC__VM__holdAPILock(JSC__VM* arg0, void* arg1, void (* ArgFn2)(void* arg0));
 CPP_DECL bool JSC__VM__isEntered(JSC__VM* arg0);
 CPP_DECL bool JSC__VM__isJITEnabled();
diff --git a/src/bun.js/bindings/headers.zig b/src/bun.js/bindings/headers.zig
index 606ae64a4..ef6257499 100644
--- a/src/bun.js/bindings/headers.zig
+++ b/src/bun.js/bindings/headers.zig
@@ -350,6 +350,7 @@ pub extern fn JSC__Exception__create(arg0: ?*JSC__JSGlobalObject, arg1: [*c]JSC_
 pub extern fn JSC__Exception__getStackTrace(arg0: [*c]JSC__Exception, arg1: [*c]ZigStackTrace) void;
 pub extern fn JSC__Exception__value(arg0: [*c]JSC__Exception) JSC__JSValue;
 pub extern fn JSC__VM__clearExecutionTimeLimit(arg0: [*c]JSC__VM) void;
+pub extern fn JSC__VM__collectAsync(arg0: [*c]JSC__VM) void;
 pub extern fn JSC__VM__create(HeapType0: u8) [*c]JSC__VM;
 pub extern fn JSC__VM__deferGC(arg0: [*c]JSC__VM, arg1: ?*anyopaque, ArgFn2: ?fn (?*anyopaque) callconv(.C) void) void;
 pub extern fn JSC__VM__deinit(arg0: [*c]JSC__VM, arg1: ?*JSC__JSGlobalObject) void;
@@ -357,6 +358,7 @@ pub extern fn JSC__VM__deleteAllCode(arg0: [*c]JSC__VM, arg1: ?*JSC__JSGlobalObj
 pub extern fn JSC__VM__doWork(arg0: [*c]JSC__VM) void;
 pub extern fn JSC__VM__drainMicrotasks(arg0: [*c]JSC__VM) void;
 pub extern fn JSC__VM__executionForbidden(arg0: [*c]JSC__VM) bool;
+pub extern fn JSC__VM__heapSize(arg0: [*c]JSC__VM) usize;
 pub extern fn JSC__VM__holdAPILock(arg0: [*c]JSC__VM, arg1: ?*anyopaque, ArgFn2: ?fn (?*anyopaque) callconv(.C) void) void;
 pub extern fn JSC__VM__isEntered(arg0: [*c]JSC__VM) bool;
 pub extern fn JSC__VM__isJITEnabled(...) bool;
diff --git a/src/bun.js/scripts/class-definitions.ts b/src/bun.js/scripts/class-definitions.ts
index 67e4cf475..f8b08e9bd 100644
--- a/src/bun.js/scripts/class-definitions.ts
+++ b/src/bun.js/scripts/class-definitions.ts
@@ -20,13 +20,23 @@ export interface ClassDefinition {
   proto: Record<string, Field>;
   JSType?: string;
   noConstructor?: boolean;
+  estimatedSize?: boolean;
+  isEventEmitter?: boolean;
 }
 
 export function define(
-  { klass = {}, proto = {}, ...rest } = {} as ClassDefinition
+  {
+    klass = {},
+    proto = {},
+    isEventEmitter = false,
+    estimatedSize = false,
+    ...rest
+  } = {} as ClassDefinition
 ): ClassDefinition {
   return {
     ...rest,
+    isEventEmitter,
+    estimatedSize,
     klass: Object.fromEntries(
       Object.entries(klass).sort(([a], [b]) => a.localeCompare(b))
     ),
diff --git a/src/bun.js/webcore/response.classes.ts b/src/bun.js/webcore/response.classes.ts
index 84bad8f24..89133979a 100644
--- a/src/bun.js/webcore/response.classes.ts
+++ b/src/bun.js/webcore/response.classes.ts
@@ -7,6 +7,7 @@ export default [
     finalize: true,
     klass: {},
     JSType: "0b11101110",
+    estimatedSize: true,
     proto: {
       text: { fn: "getText" },
       json: { fn: "getJSON" },
@@ -59,6 +60,7 @@ export default [
     construct: true,
     finalize: true,
     JSType: "0b11101110",
+    estimatedSize: true,
     klass: {
       json: {
         fn: "constructJSON",
diff --git a/src/bun.js/webcore/streams.zig b/src/bun.js/webcore/streams.zig
index f5d2be92e..2695a43e2 100644
--- a/src/bun.js/webcore/streams.zig
+++ b/src/bun.js/webcore/streams.zig
@@ -7,7 +7,7 @@ const ZigURL = @import("../../url.zig").URL;
 const HTTPClient = @import("http");
 const NetworkThread = HTTPClient.NetworkThread;
 const AsyncIO = NetworkThread.AsyncIO;
-const JSC = @import("javascript_core");
+const JSC = @import("root").JSC;
 const js = JSC.C;
 
 const Method = @import("../../http/method.zig").Method;
@@ -2796,7 +2796,7 @@ pub const ByteStream = struct {
 
         const chunk = stream.slice();
 
-        if (this.pending.state != .pending) {
+        if (this.pending.state == .pending) {
             std.debug.assert(this.buffer.items.len == 0);
             var to_copy = this.pending_buffer[0..@minimum(chunk.len, this.pending_buffer.len)];
             const pending_buffer_len = this.pending_buffer.len;
@@ -2825,7 +2825,8 @@ pub const ByteStream = struct {
             const remaining = chunk[to_copy.len..];
             if (remaining.len > 0)
                 this.append(stream, to_copy.len, allocator) catch @panic("Out of memory while copying request body");
-            resume this.pending.frame;
+
+            this.pending.run();
             return;
         }