IGNITE-28853 Harden compressed payload deserialization against corrupted streams

[anton-vinogradov]

Implements the fixes for the issues raised in the review of apache#13329 (see apache#13329 (comment)). Targets this PR's branch, so merging brings them straight into the PR.

What's fixed

• Partial inner read of a compressed payload now fails fast and drops tmpReader. The payload buffer handed to tmpReader is always complete, so lastFinished == false after the inner deserialization is stream corruption, not a lack of data. Previously it propagated as lastRead = false, making the outer parser wait for socket bytes that never come (protocol desync), while the reused tmpReader kept half-read stream state (msgTypeDone, partial msg, readSize) that reset() doesn't clear and poisoned every subsequent compressed field on the connection. Now it throws IgniteException, and tmpReader is dropped in a finally on any failure — including exceptions thrown from inside the inner deserialization.
• tmpReader no longer pins the last uncompressed payload. The reader lives in session meta for the connection lifetime, and every stream that participated in the nested read kept a reference to the payload array (megabytes for exchange messages). New releasePayload() swaps all created streams to a shared empty buffer via DirectMessageState.forEachItem().
• uncompress() validates the stream against the size header. chunks == null with dataSize > 0 throws a diagnosable IgniteException instead of returning null into ByteBuffer.wrap(null) (NPE); a 1032:1 sanity bound (max raw-deflate expansion) rejects a corrupted size header before the new byte[dataSize] allocation; a one-byte probe inflate after the main loop restores the second direction of master's length == dataSize assert — a stream that inflates to more than dataSize now fails instead of being silently truncated.
• readFrom() rejects negative dataSize (previously IllegalArgumentException/NegativeArraySizeException from the guts of the serializer) and clamps the chunk-list pre-size, so a lying header can't inflate the ArrayList capacity allocation.
• Benchmark: assert finished in @Setup replaced with an explicit check — asserts are disabled in the JMH fork, a message outgrowing the buffer would silently benchmark a truncated stub.

Tests

Five new cases in CompressedMessageTest: negative dataSize, finalChunk with no chunks, understated dataSize (single- and multi-chunk), and a full readMessage(true) path over an envelope whose payload is a truncated message serialization.

Verification: CompressedMessageTest 7/7, DirectMarshallingMessagesTest green, -Pcheckstyle clean, benchmarks module compiles with -Pbenchmarks.

Known limitations (intentional)

• The upfront new byte[dataSize] allocation is bounded only by ~1032× of the actually received compressed bytes. Full protection would need incremental allocation (reintroduces the copying this PR removes) or a transport-level message size limit; note that master had no bound on the inflated size at all.
• A decode-time IgniteException still lands in the GridNioServer worker catch block (pre-existing route, shared with this PR's own null-chunk guard).

🤖 Generated with Claude Code