Add simple poison message handling for Azure Storage

src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs

@@ -117,6 +117,11 @@ public class AzureStorageOrchestrationServiceSettings
         /// </summary>
         public int MaxConcurrentTaskEntityWorkItems { get; set; } = 100;
 
+        /// <summary>
+        /// Gets or sets the maximum dequeue count of any message before it is flagged as a "poison message".
+        /// The default value is 20.
+        /// </summary>
+        public int PoisonMessageDeuqueCountThreshold { get; set; } = 20;
         /// <summary>
         /// Gets or sets the maximum number of concurrent storage operations that can be executed in the context
         /// of a single orchestration instance.

src/DurableTask.AzureStorage/DataContractJsonConverter.cs

@@ -29,6 +29,8 @@ namespace DurableTask.AzureStorage
     /// </summary>
     internal class DataContractJsonConverter : JsonConverter
     {
+        public JsonSerializer alternativeSerializer = null;
+
         public override bool CanConvert(Type objectType)
         {
             if (objectType == null)
@@ -59,50 +61,33 @@ public override object ReadJson(
                 throw new ArgumentNullException(nameof(serializer));
             }
 
-            using (var stream = new MemoryStream())
-            using (var writer = new StreamWriter(stream))
-            using (var jsonWriter = new JsonTextWriter(writer))
+            // JsonReader is forward only, need to make a copy so we can read it twice.
+            using var stream = new MemoryStream();
+            using var writer = new StreamWriter(stream);
+            using var jsonWriter = new JsonTextWriter(writer);
+            jsonWriter.WriteToken(reader, writeChildren: true);
+            jsonWriter.Flush();
+            stream.Position = 0;
+
+            try
+            {
+                using var reader2 = new JsonTextReader(new StreamReader(stream));
+                reader2.CloseInput = false;
+                return this.alternativeSerializer.Deserialize(reader2, objectType);
+            }
+            catch
             {
-                jsonWriter.WriteToken(reader, writeChildren: true);
-                jsonWriter.Flush();
                 stream.Position = 0;
-
-                var contractSerializer = CreateSerializer(objectType, serializer);
+                DataContractJsonSerializer contractSerializer = CreateSerializer(objectType, serializer);
                 return contractSerializer.ReadObject(stream);
             }
         }
 
         /// <inheritdoc />
         public override void WriteJson(JsonWriter writer, object value, JsonSerializer serializer)
         {
-            if (writer == null)
-            {
-                throw new ArgumentNullException(nameof(writer));
-            }
-
-            if (value == null)
-            {
-                writer.WriteNull();
-                return;
-            }
-
-            if (serializer == null)
-            {
-                throw new ArgumentNullException(nameof(serializer));
-            }
-
-            using (var memoryStream = new MemoryStream())
-            {
-                var contractSerializer = CreateSerializer(value.GetType(), serializer);
-                contractSerializer.WriteObject(memoryStream, value);
-                memoryStream.Position = 0;
-
-                using (var streamReader = new StreamReader(memoryStream))
-                using (var jsonReader = new JsonTextReader(streamReader))
-                {
-                    writer.WriteToken(jsonReader, writeChildren: true);
-                }
-            }
+            // Ignore data contract, use Newtonsoft
+            this.alternativeSerializer.Serialize(writer, value);
         }
 
         private static DataContractJsonSerializer CreateSerializer(Type type, JsonSerializer serializer)
@@ -115,4 +100,4 @@ private static DataContractJsonSerializer CreateSerializer(Type type, JsonSerial
                 });
         }
     }
-}
+}

src/DurableTask.AzureStorage/DurableTask.AzureStorage.csproj

@@ -21,8 +21,9 @@
   <PropertyGroup>
     <MajorVersion>1</MajorVersion>
     <MinorVersion>17</MinorVersion>
-    <PatchVersion>3</PatchVersion>
+    <PatchVersion>4</PatchVersion>
     <VersionPrefix>$(MajorVersion).$(MinorVersion).$(PatchVersion)</VersionPrefix>
+    <VersionSuffix>poisonmessagehandler.4</VersionSuffix>
     <FileVersion>$(VersionPrefix).0</FileVersion>
     <!-- FileVersionRevision is expected to be set by the CI. This is useful for distinguishing between multiple builds of the same version. -->
     <FileVersion Condition="'$(FileVersionRevision)' != ''">$(VersionPrefix).$(FileVersionRevision)</FileVersion>

src/DurableTask.AzureStorage/EntityTrackingStoreQueries.cs

@@ -235,9 +235,10 @@ bool OrchestrationIsRunning(OrchestrationStatus? status)
             {
                 // first, retrieve the entity scheduler state (= input of the orchestration state), possibly from blob storage.
                 string serializedSchedulerState;
-                if (MessageManager.TryGetLargeMessageReference(state.Input, out Uri blobUrl))
+                if (MessageManager.TryGetLargeMessageReference(state.Input, out Uri? blobUrl))
                 {
-                    serializedSchedulerState = await this.messageManager.DownloadAndDecompressAsBytesAsync(blobUrl);
+                    // we know blobUrl is not null because TryGetLargeMessageReference returned true
+                    serializedSchedulerState = await this.messageManager.DownloadAndDecompressAsBytesAsync(blobUrl!);
                 }
                 else
                 {

src/DurableTask.AzureStorage/MessageManager.cs

@@ -10,7 +10,7 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 //  ----------------------------------------------------------------------------------
-
+#nullable enable
 namespace DurableTask.AzureStorage
 {
     using System;
@@ -50,7 +50,7 @@ public MessageManager(
         {
             this.settings = settings;
             this.azureStorageClient = azureStorageClient;
-            this.blobContainer = this.azureStorageClient.GetBlobContainerReference(blobContainerName);
+            this.blobContainer = this.azureStorageClient?.GetBlobContainerReference(blobContainerName);
             this.taskMessageSerializerSettings = new JsonSerializerSettings
             {
                 TypeNameHandling = TypeNameHandling.Objects,
@@ -61,9 +61,13 @@ public MessageManager(
 #endif
             };
 
-            if (this.settings.UseDataContractSerialization)
+            JsonSerializer newtonSoftSerializer = JsonSerializer.Create(taskMessageSerializerSettings);
+
+            if (this.settings.UseDataContractSerialization) // for hotfix to work, set setting to `true`
             {
-                this.taskMessageSerializerSettings.Converters.Add(new DataContractJsonConverter());
+                var dataConverter = new DataContractJsonConverter();
+                dataConverter.alternativeSerializer = newtonSoftSerializer;
+                this.taskMessageSerializerSettings.Converters.Add(dataConverter);
             }
 
             // We _need_ to create the Json serializer after providing the data converter,
@@ -122,17 +126,18 @@ public async Task<string> SerializeMessageDataAsync(MessageData messageData)
         /// <returns>Actual string representation of message.</returns>
         public async Task<string> FetchLargeMessageIfNecessary(string message)
         {
-            if (TryGetLargeMessageReference(message, out Uri blobUrl))
+            if (TryGetLargeMessageReference(message, out Uri? blobUrl))
             {
-                return await this.DownloadAndDecompressAsBytesAsync(blobUrl);
+                // we know blobUrl is not null because TryGetLargeMessageReference returned true
+                return await this.DownloadAndDecompressAsBytesAsync(blobUrl!);
             }
             else
             {
                 return message;
             }
         }
 
-        internal static bool TryGetLargeMessageReference(string messagePayload, out Uri blobUrl)
+        internal static bool TryGetLargeMessageReference(string messagePayload, out Uri? blobUrl)
         {
             if (Uri.IsWellFormedUriString(messagePayload, UriKind.Absolute))
             {
@@ -318,6 +323,7 @@ public async Task<int> DeleteLargeMessageBlobs(string sanitizedInstanceId)
             return storageOperationCount;
         }
     }
+#nullable disable
 
 #if NETSTANDARD2_0
     class TypeNameSerializationBinder : ISerializationBinder