fix: condition counter writes on the read-time ETag to prevent duplicate ids under multi-instance concurrency

AutoNumber/BlobOptimisticDataStore.cs

@@ -14,13 +14,32 @@
 
 namespace AutoNumber
 {
+    /// <summary>
+    /// Optimistic-concurrency counter store backed by Azure block blobs.
+    /// </summary>
+    /// <remarks>
+    /// The read-time ETag captured by <see cref="GetData"/>/<see cref="GetDataAsync"/> is stored
+    /// per block name and consumed by the next <see cref="TryOptimisticWrite"/>/
+    /// <see cref="TryOptimisticWriteAsync"/> for that block, turning the read-&gt;write cycle into a
+    /// compare-and-swap. Callers must therefore keep the read-&gt;write cycle for a given block
+    /// serialized per instance; <c>UniqueIdGenerator</c> 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.
+    /// </remarks>
     public class BlobOptimisticDataStore : IOptimisticDataStore
     {
         private const string SeedValue = "1";
         private readonly BlobContainerClient blobContainer;
         private readonly ConcurrentDictionary<string, BlockBlobClient> 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<string, ETag> readETags = new ConcurrentDictionary<string, ETag>();
+
         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<string> 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<bool> 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<bool> 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,

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<long>();
+            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()
         {