Modify QuicPacketScheduler for sending reliable resets

Summary:
I'm modifying the `QuicPacketScheduler` so that we only send RESET_STREAM_AT frames when we've sent out all of the data upto the reliable offset. While this is not something that's mandated by the spec, it greatly simplifies the accounting of flow control if we do it this way.

In more detail, if we allowed the egressing of the RESET_STREAM_AT frame before all of the reliable data was sent out, we'd need coordination between the StreamFrameScheduler and the RstStreamScheduler, so as to ensure that we are correctly accounting for flow control. In addition to that, we'd likely have to have a "flow control accounted for" offset in each stream state, so as to ensure that we don't "double count" flow control between the two schedulers. This adds significant complexity, so we're just sending RESET_STREAM_AT frames when we've sent out all of the data upto the reliable offset.

Reviewed By: mjoras

Differential Revision: D66709346

fbshipit-source-id: 01ca8659ae80fcdbd0e599f480fde10a11f0a56b

quic/api/QuicPacketScheduler.cpp

@@ -646,11 +646,21 @@ bool RstStreamScheduler::hasPendingRsts() const {
 bool RstStreamScheduler::writeRsts(PacketBuilderInterface& builder) {
   bool rstWritten = false;
   for (const auto& resetStream : conn_.pendingEvents.resets) {
-    auto bytesWritten = writeFrame(resetStream.second, builder);
-    if (!bytesWritten) {
-      break;
+    auto streamId = resetStream.first;
+    QuicStreamState* streamState = conn_.streamManager->getStream(streamId);
+    if (streamState->pendingWrites.empty() &&
+        streamState->writeBufMeta.length == 0) {
+      //    We only write a RESET_STREAM or RESET_STREAM_AT frame for a stream
+      //    once we've written out all data that needs to be delivered reliably.
+      //    While this is not something that's mandated by the spec, we're doing
+      //    it in this implementation because it dramatically simplifies flow
+      //    control accounting.
+      auto bytesWritten = writeFrame(resetStream.second, builder);
+      if (!bytesWritten) {
+        break;
+      }
+      rstWritten = true;
     }
-    rstWritten = true;
   }
   return rstWritten;
 }

quic/api/test/QuicPacketSchedulerTest.cpp

@@ -2729,4 +2729,81 @@ TEST_F(QuicPacketSchedulerTest, ImmediateAckFrameSchedulerNotRequested) {
   EXPECT_LT(packetLength, conn.udpSendPacketLen);
 }
 
+TEST_F(QuicPacketSchedulerTest, RstStreamSchedulerReliableReset) {
+  QuicClientConnectionState conn(
+      FizzClientQuicHandshakeContext::Builder().build());
+  conn.streamManager->setMaxLocalBidirectionalStreams(10);
+  conn.flowControlState.peerAdvertisedMaxOffset = 100000;
+  conn.flowControlState.peerAdvertisedInitialMaxStreamOffsetBidiRemote = 100000;
+  auto stream = conn.streamManager->createNextBidirectionalStream().value();
+  auto buf = folly::IOBuf::copyBuffer("cupcake");
+  auto bufLen = buf->computeChainDataLength();
+  writeDataToQuicStream(*stream, buf->clone(), false);
+
+  // Reliable reset with reliableSize = bufLen
+  conn.pendingEvents.resets.emplace(
+      stream->id, RstStreamFrame(stream->id, 0, bufLen, bufLen));
+
+  FrameScheduler scheduler = std::move(FrameScheduler::Builder(
+                                           conn,
+                                           EncryptionLevel::AppData,
+                                           PacketNumberSpace::AppData,
+                                           "streamScheduler")
+                                           .streamFrames()
+                                           .resetFrames())
+                                 .build();
+  auto cipherOverhead = 16;
+  PacketNum packetNum1 = 0;
+  ShortHeader header1(
+      ProtectionType::KeyPhaseOne,
+      conn.clientConnectionId.value_or(getTestConnectionId()),
+      packetNum1);
+  RegularQuicPacketBuilder builder1(
+      conn.udpSendPacketLen,
+      std::move(header1),
+      conn.ackStates.appDataAckState.largestAckedByPeer.value_or(0));
+  auto packetResult1 = scheduler.scheduleFramesForPacket(
+      std::move(builder1), conn.udpSendPacketLen - cipherOverhead);
+  auto encodedSize1 = packetResult1.packet->body.computeChainDataLength() +
+      packetResult1.packet->header.computeChainDataLength() + cipherOverhead;
+  updateConnection(
+      conn,
+      none,
+      packetResult1.packet->packet,
+      Clock::now(),
+      encodedSize1,
+      0,
+      false /* isDSRPacket */);
+
+  // We shouldn't send the reliable reset just yet, because we haven't yet
+  // egressed all the stream data upto the reliable offset.
+  EXPECT_TRUE(conn.pendingEvents.resets.contains(stream->id));
+
+  PacketNum packetNum2 = 1;
+  ShortHeader header2(
+      ProtectionType::KeyPhaseOne,
+      conn.clientConnectionId.value_or(getTestConnectionId()),
+      packetNum2);
+  RegularQuicPacketBuilder builder2(
+      conn.udpSendPacketLen,
+      std::move(header2),
+      conn.ackStates.appDataAckState.largestAckedByPeer.value_or(0));
+  auto packetResult2 = scheduler.scheduleFramesForPacket(
+      std::move(builder2), conn.udpSendPacketLen - cipherOverhead);
+  auto encodedSize2 = packetResult1.packet->body.computeChainDataLength() +
+      packetResult2.packet->header.computeChainDataLength() + cipherOverhead;
+  updateConnection(
+      conn,
+      none,
+      packetResult2.packet->packet,
+      Clock::now(),
+      encodedSize2,
+      0,
+      false /* isDSRPacket */);
+
+  // Now we should have egressed all the stream data upto the reliable offset,
+  // so we should have sent the reliable reset.
+  EXPECT_FALSE(conn.pendingEvents.resets.contains(stream->id));
+}
+
 } // namespace quic::test