Issue #7635 - add extra testing for QPACK blocked streams

Signed-off-by: Lachlan Roberts <lachlan@webtide.com>

jetty-http3/http3-qpack/src/main/java/org/eclipse/jetty/http3/qpack/QpackEncoder.java

@@ -106,6 +106,11 @@ public QpackEncoder(Instruction.Handler handler, int maxBlockedStreams)
         _parser = new EncoderInstructionParser(_instructionHandler);
     }
 
+    Map<Long, StreamInfo> getStreamInfoMap()
+    {
+        return _streamInfoMap;
+    }
+
     public int getMaxBlockedStreams()
     {
         return _maxBlockedStreams;
@@ -193,7 +198,15 @@ public void encode(ByteBuffer buffer, long streamId, MetaData metadata) throws Q
                         requiredInsertCount = entryRequiredInsertCount;
                 }
 
+                // We should not expect section acknowledgements for 0 required insert count.
                 sectionInfo.setRequiredInsertCount(requiredInsertCount);
+                if (requiredInsertCount == 0)
+                {
+                    streamInfo.remove(sectionInfo);
+                    if (streamInfo.isEmpty())
+                        _streamInfoMap.remove(streamId);
+                }
+
                 int base = dynamicTable.getBase();
                 int encodedInsertCount = encodeInsertCount(requiredInsertCount, dynamicTable.getCapacity());
                 boolean signBit = base < requiredInsertCount;
@@ -472,9 +485,13 @@ public void onSectionAcknowledgement(long streamId) throws QpackException
 
             // The KnownInsertCount should be updated to the earliest sent RequiredInsertCount on that stream.
             StreamInfo.SectionInfo sectionInfo = streamInfo.acknowledge();
+            boolean wasBlocked = sectionInfo.isBlocking();
             sectionInfo.release();
             _knownInsertCount = Math.max(_knownInsertCount, sectionInfo.getRequiredInsertCount());
 
+            if (wasBlocked && !streamInfo.isBlocked())
+                _blockedStreams--;
+
             // If we have no more outstanding section acknowledgments remove the StreamInfo.
             if (streamInfo.isEmpty())
                 _streamInfoMap.remove(streamId);

jetty-http3/http3-qpack/src/main/java/org/eclipse/jetty/http3/qpack/internal/StreamInfo.java

@@ -106,6 +106,7 @@ public void release()
             {
                 entry.release();
             }
+            _block = false;
             _entries.clear();
         }
 

jetty-http3/http3-qpack/src/main/java/org/eclipse/jetty/http3/qpack/internal/instruction/InsertCountIncrementInstruction.java

@@ -29,6 +29,11 @@ public InsertCountIncrementInstruction(int increment)
         _increment = increment;
     }
 
+    public int getIncrement()
+    {
+        return _increment;
+    }
+
     @Override
     public void encode(ByteBufferPool.Lease lease)
     {

jetty-http3/http3-qpack/src/main/java/org/eclipse/jetty/http3/qpack/internal/instruction/SectionAcknowledgmentInstruction.java

@@ -29,6 +29,11 @@ public SectionAcknowledgmentInstruction(long streamId)
         _streamId = streamId;
     }
 
+    public long getStreamId()
+    {
+        return _streamId;
+    }
+
     @Override
     public void encode(ByteBufferPool.Lease lease)
     {

jetty-http3/http3-qpack/src/test/java/org/eclipse/jetty/http3/qpack/BlockedStreamsTest.java

@@ -0,0 +1,213 @@
+package org.eclipse.jetty.http3.qpack;
+
+import java.nio.ByteBuffer;
+
+import org.eclipse.jetty.http.HttpField;
+import org.eclipse.jetty.http.MetaData;
+import org.eclipse.jetty.http3.qpack.QpackException.SessionException;
+import org.eclipse.jetty.http3.qpack.internal.instruction.IndexedNameEntryInstruction;
+import org.eclipse.jetty.http3.qpack.internal.instruction.InsertCountIncrementInstruction;
+import org.eclipse.jetty.http3.qpack.internal.instruction.LiteralNameEntryInstruction;
+import org.eclipse.jetty.http3.qpack.internal.instruction.SectionAcknowledgmentInstruction;
+import org.eclipse.jetty.http3.qpack.internal.instruction.SetCapacityInstruction;
+import org.eclipse.jetty.util.BufferUtil;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.eclipse.jetty.http3.qpack.QpackTestUtil.encode;
+import static org.eclipse.jetty.http3.qpack.QpackTestUtil.toBuffer;
+import static org.eclipse.jetty.http3.qpack.QpackTestUtil.toMetaData;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class BlockedStreamsTest
+{
+    private static final int MAX_BLOCKED_STREAMS = 5;
+    private static final int MAX_HEADER_SIZE = 1024;
+
+    private QpackEncoder _encoder;
+    private QpackDecoder _decoder;
+    private TestDecoderHandler _decoderHandler;
+    private TestEncoderHandler _encoderHandler;
+
+    @BeforeEach
+    public void before()
+    {
+        _encoderHandler = new TestEncoderHandler();
+        _decoderHandler = new TestDecoderHandler();
+        _encoder = new QpackEncoder(_encoderHandler, MAX_BLOCKED_STREAMS);
+        _decoder = new QpackDecoder(_decoderHandler, MAX_HEADER_SIZE);
+    }
+
+    @Test
+    public void testBlockedStreams() throws Exception
+    {
+        // These settings are determined by HTTP/3 settings frames.
+        _encoder.setMaxBlockedStreams(2);
+        _decoder.setMaxBlockedStreams(2);
+
+        // Set capacity of the encoder & decoder to allow entries to be added to the table.
+        int capacity = 1024;
+        _encoder.setCapacity(capacity);
+        Instruction instruction = _encoderHandler.getInstruction();
+        assertThat(instruction, instanceOf(SetCapacityInstruction.class));
+        _decoder.parseInstructions(QpackTestUtil.toBuffer(instruction));
+
+        // Encode a new field, which will be added to table. But do not forward insertion instruction to decoder,
+        // this will cause decoder to become "blocked" on stream 0 until receives the instruction.
+        HttpField entry1 = new HttpField("name1", "value1");
+        ByteBuffer buffer = encode(_encoder, 0, toMetaData("GET", "/", "http", entry1));
+        assertThat(BufferUtil.remaining(buffer), greaterThan(0L));
+        Instruction instruction1 = _encoderHandler.getInstruction();
+        assertThat(instruction1, instanceOf(LiteralNameEntryInstruction.class));
+        assertNull(_encoderHandler.getInstruction());
+
+        // Decoder will not be able to decode this header until it receives instruction.
+        boolean decoded = _decoder.decode(0, buffer, _decoderHandler);
+        assertFalse(decoded);
+        assertThat(BufferUtil.remaining(buffer), equalTo(0L));
+        assertNull(_decoderHandler.getMetaData());
+        assertNull(_decoderHandler.getInstruction());
+
+        // Encode second field with dynamic table, do not forward instruction to decoder.
+        HttpField entry2 = new HttpField("name1", "value2");
+        buffer = encode(_encoder, 1, toMetaData("GET", "/", "http", entry2));
+        assertThat(BufferUtil.remaining(buffer), greaterThan(0L));
+        Instruction instruction2 = _encoderHandler.getInstruction();
+        assertThat(instruction2, instanceOf(IndexedNameEntryInstruction.class));
+        assertNull(_encoderHandler.getInstruction());
+
+        // Decoder will not be able to decode this header until it receives instruction.
+        decoded = _decoder.decode(1, buffer, _decoderHandler);
+        assertFalse(decoded);
+        assertNull(_decoderHandler.getMetaData());
+        assertNull(_decoderHandler.getInstruction());
+
+        // Give first instruction to get first metadata.
+        _decoder.parseInstructions(QpackTestUtil.toBuffer(instruction1));
+        MetaData metaData = _decoderHandler.getMetaData();
+        assertThat(metaData.getFields().size(), equalTo(1));
+        assertThat(metaData.getFields().get(entry1.getHeader()), equalTo(entry1.getValue()));
+
+        Instruction inc1 = _decoderHandler.getInstruction();
+        assertThat(inc1, instanceOf(InsertCountIncrementInstruction.class));
+        assertThat(((InsertCountIncrementInstruction)inc1).getIncrement(), equalTo(1));
+
+        Instruction ack1 = _decoderHandler.getInstruction();
+        assertThat(ack1, instanceOf(SectionAcknowledgmentInstruction.class));
+        assertThat(((SectionAcknowledgmentInstruction)ack1).getStreamId(), equalTo(0L));
+
+        assertNull(_decoderHandler.getMetaData());
+        assertNull(_decoderHandler.getInstruction());
+
+        // Give second instruction to get second metadata.
+        _decoder.parseInstructions(QpackTestUtil.toBuffer(instruction2));
+        metaData = _decoderHandler.getMetaData();
+        assertThat(metaData.getFields().size(), equalTo(1));
+        assertThat(metaData.getFields().get(entry2.getHeader()), equalTo(entry2.getValue()));
+
+        Instruction inc2 = _decoderHandler.getInstruction();
+        assertThat(inc2, instanceOf(InsertCountIncrementInstruction.class));
+        assertThat(((InsertCountIncrementInstruction)inc2).getIncrement(), equalTo(1));
+
+        Instruction ack2 = _decoderHandler.getInstruction();
+        assertThat(ack2, instanceOf(SectionAcknowledgmentInstruction.class));
+        assertThat(((SectionAcknowledgmentInstruction)ack2).getStreamId(), equalTo(1L));
+
+        assertNull(_decoderHandler.getMetaData());
+        assertNull(_decoderHandler.getInstruction());
+
+        // The encoder hasn't received any InsertCountIncrementInstruction and so it thinks there are two streams blocked.
+        // It should only encode literal entries to not risk blocking another stream on the decoder.
+        HttpField entry3 = new HttpField("name3", "value3");
+        buffer = encode(_encoder, 3, toMetaData("GET", "/", "http", entry3));
+        assertThat(BufferUtil.remaining(buffer), greaterThan(0L));
+        instruction = _encoderHandler.getInstruction();
+        assertThat(instruction, instanceOf(LiteralNameEntryInstruction.class));
+        assertNull(_encoderHandler.getInstruction());
+
+        // Can decode literal entry immediately without any further instructions.
+        decoded = _decoder.decode(3, buffer, _decoderHandler);
+        assertTrue(decoded);
+        metaData = _decoderHandler.getMetaData();
+        assertThat(metaData.getFields().size(), equalTo(1));
+        assertThat(metaData.getFields().get(entry3.getHeader()), equalTo(entry3.getValue()));
+
+        // No longer referencing any streams that have been acknowledged.
+        buffer = toBuffer(inc1, ack1, inc2, ack2);
+        _encoder.parseInstructions(buffer);
+        assertThat(BufferUtil.remaining(buffer), equalTo(0L));
+        assertThat(_encoder.getStreamInfoMap().size(), equalTo(0));
+
+        // Encoder can now reference entries not acknowledged by the decoder again.
+        HttpField entry4 = new HttpField("name4", "value4");
+        buffer = encode(_encoder, 4, toMetaData("GET", "/", "http", entry4));
+        assertThat(BufferUtil.remaining(buffer), greaterThan(0L));
+        instruction = _encoderHandler.getInstruction();
+        assertThat(instruction, instanceOf(LiteralNameEntryInstruction.class));
+        assertNull(_encoderHandler.getInstruction());
+        decoded = _decoder.decode(4, buffer, _decoderHandler);
+        assertFalse(decoded);
+    }
+
+    @Test
+    public void testMaxBlockedStreams() throws Exception
+    {
+        // Encoder will risk blocking 1 more stream than the decoder will allow.
+        _encoder.setMaxBlockedStreams(3);
+        _decoder.setMaxBlockedStreams(2);
+
+        // Set capacity of the encoder & decoder to allow entries to be added to the table.
+        int capacity = 1024;
+        _encoder.setCapacity(capacity);
+        Instruction instruction = _encoderHandler.getInstruction();
+        assertThat(instruction, instanceOf(SetCapacityInstruction.class));
+        _decoder.parseInstructions(QpackTestUtil.toBuffer(instruction));
+
+        // Encode a new field, which will be added to table. But do not forward insertion instruction to decoder,
+        // this will cause decoder to become "blocked" on stream 0 until receives the instruction.
+        HttpField entry1 = new HttpField("name1", "value1");
+        ByteBuffer buffer = encode(_encoder, 0, toMetaData("GET", "/", "http", entry1));
+        assertThat(BufferUtil.remaining(buffer), greaterThan(0L));
+        Instruction instruction1 = _encoderHandler.getInstruction();
+        assertThat(instruction1, instanceOf(LiteralNameEntryInstruction.class));
+        assertNull(_encoderHandler.getInstruction());
+
+        // Decoder will not be able to decode this header until it receives instruction.
+        boolean decoded = _decoder.decode(0, buffer, _decoderHandler);
+        assertFalse(decoded);
+        assertThat(BufferUtil.remaining(buffer), equalTo(0L));
+        assertNull(_decoderHandler.getMetaData());
+        assertNull(_decoderHandler.getInstruction());
+
+        // Encode second field with dynamic table, do not forward instruction to decoder.
+        HttpField entry2 = new HttpField("name1", "value2");
+        buffer = encode(_encoder, 1, toMetaData("GET", "/", "http", entry2));
+        assertThat(BufferUtil.remaining(buffer), greaterThan(0L));
+        Instruction instruction2 = _encoderHandler.getInstruction();
+        assertThat(instruction2, instanceOf(IndexedNameEntryInstruction.class));
+        assertNull(_encoderHandler.getInstruction());
+
+        // Decoder will not be able to decode this header until it receives instruction.
+        decoded = _decoder.decode(1, buffer, _decoderHandler);
+        assertFalse(decoded);
+        assertNull(_decoderHandler.getMetaData());
+        assertNull(_decoderHandler.getInstruction());
+
+        // This entry will block a 3rd stream which the decoder must not allow.
+        HttpField entry3 = new HttpField("name3", "value3");
+        ByteBuffer encodedMetadata = encode(_encoder, 3, toMetaData("GET", "/", "http", entry3));
+        assertThat(BufferUtil.remaining(encodedMetadata), greaterThan(0L));
+        instruction = _encoderHandler.getInstruction();
+        assertThat(instruction, instanceOf(LiteralNameEntryInstruction.class));
+        assertNull(_encoderHandler.getInstruction());
+
+        assertThrows(SessionException.class, () -> _decoder.decode(3, encodedMetadata, _decoderHandler));
+    }
+}

jetty-http3/http3-qpack/src/test/java/org/eclipse/jetty/http3/qpack/QpackTestUtil.java

@@ -16,6 +16,10 @@
 import java.nio.ByteBuffer;
 import java.util.List;
 
+import org.eclipse.jetty.http.HttpField;
+import org.eclipse.jetty.http.HttpFields;
+import org.eclipse.jetty.http.HttpVersion;
+import org.eclipse.jetty.http.MetaData;
 import org.eclipse.jetty.io.ByteBufferPool;
 import org.eclipse.jetty.io.NullByteBufferPool;
 import org.eclipse.jetty.util.BufferUtil;
@@ -27,6 +31,23 @@
 
 public class QpackTestUtil
 {
+    public static ByteBuffer toBuffer(Instruction... instructions)
+    {
+        ByteBufferPool.Lease lease = new ByteBufferPool.Lease(new NullByteBufferPool());
+        for (Instruction instruction : instructions)
+        {
+            instruction.encode(lease);
+        }
+        ByteBuffer combinedBuffer = BufferUtil.allocate(Math.toIntExact(lease.getTotalLength()));
+        BufferUtil.clearToFill(combinedBuffer);
+        for (ByteBuffer buffer : lease.getByteBuffers())
+        {
+            combinedBuffer.put(buffer);
+        }
+        BufferUtil.flipToFlush(combinedBuffer, 0);
+        return combinedBuffer;
+    }
+
     public static Matcher<String> equalsHex(String expectedString)
     {
         expectedString = expectedString.replaceAll("\\s+", "");
@@ -56,4 +77,54 @@ public static String toHexString(Instruction instruction)
     {
         return BufferUtil.toHexString(toBuffer(List.of(instruction)));
     }
+
+    public static ByteBuffer encode(QpackEncoder encoder, long streamId, MetaData metaData) throws QpackException
+    {
+        ByteBuffer buffer = BufferUtil.allocate(1024);
+        BufferUtil.clearToFill(buffer);
+        encoder.encode(buffer, streamId, metaData);
+        BufferUtil.flipToFlush(buffer, 0);
+        return buffer;
+    }
+
+    public static HttpFields.Mutable toHttpFields(HttpField field)
+    {
+        return HttpFields.build().add(field);
+    }
+
+    public static MetaData toMetaData(String name, String value)
+    {
+        return toMetaData(toHttpFields(new HttpField(name, value)));
+    }
+
+    public static MetaData toMetaData(String method, String path, String scheme)
+    {
+        return toMetaData(method, path, scheme, (HttpField)null);
+    }
+
+    public static MetaData toMetaData(String method, String path, String scheme, HttpField ...fields)
+    {
+        HttpFields.Mutable httpFields = HttpFields.build();
+        for (HttpField field : fields)
+        {
+            httpFields.add(field);
+        }
+
+        return toMetaData(method, path, scheme, httpFields);
+    }
+
+    public static MetaData toMetaData(String method, String path, String scheme, HttpFields.Mutable fields)
+    {
+        fields = HttpFields.build()
+            .put(":scheme", scheme)
+            .put(":method", method)
+            .put(":path", path)
+            .add(fields);
+        return new MetaData(HttpVersion.HTTP_3, fields);
+    }
+
+    public static MetaData toMetaData(HttpFields fields)
+    {
+        return new MetaData(HttpVersion.HTTP_3, fields);
+    }
 }