anton-vinogradov commented on PR #13329:
URL: https://github.com/apache/ignite/pull/13329#issuecomment-4895522810
Reviewed the whole change (the `CompressedMessage`/serializer part matches
my #13325, so the findings there carry over; the `DirectMessageReader` part —
`curStream` cache and `tmpReader` reuse — is new and I looked at it from
scratch). The `curStream` cache itself checks out: all three mutation points
(`setBuffer` / `beforeNestedRead` / `afterNestedRead`) keep it in sync, depth
only ever changes inside `try/finally` in `DirectByteBufferStream`, and every
production `reset()` is followed by `setBuffer` before the next read. The
benchmark compiles with `-Pbenchmarks` and runs. Found the following, roughly
by severity.
### 1. `tmpReader` reuse: one partial inner read poisons every subsequent
compressed field on the connection (reproduced)
`reset()` is a shallow reset: `DirectMessageState.reset()` only zeroes
`stack[0].state` and asserts `pos == 0`. It does **not** clear the per-stream
parse state (`msgTypeDone`, half-read `msg`, `readSize`/`readItems`, map
iteration, `tmpArr`) — those are only cleared when a `readMessage`/`readMap`
*finishes successfully* — and `backward(false)` intentionally preserves the
state of abandoned deeper items for resume.
The old code was immune by construction (`new DirectMessageReader(...)` per
compressed field). With reuse, this sequence breaks: a `CompressedMessage`
whose inflated payload is logically incomplete (incompatible peer, corruption,
understated `dataSize` — see #4) makes `fun.apply(tmpReader)` return with
`lastFinished == false` and **no exception**, so the session stays alive. The
next, fully valid compressed field on the same per-connection reader reuses the
poisoned `tmpReader`: `readMessage` sees the stale `msgTypeDone == true`, skips
the type header of the new payload and resumes the *old* half-read message at a
wrong offset → the valid field reads as `null` with `lastRead == false` forever
(outer parser waits for bytes that never come), or a silently corrupted message
is assembled.
I verified this with a quick repro against the branch classes: after one
truncated payload, a following valid field fails to deserialize; with a fresh
reader (old behavior) the same field reads fine.
The clean fix falls out of an invariant this method already has:
`tmpReader`'s buffer is always the *complete* payload, so a partial inner read
is never "need more bytes" — it is always corruption. Instead of propagating it
as `lastRead = false`:
```java
T res = fun.apply(tmpReader);
if (!tmpReader.state.item().stream.lastFinished()) {
tmpReader = null; // Don't reuse a reader with half-read state.
throw new IgniteException("Failed to deserialize compressed payload: " +
"uncompressed data ended unexpectedly [dataSize=" + msg0.dataSize()
+ ']');
}
lastRead = true;
```
This also turns the silent protocol desync from #4 into a localized,
diagnosable error, and dropping `tmpReader` matters because the session is not
guaranteed to die: the legacy selector loop (`IGNITE_NO_SELECTOR_OPTS=true`,
`GridNioServer.processSelectedKeys`) logs the exception and keeps the session —
only the optimized path closes it.
### 2. `tmpReader` retains the last uncompressed payload for the connection
lifetime
`tmpReader` is a field of the per-connection reader (session meta,
`GridDirectParser`), and after the read its `stack[0].stream` keeps referencing
the `ByteBuffer.wrap(uncompressed)`. For exchange traffic
(`GridDhtPartitionsFullMessage` on large topologies) that's megabytes per
connection held until the *next* compressed message on that connection or TCP
session close; on the coordinator — one payload per node connection. The old
per-call reader made it garbage immediately. One line after `fun.apply`:
```java
tmpReader.setBuffer(EMPTY_BUF); // static final ByteBuffer EMPTY_BUF =
ByteBuffer.wrap(new byte[0]);
```
### Carried over from #13325 (identical `CompressedMessage`/serializer code)
These three have no benign trigger — the sender computes `dataSize =
buf.remaining()` correctly by construction — but the reworked receive path made
the wire `dataSize` load-bearing, while master was insensitive to it (`tmpBuf`
grew from actually received chunks, `readAllBytes()` sized the output,
`dataSize` was assert-only). So on a corrupted stream the failure quality
degrades vs master, and the guards are cheap:
**3. `dataSize > 0` + immediate `finalChunk` → bare NPE.** `readFrom()`
legitimately returns `true` with `chunks == null` when the first `finalChunk`
boolean is `true` (state 1 exits before state 2 runs) — the new null-chunk
guard doesn't cover this path. `uncompress()` then returns `null` and the
caller hits `ByteBuffer.wrap(null)` → NPE in the NIO worker. Master threw
`IgniteException (EOFException: Unexpected end of ZLIB input stream)` on the
same input. Fix in `uncompress()`:
```java
if (chunks == null)
throw new IgniteException("Compressed stream is truncated [expected=" +
dataSize + ", inflated=0]");
```
(The `null` return contract is dead: the caller already short-circuits
`dataSize() == 0`.)
**4. Understated `dataSize` truncates silently.** Both inflate loops are
bounded by `off < dataSize`, so a valid stream that inflates to *more* than
`dataSize` is cut short and the `off != dataSize` check can't fire — it only
covers the "shorter" direction, while master's `assert uncompressedData.length
== dataSize` covered both. The consequence is the partial-read cascade from #1.
Besides the reader-level guard above, the invariant is cheap to restore locally
with a probe inflate after the main loop (hoist the chunk index out of the
`for`):
```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 and finishes here.
Worth a unit test next to `testReadFailsOnNullChunk`: valid compressed payload
with the `dataSize` field patched smaller → expect `IgniteException`.
**5. `dataSize` drives allocations before any validation.** Negative values:
`dataSize <= -20480` → `new ArrayList<>(msg.dataSize / CHUNK_SIZE + 1)` →
`IllegalArgumentException: Illegal Capacity`; `-1..-20479` → `new
byte[dataSize]` → `NegativeArraySizeException`. Huge values: `dataSize =
Integer.MAX_VALUE` + one tiny chunk → ~2 GB upfront allocation →
`OutOfMemoryError` → failure handler. Minimal guard in `readFrom` state 0:
```java
if (msg.dataSize < 0)
throw new IgniteException("Invalid compressed message data size: " +
msg.dataSize);
```
Optionally bound the allocation by what was actually received (raw deflate
can't expand beyond ~1032:1): before `new byte[dataSize]`, check `dataSize <=
totalCompressedLen * 1032L + 64`.
### 6. Benchmark nit
`@Setup` relies on `assert finished` after `writeMessage`, but asserts are
off in the JMH fork by default — if the message ever outgrows the 4 MB buffer
(larger `entries` param), the benchmark will happily measure deserialization of
a truncated stub. Make it explicit: `if (!finished) throw new
IllegalStateException("Message doesn't fit the buffer");`
🤖 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]
