Fix BlobWriteStream: Commit block list on each flush (#23)

spotflow-io · Oct 11, 2024 · 264d2e6 · 264d2e6
1 parent 39448f7
commit 264d2e6
Show file tree

Hide file tree

Showing 2 changed files with 45 additions and 0 deletions.
diff --git a/src/Spotflow.InMemory.Azure.Storage/Blobs/Internals/BlobWriteStream.cs b/src/Spotflow.InMemory.Azure.Storage/Blobs/Internals/BlobWriteStream.cs
@@ -146,9 +146,17 @@ private async Task FlushUnsafeAsync(CancellationToken cancellationToken)
 
             _blocks.Add(blockId);
 
+            var commitResult = await blobClient.CommitBlockListAsync(_blocks, options: _commitOptions, cancellationToken);
+
+            var newETag = commitResult.Value.ETag;
+
+            _commitOptions.Conditions.IfMatch = newETag;
+            _stageOptions.Conditions.IfMatch = newETag;
+
             _bufferPosition = 0;
         }
 
+
         foreach (var interceptor in _interceptors)
         {
             await interceptor.FlushAsync();

diff --git a/tests/Tests/Storage/Blobs/BlobClientTests.cs b/tests/Tests/Storage/Blobs/BlobClientTests.cs
@@ -5,6 +5,8 @@
 using Azure.Storage.Blobs.Models;
 using Azure.Storage.Blobs.Specialized;
 
+using Spotflow.InMemory.Azure.Storage.FluentAssertions;
+
 using Tests.Utils;
 
 namespace Tests.Storage.Blobs;
@@ -326,6 +328,41 @@ public void OpenWrite_For_Existing_Blob_With_Conditions_Should_Fail(BlobClientTy
 
     }
 
+    [TestMethod]
+    [TestCategory(TestCategory.AzureInfra)]
+    [DataRow(BlobClientType.Generic)]
+    [DataRow(BlobClientType.Block)]
+    public void OpenWrite_Stream_Should_Create_Block_And_Commit_On_Each_Flush(BlobClientType clientType)
+    {
+        var containerClient = ImplementationProvider.GetBlobContainerClient();
+
+        containerClient.CreateIfNotExists();
+
+        var blobName = Guid.NewGuid().ToString();
+
+        var blobClient = containerClient.GetBlobBaseClient(blobName, clientType);
+
+        using (var stream = OpenWrite(blobClient, true))
+        {
+            blobClient.Should().HaveNoCommittedBlocks();
+
+            stream.WriteByte(1);
+            stream.Flush();
+
+            blobClient.Should().HaveCommittedBlocks(1);
+
+            stream.WriteByte(2);
+            stream.Flush();
+
+            blobClient.Should().HaveCommittedBlocks(2);
+
+            stream.WriteByte(2);
+
+        }
+
+        blobClient.Should().HaveCommittedBlocks(3);
+    }
+
 
     [TestMethod]
     [TestCategory(TestCategory.AzureInfra)]