AsyncPipe BlockedPumpTo allows writes to broken downstream after write error

Author: jasnellCreated Feb 22, 2026Updated Jul 30, 2026

###Summary

After a write through AsyncPipe's BlockedPumpTo state fails (e.g. downstream disconnect), subsequent writes can still pass through and reach the broken downstream output stream. The error handler teeExceptionPromise releases the canceler and rejects the pump fulfiller, but does not end the BlockedPumpTo state on the pipe. This leaves the pipe in a zombie state that forwards writes to a stream that may be in an inconsistent state.

###Impact

In production (Cloudflare Workers runtime), this triggers HttpOutputStream's assertion:

failed: expected !writeInProgress; concurrent write()s not allowed

at kj/compat/http.c++:2393. When a write to the HttpOutputStream fails mid-flight, writeInProgress is intentionally left true to prevent further use of the inconsistent stream. But the BlockedPumpTo bug allows a second write to reach writeBodyData() anyway, hitting the assertion.

The typical stack trace involves data flowing through an AsyncPipe (as used by http-over-capnp.c++ for response body proxying):

HttpOutputStream::writeBodyData              kj/compat/http.c++:2393
HttpFixedLengthEntityWriter::write           kj/compat/http.c++:2591
AsyncPipe::BlockedPumpTo::write              kj/async-io.c++:1257
AsyncPipe::BlockedPumpFrom::pumpTo           kj/async-io.c++:842
...
KjToCapnpHttpServiceAdapter::request         capnp/compat/http-over-capnp.c++

Root Cause

In async-io.c++, BlockedPumpTo::write():

cpp

return canceler.wrap(output.write(writeBuffer.first(actual))
    .then([this,actual,writeBuffer]() -> kj::Promise<void> {
  canceler.release();
  // ... success path: may call pipe.endState(*this) ...
}, teeExceptionPromise<void>(fulfiller, canceler)));

And teeExceptionPromise:

cpp

return [&fulfiller, &canceler](kj::Exception&& e) -> kj::Promise<T> {
    canceler.release();        // canceler becomes empty
    fulfiller.reject(kj::cp(e)); // pump promise rejected
    return kj::mv(e);
    // NOTE: pipe.endState() is never called!
};

After the error handler runs:

canceler.isEmpty() → true (released)
pipe.state → still points to BlockedPumpTo (never ended)
BlockedPumpTo object → still alive (the newAdaptedPromise node hasn't been destroyed yet)

So when a subsequent pipe.write() arrives, it dispatches to BlockedPumpTo::write(), passes the canceler.isEmpty() guard, and calls output.write() on the broken downstream.

Reproduction

The following KJ_TEST demonstrates the issue (added to async-io-test.c++):

cpp

KJ_TEST("Userland pipe pumpTo write after output error") {
  kj::EventLoop loop;
  WaitScope ws(loop);
  auto pipe = newOneWayPipe();
  // Output stream where we control when writes complete via a fulfiller.
  struct TestOutputStream final : public kj::AsyncOutputStream {
    uint writeCount = 0;
    kj::Maybe<kj::Own<kj::PromiseFulfiller<void>>> pendingFulfiller;
    kj::Promise<void> write(kj::ArrayPtr<const byte> buffer) override {
      ++writeCount;
      auto paf = kj::newPromiseAndFulfiller<void>();
      pendingFulfiller = kj::mv(paf.fulfiller);
      return kj::mv(paf.promise);
    }
    kj::Promise<void>
    write(kj::ArrayPtr<const kj::ArrayPtr<const byte>> pieces) override {
      return write(pieces[0]);
    }
    kj::Promise<void> whenWriteDisconnected() override {
      return kj::NEVER_DONE;
    }
  };
  TestOutputStream output;
  // Start pump: pipe.in -> output. Creates BlockedPumpTo state.
  auto pumpPromise = pipe.in->pumpTo(output, kj::maxValue);
  // First write reaches the output.
  auto write1 = pipe.out->write("foo"_kjb);
  KJ_ASSERT(!write1.poll(ws));
  KJ_ASSERT(output.writeCount == 1);
  // Reject the first write to simulate a downstream disconnect.
  {
    auto fulfiller = kj::mv(KJ_ASSERT_NONNULL(output.pendingFulfiller));
    output.pendingFulfiller = kj::none;
    fulfiller->reject(KJ_EXCEPTION(DISCONNECTED, "simulated disconnect"));
  }
  // Error propagates: teeExceptionPromise releases canceler, rejects pump.
  KJ_ASSERT(write1.poll(ws));
  // Second write should NOT reach the output, but it does (bug).
  auto write2 = pipe.out->write("bar"_kjb);
  KJ_EXPECT(output.writeCount == 1,  // FAILS: writeCount == 2
      "second write reached the output after an error");
}

Actual result: output.writeCount == 2 — the second write reaches the broken downstream. Expected result: output.writeCount == 1 — the pipe should stop forwarding writes after a failure.

Suggested Fix

teeExceptionPromise (or the error path in BlockedPumpTo::write()) should call pipe.endState(*this) to remove the zombie state. The same pattern likely affects other teeExceptionPromise call sites in BlockedPumpTo and BlockedPumpFrom.

Source: capnproto/capnproto

View original on GitHub View discussion on GitHub