Support "flat" encoding, and "packed-flat" too.

This seems correct, and works around a bug in Sandstorm where a capability shared between multiple components is sometimes closed by one of them. This isn't the best way to fix the bug, but the real fix will require some refactoring.

src/node-capnp/capnp.cc

@@ -47,6 +47,8 @@
 #include <set>
 #include <stdlib.h>
 #include <sys/uio.h>
+#include <sys/ioctl.h>
+#include <sys/poll.h>
 
 #include <typeinfo>
 #include <typeindex>
@@ -58,6 +60,8 @@ namespace {  // so we get warnings if anything declared in this file is left und
 typedef unsigned char byte;
 typedef unsigned int uint;
 
+bool verboseDebugLogging = false;
+
 // =======================================================================================
 // KJ <-> libuv glue.
 
@@ -145,6 +149,7 @@ public:
   }
 
   void setRunnable(bool runnable) override {
+    if (verboseDebugLogging) { KJ_LOG(INFO, "setRunnable", runnable, this->runnable, this->scheduled); }
     if (runnable != this->runnable) {
       this->runnable = runnable;
       if (runnable && !scheduled) {
@@ -153,6 +158,10 @@ public:
     }
   }
 
+  void logState() {
+    KJ_LOG(INFO, "UvEventPort state", runnable, scheduled);
+  }
+
 private:
   uv_loop_t* loop;
   UvHandle<uv_timer_t> timer;
@@ -161,11 +170,21 @@ private:
   bool scheduled = false;
 
   void schedule() {
+    if (verboseDebugLogging) { KJ_LOG(INFO, "schedule"); }
     UV_CALL(uv_timer_start(timer, &doRun, 0, 0), loop);
+    if (verboseDebugLogging) { KJ_LOG(INFO, "scheduled"); }
     scheduled = true;
   }
 
   void run() {
+    if (verboseDebugLogging) {
+      KJ_LOG(INFO, "run");
+      kj::evalLater([]() {
+        KJ_LOG(INFO, "evalLater() executed");
+      }).detach([](kj::Exception&& e) {
+        KJ_LOG(ERROR, e);
+      });
+    }
     KJ_ASSERT(scheduled);
 
     UV_CALL(uv_timer_stop(timer), loop);
@@ -183,6 +202,8 @@ private:
     } else {
       scheduled = false;
     }
+
+    if (verboseDebugLogging) { KJ_LOG(INFO, "run done"); }
   }
 
   static void doRun(uv_timer_t* handle, int status) {
@@ -253,11 +274,17 @@ public:
       : uvLoop(loop), fd(applyFlags(fd, flags)), flags(flags),
         uvPoller(uv_poll_init, uvLoop, fd) {
     uvPoller->data = this;
+    // TODO(soon): This uv_poll_start() is redundant.
     UV_CALL(uv_poll_start(uvPoller, 0, &pollCallback), uvLoop);
+    if (verboseDebugLogging) { KJ_LOG(INFO, "created socket", fd); }
   }
 
   ~OwnedFileDescriptor() noexcept(false) {
+    if (verboseDebugLogging) { KJ_LOG(INFO, "destroying socket", fd); }
+
     if (!stopped) {
+      // TODO(soon): This uv_poll_stop() is redundant so long as we can make sure that uvPoller is
+      //   destroyed before we close the descriptor.
       UV_CALL(uv_poll_stop(uvPoller), uvLoop);
     }
 
@@ -281,6 +308,8 @@ public:
     int flags = UV_READABLE | (writable == nullptr ? 0 : UV_WRITABLE);
     UV_CALL(uv_poll_start(uvPoller, flags, &pollCallback), uvLoop);
 
+    if (verboseDebugLogging) { KJ_LOG(INFO, "waiting for readable", this, fd); }
+
     return kj::mv(paf.promise);
   }
 
@@ -295,6 +324,8 @@ public:
     int flags = UV_WRITABLE | (readable == nullptr ? 0 : UV_READABLE);
     UV_CALL(uv_poll_start(uvPoller, flags, &pollCallback), uvLoop);
 
+    if (verboseDebugLogging) { KJ_LOG(INFO, "waiting for writable", this, fd); }
+
     return kj::mv(paf.promise);
   }
 
@@ -302,6 +333,22 @@ protected:
   uv_loop_t* const uvLoop;
   const int fd;
 
+  void logInfo() {
+    KJ_ASSERT(verboseDebugLogging);
+    int available;
+    KJ_SYSCALL(ioctl(fd, FIONREAD, &available));
+
+    struct pollfd pfd;
+    memset(&pfd, 0, sizeof(pfd));
+    pfd.fd = fd;
+    pfd.events = POLLIN | POLLOUT;
+    int pollResult = poll(&pfd, 1, 0);
+    KJ_LOG(INFO, "closing socket", fd, available, pollResult, pfd.revents,
+           uvPoller->io_watcher.events, uvPoller->io_watcher.pevents,
+           uvPoller->io_watcher.fd, &uvPoller->io_watcher, uvLoop->watchers[fd],
+           ngx_queue_empty(&uvPoller->io_watcher.watcher_queue));
+  }
+
 private:
   uint flags;
   kj::Maybe<kj::Own<kj::PromiseFulfiller<void>>> readable;
@@ -314,6 +361,10 @@ private:
   }
 
   void pollDone(int status, int events) {
+    if (verboseDebugLogging) {
+      KJ_LOG(INFO, "poll event", this, fd, status, events, readable == nullptr, writable == nullptr);
+    }
+
     if (status != 0) {
       // Error.  libuv produces a non-zero status if polling produced POLLERR.  The error code
       // reported by libuv is always EBADF, even if the file descriptor is perfectly legitimate but
@@ -409,6 +460,10 @@ public:
   }
 
   void shutdownWrite() override {
+    if (verboseDebugLogging) {
+      logInfo();
+    }
+
     // There's no legitimate way to get an AsyncStreamFd that isn't a socket through the
     // UnixAsyncIoProvider interface.
     KJ_SYSCALL(shutdown(fd, SHUT_WR));
@@ -426,6 +481,8 @@ private:
       return alreadyRead;
     }
 
+    if (verboseDebugLogging) { KJ_LOG(INFO, "did read", fd, n); }
+
     if (n < 0) {
       // Read would block.
       return onReadable().then([=]() {
@@ -482,6 +539,8 @@ private:
       return kj::READY_NOW;
     }
 
+    if (verboseDebugLogging) { KJ_LOG(INFO, "did write", fd, writeResult); }
+
     // A negative result means EAGAIN, which we can treat the same as having written zero bytes.
     size_t n = writeResult < 0 ? 0 : writeResult;
 
@@ -615,6 +674,8 @@ public:
     KJ_FAIL_ASSERT("Timers not implemented.");
   }
 
+  void logState() { return eventPort.logState(); }
+
 private:
   UvEventPort eventPort;
   kj::WaitScope waitScope;
@@ -2153,6 +2214,8 @@ v8::Handle<v8::Value> send(const v8::Arguments& args) {
     OwnHandle<v8::Function> callback = v8::Handle<v8::Function>(v8::Function::Cast(*args[1]));
     OwnHandle<v8::Function> errorCallback = v8::Handle<v8::Function>(v8::Function::Cast(*args[2]));
 
+    if (verboseDebugLogging) { KJ_LOG(INFO, "making call"); }
+
     auto promise = KJ_REQUIRE_NONNULL(request.request, "Request already sent.").send();
     request.request = nullptr;
 
@@ -2170,6 +2233,7 @@ v8::Handle<v8::Value> send(const v8::Arguments& args) {
         .then(kj::mvCapture(callback,
           [&context](OwnHandle<v8::Function>&& callback,
                      capnp::Response<capnp::DynamicStruct>&& response) {
+      if (verboseDebugLogging) { KJ_LOG(INFO, "call return"); }
       v8::HandleScope scope;
       v8::Handle<v8::Value> args[1] = {
         context.wrapper.wrapCopy(ClientResponse { kj::mv(response) })
@@ -2185,6 +2249,7 @@ v8::Handle<v8::Value> send(const v8::Arguments& args) {
     })).detach(kj::mvCapture(errorCallback,
           [&context](OwnHandle<v8::Function>&& errorCallback,
                      kj::Exception&& exception) {
+      if (verboseDebugLogging) { KJ_LOG(INFO, "call failed"); }
       v8::HandleScope scope;
       v8::Handle<v8::Value> args[1] = { toJsException(kj::mv(exception)) };
       v8::TryCatch tryCatch;
@@ -2348,16 +2413,7 @@ v8::Handle<v8::Value> fulfillPromisedCap(const v8::Arguments& args) {
   KJV8_UNWRAP(capnp::DynamicCapability::Client, cap, args[1]);
 
   return liftKj([&]() -> v8::Handle<v8::Value> {
-    auto schema = cap.getSchema();
-
     fulfiller->fulfill(kj::mv(cap));
-
-    // Avoid segfault if capability is called again.
-    cap = capnp::Capability::Client(
-        capnp::newBrokenCap(KJ_EXCEPTION(DISCONNECTED,
-            "Capability was used to resolve a promise capability and so has been closed.")))
-        .castAs<capnp::DynamicCapability>(schema);
-
     return v8::Undefined();
   });
 }
@@ -2465,6 +2521,45 @@ v8::Handle<v8::Value> throw_(const v8::Arguments& args) {
   });
 }
 
+v8::Handle<v8::Value> enableVerboseDebugLogging(const v8::Arguments& args) {
+  // enableVerboseDebugLogging(flag) -> void
+  //
+  // If `flag` is true, turn on advanced debug logging.
+
+  KJV8_UNWRAP(CapnpContext, context, args.Data());
+
+  return liftKj([&]() -> v8::Handle<v8::Value> {
+    bool enable = args[0]->BooleanValue();
+    if (enable != verboseDebugLogging) {
+      verboseDebugLogging = enable;
+      if (verboseDebugLogging) {
+        kj::_::Debug::setLogLevel(kj::LogSeverity::INFO);
+        KJ_LOG(INFO, "node-capnp ulta-verbose debug logging now enabled; hope you track down the bug!");
+      } else {
+        KJ_LOG(INFO, "node-capnp ulta-verbose debug logging stopping here");
+        kj::_::Debug::setLogLevel(kj::LogSeverity::WARNING);
+      }
+    }
+
+    if (enable) {
+      context.llaiop.logState();
+    }
+
+    return v8::Undefined();
+  });
+}
+
+v8::Handle<v8::Value> methodName(const v8::Arguments& args) {
+  // methodName(method) -> string
+
+  KJV8_UNWRAP(CapnpContext, context, args.Data());
+  KJV8_UNWRAP(capnp::InterfaceSchema::Method, method, args[0]);
+
+  return liftKj([&]() -> v8::Handle<v8::Value> {
+    return v8::String::NewSymbol(method.getProto().getName().cStr());
+  });
+}
+
 // -----------------------------------------------------------------------------
 
 void init(v8::Handle<v8::Object> exports) {
@@ -2510,6 +2605,9 @@ void init(v8::Handle<v8::Object> exports) {
     mapFunction("return_", return_);
     mapFunction("throw_", throw_);
 
+    mapFunction("enableVerboseDebugLogging", enableVerboseDebugLogging);
+    mapFunction("methodName", methodName);
+
     return emptyHandle;
   });
 }

src/node-capnp/capnp.js

@@ -153,14 +153,27 @@ function makeMethod(cap, method) {
     v8capnp.fromJs(req, Array.prototype.slice.call(arguments, 0), LocalCapWrapper);
     var pipeline;
     var promise = new Promise(function (resolve, reject) {
-      pipeline = v8capnp.send(req, resolve, reject, Capability);
+      pipeline = v8capnp.send(req, function (x) {
+        if (verboseDebugLogging) console.log("capnp.js: returned to JS 1", v8capnp.methodName(method));
+        resolve(x);
+      }, function (x) {
+        if (verboseDebugLogging) console.log("capnp.js: threw to JS 1", v8capnp.methodName(method), x);
+        reject(x);
+      }, Capability);
     }).then(function (response) {
+      if (verboseDebugLogging) console.log("capnp.js: returned to JS 2", v8capnp.methodName(method));
       var result = v8capnp.toJs(response, Capability);
       v8capnp.release(response);
       settleCaps(pipeline, result);
+      if (verboseDebugLogging) console.log("capnp.js: returned to JS 3", v8capnp.methodName(method));
       return result;
+    }, function (err) {
+      if (verboseDebugLogging) console.log("capnp.js: threw to JS 2", v8capnp.methodName(method));
+      throw err;
     });
+    if (verboseDebugLogging) console.log("capnp.js: before makeRemotePromise:", v8capnp.methodName(method), promise);
     makeRemotePromise(promise, pipeline);
+    if (verboseDebugLogging) console.log("capnp.js: after makeRemotePromise:", v8capnp.methodName(method), promise);
     return promise;
   }
 }
@@ -286,3 +299,9 @@ exports.bytesToPreorder = function(schema, buf, options) {
   return v8capnp.toBytes(builder, options || {});
 }
 
+var verboseDebugLogging = false;
+exports.enableVerboseDebugLogging = function (flag) {
+  verboseDebugLogging = flag;
+  // Turns out the problem isn't even in C++, so don't  bother enabling its verbose debug logging.
+  //v8capnp.enableVerboseDebugLogging(flag);
+}