From 2b9369dd6f0b3609a6b66c8844fc8894c15a510c Mon Sep 17 00:00:00 2001 From: Mirro Mutth Date: Wed, 13 Dec 2023 19:09:56 +0900 Subject: [PATCH] Correct unsigned long handling (#161) Motivation: Length-encoded integers should be handled correctly, especially those larger than `Long.MAX_VALUE`. See also #39. Modification: - Add code comments to explain how to handle the length encoded integer. - Correct `LargeFieldReader.readSlice()` method, it should use the length as an unsigned long. Result: - `LargeFieldReader.readSlice()` should now work correctly for fields those size is greater than 263-1 --- .../r2dbc/mysql/internal/util/VarIntUtils.java | 15 +++++++++++++++ .../message/server/ColumnCountMessage.java | 1 + .../server/DefinitionMetadataMessage.java | 5 +++-- .../mysql/message/server/LargeFieldReader.java | 18 +++++++++++------- .../message/server/NormalFieldReader.java | 1 + .../r2dbc/mysql/message/server/OkMessage.java | 3 ++- 6 files changed, 33 insertions(+), 10 deletions(-) diff --git a/src/main/java/io/asyncer/r2dbc/mysql/internal/util/VarIntUtils.java b/src/main/java/io/asyncer/r2dbc/mysql/internal/util/VarIntUtils.java index c5a4d530a..dba99fd72 100644 --- a/src/main/java/io/asyncer/r2dbc/mysql/internal/util/VarIntUtils.java +++ b/src/main/java/io/asyncer/r2dbc/mysql/internal/util/VarIntUtils.java @@ -47,6 +47,17 @@ public final class VarIntUtils { private static final int MEDIUM_SIZE = MEDIUM_BYTES * Byte.SIZE; + /** + * Reads a length encoded integer from the given buffers. Notice that a length encoded integer can be + * greater than {@link Long#MAX_VALUE}. In this case it should be used as an unsigned long. If we need + * assume the result as a smaller integer, add code comment to explain it. + *

+ * Note: it will change {@code firstPart} and {@code secondPart} readerIndex if necessary. + * + * @param firstPart the first part of a readable buffer include a part of the var integer. + * @param secondPart the second part of a readable buffer include subsequent part of the var integer. + * @return A var integer read from buffer. + */ public static long crossReadVarInt(ByteBuf firstPart, ByteBuf secondPart) { requireNonNull(firstPart, "firstPart must not be null"); requireNonNull(secondPart, "secondPart must not be null"); @@ -87,6 +98,10 @@ public static long crossReadVarInt(ByteBuf firstPart, ByteBuf secondPart) { } /** + * Reads a length encoded integer from the given buffer. Notice that a length encoded integer can be + * greater than {@link Long#MAX_VALUE}. In this case it should be used as an unsigned long. If we need + * assume the result as a smaller integer, add code comment to explain it. + *

* Note: it will change {@code buf} readerIndex. * * @param buf a readable buffer include a var integer. diff --git a/src/main/java/io/asyncer/r2dbc/mysql/message/server/ColumnCountMessage.java b/src/main/java/io/asyncer/r2dbc/mysql/message/server/ColumnCountMessage.java index a036e7a25..a23019a1f 100644 --- a/src/main/java/io/asyncer/r2dbc/mysql/message/server/ColumnCountMessage.java +++ b/src/main/java/io/asyncer/r2dbc/mysql/message/server/ColumnCountMessage.java @@ -40,6 +40,7 @@ public int getTotalColumns() { } static ColumnCountMessage decode(ByteBuf buf) { + // JVM does NOT support arrays longer than Integer.MAX_VALUE return new ColumnCountMessage(Math.toIntExact(VarIntUtils.readVarInt(buf))); } diff --git a/src/main/java/io/asyncer/r2dbc/mysql/message/server/DefinitionMetadataMessage.java b/src/main/java/io/asyncer/r2dbc/mysql/message/server/DefinitionMetadataMessage.java index 3fb4950ab..a2aac7a8b 100644 --- a/src/main/java/io/asyncer/r2dbc/mysql/message/server/DefinitionMetadataMessage.java +++ b/src/main/java/io/asyncer/r2dbc/mysql/message/server/DefinitionMetadataMessage.java @@ -173,7 +173,8 @@ private static DefinitionMetadataMessage decode41(ByteBuf buf, ConnectionContext String column = readVarIntSizedString(buf, charset); String originColumn = readVarIntSizedString(buf, charset); - VarIntUtils.readVarInt(buf); // skip constant 0x0c encoded by var integer + // Skip constant 0x0c encoded by var integer + VarIntUtils.readVarInt(buf); int collationId = buf.readUnsignedShortLE(); long size = buf.readUnsignedIntLE(); @@ -185,7 +186,7 @@ private static DefinitionMetadataMessage decode41(ByteBuf buf, ConnectionContext } private static String readVarIntSizedString(ByteBuf buf, Charset charset) { - // JVM can NOT support string which length upper than maximum of int32 + // JVM does NOT support strings longer than Integer.MAX_VALUE int bytes = (int) VarIntUtils.readVarInt(buf); if (bytes == 0) { diff --git a/src/main/java/io/asyncer/r2dbc/mysql/message/server/LargeFieldReader.java b/src/main/java/io/asyncer/r2dbc/mysql/message/server/LargeFieldReader.java index 3eb133aac..9842687b1 100644 --- a/src/main/java/io/asyncer/r2dbc/mysql/message/server/LargeFieldReader.java +++ b/src/main/java/io/asyncer/r2dbc/mysql/message/server/LargeFieldReader.java @@ -99,12 +99,12 @@ public FieldValue readVarIntSizedField() { fieldSize = VarIntUtils.readVarInt(currentBuf); } - // Refresh non empty buffer because current buffer has been read. + // Refresh non-empty buffer because current buffer has been read. currentBuf = nonEmptyBuffer(); List results = readSlice(currentBuf, fieldSize); - if (fieldSize > Integer.MAX_VALUE) { + if (Long.compareUnsigned(fieldSize, Integer.MAX_VALUE) > 0) { return retainedLargeField(results); } @@ -130,26 +130,30 @@ protected void deallocate() { * list instead of a single buffer. * * @param current the current {@link ByteBuf} in {@link #buffers}. - * @param length the length of read. + * @param length the length of read, it can be an unsigned long. * @return result buffer list, should NEVER retain any buffer. */ private List readSlice(ByteBuf current, long length) { ByteBuf buf = current; List results = new ArrayList<>(Math.max( - (int) Math.min((length / Envelopes.MAX_ENVELOPE_SIZE) + 2, Byte.MAX_VALUE), 10)); + (int) Math.min(Long.divideUnsigned(length, Envelopes.MAX_ENVELOPE_SIZE) + 2, Byte.MAX_VALUE), + 10 + )); long totalSize = 0; int bufReadable; // totalSize + bufReadable <= length - while (totalSize <= length - (bufReadable = buf.readableBytes())) { + while (Long.compareUnsigned(totalSize, length - (bufReadable = buf.readableBytes())) <= 0) { totalSize += bufReadable; // No need readSlice because currentBufIndex will be increment after List pushed. results.add(buf); buf = this.buffers[++this.currentBufIndex]; } - if (length > totalSize) { - // need bytes = length - `results` real length = length - (totalSize - `buf` length) + // totalSize < length + if (Long.compareUnsigned(length, totalSize) > 0) { + // need bytes = length - `results` length = length - totalSize + // length - totalSize should be an int due to while loop above. results.add(buf.readSlice((int) (length - totalSize))); } // else results has filled by prev buffer, and currentBufIndex is unread for now. diff --git a/src/main/java/io/asyncer/r2dbc/mysql/message/server/NormalFieldReader.java b/src/main/java/io/asyncer/r2dbc/mysql/message/server/NormalFieldReader.java index b65d30024..9379a0265 100644 --- a/src/main/java/io/asyncer/r2dbc/mysql/message/server/NormalFieldReader.java +++ b/src/main/java/io/asyncer/r2dbc/mysql/message/server/NormalFieldReader.java @@ -105,6 +105,7 @@ public boolean release(int decrement) { } private static ByteBuf readVarIntSizedRetained(ByteBuf buf) { + // Normal field will NEVER be greater than Integer.MAX_VALUE. int size = (int) VarIntUtils.readVarInt(buf); if (size == 0) { // Use EmptyByteBuf, new buffer no need to be retained. diff --git a/src/main/java/io/asyncer/r2dbc/mysql/message/server/OkMessage.java b/src/main/java/io/asyncer/r2dbc/mysql/message/server/OkMessage.java index f5c91a60e..52e3bce2f 100644 --- a/src/main/java/io/asyncer/r2dbc/mysql/message/server/OkMessage.java +++ b/src/main/java/io/asyncer/r2dbc/mysql/message/server/OkMessage.java @@ -160,10 +160,11 @@ static OkMessage decode(ByteBuf buf, ConnectionContext context) { if (size > sizeAfterVarInt) { information = buf.toString(readerIndex, buf.writerIndex() - readerIndex, charset); } else { + // JVM does NOT support strings longer than Integer.MAX_VALUE information = buf.toString(buf.readerIndex(), (int) size, charset); } - // Ignore session track, it is not human readable and useless for R2DBC client. + // Ignore session track, it is not human-readable and useless for R2DBC client. return new OkMessage(affectedRows, lastInsertId, serverStatuses, warnings, information); }