Conversation
ba8840b to
6d89325
Compare
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces RewindableStreamBuffer to manage a single-chunk buffer over an InputStream for resumable uploads, supporting forward compaction and topping up during recovery realignment. ResumableUploadChunkCoordinator and ChunkUploadRequest are updated to utilize this buffer to avoid unnecessary array copies. The review feedback identifies two key issues: first, unconditionally calling streamBuffer.fill(currentOffset) on every chunk transmission can overwrite compacted bytes and cause silent data corruption; second, the realignTo method in RewindableStreamBuffer is currently implemented and tested but never actually integrated into the recovery path of the coordinator.
| try { | ||
| bytesRead = ByteStreams.read(payload, buffer, 0, chunkSize); | ||
| streamBuffer.fill(currentOffset); | ||
| } catch (IOException e) { | ||
| finish(null, e); | ||
| return; | ||
| } |
There was a problem hiding this comment.
Unconditionally calling streamBuffer.fill(currentOffset) on every chunk transmission will overwrite the compacted and topped-up bytes in the buffer if the buffer was previously realigned (e.g., during recovery realignment). This leads to silent data corruption because the stream will be read from its current position (which is ahead of currentOffset), resulting in incorrect bytes being uploaded at currentOffset. We should only call fill if the buffer does not already contain the data for currentOffset.
| try { | |
| bytesRead = ByteStreams.read(payload, buffer, 0, chunkSize); | |
| streamBuffer.fill(currentOffset); | |
| } catch (IOException e) { | |
| finish(null, e); | |
| return; | |
| } | |
| try { | |
| if (streamBuffer.getBufferBaseOffset() != currentOffset || streamBuffer.isEmpty()) { | |
| streamBuffer.fill(currentOffset); | |
| } | |
| } catch (IOException e) { | |
| finish(null, e); | |
| return; | |
| } |
| * offset or beyond the current buffer window | ||
| * @throws IOException if reading from the stream fails | ||
| */ | ||
| void realignTo(long committedOffset) throws IOException { |
There was a problem hiding this comment.
The realignTo method is implemented and thoroughly tested, but it is never actually called within ResumableUploadChunkCoordinator or any other production code in this pull request. Since buffer is a private field in ResumableUploadChunkCoordinator and has no package-private getter, the recovery realignment logic remains completely unused. Please ensure that realignTo is integrated into the recovery/retry path (e.g., when handling server-committed offsets after a recoverable failure) so that the rewindable stream buffer is actually utilized for chunk recovery.
6d89325 to
da1275c
Compare
589ef8e to
f130ccc
Compare
f130ccc to
0e94df8
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.
0e94df8 to
00f7932
Compare
|
|
1 participant
Introduces
RewindableStreamBufferto manage a reusable single-chunk buffer over anInputStream. Supports compacting unacknowledged bytes forward in-place and topping up from the stream when realigning to a server-committed offset.