Address review feedback

Author: jasnell

src/workerd/api/streams/compression.c++

@@ -249,9 +249,8 @@ class CompressionStreamBase: public kj::Refcounted,
   explicit CompressionStreamBase(kj::String format,
       Context::ContextFlags flags,
       kj::Arc<const jsg::ExternalMemoryTarget>&& externalMemoryTarget)
-      : context(mode, format, flags, kj::mv(externalMemoryTarget)) {
-    state.template transitionTo<Open>();
-  }
+      : state(decltype(state)::template create<Open>()),
+        context(mode, format, flags, kj::mv(externalMemoryTarget)) {}
 
   // WritableStreamSink implementation ---------------------------------------------------
 

src/workerd/api/streams/identity-transform-stream.c++

@@ -60,9 +60,9 @@ class IdentityTransformStreamImpl final: public kj::Refcounted,
  public:
   // The limit is the maximum number of bytes that can be fed through the stream.
   // If kj::none, there is no limit.
-  explicit IdentityTransformStreamImpl(kj::Maybe<uint64_t> limit = kj::none): limit(limit) {
-    state.transitionTo<Idle>();
-  }
+  explicit IdentityTransformStreamImpl(kj::Maybe<uint64_t> limit = kj::none)
+      : limit(limit),
+        state(IdentityTransformState::create<Idle>()) {}
 
   ~IdentityTransformStreamImpl() noexcept(false) {
     // Due to the different natures of JS and C++ disposal, there is no point in enforcing the limit

src/workerd/api/streams/queue.h

@@ -343,7 +343,6 @@ class ConsumerImpl final {
       : queue(queue),
         state(ConsumerState::template create<Ready>()),
         stateListener(stateListener) {
-    state.template transitionTo<Ready>();
     queue.addConsumer(this);
   }
 

src/workerd/api/streams/readable-source-adapter.c++

@@ -719,9 +719,7 @@ ReadableSourceKjAdapter::ReadableSourceKjAdapter(
     : state(KjState::create<KjOpen>(kj::heap<Active>(js, ioContext, kj::mv(stream)))),
       options(options),
       selfRef(
-          kj::rc<WeakRef<ReadableSourceKjAdapter>>(kj::Badge<ReadableSourceKjAdapter>{}, *this)) {
-  state.transitionTo<KjOpen>(kj::heap<Active>(js, ioContext, kj::mv(stream)));
-}
+          kj::rc<WeakRef<ReadableSourceKjAdapter>>(kj::Badge<ReadableSourceKjAdapter>{}, *this)) {}
 
 ReadableSourceKjAdapter::~ReadableSourceKjAdapter() noexcept(false) {
   selfRef->invalidate();

src/workerd/api/streams/readable.c++

@@ -16,9 +16,8 @@ namespace workerd::api {
 
 ReaderImpl::ReaderImpl(ReadableStreamController::Reader& reader)
     : ioContext(tryGetIoContext()),
-      reader(reader) {
-  state.transitionTo<Initial>();
-}
+      reader(reader),
+      state(ReaderState::create<Initial>()) {}
 
 ReaderImpl::~ReaderImpl() noexcept(false) {
   KJ_IF_SOME(attached, state.tryGetActiveUnsafe()) {

src/workerd/api/streams/standard.c++

@@ -1168,11 +1168,7 @@ void ReadableImpl<Self>::pullIfNeeded(jsg::Lock& js, jsg::Ref<Self> self) {
 
 template <typename Self>
 void ReadableImpl<Self>::visitForGc(jsg::GcVisitor& visitor) {
-  KJ_IF_SOME(errored, state.template tryGetUnsafe<StreamStates::Errored>()) {
-    visitor.visit(errored);
-  } else KJ_IF_SOME(queue, state.template tryGetUnsafe<Queue>()) {
-    visitor.visit(queue);
-  }
+  state.visitForGc(visitor);
   KJ_IF_SOME(pendingCancel, maybePendingCancel) {
     visitor.visit(pendingCancel.fulfiller, pendingCancel.promise);
   }
@@ -1636,11 +1632,7 @@ jsg::Promise<void> WritableImpl<Self>::write(
 
 template <typename Self>
 void WritableImpl<Self>::visitForGc(jsg::GcVisitor& visitor) {
-  KJ_IF_SOME(error, state.template tryGetUnsafe<StreamStates::Errored>()) {
-    visitor.visit(error);
-  } else KJ_IF_SOME(erroring, state.template tryGetUnsafe<StreamStates::Erroring>()) {
-    visitor.visit(erroring.reason);
-  }
+  state.visitForGc(visitor);
   visitor.visit(inFlightWrite, inFlightClose, closeRequest, algorithms, signal);
   KJ_IF_SOME(pendingAbort, maybePendingAbort) {
     visitor.visit(*pendingAbort);

src/workerd/api/streams/writable-sink-adapter.c++

@@ -134,8 +134,6 @@ WritableStreamSinkJsAdapter::WritableStreamSinkJsAdapter(
       backpressureState(newBackpressureState(js)),
       selfRef(kj::rc<WeakRef<WritableStreamSinkJsAdapter>>(
           kj::Badge<WritableStreamSinkJsAdapter>{}, *this)) {
-  state.transitionTo<Open>(
-      ioContext.addObject(kj::heap<Active>(kj::mv(sink), kj::mv(options).orDefault({}))));
   // We want the initial backpressure state to be "ready".
   backpressureState.release(js);
 }

src/workerd/api/streams/writable.c++

@@ -10,9 +10,9 @@
 
 namespace workerd::api {
 
-WritableStreamDefaultWriter::WritableStreamDefaultWriter(): ioContext(tryGetIoContext()) {
-  state.transitionTo<Initial>();
-}
+WritableStreamDefaultWriter::WritableStreamDefaultWriter()
+    : ioContext(tryGetIoContext()),
+      state(WriterState::create<Initial>()) {}
 
 WritableStreamDefaultWriter::~WritableStreamDefaultWriter() noexcept(false) {
   KJ_IF_SOME(attached, state.tryGetActiveUnsafe()) {