DT.AzureStorage: add randomization to history blob names to avoid races

src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs

@@ -1052,8 +1052,7 @@ async Task AbandonAndReleaseSessionAsync(OrchestrationSession session)
             // will result in a duplicate replay of the orchestration with no side-effects.
             try
             {
-                session.ETag = await this.trackingStore.UpdateStateAsync(runtimeState, workItem.OrchestrationRuntimeState, instanceId, executionId, session.ETag);
-
+                session.ETag = await this.trackingStore.UpdateStateAsync(runtimeState, workItem.OrchestrationRuntimeState, instanceId, executionId, session.ETag, session.TrackingStoreContext);
                 // update the runtime state and execution id stored in the session
                 session.UpdateRuntimeState(runtimeState);
 

src/DurableTask.AzureStorage/MessageManager.cs

@@ -216,6 +216,12 @@ public Task<string> DownloadAndDecompressAsBytesAsync(Uri blobUri)
             return DownloadAndDecompressAsBytesAsync(blob);
         }
 
+        public Task<bool> DeleteBlobAsync(string blobName)
+        {
+            Blob blob = this.blobContainer.GetBlobReference(blobName);
+            return blob.DeleteIfExistsAsync(); 
+        }
+
         private async Task<string> DownloadAndDecompressAsBytesAsync(Blob blob)
         {
             using (MemoryStream memory = new MemoryStream(MaxStorageQueuePayloadSizeInBytes * 2))

src/DurableTask.AzureStorage/Messaging/OrchestrationSession.cs

@@ -38,6 +38,7 @@ sealed class OrchestrationSession : SessionBase, IOrchestrationSession
             OrchestrationRuntimeState runtimeState,
             string eTag,
             DateTime lastCheckpointTime,
+            object trackingStoreContext,
             TimeSpan idleTimeout,
             Guid traceActivityId)
             : base(settings, storageAccountName, orchestrationInstance, traceActivityId)
@@ -48,6 +49,7 @@ sealed class OrchestrationSession : SessionBase, IOrchestrationSession
             this.RuntimeState = runtimeState ?? throw new ArgumentNullException(nameof(runtimeState));
             this.ETag = eTag;
             this.LastCheckpointTime = lastCheckpointTime;
+            this.TrackingStoreContext = trackingStoreContext;
 
             this.messagesAvailableEvent = new AsyncAutoResetEvent(signaled: false);
             this.nextMessageBatch = new MessageCollection();
@@ -67,6 +69,8 @@ sealed class OrchestrationSession : SessionBase, IOrchestrationSession
 
         public DateTime LastCheckpointTime { get; }
 
+        public object TrackingStoreContext { get; }
+
         public IReadOnlyList<MessageData> PendingMessages => this.nextMessageBatch;
 
         public override int GetCurrentEpisode()

src/DurableTask.AzureStorage/OrchestrationSessionManager.cs

@@ -478,6 +478,7 @@ bool IsScheduledAfterInstanceUpdate(MessageData msg, OrchestrationState? remoteI
                     batch.OrchestrationState = new OrchestrationRuntimeState(history.Events);
                     batch.ETag = history.ETag;
                     batch.LastCheckpointTime = history.LastCheckpointTime;
+                    batch.TrackingStoreContext = history.TrackingStoreContext;
                 }
 
                 this.readyForProcessingQueue.Enqueue(node);
@@ -538,6 +539,7 @@ bool IsScheduledAfterInstanceUpdate(MessageData msg, OrchestrationState? remoteI
                             nextBatch.OrchestrationState,
                             nextBatch.ETag,
                             nextBatch.LastCheckpointTime,
+                            nextBatch.TrackingStoreContext,
                             this.settings.ExtendedSessionIdleTimeout,
                             traceActivityId);
 
@@ -683,6 +685,7 @@ public PendingMessageBatch(ControlQueue controlQueue, string instanceId, string?
 
             public string? ETag { get; set; }
             public DateTime LastCheckpointTime { get; set; }
+            public object? TrackingStoreContext { get; set; }
         }
     }
 }

src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs

@@ -14,6 +14,7 @@
 namespace DurableTask.AzureStorage.Tracking
 {
     using System;
+    using System.Collections.Concurrent;
     using System.Collections.Generic;
     using System.Diagnostics;
     using System.Linq;
@@ -158,6 +159,7 @@ public override async Task<OrchestrationHistory> GetHistoryEventsAsync(string in
             IList<HistoryEvent> historyEvents;
             string executionId;
             DynamicTableEntity sentinel = null;
+            TrackingStoreContext trackingStoreContext = new TrackingStoreContext();
             if (tableEntities.Count > 0)
             {
                 // The most recent generation will always be in the first history event.
@@ -183,7 +185,7 @@ public override async Task<OrchestrationHistory> GetHistoryEventsAsync(string in
                     }
 
                     // Some entity properties may be stored in blob storage.
-                    await this.DecompressLargeEntityProperties(entity);
+                    await this.DecompressLargeEntityProperties(entity, trackingStoreContext.Blobs);
 
                     events.Add((HistoryEvent)this.tableEntityConverter.ConvertFromTableEntity(entity, GetTypeForTableEntity));
                 }
@@ -222,7 +224,7 @@ public override async Task<OrchestrationHistory> GetHistoryEventsAsync(string in
                 eTagValue,
                 checkpointCompletionTime);
 
-            return new OrchestrationHistory(historyEvents, checkpointCompletionTime, eTagValue);
+            return new OrchestrationHistory(historyEvents, checkpointCompletionTime, eTagValue, trackingStoreContext);
         }
 
         async Task<TableEntitiesResponseInfo<DynamicTableEntity>> GetHistoryEntitiesResponseInfoAsync(string instanceId, string expectedExecutionId, IList<string> projectionColumns,  CancellationToken cancellationToken = default(CancellationToken))
@@ -843,7 +845,7 @@ public override async Task<PurgeHistoryResult> PurgeInstanceHistoryAsync(string
 
             // It is possible that the queue message was small enough to be written directly to a queue message,
             // not a blob, but is too large to be written to a table property.
-            await this.CompressLargeMessageAsync(entity);
+            await this.CompressLargeMessageAsync(entity, listOfBlobs: null);
 
             Stopwatch stopwatch = Stopwatch.StartNew();
             try
@@ -928,11 +930,13 @@ public override Task StartAsync()
             OrchestrationRuntimeState oldRuntimeState,
             string instanceId,
             string executionId,
-            string eTagValue)
+            string eTagValue,
+            object trackingStoreContext)
         {
             int estimatedBytes = 0;
             IList<HistoryEvent> newEvents = newRuntimeState.NewEvents;
             IList<HistoryEvent> allEvents = newRuntimeState.Events;
+            TrackingStoreContext context = (TrackingStoreContext) trackingStoreContext;
 
             int episodeNumber = Utils.GetEpisodeNumber(newRuntimeState);
 
@@ -953,7 +957,15 @@ public override Task StartAsync()
                     ["LastUpdatedTime"] = new EntityProperty(newEvents.Last().Timestamp),
                 }
             };
-
+
+            // check if we are replacing a previous execution with blobs; those will be deleted from the store after the update. This could occur in a ContinueAsNew scenario
+            List<string> blobsToDelete = null;
+            if (oldRuntimeState != newRuntimeState && context.Blobs.Count > 0)
+            {
+                blobsToDelete = context.Blobs;
+                context.Blobs = new List<string>();
+            }
+
             for (int i = 0; i < newEvents.Count; i++)
             {
                 bool isFinalEvent = i == newEvents.Count - 1;
@@ -969,7 +981,7 @@ public override Task StartAsync()
                 historyEntity.RowKey = sequenceNumber.ToString("X16");
                 historyEntity.Properties["ExecutionId"] = new EntityProperty(executionId);
 
-                await this.CompressLargeMessageAsync(historyEntity);
+                await this.CompressLargeMessageAsync(historyEntity, context.Blobs);
 
                 // Replacement can happen if the orchestration episode gets replayed due to a commit failure in one of the steps below.
                 historyEventBatch.InsertOrReplace(historyEntity);
@@ -1108,6 +1120,18 @@ public override Task StartAsync()
                 episodeNumber,
                 orchestrationInstanceUpdateStopwatch.ElapsedMilliseconds);
 
+            // finally, delete orphaned blobs from the previous execution history.
+            // We had to wait until the new history has committed to make sure the blobs are no longer necessary.
+            if (blobsToDelete != null)
+            {
+                var tasks = new List<Task>(blobsToDelete.Count);
+                foreach (var blobName in blobsToDelete)
+                {
+                    tasks.Add(this.messageManager.DeleteBlobAsync(blobName));
+                }
+                await Task.WhenAll(tasks);
+            }
+
             return eTagValue;
         }
 
@@ -1177,7 +1201,7 @@ Type GetTypeForTableEntity(DynamicTableEntity tableEntity)
             }
         }
 
-        async Task CompressLargeMessageAsync(DynamicTableEntity entity)
+        async Task CompressLargeMessageAsync(DynamicTableEntity entity, List<string> listOfBlobs)
         {
             foreach (string propertyName in VariableSizeEntityProperties)
             {
@@ -1194,11 +1218,14 @@ async Task CompressLargeMessageAsync(DynamicTableEntity entity)
                     string blobPropertyName = GetBlobPropertyName(propertyName);
                     entity.Properties.Add(blobPropertyName, new EntityProperty(blobName));
                     entity.Properties[propertyName].StringValue = string.Empty;
+
+                    // if necessary, keep track of all the blobs associated with this execution
+                    listOfBlobs?.Add(blobName);
                 }
             }
         }
 
-        async Task DecompressLargeEntityProperties(DynamicTableEntity entity)
+        async Task DecompressLargeEntityProperties(DynamicTableEntity entity, List<string> listOfBlobs)
         {
             // Check for entity properties stored in blob storage
             foreach (string propertyName in VariableSizeEntityProperties)
@@ -1210,6 +1237,9 @@ async Task DecompressLargeEntityProperties(DynamicTableEntity entity)
                     string decompressedMessage = await this.messageManager.DownloadAndDecompressAsBytesAsync(blobName);
                     entity.Properties[propertyName] = new EntityProperty(decompressedMessage);
                     entity.Properties.Remove(blobPropertyName);
+
+                    // keep track of all the blobs associated with this execution
+                    listOfBlobs.Add(blobName);
                 }
             }
         }
@@ -1241,7 +1271,10 @@ static string GetBlobName(DynamicTableEntity entity, string property)
                 throw new InvalidOperationException($"Could not compute the blob name for property {property}");
             }
 
-            string blobName = $"{sanitizedInstanceId}/history-{sequenceNumber}-{eventType}-{property}.json.gz";
+            // randomize the blob name to prevent accidental races in split-brain situations (#890)
+            uint random = (uint)(new Random()).Next();
+
+            string blobName = $"{sanitizedInstanceId}/history-{sequenceNumber}-{eventType}-{random:X8}-{property}.json.gz";
 
             return blobName;
         }
@@ -1346,5 +1379,10 @@ bool ExceedsMaxTablePropertySize(string data)
 
             return false;
         }
+
+        class TrackingStoreContext
+        {
+            public List<string> Blobs { get; set; } = new List<string>();
+        }
     }
 }

src/DurableTask.AzureStorage/Tracking/ITrackingStore.cs

@@ -69,7 +69,8 @@ interface ITrackingStore
         /// <param name="instanceId">InstanceId for the Orchestration Update</param>
         /// <param name="executionId">ExecutionId for the Orchestration Update</param>
         /// <param name="eTag">The ETag value to use for safe updates</param>
-        Task<string> UpdateStateAsync(OrchestrationRuntimeState newRuntimeState, OrchestrationRuntimeState oldRuntimeState, string instanceId, string executionId, string eTag);
+        /// <param name="trackingStoreContext">Additional context for the execution that is maintained by the tracking store.</param>
+        Task<string> UpdateStateAsync(OrchestrationRuntimeState newRuntimeState, OrchestrationRuntimeState oldRuntimeState, string instanceId, string executionId, string eTag, object trackingStoreContext);
 
         /// <summary>
         /// Get The Orchestration State for the Latest or All Executions

src/DurableTask.AzureStorage/Tracking/InstanceStoreBackedTrackingStore.cs

@@ -137,7 +137,7 @@ public override Task StartAsync()
         }
 
         /// <inheritdoc />
-        public override async Task<string> UpdateStateAsync(OrchestrationRuntimeState newRuntimeState, OrchestrationRuntimeState oldRuntimeState, string instanceId, string executionId, string eTag)
+        public override async Task<string> UpdateStateAsync(OrchestrationRuntimeState newRuntimeState, OrchestrationRuntimeState oldRuntimeState, string instanceId, string executionId, string eTag, object executionData)
         {
             //In case there is a runtime state for an older execution/iteration as well that needs to be committed, commit it.
             //This may be the case if a ContinueAsNew was executed on the orchestration

src/DurableTask.AzureStorage/Tracking/OrchestrationHistory.cs

@@ -30,16 +30,24 @@ public OrchestrationHistory(IList<HistoryEvent> historyEvents, DateTime lastChec
         }
 
         public OrchestrationHistory(IList<HistoryEvent> historyEvents, DateTime lastCheckpointTime, string eTag)
+            : this(historyEvents, lastCheckpointTime, eTag, null)
+        {
+        }
+
+        public OrchestrationHistory(IList<HistoryEvent> historyEvents, DateTime lastCheckpointTime, string eTag, object trackingStoreContext)
         {
             this.Events = historyEvents ?? throw new ArgumentNullException(nameof(historyEvents));
             this.LastCheckpointTime = lastCheckpointTime;
             this.ETag = eTag;
+            this.TrackingStoreContext = trackingStoreContext;
         }
 
         public IList<HistoryEvent> Events { get; }
 
         public string ETag { get; }
 
         public DateTime LastCheckpointTime { get; }
+
+        public object TrackingStoreContext { get; }
     }
 }

src/DurableTask.AzureStorage/Tracking/TrackingStoreBase.cs

@@ -111,6 +111,6 @@ public virtual Task UpdateStatusForRewindAsync(string instanceId)
         public abstract Task StartAsync();
 
         /// <inheritdoc />
-        public abstract Task<string> UpdateStateAsync(OrchestrationRuntimeState newRuntimeState, OrchestrationRuntimeState oldRuntimeState, string instanceId, string executionId, string eTag);
+        public abstract Task<string> UpdateStateAsync(OrchestrationRuntimeState newRuntimeState, OrchestrationRuntimeState oldRuntimeState, string instanceId, string executionId, string eTag, object executionData);
     }
 }