diff --git a/AutoNumber/BlobOptimisticDataStore.cs b/AutoNumber/BlobOptimisticDataStore.cs index f0e5f1e..25f0b57 100644 --- a/AutoNumber/BlobOptimisticDataStore.cs +++ b/AutoNumber/BlobOptimisticDataStore.cs @@ -14,6 +14,19 @@ namespace AutoNumber { + /// + /// Optimistic-concurrency counter store backed by Azure block blobs. + /// + /// + /// The read-time ETag captured by / is stored + /// per block name and consumed by the next / + /// for that block, turning the read->write cycle into a + /// compare-and-swap. Callers must therefore keep the read->write cycle for a given block + /// serialized per instance; UniqueIdGenerator does this via its per-scope lock. + /// Interleaving concurrent read+write for the same block on one instance can drop a read-time + /// ETag, causing a silent fallback to the blob's current ETag (fetched at write time), which can + /// hand out duplicate id ranges. + /// public class BlobOptimisticDataStore : IOptimisticDataStore { private const string SeedValue = "1"; @@ -21,6 +34,12 @@ public class BlobOptimisticDataStore : IOptimisticDataStore private readonly ConcurrentDictionary blobReferences; private readonly object blobReferencesLock = new object(); + // ETag observed by the most recent GetData per block, so TryOptimisticWrite can condition + // on the state that was actually read. Conditioning on the blob's current ETag (fetched at + // write time) does not detect writers that committed between our read and our write, which + // allows two processes to hand out the same id range. + private readonly ConcurrentDictionary readETags = new ConcurrentDictionary(); + public BlobOptimisticDataStore(BlobServiceClient blobServiceClient, string containerName) { blobContainer = blobServiceClient.GetBlobContainerClient(containerName.ToLower()); @@ -36,22 +55,20 @@ public string GetData(string blockName) { var blobReference = GetBlobReference(blockName); - using (var stream = new MemoryStream()) - { - blobReference.DownloadTo(stream); - return Encoding.UTF8.GetString(stream.ToArray()); - } + // DownloadContent returns the content and its ETag from a single response, so the + // value/ETag pair is consistent + var download = blobReference.DownloadContent(); + readETags[blockName] = download.Value.Details.ETag; + return download.Value.Content.ToString(); } public async Task GetDataAsync(string blockName) { var blobReference = GetBlobReference(blockName); - using (var stream = new MemoryStream()) - { - await blobReference.DownloadToAsync(stream).ConfigureAwait(false); - return Encoding.UTF8.GetString(stream.ToArray()); - } + var download = await blobReference.DownloadContentAsync().ConfigureAwait(false); + readETags[blockName] = download.Value.Details.ETag; + return download.Value.Content.ToString(); } public async Task InitAsync() @@ -69,11 +86,21 @@ public bool Init() public bool TryOptimisticWrite(string blockName, string data) { var blobReference = GetBlobReference(blockName); + + // Condition the write on the ETag captured by the preceding GetData, so any writer that + // committed after our read fails this write with 412 and the caller re-reads and retries. + // Each read ETag is consumed at most once; callers without a preceding read fall back to + // the blob's current ETag (previous behaviour). + if (!readETags.TryRemove(blockName, out var readETag)) + { + readETag = blobReference.GetProperties().Value.ETag; + } + try { var blobRequestCondition = new BlobRequestConditions { - IfMatch = (blobReference.GetProperties()).Value.ETag + IfMatch = readETag }; UploadText( blobReference, @@ -94,11 +121,17 @@ public bool TryOptimisticWrite(string blockName, string data) public async Task TryOptimisticWriteAsync(string blockName, string data) { var blobReference = GetBlobReference(blockName); + + if (!readETags.TryRemove(blockName, out var readETag)) + { + readETag = (await blobReference.GetPropertiesAsync().ConfigureAwait(false)).Value.ETag; + } + try { var blobRequestCondition = new BlobRequestConditions { - IfMatch = (await blobReference.GetPropertiesAsync()).Value.ETag + IfMatch = readETag }; await UploadTextAsync( blobReference, diff --git a/IntegrationTests/Azure.cs b/IntegrationTests/Azure.cs index f3755ab..f6fc3b0 100644 --- a/IntegrationTests/Azure.cs +++ b/IntegrationTests/Azure.cs @@ -159,6 +159,38 @@ public void ShouldReturnIdsAcrossMultipleGenerators() Assert.Equal(expected, generatedIds); } + [Fact] + public void ShouldNotGenerateDuplicateIdsAcrossGeneratorsUnderContention() + { + // Regression test: TryOptimisticWrite used to condition on the blob's CURRENT ETag + // (fetched at write time) instead of the ETag observed by the preceding GetData, so a + // competing generator that committed between our read and our write went undetected and + // both handed out the same ids. Independent stores/generators simulate separate + // processes; BatchSize = 1 maximises contention on the counter blob. + using var testScope = BuildTestScope(); + const int generatorCount = 4; + const int idsPerGenerator = 100; + + var generators = Enumerable.Range(0, generatorCount) + .Select(_ => new UniqueIdGenerator(BuildStore(testScope)) + { + BatchSize = 1, + MaxWriteAttempts = 100 + }) + .ToArray(); + + var generatedIds = new ConcurrentQueue(); + var scopeName = testScope.IdScopeName; + Parallel.ForEach(generators, generator => + { + for (var i = 0; i < idsPerGenerator; i++) + generatedIds.Enqueue(generator.NextId(scopeName)); + }); + + Assert.Equal(generatorCount * idsPerGenerator, generatedIds.Count); + Assert.Equal(generatedIds.Count, generatedIds.Distinct().Count()); + } + [Fact] public void ShouldSupportUsingOneGeneratorFromMultipleThreads() {