quic: getting application working, fixing bugs

jasnell · jasnell · commit cffe9b2357b0 · 2024-01-06T08:54:42.000-08:00
diff --git a/src/quic/application.cc b/src/quic/application.cc
@@ -1,3 +1,4 @@
+#include "ngtcp2/ngtcp2.h"
 #if HAVE_OPENSSL && NODE_OPENSSL_HAS_QUIC
 
 #include "application.h"
@@ -95,6 +96,20 @@ Maybe<Session::Application_Options> Session::Application_Options::From(
   return Just<Application_Options>(options);
 }
 
+// ============================================================================
+
+std::string Session::Application::StreamData::ToString() const {
+  DebugIndentScope indent;
+  auto prefix = indent.Prefix();
+  std::string res("{");
+  res += prefix + "count: " + std::to_string(count);
+  res += prefix + "remaining: " + std::to_string(remaining);
+  res += prefix + "id: " + std::to_string(id);
+  res += prefix + "fin: " + std::to_string(fin);
+  res += indent.Close();
+  return res;
+}
+
 Session::Application::Application(Session* session, const Options& options)
     : session_(session) {}
 
@@ -251,13 +266,15 @@ void Session::Application::SendPendingData() {
   };
 
   for (;;) {
-    ssize_t ndatalen;
+    ssize_t ndatalen = 0;
     StreamData stream_data;
 
     err = GetStreamData(&stream_data);
 
     if (err < 0) {
       session_->last_error_ = QuicError::ForNgtcp2Error(NGTCP2_ERR_INTERNAL);
+      Debug(session_, "Application failed to get stream data with error %s",
+            session_->last_error_);
       return session_->Close(Session::CloseMethod::SILENT);
     }
 
@@ -270,13 +287,19 @@ void Session::Application::SendPendingData() {
       pos = ngtcp2_vec(*packet).base;
     }
 
+    Debug(session_, "Application using stream data: %s", stream_data);
+    DCHECK_NOT_NULL(pos);
+    DCHECK_NOT_NULL(stream_data.buf);
     ssize_t nwrite = WriteVStream(&path, pos, &ndatalen, stream_data);
+    Debug(session_, "Application accepted %zu bytes", ndatalen);
 
     if (nwrite <= 0) {
       switch (nwrite) {
-        case 0:
+        case 0: {
+          Debug(session_, "Congestion limited. Sending nothing.");
           if (stream_data.id >= 0) ResumeStream(stream_data.id);
           return congestionLimited(std::move(packet));
+        }
         case NGTCP2_ERR_STREAM_DATA_BLOCKED: {
           session().StreamDataBlocked(stream_data.id);
           if (session().max_data_left() == 0) {
@@ -287,6 +310,8 @@ void Session::Application::SendPendingData() {
           continue;
         }
         case NGTCP2_ERR_STREAM_SHUT_WR: {
+          Debug(session_, "Stream %" PRIi64 " is closed for writing",
+                stream_data.id);
           // Indicates that the writable side of the stream has been closed
           // locally or the stream is being reset. In either case, we can't send
           // any stream data!
@@ -299,6 +324,7 @@ void Session::Application::SendPendingData() {
           continue;
         }
         case NGTCP2_ERR_WRITE_MORE: {
+          Debug(session_, "Application should write more to packet");
           CHECK_GT(ndatalen, 0);
           if (!StreamCommit(&stream_data, ndatalen)) return session_->Close();
           pos += ndatalen;
@@ -308,6 +334,8 @@ void Session::Application::SendPendingData() {
 
       packet->Done(UV_ECANCELED);
       session_->last_error_ = QuicError::ForNgtcp2Error(nwrite);
+      Debug(session_, "Application failed to write stream data with error %s",
+            session_->last_error_);
       return session_->Close(Session::CloseMethod::SILENT);
     }
 
@@ -322,9 +350,10 @@ void Session::Application::SendPendingData() {
 
     if (stream_data.id >= 0 && ndatalen < 0) ResumeStream(stream_data.id);
 
+    Debug(session_, "Packet ready to send with %zu bytes", nwrite);
     packet->Truncate(nwrite);
-    session_->Send(std::move(packet), path);
-
+    session_->Send(packet, path);
+    packet = nullptr;
     pos = nullptr;
 
     if (++packetSendCount == maxPacketCount) {
@@ -343,10 +372,12 @@ ssize_t Session::Application::WriteVStream(PathStorage* path,
   uint32_t flags = NGTCP2_WRITE_STREAM_FLAG_NONE;
   if (stream_data.remaining > 0) flags |= NGTCP2_WRITE_STREAM_FLAG_MORE;
   if (stream_data.fin) flags |= NGTCP2_WRITE_STREAM_FLAG_FIN;
+  ngtcp2_pkt_info pi;
+
   ssize_t ret = ngtcp2_conn_writev_stream(
       *session_,
       &path->path,
-      nullptr,
+      &pi,
       buf,
       ngtcp2_conn_get_max_tx_udp_payload_size(*session_),
       ndatalen,
diff --git a/src/quic/application.h b/src/quic/application.h
@@ -151,6 +151,8 @@ struct Session::Application::StreamData final {
   BaseObjectPtr<Stream> stream;
 
   inline operator nghttp3_vec() const { return {data[0].base, data[0].len}; }
+
+  std::string ToString() const;
 };
 
 }  // namespace quic
diff --git a/src/quic/cid.cc b/src/quic/cid.cc
@@ -114,15 +114,16 @@ class RandomCIDFactory : public CID::Factory {
     return CID(start, length_hint);
   }
 
-  void GenerateInto(ngtcp2_cid* cid,
-                    size_t length_hint = CID::kMaxLength) const override {
+  CID GenerateInto(ngtcp2_cid* cid,
+                   size_t length_hint = CID::kMaxLength) const override {
     DCHECK_GE(length_hint, CID::kMinLength);
     DCHECK_LE(length_hint, CID::kMaxLength);
     Mutex::ScopedLock lock(mutex_);
     maybe_refresh_pool(length_hint);
     auto start = pool_ + pos_;
     pos_ += length_hint;
     ngtcp2_cid_init(cid, start, length_hint);
+    return CID(cid);
   }
 
  private:
diff --git a/src/quic/cid.h b/src/quic/cid.h
@@ -111,10 +111,9 @@ class CID::Factory {
   virtual CID Generate(size_t length_hint = CID::kMaxLength) const = 0;
 
   // Generate a new CID into the given ngtcp2_cid. This variation of
-  // Generate should be used far less commonly. It is provided largely
-  // for a couple of internal cases.
-  virtual void GenerateInto(ngtcp2_cid* cid,
-                            size_t length_hint = CID::kMaxLength) const = 0;
+  // Generate should be used far less commonly.
+  virtual CID GenerateInto(ngtcp2_cid* cid,
+                           size_t length_hint = CID::kMaxLength) const = 0;
 
   // The default random CID generator instance.
   static const Factory& random();
diff --git a/src/quic/endpoint.cc b/src/quic/endpoint.cc
@@ -1236,7 +1236,7 @@ void Endpoint::Receive(const uv_buf_t& buf,
     // identifies us. config.dcid should equal scid. config.scid should *not*
     // equal dcid.
     DCHECK(config.dcid == scid);
-    DCHECK(config.scid != dcid);
+    DCHECK(config.scid == dcid);
 
     const auto is_remote_address_validated = ([&] {
       auto info = addrLRU_.Peek(remote_address);
diff --git a/src/quic/session.cc b/src/quic/session.cc
@@ -1406,7 +1406,7 @@ void Session::DatagramReceived(const uint8_t* data,
 bool Session::GenerateNewConnectionId(ngtcp2_cid* cid,
                                       size_t len,
                                       uint8_t* token) {
-  CID cid_ = config_.options.cid_factory->Generate(len);
+  CID cid_ = config_.options.cid_factory->GenerateInto(cid, len);
   Debug(this, "Generated new connection id %s", cid_);
   StatelessResetToken new_token(
       token, endpoint_->options().reset_token_secret, cid_);
@@ -1416,10 +1416,11 @@ bool Session::GenerateNewConnectionId(ngtcp2_cid* cid,
 }
 
 bool Session::HandshakeCompleted() {
+  Debug(this, "Session handshake completed");
+
   if (state_->handshake_completed) return false;
-  state_->handshake_completed = true;
+  state_->handshake_completed = 1;
 
-  Debug(this, "Session handshake completed");
   STAT_RECORD_TIMESTAMP(Stats, handshake_completed_at);
 
   if (!tls_session().early_data_was_accepted())
@@ -2012,17 +2013,17 @@ struct Session::Impl {
                                void* user_data) {
     auto session = Impl::From(conn, user_data);
     if (UNLIKELY(session->is_destroyed())) return NGTCP2_ERR_CALLBACK_FAILURE;
+    CHECK(!session->is_server());
+
+    if (level != NGTCP2_ENCRYPTION_LEVEL_1RTT) return NGTCP2_SUCCESS;
 
     Debug(session,
           "Receiving RX key for level %d for dcid %s",
           to_string(level),
           session->config().dcid);
 
-    if (!session->is_server() && (level == NGTCP2_ENCRYPTION_LEVEL_0RTT ||
-                                  level == NGTCP2_ENCRYPTION_LEVEL_1RTT)) {
-      if (!session->application().Start()) return NGTCP2_ERR_CALLBACK_FAILURE;
-    }
-    return NGTCP2_SUCCESS;
+    return session->application().Start() ?
+        NGTCP2_SUCCESS : NGTCP2_ERR_CALLBACK_FAILURE;
   }
 
   static int on_receive_stateless_reset(ngtcp2_conn* conn,
@@ -2071,17 +2072,16 @@ struct Session::Impl {
                                void* user_data) {
     auto session = Impl::From(conn, user_data);
     if (UNLIKELY(session->is_destroyed())) return NGTCP2_ERR_CALLBACK_FAILURE;
+    CHECK(session->is_server());
+
+    if (level != NGTCP2_ENCRYPTION_LEVEL_1RTT) return NGTCP2_SUCCESS;
 
     Debug(session,
           "Receiving TX key for level %d for dcid %s",
           to_string(level),
           session->config().dcid);
-
-    if (session->is_server() && (level == NGTCP2_ENCRYPTION_LEVEL_0RTT ||
-                                 level == NGTCP2_ENCRYPTION_LEVEL_1RTT)) {
-      if (!session->application().Start()) return NGTCP2_ERR_CALLBACK_FAILURE;
-    }
-    return NGTCP2_SUCCESS;
+    return session->application().Start() ?
+        NGTCP2_SUCCESS : NGTCP2_ERR_CALLBACK_FAILURE;
   }
 
   static int on_receive_version_negotiation(ngtcp2_conn* conn,
@@ -2206,7 +2206,7 @@ struct Session::Impl {
       on_stream_stop_sending,
       ngtcp2_crypto_version_negotiation_cb,
       on_receive_rx_key,
-      on_receive_tx_key,
+      nullptr,
       on_early_data_rejected};
 
   static constexpr ngtcp2_callbacks SERVER = {
@@ -2247,7 +2247,7 @@ struct Session::Impl {
       ngtcp2_crypto_get_path_challenge_data_cb,
       on_stream_stop_sending,
       ngtcp2_crypto_version_negotiation_cb,
-      on_receive_rx_key,
+      nullptr,
       on_receive_tx_key,
       on_early_data_rejected};
 };
