Revise how entity batches are executed and handle failures (#954)

* revise task entity definition

* commit change that had been accidentally committed to a different PR.

* Apply suggestions from code review

Co-authored-by: David Justo <david.justo.1996@gmail.com>

* Apply suggestions from code review

Co-authored-by: Jacob Viau <javia@microsoft.com>

---------

Co-authored-by: David Justo <david.justo.1996@gmail.com>
Co-authored-by: Jacob Viau <javia@microsoft.com>

src/DurableTask.Core/Entities/OperationFormat/EntityBatchResult.cs

@@ -24,10 +24,9 @@ public class EntityBatchResult
         //       To ensure maximum compatibility, all properties should be public and settable by default.
 
         /// <summary>
-        /// The results of executing the operations in the batch. The length of this list must match
-        /// the size of the batch if all messages were processed; In particular, all execution errors must be reported as a result.
-        /// However, this list of results can be shorter than the list of operations if
-        /// some suffix of the operation list was skipped, e.g. due to shutdown, send throttling, or timeouts.  
+        /// The results of executing the operations in the batch. If there were (non-application-level) errors, the length of this list may
+        /// be shorter than the number of requests. In that case, <see cref="FailureDetails"/> contains the reason why not all requests 
+        /// were processed.
         /// </summary>
         public List<OperationResult>? Results { get; set; }
 
@@ -41,5 +40,10 @@ public class EntityBatchResult
         /// or null if the entity has no state (e.g. if it has been deleted).
         /// </summary>
         public string? EntityState { get; set; }
+
+        /// <summary>
+        /// Contains the failure details, if there was a failure to process all requests (fewer results were returned than requests) 
+        /// </summary>
+        public FailureDetails? FailureDetails { get; set; }
     }
 }

src/DurableTask.Core/Entities/TaskEntity.cs

@@ -25,6 +25,6 @@ public abstract class TaskEntity
         /// <summary>
         /// Execute a batch of operations on an entity.
         /// </summary>
-        public abstract Task<EntityBatchResult> ExecuteOperationBatchAsync(EntityBatchRequest operations, EntityExecutionOptions options);
+        public abstract Task<EntityBatchResult> ExecuteOperationBatchAsync(EntityBatchRequest operations);
     }
 }

src/DurableTask.Core/TaskEntityDispatcher.cs

@@ -246,6 +246,24 @@ protected async Task<bool> OnProcessWorkItemAsync(TaskOrchestrationWorkItem work
                     {
                         // execute the user-defined operations on this entity, via the middleware
                         var result = await this.ExecuteViaMiddlewareAsync(workToDoNow, runtimeState.OrchestrationInstance, schedulerState.EntityState);
+                        var operationResults = result.Results!;
+
+                        // if we encountered an error, record it as the result of the operations
+                        // so that callers are notified that the operation did not succeed.
+                        if (result.FailureDetails != default(FailureDetails))
+                        {
+                            OperationResult errorResult = new OperationResult()
+                            {
+                                Result = null,
+                                ErrorMessage = "entity dispatch failed",
+                                FailureDetails = result.FailureDetails,
+                            };
+
+                            for (int i = operationResults.Count; i < workToDoNow.OperationCount; i++)
+                            {
+                                operationResults.Add(errorResult);
+                            }
+                        }
 
                         // go through all results
                         // for each operation that is not a signal, send a result message back to the calling orchestrator
@@ -260,7 +278,8 @@ protected async Task<bool> OnProcessWorkItemAsync(TaskOrchestrationWorkItem work
 
                         if (result.Results.Count < workToDoNow.OperationCount)
                         {
-                            // some operations were not processed
+                            // some requests were not processed (e.g. due to shutdown or timeout)
+                            // in this case we just defer the work so it can be retried
                             var deferred = workToDoNow.RemoveDeferredWork(result.Results.Count);
                             schedulerState.PutBack(deferred);
                             workToDoNow.ToBeContinued(schedulerState);
@@ -851,7 +870,7 @@ await this.dispatchPipeline.RunAsync(dispatchContext, async _ =>
                     ErrorPropagationMode = this.errorPropagationMode,
                 };
 
-                var result = await taskEntity.ExecuteOperationBatchAsync(request, options);
+                var result = await taskEntity.ExecuteOperationBatchAsync(request);
 
                 dispatchContext.SetProperty(result);
             });