Skip to content

Commit 99c6617

Browse files
refactor(Storage): Modify HashingStream class to accomodate all retry scenarios
Add tests for retry scenarios
1 parent c4da0a1 commit 99c6617

2 files changed

Lines changed: 70 additions & 33 deletions

File tree

apis/Google.Cloud.Storage.V1/Google.Cloud.Storage.V1.IntegrationTests/UploadObjectTest.cs

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
using System.IO;
2424
using System.Net;
2525
using System.Net.Http;
26+
using System.Text;
2627
using System.Threading;
2728
using System.Threading.Tasks;
2829
using Xunit;
@@ -426,6 +427,52 @@ public async Task InterceptAsync(HttpRequestMessage request, CancellationToken c
426427
}
427428
}
428429

430+
[Fact]
431+
public void HashingStream_ShouldHandleRetries_WhenRestartedFromBeginning()
432+
{
433+
var data = Encoding.UTF8.GetBytes("The quick brown fox jumps over the lazy dog");
434+
var baseStream = new MemoryStream(data);
435+
var hashingStream = new CustomMediaUpload.HashingStream(baseStream);
436+
var buffer = new byte[data.Length];
437+
438+
hashingStream.Read(buffer, 0, 10);
439+
var hashAfterPartial = hashingStream.GetBase64Hash();
440+
441+
// Simulate the Retry logic: Seek back to the beginning
442+
hashingStream.Position = 0;
443+
444+
hashingStream.Read(buffer, 0, data.Length);
445+
var finalHash = hashingStream.GetBase64Hash();
446+
447+
var expectedHasher = new Crc32c();
448+
expectedHasher.UpdateHash(data, 0, data.Length);
449+
var expectedHash = Convert.ToBase64String(expectedHasher.GetHash());
450+
Assert.Equal(expectedHash, finalHash);
451+
}
452+
453+
[Fact]
454+
public void HashingStream_ShouldHandleRetries_WhenSeekingBackwardsToIntermediatePoint()
455+
{
456+
var data = Encoding.UTF8.GetBytes("The quick brown fox jumps over the lazy dog");
457+
var baseStream = new MemoryStream(data);
458+
var hashingStream = new CustomMediaUpload.HashingStream(baseStream);
459+
var buffer = new byte[data.Length];
460+
461+
hashingStream.Read(buffer, 0, 10);
462+
var hashAfterPartial = hashingStream.GetBase64Hash();
463+
464+
// Simulate the Retry logic: Seek back to the intermediate point.
465+
hashingStream.Position = 5;
466+
467+
hashingStream.Read(buffer, 0, data.Length);
468+
var finalHash = hashingStream.GetBase64Hash();
469+
470+
var expectedHasher = new Crc32c();
471+
expectedHasher.UpdateHash(data, 0, data.Length);
472+
var expectedHash = Convert.ToBase64String(expectedHasher.GetHash());
473+
Assert.Equal(expectedHash, finalHash);
474+
}
475+
429476
private Object GetExistingObject()
430477
{
431478
var obj = _fixture.Client.UploadObject(_fixture.MultiVersionBucket, IdGenerator.FromGuid(), "application/octet-stream", GenerateData(100));

apis/Google.Cloud.Storage.V1/Google.Cloud.Storage.V1/CustomMediaUpload.cs

Lines changed: 23 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -149,10 +149,11 @@ private bool IsFinalChunk(string rangeHeader)
149149
}
150150
}
151151

152-
private sealed class HashingStream : Stream
152+
internal sealed class HashingStream : Stream
153153
{
154154
private readonly Stream _stream;
155-
private Crc32c _hasher;
155+
private readonly Crc32c _hasher;
156+
private long _maxPositionHashed = 0;
156157

157158
public HashingStream(Stream stream)
158159
{
@@ -162,55 +163,44 @@ public HashingStream(Stream stream)
162163

163164
public override int Read(byte[] buffer, int offset, int count)
164165
{
166+
long startingPos = _stream.Position;
165167
int bytesRead = _stream.Read(buffer, offset, count);
166-
if (bytesRead > 0)
167-
{
168-
_hasher.UpdateHash(buffer, offset, bytesRead);
169-
}
168+
ProcessBytes(buffer, offset, bytesRead, startingPos);
170169
return bytesRead;
171170
}
172171

173172
public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
174173
{
174+
long startingPos = _stream.Position;
175175
int bytesRead = await _stream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
176-
if (bytesRead > 0)
177-
{
178-
_hasher.UpdateHash(buffer, offset, bytesRead);
179-
}
176+
ProcessBytes(buffer, offset, bytesRead, startingPos);
180177
return bytesRead;
181178
}
182179

183-
public override long Position
180+
private void ProcessBytes(byte[] buffer, int offset, int bytesRead, long startingPos)
184181
{
185-
get => _stream.Position;
186-
set
182+
if (bytesRead <= 0) return;
183+
184+
// Only hash bytes that are beyond the furthest point we've already hashed.
185+
// This handles the rewind and re-read scenario during retries.
186+
if (startingPos + bytesRead > _maxPositionHashed)
187187
{
188-
if (value < _stream.Position)
189-
{
190-
_hasher = new Crc32c();
191-
}
192-
_stream.Position = value;
188+
long newBytesStart = Math.Max(startingPos, _maxPositionHashed);
189+
int actuallyNewCount = (int) ((startingPos + bytesRead) - newBytesStart);
190+
int bufferOffset = offset + (int) (newBytesStart - startingPos);
191+
192+
_hasher.UpdateHash(buffer, bufferOffset, actuallyNewCount);
193+
_maxPositionHashed = startingPos + bytesRead;
193194
}
194195
}
195196

196-
public override long Seek(long offset, SeekOrigin origin)
197+
public override long Position
197198
{
198-
long target;
199-
switch (origin)
200-
{
201-
case SeekOrigin.Begin: target = offset; break;
202-
case SeekOrigin.Current: target = _stream.Position + offset; break;
203-
case SeekOrigin.End: target = _stream.Length + offset; break;
204-
default: target = _stream.Position; break;
205-
}
206-
207-
if (target < _stream.Position)
208-
{
209-
_hasher = new Crc32c();
210-
}
211-
return _stream.Seek(offset, origin);
199+
get => _stream.Position;
200+
set => _stream.Position = value;
212201
}
213202

203+
public override long Seek(long offset, SeekOrigin origin) => _stream.Seek(offset, origin);
214204
public string GetBase64Hash() => Convert.ToBase64String(_hasher.GetHash());
215205
public override bool CanRead => _stream.CanRead;
216206
public override bool CanSeek => _stream.CanSeek;

0 commit comments

Comments
 (0)