Conversation
ba8840b to
6d89325
Compare
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
6d89325 to
da1275c
Compare
589ef8e to
f130ccc
Compare
f130ccc to
0e94df8
Compare
0e94df8 to
00f7932
Compare
00f7932 to
c6c28b1
Compare
c6c28b1 to
4a4b890
Compare
4a4b890 to
14f6204
Compare
14f6204 to
0a389c4
Compare
0a389c4 to
7d192ff
Compare
whowes
changed the title
7d192ff to
33b9862
Compare
Introduce RewindableStreamBuffer managing a single-chunk buffer over an InputStream, supporting forward compaction and topping up upon recovery realignment without mark()/reset(). Enforces boundaries by throwing FailedPreconditionException when a server offset is below the base offset or beyond the current buffer window. Use payloadLength in ChunkUploadRequest to avoid allocating temporary byte arrays for full-sized chunks while reusing a single backing array.
33b9862 to
12f0b82
Compare
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request introduces a new RewindableStreamBuffer class to manage a single-chunk buffer over an InputStream for resumable uploads, supporting forward compaction and topping up upon recovery realignment. It also updates ResumableUploadChunkCoordinator to use this new buffer, and modifies ChunkUploadRequest to track and validate payloadLength. However, a critical issue was identified in ResumableUploadChunkCoordinator where unconditionally calling buffer.fill(currentOffset) inside transmitChunk would overwrite the buffer even if it had already been populated and realigned during recovery, potentially causing data corruption. A conditional check should be added to only fill the buffer when necessary.
| try { | ||
| bytesRead = ByteStreams.read(payload, buffer, 0, chunkSize); | ||
| buffer.fill(currentOffset); | ||
| } catch (IOException e) { | ||
| result.setException(e); | ||
| return; | ||
| } |
There was a problem hiding this comment.
Unconditionally calling buffer.fill(currentOffset) inside transmitChunk will overwrite the buffer even if it has already been correctly populated and realigned (e.g., via realignTo(committedOffset) during a recovery flow). This defeats the purpose of the realignTo method and would result in data corruption or unnecessary stream reads during recovery.
We should only call buffer.fill(currentOffset) if the buffer does not already contain the data for the requested currentOffset (i.e., when buffer.getBufferBaseOffset() != currentOffset or when the buffer is in its initial empty state).
| try { | |
| bytesRead = ByteStreams.read(payload, buffer, 0, chunkSize); | |
| buffer.fill(currentOffset); | |
| } catch (IOException e) { | |
| result.setException(e); | |
| return; | |
| } | |
| if (buffer.getBufferBaseOffset() != currentOffset || (buffer.isEmpty() && !buffer.isFinal())) { | |
| try { | |
| buffer.fill(currentOffset); | |
| } catch (IOException e) { | |
| result.setException(e); | |
| return; | |
| } | |
| } |
|
|
1 participant
Introduces
RewindableStreamBufferto manage a single-chunk buffer for the user-providedInputStream. This will be used in the next PR to "rewind" the stream as part of the query-command-based recovery process.The length of upload request payloads is now included in the
ChunkUploadRequestto allow for more efficient propagation of the content bytes from stream -> buffer -> wire.