Manually handle gzipped responses (#281)

A workaround for DynamoDB quirk - it doesn't actually gzip large error responses but provides gzip header which breaks automatic decompression.

Author: firenero

src/EfficientDynamoDb.IntegrationTests/DataPlane/TransactWrite/TestUser.cs

@@ -0,0 +1,25 @@
+using EfficientDynamoDb.Attributes;
+
+namespace EfficientDynamoDb.IntegrationTests.DataPlane.TransactWrite;
+
+[DynamoDbTable(TestHelper.TestTableName)]
+public record TestUser
+{
+    [DynamoDbProperty("pk", DynamoDbAttributeType.PartitionKey)]
+    public required string PartitionKey { get; init; }
+
+    [DynamoDbProperty("sk", DynamoDbAttributeType.SortKey)]
+    public required string SortKey { get; init; }
+
+    [DynamoDbProperty("name")]
+    public string Name { get; init; } = "";
+
+    [DynamoDbProperty("age")]
+    public int Age { get; init; }
+
+    [DynamoDbProperty("email")]
+    public string Email { get; init; } = "";
+
+    [DynamoDbProperty("large_data")]
+    public string LargeData { get; init; } = "";
+}

src/EfficientDynamoDb.IntegrationTests/DataPlane/TransactWrite/TransactWriteShould.cs

@@ -0,0 +1,140 @@
+using EfficientDynamoDb.Exceptions;
+using NUnit.Framework;
+using Shouldly;
+
+namespace EfficientDynamoDb.IntegrationTests.DataPlane.TransactWrite;
+
+[TestFixture]
+public class TransactWriteShould
+{
+    private const string KeyPrefix = "effddb_tests-transact_write";
+    private static readonly Random Random = new(42);
+
+    private DynamoDbContext _context = null!;
+    private List<TestUser> _itemsToCleanup = null!;
+
+    [SetUp]
+    public void SetUp()
+    {
+        _context = TestHelper.CreateContext();
+        _itemsToCleanup = [];
+    }
+
+    [TearDown]
+    public async Task TearDown()
+    {
+        if (_itemsToCleanup.Count != 0)
+        {
+            await _context.BatchWrite()
+                .WithItems(_itemsToCleanup.Select(user => Batch.DeleteItem<TestUser>().WithPrimaryKey(user.PartitionKey, user.SortKey)))
+                .ExecuteAsync();
+        }
+    }
+
+    [Test]
+    public async Task PutItemsSuccessfully()
+    {
+        var item1 = new TestUser
+        {
+            PartitionKey = $"{KeyPrefix}-put-pk-1",
+            SortKey = $"{KeyPrefix}-put-sk-1",
+            Name = "Transact User 1",
+            Age = 25,
+            Email = "transact1@example.com"
+        };
+        var item2 = new TestUser
+        {
+            PartitionKey = $"{KeyPrefix}-put-pk-2",
+            SortKey = $"{KeyPrefix}-put-sk-2",
+            Name = "Transact User 2",
+            Age = 30,
+            Email = "transact2@example.com"
+        };
+
+        _itemsToCleanup.Add(item1);
+        _itemsToCleanup.Add(item2);
+
+        await _context.TransactWrite()
+            .WithItems(
+                Transact.PutItem(item1),
+                Transact.PutItem(item2)
+            )
+            .ExecuteAsync();
+
+        var retrieved1 = await _context.GetItem<TestUser>()
+            .WithPrimaryKey(item1.PartitionKey, item1.SortKey)
+            .WithConsistentRead(true)
+            .ToItemAsync();
+        var retrieved2 = await _context.GetItem<TestUser>()
+            .WithPrimaryKey(item2.PartitionKey, item2.SortKey)
+            .WithConsistentRead(true)
+            .ToItemAsync();
+
+        retrieved1.ShouldBe(item1);
+        retrieved2.ShouldBe(item2);
+    }
+
+    [Test]
+    public async Task ThrowTransactionCanceledException_WhenConditionCheckFails()
+    {
+        var existingItem = new TestUser
+        {
+            PartitionKey = $"{KeyPrefix}-cond-pk",
+            SortKey = $"{KeyPrefix}-cond-sk",
+            Name = "Existing User",
+            Age = 30,
+            Email = "existing@example.com"
+        };
+
+        _itemsToCleanup.Add(existingItem);
+        await _context.PutItemAsync(existingItem);
+
+        var ex = await Should.ThrowAsync<TransactionCanceledException>(() =>
+            _context.TransactWrite()
+                .WithItems(
+                    Transact.PutItem(existingItem).WithCondition(c => c.On(x => x.PartitionKey).NotExists())
+                )
+                .ExecuteAsync());
+
+        ex.CancellationReasons.ShouldNotBeEmpty();
+        ex.CancellationReasons[0].Code.ShouldBe("ConditionalCheckFailed");
+    }
+
+    [Test]
+    public async Task ThrowTransactionCanceledException_WithMultipleItems_WhenConditionCheckFails()
+    {
+        var items = Enumerable.Range(1, 25)
+            .Select(i => new TestUser
+            {
+                PartitionKey = $"{KeyPrefix}-multi-pk-{i:D3}",
+                SortKey = $"{KeyPrefix}-multi-sk-{i:D3}",
+                Name = $"Multi User {i}",
+                Age = 20 + i,
+                Email = $"multi{i}@example.com",
+                LargeData = GenerateLargeString(1000)
+            })
+            .ToList();
+
+        _itemsToCleanup.AddRange(items);
+
+        // Pre-create first item so the condition check fails
+        await _context.PutItemAsync(items[0]);
+
+        var ex = await Should.ThrowAsync<TransactionCanceledException>(() =>
+            _context.TransactWrite()
+                .WithItems(items.Select(x => Transact.PutItem(x).WithCondition(c => c.On(p => p.PartitionKey).NotExists())))
+                .ExecuteAsync());
+
+        ex.CancellationReasons.ShouldNotBeEmpty();
+        ex.CancellationReasons.Any(r => r.Code == "ConditionalCheckFailed").ShouldBeTrue();
+    }
+
+    private static string GenerateLargeString(int approximateLength)
+    {
+        const string chars = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789";
+        var result = new char[approximateLength];
+        for (var i = 0; i < approximateLength; i++)
+            result[i] = chars[Random.Next(chars.Length)];
+        return new string(result);
+    }
+}

src/EfficientDynamoDb/Configs/Http/DefaultHttpClientFactory.cs

@@ -13,7 +13,7 @@ private DefaultHttpClientFactory()
         {
             _httpClient = new HttpClient(new HttpClientHandler
             {
-                AutomaticDecompression = DecompressionMethods.GZip
+                AutomaticDecompression = DecompressionMethods.None
             });
         }
 

src/EfficientDynamoDb/Configs/Http/IHttpClientFactory.cs

@@ -4,6 +4,14 @@ namespace EfficientDynamoDb.Configs.Http
 {
     public interface IHttpClientFactory
     {
+        /// <summary>
+        /// Creates or returns an <see cref="HttpClient"/> used for DynamoDB requests.
+        /// </summary>
+        /// <remarks>
+        /// The returned <see cref="HttpClient"/> must have <c>AutomaticDecompression</c> set to <see cref="System.Net.DecompressionMethods.None"/>.
+        /// EfficientDynamoDb handles gzip decompression manually, and enabling automatic decompression on the handler
+        /// will interfere with CRC verification and error response parsing, leading to incorrect behavior.
+        /// </remarks>
         HttpClient CreateHttpClient();
     }
 }

src/EfficientDynamoDb/DynamoDbContext/DynamoDbContext.cs

@@ -43,7 +43,7 @@ async Task<TResponse> IDynamoDbContext.ExecuteAsync<TResponse>(HttpContent httpC
 
         private async ValueTask<TResult> ReadAsync<TResult>(HttpResponseMessage response, CancellationToken cancellationToken = default) where TResult : class
         {
-            await using var responseStream = await response.Content.ReadAsStreamAsync().ConfigureAwait(false);
+            await using var responseStream = await response.GetDecodedStreamAsync().ConfigureAwait(false);
 
             var expectedCrc = GetExpectedCrc(response);
             var classInfo = Config.Metadata.GetOrAddClassInfo(typeof(TResult), typeof(JsonObjectDdbConverter<TResult>));

src/EfficientDynamoDb/DynamoDbContext/DynamoDbLowLevelContext.cs

@@ -203,7 +203,7 @@ private async ValueTask<HttpContent> BuildHttpContentAsync(UpdateItemRequest req
 
         internal static async ValueTask<Document?> ReadDocumentAsync(HttpResponseMessage response, IParsingOptions options, CancellationToken cancellationToken = default)
         {
-            await using var responseStream = await response.Content.ReadAsStreamAsync().ConfigureAwait(false);
+            await using var responseStream = await response.GetDecodedStreamAsync().ConfigureAwait(false);
 
             var expectedCrc = GetExpectedCrc(response);
             var result = await DdbJsonReader.ReadAsync(responseStream, options, expectedCrc.HasValue, cancellationToken).ConfigureAwait(false);
@@ -216,6 +216,11 @@ private async ValueTask<HttpContent> BuildHttpContentAsync(UpdateItemRequest req
 
         internal static uint? GetExpectedCrc(HttpResponseMessage response)
         {
+            // Unable to verify crc for gzipped content because DDB provides expected crc for the compressed data instead of decompressed.
+            // Validating it would require an additional pass over the compressed data to calculate crc.
+            if (response.Content.Headers.ContentEncoding.Contains("gzip"))
+                return null;
+            
             if (!response.Content.Headers.ContentLength.HasValue)
                 return null;
 

src/EfficientDynamoDb/DynamoDbContextConfig.cs

@@ -21,6 +21,16 @@ public class DynamoDbContextConfig
 
         public IAwsCredentialsProvider CredentialsProvider { get; }
 
+        /// <summary>
+        /// Factory used to create the <see cref="System.Net.Http.HttpClient"/> for DynamoDB requests.
+        /// Defaults to an internal factory with a shared <see cref="System.Net.Http.HttpClient"/>.
+        /// </summary>
+        /// <remarks>
+        /// When providing a custom factory, ensure the returned <see cref="System.Net.Http.HttpClient"/> has
+        /// <c>AutomaticDecompression</c> set to <see cref="System.Net.DecompressionMethods.None"/>.
+        /// EfficientDynamoDb manages gzip decompression manually; enabling automatic decompression on the handler
+        /// will interfere with CRC verification and error response parsing, leading to incorrect behavior.
+        /// </remarks>
         public IHttpClientFactory HttpClientFactory { get; set; } = DefaultHttpClientFactory.Instance;
 
         public IReadOnlyCollection<DdbConverter> Converters

src/EfficientDynamoDb/Internal/ErrorHandler.cs

@@ -1,5 +1,6 @@
 using System;
 using System.IO;
+using System.IO.Compression;
 using System.Net;
 using System.Net.Http;
 using System.Text.Json;
@@ -21,7 +22,7 @@ internal static class ErrorHandler
         {
             PropertyNameCaseInsensitive = true,
         };
-        
+
         public static async Task<DdbException> ProcessErrorAsync(DynamoDbContextMetadata metadata, HttpResponseMessage response, CancellationToken cancellationToken = default)
         {
             try
@@ -30,37 +31,71 @@ public static async Task<DdbException> ProcessErrorAsync(DynamoDbContextMetadata
                 if (response.StatusCode == HttpStatusCode.ServiceUnavailable)
                     return new ServiceUnavailableException("DynamoDB is currently unavailable. (This should be a temporary state.)");
 
-                var recyclableStream = new RecyclableMemoryStream(DynamoDbHttpContent.MemoryStreamManager);
-                try
-                {
-                    await responseStream.CopyToAsync(recyclableStream, cancellationToken).ConfigureAwait(false);
-
-                    recyclableStream.Position = 0;
-                    var error = await JsonSerializer.DeserializeAsync<Error>(recyclableStream, SerializerOptions, cancellationToken).ConfigureAwait(false);
-                    recyclableStream.Position = 0;
-                    
-                    switch (response.StatusCode)
-                    {
-                        case HttpStatusCode.BadRequest:
-                            return await ProcessBadRequestAsync(metadata, recyclableStream, error, cancellationToken).ConfigureAwait(false);
-                        case HttpStatusCode.InternalServerError:
-                            return new InternalServerErrorException(error.Message);
-                        default:
-                            return new DdbException(error.Message);
-                    }
-                }
-                finally
+                var hasGzipHeader = response.Content.Headers.ContentEncoding.Contains("gzip");
+                await using var parsedStreamOwner = await DecodeErrorStreamAsync(responseStream, hasGzipHeader, cancellationToken).ConfigureAwait(false);
+
+                var errorStream = parsedStreamOwner.Stream;
+                var error = await JsonSerializer.DeserializeAsync<Error>(errorStream, SerializerOptions, cancellationToken).ConfigureAwait(false);
+                errorStream.Position = 0;
+
+                return response.StatusCode switch
                 {
-                    await recyclableStream.DisposeAsync().ConfigureAwait(false);
-                }
+                    HttpStatusCode.BadRequest => await ProcessBadRequestAsync(metadata, errorStream, error, cancellationToken).ConfigureAwait(false),
+                    HttpStatusCode.InternalServerError => new InternalServerErrorException(error.Message),
+                    _ => new DdbException(error.Message)
+                };
             }
             finally
             {
                 response.Dispose();
             }
         }
+        
+        private static bool IsGzipEncoded(RecyclableMemoryStream stream)
+        {
+            Span<byte> magic = stackalloc byte[2];
+            _ = stream.Read(magic);
+            stream.Position = 0;
+            return magic[0] == 0x1F && magic[1] == 0x8B;
+        }
+
+        private static async Task<DecodedStreamOwner> DecodeErrorStreamAsync(Stream responseStream, bool hasGzipHeader, CancellationToken ct = default)
+        {
+            var recyclableStream = new RecyclableMemoryStream(DynamoDbHttpContent.MemoryStreamManager);
+            RecyclableMemoryStream? decompressedStream = null;
+            try
+            {
+                await responseStream.CopyToAsync(recyclableStream, ct).ConfigureAwait(false);
+                recyclableStream.Position = 0;
+                if (!hasGzipHeader || recyclableStream.Length < 2)
+                    return new(recyclableStream, null);
+
+                // DynamoDB lies: error responses may carry Content-Encoding: gzip but the body is NOT gzipped.
+                // When the header claims gzip, detect actual gzip by inspecting magic bytes (0x1F 0x8B).
+                if (!IsGzipEncoded(recyclableStream))
+                    return new(recyclableStream, null);
+
+                // Rare path: error response was actually gzip-encoded — decompress eagerly into a pooled stream.
+                decompressedStream = new RecyclableMemoryStream(DynamoDbHttpContent.MemoryStreamManager);
+                await using (var gz = new GZipStream(recyclableStream, CompressionMode.Decompress, leaveOpen: true))
+                {
+                    await gz.CopyToAsync(decompressedStream, ct).ConfigureAwait(false);
+                }
+
+                recyclableStream.Position = 0;
+                decompressedStream.Position = 0;
+                return new(recyclableStream, decompressedStream);
+            }
+            catch (Exception)
+            {
+                await recyclableStream.DisposeAsync().ConfigureAwait(false);
+                if (decompressedStream is not null)
+                    await decompressedStream.DisposeAsync().ConfigureAwait(false);
+                throw;
+            }
+        }
 
-        private static ValueTask<DdbException> ProcessBadRequestAsync(DynamoDbContextMetadata metadata, MemoryStream recyclableStream, Error error, CancellationToken cancellationToken)
+        private static ValueTask<DdbException> ProcessBadRequestAsync(DynamoDbContextMetadata metadata, Stream recyclableStream, Error error, CancellationToken cancellationToken)
         {
             if (error.Type is null)
                 return new(new DdbException(string.Empty));
@@ -122,6 +157,33 @@ async ValueTask<DdbException> ParseConditionalCheckFailedException()
                 return new ConditionalCheckFailedException(conditionalCheckFailedResponse.Value!.Item, error.Message);
             }
         }
+        
+        private readonly struct DecodedStreamOwner : IAsyncDisposable, IDisposable
+        {
+            private readonly RecyclableMemoryStream _original;
+            private readonly RecyclableMemoryStream? _decompressed;
+
+            public DecodedStreamOwner(RecyclableMemoryStream original, RecyclableMemoryStream? decompressed)
+            {
+                _original = original;
+                _decompressed = decompressed;
+            }
+
+            public Stream Stream => _decompressed ?? _original;
+
+            public async ValueTask DisposeAsync()
+            {
+                await _original.DisposeAsync().ConfigureAwait(false);
+                if (_decompressed is not null)
+                    await _decompressed.DisposeAsync().ConfigureAwait(false);
+            }
+
+            public void Dispose()
+            {
+                _original.Dispose();
+                _decompressed?.Dispose();
+            }
+        }
 
         private readonly struct Error
         {