Disallow non-consecutive duplicate byte event registrations

Summary:
As per design, we are not supposed to honor duplicate byte event registrations. Duplicate is defined by a registration for the same <stream ID, stream offset, callback*> tuple twice.
Currently, there is a bug where-in we can register the same byte event a second time *if* there is another byte event for a higher offset that is registered in-between, for the same stream.
Example Sequence that triggers the bug:
1. Registration for Stream ID =10, offset = 50, callback = A. This will be successful.
2. Registration for Stream ID = 10, offset = 55, callback = <doesn't matter>. This will be successful.
3. Registration for Stream ID = 10, offset = 50, callback = A. This is a duplicate of #1. This *should not* be honored, but it was being honored (this was the bug).

This commit fixes the bug, #3 will return INVALID_OPERATION.

Reviewed By: bschlinker, lnicco

Differential Revision: D27946237

fbshipit-source-id: f9751ac90159284f86a285e5e0ebb23b83041d2a

quic/api/QuicTransportBase.cpp

@@ -2110,8 +2110,13 @@ QuicTransportBase::registerByteEventCallback(
         offset,
         [&](uint64_t o, const ByteEventDetail& p) { return o < p.offset; });
     if (pos != byteEventMapIt->second.begin()) {
-      auto prev = std::prev(pos);
-      if ((prev->offset == offset) && (prev->callback == cb)) {
+      auto matchingEvent = std::find_if(
+          byteEventMapIt->second.begin(),
+          pos,
+          [offset, cb](const ByteEventDetail& p) {
+            return ((p.offset == offset) && (p.callback == cb));
+          });
+      if (matchingEvent != pos) {
         // ByteEvent has been already registered for the same type, id,
         // offset and for the same recipient, return an INVALID_OPERATION error
         // to prevent duplicate registrations.

quic/api/test/QuicTransportBaseTest.cpp

@@ -1672,6 +1672,78 @@ TEST_F(
   Mock::VerifyAndClearExpectations(&dcb);
 }
 
+TEST_F(QuicTransportImplTest, RegisterDeliveryCallbackMultipleRegistrationsTx) {
+  auto stream = transport->createBidirectionalStream().value();
+  StrictMock<MockByteEventCallback> txcb1;
+  StrictMock<MockByteEventCallback> txcb2;
+
+  // Set the current write offset to 7.
+  auto streamState = transport->transportConn->streamManager->getStream(stream);
+  streamState->currentWriteOffset = 7;
+  streamState->ackedIntervals.insert(0, 6);
+
+  // Have 2 different recipients register for a callback on the same stream ID
+  // and offset that is before the curernt write offset, they will both be
+  // scheduled for immediate delivery.
+  EXPECT_CALL(txcb1, onByteEventRegistered(getTxMatcher(stream, 3)));
+  EXPECT_CALL(txcb2, onByteEventRegistered(getTxMatcher(stream, 3)));
+  transport->registerTxCallback(stream, 3, &txcb1);
+  transport->registerTxCallback(stream, 3, &txcb2);
+  Mock::VerifyAndClearExpectations(&txcb1);
+  Mock::VerifyAndClearExpectations(&txcb2);
+
+  // Now, re-register the same callbacks, it should not go through.
+  auto ret = transport->registerTxCallback(stream, 3, &txcb1);
+  EXPECT_EQ(LocalErrorCode::INVALID_OPERATION, ret.error());
+  ret = transport->registerTxCallback(stream, 3, &txcb2);
+  EXPECT_EQ(LocalErrorCode::INVALID_OPERATION, ret.error());
+
+  // Deliver the first set of registrations.
+  EXPECT_CALL(txcb1, onByteEvent(getTxMatcher(stream, 3))).Times(1);
+  EXPECT_CALL(txcb2, onByteEvent(getTxMatcher(stream, 3))).Times(1);
+  evb->loopOnce();
+  Mock::VerifyAndClearExpectations(&txcb1);
+  Mock::VerifyAndClearExpectations(&txcb2);
+}
+
+TEST_F(
+    QuicTransportImplTest,
+    RegisterDeliveryCallbackMultipleRegistrationsAck) {
+  auto stream = transport->createBidirectionalStream().value();
+  StrictMock<MockByteEventCallback> txcb1;
+  StrictMock<MockByteEventCallback> txcb2;
+
+  // Set the current write offset to 7.
+  auto streamState = transport->transportConn->streamManager->getStream(stream);
+  streamState->currentWriteOffset = 7;
+  streamState->ackedIntervals.insert(0, 6);
+
+  // Have 2 different recipients register for a callback on the same stream ID
+  // and offset that is before the curernt write offset, they will both be
+  // scheduled for immediate delivery.
+  EXPECT_CALL(txcb1, onByteEventRegistered(getAckMatcher(stream, 3)));
+  EXPECT_CALL(txcb2, onByteEventRegistered(getAckMatcher(stream, 3)));
+  transport->registerByteEventCallback(ByteEvent::Type::ACK, stream, 3, &txcb1);
+  transport->registerByteEventCallback(ByteEvent::Type::ACK, stream, 3, &txcb2);
+  Mock::VerifyAndClearExpectations(&txcb1);
+  Mock::VerifyAndClearExpectations(&txcb2);
+
+  // Now, re-register the same callbacks, it should not go through.
+  auto ret = transport->registerByteEventCallback(
+      ByteEvent::Type::ACK, stream, 3, &txcb1);
+  EXPECT_EQ(LocalErrorCode::INVALID_OPERATION, ret.error());
+  ret = transport->registerByteEventCallback(
+      ByteEvent::Type::ACK, stream, 3, &txcb2);
+  EXPECT_EQ(LocalErrorCode::INVALID_OPERATION, ret.error());
+
+  // Deliver the first set of registrations.
+  EXPECT_CALL(txcb1, onByteEvent(getAckMatcher(stream, 3))).Times(1);
+  EXPECT_CALL(txcb2, onByteEvent(getAckMatcher(stream, 3))).Times(1);
+  evb->loopOnce();
+  Mock::VerifyAndClearExpectations(&txcb1);
+  Mock::VerifyAndClearExpectations(&txcb2);
+}
+
 TEST_F(QuicTransportImplTest, RegisterDeliveryCallbackMultipleRecipientsTx) {
   auto stream = transport->createBidirectionalStream().value();
   StrictMock<MockByteEventCallback> txcb1;