Fixed the WorkflowInstanceHandler to handle exceptions thrown while persisting log batches, which were crashing the operator

src/operator/Synapse.Operator/Services/WorkflowInstanceHandler.cs

@@ -201,13 +201,24 @@ protected virtual async Task OnCorrelationContextsChangedAsync((EquatableDiction
     /// <returns>A new awaitable <see cref="Task"/></returns>
     protected virtual async Task OnPersistLogBatchAsync()
     {
-        if (this._persistingLogs || this.LogBatchQueue.Count < 1) return;
+        if (this._disposed || this._persistingLogs || this.LogBatchQueue.Count < 1) return;
         this._persistingLogs = true;
-        var stringBuilder = new StringBuilder();
-        while (this.LogBatchQueue.TryDequeue(out var log)) stringBuilder.AppendLine(log);
-        var logs = stringBuilder.ToString();
-        await this.Logs.AppendAsync($"{this.WorkflowInstance.Resource.GetName()}.{this.WorkflowInstance.Resource.GetNamespace()}", logs, this.CancellationTokenSource.Token).ConfigureAwait(false);
-        this._persistingLogs = false;
+        try
+        {
+            var stringBuilder = new StringBuilder();
+            while (this.LogBatchQueue.TryDequeue(out var log)) stringBuilder.AppendLine(log);
+            var logs = stringBuilder.ToString();
+            await this.Logs.AppendAsync($"{this.WorkflowInstance.Resource.GetName()}.{this.WorkflowInstance.Resource.GetNamespace()}", logs, this.CancellationTokenSource.Token).ConfigureAwait(false);
+        }
+        catch (Exception ex) when (ex is ObjectDisposedException || ex is OperationCanceledException) { }
+        catch (Exception ex)
+        {
+            this.Logger.LogError("An error occurred while persisting the logs of workflow instance '{workflowInstance}': {ex}", this.WorkflowInstance.Resource.GetQualifiedName(), ex);
+        }
+        finally
+        {
+            this._persistingLogs = false;
+        }
     }
 
     /// <summary>

tests/Synapse.UnitTests/Cases/Operator/WorkflowInstanceHandlerTests.cs

@@ -0,0 +1,81 @@
+// Copyright © 2024-Present The Synapse Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License"),
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using Microsoft.Extensions.Logging.Abstractions;
+using Moq;
+using Neuroglia.Data.Infrastructure.ResourceOriented;
+using Neuroglia.Data.Infrastructure.ResourceOriented.Services;
+using Neuroglia.Data.Infrastructure.Services;
+using Synapse.Operator.Configuration;
+using Synapse.Operator.Services;
+using Synapse.Runtime.Services;
+
+namespace Synapse.UnitTests.Cases.Operator;
+
+public class WorkflowInstanceHandlerTests
+{
+
+    [Fact]
+    public async Task Persist_Log_Batch_After_Dispose_Should_Not_Throw()
+    {
+        //arrange
+        var handler = new TestWorkflowInstanceHandler(Mock.Of<ITextDocumentRepository>());
+        handler.EnqueueLog("log");
+        await handler.DisposeAsync();
+
+        //act
+        var persist = () => handler.PersistLogBatchAsync();
+
+        //assert
+        await persist.Should().NotThrowAsync();
+    }
+
+    [Fact]
+    public async Task Persist_Log_Batch_When_Append_Fails_Should_Not_Throw_And_Should_Retry_Next_Batch()
+    {
+        //arrange
+        var logs = new Mock<ITextDocumentRepository>();
+        logs.SetupSequence(l => l.AppendAsync(It.IsAny<object>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
+            .ThrowsAsync(new IOException())
+            .Returns(Task.CompletedTask);
+        var handler = new TestWorkflowInstanceHandler(logs.Object);
+        handler.EnqueueLog("first");
+
+        //act
+        var persist = () => handler.PersistLogBatchAsync();
+        await persist.Should().NotThrowAsync();
+        handler.EnqueueLog("second");
+        await handler.PersistLogBatchAsync();
+
+        //assert
+        logs.Verify(l => l.AppendAsync(It.IsAny<object>(), It.IsAny<string>(), It.IsAny<CancellationToken>()), Times.Exactly(2));
+    }
+
+    class TestWorkflowInstanceHandler(ITextDocumentRepository logs)
+        : WorkflowInstanceHandler(NullLogger<WorkflowInstanceHandler>.Instance, Microsoft.Extensions.Options.Options.Create(new OperatorOptions()), Mock.Of<IResourceRepository>(), logs, Mock.Of<IWorkflowRuntime>(), Mock.Of<IJsonSerializer>(), CreateWorkflowInstanceMonitor())
+    {
+
+        public void EnqueueLog(string log) => this.LogBatchQueue.Enqueue(log);
+
+        public Task PersistLogBatchAsync() => this.OnPersistLogBatchAsync();
+
+        static IResourceMonitor<WorkflowInstance> CreateWorkflowInstanceMonitor()
+        {
+            var monitor = new Mock<IResourceMonitor<WorkflowInstance>>();
+            monitor.Setup(m => m.Resource).Returns(new WorkflowInstance(new ResourceMetadata("fake-instance", "default"), new WorkflowInstanceSpec()));
+            return monitor.Object;
+        }
+
+    }
+
+}

tests/Synapse.UnitTests/Services/MockCloudFlowsApiClient.cs

@@ -28,7 +28,7 @@ internal class MockSynapseApiClient(IServiceProvider serviceProvider)
 
     public IClusterResourceApiClient<Namespace> Namespaces { get; } = ActivatorUtilities.CreateInstance<MockClusterResourceApiClient<Namespace>>(serviceProvider);
 
-    public INamespacedResourceApiClient<Operator> Operators { get; } = ActivatorUtilities.CreateInstance<MockNamespacedResourceApiClient<Operator>>(serviceProvider);
+    public INamespacedResourceApiClient<Synapse.Resources.Operator> Operators { get; } = ActivatorUtilities.CreateInstance<MockNamespacedResourceApiClient<Synapse.Resources.Operator>>(serviceProvider);
 
     public INamespacedResourceApiClient<Workflow> Workflows { get; } = ActivatorUtilities.CreateInstance<MockNamespacedResourceApiClient<Workflow>>(serviceProvider);
 

tests/Synapse.UnitTests/Synapse.UnitTests.csproj

@@ -38,6 +38,7 @@
   <ItemGroup>
     <ProjectReference Include="..\..\src\core\Synapse.Core.Infrastructure.Containers.Docker\Synapse.Core.Infrastructure.Containers.Docker.csproj" />
     <ProjectReference Include="..\..\src\correlator\Synapse.Correlator\Synapse.Correlator.csproj" />
+    <ProjectReference Include="..\..\src\operator\Synapse.Operator\Synapse.Operator.csproj" />
     <ProjectReference Include="..\..\src\runner\Synapse.Runner\Synapse.Runner.csproj" />
     <ProjectReference Include="..\..\src\runtime\Synapse.Runtime.Native\Synapse.Runtime.Native.csproj" />
   </ItemGroup>