Introduce `FileSink.ref()` and `FileSink.unref()`

10 files changed, 382 insertions, 72 deletions

diff --git a/packages/bun-types/bun.d.ts b/packages/bun-types/bun.d.ts
index a0d9f8f15..ccdcf2c2c 100644
--- a/packages/bun-types/bun.d.ts
+++ b/packages/bun-types/bun.d.ts
@@ -411,6 +411,43 @@ declare module "bun" {
      * Close the file descriptor. This also flushes the internal buffer.
      */
     end(error?: Error): number | Promise<number>;
+
+    start(options?: {
+      /**
+       * Preallocate an internal buffer of this size
+       * This can significantly improve performance when the chunk size is small
+       */
+      highWaterMark?: number;
+    }): void;
+
+    /**
+     * For FIFOs & pipes, this lets you decide whether Bun's process should
+     * remain alive until the pipe is closed.
+     *
+     * By default, it is automatically managed. While the stream is open, the
+     * process remains alive and once the other end hangs up or the stream
+     * closes, the process exits.
+     *
+     * If you previously called {@link unref}, you can call this again to re-enable automatic management.
+     *
+     * Internally, it will reference count the number of times this is called. By default, that number is 1
+     *
+     * If the file is not a FIFO or pipe, {@link ref} and {@link unref} do
+     * nothing. If the pipe is already closed, this does nothing.
+     */
+    ref(): void;
+
+    /**
+     * For FIFOs & pipes, this lets you decide whether Bun's process should
+     * remain alive until the pipe is closed.
+     *
+     * If you want to allow Bun's process to terminate while the stream is open,
+     * call this.
+     *
+     * If the file is not a FIFO or pipe, {@link ref} and {@link unref} do
+     * nothing. If the pipe is already closed, this does nothing.
+     */
+    unref(): void;
   }
 
   /**
diff --git a/src/bun.js/bindings/JSSink.cpp b/src/bun.js/bindings/JSSink.cpp
index ed3b6a5ac..a93e4e298 100644
--- a/src/bun.js/bindings/JSSink.cpp
+++ b/src/bun.js/bindings/JSSink.cpp
@@ -1,6 +1,6 @@
 
 // AUTO-GENERATED FILE. DO NOT EDIT.
-// Generated by 'make generate-sink' at 2022-11-20T12:51:04.404Z
+// Generated by 'make generate-sink' at 2022-11-25T07:36:18.561Z
 // To regenerate this file, run:
 //
 //   make generate-sink
@@ -127,6 +127,48 @@ JSC_DEFINE_HOST_FUNCTION(functionStartDirectStream, (JSC::JSGlobalObject * lexic
     RELEASE_AND_RETURN(scope, JSC::JSValue::encode(JSC::jsUndefined()));
 }
 
+void JSArrayBufferSink::ref()
+{
+    if (!m_sinkPtr)
+        return;
+
+    m_refCount++;
+    if (m_refCount == 1) {
+        ArrayBufferSink__updateRef(m_sinkPtr, true);
+    }
+}
+
+void JSArrayBufferSink::unref()
+{
+    if (!m_sinkPtr)
+        return;
+
+    m_refCount = std::max(0, m_refCount - 1);
+    if (!m_refCount) {
+        ArrayBufferSink__updateRef(m_sinkPtr, false);
+    }
+}
+
+JSC_DEFINE_HOST_FUNCTION(ArrayBufferSink__ref, (JSC::JSGlobalObject * lexicalGlobalObject, JSC::CallFrame* callFrame))
+{
+    auto& vm = lexicalGlobalObject->vm();
+    auto* sink = jsDynamicCast<WebCore::JSArrayBufferSink*>(callFrame->thisValue());
+    if (LIKELY(sink)) {
+        sink->ref();
+    }
+    return JSC::JSValue::encode(JSC::jsUndefined());
+}
+
+JSC_DEFINE_HOST_FUNCTION(ArrayBufferSink__unref, (JSC::JSGlobalObject * lexicalGlobalObject, JSC::CallFrame* callFrame))
+{
+    auto& vm = lexicalGlobalObject->vm();
+    auto* sink = jsDynamicCast<WebCore::JSArrayBufferSink*>(callFrame->thisValue());
+    if (LIKELY(sink)) {
+        sink->unref();
+    }
+    return JSC::JSValue::encode(JSC::jsUndefined());
+}
+
 JSC_DEFINE_CUSTOM_GETTER(functionArrayBufferSink__getter, (JSC::JSGlobalObject * lexicalGlobalObject, JSC::EncodedJSValue thisValue, JSC::PropertyName))
 {
     auto& vm = lexicalGlobalObject->vm();
@@ -203,6 +245,48 @@ JSC_DEFINE_HOST_FUNCTION(ArrayBufferSink__doClose, (JSC::JSGlobalObject * lexica
     return JSC::JSValue::encode(JSC::jsUndefined());
 }
 
+void JSFileSink::ref()
+{
+    if (!m_sinkPtr)
+        return;
+
+    m_refCount++;
+    if (m_refCount == 1) {
+        FileSink__updateRef(m_sinkPtr, true);
+    }
+}
+
+void JSFileSink::unref()
+{
+    if (!m_sinkPtr)
+        return;
+
+    m_refCount = std::max(0, m_refCount - 1);
+    if (!m_refCount) {
+        FileSink__updateRef(m_sinkPtr, false);
+    }
+}
+
+JSC_DEFINE_HOST_FUNCTION(FileSink__ref, (JSC::JSGlobalObject * lexicalGlobalObject, JSC::CallFrame* callFrame))
+{
+    auto& vm = lexicalGlobalObject->vm();
+    auto* sink = jsDynamicCast<WebCore::JSFileSink*>(callFrame->thisValue());
+    if (LIKELY(sink)) {
+        sink->ref();
+    }
+    return JSC::JSValue::encode(JSC::jsUndefined());
+}
+
+JSC_DEFINE_HOST_FUNCTION(FileSink__unref, (JSC::JSGlobalObject * lexicalGlobalObject, JSC::CallFrame* callFrame))
+{
+    auto& vm = lexicalGlobalObject->vm();
+    auto* sink = jsDynamicCast<WebCore::JSFileSink*>(callFrame->thisValue());
+    if (LIKELY(sink)) {
+        sink->unref();
+    }
+    return JSC::JSValue::encode(JSC::jsUndefined());
+}
+
 JSC_DEFINE_CUSTOM_GETTER(functionFileSink__getter, (JSC::JSGlobalObject * lexicalGlobalObject, JSC::EncodedJSValue thisValue, JSC::PropertyName))
 {
     auto& vm = lexicalGlobalObject->vm();
@@ -279,6 +363,48 @@ JSC_DEFINE_HOST_FUNCTION(FileSink__doClose, (JSC::JSGlobalObject * lexicalGlobal
     return JSC::JSValue::encode(JSC::jsUndefined());
 }
 
+void JSHTTPResponseSink::ref()
+{
+    if (!m_sinkPtr)
+        return;
+
+    m_refCount++;
+    if (m_refCount == 1) {
+        HTTPResponseSink__updateRef(m_sinkPtr, true);
+    }
+}
+
+void JSHTTPResponseSink::unref()
+{
+    if (!m_sinkPtr)
+        return;
+
+    m_refCount = std::max(0, m_refCount - 1);
+    if (!m_refCount) {
+        HTTPResponseSink__updateRef(m_sinkPtr, false);
+    }
+}
+
+JSC_DEFINE_HOST_FUNCTION(HTTPResponseSink__ref, (JSC::JSGlobalObject * lexicalGlobalObject, JSC::CallFrame* callFrame))
+{
+    auto& vm = lexicalGlobalObject->vm();
+    auto* sink = jsDynamicCast<WebCore::JSHTTPResponseSink*>(callFrame->thisValue());
+    if (LIKELY(sink)) {
+        sink->ref();
+    }
+    return JSC::JSValue::encode(JSC::jsUndefined());
+}
+
+JSC_DEFINE_HOST_FUNCTION(HTTPResponseSink__unref, (JSC::JSGlobalObject * lexicalGlobalObject, JSC::CallFrame* callFrame))
+{
+    auto& vm = lexicalGlobalObject->vm();
+    auto* sink = jsDynamicCast<WebCore::JSHTTPResponseSink*>(callFrame->thisValue());
+    if (LIKELY(sink)) {
+        sink->unref();
+    }
+    return JSC::JSValue::encode(JSC::jsUndefined());
+}
+
 JSC_DEFINE_CUSTOM_GETTER(functionHTTPResponseSink__getter, (JSC::JSGlobalObject * lexicalGlobalObject, JSC::EncodedJSValue thisValue, JSC::PropertyName))
 {
     auto& vm = lexicalGlobalObject->vm();
@@ -355,6 +481,48 @@ JSC_DEFINE_HOST_FUNCTION(HTTPResponseSink__doClose, (JSC::JSGlobalObject * lexic
     return JSC::JSValue::encode(JSC::jsUndefined());
 }
 
+void JSHTTPSResponseSink::ref()
+{
+    if (!m_sinkPtr)
+        return;
+
+    m_refCount++;
+    if (m_refCount == 1) {
+        HTTPSResponseSink__updateRef(m_sinkPtr, true);
+    }
+}
+
+void JSHTTPSResponseSink::unref()
+{
+    if (!m_sinkPtr)
+        return;
+
+    m_refCount = std::max(0, m_refCount - 1);
+    if (!m_refCount) {
+        HTTPSResponseSink__updateRef(m_sinkPtr, false);
+    }
+}
+
+JSC_DEFINE_HOST_FUNCTION(HTTPSResponseSink__ref, (JSC::JSGlobalObject * lexicalGlobalObject, JSC::CallFrame* callFrame))
+{
+    auto& vm = lexicalGlobalObject->vm();
+    auto* sink = jsDynamicCast<WebCore::JSHTTPSResponseSink*>(callFrame->thisValue());
+    if (LIKELY(sink)) {
+        sink->ref();
+    }
+    return JSC::JSValue::encode(JSC::jsUndefined());
+}
+
+JSC_DEFINE_HOST_FUNCTION(HTTPSResponseSink__unref, (JSC::JSGlobalObject * lexicalGlobalObject, JSC::CallFrame* callFrame))
+{
+    auto& vm = lexicalGlobalObject->vm();
+    auto* sink = jsDynamicCast<WebCore::JSHTTPSResponseSink*>(callFrame->thisValue());
+    if (LIKELY(sink)) {
+        sink->unref();
+    }
+    return JSC::JSValue::encode(JSC::jsUndefined());
+}
+
 JSC_DEFINE_CUSTOM_GETTER(functionHTTPSResponseSink__getter, (JSC::JSGlobalObject * lexicalGlobalObject, JSC::EncodedJSValue thisValue, JSC::PropertyName))
 {
     auto& vm = lexicalGlobalObject->vm();
@@ -440,6 +608,8 @@ JSC_DEFINE_HOST_FUNCTION(HTTPSResponseSink__doClose, (JSC::JSGlobalObject * lexi
   end        ArrayBufferSink__end             ReadOnly|DontDelete|Function 0
   start      ArrayBufferSink__start           ReadOnly|DontDelete|Function 1
   write      ArrayBufferSink__write           ReadOnly|DontDelete|Function 1
+  ref        ArrayBufferSink__ref             ReadOnly|DontDelete|Function 0
+  unref      ArrayBufferSink__unref           ReadOnly|DontDelete|Function 0
 @end
 */
 
@@ -460,6 +630,8 @@ JSC_DEFINE_HOST_FUNCTION(HTTPSResponseSink__doClose, (JSC::JSGlobalObject * lexi
   end        FileSink__end             ReadOnly|DontDelete|Function 0
   start      FileSink__start           ReadOnly|DontDelete|Function 1
   write      FileSink__write           ReadOnly|DontDelete|Function 1
+  ref        FileSink__ref             ReadOnly|DontDelete|Function 0
+  unref      FileSink__unref           ReadOnly|DontDelete|Function 0
 @end
 */
 
@@ -480,6 +652,8 @@ JSC_DEFINE_HOST_FUNCTION(HTTPSResponseSink__doClose, (JSC::JSGlobalObject * lexi
   end        HTTPResponseSink__end             ReadOnly|DontDelete|Function 0
   start      HTTPResponseSink__start           ReadOnly|DontDelete|Function 1
   write      HTTPResponseSink__write           ReadOnly|DontDelete|Function 1
+  ref        HTTPResponseSink__ref             ReadOnly|DontDelete|Function 0
+  unref      HTTPResponseSink__unref           ReadOnly|DontDelete|Function 0
 @end
 */
 
@@ -500,6 +674,8 @@ JSC_DEFINE_HOST_FUNCTION(HTTPSResponseSink__doClose, (JSC::JSGlobalObject * lexi
   end        HTTPSResponseSink__end             ReadOnly|DontDelete|Function 0
   start      HTTPSResponseSink__start           ReadOnly|DontDelete|Function 1
   write      HTTPSResponseSink__write           ReadOnly|DontDelete|Function 1
+  ref        HTTPSResponseSink__ref             ReadOnly|DontDelete|Function 0
+  unref      HTTPSResponseSink__unref           ReadOnly|DontDelete|Function 0
 @end
 */
 
@@ -1603,8 +1779,7 @@ extern "C" JSC__JSValue ArrayBufferSink__createObject(JSC__JSGlobalObject* arg0,
 {
     auto& vm = arg0->vm();
     Zig::GlobalObject* globalObject = reinterpret_cast<Zig::GlobalObject*>(arg0);
-    JSC::JSValue prototype = globalObject->ArrayBufferSinkPrototype();
-    JSC::Structure* structure = WebCore::JSArrayBufferSink::createStructure(vm, globalObject, prototype);
+    JSC::Structure* structure = globalObject->ArrayBufferSinkStructure();
     return JSC::JSValue::encode(WebCore::JSArrayBufferSink::create(vm, globalObject, structure, sinkPtr));
 }
 
@@ -1690,8 +1865,7 @@ extern "C" JSC__JSValue FileSink__createObject(JSC__JSGlobalObject* arg0, void*
 {
     auto& vm = arg0->vm();
     Zig::GlobalObject* globalObject = reinterpret_cast<Zig::GlobalObject*>(arg0);
-    JSC::JSValue prototype = globalObject->FileSinkPrototype();
-    JSC::Structure* structure = WebCore::JSFileSink::createStructure(vm, globalObject, prototype);
+    JSC::Structure* structure = globalObject->FileSinkStructure();
     return JSC::JSValue::encode(WebCore::JSFileSink::create(vm, globalObject, structure, sinkPtr));
 }
 
@@ -1777,8 +1951,7 @@ extern "C" JSC__JSValue HTTPResponseSink__createObject(JSC__JSGlobalObject* arg0
 {
     auto& vm = arg0->vm();
     Zig::GlobalObject* globalObject = reinterpret_cast<Zig::GlobalObject*>(arg0);
-    JSC::JSValue prototype = globalObject->HTTPResponseSinkPrototype();
-    JSC::Structure* structure = WebCore::JSHTTPResponseSink::createStructure(vm, globalObject, prototype);
+    JSC::Structure* structure = globalObject->HTTPResponseSinkStructure();
     return JSC::JSValue::encode(WebCore::JSHTTPResponseSink::create(vm, globalObject, structure, sinkPtr));
 }
 
@@ -1864,8 +2037,7 @@ extern "C" JSC__JSValue HTTPSResponseSink__createObject(JSC__JSGlobalObject* arg
 {
     auto& vm = arg0->vm();
     Zig::GlobalObject* globalObject = reinterpret_cast<Zig::GlobalObject*>(arg0);
-    JSC::JSValue prototype = globalObject->HTTPSResponseSinkPrototype();
-    JSC::Structure* structure = WebCore::JSHTTPSResponseSink::createStructure(vm, globalObject, prototype);
+    JSC::Structure* structure = globalObject->HTTPSResponseSinkStructure();
     return JSC::JSValue::encode(WebCore::JSHTTPSResponseSink::create(vm, globalObject, structure, sinkPtr));
 }
 
diff --git a/src/bun.js/bindings/JSSink.h b/src/bun.js/bindings/JSSink.h
index 68caf12fc..3bdfaec9a 100644
--- a/src/bun.js/bindings/JSSink.h
+++ b/src/bun.js/bindings/JSSink.h
@@ -1,6 +1,6 @@
 
 // AUTO-GENERATED FILE. DO NOT EDIT.
-// Generated by 'make generate-sink' at 2022-11-20T12:51:04.403Z
+// Generated by 'make generate-sink' at 2022-11-25T07:36:18.559Z
 //
 #pragma once
 
@@ -97,7 +97,11 @@ public:
 
     static void analyzeHeap(JSCell*, JSC::HeapAnalyzer&);
 
+    void ref();
+    void unref();
+
     void* m_sinkPtr;
+    int m_refCount { 1 };
 
     JSArrayBufferSink(JSC::VM& vm, JSC::Structure* structure, void* sinkPtr)
         : Base(vm, structure)
@@ -240,7 +244,11 @@ public:
 
     static void analyzeHeap(JSCell*, JSC::HeapAnalyzer&);
 
+    void ref();
+    void unref();
+
     void* m_sinkPtr;
+    int m_refCount { 1 };
 
     JSFileSink(JSC::VM& vm, JSC::Structure* structure, void* sinkPtr)
         : Base(vm, structure)
@@ -383,7 +391,11 @@ public:
 
     static void analyzeHeap(JSCell*, JSC::HeapAnalyzer&);
 
+    void ref();
+    void unref();
+
     void* m_sinkPtr;
+    int m_refCount { 1 };
 
     JSHTTPResponseSink(JSC::VM& vm, JSC::Structure* structure, void* sinkPtr)
         : Base(vm, structure)
@@ -526,7 +538,11 @@ public:
 
     static void analyzeHeap(JSCell*, JSC::HeapAnalyzer&);
 
+    void ref();
+    void unref();
+
     void* m_sinkPtr;
+    int m_refCount { 1 };
 
     JSHTTPSResponseSink(JSC::VM& vm, JSC::Structure* structure, void* sinkPtr)
         : Base(vm, structure)
diff --git a/src/bun.js/bindings/JSSinkLookupTable.h b/src/bun.js/bindings/JSSinkLookupTable.h
index 14d547708..a4ace6dc3 100644
--- a/src/bun.js/bindings/JSSinkLookupTable.h
+++ b/src/bun.js/bindings/JSSinkLookupTable.h
@@ -10,7 +10,7 @@ static const struct CompactHashIndex JSArrayBufferSinkPrototypeTableIndex[19] =
     { -1, -1 },
     { -1, -1 },
     { -1, -1 },
-    { -1, -1 },
+    { 6, -1 },
     { -1, -1 },
     { -1, -1 },
     { -1, -1 },
@@ -19,23 +19,25 @@ static const struct CompactHashIndex JSArrayBufferSinkPrototypeTableIndex[19] =
     { -1, -1 },
     { -1, -1 },
     { -1, -1 },
-    { -1, -1 },
+    { 5, -1 },
     { 4, -1 },
     { 1, 17 },
     { 2, 18 },
     { 3, -1 },
 };
 
-static const struct HashTableValue JSArrayBufferSinkPrototypeTableValues[5] = {
+static const struct HashTableValue JSArrayBufferSinkPrototypeTableValues[7] = {
    { "close"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, ArrayBufferSink__doClose, 0 } },
    { "flush"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, ArrayBufferSink__flush, 1 } },
    { "end"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, ArrayBufferSink__end, 0 } },
    { "start"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, ArrayBufferSink__start, 1 } },
    { "write"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, ArrayBufferSink__write, 1 } },
+   { "ref"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, ArrayBufferSink__ref, 0 } },
+   { "unref"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, ArrayBufferSink__unref, 0 } },
 };
 
 static const struct HashTable JSArrayBufferSinkPrototypeTable =
-    { 5, 15, false, nullptr, JSArrayBufferSinkPrototypeTableValues, JSArrayBufferSinkPrototypeTableIndex };
+    { 7, 15, false, nullptr, JSArrayBufferSinkPrototypeTableValues, JSArrayBufferSinkPrototypeTableIndex };
 
 
 
@@ -88,7 +90,7 @@ static const struct CompactHashIndex JSFileSinkPrototypeTableIndex[19] = {
     { -1, -1 },
     { -1, -1 },
     { -1, -1 },
-    { -1, -1 },
+    { 6, -1 },
     { -1, -1 },
     { -1, -1 },
     { -1, -1 },
@@ -97,23 +99,25 @@ static const struct CompactHashIndex JSFileSinkPrototypeTableIndex[19] = {
     { -1, -1 },
     { -1, -1 },
     { -1, -1 },
-    { -1, -1 },
+    { 5, -1 },
     { 4, -1 },
     { 1, 17 },
     { 2, 18 },
     { 3, -1 },
 };
 
-static const struct HashTableValue JSFileSinkPrototypeTableValues[5] = {
+static const struct HashTableValue JSFileSinkPrototypeTableValues[7] = {
    { "close"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, FileSink__doClose, 0 } },
    { "flush"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, FileSink__flush, 1 } },
    { "end"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, FileSink__end, 0 } },
    { "start"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, FileSink__start, 1 } },
    { "write"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, FileSink__write, 1 } },
+   { "ref"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, FileSink__ref, 0 } },
+   { "unref"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, FileSink__unref, 0 } },
 };
 
 static const struct HashTable JSFileSinkPrototypeTable =
-    { 5, 15, false, nullptr, JSFileSinkPrototypeTableValues, JSFileSinkPrototypeTableIndex };
+    { 7, 15, false, nullptr, JSFileSinkPrototypeTableValues, JSFileSinkPrototypeTableIndex };
 
 
 
@@ -166,7 +170,7 @@ static const struct CompactHashIndex JSHTTPResponseSinkPrototypeTableIndex[19] =
     { -1, -1 },
     { -1, -1 },
     { -1, -1 },
-    { -1, -1 },
+    { 6, -1 },
     { -1, -1 },
     { -1, -1 },
     { -1, -1 },
@@ -175,23 +179,25 @@ static const struct CompactHashIndex JSHTTPResponseSinkPrototypeTableIndex[19] =
     { -1, -1 },
     { -1, -1 },
     { -1, -1 },
-    { -1, -1 },
+    { 5, -1 },
     { 4, -1 },
     { 1, 17 },
     { 2, 18 },
     { 3, -1 },
 };
 
-static const struct HashTableValue JSHTTPResponseSinkPrototypeTableValues[5] = {
+static const struct HashTableValue JSHTTPResponseSinkPrototypeTableValues[7] = {
    { "close"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, HTTPResponseSink__doClose, 0 } },
    { "flush"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, HTTPResponseSink__flush, 1 } },
    { "end"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, HTTPResponseSink__end, 0 } },
    { "start"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, HTTPResponseSink__start, 1 } },
    { "write"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, HTTPResponseSink__write, 1 } },
+   { "ref"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, HTTPResponseSink__ref, 0 } },
+   { "unref"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, HTTPResponseSink__unref, 0 } },
 };
 
 static const struct HashTable JSHTTPResponseSinkPrototypeTable =
-    { 5, 15, false, nullptr, JSHTTPResponseSinkPrototypeTableValues, JSHTTPResponseSinkPrototypeTableIndex };
+    { 7, 15, false, nullptr, JSHTTPResponseSinkPrototypeTableValues, JSHTTPResponseSinkPrototypeTableIndex };
 
 
 
@@ -244,7 +250,7 @@ static const struct CompactHashIndex JSHTTPSResponseSinkPrototypeTableIndex[19]
     { -1, -1 },
     { -1, -1 },
     { -1, -1 },
-    { -1, -1 },
+    { 6, -1 },
     { -1, -1 },
     { -1, -1 },
     { -1, -1 },
@@ -253,23 +259,25 @@ static const struct CompactHashIndex JSHTTPSResponseSinkPrototypeTableIndex[19]
     { -1, -1 },
     { -1, -1 },
     { -1, -1 },
-    { -1, -1 },
+    { 5, -1 },
     { 4, -1 },
     { 1, 17 },
     { 2, 18 },
     { 3, -1 },
 };
 
-static const struct HashTableValue JSHTTPSResponseSinkPrototypeTableValues[5] = {
+static const struct HashTableValue JSHTTPSResponseSinkPrototypeTableValues[7] = {
    { "close"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, HTTPSResponseSink__doClose, 0 } },
    { "flush"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, HTTPSResponseSink__flush, 1 } },
    { "end"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, HTTPSResponseSink__end, 0 } },
    { "start"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, HTTPSResponseSink__start, 1 } },
    { "write"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, HTTPSResponseSink__write, 1 } },
+   { "ref"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, HTTPSResponseSink__ref, 0 } },
+   { "unref"_s, static_cast<unsigned>(PropertyAttribute::ReadOnly|PropertyAttribute::DontDelete|PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, HTTPSResponseSink__unref, 0 } },
 };
 
 static const struct HashTable JSHTTPSResponseSinkPrototypeTable =
-    { 5, 15, false, nullptr, JSHTTPSResponseSinkPrototypeTableValues, JSHTTPSResponseSinkPrototypeTableIndex };
+    { 7, 15, false, nullptr, JSHTTPSResponseSinkPrototypeTableValues, JSHTTPSResponseSinkPrototypeTableIndex };
 
 
 
diff --git a/src/bun.js/bindings/headers.h b/src/bun.js/bindings/headers.h
index 67dc9e393..419a8965a 100644
--- a/src/bun.js/bindings/headers.h
+++ b/src/bun.js/bindings/headers.h
@@ -1,5 +1,5 @@
 // clang-format off
-//-- AUTOGENERATED FILE -- 1669267948
+//-- AUTOGENERATED FILE -- 1669352137
 #pragma once
 
 #include <stddef.h>
@@ -30,7 +30,7 @@ typedef void* JSClassRef;
  typedef char* bJSC__SourceCode_buf;
  typedef struct bWTF__URL { unsigned char bytes[40]; } bWTF__URL;
  typedef char* bWTF__URL_buf;
- typedef struct bJSC__JSModuleRecord { unsigned char bytes[208]; } bJSC__JSModuleRecord;
+ typedef struct bJSC__JSModuleRecord { unsigned char bytes[216]; } bJSC__JSModuleRecord;
  typedef char* bJSC__JSModuleRecord_buf;
  typedef struct bJSC__ThrowScope { unsigned char bytes[8]; } bJSC__ThrowScope;
  typedef char* bJSC__ThrowScope_buf;
@@ -38,7 +38,7 @@ typedef void* JSClassRef;
  typedef char* bJSC__PropertyName_buf;
  typedef struct bJSC__JSFunction { unsigned char bytes[32]; } bJSC__JSFunction;
  typedef char* bJSC__JSFunction_buf;
- typedef struct bJSC__JSGlobalObject { unsigned char bytes[2840]; } bJSC__JSGlobalObject;
+ typedef struct bJSC__JSGlobalObject { unsigned char bytes[3128]; } bJSC__JSGlobalObject;
  typedef char* bJSC__JSGlobalObject_buf;
  typedef struct bJSC__JSCell { unsigned char bytes[8]; } bJSC__JSCell;
  typedef char* bJSC__JSCell_buf;
@@ -54,7 +54,7 @@ typedef void* JSClassRef;
  typedef char* bInspector__ScriptArguments_buf;
  typedef struct bJSC__Exception { unsigned char bytes[40]; } bJSC__Exception;
  typedef char* bJSC__Exception_buf;
- typedef struct bJSC__VM { unsigned char bytes[52008]; } bJSC__VM;
+ typedef struct bJSC__VM { unsigned char bytes[52176]; } bJSC__VM;
  typedef char* bJSC__VM_buf;
  typedef struct bJSC__JSString { unsigned char bytes[16]; } bJSC__JSString;
  typedef char* bJSC__JSString_buf;
@@ -853,6 +853,7 @@ ZIG_DECL JSC__JSValue ArrayBufferSink__endWithSink(void* arg0, JSC__JSGlobalObje
 ZIG_DECL void ArrayBufferSink__finalize(void* arg0);
 ZIG_DECL JSC__JSValue ArrayBufferSink__flush(JSC__JSGlobalObject* arg0, JSC__CallFrame* arg1);
 ZIG_DECL JSC__JSValue ArrayBufferSink__start(JSC__JSGlobalObject* arg0, JSC__CallFrame* arg1);
+ZIG_DECL void ArrayBufferSink__updateRef(void* arg0, bool arg1);
 ZIG_DECL JSC__JSValue ArrayBufferSink__write(JSC__JSGlobalObject* arg0, JSC__CallFrame* arg1);
 
 #endif
@@ -872,6 +873,7 @@ ZIG_DECL JSC__JSValue HTTPSResponseSink__endWithSink(void* arg0, JSC__JSGlobalOb
 ZIG_DECL void HTTPSResponseSink__finalize(void* arg0);
 ZIG_DECL JSC__JSValue HTTPSResponseSink__flush(JSC__JSGlobalObject* arg0, JSC__CallFrame* arg1);
 ZIG_DECL JSC__JSValue HTTPSResponseSink__start(JSC__JSGlobalObject* arg0, JSC__CallFrame* arg1);
+ZIG_DECL void HTTPSResponseSink__updateRef(void* arg0, bool arg1);
 ZIG_DECL JSC__JSValue HTTPSResponseSink__write(JSC__JSGlobalObject* arg0, JSC__CallFrame* arg1);
 
 #endif
@@ -891,6 +893,7 @@ ZIG_DECL JSC__JSValue HTTPResponseSink__endWithSink(void* arg0, JSC__JSGlobalObj
 ZIG_DECL void HTTPResponseSink__finalize(void* arg0);
 ZIG_DECL JSC__JSValue HTTPResponseSink__flush(JSC__JSGlobalObject* arg0, JSC__CallFrame* arg1);
 ZIG_DECL JSC__JSValue HTTPResponseSink__start(JSC__JSGlobalObject* arg0, JSC__CallFrame* arg1);
+ZIG_DECL void HTTPResponseSink__updateRef(void* arg0, bool arg1);
 ZIG_DECL JSC__JSValue HTTPResponseSink__write(JSC__JSGlobalObject* arg0, JSC__CallFrame* arg1);
 
 #endif
@@ -910,6 +913,7 @@ ZIG_DECL JSC__JSValue FileSink__endWithSink(void* arg0, JSC__JSGlobalObject* arg
 ZIG_DECL void FileSink__finalize(void* arg0);
 ZIG_DECL JSC__JSValue FileSink__flush(JSC__JSGlobalObject* arg0, JSC__CallFrame* arg1);
 ZIG_DECL JSC__JSValue FileSink__start(JSC__JSGlobalObject* arg0, JSC__CallFrame* arg1);
+ZIG_DECL void FileSink__updateRef(void* arg0, bool arg1);
 ZIG_DECL JSC__JSValue FileSink__write(JSC__JSGlobalObject* arg0, JSC__CallFrame* arg1);
 
 #endif
diff --git a/src/bun.js/scripts/generate-jssink.js b/src/bun.js/scripts/generate-jssink.js
index 174efcb06..b9b6b2c17 100644
--- a/src/bun.js/scripts/generate-jssink.js
+++ b/src/bun.js/scripts/generate-jssink.js
@@ -102,11 +102,15 @@ function header() {
             void detach() {
                 m_sinkPtr = nullptr;
 
-            }                       
+            }
+
+            static void analyzeHeap(JSCell*, JSC::HeapAnalyzer&);
 
-            static void analyzeHeap(JSCell*, JSC::HeapAnalyzer&);                                                                                                                   
+            void ref();
+            void unref();
                                                                                                                                                                                     
             void* m_sinkPtr;
+            int m_refCount { 1 };
                                                                                                                                                                                     
             ${className}(JSC::VM& vm, JSC::Structure* structure, void* sinkPtr)                                                                                                    
                 : Base(vm, structure)                                                                                                                                               
@@ -355,6 +359,51 @@ JSC_DEFINE_HOST_FUNCTION(functionStartDirectStream, (JSC::JSGlobalObject * lexic
     const protopad = `${controller}__close`.length;
     const padding = `${name}__doClose`.length;
     templ += `
+
+  void ${className}::ref() {
+    if (!m_sinkPtr)
+      return;
+
+    m_refCount++;
+    if (m_refCount == 1) {
+      ${name}__updateRef(m_sinkPtr, true);
+    }
+  }
+
+  void ${className}::unref() {
+    if (!m_sinkPtr)
+      return;
+
+      m_refCount = std::max(0, m_refCount - 1);
+      if (!m_refCount)
+      {
+        ${name}__updateRef(m_sinkPtr, false);
+      }
+  }
+
+JSC_DEFINE_HOST_FUNCTION(${name}__ref, (JSC::JSGlobalObject * lexicalGlobalObject, JSC::CallFrame *callFrame))
+{
+    auto& vm = lexicalGlobalObject->vm();
+    auto* sink = jsDynamicCast<WebCore::${className}*>(callFrame->thisValue());
+    if (LIKELY(sink)) {
+        sink->ref();
+    }
+    return JSC::JSValue::encode(JSC::jsUndefined());
+
+}
+
+
+JSC_DEFINE_HOST_FUNCTION(${name}__unref, (JSC::JSGlobalObject * lexicalGlobalObject, JSC::CallFrame *callFrame))
+{
+    auto& vm = lexicalGlobalObject->vm();
+    auto* sink = jsDynamicCast<WebCore::${className}*>(callFrame->thisValue());
+    if (LIKELY(sink)) {
+        sink->unref();
+    }
+    return JSC::JSValue::encode(JSC::jsUndefined());
+
+}
+
 JSC_DEFINE_CUSTOM_GETTER(function${name}__getter, (JSC::JSGlobalObject * lexicalGlobalObject, JSC::EncodedJSValue thisValue, JSC::PropertyName))
 {
     auto& vm = lexicalGlobalObject->vm();
@@ -472,6 +521,12 @@ JSC_DEFINE_HOST_FUNCTION(${name}__doClose, (JSC::JSGlobalObject * lexicalGlobalO
   write      ${`${name}__write`.padEnd(
     padding + 8,
   )} ReadOnly|DontDelete|Function 1
+  ref        ${`${name}__ref`.padEnd(
+    padding + 8,
+  )} ReadOnly|DontDelete|Function 0
+  unref      ${`${name}__unref`.padEnd(
+    padding + 8,
+  )} ReadOnly|DontDelete|Function 0
 @end
 */
 
@@ -849,8 +904,7 @@ extern "C" JSC__JSValue ${name}__createObject(JSC__JSGlobalObject* arg0, void* s
 {
     auto& vm = arg0->vm();
     Zig::GlobalObject* globalObject = reinterpret_cast<Zig::GlobalObject*>(arg0);
-    JSC::JSValue prototype = globalObject->${name}Prototype();
-    JSC::Structure* structure = WebCore::JS${name}::createStructure(vm, globalObject, prototype);
+    JSC::Structure* structure = globalObject->${name}Structure();
     return JSC::JSValue::encode(WebCore::JS${name}::create(vm, globalObject, structure, sinkPtr));
 }
 
diff --git a/src/global.zig b/src/global.zig
index 1dccb15e7..635e759ce 100644
--- a/src/global.zig
+++ b/src/global.zig
@@ -341,24 +341,27 @@ pub fn ensureNonBlocking(fd: anytype) void {
 }
 
 const global_scope_log = Output.scoped(.bun, false);
-pub fn isReadable(fd: std.os.fd_t) bool {
-    _ = fd;
-    return false;
-    // var polls = &[_]std.os.pollfd{
-    //     .{
-    //         .fd = fd,
-    //         .events = std.os.POLL.IN | std.os.POLL.ERR,
-    //         .revents = 0,
-    //     },
-    // };
-
-    // const result = (std.os.poll(polls, 0) catch 0) != 0;
-    // global_scope_log("isReadable: {d}", .{result});
-    // return result;
+pub fn isReadable(fd: std.os.fd_t) PollFlag {
+    var polls = &[_]std.os.pollfd{
+        .{
+            .fd = fd,
+            .events = std.os.POLL.IN | std.os.POLL.ERR,
+            .revents = 0,
+        },
+    };
+
+    const result = (std.os.poll(polls, 0) catch 0) != 0;
+    global_scope_log("isReadable: {d} ({d})", .{ result, polls[0].revents });
+    return if (result and polls[0].revents & std.os.POLL.HUP != 0)
+        PollFlag.hup
+    else if (result)
+        PollFlag.ready
+    else
+        PollFlag.not_ready;
 }
 
-pub const WritableFlag = enum { writable, not_writable, hup };
-pub fn isWritable(fd: std.os.fd_t) WritableFlag {
+pub const PollFlag = enum { ready, not_ready, hup };
+pub fn isWritable(fd: std.os.fd_t) PollFlag {
     var polls = &[_]std.os.pollfd{
         .{
             .fd = fd,
@@ -370,11 +373,11 @@ pub fn isWritable(fd: std.os.fd_t) WritableFlag {
     const result = (std.os.poll(polls, 0) catch 0) != 0;
     global_scope_log("isWritable: {d} ({d})", .{ result, polls[0].revents });
     if (result and polls[0].revents & std.os.POLL.HUP != 0) {
-        return WritableFlag.hup;
+        return PollFlag.hup;
     } else if (result) {
-        return WritableFlag.writable;
+        return PollFlag.ready;
     } else {
-        return WritableFlag.not_writable;
+        return PollFlag.not_ready;
     }
 }
 
diff --git a/test/bun.js/spawn-streaming-stdin.test.ts b/test/bun.js/spawn-streaming-stdin.test.ts
index 953548071..6424eadfc 100644
--- a/test/bun.js/spawn-streaming-stdin.test.ts
+++ b/test/bun.js/spawn-streaming-stdin.test.ts
@@ -12,30 +12,40 @@ test("spawn can write to stdin multiple chunks", async () => {
         cmd: [bunExe(), import.meta.dir + "/stdin-repro.js"],
         stdout: "pipe",
         stdin: "pipe",
-        stderr: "inherit",
+        stderr: Bun.file("/tmp/out.log"),
         env: {
           BUN_DEBUG_QUIET_LOGS: 1,
         },
       });
+      // (async function () {
+      //   for await (var chunk of proc.stderr) {
+      //     console.error("[stderr]", new TextDecoder().decode(chunk));
+      //   }
+      // })();
       exited = proc.exited;
       var counter = 0;
       var inCounter = 0;
       const prom2 = (async function () {
-        while (inCounter++ < 4) {
+        while (true) {
           await new Promise((resolve, reject) => setTimeout(resolve, 8));
           proc.stdin.write("Wrote to stdin!");
-          await proc.stdin.flush();
+          inCounter++;
+
+          if (inCounter === 4) break;
         }
-        await proc.stdin.end();
+        await new Promise((resolve) =>
+          Promise.resolve(proc.stdin.end()).then(resolve),
+        );
       })();
 
+      var chunks = [];
       const prom = (async function () {
         try {
           for await (var chunk of proc.stdout) {
-            expect(new TextDecoder().decode(chunk)).toBe("Wrote to stdin!\n");
+            chunks.push(chunk);
             counter++;
 
-            if (counter > 3) break;
+            if (counter === 4) break;
           }
         } catch (e) {
           console.log(e.stack);
@@ -43,11 +53,15 @@ test("spawn can write to stdin multiple chunks", async () => {
         }
       })();
       await Promise.all([prom, prom2]);
+      const code = await exited;
+      console.log(code);
       expect(counter).toBe(4);
+      expect(Buffer.concat(chunks).toString().trim()).toBe(
+        "Wrote to stdin!\n".repeat(4).trim(),
+      );
       //   proc.kill();
+
+      gcTick(true);
     })();
-    await exited;
   }
-
-  gcTick(true);
 });
diff --git a/test/bun.js/stdin-repro.js b/test/bun.js/stdin-repro.js
index 05daf0637..5f945be7c 100644
--- a/test/bun.js/stdin-repro.js
+++ b/test/bun.js/stdin-repro.js
@@ -1,5 +1,7 @@
-while (true) {
-  for await (let chunk of Bun.stdin.stream()) {
-    console.log(new Buffer(chunk).toString());
-  }
+var count = 5;
+for await (let chunk of Bun.stdin.stream()) {
+  const str = new Buffer(chunk).toString();
+  console.error("how many?", count, chunk.byteLength);
+  count -= str.split("\n").length;
+  console.log(str);
 }
diff --git a/test/bun.js/streams.test.js b/test/bun.js/streams.test.js
index a872b7701..406c80852 100644
--- a/test/bun.js/streams.test.js
+++ b/test/bun.js/streams.test.js
@@ -225,7 +225,7 @@ it("Bun.file() read text from pipe", async () => {
   const large = "HELLO!".repeat((((1024 * 65) / "HELLO!".length) | 0) + 1);
 
   const chunks = [];
-  var out = Bun.file("/tmp/fifo").stream();
+
   const proc = Bun.spawn({
     cmd: [
       "bash",
@@ -244,17 +244,17 @@ it("Bun.file() read text from pipe", async () => {
 
   const prom = (async function () {
     while (chunks.length === 0) {
+      var out = Bun.file("/tmp/fifo").stream();
       for await (const chunk of out) {
         chunks.push(chunk);
       }
-      console.log("done");
     }
+    return Buffer.concat(chunks).toString();
   })();
 
   const [status, output] = await Promise.all([exited, prom]);
-  console.log("here");
-  expect(output.length).toBe(large.length);
-  expect(output).toBe(large);
+  expect(output.length).toBe(large.length + 1);
+  expect(output).toBe(large + "\n");
   expect(status).toBe(0);
 });
 
@@ -452,7 +452,7 @@ it("ReadableStream for Blob", async () => {
 
 it("ReadableStream for File", async () => {
   var blob = file(import.meta.dir + "/fetch.js.txt");
-  var stream = blob.stream(24);
+  var stream = blob.stream();
   const chunks = [];
   var reader = stream.getReader();
   stream = undefined;