From 7f4c792e579a9672c954c1b41b46da370a0ae575 Mon Sep 17 00:00:00 2001 From: Mykhailo Matviiv Date: Mon, 23 Feb 2026 00:18:19 +0100 Subject: [PATCH 1/3] Manual GZip compression --- .../Configs/Http/DefaultHttpClientFactory.cs | 2 +- .../DynamoDbContext/DynamoDbContext.cs | 2 +- .../DynamoDbLowLevelContext.cs | 7 +- .../Internal/ErrorHandler.cs | 110 ++++++++++++++---- src/EfficientDynamoDb/Internal/HttpApi.cs | 4 +- .../Internal/HttpResponseExtensions.cs | 24 ++++ 6 files changed, 121 insertions(+), 28 deletions(-) create mode 100644 src/EfficientDynamoDb/Internal/HttpResponseExtensions.cs diff --git a/src/EfficientDynamoDb/Configs/Http/DefaultHttpClientFactory.cs b/src/EfficientDynamoDb/Configs/Http/DefaultHttpClientFactory.cs index b529217e..603ff6c9 100644 --- a/src/EfficientDynamoDb/Configs/Http/DefaultHttpClientFactory.cs +++ b/src/EfficientDynamoDb/Configs/Http/DefaultHttpClientFactory.cs @@ -13,7 +13,7 @@ private DefaultHttpClientFactory() { _httpClient = new HttpClient(new HttpClientHandler { - AutomaticDecompression = DecompressionMethods.GZip + AutomaticDecompression = DecompressionMethods.None }); } diff --git a/src/EfficientDynamoDb/DynamoDbContext/DynamoDbContext.cs b/src/EfficientDynamoDb/DynamoDbContext/DynamoDbContext.cs index 2b448172..591a655e 100644 --- a/src/EfficientDynamoDb/DynamoDbContext/DynamoDbContext.cs +++ b/src/EfficientDynamoDb/DynamoDbContext/DynamoDbContext.cs @@ -43,7 +43,7 @@ async Task IDynamoDbContext.ExecuteAsync(HttpContent httpC private async ValueTask ReadAsync(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)); diff --git a/src/EfficientDynamoDb/DynamoDbContext/DynamoDbLowLevelContext.cs b/src/EfficientDynamoDb/DynamoDbContext/DynamoDbLowLevelContext.cs index deebe4a6..fd12a204 100644 --- a/src/EfficientDynamoDb/DynamoDbContext/DynamoDbLowLevelContext.cs +++ b/src/EfficientDynamoDb/DynamoDbContext/DynamoDbLowLevelContext.cs @@ -203,7 +203,7 @@ private async ValueTask BuildHttpContentAsync(UpdateItemRequest req internal static async ValueTask 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 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; diff --git a/src/EfficientDynamoDb/Internal/ErrorHandler.cs b/src/EfficientDynamoDb/Internal/ErrorHandler.cs index 7a46e49d..f16bd3b1 100644 --- a/src/EfficientDynamoDb/Internal/ErrorHandler.cs +++ b/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 ProcessErrorAsync(DynamoDbContextMetadata metadata, HttpResponseMessage response, CancellationToken cancellationToken = default) { try @@ -30,37 +31,71 @@ public static async Task 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(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(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 magic = stackalloc byte[2]; + _ = stream.Read(magic); + stream.Position = 0; + return magic[0] == 0x1F && magic[1] == 0x8B; + } + + private static async Task 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 ProcessBadRequestAsync(DynamoDbContextMetadata metadata, MemoryStream recyclableStream, Error error, CancellationToken cancellationToken) + private static ValueTask 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 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 { diff --git a/src/EfficientDynamoDb/Internal/HttpApi.cs b/src/EfficientDynamoDb/Internal/HttpApi.cs index a722b7c4..6d6555e2 100644 --- a/src/EfficientDynamoDb/Internal/HttpApi.cs +++ b/src/EfficientDynamoDb/Internal/HttpApi.cs @@ -1,6 +1,7 @@ using System; using System.IO; using System.Net.Http; +using System.Net.Http.Headers; using System.Net.Sockets; using System.Text.Json; using System.Threading; @@ -49,6 +50,7 @@ public async ValueTask SendAsync(DynamoDbContextConfig conf { using var request = new HttpRequestMessage(HttpMethod.Post, config.RegionEndpoint.RequestUri); request.Content = httpContent; + request.Headers.AcceptEncoding.Add(new("gzip")); try { @@ -108,7 +110,7 @@ public async ValueTask SendAsync(DynamoDbContextConfig con { using var response = await SendAsync(config, httpContent, cancellationToken).ConfigureAwait(false); - await using var responseStream = await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false); + await using var responseStream = await response.GetDecodedStreamAsync().ConfigureAwait(false); return (await JsonSerializer.DeserializeAsync(responseStream, new JsonSerializerOptions {Converters = {new DdbEnumJsonConverterFactory(), new UnixDateTimeJsonConverter()}}, cancellationToken).ConfigureAwait(false))!; } diff --git a/src/EfficientDynamoDb/Internal/HttpResponseExtensions.cs b/src/EfficientDynamoDb/Internal/HttpResponseExtensions.cs new file mode 100644 index 00000000..1c9a879f --- /dev/null +++ b/src/EfficientDynamoDb/Internal/HttpResponseExtensions.cs @@ -0,0 +1,24 @@ +using System.IO; +using System.IO.Compression; +using System.Net.Http; +using System.Threading; +using System.Threading.Tasks; + +namespace EfficientDynamoDb.Internal +{ + internal static class HttpResponseExtensions + { + /// + /// Returns a decoding stream for a successful response. + /// Trusts the Content-Encoding header (DynamoDB is honest for 2xx responses). + /// The returned stream owns the underlying stream when gzip is used. + /// + internal static async Task GetDecodedStreamAsync(this HttpResponseMessage response, CancellationToken cancellationToken = default) + { + var rawStream = await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false); + return !response.Content.Headers.ContentEncoding.Contains("gzip") + ? rawStream + : new GZipStream(rawStream, CompressionMode.Decompress, leaveOpen: false); + } + } +} From e38ebdc5484985d41ebb938008b557e361f434fb Mon Sep 17 00:00:00 2001 From: Mykhailo Matviiv Date: Sat, 14 Mar 2026 17:00:37 +0100 Subject: [PATCH 2/3] Add integration tests for batch write --- .../DataPlane/TransactWrite/TestUser.cs | 25 ++++ .../TransactWrite/TransactWriteShould.cs | 140 ++++++++++++++++++ 2 files changed, 165 insertions(+) create mode 100644 src/EfficientDynamoDb.IntegrationTests/DataPlane/TransactWrite/TestUser.cs create mode 100644 src/EfficientDynamoDb.IntegrationTests/DataPlane/TransactWrite/TransactWriteShould.cs diff --git a/src/EfficientDynamoDb.IntegrationTests/DataPlane/TransactWrite/TestUser.cs b/src/EfficientDynamoDb.IntegrationTests/DataPlane/TransactWrite/TestUser.cs new file mode 100644 index 00000000..412d02a3 --- /dev/null +++ b/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; } = ""; +} diff --git a/src/EfficientDynamoDb.IntegrationTests/DataPlane/TransactWrite/TransactWriteShould.cs b/src/EfficientDynamoDb.IntegrationTests/DataPlane/TransactWrite/TransactWriteShould.cs new file mode 100644 index 00000000..82a8009b --- /dev/null +++ b/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 _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().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() + .WithPrimaryKey(item1.PartitionKey, item1.SortKey) + .WithConsistentRead(true) + .ToItemAsync(); + var retrieved2 = await _context.GetItem() + .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(() => + _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(() => + _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); + } +} From e24793e7e040ab47dc9daae47e6626f1bf8df30c Mon Sep 17 00:00:00 2001 From: Mykhailo Matviiv Date: Sun, 15 Mar 2026 22:53:48 +0100 Subject: [PATCH 3/3] Update docs regarding automatic decompression --- .../Configs/Http/IHttpClientFactory.cs | 8 ++++++++ src/EfficientDynamoDb/DynamoDbContextConfig.cs | 10 ++++++++++ website/docs/dev_guide/getting-started.md | 5 +++++ 3 files changed, 23 insertions(+) diff --git a/src/EfficientDynamoDb/Configs/Http/IHttpClientFactory.cs b/src/EfficientDynamoDb/Configs/Http/IHttpClientFactory.cs index 5b3252f5..e099349d 100644 --- a/src/EfficientDynamoDb/Configs/Http/IHttpClientFactory.cs +++ b/src/EfficientDynamoDb/Configs/Http/IHttpClientFactory.cs @@ -4,6 +4,14 @@ namespace EfficientDynamoDb.Configs.Http { public interface IHttpClientFactory { + /// + /// Creates or returns an used for DynamoDB requests. + /// + /// + /// The returned must have AutomaticDecompression set to . + /// 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. + /// HttpClient CreateHttpClient(); } } \ No newline at end of file diff --git a/src/EfficientDynamoDb/DynamoDbContextConfig.cs b/src/EfficientDynamoDb/DynamoDbContextConfig.cs index 940f7331..f661cd66 100644 --- a/src/EfficientDynamoDb/DynamoDbContextConfig.cs +++ b/src/EfficientDynamoDb/DynamoDbContextConfig.cs @@ -21,6 +21,16 @@ public class DynamoDbContextConfig public IAwsCredentialsProvider CredentialsProvider { get; } + /// + /// Factory used to create the for DynamoDB requests. + /// Defaults to an internal factory with a shared . + /// + /// + /// When providing a custom factory, ensure the returned has + /// AutomaticDecompression set to . + /// EfficientDynamoDb manages gzip decompression manually; enabling automatic decompression on the handler + /// will interfere with CRC verification and error response parsing, leading to incorrect behavior. + /// public IHttpClientFactory HttpClientFactory { get; set; } = DefaultHttpClientFactory.Instance; public IReadOnlyCollection Converters diff --git a/website/docs/dev_guide/getting-started.md b/website/docs/dev_guide/getting-started.md index aedb44b4..4ded2277 100644 --- a/website/docs/dev_guide/getting-started.md +++ b/website/docs/dev_guide/getting-started.md @@ -44,6 +44,11 @@ var config = new DynamoDbContextConfig(RegionEndpoint.USEast1, credentials) config.RetryStrategies.ThrottlingStrategy = DefaultRetryStrategy.Instance; ``` +:::caution +When providing a custom `HttpClientFactory`, make sure the `HttpClient` it returns has `AutomaticDecompression` set to `DecompressionMethods.None`. +EfficientDynamoDb handles gzip decompression manually, and enabling automatic decompression will interfere with CRC verification and error response parsing. +::: + For more info about retry strategies and possible options, check the [retry strategies guide](./configuration/retry-strategies.md). ## Region