Merge pull request #5166 from cloudflare/felix/092325-exception-handling

EW-9453 Implement proper STW exception propagation

Author: fhanau

src/workerd/io/trace-stream.c++

@@ -12,9 +12,6 @@
 namespace workerd::tracing {
 namespace {
 
-// Uniquely identifies js tail session failures
-constexpr kj::Exception::DetailTypeId TAIL_STREAM_JS_FAILURE = 0xcde53d65a46183f7;
-
 #define STRS(V) \
  V(ALARM, "alarm") \
  V(ATTRIBUTES, "attributes") \
@@ -606,10 +603,9 @@ class TailStreamTarget final: public rpc::TailStreamTarget::Server {
 
  auto ownReportContext = capnp::CallContextHook::from(reportContext).addRef();
 
- auto promise =
- ioContext
- .run([this, &ioContext, reportContext, ownReportContext = kj::mv(ownReportContext)](
- Worker::Lock& lock) mutable -> kj::Promise<void> {
+ auto promise = ioContext.run(
+ [this, &ioContext, reportContext, ownReportContext = kj::mv(ownReportContext)](
+ Worker::Lock& lock) mutable -> kj::Promise<void> {
  auto params = reportContext.getParams();
  KJ_ASSERT(params.hasEvents(), "Events are required.");
  auto eventReaders = params.getEvents();
@@ -638,19 +634,26 @@ class TailStreamTarget final: public rpc::TailStreamTarget::Server {
  } else {
  return kj::mv(result);
  }
- }).catch_([](kj::Exception&& e) {
+ });
+
+ auto paf = kj::newPromiseAndFulfiller<void>();
+ promise = promise.then([&fulfiller = *paf.fulfiller]() { fulfiller.fulfill(); },
+ [&, &fulfiller = *paf.fulfiller](kj::Exception&& e) {
+ // This is the top level exception catcher for tail events being delivered. We do not want to
+ // propagate JS exceptions to the client side here, all exceptions should stay within this
+ // customEvent. Instead, we propagate the exception to the doneFulfiller, where it is used to
+ // set the right outcome code and re-thrown if appropriate. By rejecting the doneFulfiller, we
+ // also ensure that no more tail events get delivered.
  if (jsg::isTunneledException(e.getDescription())) {
  auto description = jsg::stripRemoteExceptionPrefix(e.getDescription());
  if (!description.startsWith("remote.")) {
  e.setDescription(kj::str("remote.", description));
  }
  }
- kj::throwFatalException(kj::mv(e));
+ // We still fulfill this fulfiller to disarm the cancellation check below
+ fulfiller.fulfill();
+ doneFulfiller->reject(kj::mv(e));
  });
-
- auto paf = kj::newPromiseAndFulfiller<void>();
- promise = promise.then([&fulfiller = *paf.fulfiller]() { fulfiller.fulfill(); },
- [&fulfiller = *paf.fulfiller](kj::Exception&& e) { fulfiller.reject(kj::mv(e)); });
  promise = promise.attach(kj::defer([fulfiller = kj::mv(paf.fulfiller)]() mutable {
  if (fulfiller->isWaiting()) {
  fulfiller->reject(JSG_KJ_EXCEPTION(FAILED, Error,
@@ -764,10 +767,11 @@ class TailStreamTarget final: public rpc::TailStreamTarget::Server {
  results->get().setStop(true);
  doneFulfiller->fulfill();
  }),
- ioContext.addFunctor([this, results = sharedResults.addRef()](
- jsg::Lock& js, jsg::Value&& error) mutable {
+ ioContext.addFunctor(
+ [results = sharedResults.addRef()](jsg::Lock& js, jsg::Value&& error) mutable {
+ // Received a JS error. Do not reject doneFulfiller yet, this will be handled when we catch
+ // the exception later.
  results->get().setStop(true);
- doneFulfiller->fulfill();
  js.throwException(kj::mv(error));
  })));
  } catch (...) {
@@ -860,15 +864,11 @@ class TailStreamTarget final: public rpc::TailStreamTarget::Server {
  // the doneFulfiller afterwards, indicating that TailStreamTarget has received all events over
  // the stream and has done all its work, that the stream self-evidently did not get canceled
  // prematurely. This applies even if promises were rejected.
+ // No need to catch exceptions here: They will be handled in report() alongside exceptions
+ // from the onset event etc. JSG knows how JS exceptions look like, so we don't need an
+ // identifier for them.
  if (doFulfill) {
- p = p.then(js, [&](jsg::Lock& js) { doneFulfiller->fulfill(); },
- [&](jsg::Lock& js, jsg::Value&& value) {
- // Convert the JS exception to a KJ exception, preserving all details
- kj::Exception exception = js.exceptionToKj(kj::mv(value));
- // Mark this as a tail stream failure for proper classification
- exception.setDetail(TAIL_STREAM_JS_FAILURE, kj::heapArray<kj::byte>(0));
- doneFulfiller->reject(kj::mv(exception));
- });
+ p = p.then(js, [&](jsg::Lock& js) { doneFulfiller->fulfill(); });
  }
  return ioContext.awaitJs(js, kj::mv(p));
  }
@@ -916,8 +916,12 @@ kj::Promise<WorkerInterface::CustomEvent::Result> TailStreamCustomEventImpl::run
 
  auto eventOutcome = co_await donePromise.exclusiveJoin(ioContext.onAbort()).then([&]() {
  return ioContext.waitUntilStatus();
- }, [](kj::Exception&& e) {
- if (e.getDetail(TAIL_STREAM_JS_FAILURE) != kj::none) {
+ }, [&incomingRequest](kj::Exception&& e) {
+ // If we have a JSG exception, just set the appropriate return code – this will already have
+ // been logged and we do not need to treat it like a KJ exception. Otherwise, re-throw the
+ // exception.
+ if (jsg::isTunneledException(e.getDescription())) {
+ incomingRequest->getMetrics().reportFailure(e);
  return EventOutcome::EXCEPTION;
  }
  kj::throwRecoverableException(kj::mv(e));