Fix segfault due to GC and some more helper functions (#1221)

* Fix segfault due to GC and some more helper functions

* fix upon reviews

* add visitChildren

1 files changed, 103 insertions, 83 deletions

diff --git a/src/bun.js/bindings/JSReadableHelper.cpp b/src/bun.js/bindings/JSReadableHelper.cpp
index ef17982d4..de1c30799 100644
--- a/src/bun.js/bindings/JSReadableHelper.cpp
+++ b/src/bun.js/bindings/JSReadableHelper.cpp
@@ -3,6 +3,7 @@
 #include "JSBufferList.h"
 #include "JSBuffer.h"
 #include "JSEventEmitter.h"
+#include "JSStringDecoder.h"
 #include "JavaScriptCore/Lookup.h"
 #include "JavaScriptCore/ObjectConstructor.h"
 #include "ZigGlobalObject.h"
@@ -14,24 +15,27 @@
 namespace WebCore {
 using namespace JSC;
 
+#define JSReadableHelper_EXTRACT_STREAM_STATE \
+    VM& vm = lexicalGlobalObject->vm(); \
+    auto throwScope = DECLARE_THROW_SCOPE(vm); \
+    \
+    if (callFrame->argumentCount() < 2) { \
+        throwTypeError(lexicalGlobalObject, throwScope, "Not enough arguments"_s); \
+        return JSValue::encode(jsUndefined()); \
+    } \
+    \
+    JSObject* stream = callFrame->uncheckedArgument(0).toObject(lexicalGlobalObject); \
+    RETURN_IF_EXCEPTION(throwScope, JSValue::encode(jsUndefined())); \
+    JSReadableState* state = jsCast<JSReadableState*>(callFrame->uncheckedArgument(1)); \
+    if (!state) { \
+        throwTypeError(lexicalGlobalObject, throwScope, "Second argument not ReadableState"_s); \
+        return JSValue::encode(jsUndefined()); \
+    }
+
 static JSC_DECLARE_HOST_FUNCTION(jsReadable_maybeReadMore_);
 JSC_DEFINE_HOST_FUNCTION(jsReadable_maybeReadMore_, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame))
 {
-    VM& vm = lexicalGlobalObject->vm();
-    auto throwScope = DECLARE_THROW_SCOPE(vm);
-
-    if (callFrame->argumentCount() < 2) {
-        throwTypeError(lexicalGlobalObject, throwScope, "Not enough arguments"_s);
-        return JSValue::encode(jsUndefined());
-    }
-
-    JSObject* stream = callFrame->uncheckedArgument(0).toObject(lexicalGlobalObject);
-    RETURN_IF_EXCEPTION(throwScope, JSValue::encode(jsUndefined()));
-    JSReadableState* state = jsDynamicCast<JSReadableState*>(callFrame->uncheckedArgument(1));
-    if (!state) {
-        throwTypeError(lexicalGlobalObject, throwScope, "Second argument not ReadableState"_s);
-        return JSValue::encode(jsUndefined());
-    }
+    JSReadableHelper_EXTRACT_STREAM_STATE
 
     auto read = stream->get(lexicalGlobalObject, Identifier::fromString(vm, "read"_s));
     auto callData = JSC::getCallData(read);
@@ -43,8 +47,8 @@ JSC_DEFINE_HOST_FUNCTION(jsReadable_maybeReadMore_, (JSGlobalObject * lexicalGlo
     args.append(jsNumber(0));
 
     while (
-        !state->m_reading &&
-        !state->m_ended &&
+        !state->getBool(JSReadableState::reading) &&
+        !state->getBool(JSReadableState::ended) &&
         (state->m_length < state->m_highWaterMark || (state->m_flowing > 0 && state->m_length == 0))) {
         int64_t len = state->m_length;
 
@@ -58,22 +62,13 @@ JSC_DEFINE_HOST_FUNCTION(jsReadable_maybeReadMore_, (JSGlobalObject * lexicalGlo
 
 JSC_DEFINE_HOST_FUNCTION(jsReadable_maybeReadMore, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame))
 {
-    VM& vm = lexicalGlobalObject->vm();
-    auto throwScope = DECLARE_THROW_SCOPE(vm);
-
-    if (callFrame->argumentCount() < 2) {
-        throwTypeError(lexicalGlobalObject, throwScope, "Not enough arguments"_s);
-        return JSValue::encode(jsUndefined());
-    }
-
-    JSValue streamVal = callFrame->uncheckedArgument(0);
-    JSValue stateVal = callFrame->uncheckedArgument(1);
+    JSReadableHelper_EXTRACT_STREAM_STATE
 
     // make this static?
     JSFunction* maybeReadMore_ = JSC::JSFunction::create(
         vm, lexicalGlobalObject, 0, "maybeReadMore_"_s, jsReadable_maybeReadMore_, ImplementationVisibility::Public);
 
-    lexicalGlobalObject->queueMicrotask(maybeReadMore_, streamVal, stateVal, JSValue{}, JSValue{});
+    lexicalGlobalObject->queueMicrotask(maybeReadMore_, JSValue(stream), JSValue(state), JSValue{}, JSValue{});
     RELEASE_AND_RETURN(throwScope, JSValue::encode(jsUndefined()));
 }
 
@@ -100,23 +95,9 @@ void flow(JSGlobalObject* lexicalGlobalObject, JSObject* stream, JSReadableState
 static JSC_DECLARE_HOST_FUNCTION(jsReadable_resume_);
 JSC_DEFINE_HOST_FUNCTION(jsReadable_resume_, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame))
 {
-    VM& vm = lexicalGlobalObject->vm();
-    auto throwScope = DECLARE_THROW_SCOPE(vm);
+    JSReadableHelper_EXTRACT_STREAM_STATE
 
-    if (callFrame->argumentCount() < 2) {
-        throwTypeError(lexicalGlobalObject, throwScope, "Not enough arguments"_s);
-        return JSValue::encode(jsUndefined());
-    }
-
-    JSObject* stream = callFrame->uncheckedArgument(0).toObject(lexicalGlobalObject);
-    RETURN_IF_EXCEPTION(throwScope, JSValue::encode(jsUndefined()));
-    JSReadableState* state = jsDynamicCast<JSReadableState*>(callFrame->uncheckedArgument(1));
-    if (!state) {
-        throwTypeError(lexicalGlobalObject, throwScope, "Second argument not ReadableState"_s);
-        return JSValue::encode(jsUndefined());
-    }
-
-    if (!state->m_reading) {
+    if (!state->getBool(JSReadableState::reading)) {
         // stream.read(0)
         auto read = stream->get(lexicalGlobalObject, Identifier::fromString(vm, "read"_s));
         auto callData = JSC::getCallData(read);
@@ -129,7 +110,7 @@ JSC_DEFINE_HOST_FUNCTION(jsReadable_resume_, (JSGlobalObject * lexicalGlobalObje
         JSC::call(lexicalGlobalObject, read, callData, JSValue(stream), args);
     }
 
-    state->m_resumeScheduled = true;
+    state->setBool(JSReadableState::resumeScheduled, true);
     // stream.emit('resume')
     auto eventType = Identifier::fromString(vm, "resume"_s);
     MarkedArgumentBuffer args;
@@ -142,7 +123,7 @@ JSC_DEFINE_HOST_FUNCTION(jsReadable_resume_, (JSGlobalObject * lexicalGlobalObje
 
     flow(lexicalGlobalObject, stream, state);
 
-    if (state->m_flowing && !state->m_reading) {
+    if (state->m_flowing > 0 && !state->getBool(JSReadableState::reading)) {
         // stream.read(0)
         auto read = stream->get(lexicalGlobalObject, Identifier::fromString(vm, "read"_s));
         auto callData = JSC::getCallData(read);
@@ -159,54 +140,26 @@ JSC_DEFINE_HOST_FUNCTION(jsReadable_resume_, (JSGlobalObject * lexicalGlobalObje
 
 JSC_DEFINE_HOST_FUNCTION(jsReadable_resume, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame))
 {
-    VM& vm = lexicalGlobalObject->vm();
-    auto throwScope = DECLARE_THROW_SCOPE(vm);
+    JSReadableHelper_EXTRACT_STREAM_STATE
 
-    if (callFrame->argumentCount() < 2) {
-        throwTypeError(lexicalGlobalObject, throwScope, "Not enough arguments"_s);
-        return JSValue::encode(jsUndefined());
-    }
-
-    JSValue streamVal = callFrame->uncheckedArgument(0);
-    JSValue stateVal = callFrame->uncheckedArgument(1);
-
-    JSReadableState* state = jsDynamicCast<JSReadableState*>(callFrame->uncheckedArgument(1));
-    if (!state) {
-        throwTypeError(lexicalGlobalObject, throwScope, "Second argument not ReadableState"_s);
-        return JSValue::encode(jsUndefined());
-    }
-
-    if (!state->m_resumeScheduled) {
-        state->m_resumeScheduled = true;
+    if (!state->getBool(JSReadableState::resumeScheduled)) {
+        state->setBool(JSReadableState::resumeScheduled, true);
         // make this static?
         JSFunction* resume_ = JSC::JSFunction::create(
             vm, lexicalGlobalObject, 0, "resume_"_s, jsReadable_resume_, ImplementationVisibility::Public);
 
-        lexicalGlobalObject->queueMicrotask(resume_, streamVal, stateVal, JSValue{}, JSValue{});
+        lexicalGlobalObject->queueMicrotask(resume_, JSValue(stream), JSValue(state), JSValue{}, JSValue{});
     }
     RELEASE_AND_RETURN(throwScope, JSValue::encode(jsUndefined()));
 }
 
-JSC_DEFINE_HOST_FUNCTION(jsReadable_emitReadable_, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame))
+EncodedJSValue emitReadable_(JSGlobalObject* lexicalGlobalObject, JSObject* stream, JSReadableState* state)
 {
     VM& vm = lexicalGlobalObject->vm();
     auto throwScope = DECLARE_THROW_SCOPE(vm);
 
-    if (callFrame->argumentCount() < 2) {
-        throwTypeError(lexicalGlobalObject, throwScope, "Not enough arguments"_s);
-        return JSValue::encode(jsUndefined());
-    }
-
-    JSObject* stream = callFrame->uncheckedArgument(0).toObject(lexicalGlobalObject);
-    RETURN_IF_EXCEPTION(throwScope, JSValue::encode(jsUndefined()));
-    JSReadableState* state = jsDynamicCast<JSReadableState*>(callFrame->uncheckedArgument(1));
-    if (!state) {
-        throwTypeError(lexicalGlobalObject, throwScope, "Second argument not ReadableState"_s);
-        return JSValue::encode(jsUndefined());
-    }
-
-    JSValue errored = state->getDirect(vm, JSC::Identifier::fromString(vm, "errored"_s));
-    if (!state->m_destroyed && !errored.toBoolean(lexicalGlobalObject) && (state->m_length || state->m_ended)) {
+    JSValue errored = state->m_errored.get();
+    if (!state->getBool(JSReadableState::destroyed) && !errored.toBoolean(lexicalGlobalObject) && (state->m_length || state->getBool(JSReadableState::ended))) {
         // stream.emit('readable')
         auto eventType = Identifier::fromString(vm, "readable"_s);
         MarkedArgumentBuffer args;
@@ -217,12 +170,79 @@ JSC_DEFINE_HOST_FUNCTION(jsReadable_emitReadable_, (JSGlobalObject * lexicalGlob
         }
         emitter->wrapped().emitForBindings(eventType, args);
 
-        state->m_emittedReadable = false;
+        state->setBool(JSReadableState::emittedReadable, false);
     }
 
-    state->m_needReadable = state->m_flowing <= 0 && !state->m_ended && state->m_length <= state->m_highWaterMark;
+    state->setBool(JSReadableState::needReadable, state->m_flowing <= 0 && !state->getBool(JSReadableState::ended) && state->m_length <= state->m_highWaterMark);
     flow(lexicalGlobalObject, stream, state);
+    return JSValue::encode(jsUndefined());
+}
+
+JSC_DECLARE_HOST_FUNCTION(jsReadable_emitReadable_);
+JSC_DEFINE_HOST_FUNCTION(jsReadable_emitReadable_, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame))
+{
+    JSReadableHelper_EXTRACT_STREAM_STATE
+
+    emitReadable_(lexicalGlobalObject, stream, state);
+
     RELEASE_AND_RETURN(throwScope, JSValue::encode(jsUndefined()));
 }
 
+EncodedJSValue emitReadable(JSGlobalObject* lexicalGlobalObject, JSObject* stream, JSReadableState* state)
+{
+    VM& vm = lexicalGlobalObject->vm();
+
+    state->setBool(JSReadableState::needReadable, false);
+    if (!state->getBool(JSReadableState::emittedReadable)) {
+        state->setBool(JSReadableState::emittedReadable, true);
+        // make this static?
+        JSFunction* emitReadable_ = JSC::JSFunction::create(
+            vm, lexicalGlobalObject, 0, "emitReadable_"_s, jsReadable_emitReadable_, ImplementationVisibility::Public);
+
+        lexicalGlobalObject->queueMicrotask(emitReadable_, JSValue(stream), JSValue(state), JSValue{}, JSValue{});
+    }
+    return JSValue::encode(jsUndefined());
+}
+
+JSC_DEFINE_HOST_FUNCTION(jsReadable_emitReadable, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame))
+{
+    JSReadableHelper_EXTRACT_STREAM_STATE
+
+    RELEASE_AND_RETURN(throwScope, emitReadable(lexicalGlobalObject, stream, state));
+}
+
+JSC_DEFINE_HOST_FUNCTION(jsReadable_onEofChunk, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame))
+{
+    JSReadableHelper_EXTRACT_STREAM_STATE
+
+    if (state->getBool(JSReadableState::ended))
+        RELEASE_AND_RETURN(throwScope, JSValue::encode(jsUndefined()));
+
+    auto decoder = jsDynamicCast<JSStringDecoder*>(state->m_decoder.get());
+    if (decoder) {
+        JSString* chunk = jsDynamicCast<JSString*>(decoder->end(vm, lexicalGlobalObject, nullptr, 0));
+        if (chunk && chunk->length()) {
+            auto buffer = jsDynamicCast<JSBufferList*>(state->m_buffer.get());
+            if (!buffer) {
+                throwTypeError(lexicalGlobalObject, throwScope, "Not buffer on stream"_s);
+                return JSValue::encode(jsUndefined());
+            }
+            buffer->push(vm, JSValue(chunk));
+            state->m_length += state->getBool(JSReadableState::objectMode) ? 1 : chunk->length();
+        }
+    }
+
+    state->setBool(JSReadableState::ended, true);
+
+    if (state->getBool(JSReadableState::sync)) {
+        RELEASE_AND_RETURN(throwScope, emitReadable(lexicalGlobalObject, stream, state));
+    } else {
+        state->setBool(JSReadableState::needReadable, false);
+        state->setBool(JSReadableState::emittedReadable, true);
+        RELEASE_AND_RETURN(throwScope, emitReadable_(lexicalGlobalObject, stream, state));
+    }
+}
+
+#undef JSReadableHelper_EXTRACT_STREAM_STATE
+
 } // namespace WebCore