refactor(Storage): Modify HashingStream class to accomodate all retry scenarios

Add tests for retry scenarios

Author: mahendra-google

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

@@ -24,6 +24,7 @@
 using System.Linq;
 using System.Net;
 using System.Net.Http;
+using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 using Xunit;
@@ -454,6 +455,52 @@ public async Task InterceptAsync(HttpRequestMessage request, CancellationToken c
             }
         }
 
+        [Fact]
+        public void HashingStream_ShouldHandleRetries_WhenRestartedFromBeginning()
+        {
+            var data = Encoding.UTF8.GetBytes("The quick brown fox jumps over the lazy dog");
+            var baseStream = new MemoryStream(data);
+            var hashingStream = new CustomMediaUpload.HashingStream(baseStream);
+            var buffer = new byte[data.Length];
+
+            hashingStream.Read(buffer, 0, 10);
+            var hashAfterPartial = hashingStream.GetBase64Hash();
+
+            // Simulate the Retry logic: Seek back to the beginning
+            hashingStream.Position = 0;
+
+            hashingStream.Read(buffer, 0, data.Length);
+            var finalHash = hashingStream.GetBase64Hash();
+
+            var expectedHasher = new Crc32c();
+            expectedHasher.UpdateHash(data, 0, data.Length);
+            var expectedHash = Convert.ToBase64String(expectedHasher.GetHash());
+            Assert.Equal(expectedHash, finalHash);
+        }
+
+        [Fact]
+        public void HashingStream_ShouldHandleRetries_WhenSeekingBackwardsToIntermediatePoint()
+        {
+            var data = Encoding.UTF8.GetBytes("The quick brown fox jumps over the lazy dog");
+            var baseStream = new MemoryStream(data);
+            var hashingStream = new CustomMediaUpload.HashingStream(baseStream);
+            var buffer = new byte[data.Length];
+
+            hashingStream.Read(buffer, 0, 10);
+            var hashAfterPartial = hashingStream.GetBase64Hash();
+
+            // Simulate the Retry logic: Seek back to the intermediate point.
+            hashingStream.Position = 5;
+
+            hashingStream.Read(buffer, 0, data.Length);
+            var finalHash = hashingStream.GetBase64Hash();
+
+            var expectedHasher = new Crc32c();
+            expectedHasher.UpdateHash(data, 0, data.Length);
+            var expectedHash = Convert.ToBase64String(expectedHasher.GetHash());
+            Assert.Equal(expectedHash, finalHash);
+        }
+
         private Object GetExistingObject()
         {
             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

@@ -149,10 +149,11 @@ private bool IsFinalChunk(string rangeHeader)
             }
         }
 
-        private sealed class HashingStream : Stream
+        internal sealed class HashingStream : Stream
         {
             private readonly Stream _stream;
-            private Crc32c _hasher;
+            private readonly Crc32c _hasher;
+            private long _maxPositionHashed = 0;
 
             public HashingStream(Stream stream)
             {
@@ -162,55 +163,44 @@ public HashingStream(Stream stream)
 
             public override int Read(byte[] buffer, int offset, int count)
             {
+                long startingPos = _stream.Position;
                 int bytesRead = _stream.Read(buffer, offset, count);
-                if (bytesRead > 0)
-                {
-                    _hasher.UpdateHash(buffer, offset, bytesRead);
-                }
+                ProcessBytes(buffer, offset, bytesRead, startingPos);
                 return bytesRead;
             }
 
             public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
             {
+                long startingPos = _stream.Position;
                 int bytesRead = await _stream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
-                if (bytesRead > 0)
-                {
-                    _hasher.UpdateHash(buffer, offset, bytesRead);
-                }
+                ProcessBytes(buffer, offset, bytesRead, startingPos);
                 return bytesRead;
             }
 
-            public override long Position
+            private void ProcessBytes(byte[] buffer, int offset, int bytesRead, long startingPos)
             {
-                get => _stream.Position;
-                set
+                if (bytesRead <= 0) return;
+
+                // Only hash bytes that are beyond the furthest point we've already hashed.
+                // This handles the rewind and re-read scenario during retries.
+                if (startingPos + bytesRead > _maxPositionHashed)
                 {
-                    if (value < _stream.Position)
-                    {
-                        _hasher = new Crc32c();
-                    }
-                    _stream.Position = value;
+                    long newBytesStart = Math.Max(startingPos, _maxPositionHashed);
+                    int actuallyNewCount = (int) ((startingPos + bytesRead) - newBytesStart);
+                    int bufferOffset = offset + (int) (newBytesStart - startingPos);
+
+                    _hasher.UpdateHash(buffer, bufferOffset, actuallyNewCount);
+                    _maxPositionHashed = startingPos + bytesRead;
                 }
             }
 
-            public override long Seek(long offset, SeekOrigin origin)
+            public override long Position
             {
-                long target;
-                switch (origin)
-                {
-                    case SeekOrigin.Begin: target = offset; break;
-                    case SeekOrigin.Current: target = _stream.Position + offset; break;
-                    case SeekOrigin.End: target = _stream.Length + offset; break;
-                    default: target = _stream.Position; break;
-                }
-
-                if (target < _stream.Position)
-                {
-                    _hasher = new Crc32c();
-                }
-                return _stream.Seek(offset, origin);
+                get => _stream.Position;
+                set => _stream.Position = value;
             }
 
+            public override long Seek(long offset, SeekOrigin origin) => _stream.Seek(offset, origin);
             public string GetBase64Hash() => Convert.ToBase64String(_hasher.GetHash());
             public override bool CanRead => _stream.CanRead;
             public override bool CanSeek => _stream.CanSeek;