fix(Runner): Fixed TaskExecutor implementations to not dispose of child executors

fix(Runner): Fixed the TaskExecutor to catch uncaught initialization and execution exceptions, thus enabling faulting the processed task

Signed-off-by: Charles d'Avernas <charles.davernas@neuroglia.io>

Author: cdavernas

src/core/Synapse.Core/SynapseDefaults.cs

@@ -11,8 +11,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-using Synapse.Resources;
 using Neuroglia.Data.Infrastructure.ResourceOriented;
+using Synapse.Resources;
 using System.Diagnostics;
 using System.Reflection;
 

src/runner/Synapse.Runner/AuthorizationInfo.cs

@@ -11,7 +11,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-using Docker.DotNet.Models;
 using Neuroglia.Data.Infrastructure.ResourceOriented;
 using ServerlessWorkflow.Sdk.Models.Authentication;
 using System.Text;

src/runner/Synapse.Runner/Services/Executors/ForTaskExecutor.cs

@@ -105,7 +105,6 @@ protected virtual async Task OnIterationFaultAsync(ITaskExecutor executor, Cance
         ArgumentNullException.ThrowIfNull(executor);
         var error = executor.Task.Instance.Error ?? throw new NullReferenceException();
         this.Executors.Remove(executor);
-        await executor.DisposeAsync().ConfigureAwait(false);
         await this.SetErrorAsync(error, cancellationToken).ConfigureAwait(false);
     }
 
@@ -123,7 +122,6 @@ protected virtual async Task OnIterationCompletedAsync(ITaskExecutor executor, C
         var output = executor.Task.Output!;
         this.Executors.Remove(executor);
         if (this.Task.ContextData != executor.Task.ContextData) await this.Task.SetContextDataAsync(executor.Task.ContextData, cancellationToken).ConfigureAwait(false);
-        await executor.DisposeAsync().ConfigureAwait(false);
         var index = int.Parse(last.Reference.OriginalString.Split('/', StringSplitOptions.RemoveEmptyEntries)[^2]) + 1;
         if (index == this.Collection.Count)
         {

src/runner/Synapse.Runner/Services/Executors/ForkTaskExecutor.cs

@@ -80,11 +80,9 @@ protected virtual async Task OnSubTaskFaultAsync(ITaskExecutor executor, Cancell
         using var @lock = await this.Lock.LockAsync(cancellationToken).ConfigureAwait(false);
         var error = executor.Task.Instance.Error ?? throw new NullReferenceException();
         this.Executors.Remove(executor);
-        await executor.DisposeAsync().ConfigureAwait(false);
         foreach (var subExecutor in this.Executors)
         {
             await subExecutor.CancelAsync(cancellationToken).ConfigureAwait(false);
-            await subExecutor.DisposeAsync().ConfigureAwait(false);
             this.Executors.Remove(executor);
         }
         await this.SetErrorAsync(error, cancellationToken).ConfigureAwait(false);
@@ -102,11 +100,7 @@ protected virtual async Task OnSubTaskCompletedAsync(ITaskExecutor executor, Can
         using var @lock = await this.Lock.LockAsync(cancellationToken).ConfigureAwait(false);
         if (this.Task.Instance.Status != TaskInstanceStatus.Running)
         {
-            if (this.Executors.Remove(executor))
-            {
-                await executor.CancelAsync(cancellationToken).ConfigureAwait(false);
-                await executor.DisposeAsync().ConfigureAwait(false);
-            }
+            if (this.Executors.Remove(executor)) await executor.CancelAsync(cancellationToken).ConfigureAwait(false);
         }
         if (this.Task.Definition.Fork.Compete == true)
         {
@@ -115,7 +109,6 @@ protected virtual async Task OnSubTaskCompletedAsync(ITaskExecutor executor, Can
             {
                 this.Executors.Remove(concurrentTaskExecutor);
                 await concurrentTaskExecutor.CancelAsync(cancellationToken).ConfigureAwait(false);
-                await concurrentTaskExecutor.DisposeAsync().ConfigureAwait(false);
             }
             await this.SetResultAsync(output, this.Task.Definition.Then, cancellationToken).ConfigureAwait(false);
         }

src/runner/Synapse.Runner/Services/Executors/FunctionCallExecutor.cs

@@ -115,7 +115,6 @@ protected virtual async Task OnSubTaskFaultAsync(ITaskExecutor executor, Cancell
         ArgumentNullException.ThrowIfNull(executor);
         var error = executor.Task.Instance.Error ?? throw new NullReferenceException();
         this.Executors.Remove(executor);
-        await executor.DisposeAsync().ConfigureAwait(false);
         await this.SetErrorAsync(error, cancellationToken).ConfigureAwait(false);
     }
 

src/runner/Synapse.Runner/Services/TaskExecutor.cs

@@ -115,9 +115,34 @@ public abstract class TaskExecutor<TDefinition>(IServiceProvider serviceProvider
     /// <inheritdoc/>
     public virtual async Task InitializeAsync(CancellationToken cancellationToken = default)
     {
-        await this.DoInitializeAsync(cancellationToken).ConfigureAwait(false);
-        await this.Task.InitializeAsync(cancellationToken).ConfigureAwait(false);
-        this.Subject.OnNext(new TaskLifeCycleEvent(TaskLifeCycleEventType.Initialized));
+        try
+        {
+            await this.DoInitializeAsync(cancellationToken).ConfigureAwait(false);
+            await this.Task.InitializeAsync(cancellationToken).ConfigureAwait(false);
+            this.Subject.OnNext(new TaskLifeCycleEvent(TaskLifeCycleEventType.Initialized));
+        }
+        catch(HttpRequestException ex)
+        {
+            await this.SetErrorAsync(new Error()
+            {
+                Type = ErrorType.Communication,
+                Title = ErrorTitle.Communication,
+                Status = ex.StatusCode.HasValue ? (ushort)ex.StatusCode : (ushort)ErrorStatus.Communication,
+                Detail = ex.Message,
+                Instance = this.Task.Instance.Reference
+            }, cancellationToken).ConfigureAwait(false);
+        }
+        catch(Exception ex)
+        {
+            await this.SetErrorAsync(new Error()
+            {
+                Type = ErrorType.Runtime,
+                Title = ErrorTitle.Runtime,
+                Status = ErrorStatus.Runtime,
+                Detail = ex.Message,
+                Instance = this.Task.Instance.Reference
+            }, cancellationToken).ConfigureAwait(false);
+        } 
     }
 
     /// <summary>
@@ -171,6 +196,28 @@ await this.SetErrorAsync(new()
             await this.TaskCompletionSource.Task.ConfigureAwait(false);
         }
         catch (OperationCanceledException) { }
+        catch (HttpRequestException ex)
+        {
+            await this.SetErrorAsync(new Error()
+            {
+                Type = ErrorType.Communication,
+                Title = ErrorTitle.Communication,
+                Status = ex.StatusCode.HasValue ? (ushort)ex.StatusCode : (ushort)ErrorStatus.Communication,
+                Detail = ex.Message,
+                Instance = this.Task.Instance.Reference
+            }, cancellationToken).ConfigureAwait(false);
+        }
+        catch (Exception ex)
+        {
+            await this.SetErrorAsync(new Error()
+            {
+                Type = ErrorType.Runtime,
+                Title = ErrorTitle.Runtime,
+                Status = ErrorStatus.Runtime,
+                Detail = ex.Message,
+                Instance = this.Task.Instance.Reference
+            }, cancellationToken).ConfigureAwait(false);
+        }
     }
 
     /// <summary>
@@ -198,7 +245,6 @@ protected virtual async Task BeforeExecuteAsync(CancellationToken cancellationTo
             }
             input = executor.Task.Output ?? new();
             this.Executors.Remove(executor);
-            await executor.DisposeAsync().ConfigureAwait(false);
         }
     }
 

src/runner/Synapse.Runner/Services/WorkflowExecutor.cs

@@ -272,7 +272,6 @@ protected virtual async Task OnTaskCompletedAsync(ITaskExecutor executor, Cancel
         };
         var completedTask = executor.Task;
         this.Executors.Remove(executor);
-        await executor.DisposeAsync().ConfigureAwait(false);
         if (nextDefinition == null)
         {
             await this.SetResultAsync(completedTask.Output, cancellationToken).ConfigureAwait(false);