From 149f6f200f7845ec54a7c1ab3d3171c9cf3d1b60 Mon Sep 17 00:00:00 2001
From: Raphael Fakhri <153192858+RaphaelFakhri@users.noreply.github.com>
Date: Tue, 3 Nov 2026 11:13:37 +0000
Subject: [PATCH] Fixed the `WorkflowInstanceHandler` to handle exceptions
thrown while persisting log batches, which were crashing the operator
Fixes #564
---
.../Services/WorkflowInstanceHandler.cs | 23 ++++--
.../Operator/WorkflowInstanceHandlerTests.cs | 81 +++++++++++++++++++
.../Services/MockCloudFlowsApiClient.cs | 2 +-
.../Synapse.UnitTests.csproj | 1 +
4 files changed, 100 insertions(+), 7 deletions(-)
create mode 100644 tests/Synapse.UnitTests/Cases/Operator/WorkflowInstanceHandlerTests.cs
diff --git a/src/operator/Synapse.Operator/Services/WorkflowInstanceHandler.cs b/src/operator/Synapse.Operator/Services/WorkflowInstanceHandler.cs
index 768893a5c..46fd0b98f 100644
--- a/src/operator/Synapse.Operator/Services/WorkflowInstanceHandler.cs
+++ b/src/operator/Synapse.Operator/Services/WorkflowInstanceHandler.cs
@@ -201,13 +201,24 @@ protected virtual async Task OnCorrelationContextsChangedAsync((EquatableDiction
/// A new awaitable
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;
+ }
}
///
diff --git a/tests/Synapse.UnitTests/Cases/Operator/WorkflowInstanceHandlerTests.cs b/tests/Synapse.UnitTests/Cases/Operator/WorkflowInstanceHandlerTests.cs
new file mode 100644
index 000000000..f151d7788
--- /dev/null
+++ b/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());
+ 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();
+ logs.SetupSequence(l => l.AppendAsync(It.IsAny