anton-vinogradov commented on PR #13325:
URL: https://github.com/apache/ignite/pull/13325#issuecomment-4894905352
**Self-review, round 2: the new receive path made `dataSize` load-bearing,
and three malformed-stream cases now fail worse than on master.**
None of these have a benign trigger — the sender computes `dataSize =
buf.remaining()` and deflates exactly that buffer, so hitting them requires a
corrupted stream or a misbehaving peer (rolling upgrade is safe: old and new
senders produce the same wire contract). But on master the receive path was
*insensitive* to the header: `tmpBuf` grew from actually received chunks,
`readAllBytes()` sized the output, and `dataSize` was only checked by an
assert. Now the header drives allocations and the inflate loop bound, so the
failure quality degraded in all three cases. All are cheap to fix without
losing the perf gains.
### 1. `dataSize > 0` + `finalChunk` before any chunk → bare NPE instead of
a diagnosable error
`readFrom()` returns `true` with `msg.chunks == null` when the first
`finalChunk` boolean reads `true` (state 1 exits before state 2 ever runs).
`uncompress()` then returns `null`, and the only caller does
`tmpReader.setBuffer(ByteBuffer.wrap(uncompressed))` → NPE in the NIO worker
(`DirectMessageReader.java:498`). Master threw a meaningful `IgniteException`
(caused by `EOFException: Unexpected end of ZLIB input stream`) on this very
input.
**Fix** — treat it as the truncation it is, in `uncompress()`:
```java
if (chunks == null)
throw new IgniteException("Compressed stream is truncated [expected=" +
dataSize + ", inflated=0]");
```
The `null` return contract is dead anyway: `uncompress()` is only reachable
with `dataSize > 0` (the caller short-circuits `dataSize() == 0`), and it's
called exactly once per message.
### 2. `dataSize` drives allocations before any validation
* `dataSize <= -20480` → `new ArrayList<>(msg.dataSize / CHUNK_SIZE + 1)` →
`IllegalArgumentException: Illegal Capacity` in the NIO thread; `-1..-20479` →
`new byte[dataSize]` → `NegativeArraySizeException`.
* `dataSize = Integer.MAX_VALUE` + one tiny chunk → ~2 GB upfront allocation
in `uncompress()` → `OutOfMemoryError` → failure handler may stop the node.
Master only ever allocated proportionally to what was actually received.
**Fix** — validate the header where it's read (`readFrom`, state 0):
```java
if (msg.dataSize < 0)
throw new IgniteException("Invalid compressed message data size: " +
msg.dataSize);
```
Optionally, bound the upfront allocation by what was actually received
before `new byte[dataSize]` (raw deflate can't expand beyond ~1032:1):
```java
long compressedTotal = 0;
for (byte[] chunk : chunks)
compressedTotal += chunk.length;
if (dataSize > compressedTotal * 1032L + 64)
throw new IgniteException("Compressed stream is truncated [expected=" +
dataSize +
", compressedBytes=" + compressedTotal + ']');
```
### 3. Understated `dataSize` now truncates silently — and the failure mode
is a protocol desync
Both inflate loops are bounded by `off < dataSize`, so when a *valid* stream
inflates to more than `dataSize`, the surplus is silently dropped and the `off
!= dataSize` check can't fire — it only covers the "shorter" direction. The
truncated prefix then goes to `tmpReader`; the nested serializer hits
end-of-buffer, `isLastRead() == false` propagates, the enclosing generated
serializer returns `false`, and `GridDirectParser` keeps waiting for socket
bytes that will never come for this message. The next message's bytes then get
parsed as a continuation → garbage direct type / connection teardown far from
the actual fault, or an idle hang if there's no more traffic. Master returned
the full `readAllBytes()` result and handled this input correctly (and with
`-ea` gave a clear `AssertionError` right at the spot).
**Fix** — restore the second direction of the old `length == dataSize`
invariant with a probe inflate after the main loop (this is exactly what
`readAllBytes()` used to provide). Hoist the chunk index out of the `for` and
append:
```java
byte[] probe = new byte[1];
while (!inflater.finished()) {
if (inflater.needsInput()) {
if (i == chunks.size())
break;
inflater.setInput(chunks.get(i++));
}
if (inflater.inflate(probe, 0, 1) > 0)
throw new IgniteException("Compressed stream is longer than expected
[expected=" + dataSize + ']');
}
```
A well-formed stream just consumes the deflate terminator here and finishes;
any extra byte means the header lied. Worth a unit test next to the truncation
one: valid compressed payload, `dataSize` field patched to a smaller value →
expect `IgniteException`, not a silently truncated result.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
