CASSANDRA-21575: Keep message ids unsigned so they do not inflate every message header

CHANGES.txt

@@ -1,4 +1,5 @@
 7.0
+ * Keep message ids unsigned so they do not inflate every message header once the id counter wraps (CASSANDRA-21575)
  * Add prepared statement cache stats to nodetool info (CASSANDRA-14366)
  * Don't increment client metrics on messaging service connection unpause (CASSANDRA-21491)
  * Add nodetool getreplicas (CASSANDRA-17665)

src/java/org/apache/cassandra/net/ForwardingInfo.java

@@ -108,7 +108,7 @@ public ForwardingInfo deserialize(DataInputPlus in, int version) throws IOExcept
             for (int i = 0; i < count; i++)
             {
                 targets.add(inetAddressAndPortSerializer.deserialize(in, version));
-                ids[i] = in.readUnsignedVInt32();
+                ids[i] = in.readUnsignedVInt();
             }
 
             return new ForwardingInfo(targets, ids);

src/java/org/apache/cassandra/net/Message.java

@@ -483,13 +483,27 @@ private static long nextId()
         long id;
         do
         {
-            id = nextId.incrementAndGet();
+            id = toUnsignedId(nextId.incrementAndGet());
         }
         while (id == NO_ID);
 
         return id;
     }
 
+    /**
+     * Widens the id counter as unsigned.
+     * <p>
+     * The counter is an {@code int} and wraps to {@link Integer#MIN_VALUE} once it passes
+     * {@link Integer#MAX_VALUE}. Widening it directly would sign-extend, and {@link Serializer} writes the id as an
+     * unsigned vint, so every negative id occupies the maximum width of 9 bytes instead of at most 5. Masking keeps
+     * ids in {@code [0, 2^32)}, which covers the same number of distinct values while staying cheap to encode.
+     */
+    @VisibleForTesting
+    static long toUnsignedId(int counter)
+    {
+        return counter & 0xFFFFFFFFL;
+    }
+
     /**
      * WARNING: this is inaccurate for messages from pre40 nodes, which can use 0 as an id (but will do so rarely)
      */

test/unit/org/apache/cassandra/net/ForwardingInfoTest.java

@@ -44,6 +44,42 @@ public void testSupportedVersions() throws Exception
             testVersion(version.value);
     }
 
+    /**
+     * Message ids are drawn from an unsigned 32-bit counter, so they span the whole [1, 2^32) range. They are written
+     * as 64-bit unsigned vints, and the reader has to use the matching width: reading them back as a 32-bit vint
+     * throws {@link org.apache.cassandra.utils.vint.VIntOutOfRangeException} for any id above Integer.MAX_VALUE.
+     */
+    @Test
+    public void testLargeMessageIdsRoundTrip() throws Exception
+    {
+        InetAddressAndPort.initializeDefaultPort(65532);
+        List<InetAddressAndPort> addresses = ImmutableList.of(InetAddressAndPort.getByName("127.0.0.1:7000"),
+                                                              InetAddressAndPort.getByName("127.0.0.2:7000"),
+                                                              InetAddressAndPort.getByName("127.0.0.3:7000"),
+                                                              InetAddressAndPort.getByName("127.0.0.4:7000"));
+
+        long[] ids = { 1L, Integer.MAX_VALUE, Integer.MAX_VALUE + 1L, 0xFFFFFFFFL };
+        ForwardingInfo forwardingInfo = new ForwardingInfo(addresses, ids);
+
+        for (MessagingService.Version version : MessagingService.Version.supportedVersions())
+        {
+            ByteBuffer buffer;
+            try (DataOutputBuffer dob = new DataOutputBuffer())
+            {
+                ForwardingInfo.serializer.serialize(forwardingInfo, dob, version.value);
+                buffer = dob.buffer();
+            }
+
+            assertEquals(buffer.remaining(), ForwardingInfo.serializer.serializedSize(forwardingInfo, version.value));
+
+            try (DataInputBuffer dib = new DataInputBuffer(buffer, false))
+            {
+                ForwardingInfo deserialized = ForwardingInfo.serializer.deserialize(dib, version.value);
+                assertTrue(Arrays.equals(ids, deserialized.messageIds));
+            }
+        }
+    }
+
     private void testVersion(int version) throws Exception
     {
         InetAddressAndPort.initializeDefaultPort(65532);

test/unit/org/apache/cassandra/net/MessageTest.java

@@ -46,6 +46,7 @@
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.FreeRunningClock;
 import org.apache.cassandra.utils.TimeUUID;
+import org.apache.cassandra.utils.vint.VIntCoding;
 
 import static com.google.common.base.Throwables.getStackTraceAsString;
 import static org.apache.cassandra.exceptions.RemoteExceptionTest.normalizeThrowable;
@@ -368,4 +369,55 @@ public void testCreationTime()
         long localTimeNanos = localClock.now();
         assertTrue( Message.Serializer.calculateCreationTimeNanos(remoteCreatedAt, localClock.translate(), localTimeNanos) > 0);
     }
+
+    /**
+     * The id counter is an int, so it wraps to Integer.MIN_VALUE after Integer.MAX_VALUE. Ids must stay within
+     * [0, 2^32) across that wrap: the serializer writes them as unsigned vints, so a sign-extended negative id would
+     * occupy the maximum width of 9 bytes rather than at most 5.
+     */
+    @Test
+    public void testIdsRemainUnsignedAcrossCounterWrap()
+    {
+        for (int counter : new int[]{ 1, Integer.MAX_VALUE - 1, Integer.MAX_VALUE, Integer.MIN_VALUE, Integer.MIN_VALUE + 1, -1 })
+        {
+            long id = Message.toUnsignedId(counter);
+            assertTrue("id must not be negative, got " + id + " for counter " + counter, id >= 0);
+            assertTrue("id must fit in 32 unsigned bits, got " + id, id <= 0xFFFFFFFFL);
+            assertTrue("id must encode within 5 bytes, got " + VIntCoding.computeUnsignedVIntSize(id),
+                       VIntCoding.computeUnsignedVIntSize(id) <= 5);
+        }
+
+        // the mapping stays injective, so ids remain as distinct as the counter itself
+        assertEquals(Integer.MAX_VALUE + 1L, Message.toUnsignedId(Integer.MIN_VALUE));
+        assertEquals(0xFFFFFFFFL, Message.toUnsignedId(-1));
+    }
+
+    /**
+     * Ids now span the whole unsigned 32-bit range, so the message header must round-trip the upper half too.
+     */
+    @Test
+    public void testLargeIdRoundTrips() throws IOException
+    {
+        for (Version version : Version.supportedVersions())
+        {
+            for (long id : new long[]{ 1L, Integer.MAX_VALUE, Integer.MAX_VALUE + 1L, 0xFFFFFFFFL })
+            {
+                Message<NoPayload> msg = Message.builder(Verb._TEST_1, noPayload)
+                                                .withId(id)
+                                                .from(FBUtilities.getBroadcastAddressAndPort())
+                                                .build();
+
+                try (DataOutputBuffer out = new DataOutputBuffer())
+                {
+                    serializer.serialize(msg, out, version.value);
+                    assertEquals(msg.serializedSize(version.value), out.getLength());
+
+                    try (DataInputBuffer in = new DataInputBuffer(out.buffer(), false))
+                    {
+                        assertEquals(id, serializer.deserialize(in, msg.from(), version.value).id());
+                    }
+                }
+            }
+        }
+    }
 }