text: // Copyright (c) 2013-2014 Sandstorm Development Group, Inc. and contributors
  // Licensed under the MIT License:
  //
  // Permission is hereby granted, free of charge, to any person obtaining a copy
  // of this software and associated documentation files (the "Software"), to deal
  // in the Software without restriction, including without limitation the rights
  // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  // copies of the Software, and to permit persons to whom the Software is
  // furnished to do so, subject to the following conditions:
  //
  // The above copyright notice and this permission notice shall be included in
  // all copies or substantial portions of the Software.
  //
  // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  // THE SOFTWARE.
  
  #include "rpc.h"
  #include "message.h"
  #include "kj/debug.h"
  #include "kj/vector.h"
  #include "kj/async.h"
  #include "kj/one-of.h"
  #include "kj/function.h"
  #include <functional>  // std::greater
  #include <unordered_map>
  #include <map>
  #include <queue>
  #include "capnp/rpc.capnp.h"
  #include "kj/io.h"
  #include "kj/map.h"
  
  namespace capnp {
  namespace _ {  // private
  
  namespace {
  
  template <typename T>
  inline constexpr uint messageSizeHint() {
    return 1 + sizeInWords<rpc::Message>() + sizeInWords<T>();
  }
  template <>
  inline constexpr uint messageSizeHint<void>() {
    return 1 + sizeInWords<rpc::Message>();
  }
  
  constexpr const uint MESSAGE_TARGET_SIZE_HINT = sizeInWords<rpc::MessageTarget>() +
      sizeInWords<rpc::PromisedAnswer>() + 16;  // +16 for ops; hope that's enough
  
  constexpr const uint CAP_DESCRIPTOR_SIZE_HINT = sizeInWords<rpc::CapDescriptor>() +
      sizeInWords<rpc::PromisedAnswer>();
  
  constexpr const uint64_t MAX_SIZE_HINT = 1 << 20;
  
  uint copySizeHint(MessageSize size) {
    uint64_t sizeHint = size.wordCount + size.capCount * CAP_DESCRIPTOR_SIZE_HINT
                      // if capCount > 0, the cap descriptor list has a 1-word tag
                      + (size.capCount > 0);
    return kj::min(MAX_SIZE_HINT, sizeHint);
  }
  
  uint firstSegmentSize(kj::Maybe<MessageSize> sizeHint, uint additional) {
    KJ_IF_MAYBE(s, sizeHint) {
      return copySizeHint(*s) + additional;
    } else {
      return 0;
    }
  }
  
  kj::Maybe<kj::Array<PipelineOp>> toPipelineOps(List<rpc::PromisedAnswer::Op>::Reader ops) {
    auto result = kj::heapArrayBuilder<PipelineOp>(ops.size());
    for (auto opReader: ops) {
      PipelineOp op;
      switch (opReader.which()) {
        case rpc::PromisedAnswer::Op::NOOP:
          op.type = PipelineOp::NOOP;
          break;
        case rpc::PromisedAnswer::Op::GET_POINTER_FIELD:
          op.type = PipelineOp::GET_POINTER_FIELD;
          op.pointerIndex = opReader.getGetPointerField();
          break;
        default:
          KJ_FAIL_REQUIRE("Unsupported pipeline op.", (uint)opReader.which()) {
            return nullptr;
          }
      }
      result.add(op);
    }
    return result.finish();
  }
  
  Orphan<List<rpc::PromisedAnswer::Op>> fromPipelineOps(
      Orphanage orphanage, kj::ArrayPtr<const PipelineOp> ops) {
    auto result = orphanage.newOrphan<List<rpc::PromisedAnswer::Op>>(ops.size());
    auto builder = result.get();
    for (uint i: kj::indices(ops)) {
      rpc::PromisedAnswer::Op::Builder opBuilder = builder[i];
      switch (ops[i].type) {
        case PipelineOp::NOOP:
          opBuilder.setNoop();
          break;
        case PipelineOp::GET_POINTER_FIELD:
          opBuilder.setGetPointerField(ops[i].pointerIndex);
          break;
      }
    }
    return result;
  }
  
  kj::Exception toException(const rpc::Exception::Reader& exception) {
    auto reason = [&]() {
      if (exception.getReason().startsWith("remote exception: ")) {
        return kj::str(exception.getReason());
      } else {
        return kj::str("remote exception: ", exception.getReason());
      }
    }();
  
    kj::Exception result(static_cast<kj::Exception::Type>(exception.getType()),
        "(remote)", 0, kj::mv(reason));
    if (exception.hasTrace()) {
      result.setRemoteTrace(kj::str(exception.getTrace()));
    }
    return result;
  }
  
  void fromException(const kj::Exception& exception, rpc::Exception::Builder builder,
                     kj::Maybe<kj::Function<kj::String(const kj::Exception&)>&> traceEncoder) {
    kj::StringPtr description = exception.getDescription();
  
    // Include context, if any.
    kj::Vector<kj::String> contextLines;
    for (auto context = exception.getContext();;) {
      KJ_IF_MAYBE(c, context) {
        contextLines.add(kj::str("context: ", c->file, ": ", c->line, ": ", c->description));
        context = c->next;
      } else {
        break;
      }
    }
    kj::String scratch;
    if (contextLines.size() > 0) {
      scratch = kj::str(description, '\n', kj::strArray(contextLines, "\n"));
      description = scratch;
    }
  
    builder.setReason(description);
    builder.setType(static_cast<rpc::Exception::Type>(exception.getType()));
  
    KJ_IF_MAYBE(t, traceEncoder) {
      builder.setTrace((*t)(exception));
    }
  
    if (exception.getType() == kj::Exception::Type::FAILED &&
        !exception.getDescription().startsWith("remote exception:")) {
      KJ_LOG(INFO, "returning failure over rpc", exception);
    }
  }
  
  uint exceptionSizeHint(const kj::Exception& exception) {
    return sizeInWords<rpc::Exception>() + exception.getDescription().size() / sizeof(word) + 1;
  }
  
  // =======================================================================================
  
  template <typename Id, typename T>
  class ExportTable {
    // Table mapping integers to T, where the integers are chosen locally.
  
  public:
    kj::Maybe<T&> find(Id id) {
      if (id < slots.size() && slots[id] != nullptr) {
        return slots[id];
      } else {
        return nullptr;
      }
    }
  
    T erase(Id id, T& entry) {
      // Remove an entry from the table and return it.  We return it so that the caller can be
      // careful to release it (possibly invoking arbitrary destructors) at a time that makes sense.
      // `entry` is a reference to the entry being released -- we require this in order to prove
      // that the caller has already done a find() to check that this entry exists.  We can't check
      // ourselves because the caller may have nullified the entry in the meantime.
      KJ_DREQUIRE(&entry == &slots[id]);
      T toRelease = kj::mv(slots[id]);
      slots[id] = T();
      freeIds.push(id);
      return toRelease;
    }
  
    T& next(Id& id) {
      if (freeIds.empty()) {
        id = slots.size();
        return slots.add();
      } else {
        id = freeIds.top();
        freeIds.pop();
        return slots[id];
      }
    }
  
    template <typename Func>
    void forEach(Func&& func) {
      for (Id i = 0; i < slots.size(); i++) {
        if (slots[i] != nullptr) {
          func(i, slots[i]);
        }
      }
    }
  
  private:
    kj::Vector<T> slots;
    std::priority_queue<Id, std::vector<Id>, std::greater<Id>> freeIds;
  };
  
  template <typename Id, typename T>
  class ImportTable {
    // Table mapping integers to T, where the integers are chosen remotely.
  
  public:
    T& operator[](Id id) {
      if (id < kj::size(low)) {
        return low[id];
      } else {
        return high[id];
      }
    }
  
    kj::Maybe<T&> find(Id id) {
      if (id < kj::size(low)) {
        return low[id];
      } else {
        auto iter = high.find(id);
        if (iter == high.end()) {
          return nullptr;
        } else {
          return iter->second;
        }
      }
    }
  
    T erase(Id id) {
      // Remove an entry from the table and return it.  We return it so that the caller can be
      // careful to release it (possibly invoking arbitrary destructors) at a time that makes sense.
      if (id < kj::size(low)) {
        T toRelease = kj::mv(low[id]);
        low[id] = T();
        return toRelease;
      } else {
        T toRelease = kj::mv(high[id]);
        high.erase(id);
        return toRelease;
      }
    }
  
    template <typename Func>
    void forEach(Func&& func) {
      for (Id i: kj::indices(low)) {
        func(i, low[i]);
      }
      for (auto& entry: high) {
        func(entry.first, entry.second);
      }
    }
  
  private:
    T low[16];
    std::unordered_map<Id, T> high;
  };
  
  // =======================================================================================
  
  class RpcConnectionState final: public kj::TaskSet::ErrorHandler, public kj::Refcounted {
  public:
    struct DisconnectInfo {
      kj::Promise<void> shutdownPromise;
      // Task which is working on sending an abort message and cleanly ending the connection.
    };
  
    RpcConnectionState(BootstrapFactoryBase& bootstrapFactory,
                       kj::Maybe<SturdyRefRestorerBase&> restorer,
                       kj::Own<VatNetworkBase::Connection>&& connectionParam,
                       kj::Own<kj::PromiseFulfiller<DisconnectInfo>>&& disconnectFulfiller,
                       size_t flowLimit,
                       kj::Maybe<kj::Function<kj::String(const kj::Exception&)>&> traceEncoder)
        : bootstrapFactory(bootstrapFactory),
          restorer(restorer), disconnectFulfiller(kj::mv(disconnectFulfiller)), flowLimit(flowLimit),
          traceEncoder(traceEncoder), tasks(*this) {
      connection.init<Connected>(kj::mv(connectionParam));
      tasks.add(messageLoop());
    }
  
    kj::Own<ClientHook> restore(AnyPointer::Reader objectId) {
      if (connection.is<Disconnected>()) {
        return newBrokenCap(kj::cp(connection.get<Disconnected>()));
      }
  
      QuestionId questionId;
      auto& question = questions.next(questionId);
  
      question.isAwaitingReturn = true;
  
      auto paf = kj::newPromiseAndFulfiller<kj::Promise<kj::Own<RpcResponse>>>();
  
      auto questionRef = kj::refcounted<QuestionRef>(*this, questionId, kj::mv(paf.fulfiller));
      question.selfRef = *questionRef;
  
      paf.promise = paf.promise.attach(kj::addRef(*questionRef));
  
      {
        auto message = connection.get<Connected>()->newOutgoingMessage(
            objectId.targetSize().wordCount + messageSizeHint<rpc::Bootstrap>());
  
        auto builder = message->getBody().initAs<rpc::Message>().initBootstrap();
        builder.setQuestionId(questionId);
        builder.getDeprecatedObjectId().set(objectId);
  
        message->send();
      }
  
      auto pipeline = kj::refcounted<RpcPipeline>(*this, kj::mv(questionRef), kj::mv(paf.promise));
  
      return pipeline->getPipelinedCap(kj::Array<const PipelineOp>(nullptr));
    }
  
    void taskFailed(kj::Exception&& exception) override {
      disconnect(kj::mv(exception));
    }
  
    void disconnect(kj::Exception&& exception) {
      // After disconnect(), the RpcSystem could be destroyed, making `traceEncoder` a dangling
      // reference, so null it out before we return from here. We don't need it anymore once
      // disconnected anyway.
      KJ_DEFER(traceEncoder = nullptr);
  
      if (!connection.is<Connected>()) {
        // Already disconnected.
        return;
      }
  
      kj::Exception networkException(kj::Exception::Type::DISCONNECTED,
          exception.getFile(), exception.getLine(), kj::heapString(exception.getDescription()));
  
      // Don't throw away the stack trace.
      if (exception.getRemoteTrace() != nullptr) {
        networkException.setRemoteTrace(kj::str(exception.getRemoteTrace()));
      }
      for (void* addr: exception.getStackTrace()) {
        networkException.addTrace(addr);
      }
      // If your stack trace points here, it means that the exception became the reason that the
      // RPC connection was disconnected. The exception was then thrown by all in-flight calls and
      // all future calls on this connection.
      networkException.addTraceHere();
  
      KJ_IF_MAYBE(newException, kj::runCatchingExceptions([&]() {
        // Carefully pull all the objects out of the tables prior to releasing them because their
        // destructors could come back and mess with the tables.
        kj::Vector<kj::Own<PipelineHook>> pipelinesToRelease;
        kj::Vector<kj::Own<ClientHook>> clientsToRelease;
        kj::Vector<kj::Promise<kj::Own<RpcResponse>>> tailCallsToRelease;
        kj::Vector<kj::Promise<void>> resolveOpsToRelease;
  
        // All current questions complete with exceptions.
        questions.forEach([&](QuestionId id, Question& question) {
          KJ_IF_MAYBE(questionRef, question.selfRef) {
            // QuestionRef still present.
            questionRef->reject(kj::cp(networkException));
          }
        });
  
        answers.forEach([&](AnswerId id, Answer& answer) {
          KJ_IF_MAYBE(p, answer.pipeline) {
            pipelinesToRelease.add(kj::mv(*p));
          }
  
          KJ_IF_MAYBE(promise, answer.redirectedResults) {
            tailCallsToRelease.add(kj::mv(*promise));
          }
  
          KJ_IF_MAYBE(context, answer.callContext) {
            context->requestCancel();
          }
        });
  
        exports.forEach([&](ExportId id, Export& exp) {
          clientsToRelease.add(kj::mv(exp.clientHook));
          KJ_IF_MAYBE(op, exp.resolveOp) {
            resolveOpsToRelease.add(kj::mv(*op));
          }
          exp = Export();
        });
  
        imports.forEach([&](ImportId id, Import& import) {
          KJ_IF_MAYBE(f, import.promiseFulfiller) {
            f->get()->reject(kj::cp(networkException));
          }
        });
  
        embargoes.forEach([&](EmbargoId id, Embargo& embargo) {
          KJ_IF_MAYBE(f, embargo.fulfiller) {
            f->get()->reject(kj::cp(networkException));
          }
        });
      })) {
        // Some destructor must have thrown an exception.  There is no appropriate place to report
        // these errors.
        KJ_LOG(ERROR, "Uncaught exception when destroying capabilities dropped by disconnect.",
               *newException);
      }
  
      // Send an abort message, but ignore failure.
      kj::runCatchingExceptions([&]() {
        auto message = connection.get<Connected>()->newOutgoingMessage(
            messageSizeHint<void>() + exceptionSizeHint(exception));
        fromException(exception, message->getBody().getAs<rpc::Message>().initAbort());
        message->send();
      });
  
      // Indicate disconnect.
      auto shutdownPromise = connection.get<Connected>()->shutdown()
          .attach(kj::mv(connection.get<Connected>()))
          .then([]() -> kj::Promise<void> { return kj::READY_NOW; },
                [origException = kj::mv(exception)](kj::Exception&& e) -> kj::Promise<void> {
            // Don't report disconnects as an error.
            if (e.getType() == kj::Exception::Type::DISCONNECTED) {
              return kj::READY_NOW;
            }
            // If the error is just what was passed in to disconnect(), don't report it back out
            // since it shouldn't be anything the caller doesn't already know about.
            if (e.getType() == origException.getType() &&
                e.getDescription() == origException.getDescription()) {
              return kj::READY_NOW;
            }
            return kj::mv(e);
          });
      disconnectFulfiller->fulfill(DisconnectInfo { kj::mv(shutdownPromise) });
      connection.init<Disconnected>(kj::mv(networkException));
      canceler.cancel(networkException);
    }
  
    void setFlowLimit(size_t words) {
      flowLimit = words;
      maybeUnblockFlow();
    }
  
  private:
    class RpcClient;
    class ImportClient;
    class PromiseClient;
    class QuestionRef;
    class RpcPipeline;
    class RpcCallContext;
    class RpcResponse;
  
    // =======================================================================================
    // The Four Tables entry types
    //
    // We have to define these before we can define the class's fields.
  
    typedef uint32_t QuestionId;
    typedef QuestionId AnswerId;
    typedef uint32_t ExportId;
    typedef ExportId ImportId;
    // See equivalent definitions in rpc.capnp.
    //
    // We always use the type that refers to the local table of the same name.  So e.g. although
    // QuestionId and AnswerId are the same type, we use QuestionId when referring to an entry in
    // the local question table (which corresponds to the peer's answer table) and use AnswerId
    // to refer to an entry in our answer table (which corresponds to the peer's question table).
    // Since all messages in the RPC protocol are defined from the sender's point of view, this
    // means that any time we read an ID from a received message, its type should invert.
    // TODO(cleanup):  Perhaps we could enforce that in a type-safe way?  Hmm...
  
    struct Question {
      kj::Array<ExportId> paramExports;
      // List of exports that were sent in the request.  If the response has `releaseParamCaps` these
      // will need to be released.
  
      kj::Maybe<QuestionRef&> selfRef;
      // The local QuestionRef, set to nullptr when it is destroyed, which is also when `Finish` is
      // sent.
  
      bool isAwaitingReturn = false;
      // True from when `Call` is sent until `Return` is received.
  
      bool isTailCall = false;
      // Is this a tail call?  If so, we don't expect to receive results in the `Return`.
  
      bool skipFinish = false;
      // If true, don't send a Finish message.
  
      inline bool operator==(decltype(nullptr)) const {
        return !isAwaitingReturn && selfRef == nullptr;
      }
      inline bool operator!=(decltype(nullptr)) const { return !operator==(nullptr); }
    };
  
    struct Answer {
      Answer() = default;
      Answer(const Answer&) = delete;
      Answer(Answer&&) = default;
      Answer& operator=(Answer&&) = default;
      // If we don't explicitly write all this, we get some stupid error deep in STL.
  
      bool active = false;
      // True from the point when the Call message is received to the point when both the `Finish`
      // message has been received and the `Return` has been sent.
  
      kj::Maybe<kj::Own<PipelineHook>> pipeline;
      // Send pipelined calls here.  Becomes null as soon as a `Finish` is received.
  
      kj::Maybe<kj::Promise<kj::Own<RpcResponse>>> redirectedResults;
      // For locally-redirected calls (Call.sendResultsTo.yourself), this is a promise for the call
      // result, to be picked up by a subsequent `Return`.
  
      kj::Maybe<RpcCallContext&> callContext;
      // The call context, if it's still active.  Becomes null when the `Return` message is sent.
      // This object, if non-null, is owned by `asyncOp`.
  
      kj::Array<ExportId> resultExports;
      // List of exports that were sent in the results.  If the finish has `releaseResultCaps` these
      // will need to be released.
    };
  
    struct Export {
      uint refcount = 0;
      // When this reaches 0, drop `clientHook` and free this export.
  
      kj::Own<ClientHook> clientHook;
  
      kj::Maybe<kj::Promise<void>> resolveOp = nullptr;
      // If this export is a promise (not a settled capability), the `resolveOp` represents the
      // ongoing operation to wait for that promise to resolve and then send a `Resolve` message.
  
      inline bool operator==(decltype(nullptr)) const { return refcount == 0; }
      inline bool operator!=(decltype(nullptr)) const { return refcount != 0; }
    };
  
    struct Import {
      Import() = default;
      Import(const Import&) = delete;
      Import(Import&&) = default;
      Import& operator=(Import&&) = default;
      // If we don't explicitly write all this, we get some stupid error deep in STL.
  
      kj::Maybe<ImportClient&> importClient;
      // Becomes null when the import is destroyed.
  
      kj::Maybe<RpcClient&> appClient;
      // Either a copy of importClient, or, in the case of promises, the wrapping PromiseClient.
      // Becomes null when it is discarded *or* when the import is destroyed (e.g. the promise is
      // resolved and the import is no longer needed).
  
      kj::Maybe<kj::Own<kj::PromiseFulfiller<kj::Own<ClientHook>>>> promiseFulfiller;
      // If non-null, the import is a promise.
    };
  
    typedef uint32_t EmbargoId;
  
    struct Embargo {
      // For handling the `Disembargo` message when looping back to self.
  
      kj::Maybe<kj::Own<kj::PromiseFulfiller<void>>> fulfiller;
      // Fulfill this when the Disembargo arrives.
  
      inline bool operator==(decltype(nullptr)) const { return fulfiller == nullptr; }
      inline bool operator!=(decltype(nullptr)) const { return fulfiller != nullptr; }
    };
  
    // =======================================================================================
    // OK, now we can define RpcConnectionState's member data.
  
    BootstrapFactoryBase& bootstrapFactory;
    kj::Maybe<SturdyRefRestorerBase&> restorer;
  
    typedef kj::Own<VatNetworkBase::Connection> Connected;
    typedef kj::Exception Disconnected;
    kj::OneOf<Connected, Disconnected> connection;
    // Once the connection has failed, we drop it and replace it with an exception, which will be
    // thrown from all further calls.
  
    kj::Canceler canceler;
    // Will be canceled if and when `connection` is changed from `Connected` to `Disconnected`.
    // TODO(cleanup): `Connected` should be a struct that contains the connection and the Canceler,
    //   but that's more refactoring than I want to do right now.
  
    kj::Own<kj::PromiseFulfiller<DisconnectInfo>> disconnectFulfiller;
  
    ExportTable<ExportId, Export> exports;
    ExportTable<QuestionId, Question> questions;
    ImportTable<AnswerId, Answer> answers;
    ImportTable<ImportId, Import> imports;
    // The Four Tables!
    // The order of the tables is important for correct destruction.
  
    std::unordered_map<ClientHook*, ExportId> exportsByCap;
    // Maps already-exported ClientHook objects to their ID in the export table.
  
    ExportTable<EmbargoId, Embargo> embargoes;
    // There are only four tables.  This definitely isn't a fifth table.  I don't know what you're
    // talking about.
  
    size_t flowLimit;
    size_t callWordsInFlight = 0;
  
    kj::Maybe<kj::Own<kj::PromiseFulfiller<void>>> flowWaiter;
    // If non-null, we're currently blocking incoming messages waiting for callWordsInFlight to drop
    // below flowLimit. Fulfill this to un-block.
  
    kj::Maybe<kj::Function<kj::String(const kj::Exception&)>&> traceEncoder;
  
    kj::TaskSet tasks;
  
    // =====================================================================================
    // ClientHook implementations
  
    class RpcClient: public ClientHook, public kj::Refcounted {
    public:
      RpcClient(RpcConnectionState& connectionState)
          : connectionState(kj::addRef(connectionState)) {}
  
      virtual kj::Maybe<ExportId> writeDescriptor(rpc::CapDescriptor::Builder descriptor,
                                                  kj::Vector<int>& fds) = 0;
      // Writes a CapDescriptor referencing this client.  The CapDescriptor must be sent as part of
      // the very next message sent on the connection, as it may become invalid if other things
      // happen.
      //
      // If writing the descriptor adds a new export to the export table, or increments the refcount
      // on an existing one, then the ID is returned and the caller is responsible for removing it
      // later.
  
      virtual kj::Maybe<kj::Own<ClientHook>> writeTarget(
          rpc::MessageTarget::Builder target) = 0;
      // Writes the appropriate call target for calls to this capability and returns null.
      //
      // - OR -
      //
      // If calls have been redirected to some other local ClientHook, returns that hook instead.
      // This can happen if the capability represents a promise that has been resolved.
  
      virtual kj::Own<ClientHook> getInnermostClient() = 0;
      // If this client just wraps some other client -- even if it is only *temporarily* wrapping
      // that other client -- return a reference to the other client, transitively.  Otherwise,
      // return a new reference to *this.
  
      virtual void adoptFlowController(kj::Own<RpcFlowController> flowController) {
        // Called when a PromiseClient resolves to another RpcClient. If streaming calls were
        // outstanding on the old client, we'd like to keep using the same FlowController on the new
        // client, so as to keep the flow steady.
  
        if (this->flowController == nullptr) {
          // We don't have any existing flowController so we can adopt this one, yay!
          this->flowController = kj::mv(flowController);
        } else {
          // Apparently, there is an existing flowController. This is an unusual scenario: Apparently
          // we had two stream capabilities, we were streaming to both of them, and they later
          // resolved to the same capability. This probably never happens because streaming use cases
          // normally call for there to be only one client. But, it's certainly possible, and we need
          // to handle it. We'll do the conservative thing and just make sure that all the calls
          // finish. This may mean we'll over-buffer temporarily; oh well.
          connectionState->tasks.add(flowController->waitAllAcked().attach(kj::mv(flowController)));
        }
      }
  
      // implements ClientHook -----------------------------------------
  
      Request<AnyPointer, AnyPointer> newCall(
          uint64_t interfaceId, uint16_t methodId, kj::Maybe<MessageSize> sizeHint) override {
        return newCallNoIntercept(interfaceId, methodId, sizeHint);
      }
  
      Request<AnyPointer, AnyPointer> newCallNoIntercept(
          uint64_t interfaceId, uint16_t methodId, kj::Maybe<MessageSize> sizeHint) {
        if (!connectionState->connection.is<Connected>()) {
          return newBrokenRequest(kj::cp(connectionState->connection.get<Disconnected>()), sizeHint);
        }
  
        auto request = kj::heap<RpcRequest>(
            *connectionState, *connectionState->connection.get<Connected>(),
            sizeHint, kj::addRef(*this));
        auto callBuilder = request->getCall();
  
        callBuilder.setInterfaceId(interfaceId);
        callBuilder.setMethodId(methodId);
  
        auto root = request->getRoot();
        return Request<AnyPointer, AnyPointer>(root, kj::mv(request));
      }
  
      VoidPromiseAndPipeline call(uint64_t interfaceId, uint16_t methodId,
                                  kj::Own<CallContextHook>&& context) override {
        return callNoIntercept(interfaceId, methodId, kj::mv(context));
      }
  
      VoidPromiseAndPipeline callNoIntercept(uint64_t interfaceId, uint16_t methodId,
                                             kj::Own<CallContextHook>&& context) {
        // Implement call() by copying params and results messages.
  
        auto params = context->getParams();
        auto request = newCallNoIntercept(interfaceId, methodId, params.targetSize());
  
        request.set(params);
        context->releaseParams();
  
        // We can and should propagate cancellation.
        context->allowCancellation();
  
        return context->directTailCall(RequestHook::from(kj::mv(request)));
      }
  
      kj::Own<ClientHook> addRef() override {
        return kj::addRef(*this);
      }
      const void* getBrand() override {
        return connectionState.get();
      }
  
      kj::Own<RpcConnectionState> connectionState;
  
      kj::Maybe<kj::Own<RpcFlowController>> flowController;
      // Becomes non-null the first time a streaming call is made on this capability.
    };
  
    class ImportClient final: public RpcClient {
      // A ClientHook that wraps an entry in the import table.
  
    public:
      ImportClient(RpcConnectionState& connectionState, ImportId importId,
                   kj::Maybe<kj::AutoCloseFd> fd)
          : RpcClient(connectionState), importId(importId), fd(kj::mv(fd)) {}
  
      ~ImportClient() noexcept(false) {
        unwindDetector.catchExceptionsIfUnwinding([&]() {
          // Remove self from the import table, if the table is still pointing at us.
          KJ_IF_MAYBE(import, connectionState->imports.find(importId)) {
            KJ_IF_MAYBE(i, import->importClient) {
              if (i == this) {
                connectionState->imports.erase(importId);
              }
            }
          }
  
          // Send a message releasing our remote references.
          if (remoteRefcount > 0 && connectionState->connection.is<Connected>()) {
            auto message = connectionState->connection.get<Connected>()->newOutgoingMessage(
                messageSizeHint<rpc::Release>());
            rpc::Release::Builder builder = message->getBody().initAs<rpc::Message>().initRelease();
            builder.setId(importId);
            builder.setReferenceCount(remoteRefcount);
            message->send();
          }
        });
      }
  
      void setFdIfMissing(kj::Maybe<kj::AutoCloseFd> newFd) {
        if (fd == nullptr) {
          fd = kj::mv(newFd);
        }
      }
  
      void addRemoteRef() {
        // Add a new RemoteRef and return a new ref to this client representing it.
        ++remoteRefcount;
      }
  
      kj::Maybe<ExportId> writeDescriptor(rpc::CapDescriptor::Builder descriptor,
                                          kj::Vector<int>& fds) override {
        descriptor.setReceiverHosted(importId);
        return nullptr;
      }
  
      kj::Maybe<kj::Own<ClientHook>> writeTarget(
          rpc::MessageTarget::Builder target) override {
        target.setImportedCap(importId);
        return nullptr;
      }
  
      kj::Own<ClientHook> getInnermostClient() override {
        return kj::addRef(*this);
      }
  
      // implements ClientHook -----------------------------------------
  
      kj::Maybe<ClientHook&> getResolved() override {
        return nullptr;
      }
  
      kj::Maybe<kj::Promise<kj::Own<ClientHook>>> whenMoreResolved() override {
        return nullptr;
      }
  
      kj::Maybe<int> getFd() override {
        return fd.map([](auto& f) { return f.get(); });
      }
  
    private:
      ImportId importId;
      kj::Maybe<kj::AutoCloseFd> fd;
  
      uint remoteRefcount = 0;
      // Number of times we've received this import from the peer.
  
      kj::UnwindDetector unwindDetector;
    };
  
    class PipelineClient final: public RpcClient {
      // A ClientHook representing a pipelined promise.  Always wrapped in PromiseClient.
  
    public:
      PipelineClient(RpcConnectionState& connectionState,
                     kj::Own<QuestionRef>&& questionRef,
                     kj::Array<PipelineOp>&& ops)
          : RpcClient(connectionState), questionRef(kj::mv(questionRef)), ops(kj::mv(ops)) {}
  
     kj::Maybe<ExportId> writeDescriptor(rpc::CapDescriptor::Builder descriptor,
                                         kj::Vector<int>& fds) override {
        auto promisedAnswer = descriptor.initReceiverAnswer();
        promisedAnswer.setQuestionId(questionRef->getId());
        promisedAnswer.adoptTransform(fromPipelineOps(
            Orphanage::getForMessageContaining(descriptor), ops));
        return nullptr;
      }
  
      kj::Maybe<kj::Own<ClientHook>> writeTarget(
          rpc::MessageTarget::Builder target) override {
        auto builder = target.initPromisedAnswer();
        builder.setQuestionId(questionRef->getId());
        builder.adoptTransform(fromPipelineOps(Orphanage::getForMessageContaining(builder), ops));
        return nullptr;
      }
  
      kj::Own<ClientHook> getInnermostClient() override {
        return kj::addRef(*this);
      }
  
      // implements ClientHook -----------------------------------------
  
      kj::Maybe<ClientHook&> getResolved() override {
        return nullptr;
      }
  
      kj::Maybe<kj::Promise<kj::Own<ClientHook>>> whenMoreResolved() override {
        return nullptr;
      }
  
      kj::Maybe<int> getFd() override {
        return nullptr;
      }
  
    private:
      kj::Own<QuestionRef> questionRef;
      kj::Array<PipelineOp> ops;
    };
  
    class PromiseClient final: public RpcClient {
      // A ClientHook that initially wraps one client (in practice, an ImportClient or a
      // PipelineClient) and then, later on, redirects to some other client.
  
    public:
      PromiseClient(RpcConnectionState& connectionState,
                    kj::Own<RpcClient> initial,
                    kj::Promise<kj::Own<ClientHook>> eventual,
                    kj::Maybe<ImportId> importId)
          : RpcClient(connectionState),
            cap(kj::mv(initial)),
            importId(importId),
            fork(eventual.then(
                [this](kj::Own<ClientHook>&& resolution) {
                  return resolve(kj::mv(resolution));
                }, [this](kj::Exception&& exception) {
                  return resolve(newBrokenCap(kj::mv(exception)));
                }).catch_([&](kj::Exception&& e) {
                  // Make any exceptions thrown from resolve() go to the connection's TaskSet which
                  // will cause the connection to be terminated.
                  connectionState.tasks.add(kj::cp(e));
                  return newBrokenCap(kj::mv(e));
                }).fork()) {}
      // Create a client that starts out forwarding all calls to `initial` but, once `eventual`
      // resolves, will forward there instead.
  
      ~PromiseClient() noexcept(false) {
        KJ_IF_MAYBE(id, importId) {
          // This object is representing an import promise.  That means the import table may still
          // contain a pointer back to it.  Remove that pointer.  Note that we have to verify that
          // the import still exists and the pointer still points back to this object because this
          // object may actually outlive the import.
          KJ_IF_MAYBE(import, connectionState->imports.find(*id)) {
            KJ_IF_MAYBE(c, import->appClient) {
              if (c == this) {
                import->appClient = nullptr;
              }
            }
          }
        }
      }
  
      kj::Maybe<ExportId> writeDescriptor(rpc::CapDescriptor::Builder descriptor,
                                          kj::Vector<int>& fds) override {
        receivedCall = true;
        return connectionState->writeDescriptor(*cap, descriptor, fds);
      }
  
      kj::Maybe<kj::Own<ClientHook>> writeTarget(
          rpc::MessageTarget::Builder target) override {
        receivedCall = true;
        return connectionState->writeTarget(*cap, target);
      }
  
      kj::Own<ClientHook> getInnermostClient() override {
        receivedCall = true;
        return connectionState->getInnermostClient(*cap);
      }
  
      void adoptFlowController(kj::Own<RpcFlowController> flowController) override {
        if (cap->getBrand() == connectionState.get()) {
          // Pass the flow controller on to our inner cap.
          kj::downcast<RpcClient>(*cap).adoptFlowController(kj::mv(flowController));
        } else {
          // We resolved to a capability that isn't another RPC capability. We should simply make
          // sure that all the calls complete.
          connectionState->tasks.add(flowController->waitAllAcked().attach(kj::mv(flowController)));
        }
      }
  
      // implements ClientHook -----------------------------------------
  
      Request<AnyPointer, AnyPointer> newCall(
          uint64_t interfaceId, uint16_t methodId, kj::Maybe<MessageSize> sizeHint) override {
        receivedCall = true;
  
        // IMPORTANT: We must call our superclass's version of newCall(), NOT cap->newCall(), because
        //   the Request object we create needs to check at send() time whether the promise has
        //   resolved and, if so, redirect to the new target.
        return RpcClient::newCall(interfaceId, methodId, sizeHint);
      }
  
      VoidPromiseAndPipeline call(uint64_t interfaceId, uint16_t methodId,
                                  kj::Own<CallContextHook>&& context) override {
        receivedCall = true;
        return cap->call(interfaceId, methodId, kj::mv(context));
      }
  
      kj::Maybe<ClientHook&> getResolved() override {
        if (isResolved()) {
          return *cap;
        } else {
          return nullptr;
        }
      }
  
      kj::Maybe<kj::Promise<kj::Own<ClientHook>>> whenMoreResolved() override {
        return fork.addBranch();
      }
  
      kj::Maybe<int> getFd() override {
        if (isResolved()) {
          return cap->getFd();
        } else {
          // In theory, before resolution, the ImportClient for the promise could have an FD
          // attached, if the promise itself was presented with an attached FD. However, we can't
          // really return that one here because it may be closed when we get the Resolve message
          // later. In theory we could have the PromiseClient itself take ownership of an FD that
          // arrived attached to a promise cap, but the use case for that is questionable. I'm
          // keeping it simple for now.
          return nullptr;
        }
      }
  
    private:
      kj::Own<ClientHook> cap;
  
      kj::Maybe<ImportId> importId;
      kj::ForkedPromise<kj::Own<ClientHook>> fork;
  
      bool receivedCall = false;
  
      enum {
        UNRESOLVED,
        // Not resolved at all yet.
  
        REMOTE,
        // Remote promise resolved to a remote settled capability (or null/error).
  
        REFLECTED,
        // Remote promise resolved to one of our own exports.
  
        MERGED,
        // Remote promise resolved to another remote promise which itself wasn't resolved yet, so we
        // merged them. In this case, `cap` is guaranteed to point to another PromiseClient.
  
        BROKEN
        // Resolved to null or error.
      } resolutionType = UNRESOLVED;
  
      inline bool isResolved() {
        return resolutionType != UNRESOLVED;
      }
  
      kj::Promise<kj::Own<ClientHook>> resolve(kj::Own<ClientHook> replacement) {
        KJ_DASSERT(!isResolved());
  
        const void* replacementBrand = replacement->getBrand();
        bool isSameConnection = replacementBrand == connectionState.get();
        if (isSameConnection) {
          // We resolved to some other RPC capability hosted by the same peer.
          KJ_IF_MAYBE(promise, replacement->whenMoreResolved()) {
            // We resolved to another remote promise. If *that* promise eventually resolves back
            // to us, we'll need a disembargo. Possibilities:
            // 1. The other promise hasn't resolved at all yet. In that case we can simply set its
            //    `receivedCall` flag and let it handle the disembargo later.
            // 2. The other promise has received a Resolve message and decided to initiate a
            //    disembargo which it is still waiting for. In that case we will certainly also need
            //    a disembargo for the same reason that the other promise did. And, we can't simply
            //    wait for their disembargo; we need to start a new one of our own.
            // 3. The other promise has resolved already (with or without a disembargo). In this
            //    case we should treat it as if we resolved directly to the other promise's result,
            //    possibly requiring a disembargo under the same conditions.
  
            // We know the other object is a PromiseClient because it's the only ClientHook
            // type in the RPC implementation which returns non-null for `whenMoreResolved()`.
            PromiseClient* other = &kj::downcast<PromiseClient>(*replacement);
            while (other->resolutionType == MERGED) {
              // There's no need to resolve to a thing that's just going to resolve to another thing.
              replacement = other->cap->addRef();
              other = &kj::downcast<PromiseClient>(*replacement);
  
              // Note that replacementBrand is unchanged since we'd only merge with other
              // PromiseClients on the same connection.
              KJ_DASSERT(replacement->getBrand() == replacementBrand);
            }
  
            if (other->isResolved()) {
              // The other capability resolved already. If it determined that it resolved as
              // reflected, then we determine the same.
              resolutionType = other->resolutionType;
            } else {
              // The other capability hasn't resolved yet, so we can safely merge with it and do a
              // single combined disembargo if needed later.
              other->receivedCall = other->receivedCall || receivedCall;
              resolutionType = MERGED;
            }
          } else {
            resolutionType = REMOTE;
          }
        } else {
          if (replacementBrand == &ClientHook::NULL_CAPABILITY_BRAND ||
              replacementBrand == &ClientHook::BROKEN_CAPABILITY_BRAND) {
            // We don't consider null or broken capabilities as "reflected" because they may have
            // been communicated to us literally as a null pointer or an exception on the wire,
            // rather than as a reference to one of our exports, in which case a disembargo won't
            // work. But also, call ordering is completely irrelevant with these so there's no need
            // to disembargo anyway.
            resolutionType = BROKEN;
          } else {
            resolutionType = REFLECTED;
          }
        }
  
        // Every branch above ends by setting resolutionType to something other than UNRESOLVED.
        KJ_DASSERT(isResolved());
  
        // If the original capability was used for streaming calls, it will have a
        // `flowController` that might still be shepherding those calls. We'll need make sure that
        // it doesn't get thrown away. Note that we know that *cap is an RpcClient because resolve()
        // is only called once and our constructor required that the initial capability is an
        // RpcClient.
        KJ_IF_MAYBE(f, kj::downcast<RpcClient>(*cap).flowController) {
          if (isSameConnection) {
            // The new target is on the same connection. It would make a lot of sense to keep using
            // the same flow controller if possible.
            kj::downcast<RpcClient>(*replacement).adoptFlowController(kj::mv(*f));
          } else {
            // The new target is something else. The best we can do is wait for the controller to
            // drain. New calls will be flow-controlled in a new way without knowing about the old
            // controller.
            connectionState->tasks.add(f->get()->waitAllAcked().attach(kj::mv(*f)));
          }
        }
  
        if (resolutionType == REFLECTED && receivedCall &&
            connectionState->connection.is<Connected>()) {
          // The new capability is hosted locally, not on the remote machine.  And, we had made calls
          // to the promise.  We need to make sure those calls echo back to us before we allow new
          // calls to go directly to the local capability, so we need to set a local embargo and send
          // a `Disembargo` to echo through the peer.
  
          auto message = connectionState->connection.get<Connected>()->newOutgoingMessage(
              messageSizeHint<rpc::Disembargo>() + MESSAGE_TARGET_SIZE_HINT);
  
          auto disembargo = message->getBody().initAs<rpc::Message>().initDisembargo();
  
          {
            auto redirect = connectionState->writeTarget(*cap, disembargo.initTarget());
            KJ_ASSERT(redirect == nullptr,
                      "Original promise target should always be from this RPC connection.");
          }
  
          EmbargoId embargoId;
          Embargo& embargo = connectionState->embargoes.next(embargoId);
  
          disembargo.getContext().setSenderLoopback(embargoId);
  
          auto paf = kj::newPromiseAndFulfiller<void>();
          embargo.fulfiller = kj::mv(paf.fulfiller);
  
          // Make a promise which resolves to `replacement` as soon as the `Disembargo` comes back.
          auto embargoPromise = paf.promise.then([replacement = kj::mv(replacement)]() mutable {
            return kj::mv(replacement);
          });
  
          // We need to queue up calls in the meantime, so we'll resolve ourselves to a local promise
          // client instead.
          replacement = newLocalPromiseClient(kj::mv(embargoPromise));
  
          // Send the `Disembargo`.
          message->send();
        }
  
        cap = replacement->addRef();
  
        return kj::mv(replacement);
      }
    };
  
    kj::Maybe<ExportId> writeDescriptor(ClientHook& cap, rpc::CapDescriptor::Builder descriptor,
                                        kj::Vector<int>& fds) {
      // Write a descriptor for the given capability.
  
      // Find the innermost wrapped capability.
      ClientHook* inner = &cap;
      for (;;) {
        KJ_IF_MAYBE(resolved, inner->getResolved()) {
          inner = resolved;
        } else {
          break;
        }
      }
  
      KJ_IF_MAYBE(fd, inner->getFd()) {
        descriptor.setAttachedFd(fds.size());
        fds.add(kj::mv(*fd));
      }
  
      if (inner->getBrand() == this) {
        return kj::downcast<RpcClient>(*inner).writeDescriptor(descriptor, fds);
      } else {
        auto iter = exportsByCap.find(inner);
        if (iter != exportsByCap.end()) {
          // We've already seen and exported this capability before.  Just up the refcount.
          auto& exp = KJ_ASSERT_NONNULL(exports.find(iter->second));
          ++exp.refcount;
          if (exp.resolveOp == nullptr) {
            descriptor.setSenderHosted(iter->second);
          } else {
            descriptor.setSenderPromise(iter->second);
          }
          return iter->second;
        } else {
          // This is the first time we've seen this capability.
          ExportId exportId;
          auto& exp = exports.next(exportId);
          exportsByCap[inner] = exportId;
          exp.refcount = 1;
          exp.clientHook = inner->addRef();
  
          KJ_IF_MAYBE(wrapped, inner->whenMoreResolved()) {
            // This is a promise.  Arrange for the `Resolve` message to be sent later.
            exp.resolveOp = resolveExportedPromise(exportId, kj::mv(*wrapped));
            descriptor.setSenderPromise(exportId);
          } else {
            descriptor.setSenderHosted(exportId);
          }
  
          return exportId;
        }
      }
    }
  
    kj::Array<ExportId> writeDescriptors(kj::ArrayPtr<kj::Maybe<kj::Own<ClientHook>>> capTable,
                                         rpc::Payload::Builder payload, kj::Vector<int>& fds) {
      if (capTable.size() == 0) {
        // Calling initCapTable(0) will still allocate a 1-word tag, which we'd like to avoid...
        return nullptr;
      }
  
      auto capTableBuilder = payload.initCapTable(capTable.size());
      kj::Vector<ExportId> exports(capTable.size());
      for (uint i: kj::indices(capTable)) {
        KJ_IF_MAYBE(cap, capTable[i]) {
          KJ_IF_MAYBE(exportId, writeDescriptor(**cap, capTableBuilder[i], fds)) {
            exports.add(*exportId);
          }
        } else {
          capTableBuilder[i].setNone();
        }
      }
      return exports.releaseAsArray();
    }
  
    kj::Maybe<kj::Own<ClientHook>> writeTarget(ClientHook& cap, rpc::MessageTarget::Builder target) {
      // If calls to the given capability should pass over this connection, fill in `target`
      // appropriately for such a call and return nullptr.  Otherwise, return a `ClientHook` to which
      // the call should be forwarded; the caller should then delegate the call to that `ClientHook`.
      //
      // The main case where this ends up returning non-null is if `cap` is a promise that has
      // recently resolved.  The application might have started building a request before the promise
      // resolved, and so the request may have been built on the assumption that it would be sent over
      // this network connection, but then the promise resolved to point somewhere else before the
      // request was sent.  Now the request has to be redirected to the new target instead.
  
      if (cap.getBrand() == this) {
        return kj::downcast<RpcClient>(cap).writeTarget(target);
      } else {
        return cap.addRef();
      }
    }
  
    kj::Own<ClientHook> getInnermostClient(ClientHook& client) {
      ClientHook* ptr = &client;
      for (;;) {
        KJ_IF_MAYBE(inner, ptr->getResolved()) {
          ptr = inner;
        } else {
          break;
        }
      }
  
      if (ptr->getBrand() == this) {
        return kj::downcast<RpcClient>(*ptr).getInnermostClient();
      } else {
        return ptr->addRef();
      }
    }
  
    kj::Promise<void> resolveExportedPromise(
        ExportId exportId, kj::Promise<kj::Own<ClientHook>>&& promise) {
      // Implements exporting of a promise.  The promise has been exported under the given ID, and is
      // to eventually resolve to the ClientHook produced by `promise`.  This method waits for that
      // resolve to happen and then sends the appropriate `Resolve` message to the peer.
  
      return promise.then(
          [this,exportId](kj::Own<ClientHook>&& resolution) -> kj::Promise<void> {
        // Successful resolution.
  
        KJ_ASSERT(connection.is<Connected>(),
                  "Resolving export should have been canceled on disconnect.") {
          return kj::READY_NOW;
        }
  
        // Get the innermost ClientHook backing the resolved client.  This includes traversing
        // PromiseClients that haven't resolved yet to their underlying ImportClient or
        // PipelineClient, so that we get a remote promise that might resolve later.  This is
        // important to make sure that if the peer sends a `Disembargo` back to us, it bounces back
        // correctly instead of going to the result of some future resolution.  See the documentation
        // for `Disembargo` in `rpc.capnp`.
        resolution = getInnermostClient(*resolution);
  
        // Update the export table to point at this object instead.  We know that our entry in the
        // export table is still live because when it is destroyed the asynchronous resolution task
        // (i.e. this code) is canceled.
        auto& exp = KJ_ASSERT_NONNULL(exports.find(exportId));
        exportsByCap.erase(exp.clientHook);
        exp.clientHook = kj::mv(resolution);
  
        if (exp.clientHook->getBrand() != this) {
          // We're resolving to a local capability.  If we're resolving to a promise, we might be
          // able to reuse our export table entry and avoid sending a message.
  
          KJ_IF_MAYBE(promise, exp.clientHook->whenMoreResolved()) {
            // We're replacing a promise with another local promise.  In this case, we might actually
            // be able to just reuse the existing export table entry to represent the new promise --
            // unless it already has an entry.  Let's check.
  
            auto insertResult = exportsByCap.insert(std::make_pair(exp.clientHook.get(), exportId));
  
            if (insertResult.second) {
              // The new promise was not already in the table, therefore the existing export table
              // entry has now been repurposed to represent it.  There is no need to send a resolve
              // message at all.  We do, however, have to start resolving the next promise.
              return resolveExportedPromise(exportId, kj::mv(*promise));
            }
          }
        }
  
        // OK, we have to send a `Resolve` message.
        auto message = connection.get<Connected>()->newOutgoingMessage(
            messageSizeHint<rpc::Resolve>() + sizeInWords<rpc::CapDescriptor>() + 16);
        auto resolve = message->getBody().initAs<rpc::Message>().initResolve();
        resolve.setPromiseId(exportId);
        kj::Vector<int> fds;
        writeDescriptor(*exp.clientHook, resolve.initCap(), fds);
        message->setFds(fds.releaseAsArray());
        message->send();
  
        return kj::READY_NOW;
      }, [this,exportId](kj::Exception&& exception) {
        // send error resolution
        auto message = connection.get<Connected>()->newOutgoingMessage(
            messageSizeHint<rpc::Resolve>() + exceptionSizeHint(exception) + 8);
        auto resolve = message->getBody().initAs<rpc::Message>().initResolve();
        resolve.setPromiseId(exportId);
        fromException(exception, resolve.initException());
        message->send();
      }).eagerlyEvaluate([this](kj::Exception&& exception) {
        // Put the exception on the TaskSet which will cause the connection to be terminated.
        tasks.add(kj::mv(exception));
      });
    }
  
    void fromException(const kj::Exception& exception, rpc::Exception::Builder builder) {
      _::fromException(exception, builder, traceEncoder);
    }
  
    // =====================================================================================
    // Interpreting CapDescriptor
  
    kj::Own<ClientHook> import(ImportId importId, bool isPromise, kj::Maybe<kj::AutoCloseFd> fd) {
      // Receive a new import.
  
      auto& import = imports[importId];
      kj::Own<ImportClient> importClient;
  
      // Create the ImportClient, or if one already exists, use it.
      KJ_IF_MAYBE(c, import.importClient) {
        importClient = kj::addRef(*c);
  
        // If the same import is introduced multiple times, and it is missing an FD the first time,
        // but it has one on a later attempt, we want to attach the later one. This could happen
        // because the first introduction was part of a message that had too many other FDs and went
        // over the per-message limit. Perhaps the protocol design is such that this other message
        // doesn't really care if the FDs are transferred or not, but the later message really does
        // care; it would be bad if the previous message blocked later messages from delivering the
        // FD just because it happened to reference the same capability.
        importClient->setFdIfMissing(kj::mv(fd));
      } else {
        importClient = kj::refcounted<ImportClient>(*this, importId, kj::mv(fd));
        import.importClient = *importClient;
      }
  
      // We just received a copy of this import ID, so the remote refcount has gone up.
      importClient->addRemoteRef();
  
      if (isPromise) {
        // We need to construct a PromiseClient around this import, if we haven't already.
        KJ_IF_MAYBE(c, import.appClient) {
          // Use the existing one.
          return kj::addRef(*c);
        } else {
          // Create a promise for this import's resolution.
          auto paf = kj::newPromiseAndFulfiller<kj::Own<ClientHook>>();
          import.promiseFulfiller = kj::mv(paf.fulfiller);
  
          // Make sure the import is not destroyed while this promise exists.
          paf.promise = paf.promise.attach(kj::addRef(*importClient));
  
          // Create a PromiseClient around it and return it.
          auto result = kj::refcounted<PromiseClient>(
              *this, kj::mv(importClient), kj::mv(paf.promise), importId);
          import.appClient = *result;
          return kj::mv(result);
        }
      } else {
        import.appClient = *importClient;
        return kj::mv(importClient);
      }
    }
  
    class TribbleRaceBlocker: public ClientHook, public kj::Refcounted {
      // Hack to work around a problem that arises during the Tribble 4-way Race Condition as
      // described in rpc.capnp in the documentation for the `Disembargo` message.
      //
      // Consider a remote promise that is resolved by a `Resolve` message. PromiseClient::resolve()
      // is eventually called and given the `ClientHook` for the resolution. Imagine that the
      // `ClientHook` it receives turns out to be an `ImportClient`. There are two ways this could
      // have happened:
      //
      // 1. The `Resolve` message contained a `CapDescriptor` of type `senderHosted`, naming an entry
      //    in the sender's export table, and the `ImportClient` refers to the corresponding slot on
      //    the receiver's import table. In this case, no embargo is needed, because messages to the
      //    resolved location traverse the same path as messages to the promise would have.
      //
      // 2. The `Resolve` message contained a `CapDescriptor` of type `receiverHosted`, naming an
      //    entry in the receiver's export table. That entry just happened to contain an
      //    `ImportClient` referring back to the sender. This specifically happens when the entry
      //    in question had previously itself referred to a promise, and that promise has since
      //    resolved to a remote capability, at which point the export table entry was replaced by
      //    the appropriate `ImportClient` representing that. Presumably, the peer *did not yet know*
      //    about this resolution, which is why it sent a `receiverHosted` pointing to something that
      //    reflects back to the sender, rather than sending `senderHosted` in the first place.
      //
      //    In this case, an embargo *is* required, because peer may still be reflecting messages
      //    sent to this promise back to us. In fact, the peer *must* continue reflecting messages,
      //    even when it eventually learns that the eventual destination is one of its own
      //    capabilities, due to the Tribble 4-way Race Condition rule.
      //
      //    Since this case requires an embargo, somehow PromiseClient::resolve() must be able to
      //    distinguish it from the case (1). One solution would be for us to pass some extra flag
      //    all the way from where the `Resolve` messages is received to `PromiseClient::resolve()`.
      //    That solution is reasonably easy in the `Resolve` case, but gets notably more difficult
      //    in the case of `Return`s, which also resolve promises and are subject to all the same
      //    problems. In the case of a `Return`, some non-RPC-specific code is involved in the
      //    resolution, making it harder to pass along a flag.
      //
      //    Instead, we use this hack: When we read an entry in the export table and discover that
      //    it actually contains an `ImportClient` or a `PipelineClient` reflecting back over our
      //    own connection, then we wrap it in a `TribbleRaceBlocker`. This wrapper prevents
      //    `PromiseClient` from recognizing the capability as being remote, so it instead treats it
      //    as local. That causes it to set up an embargo as desired.
      //
      // TODO(perf): This actually blocks further promise resolution in the case where the
      //   ImportClient or PipelineClient itself ends up being yet another promise that resolves
      //   back over the connection again. What we probably really need to do here is, instead of
      //   placing `ImportClient` or `PipelineClient` on the export table, place a special type there
      //   that both knows what to do with future incoming messages to that export ID, but also knows
      //   what to do when that export is the subject of a `Resolve`.
  
    public:
      TribbleRaceBlocker(kj::Own<ClientHook> inner): inner(kj::mv(inner)) {}
  
      Request<AnyPointer, AnyPointer> newCall(
          uint64_t interfaceId, uint16_t methodId, kj::Maybe<MessageSize> sizeHint) override {
        return inner->newCall(interfaceId, methodId, sizeHint);
      }
      VoidPromiseAndPipeline call(uint64_t interfaceId, uint16_t methodId,
                                  kj::Own<CallContextHook>&& context) override {
        return inner->call(interfaceId, methodId, kj::mv(context));
      }
      kj::Maybe<ClientHook&> getResolved() override {
        // We always wrap either PipelineClient or ImportClient, both of which return null for this
        // anyway.
        return nullptr;
      }
      kj::Maybe<kj::Promise<kj::Own<ClientHook>>> whenMoreResolved() override {
        // We always wrap either PipelineClient or ImportClient, both of which return null for this
        // anyway.
        return nullptr;
      }
      kj::Own<ClientHook> addRef() override {
        return kj::addRef(*this);
      }
      const void* getBrand() override {
        return nullptr;
      }
      kj::Maybe<int> getFd() override {
        return inner->getFd();
      }
  
    private:
      kj::Own<ClientHook> inner;
    };
  
    kj::Maybe<kj::Own<ClientHook>> receiveCap(rpc::CapDescriptor::Reader descriptor,
                                              kj::ArrayPtr<kj::AutoCloseFd> fds) {
      uint fdIndex = descriptor.getAttachedFd();
      kj::Maybe<kj::AutoCloseFd> fd;
      if (fdIndex < fds.size() && fds[fdIndex] != nullptr) {
        fd = kj::mv(fds[fdIndex]);
      }
  
      switch (descriptor.which()) {
        case rpc::CapDescriptor::NONE:
          return nullptr;
  
        case rpc::CapDescriptor::SENDER_HOSTED:
          return import(descriptor.getSenderHosted(), false, kj::mv(fd));
        case rpc::CapDescriptor::SENDER_PROMISE:
          return import(descriptor.getSenderPromise(), true, kj::mv(fd));
  
        case rpc::CapDescriptor::RECEIVER_HOSTED:
          KJ_IF_MAYBE(exp, exports.find(descriptor.getReceiverHosted())) {
            auto result = exp->clientHook->addRef();
            if (result->getBrand() == this) {
              result = kj::refcounted<TribbleRaceBlocker>(kj::mv(result));
            }
            return kj::mv(result);
          } else {
            return newBrokenCap("invalid 'receiverHosted' export ID");
          }
  
        case rpc::CapDescriptor::RECEIVER_ANSWER: {
          auto promisedAnswer = descriptor.getReceiverAnswer();
  
          KJ_IF_MAYBE(answer, answers.find(promisedAnswer.getQuestionId())) {
            if (answer->active) {
              KJ_IF_MAYBE(pipeline, answer->pipeline) {
                KJ_IF_MAYBE(ops, toPipelineOps(promisedAnswer.getTransform())) {
                  auto result = pipeline->get()->getPipelinedCap(*ops);
                  if (result->getBrand() == this) {
                    result = kj::refcounted<TribbleRaceBlocker>(kj::mv(result));
                  }
                  return kj::mv(result);
                } else {
                  return newBrokenCap("unrecognized pipeline ops");
                }
              }
            }
          }
  
          return newBrokenCap("invalid 'receiverAnswer'");
        }
  
        case rpc::CapDescriptor::THIRD_PARTY_HOSTED:
          // We don't support third-party caps, so use the vine instead.
          return import(descriptor.getThirdPartyHosted().getVineId(), false, kj::mv(fd));
  
        default:
          KJ_FAIL_REQUIRE("unknown CapDescriptor type") { break; }
          return newBrokenCap("unknown CapDescriptor type");
      }
    }
  
    kj::Array<kj::Maybe<kj::Own<ClientHook>>> receiveCaps(List<rpc::CapDescriptor>::Reader capTable,
                                                          kj::ArrayPtr<kj::AutoCloseFd> fds) {
      auto result = kj::heapArrayBuilder<kj::Maybe<kj::Own<ClientHook>>>(capTable.size());
      for (auto cap: capTable) {
        result.add(receiveCap(cap, fds));
      }
      return result.finish();
    }
  
    // =====================================================================================
    // RequestHook/PipelineHook/ResponseHook implementations
  
    class QuestionRef: public kj::Refcounted {
      // A reference to an entry on the question table.  Used to detect when the `Finish` message
      // can be sent.
  
    public:
      inline QuestionRef(
          RpcConnectionState& connectionState, QuestionId id,
          kj::Own<kj::PromiseFulfiller<kj::Promise<kj::Own<RpcResponse>>>> fulfiller)
          : connectionState(kj::addRef(connectionState)), id(id), fulfiller(kj::mv(fulfiller)) {}
  
      ~QuestionRef() noexcept {
        // Contrary to KJ style, we declare this destructor `noexcept` because if anything in here
        // throws (without being caught) we're probably in pretty bad shape and going to be crashing
        // later anyway. Better to abort now.
  
        auto& question = KJ_ASSERT_NONNULL(
            connectionState->questions.find(id), "Question ID no longer on table?");
  
        // Send the "Finish" message (if the connection is not already broken).
        if (connectionState->connection.is<Connected>() && !question.skipFinish) {
          KJ_IF_MAYBE(e, kj::runCatchingExceptions([&]() {
            auto message = connectionState->connection.get<Connected>()->newOutgoingMessage(
                messageSizeHint<rpc::Finish>());
            auto builder = message->getBody().getAs<rpc::Message>().initFinish();
            builder.setQuestionId(id);
            // If we're still awaiting a return, then this request is being canceled, and we're going
            // to ignore any capabilities in the return message, so set releaseResultCaps true. If we
            // already received the return, then we've already built local proxies for the caps and
            // will send Release messages when those are destroyed.
            builder.setReleaseResultCaps(question.isAwaitingReturn);
            message->send();
          })) {
            connectionState->disconnect(kj::mv(*e));
          }
        }
  
        // Check if the question has returned and, if so, remove it from the table.
        // Remove question ID from the table.  Must do this *after* sending `Finish` to ensure that
        // the ID is not re-allocated before the `Finish` message can be sent.
        if (question.isAwaitingReturn) {
          // Still waiting for return, so just remove the QuestionRef pointer from the table.
          question.selfRef = nullptr;
        } else {
          // Call has already returned, so we can now remove it from the table.
          connectionState->questions.erase(id, question);
        }
      }
  
      inline QuestionId getId() const { return id; }
  
      void fulfill(kj::Own<RpcResponse>&& response) {
        fulfiller->fulfill(kj::mv(response));
      }
  
      void fulfill(kj::Promise<kj::Own<RpcResponse>>&& promise) {
        fulfiller->fulfill(kj::mv(promise));
      }
  
      void reject(kj::Exception&& exception) {
        fulfiller->reject(kj::mv(exception));
      }
  
    private:
      kj::Own<RpcConnectionState> connectionState;
      QuestionId id;
      kj::Own<kj::PromiseFulfiller<kj::Promise<kj::Own<RpcResponse>>>> fulfiller;
    };
  
    class RpcRequest final: public RequestHook {
    public:
      RpcRequest(RpcConnectionState& connectionState, VatNetworkBase::Connection& connection,
                 kj::Maybe<MessageSize> sizeHint, kj::Own<RpcClient>&& target)
          : connectionState(kj::addRef(connectionState)),
            target(kj::mv(target)),
            message(connection.newOutgoingMessage(
                firstSegmentSize(sizeHint, messageSizeHint<rpc::Call>() +
                    sizeInWords<rpc::Payload>() + MESSAGE_TARGET_SIZE_HINT))),
            callBuilder(message->getBody().getAs<rpc::Message>().initCall()),
            paramsBuilder(capTable.imbue(callBuilder.getParams().getContent())) {}
  
      inline AnyPointer::Builder getRoot() {
        return paramsBuilder;
      }
      inline rpc::Call::Builder getCall() {
        return callBuilder;
      }
  
      RemotePromise<AnyPointer> send() override {
        if (!connectionState->connection.is<Connected>()) {
          // Connection is broken.
          const kj::Exception& e = connectionState->connection.get<Disconnected>();
          return RemotePromise<AnyPointer>(
              kj::Promise<Response<AnyPointer>>(kj::cp(e)),
              AnyPointer::Pipeline(newBrokenPipeline(kj::cp(e))));
        }
  
        KJ_IF_MAYBE(redirect, target->writeTarget(callBuilder.getTarget())) {
          // Whoops, this capability has been redirected while we were building the request!
          // We'll have to make a new request and do a copy.  Ick.
  
          auto replacement = redirect->get()->newCall(
              callBuilder.getInterfaceId(), callBuilder.getMethodId(), paramsBuilder.targetSize());
          replacement.set(paramsBuilder);
          return replacement.send();
        } else {
          auto sendResult = sendInternal(false);
  
          auto forkedPromise = sendResult.promise.fork();
  
          // The pipeline must get notified of resolution before the app does to maintain ordering.
          auto pipeline = kj::refcounted<RpcPipeline>(
              *connectionState, kj::mv(sendResult.questionRef), forkedPromise.addBranch());
  
          auto appPromise = forkedPromise.addBranch().then(
              [=](kj::Own<RpcResponse>&& response) {
                auto reader = response->getResults();
                return Response<AnyPointer>(reader, kj::mv(response));
              });
  
          return RemotePromise<AnyPointer>(
              kj::mv(appPromise),
              AnyPointer::Pipeline(kj::mv(pipeline)));
        }
      }
  
      kj::Promise<void> sendStreaming() override {
        if (!connectionState->connection.is<Connected>()) {
          // Connection is broken.
          return kj::cp(connectionState->connection.get<Disconnected>());
        }
  
        KJ_IF_MAYBE(redirect, target->writeTarget(callBuilder.getTarget())) {
          // Whoops, this capability has been redirected while we were building the request!
          // We'll have to make a new request and do a copy.  Ick.
  
          auto replacement = redirect->get()->newCall(
              callBuilder.getInterfaceId(), callBuilder.getMethodId(), paramsBuilder.targetSize());
          replacement.set(paramsBuilder);
          return RequestHook::from(kj::mv(replacement))->sendStreaming();
        } else {
          return sendStreamingInternal(false);
        }
      }
  
      struct TailInfo {
        QuestionId questionId;
        kj::Promise<void> promise;
        kj::Own<PipelineHook> pipeline;
      };
  
      kj::Maybe<TailInfo> tailSend() {
        // Send the request as a tail call.
        //
        // Returns null if for some reason a tail call is not possible and the caller should fall
        // back to using send() and copying the response.
  
        SendInternalResult sendResult;
  
        if (!connectionState->connection.is<Connected>()) {
          // Disconnected; fall back to a regular send() which will fail appropriately.
          return nullptr;
        }
  
        KJ_IF_MAYBE(redirect, target->writeTarget(callBuilder.getTarget())) {
          // Whoops, this capability has been redirected while we were building the request!
          // Fall back to regular send().
          return nullptr;
        } else {
          sendResult = sendInternal(true);
        }
  
        auto promise = sendResult.promise.then([](kj::Own<RpcResponse>&& response) {
          // Response should be null if `Return` handling code is correct.
          KJ_ASSERT(!response) { break; }
        });
  
        QuestionId questionId = sendResult.questionRef->getId();
  
        auto pipeline = kj::refcounted<RpcPipeline>(*connectionState, kj::mv(sendResult.questionRef));
  
        return TailInfo { questionId, kj::mv(promise), kj::mv(pipeline) };
      }
  
      const void* getBrand() override {
        return connectionState.get();
      }
  
    private:
      kj::Own<RpcConnectionState> connectionState;
  
      kj::Own<RpcClient> target;
      kj::Own<OutgoingRpcMessage> message;
      BuilderCapabilityTable capTable;
      rpc::Call::Builder callBuilder;
      AnyPointer::Builder paramsBuilder;
  
      struct SendInternalResult {
        kj::Own<QuestionRef> questionRef;
        kj::Promise<kj::Own<RpcResponse>> promise = nullptr;
      };
  
      struct SetupSendResult: public SendInternalResult {
        QuestionId questionId;
        Question& question;
  
        SetupSendResult(SendInternalResult&& super, QuestionId questionId, Question& question)
            : SendInternalResult(kj::mv(super)), questionId(questionId), question(question) {}
        // TODO(cleanup): This constructor is implicit in C++17.
      };
  
      SetupSendResult setupSend(bool isTailCall) {
        // Build the cap table.
        kj::Vector<int> fds;
        auto exports = connectionState->writeDescriptors(
            capTable.getTable(), callBuilder.getParams(), fds);
        message->setFds(fds.releaseAsArray());
  
        // Init the question table.  Do this after writing descriptors to avoid interference.
        QuestionId questionId;
        auto& question = connectionState->questions.next(questionId);
        question.isAwaitingReturn = true;
        question.paramExports = kj::mv(exports);
        question.isTailCall = isTailCall;
  
        // Make the QuentionRef and result promise.
        SendInternalResult result;
        auto paf = kj::newPromiseAndFulfiller<kj::Promise<kj::Own<RpcResponse>>>();
        result.questionRef = kj::refcounted<QuestionRef>(
            *connectionState, questionId, kj::mv(paf.fulfiller));
        question.selfRef = *result.questionRef;
        result.promise = paf.promise.attach(kj::addRef(*result.questionRef));
  
        return { kj::mv(result), questionId, question };
      }
  
      SendInternalResult sendInternal(bool isTailCall) {
        auto result = setupSend(isTailCall);
  
        // Finish and send.
        callBuilder.setQuestionId(result.questionId);
        if (isTailCall) {
          callBuilder.getSendResultsTo().setYourself();
        }
        KJ_IF_MAYBE(exception, kj::runCatchingExceptions([&]() {
          KJ_CONTEXT("sending RPC call",
             callBuilder.getInterfaceId(), callBuilder.getMethodId());
          message->send();
        })) {
          // We can't safely throw the exception from here since we've already modified the question
          // table state. We'll have to reject the promise instead.
          result.question.isAwaitingReturn = false;
          result.question.skipFinish = true;
          connectionState->releaseExports(result.question.paramExports);
          result.questionRef->reject(kj::mv(*exception));
        }
  
        // Send and return.
        return kj::mv(result);
      }
  
      kj::Promise<void> sendStreamingInternal(bool isTailCall) {
        auto setup = setupSend(isTailCall);
  
        // Finish and send.
        callBuilder.setQuestionId(setup.questionId);
        if (isTailCall) {
          callBuilder.getSendResultsTo().setYourself();
        }
        kj::Promise<void> flowPromise = nullptr;
        KJ_IF_MAYBE(exception, kj::runCatchingExceptions([&]() {
          KJ_CONTEXT("sending RPC call",
             callBuilder.getInterfaceId(), callBuilder.getMethodId());
          RpcFlowController* flow;
          KJ_IF_MAYBE(f, target->flowController) {
            flow = *f;
          } else {
            flow = target->flowController.emplace(
                connectionState->connection.get<Connected>()->newStream());
          }
          flowPromise = flow->send(kj::mv(message), setup.promise.ignoreResult());
        })) {
          // We can't safely throw the exception from here since we've already modified the question
          // table state. We'll have to reject the promise instead.
          setup.question.isAwaitingReturn = false;
          setup.question.skipFinish = true;
          setup.questionRef->reject(kj::cp(*exception));
          return kj::mv(*exception);
        }
  
        return kj::mv(flowPromise);
      }
    };
  
    class RpcPipeline final: public PipelineHook, public kj::Refcounted {
    public:
      RpcPipeline(RpcConnectionState& connectionState, kj::Own<QuestionRef>&& questionRef,
                  kj::Promise<kj::Own<RpcResponse>>&& redirectLaterParam)
          : connectionState(kj::addRef(connectionState)),
            redirectLater(redirectLaterParam.fork()),
            resolveSelfPromise(KJ_ASSERT_NONNULL(redirectLater).addBranch().then(
                [this](kj::Own<RpcResponse>&& response) {
                  resolve(kj::mv(response));
                }, [this](kj::Exception&& exception) {
                  resolve(kj::mv(exception));
                }).eagerlyEvaluate([&](kj::Exception&& e) {
                  // Make any exceptions thrown from resolve() go to the connection's TaskSet which
                  // will cause the connection to be terminated.
                  connectionState.tasks.add(kj::mv(e));
                })) {
        // Construct a new RpcPipeline.
  
        state.init<Waiting>(kj::mv(questionRef));
      }
  
      RpcPipeline(RpcConnectionState& connectionState, kj::Own<QuestionRef>&& questionRef)
          : connectionState(kj::addRef(connectionState)),
            resolveSelfPromise(nullptr) {
        // Construct a new RpcPipeline that is never expected to resolve.
  
        state.init<Waiting>(kj::mv(questionRef));
      }
  
      // implements PipelineHook ---------------------------------------
  
      kj::Own<PipelineHook> addRef() override {
        return kj::addRef(*this);
      }
  
      kj::Own<ClientHook> getPipelinedCap(kj::ArrayPtr<const PipelineOp> ops) override {
        auto copy = kj::heapArrayBuilder<PipelineOp>(ops.size());
        for (auto& op: ops) {
          copy.add(op);
        }
        return getPipelinedCap(copy.finish());
      }
  
      kj::Own<ClientHook> getPipelinedCap(kj::Array<PipelineOp>&& ops) override {
        return clientMap.findOrCreate(ops.asPtr(), [&]() {
          if (state.is<Waiting>()) {
            // Wrap a PipelineClient in a PromiseClient.
            auto pipelineClient = kj::refcounted<PipelineClient>(
                *connectionState, kj::addRef(*state.get<Waiting>()), kj::heapArray(ops.asPtr()));
  
            KJ_IF_MAYBE(r, redirectLater) {
              auto resolutionPromise = r->addBranch().then(
                  [ops = kj::heapArray(ops.asPtr())](kj::Own<RpcResponse>&& response) {
                    return response->getResults().getPipelinedCap(kj::mv(ops));
                  });
  
              return kj::HashMap<kj::Array<PipelineOp>, kj::Own<ClientHook>>::Entry {
                kj::mv(ops),
                kj::refcounted<PromiseClient>(
                    *connectionState, kj::mv(pipelineClient), kj::mv(resolutionPromise), nullptr)
              };
            } else {
              // Oh, this pipeline will never get redirected, so just return the PipelineClient.
              return kj::HashMap<kj::Array<PipelineOp>, kj::Own<ClientHook>>::Entry {
                kj::mv(ops), kj::mv(pipelineClient)
              };
            }
          } else if (state.is<Resolved>()) {
            auto pipelineClient = state.get<Resolved>()->getResults().getPipelinedCap(ops);
            return kj::HashMap<kj::Array<PipelineOp>, kj::Own<ClientHook>>::Entry {
              kj::mv(ops), kj::mv(pipelineClient)
            };
          } else {
            return kj::HashMap<kj::Array<PipelineOp>, kj::Own<ClientHook>>::Entry {
              kj::mv(ops), newBrokenCap(kj::cp(state.get<Broken>()))
            };
          }
        })->addRef();
      }
  
    private:
      kj::Own<RpcConnectionState> connectionState;
      kj::Maybe<kj::ForkedPromise<kj::Own<RpcResponse>>> redirectLater;
  
      typedef kj::Own<QuestionRef> Waiting;
      typedef kj::Own<RpcResponse> Resolved;
      typedef kj::Exception Broken;
      kj::OneOf<Waiting, Resolved, Broken> state;
  
      kj::HashMap<kj::Array<PipelineOp>, kj::Own<ClientHook>> clientMap;
      // See QueuedPipeline::clientMap in capability.c++ for a discussion of why we must memoize
      // the results of getPipelinedCap(). RpcPipeline has a similar problem when a capability we
      // return is later subject to an embargo. It's important that the embargo is correctly applied
      // across all calls to the same capability.
  
      // Keep this last, because the continuation uses *this, so it should be destroyed first to
      // ensure the continuation is not still running.
      kj::Promise<void> resolveSelfPromise;
  
      void resolve(kj::Own<RpcResponse>&& response) {
        KJ_ASSERT(state.is<Waiting>(), "Already resolved?");
        state.init<Resolved>(kj::mv(response));
      }
  
      void resolve(const kj::Exception&& exception) {
        KJ_ASSERT(state.is<Waiting>(), "Already resolved?");
        state.init<Broken>(kj::mv(exception));
      }
    };
  
    class RpcResponse: public ResponseHook {
    public:
      virtual AnyPointer::Reader getResults() = 0;
      virtual kj::Own<RpcResponse> addRef() = 0;
    };
  
    class RpcResponseImpl final: public RpcResponse, public kj::Refcounted {
    public:
      RpcResponseImpl(RpcConnectionState& connectionState,
                      kj::Own<QuestionRef>&& questionRef,
                      kj::Own<IncomingRpcMessage>&& message,
                      kj::Array<kj::Maybe<kj::Own<ClientHook>>> capTableArray,
                      AnyPointer::Reader results)
          : connectionState(kj::addRef(connectionState)),
            message(kj::mv(message)),
            capTable(kj::mv(capTableArray)),
            reader(capTable.imbue(results)),
            questionRef(kj::mv(questionRef)) {}
  
      AnyPointer::Reader getResults() override {
        return reader;
      }
  
      kj::Own<RpcResponse> addRef() override {
        return kj::addRef(*this);
      }
  
    private:
      kj::Own<RpcConnectionState> connectionState;
      kj::Own<IncomingRpcMessage> message;
      ReaderCapabilityTable capTable;
      AnyPointer::Reader reader;
      kj::Own<QuestionRef> questionRef;
    };
  
    // =====================================================================================
    // CallContextHook implementation
  
    class RpcServerResponse {
    public:
      virtual AnyPointer::Builder getResultsBuilder() = 0;
    };
  
    class RpcServerResponseImpl final: public RpcServerResponse {
    public:
      RpcServerResponseImpl(RpcConnectionState& connectionState,
                            kj::Own<OutgoingRpcMessage>&& message,
                            rpc::Payload::Builder payload)
          : connectionState(connectionState),
            message(kj::mv(message)),
            payload(payload) {}
  
      AnyPointer::Builder getResultsBuilder() override {
        return capTable.imbue(payload.getContent());
      }
  
      kj::Maybe<kj::Array<ExportId>> send() {
        // Send the response and return the export list.  Returns nullptr if there were no caps.
        // (Could return a non-null empty array if there were caps but none of them were exports.)
  
        // Build the cap table.
        auto capTable = this->capTable.getTable();
        kj::Vector<int> fds;
        auto exports = connectionState.writeDescriptors(capTable, payload, fds);
        message->setFds(fds.releaseAsArray());
  
        // Capabilities that we are returning are subject to embargos. See `Disembargo` in rpc.capnp.
        // As explained there, in order to deal with the Tribble 4-way race condition, we need to
        // make sure that if we're returning any remote promises, that we ignore any subsequent
        // resolution of those promises for the purpose of pipelined requests on this answer. Luckily,
        // we can modify the cap table in-place.
        for (auto& slot: capTable) {
          KJ_IF_MAYBE(cap, slot) {
            slot = connectionState.getInnermostClient(**cap);
          }
        }
  
        message->send();
        if (capTable.size() == 0) {
          return nullptr;
        } else {
          return kj::mv(exports);
        }
      }
  
    private:
      RpcConnectionState& connectionState;
      kj::Own<OutgoingRpcMessage> message;
      BuilderCapabilityTable capTable;
      rpc::Payload::Builder payload;
    };
  
    class LocallyRedirectedRpcResponse final
        : public RpcResponse, public RpcServerResponse, public kj::Refcounted{
    public:
      LocallyRedirectedRpcResponse(kj::Maybe<MessageSize> sizeHint)
          : message(sizeHint.map([](MessageSize size) { return size.wordCount; })
                            .orDefault(SUGGESTED_FIRST_SEGMENT_WORDS)) {}
  
      AnyPointer::Builder getResultsBuilder() override {
        return message.getRoot<AnyPointer>();
      }
  
      AnyPointer::Reader getResults() override {
        return message.getRoot<AnyPointer>();
      }
  
      kj::Own<RpcResponse> addRef() override {
        return kj::addRef(*this);
      }
  
    private:
      MallocMessageBuilder message;
    };
  
    class RpcCallContext final: public CallContextHook, public kj::Refcounted {
    public:
      RpcCallContext(RpcConnectionState& connectionState, AnswerId answerId,
                     kj::Own<IncomingRpcMessage>&& request,
                     kj::Array<kj::Maybe<kj::Own<ClientHook>>> capTableArray,
                     const AnyPointer::Reader& params,
                     bool redirectResults, kj::Own<kj::PromiseFulfiller<void>>&& cancelFulfiller,
                     uint64_t interfaceId, uint16_t methodId)
          : connectionState(kj::addRef(connectionState)),
            answerId(answerId),
            interfaceId(interfaceId),
            methodId(methodId),
            requestSize(request->sizeInWords()),
            request(kj::mv(request)),
            paramsCapTable(kj::mv(capTableArray)),
            params(paramsCapTable.imbue(params)),
            returnMessage(nullptr),
            redirectResults(redirectResults),
            cancelFulfiller(kj::mv(cancelFulfiller)) {
        connectionState.callWordsInFlight += requestSize;
      }
  
      ~RpcCallContext() noexcept(false) {
        if (isFirstResponder()) {
          // We haven't sent a return yet, so we must have been canceled.  Send a cancellation return.
          unwindDetector.catchExceptionsIfUnwinding([&]() {
            // Don't send anything if the connection is broken.
            bool shouldFreePipeline = true;
            if (connectionState->connection.is<Connected>()) {
              auto message = connectionState->connection.get<Connected>()->newOutgoingMessage(
                  messageSizeHint<rpc::Return>() + sizeInWords<rpc::Payload>());
              auto builder = message->getBody().initAs<rpc::Message>().initReturn();
  
              builder.setAnswerId(answerId);
              builder.setReleaseParamCaps(false);
  
              if (redirectResults) {
                // The reason we haven't sent a return is because the results were sent somewhere
                // else.
                builder.setResultsSentElsewhere();
  
                // The pipeline could still be valid and in-use in this case.
                shouldFreePipeline = false;
              } else {
                builder.setCanceled();
              }
  
              message->send();
            }
  
            cleanupAnswerTable(nullptr, shouldFreePipeline);
          });
        }
      }
  
      kj::Own<RpcResponse> consumeRedirectedResponse() {
        KJ_ASSERT(redirectResults);
  
        if (response == nullptr) getResults(MessageSize{0, 0});  // force initialization of response
  
        // Note that the context needs to keep its own reference to the response so that it doesn't
        // get GC'd until the PipelineHook drops its reference to the context.
        return kj::downcast<LocallyRedirectedRpcResponse>(*KJ_ASSERT_NONNULL(response)).addRef();
      }
  
      void sendReturn() {
        KJ_ASSERT(!redirectResults);
  
        // Avoid sending results if canceled so that we don't have to figure out whether or not
        // `releaseResultCaps` was set in the already-received `Finish`.
        if (!(cancellationFlags & CANCEL_REQUESTED) && isFirstResponder()) {
          KJ_ASSERT(connectionState->connection.is<Connected>(),
                    "Cancellation should have been requested on disconnect.") {
            return;
          }
  
          if (response == nullptr) getResults(MessageSize{0, 0});  // force initialization of response
  
          returnMessage.setAnswerId(answerId);
          returnMessage.setReleaseParamCaps(false);
  
          kj::Maybe<kj::Array<ExportId>> exports;
          KJ_IF_MAYBE(exception, kj::runCatchingExceptions([&]() {
            // Debug info in case send() fails due to overside message.
            KJ_CONTEXT("returning from RPC call", interfaceId, methodId);
            exports = kj::downcast<RpcServerResponseImpl>(*KJ_ASSERT_NONNULL(response)).send();
          })) {
            responseSent = false;
            sendErrorReturn(kj::mv(*exception));
            return;
          }
  
          KJ_IF_MAYBE(e, exports) {
            // Caps were returned, so we can't free the pipeline yet.
            cleanupAnswerTable(kj::mv(*e), false);
          } else {
            // No caps in the results, therefore the pipeline is irrelevant.
            cleanupAnswerTable(nullptr, true);
          }
        }
      }
      void sendErrorReturn(kj::Exception&& exception) {
        KJ_ASSERT(!redirectResults);
        if (isFirstResponder()) {
          if (connectionState->connection.is<Connected>()) {
            auto message = connectionState->connection.get<Connected>()->newOutgoingMessage(
                messageSizeHint<rpc::Return>() + exceptionSizeHint(exception));
            auto builder = message->getBody().initAs<rpc::Message>().initReturn();
  
            builder.setAnswerId(answerId);
            builder.setReleaseParamCaps(false);
            connectionState->fromException(exception, builder.initException());
  
            message->send();
          }
  
          // Do not allow releasing the pipeline because we want pipelined calls to propagate the
          // exception rather than fail with a "no such field" exception.
          cleanupAnswerTable(nullptr, false);
        }
      }
      void sendRedirectReturn() {
        KJ_ASSERT(redirectResults);
  
        if (isFirstResponder()) {
          auto message = connectionState->connection.get<Connected>()->newOutgoingMessage(
              messageSizeHint<rpc::Return>());
          auto builder = message->getBody().initAs<rpc::Message>().initReturn();
  
          builder.setAnswerId(answerId);
          builder.setReleaseParamCaps(false);
          builder.setResultsSentElsewhere();
  
          message->send();
  
          cleanupAnswerTable(nullptr, false);
        }
      }
  
      void requestCancel() {
        // Hints that the caller wishes to cancel this call.  At the next time when cancellation is
        // deemed safe, the RpcCallContext shall send a canceled Return -- or if it never becomes
        // safe, the RpcCallContext will send a normal return when the call completes.  Either way
        // the RpcCallContext is now responsible for cleaning up the entry in the answer table, since
        // a Finish message was already received.
  
        bool previouslyAllowedButNotRequested = cancellationFlags == CANCEL_ALLOWED;
        cancellationFlags |= CANCEL_REQUESTED;
  
        if (previouslyAllowedButNotRequested) {
          // We just set CANCEL_REQUESTED, and CANCEL_ALLOWED was already set previously.  Initiate
          // the cancellation.
          cancelFulfiller->fulfill();
        }
      }
  
      // implements CallContextHook ------------------------------------
  
      AnyPointer::Reader getParams() override {
        KJ_REQUIRE(request != nullptr, "Can't call getParams() after releaseParams().");
        return params;
      }
      void releaseParams() override {
        request = nullptr;
      }
      AnyPointer::Builder getResults(kj::Maybe<MessageSize> sizeHint) override {
        KJ_IF_MAYBE(r, response) {
          return r->get()->getResultsBuilder();
        } else {
          kj::Own<RpcServerResponse> response;
  
          if (redirectResults || !connectionState->connection.is<Connected>()) {
            response = kj::refcounted<LocallyRedirectedRpcResponse>(sizeHint);
          } else {
            auto message = connectionState->connection.get<Connected>()->newOutgoingMessage(
                firstSegmentSize(sizeHint, messageSizeHint<rpc::Return>() +
                                 sizeInWords<rpc::Payload>()));
            returnMessage = message->getBody().initAs<rpc::Message>().initReturn();
            response = kj::heap<RpcServerResponseImpl>(
                *connectionState, kj::mv(message), returnMessage.getResults());
          }
  
          auto results = response->getResultsBuilder();
          this->response = kj::mv(response);
          return results;
        }
      }
      void setPipeline(kj::Own<PipelineHook>&& pipeline) override {
        KJ_IF_MAYBE(f, tailCallPipelineFulfiller) {
          f->get()->fulfill(AnyPointer::Pipeline(kj::mv(pipeline)));
        }
      }
      kj::Promise<void> tailCall(kj::Own<RequestHook>&& request) override {
        auto result = directTailCall(kj::mv(request));
        KJ_IF_MAYBE(f, tailCallPipelineFulfiller) {
          f->get()->fulfill(AnyPointer::Pipeline(kj::mv(result.pipeline)));
        }
        return kj::mv(result.promise);
      }
      ClientHook::VoidPromiseAndPipeline directTailCall(kj::Own<RequestHook>&& request) override {
        KJ_REQUIRE(response == nullptr,
                   "Can't call tailCall() after initializing the results struct.");
  
        if (request->getBrand() == connectionState.get() && !redirectResults) {
          // The tail call is headed towards the peer that called us in the first place, so we can
          // optimize out the return trip.
  
          KJ_IF_MAYBE(tailInfo, kj::downcast<RpcRequest>(*request).tailSend()) {
            if (isFirstResponder()) {
              if (connectionState->connection.is<Connected>()) {
                auto message = connectionState->connection.get<Connected>()->newOutgoingMessage(
                    messageSizeHint<rpc::Return>());
                auto builder = message->getBody().initAs<rpc::Message>().initReturn();
  
                builder.setAnswerId(answerId);
                builder.setReleaseParamCaps(false);
                builder.setTakeFromOtherQuestion(tailInfo->questionId);
  
                message->send();
              }
  
              // There are no caps in our return message, but of course the tail results could have
              // caps, so we must continue to honor pipeline calls (and just bounce them back).
              cleanupAnswerTable(nullptr, false);
            }
            return { kj::mv(tailInfo->promise), kj::mv(tailInfo->pipeline) };
          }
        }
  
        // Just forwarding to another local call.
        auto promise = request->send();
  
        // Wait for response.
        auto voidPromise = promise.then([this](Response<AnyPointer>&& tailResponse) {
          // Copy the response.
          // TODO(perf):  It would be nice if we could somehow make the response get built in-place
          //   but requires some refactoring.
          getResults(tailResponse.targetSize()).set(tailResponse);
        });
  
        return { kj::mv(voidPromise), PipelineHook::from(kj::mv(promise)) };
      }
      kj::Promise<AnyPointer::Pipeline> onTailCall() override {
        auto paf = kj::newPromiseAndFulfiller<AnyPointer::Pipeline>();
        tailCallPipelineFulfiller = kj::mv(paf.fulfiller);
        return kj::mv(paf.promise);
      }
      void allowCancellation() override {
        bool previouslyRequestedButNotAllowed = cancellationFlags == CANCEL_REQUESTED;
        cancellationFlags |= CANCEL_ALLOWED;
  
        if (previouslyRequestedButNotAllowed) {
          // We just set CANCEL_ALLOWED, and CANCEL_REQUESTED was already set previously.  Initiate
          // the cancellation.
          cancelFulfiller->fulfill();
        }
      }
      kj::Own<CallContextHook> addRef() override {
        return kj::addRef(*this);
      }
  
    private:
      kj::Own<RpcConnectionState> connectionState;
      AnswerId answerId;
  
      uint64_t interfaceId;
      uint16_t methodId;
      // For debugging.
  
      // Request ---------------------------------------------
  
      size_t requestSize;  // for flow limit purposes
      kj::Maybe<kj::Own<IncomingRpcMessage>> request;
      ReaderCapabilityTable paramsCapTable;
      AnyPointer::Reader params;
  
      // Response --------------------------------------------
  
      kj::Maybe<kj::Own<RpcServerResponse>> response;
      rpc::Return::Builder returnMessage;
      bool redirectResults = false;
      bool responseSent = false;
      kj::Maybe<kj::Own<kj::PromiseFulfiller<AnyPointer::Pipeline>>> tailCallPipelineFulfiller;
  
      // Cancellation state ----------------------------------
  
      enum CancellationFlags {
        CANCEL_REQUESTED = 1,
        CANCEL_ALLOWED = 2
      };
  
      uint8_t cancellationFlags = 0;
      // When both flags are set, the cancellation process will begin.
  
      kj::Own<kj::PromiseFulfiller<void>> cancelFulfiller;
      // Fulfilled when cancellation has been both requested and permitted.  The fulfilled promise is
      // exclusive-joined with the outermost promise waiting on the call return, so fulfilling it
      // cancels that promise.
  
      kj::UnwindDetector unwindDetector;
  
      // -----------------------------------------------------
  
      bool isFirstResponder() {
        if (responseSent) {
          return false;
        } else {
          responseSent = true;
          return true;
        }
      }
  
      void cleanupAnswerTable(kj::Array<ExportId> resultExports, bool shouldFreePipeline) {
        // We need to remove the `callContext` pointer -- which points back to us -- from the
        // answer table.  Or we might even be responsible for removing the entire answer table
        // entry.
  
        if (cancellationFlags & CANCEL_REQUESTED) {
          // Already received `Finish` so it's our job to erase the table entry. We shouldn't have
          // sent results if canceled, so we shouldn't have an export list to deal with.
          KJ_ASSERT(resultExports.size() == 0);
          connectionState->answers.erase(answerId);
        } else {
          // We just have to null out callContext and set the exports.
          auto& answer = connectionState->answers[answerId];
          answer.callContext = nullptr;
          answer.resultExports = kj::mv(resultExports);
  
          if (shouldFreePipeline) {
            // We can free the pipeline early, because we know all pipeline calls are invalid (e.g.
            // because there are no caps in the result to receive pipeline requests).
            KJ_ASSERT(resultExports.size() == 0);
            answer.pipeline = nullptr;
          }
        }
  
        // Also, this is the right time to stop counting the call against the flow limit.
        connectionState->callWordsInFlight -= requestSize;
        connectionState->maybeUnblockFlow();
      }
    };
  
    // =====================================================================================
    // Message handling
  
    void maybeUnblockFlow() {
      if (callWordsInFlight < flowLimit) {
        KJ_IF_MAYBE(w, flowWaiter) {
          w->get()->fulfill();
          flowWaiter = nullptr;
        }
      }
    }
  
    kj::Promise<void> messageLoop() {
      if (!connection.is<Connected>()) {
        return kj::READY_NOW;
      }
  
      if (callWordsInFlight > flowLimit) {
        auto paf = kj::newPromiseAndFulfiller<void>();
        flowWaiter = kj::mv(paf.fulfiller);
        return paf.promise.then([this]() {
          return messageLoop();
        });
      }
  
      return canceler.wrap(connection.get<Connected>()->receiveIncomingMessage()).then(
          [this](kj::Maybe<kj::Own<IncomingRpcMessage>>&& message) {
        KJ_IF_MAYBE(m, message) {
          handleMessage(kj::mv(*m));
          return true;
        } else {
          disconnect(KJ_EXCEPTION(DISCONNECTED, "Peer disconnected."));
          return false;
        }
      }).then([this](bool keepGoing) {
        // No exceptions; continue loop.
        //
        // (We do this in a separate continuation to handle the case where exceptions are
        // disabled.)
        //
        // TODO(perf): We add an evalLater() here so that anything we needed to do in reaction to
        //   the previous message has a chance to complete before the next message is handled. In
        //   particular, without this, I observed an ordering problem: I saw a case where a `Return`
        //   message was followed by a `Resolve` message, but the `PromiseClient` associated with the
        //   `Resolve` had its `resolve()` method invoked _before_ any `PromiseClient`s associated
        //   with pipelined capabilities resolved by the `Return`. This could lead to an
        //   incorrectly-ordered interaction between `PromiseClient`s when they resolve to each
        //   other. This is probably really a bug in the way `Return`s are handled -- apparently,
        //   resolution of `PromiseClient`s based on returned capabilities does not occur in a
        //   depth-first way, when it should. If we could fix that then we can probably remove this
        //   `evalLater()`. However, the `evalLater()` is not that bad and solves the problem...
        if (keepGoing) tasks.add(kj::evalLater([this]() { return messageLoop(); }));
      });
    }
  
    void handleMessage(kj::Own<IncomingRpcMessage> message) {
      auto reader = message->getBody().getAs<rpc::Message>();
  
      switch (reader.which()) {
        case rpc::Message::UNIMPLEMENTED:
          handleUnimplemented(reader.getUnimplemented());
          break;
  
        case rpc::Message::ABORT:
          handleAbort(reader.getAbort());
          break;
  
        case rpc::Message::BOOTSTRAP:
          handleBootstrap(kj::mv(message), reader.getBootstrap());
          break;
  
        case rpc::Message::CALL:
          handleCall(kj::mv(message), reader.getCall());
          break;
  
        case rpc::Message::RETURN:
          handleReturn(kj::mv(message), reader.getReturn());
          break;
  
        case rpc::Message::FINISH:
          handleFinish(reader.getFinish());
          break;
  
        case rpc::Message::RESOLVE:
          handleResolve(kj::mv(message), reader.getResolve());
          break;
  
        case rpc::Message::RELEASE:
          handleRelease(reader.getRelease());
          break;
  
        case rpc::Message::DISEMBARGO:
          handleDisembargo(reader.getDisembargo());
          break;
  
        default: {
          if (connection.is<Connected>()) {
            auto message = connection.get<Connected>()->newOutgoingMessage(
                firstSegmentSize(reader.totalSize(), messageSizeHint<void>()));
            message->getBody().initAs<rpc::Message>().setUnimplemented(reader);
            message->send();
          }
          break;
        }
      }
    }
  
    void handleUnimplemented(const rpc::Message::Reader& message) {
      switch (message.which()) {
        case rpc::Message::RESOLVE: {
          auto resolve = message.getResolve();
          switch (resolve.which()) {
            case rpc::Resolve::CAP: {
              auto cap = resolve.getCap();
              switch (cap.which()) {
                case rpc::CapDescriptor::NONE:
                  // Nothing to do (but this ought never to happen).
                  break;
                case rpc::CapDescriptor::SENDER_HOSTED:
                  releaseExport(cap.getSenderHosted(), 1);
                  break;
                case rpc::CapDescriptor::SENDER_PROMISE:
                  releaseExport(cap.getSenderPromise(), 1);
                  break;
                case rpc::CapDescriptor::RECEIVER_ANSWER:
                case rpc::CapDescriptor::RECEIVER_HOSTED:
                  // Nothing to do.
                  break;
                case rpc::CapDescriptor::THIRD_PARTY_HOSTED:
                  releaseExport(cap.getThirdPartyHosted().getVineId(), 1);
                  break;
              }
              break;
            }
            case rpc::Resolve::EXCEPTION:
              // Nothing to do.
              break;
          }
          break;
        }
  
        default:
          KJ_FAIL_ASSERT("Peer did not implement required RPC message type.", (uint)message.which());
          break;
      }
    }
  
    void handleAbort(const rpc::Exception::Reader& exception) {
      kj::throwRecoverableException(toException(exception));
    }
  
    // ---------------------------------------------------------------------------
    // Level 0
  
    class SingleCapPipeline: public PipelineHook, public kj::Refcounted {
    public:
      SingleCapPipeline(kj::Own<ClientHook>&& cap)
          : cap(kj::mv(cap)) {}
  
      kj::Own<PipelineHook> addRef() override {
        return kj::addRef(*this);
      }
  
      kj::Own<ClientHook> getPipelinedCap(kj::ArrayPtr<const PipelineOp> ops) override {
        if (ops.size() == 0) {
          return cap->addRef();
        } else {
          return newBrokenCap("Invalid pipeline transform.");
        }
      }
  
    private:
      kj::Own<ClientHook> cap;
    };
  
    void handleBootstrap(kj::Own<IncomingRpcMessage>&& message,
                         const rpc::Bootstrap::Reader& bootstrap) {
      AnswerId answerId = bootstrap.getQuestionId();
  
      if (!connection.is<Connected>()) {
        // Disconnected; ignore.
        return;
      }
  
      VatNetworkBase::Connection& conn = *connection.get<Connected>();
      auto response = conn.newOutgoingMessage(
          messageSizeHint<rpc::Return>() + sizeInWords<rpc::CapDescriptor>() + 32);
  
      rpc::Return::Builder ret = response->getBody().getAs<rpc::Message>().initReturn();
      ret.setAnswerId(answerId);
  
      kj::Own<ClientHook> capHook;
      kj::Array<ExportId> resultExports;
      KJ_DEFER(releaseExports(resultExports));  // in case something goes wrong
  
      // Call the restorer and initialize the answer.
      KJ_IF_MAYBE(exception, kj::runCatchingExceptions([&]() {
        Capability::Client cap = nullptr;
  
        if (bootstrap.hasDeprecatedObjectId()) {
          KJ_IF_MAYBE(r, restorer) {
            cap = r->baseRestore(bootstrap.getDeprecatedObjectId());
          } else {
            KJ_FAIL_REQUIRE("This vat only supports a bootstrap interface, not the old "
                            "Cap'n-Proto-0.4-style named exports.") { return; }
          }
        } else {
          cap = bootstrapFactory.baseCreateFor(conn.baseGetPeerVatId());
        }
  
        BuilderCapabilityTable capTable;
        auto payload = ret.initResults();
        capTable.imbue(payload.getContent()).setAs<Capability>(kj::mv(cap));
  
        auto capTableArray = capTable.getTable();
        KJ_DASSERT(capTableArray.size() == 1);
        kj::Vector<int> fds;
        resultExports = writeDescriptors(capTableArray, payload, fds);
        response->setFds(fds.releaseAsArray());
        capHook = KJ_ASSERT_NONNULL(capTableArray[0])->addRef();
      })) {
        fromException(*exception, ret.initException());
        capHook = newBrokenCap(kj::mv(*exception));
      }
  
      message = nullptr;
  
      // Add the answer to the answer table for pipelining and send the response.
      auto& answer = answers[answerId];
      KJ_REQUIRE(!answer.active, "questionId is already in use", answerId) {
        return;
      }
  
      answer.resultExports = kj::mv(resultExports);
      answer.active = true;
      answer.pipeline = kj::Own<PipelineHook>(kj::refcounted<SingleCapPipeline>(kj::mv(capHook)));
  
      response->send();
    }
  
    void handleCall(kj::Own<IncomingRpcMessage>&& message, const rpc::Call::Reader& call) {
      kj::Own<ClientHook> capability;
  
      KJ_IF_MAYBE(t, getMessageTarget(call.getTarget())) {
        capability = kj::mv(*t);
      } else {
        // Exception already reported.
        return;
      }
  
      bool redirectResults;
      switch (call.getSendResultsTo().which()) {
        case rpc::Call::SendResultsTo::CALLER:
          redirectResults = false;
          break;
        case rpc::Call::SendResultsTo::YOURSELF:
          redirectResults = true;
          break;
        default:
          KJ_FAIL_REQUIRE("Unsupported `Call.sendResultsTo`.") { return; }
      }
  
      auto payload = call.getParams();
      auto capTableArray = receiveCaps(payload.getCapTable(), message->getAttachedFds());
      auto cancelPaf = kj::newPromiseAndFulfiller<void>();
  
      AnswerId answerId = call.getQuestionId();
  
      auto context = kj::refcounted<RpcCallContext>(
          *this, answerId, kj::mv(message), kj::mv(capTableArray), payload.getContent(),
          redirectResults, kj::mv(cancelPaf.fulfiller),
          call.getInterfaceId(), call.getMethodId());
  
      // No more using `call` after this point, as it now belongs to the context.
  
      {
        auto& answer = answers[answerId];
  
        KJ_REQUIRE(!answer.active, "questionId is already in use") {
          return;
        }
  
        answer.active = true;
        answer.callContext = *context;
      }
  
      auto promiseAndPipeline = startCall(
          call.getInterfaceId(), call.getMethodId(), kj::mv(capability), context->addRef());
  
      // Things may have changed -- in particular if startCall() immediately called
      // context->directTailCall().
  
      {
        auto& answer = answers[answerId];
  
        answer.pipeline = kj::mv(promiseAndPipeline.pipeline);
  
        if (redirectResults) {
          auto resultsPromise = promiseAndPipeline.promise.then(
              kj::mvCapture(context, [](kj::Own<RpcCallContext>&& context) {
                return context->consumeRedirectedResponse();
              }));
  
          // If the call that later picks up `redirectedResults` decides to discard it, we need to
          // make sure our call is not itself canceled unless it has called allowCancellation().
          // So we fork the promise and join one branch with the cancellation promise, in order to
          // hold on to it.
          auto forked = resultsPromise.fork();
          answer.redirectedResults = forked.addBranch();
  
          cancelPaf.promise
              .exclusiveJoin(forked.addBranch().then([](kj::Own<RpcResponse>&&){}))
              .detach([](kj::Exception&&) {});
        } else {
          // Hack:  Both the success and error continuations need to use the context.  We could
          //   refcount, but both will be destroyed at the same time anyway.
          RpcCallContext* contextPtr = context;
  
          promiseAndPipeline.promise.then(
              [contextPtr]() {
                contextPtr->sendReturn();
              }, [contextPtr](kj::Exception&& exception) {
                contextPtr->sendErrorReturn(kj::mv(exception));
              }).catch_([&](kj::Exception&& exception) {
                // Handle exceptions that occur in sendReturn()/sendErrorReturn().
                taskFailed(kj::mv(exception));
              }).attach(kj::mv(context))
              .exclusiveJoin(kj::mv(cancelPaf.promise))
              .detach([](kj::Exception&&) {});
        }
      }
    }
  
    ClientHook::VoidPromiseAndPipeline startCall(
        uint64_t interfaceId, uint64_t methodId,
        kj::Own<ClientHook>&& capability, kj::Own<CallContextHook>&& context) {
      return capability->call(interfaceId, methodId, kj::mv(context));
    }
  
    kj::Maybe<kj::Own<ClientHook>> getMessageTarget(const rpc::MessageTarget::Reader& target) {
      switch (target.which()) {
        case rpc::MessageTarget::IMPORTED_CAP: {
          KJ_IF_MAYBE(exp, exports.find(target.getImportedCap())) {
            return exp->clientHook->addRef();
          } else {
            KJ_FAIL_REQUIRE("Message target is not a current export ID.") {
              return nullptr;
            }
          }
          break;
        }
  
        case rpc::MessageTarget::PROMISED_ANSWER: {
          auto promisedAnswer = target.getPromisedAnswer();
          kj::Own<PipelineHook> pipeline;
  
          auto& base = answers[promisedAnswer.getQuestionId()];
          KJ_REQUIRE(base.active, "PromisedAnswer.questionId is not a current question.") {
            return nullptr;
          }
          KJ_IF_MAYBE(p, base.pipeline) {
            pipeline = p->get()->addRef();
          } else {
            pipeline = newBrokenPipeline(KJ_EXCEPTION(FAILED,
                "Pipeline call on a request that returned no capabilities or was already closed."));
          }
  
          KJ_IF_MAYBE(ops, toPipelineOps(promisedAnswer.getTransform())) {
            return pipeline->getPipelinedCap(*ops);
          } else {
            // Exception already thrown.
            return nullptr;
          }
        }
  
        default:
          KJ_FAIL_REQUIRE("Unknown message target type.", target) {
            return nullptr;
          }
      }
  
      KJ_UNREACHABLE;
    }
  
    void handleReturn(kj::Own<IncomingRpcMessage>&& message, const rpc::Return::Reader& ret) {
      // Transitive destructors can end up manipulating the question table and invalidating our
      // pointer into it, so make sure these destructors run later.
      kj::Array<ExportId> exportsToRelease;
      KJ_DEFER(releaseExports(exportsToRelease));
      kj::Maybe<kj::Promise<kj::Own<RpcResponse>>> promiseToRelease;
  
      KJ_IF_MAYBE(question, questions.find(ret.getAnswerId())) {
        KJ_REQUIRE(question->isAwaitingReturn, "Duplicate Return.") { return; }
        question->isAwaitingReturn = false;
  
        if (ret.getReleaseParamCaps()) {
          exportsToRelease = kj::mv(question->paramExports);
        } else {
          question->paramExports = nullptr;
        }
  
        KJ_IF_MAYBE(questionRef, question->selfRef) {
          switch (ret.which()) {
            case rpc::Return::RESULTS: {
              KJ_REQUIRE(!question->isTailCall,
                  "Tail call `Return` must set `resultsSentElsewhere`, not `results`.") {
                return;
              }
  
              auto payload = ret.getResults();
              auto capTableArray = receiveCaps(payload.getCapTable(), message->getAttachedFds());
              questionRef->fulfill(kj::refcounted<RpcResponseImpl>(
                  *this, kj::addRef(*questionRef), kj::mv(message),
                  kj::mv(capTableArray), payload.getContent()));
              break;
            }
  
            case rpc::Return::EXCEPTION:
              KJ_REQUIRE(!question->isTailCall,
                  "Tail call `Return` must set `resultsSentElsewhere`, not `exception`.") {
                return;
              }
  
              questionRef->reject(toException(ret.getException()));
              break;
  
            case rpc::Return::CANCELED:
              KJ_FAIL_REQUIRE("Return message falsely claims call was canceled.") { return; }
              break;
  
            case rpc::Return::RESULTS_SENT_ELSEWHERE:
              KJ_REQUIRE(question->isTailCall,
                  "`Return` had `resultsSentElsewhere` but this was not a tail call.") {
                return;
              }
  
              // Tail calls are fulfilled with a null pointer.
              questionRef->fulfill(kj::Own<RpcResponse>());
              break;
  
            case rpc::Return::TAKE_FROM_OTHER_QUESTION:
              KJ_IF_MAYBE(answer, answers.find(ret.getTakeFromOtherQuestion())) {
                KJ_IF_MAYBE(response, answer->redirectedResults) {
                  questionRef->fulfill(kj::mv(*response));
                  answer->redirectedResults = nullptr;
  
                  KJ_IF_MAYBE(context, answer->callContext) {
                    // Send the `Return` message  for the call of which we're taking ownership, so
                    // that the peer knows it can now tear down the call state.
                    context->sendRedirectReturn();
  
                    // There are three conditions, all of which must be true, before a call is
                    // canceled:
                    // 1. The RPC opts in by calling context->allowCancellation().
                    // 2. We request cancellation with context->requestCancel().
                    // 3. The final response promise -- which we passed to questionRef->fulfill()
                    //    above -- must be dropped.
                    //
                    // We would like #3 to imply #2. So... we can just make #2 be true.
                    context->requestCancel();
                  }
                } else {
                  KJ_FAIL_REQUIRE("`Return.takeFromOtherQuestion` referenced a call that did not "
                                  "use `sendResultsTo.yourself`.") { return; }
                }
              } else {
                KJ_FAIL_REQUIRE("`Return.takeFromOtherQuestion` had invalid answer ID.") { return; }
              }
  
              break;
  
            default:
              KJ_FAIL_REQUIRE("Unknown 'Return' type.") { return; }
          }
        } else {
          // This is a response to a question that we canceled earlier.
  
          if (ret.isTakeFromOtherQuestion()) {
            // This turned out to be a tail call back to us! We now take ownership of the tail call.
            // Since the caller canceled, we need to cancel out the tail call, if it still exists.
  
            KJ_IF_MAYBE(answer, answers.find(ret.getTakeFromOtherQuestion())) {
              // Indeed, it does still exist.
  
              // Throw away the result promise.
              promiseToRelease = kj::mv(answer->redirectedResults);
  
              KJ_IF_MAYBE(context, answer->callContext) {
                // Send the `Return` message  for the call of which we're taking ownership, so
                // that the peer knows it can now tear down the call state.
                context->sendRedirectReturn();
  
                // Since the caller has been canceled, make sure the callee that we're tailing to
                // gets canceled.
                context->requestCancel();
              }
            }
          }
  
          // Looks like this question was canceled earlier, so `Finish` was already sent, with
          // `releaseResultCaps` set true so that we don't have to release them here.  We can go
          // ahead and delete it from the table.
          questions.erase(ret.getAnswerId(), *question);
        }
  
      } else {
        KJ_FAIL_REQUIRE("Invalid question ID in Return message.") { return; }
      }
    }
  
    void handleFinish(const rpc::Finish::Reader& finish) {
      // Delay release of these things until return so that transitive destructors don't accidentally
      // modify the answer table and invalidate our pointer into it.
      kj::Array<ExportId> exportsToRelease;
      KJ_DEFER(releaseExports(exportsToRelease));
      Answer answerToRelease;
      kj::Maybe<kj::Own<PipelineHook>> pipelineToRelease;
  
      KJ_IF_MAYBE(answer, answers.find(finish.getQuestionId())) {
        KJ_REQUIRE(answer->active, "'Finish' for invalid question ID.") { return; }
  
        if (finish.getReleaseResultCaps()) {
          exportsToRelease = kj::mv(answer->resultExports);
        } else {
          answer->resultExports = nullptr;
        }
  
        pipelineToRelease = kj::mv(answer->pipeline);
  
        // If the call isn't actually done yet, cancel it.  Otherwise, we can go ahead and erase the
        // question from the table.
        KJ_IF_MAYBE(context, answer->callContext) {
          context->requestCancel();
        } else {
          answerToRelease = answers.erase(finish.getQuestionId());
        }
      } else {
        KJ_FAIL_REQUIRE("'Finish' for invalid question ID.") { return; }
      }
    }
  
    // ---------------------------------------------------------------------------
    // Level 1
  
    void handleResolve(kj::Own<IncomingRpcMessage>&& message, const rpc::Resolve::Reader& resolve) {
      kj::Own<ClientHook> replacement;
      kj::Maybe<kj::Exception> exception;
  
      // Extract the replacement capability.
      switch (resolve.which()) {
        case rpc::Resolve::CAP:
          KJ_IF_MAYBE(cap, receiveCap(resolve.getCap(), message->getAttachedFds())) {
            replacement = kj::mv(*cap);
          } else {
            KJ_FAIL_REQUIRE("'Resolve' contained 'CapDescriptor.none'.") { return; }
          }
          break;
  
        case rpc::Resolve::EXCEPTION:
          // We can't set `replacement` to a new broken cap here because this will confuse
          // PromiseClient::Resolve() into thinking that the remote promise resolved to a local
          // capability and therefore a Disembargo is needed. We must actually reject the promise.
          exception = toException(resolve.getException());
          break;
  
        default:
          KJ_FAIL_REQUIRE("Unknown 'Resolve' type.") { return; }
      }
  
      // If the import is on the table, fulfill it.
      KJ_IF_MAYBE(import, imports.find(resolve.getPromiseId())) {
        KJ_IF_MAYBE(fulfiller, import->promiseFulfiller) {
          // OK, this is in fact an unfulfilled promise!
          KJ_IF_MAYBE(e, exception) {
            fulfiller->get()->reject(kj::mv(*e));
          } else {
            fulfiller->get()->fulfill(kj::mv(replacement));
          }
        } else if (import->importClient != nullptr) {
          // It appears this is a valid entry on the import table, but was not expected to be a
          // promise.
          KJ_FAIL_REQUIRE("Got 'Resolve' for a non-promise import.") { break; }
        }
      }
    }
  
    void handleRelease(const rpc::Release::Reader& release) {
      releaseExport(release.getId(), release.getReferenceCount());
    }
  
    void releaseExport(ExportId id, uint refcount) {
      KJ_IF_MAYBE(exp, exports.find(id)) {
        KJ_REQUIRE(refcount <= exp->refcount, "Tried to drop export's refcount below zero.") {
          return;
        }
  
        exp->refcount -= refcount;
        if (exp->refcount == 0) {
          exportsByCap.erase(exp->clientHook);
          exports.erase(id, *exp);
        }
      } else {
        KJ_FAIL_REQUIRE("Tried to release invalid export ID.") {
          return;
        }
      }
    }
  
    void releaseExports(kj::ArrayPtr<ExportId> exports) {
      for (auto exportId: exports) {
        releaseExport(exportId, 1);
      }
    }
  
    void handleDisembargo(const rpc::Disembargo::Reader& disembargo) {
      auto context = disembargo.getContext();
      switch (context.which()) {
        case rpc::Disembargo::Context::SENDER_LOOPBACK: {
          kj::Own<ClientHook> target;
  
          KJ_IF_MAYBE(t, getMessageTarget(disembargo.getTarget())) {
            target = kj::mv(*t);
          } else {
            // Exception already reported.
            return;
          }
  
          for (;;) {
            KJ_IF_MAYBE(r, target->getResolved()) {
              target = r->addRef();
            } else {
              break;
            }
          }
  
          KJ_REQUIRE(target->getBrand() == this,
                     "'Disembargo' of type 'senderLoopback' sent to an object that does not point "
                     "back to the sender.") {
            return;
          }
  
          EmbargoId embargoId = context.getSenderLoopback();
  
          // We need to insert an evalLast() here to make sure that any pending calls towards this
          // cap have had time to find their way through the event loop.
          tasks.add(canceler.wrap(kj::evalLast(kj::mvCapture(
              target, [this,embargoId](kj::Own<ClientHook>&& target) {
            if (!connection.is<Connected>()) {
              return;
            }
  
            RpcClient& downcasted = kj::downcast<RpcClient>(*target);
  
            auto message = connection.get<Connected>()->newOutgoingMessage(
                messageSizeHint<rpc::Disembargo>() + MESSAGE_TARGET_SIZE_HINT);
            auto builder = message->getBody().initAs<rpc::Message>().initDisembargo();
  
            {
              auto redirect = downcasted.writeTarget(builder.initTarget());
  
              // Disembargoes should only be sent to capabilities that were previously the subject of
              // a `Resolve` message.  But `writeTarget` only ever returns non-null when called on
              // a PromiseClient.  The code which sends `Resolve` and `Return` should have replaced
              // any promise with a direct node in order to solve the Tribble 4-way race condition.
              // See the documentation of Disembargo in rpc.capnp for more.
              KJ_REQUIRE(redirect == nullptr,
                         "'Disembargo' of type 'senderLoopback' sent to an object that does not "
                         "appear to have been the subject of a previous 'Resolve' message.") {
                return;
              }
            }
  
            builder.getContext().setReceiverLoopback(embargoId);
  
            message->send();
          }))));
  
          break;
        }
  
        case rpc::Disembargo::Context::RECEIVER_LOOPBACK: {
          KJ_IF_MAYBE(embargo, embargoes.find(context.getReceiverLoopback())) {
            KJ_ASSERT_NONNULL(embargo->fulfiller)->fulfill();
            embargoes.erase(context.getReceiverLoopback(), *embargo);
          } else {
            KJ_FAIL_REQUIRE("Invalid embargo ID in 'Disembargo.context.receiverLoopback'.") {
              return;
            }
          }
          break;
        }
  
        default:
          KJ_FAIL_REQUIRE("Unimplemented Disembargo type.", disembargo) { return; }
      }
    }
  
    // ---------------------------------------------------------------------------
    // Level 2
  };
  
  }  // namespace
  
  class RpcSystemBase::Impl final: private BootstrapFactoryBase, private kj::TaskSet::ErrorHandler {
  public:
    Impl(VatNetworkBase& network, kj::Maybe<Capability::Client> bootstrapInterface)
        : network(network), bootstrapInterface(kj::mv(bootstrapInterface)),
          bootstrapFactory(*this), tasks(*this) {
      acceptLoopPromise = acceptLoop().eagerlyEvaluate([](kj::Exception&& e) { KJ_LOG(ERROR, e); });
    }
    Impl(VatNetworkBase& network, BootstrapFactoryBase& bootstrapFactory)
        : network(network), bootstrapFactory(bootstrapFactory), tasks(*this) {
      acceptLoopPromise = acceptLoop().eagerlyEvaluate([](kj::Exception&& e) { KJ_LOG(ERROR, e); });
    }
    Impl(VatNetworkBase& network, SturdyRefRestorerBase& restorer)
        : network(network), bootstrapFactory(*this), restorer(restorer), tasks(*this) {
      acceptLoopPromise = acceptLoop().eagerlyEvaluate([](kj::Exception&& e) { KJ_LOG(ERROR, e); });
    }
  
    ~Impl() noexcept(false) {
      unwindDetector.catchExceptionsIfUnwinding([&]() {
        // std::unordered_map doesn't like it when elements' destructors throw, so carefully
        // disassemble it.
        if (!connections.empty()) {
          kj::Vector<kj::Own<RpcConnectionState>> deleteMe(connections.size());
          kj::Exception shutdownException = KJ_EXCEPTION(DISCONNECTED, "RpcSystem was destroyed.");
          for (auto& entry: connections) {
            entry.second->disconnect(kj::cp(shutdownException));
            deleteMe.add(kj::mv(entry.second));
          }
        }
      });
    }
  
    Capability::Client bootstrap(AnyStruct::Reader vatId) {
      // For now we delegate to restore() since it's equivalent, but eventually we'll remove restore()
      // and implement bootstrap() directly.
      return restore(vatId, AnyPointer::Reader());
    }
  
    Capability::Client restore(AnyStruct::Reader vatId, AnyPointer::Reader objectId) {
      KJ_IF_MAYBE(connection, network.baseConnect(vatId)) {
        auto& state = getConnectionState(kj::mv(*connection));
        return Capability::Client(state.restore(objectId));
      } else if (objectId.isNull()) {
        // Turns out `vatId` refers to ourselves, so we can also pass it as the client ID for
        // baseCreateFor().
        return bootstrapFactory.baseCreateFor(vatId);
      } else KJ_IF_MAYBE(r, restorer) {
        return r->baseRestore(objectId);
      } else {
        return Capability::Client(newBrokenCap(
            "This vat only supports a bootstrap interface, not the old Cap'n-Proto-0.4-style "
            "named exports."));
      }
    }
  
    void setFlowLimit(size_t words) {
      flowLimit = words;
  
      for (auto& conn: connections) {
        conn.second->setFlowLimit(words);
      }
    }
  
    void setTraceEncoder(kj::Function<kj::String(const kj::Exception&)> func) {
      traceEncoder = kj::mv(func);
    }
  
    kj::Promise<void> run() { return kj::mv(acceptLoopPromise); }
  
  private:
    VatNetworkBase& network;
    kj::Maybe<Capability::Client> bootstrapInterface;
    BootstrapFactoryBase& bootstrapFactory;
    kj::Maybe<SturdyRefRestorerBase&> restorer;
    size_t flowLimit = kj::maxValue;
    kj::Maybe<kj::Function<kj::String(const kj::Exception&)>> traceEncoder;
    kj::Promise<void> acceptLoopPromise = nullptr;
    kj::TaskSet tasks;
  
    typedef std::unordered_map<VatNetworkBase::Connection*, kj::Own<RpcConnectionState>>
        ConnectionMap;
    ConnectionMap connections;
  
    kj::UnwindDetector unwindDetector;
  
    RpcConnectionState& getConnectionState(kj::Own<VatNetworkBase::Connection>&& connection) {
      auto iter = connections.find(connection);
      if (iter == connections.end()) {
        VatNetworkBase::Connection* connectionPtr = connection;
        auto onDisconnect = kj::newPromiseAndFulfiller<RpcConnectionState::DisconnectInfo>();
        tasks.add(onDisconnect.promise
            .then([this,connectionPtr](RpcConnectionState::DisconnectInfo info) {
          connections.erase(connectionPtr);
          tasks.add(kj::mv(info.shutdownPromise));
        }));
        auto newState = kj::refcounted<RpcConnectionState>(
            bootstrapFactory, restorer, kj::mv(connection),
            kj::mv(onDisconnect.fulfiller), flowLimit, traceEncoder);
        RpcConnectionState& result = *newState;
        connections.insert(std::make_pair(connectionPtr, kj::mv(newState)));
        return result;
      } else {
        return *iter->second;
      }
    }
  
    kj::Promise<void> acceptLoop() {
      return network.baseAccept().then(
          [this](kj::Own<VatNetworkBase::Connection>&& connection) {
        getConnectionState(kj::mv(connection));
        return acceptLoop();
      });
    }
  
    Capability::Client baseCreateFor(AnyStruct::Reader clientId) override {
      // Implements BootstrapFactory::baseCreateFor() in terms of `bootstrapInterface` or `restorer`,
      // for use when we were given one of those instead of an actual `bootstrapFactory`.
  
      KJ_IF_MAYBE(cap, bootstrapInterface) {
        return *cap;
      } else KJ_IF_MAYBE(r, restorer) {
        return r->baseRestore(AnyPointer::Reader());
      } else {
        return KJ_EXCEPTION(FAILED, "This vat does not expose any public/bootstrap interfaces.");
      }
    }
  
    void taskFailed(kj::Exception&& exception) override {
      KJ_LOG(ERROR, exception);
    }
  };
  
  RpcSystemBase::RpcSystemBase(VatNetworkBase& network,
                               kj::Maybe<Capability::Client> bootstrapInterface)
      : impl(kj::heap<Impl>(network, kj::mv(bootstrapInterface))) {}
  RpcSystemBase::RpcSystemBase(VatNetworkBase& network,
                               BootstrapFactoryBase& bootstrapFactory)
      : impl(kj::heap<Impl>(network, bootstrapFactory)) {}
  RpcSystemBase::RpcSystemBase(VatNetworkBase& network, SturdyRefRestorerBase& restorer)
      : impl(kj::heap<Impl>(network, restorer)) {}
  RpcSystemBase::RpcSystemBase(RpcSystemBase&& other) noexcept = default;
  RpcSystemBase::~RpcSystemBase() noexcept(false) {}
  
  Capability::Client RpcSystemBase::baseBootstrap(AnyStruct::Reader vatId) {
    return impl->bootstrap(vatId);
  }
  
  Capability::Client RpcSystemBase::baseRestore(
      AnyStruct::Reader hostId, AnyPointer::Reader objectId) {
    return impl->restore(hostId, objectId);
  }
  
  void RpcSystemBase::baseSetFlowLimit(size_t words) {
    return impl->setFlowLimit(words);
  }
  
  void RpcSystemBase::setTraceEncoder(kj::Function<kj::String(const kj::Exception&)> func) {
    impl->setTraceEncoder(kj::mv(func));
  }
  
  kj::Promise<void> RpcSystemBase::run() {
    return impl->run();
  }
  
  }  // namespace _ (private)
  
  // =======================================================================================
  
  namespace {
  
  class WindowFlowController final: public RpcFlowController, private kj::TaskSet::ErrorHandler {
  public:
    WindowFlowController(RpcFlowController::WindowGetter& windowGetter)
        : windowGetter(windowGetter), tasks(*this) {
      state.init<Running>();
    }
  
    kj::Promise<void> send(kj::Own<OutgoingRpcMessage> message, kj::Promise<void> ack) override {
      auto size = message->sizeInWords() * sizeof(capnp::word);
      maxMessageSize = kj::max(size, maxMessageSize);
  
      // We are REQUIRED to send the message NOW to maintain correct ordering.
      message->send();
  
      inFlight += size;
      tasks.add(ack.then([this, size]() {
        inFlight -= size;
        KJ_SWITCH_ONEOF(state) {
          KJ_CASE_ONEOF(blockedSends, Running) {
            if (isReady()) {
              // Release all fulfillers.
              for (auto& fulfiller: blockedSends) {
                fulfiller->fulfill();
              }
              blockedSends.clear();
  
            }
  
            KJ_IF_MAYBE(f, emptyFulfiller) {
              if (inFlight == 0) {
                f->get()->fulfill(tasks.onEmpty());
              }
            }
          }
          KJ_CASE_ONEOF(exception, kj::Exception) {
            // A previous call failed, but this one -- which was already in-flight at the time --
            // ended up succeeding. That may indicate that the server side is not properly
            // handling streaming error propagation. Nothing much we can do about it here though.
          }
        }
      }));
  
      KJ_SWITCH_ONEOF(state) {
        KJ_CASE_ONEOF(blockedSends, Running) {
          if (isReady()) {
            return kj::READY_NOW;
          } else {
            auto paf = kj::newPromiseAndFulfiller<void>();
            blockedSends.add(kj::mv(paf.fulfiller));
            return kj::mv(paf.promise);
          }
        }
        KJ_CASE_ONEOF(exception, kj::Exception) {
          return kj::cp(exception);
        }
      }
      KJ_UNREACHABLE;
    }
  
    kj::Promise<void> waitAllAcked() override {
      KJ_IF_MAYBE(q, state.tryGet<Running>()) {
        if (!q->empty()) {
          auto paf = kj::newPromiseAndFulfiller<kj::Promise<void>>();
          emptyFulfiller = kj::mv(paf.fulfiller);
          return kj::mv(paf.promise);
        }
      }
      return tasks.onEmpty();
    }
  
  private:
    RpcFlowController::WindowGetter& windowGetter;
    size_t inFlight = 0;
    size_t maxMessageSize = 0;
  
    typedef kj::Vector<kj::Own<kj::PromiseFulfiller<void>>> Running;
    kj::OneOf<Running, kj::Exception> state;
  
    kj::Maybe<kj::Own<kj::PromiseFulfiller<kj::Promise<void>>>> emptyFulfiller;
  
    kj::TaskSet tasks;
  
    void taskFailed(kj::Exception&& exception) override {
      KJ_SWITCH_ONEOF(state) {
        KJ_CASE_ONEOF(blockedSends, Running) {
          // Fail out all pending sends.
          for (auto& fulfiller: blockedSends) {
            fulfiller->reject(kj::cp(exception));
          }
          // Fail out all future sends.
          state = kj::mv(exception);
        }
        KJ_CASE_ONEOF(exception, kj::Exception) {
          // ignore redundant exception
        }
      }
    }
  
    bool isReady() {
      // We extend the window by maxMessageSize to avoid a pathological situation when a message
      // is larger than the window size. Otherwise, after sending that message, we would end up
      // not sending any others until the ack was received, wasting a round trip's worth of
      // bandwidth.
      return inFlight <= maxMessageSize  // avoid getWindow() call if unnecessary
          || inFlight < windowGetter.getWindow() + maxMessageSize;
    }
  };
  
  class FixedWindowFlowController final
      : public RpcFlowController, public RpcFlowController::WindowGetter {
  public:
    FixedWindowFlowController(size_t windowSize): windowSize(windowSize), inner(*this) {}
  
    kj::Promise<void> send(kj::Own<OutgoingRpcMessage> message, kj::Promise<void> ack) override {
      return inner.send(kj::mv(message), kj::mv(ack));
    }
  
    kj::Promise<void> waitAllAcked() override {
      return inner.waitAllAcked();
    }
  
    size_t getWindow() override { return windowSize; }
  
  private:
    size_t windowSize;
    WindowFlowController inner;
  };
  
  }  // namespace
  
  kj::Own<RpcFlowController> RpcFlowController::newFixedWindowController(size_t windowSize) {
    return kj::heap<FixedWindowFlowController>(windowSize);
  }
  kj::Own<RpcFlowController> RpcFlowController::newVariableWindowController(WindowGetter& getter) {
    return kj::heap<WindowFlowController>(getter);
  }
  
  }  // namespace capnp
  