From e9368e32cad3af5846c0d36bb4792d15bffe438f Mon Sep 17 00:00:00 2001
From: Evgeny Malygin <emalygin@bloomberg.net>
Date: Fri, 24 Jul 2026 11:36:46 -0400
Subject: [PATCH] Fix[Options]: correct handling of compressed options

Signed-off-by: Evgeny Malygin <emalygin@bloomberg.net>
---
 .../bmq/impl/infr/proto/Options.java          | 25 ++++++------
 .../bmq/impl/infr/proto/OptionsTest.java      | 38 +++++++++++++++++++
 2 files changed, 52 insertions(+), 11 deletions(-)

diff --git a/bmq-sdk/src/main/java/com/bloomberg/bmq/impl/infr/proto/Options.java b/bmq-sdk/src/main/java/com/bloomberg/bmq/impl/infr/proto/Options.java
index 48e8f765..c12abe12 100644
--- a/bmq-sdk/src/main/java/com/bloomberg/bmq/impl/infr/proto/Options.java
+++ b/bmq-sdk/src/main/java/com/bloomberg/bmq/impl/infr/proto/Options.java
@@ -43,6 +43,12 @@ public void streamIn(int size, ByteBufferInputStream bbis) throws IOException {
         while (size > 0) {
             OptionHeader header = new OptionHeader();
             header.streamIn(bbis);
+
+            int totalSize =
+                    header.packed()
+                            ? OptionHeader.HEADER_SIZE
+                            : header.words() * Protocol.WORD_SIZE;
+
             switch (header.type()) {
                 case SUB_QUEUE_INFOS:
                     if (subQueueInfosOption == null) {
@@ -51,32 +57,29 @@ public void streamIn(int size, ByteBufferInputStream bbis) throws IOException {
                         logger.debug("New options: {}", subQueueInfosOption);
                     } else {
                         logger.warn("Multiple SubQueueInfos option: {}", header.type());
-
-                        int numSkip =
-                                header.packed()
-                                        ? 0
-                                        : header.words() * Protocol.WORD_SIZE
-                                                - OptionHeader.HEADER_SIZE;
-
-                        bbis.skip(numSkip);
+                        bbis.skip(totalSize - OptionHeader.HEADER_SIZE);
                     }
                     break;
                 case SUB_QUEUE_IDS_OLD:
+                    // Only SUB_QUEUE_INFOS options may be packed.
+                    assert !header.packed();
                     if (subQueueIdsOption == null) {
                         subQueueIdsOption = new SubQueueIdsOption(header);
                         subQueueIdsOption.streamIn(bbis);
                         logger.debug("Old options: {}", subQueueIdsOption);
                     } else {
                         logger.warn("Multiple SubQueueIds option: {}", header.type());
-                        bbis.skip(header.words() * Protocol.WORD_SIZE - OptionHeader.HEADER_SIZE);
+                        bbis.skip(totalSize - OptionHeader.HEADER_SIZE);
                     }
                     break;
                 default:
+                    // Only SUB_QUEUE_INFOS options may be packed.
+                    assert !header.packed();
                     logger.warn("Unsupported option type: {}", header.type());
-                    bbis.skip(header.words() * Protocol.WORD_SIZE - OptionHeader.HEADER_SIZE);
+                    bbis.skip(totalSize - OptionHeader.HEADER_SIZE);
                     break;
             }
-            size -= header.words() * Protocol.WORD_SIZE;
+            size -= totalSize;
         }
 
         if (subQueueIdsOption != null && subQueueInfosOption != null) {
diff --git a/bmq-sdk/src/test/java/com/bloomberg/bmq/impl/infr/proto/OptionsTest.java b/bmq-sdk/src/test/java/com/bloomberg/bmq/impl/infr/proto/OptionsTest.java
index c0cacefc..147daa1d 100644
--- a/bmq-sdk/src/test/java/com/bloomberg/bmq/impl/infr/proto/OptionsTest.java
+++ b/bmq-sdk/src/test/java/com/bloomberg/bmq/impl/infr/proto/OptionsTest.java
@@ -96,6 +96,44 @@ void testStreamInInfos() throws IOException {
         assertEquals(0, bbis.available());
     }
 
+    @Test
+    void testStreamInPackedInfosThenIds() throws IOException {
+        // A *packed* SUB_QUEUE_INFOS option occupies exactly one word (4 bytes): the
+        // header only, with no option content. Its 'words' field is reinterpreted as
+        // the RDA counter, so 'size' must be decremented by HEADER_SIZE, not by
+        // words()*WORD_SIZE. If a second option follows, an incorrect decrement causes
+        // the loop to exit early and the trailing option to be misread/dropped.
+        ByteBuffer bb =
+                ByteBuffer.wrap(
+                        new byte[] {
+                            // Packed Infos: type=SUB_QUEUE_INFOS(3), packed=1,
+                            // words field reused as RDA counter (5)
+                            0b00001110, 0b00000000, 0b00000000, 0b00000101, // header only
+
+                            // Ids
+                            0b00000100, 0b00000000, 0b00000000, 0b00000010, // header
+                            0b00000000, 0b00000000, 0b00000000, 0b00000111, // subqueue id 7
+                        });
+        ByteBufferInputStream bbis = new ByteBufferInputStream(bb);
+
+        Options options = new Options();
+        options.streamIn(12, bbis);
+
+        logger.info("Options: {}", options);
+
+        assertNotNull(options.subQueueInfosOption());
+        assertArrayEquals(
+                new Integer[] {com.bloomberg.bmq.impl.QueueId.k_DEFAULT_SUBQUEUE_ID},
+                options.subQueueInfosOption().subQueueIds());
+
+        // The option after the packed one must still be parsed.
+        assertNotNull(options.subQueueIdsOption());
+        assertArrayEquals(new Integer[] {7}, options.subQueueIdsOption().subQueueIds());
+
+        // Check that input stream is empty
+        assertEquals(0, bbis.available());
+    }
+
     @Test
     void testStreamInInfosIds() throws IOException {
         ByteBuffer bb =