Skip to content

Commit 3e04a16

Browse files
cdavernascdavernas
committed
fix(Runner): Fixed the TryTaskExecutor by updating context data after running nested try and catch tasks
fix(Api): Fixed the cluster and namespaced resource controllers to handle cancellation-related exceptions when enumerating SSE events fix(Operator): Fixed the `WorkflowController` not to throw when version compared version lists have no diff patch operations fix(Operator): Fixed the `WorkflowInstanceController` to delete all related documents Signed-off-by: Charles d'Avernas <[email protected]>
1 parent 7fd2983 commit 3e04a16

File tree

7 files changed

+87
-32
lines changed

7 files changed

+87
-32
lines changed

‎src/api/Synapse.Api.Http/ClusterResourceController.cs

Lines changed: 16 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -109,12 +109,16 @@ public virtual async Task<IActionResult> WatchResourcesUsingSSE(string? labelSel
109109
this.Response.Headers.CacheControl = "no-cache";
110110
this.Response.Headers.Connection = "keep-alive";
111111
await this.Response.Body.FlushAsync(cancellationToken).ConfigureAwait(false);
112-
await foreach (var e in response.Data!.WithCancellation(cancellationToken))
112+
try
113113
{
114-
var sseMessage = $"data: {this.JsonSerializer.SerializeToText(e)}\n\n";
115-
await this.Response.Body.WriteAsync(Encoding.UTF8.GetBytes(sseMessage), cancellationToken).ConfigureAwait(false);
116-
await this.Response.Body.FlushAsync(cancellationToken).ConfigureAwait(false);
114+
await foreach (var e in response.Data!.WithCancellation(cancellationToken))
115+
{
116+
var sseMessage = $"data: {this.JsonSerializer.SerializeToText(e)}\n\n";
117+
await this.Response.Body.WriteAsync(Encoding.UTF8.GetBytes(sseMessage), cancellationToken).ConfigureAwait(false);
118+
await this.Response.Body.FlushAsync(cancellationToken).ConfigureAwait(false);
119+
}
117120
}
121+
catch (Exception ex) when (ex is TaskCanceledException || ex is OperationCanceledException) { }
118122
return this.Ok();
119123
}
120124

@@ -149,12 +153,16 @@ public virtual async Task<IActionResult> MonitorResourceUsingSSE(string name, Ca
149153
this.Response.Headers.CacheControl = "no-cache";
150154
this.Response.Headers.Connection = "keep-alive";
151155
await this.Response.Body.FlushAsync(cancellationToken).ConfigureAwait(false);
152-
await foreach (var e in response.Data!.WithCancellation(cancellationToken))
156+
try
153157
{
154-
var sseMessage = $"data: {this.JsonSerializer.SerializeToText(e)}\n\n";
155-
await this.Response.Body.WriteAsync(Encoding.UTF8.GetBytes(sseMessage), cancellationToken).ConfigureAwait(false);
156-
await this.Response.Body.FlushAsync(cancellationToken).ConfigureAwait(false);
158+
await foreach (var e in response.Data!.WithCancellation(cancellationToken))
159+
{
160+
var sseMessage = $"data: {this.JsonSerializer.SerializeToText(e)}\n\n";
161+
await this.Response.Body.WriteAsync(Encoding.UTF8.GetBytes(sseMessage), cancellationToken).ConfigureAwait(false);
162+
await this.Response.Body.FlushAsync(cancellationToken).ConfigureAwait(false);
163+
}
157164
}
165+
catch (Exception ex) when (ex is TaskCanceledException || ex is OperationCanceledException) { }
158166
return this.Ok();
159167
}
160168

‎src/api/Synapse.Api.Http/NamespacedResourceController.cs

Lines changed: 16 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -11,8 +11,6 @@
1111
// See the License for the specific language governing permissions and
1212
// limitations under the License.
1313

14-
using Neuroglia.Data.Infrastructure.ResourceOriented;
15-
1614
namespace Synapse.Api.Http;
1715

1816
/// <summary>
@@ -164,12 +162,16 @@ public virtual async Task<IActionResult> WatchResourcesUsingSSE(string @namespac
164162
this.Response.Headers.CacheControl = "no-cache";
165163
this.Response.Headers.Connection = "keep-alive";
166164
await this.Response.Body.FlushAsync(cancellationToken).ConfigureAwait(false);
167-
await foreach (var e in response.Data!.WithCancellation(cancellationToken))
165+
try
168166
{
169-
var sseMessage = $"data: {this.JsonSerializer.SerializeToText(e)}\n\n";
170-
await this.Response.Body.WriteAsync(Encoding.UTF8.GetBytes(sseMessage), cancellationToken).ConfigureAwait(false);
171-
await this.Response.Body.FlushAsync(cancellationToken).ConfigureAwait(false);
167+
await foreach (var e in response.Data!.WithCancellation(cancellationToken))
168+
{
169+
var sseMessage = $"data: {this.JsonSerializer.SerializeToText(e)}\n\n";
170+
await this.Response.Body.WriteAsync(Encoding.UTF8.GetBytes(sseMessage), cancellationToken).ConfigureAwait(false);
171+
await this.Response.Body.FlushAsync(cancellationToken).ConfigureAwait(false);
172+
}
172173
}
174+
catch (Exception ex) when(ex is TaskCanceledException || ex is OperationCanceledException) { }
173175
return this.Ok();
174176
}
175177

@@ -206,12 +208,16 @@ public virtual async Task<IActionResult> MonitorResourceUsingSSE(string name, st
206208
this.Response.Headers.CacheControl = "no-cache";
207209
this.Response.Headers.Connection = "keep-alive";
208210
await this.Response.Body.FlushAsync(cancellationToken).ConfigureAwait(false);
209-
await foreach(var e in response.Data!.WithCancellation(cancellationToken))
211+
try
210212
{
211-
var sseMessage = $"data: {this.JsonSerializer.SerializeToText(e)}\n\n";
212-
await this.Response.Body.WriteAsync(Encoding.UTF8.GetBytes(sseMessage), cancellationToken).ConfigureAwait(false);
213-
await this.Response.Body.FlushAsync(cancellationToken).ConfigureAwait(false);
213+
await foreach (var e in response.Data!.WithCancellation(cancellationToken))
214+
{
215+
var sseMessage = $"data: {this.JsonSerializer.SerializeToText(e)}\n\n";
216+
await this.Response.Body.WriteAsync(Encoding.UTF8.GetBytes(sseMessage), cancellationToken).ConfigureAwait(false);
217+
await this.Response.Body.FlushAsync(cancellationToken).ConfigureAwait(false);
218+
}
214219
}
220+
catch (Exception ex) when (ex is TaskCanceledException || ex is OperationCanceledException) { }
215221
return this.Ok();
216222
}
217223

‎src/operator/Synapse.Operator/Services/WorkflowController.cs

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -55,7 +55,7 @@ public override async Task StartAsync(CancellationToken cancellationToken)
5555
this.Operator!.Select(b => b.Resource.Spec.Selector).DistinctUntilChanged().SubscribeAsync(this.OnResourceSelectorChangedAsync, cancellationToken: cancellationToken);
5656
this.Where(e => e.Type == ResourceWatchEventType.Updated)
5757
.Select(e => new { Workflow = e.Resource, e.Resource.Spec.Versions })
58-
.DistinctUntilChanged()
58+
.DistinctUntilChanged(s => s.Versions)
5959
.Scan((Previous: (EquatableList<WorkflowDefinition>)null!, Current: (EquatableList<WorkflowDefinition>)null!, Workflow: (Workflow)null!), (accumulator, current) => (accumulator.Current, current.Versions, current.Workflow))
6060
.SubscribeAsync(async value => await this.OnWorkflowVersionChangedAsync(value.Workflow, value.Previous, value.Current).ConfigureAwait(false), cancellationToken: cancellationToken);
6161
await this.OnResourceSelectorChangedAsync(this.Operator!.Resource.Spec.Selector).ConfigureAwait(false);
@@ -190,6 +190,7 @@ protected virtual async Task OnWorkflowVersionChangedAsync(Workflow workflow, Eq
190190
if (workflow.Metadata.Labels == null || !workflow.Metadata.Labels.TryGetValue(SynapseDefaults.Resources.Labels.Operator, out _)) if (!await this.TryClaimAsync(workflow, this.CancellationTokenSource.Token).ConfigureAwait(false)) return;
191191
if (workflow.Metadata.Labels?[SynapseDefaults.Resources.Labels.Operator] != this.Operator.Resource.GetQualifiedName()) return;
192192
var diffPatch = JsonPatchUtility.CreateJsonPatchFromDiff(previous, current);
193+
if (diffPatch.Operations.Count < 1) return;
193194
var operation = diffPatch.Operations[0].Op;
194195
if (this.Schedulers.TryRemove(workflow.GetQualifiedName(), out var scheduler)) await scheduler.DisposeAsync().ConfigureAwait(false);
195196
if (operation == OperationType.Remove) return;

‎src/operator/Synapse.Operator/Services/WorkflowInstanceController.cs

Lines changed: 49 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,8 @@
1111
// See the License for the specific language governing permissions and
1212
// limitations under the License.
1313

14+
using Neuroglia.Data.Infrastructure.Services;
15+
1416
namespace Synapse.Operator.Services;
1517

1618
/// <summary>
@@ -21,7 +23,8 @@ namespace Synapse.Operator.Services;
2123
/// <param name="controllerOptions">The service used to access the current <see cref="IOptions{TOptions}"/></param>
2224
/// <param name="repository">The service used to manage <see cref="IResource"/>s</param>
2325
/// <param name="operatorController">The service used to access the current <see cref="Resources.Operator"/></param>
24-
public class WorkflowInstanceController(IServiceProvider serviceProvider, ILoggerFactory loggerFactory, IOptions<ResourceControllerOptions<WorkflowInstance>> controllerOptions, IResourceRepository repository, IOperatorController operatorController)
26+
/// <param name="documents">The <see cref="IRepository"/> used to manage <see cref="Document"/>s</param>
27+
public class WorkflowInstanceController(IServiceProvider serviceProvider, ILoggerFactory loggerFactory, IOptions<ResourceControllerOptions<WorkflowInstance>> controllerOptions, IResourceRepository repository, IOperatorController operatorController, IRepository<Document, string> documents)
2528
: ResourceController<WorkflowInstance>(loggerFactory, controllerOptions, repository)
2629
{
2730

@@ -35,6 +38,11 @@ public class WorkflowInstanceController(IServiceProvider serviceProvider, ILogge
3538
/// </summary>
3639
protected IResourceMonitor<Resources.Operator> Operator => operatorController.Operator;
3740

41+
/// <summary>
42+
/// Gets the <see cref="IRepository"/> used to manage <see cref="Document"/>s
43+
/// </summary>
44+
protected IRepository<Document, string> Documents => documents;
45+
3846
/// <summary>
3947
/// Gets a <see cref="ConcurrentDictionary{TKey, TValue}"/> that contains current <see cref="WorkflowInstanceHandler"/>es
4048
/// </summary>
@@ -139,24 +147,54 @@ public override async Task StopAsync(CancellationToken cancellationToken)
139147
/// <inheritdoc/>
140148
protected override async Task OnResourceCreatedAsync(WorkflowInstance workflowInstance, CancellationToken cancellationToken = default)
141149
{
142-
await base.OnResourceCreatedAsync(workflowInstance, cancellationToken).ConfigureAwait(false);
143-
if (!await this.TryClaimAsync(workflowInstance, cancellationToken).ConfigureAwait(false)) return;
144-
var handler = await this.CreateWorkflowInstanceHandlerAsync(workflowInstance, cancellationToken).ConfigureAwait(false);
145-
await handler.HandleAsync(cancellationToken).ConfigureAwait(false);
150+
try
151+
{
152+
await base.OnResourceCreatedAsync(workflowInstance, cancellationToken).ConfigureAwait(false);
153+
if (!await this.TryClaimAsync(workflowInstance, cancellationToken).ConfigureAwait(false)) return;
154+
var handler = await this.CreateWorkflowInstanceHandlerAsync(workflowInstance, cancellationToken).ConfigureAwait(false);
155+
await handler.HandleAsync(cancellationToken).ConfigureAwait(false);
156+
}
157+
catch(Exception ex)
158+
{
159+
this.Logger.LogError("An error occured while handling the creation of workflow instance '{workflowInstance}': {ex}", workflowInstance.GetQualifiedName(), ex);
160+
}
146161
}
147162

148163
/// <inheritdoc/>
149164
protected override async Task OnResourceDeletedAsync(WorkflowInstance workflowInstance, CancellationToken cancellationToken = default)
150165
{
151-
await base.OnResourceDeletedAsync(workflowInstance, cancellationToken).ConfigureAwait(false);
152-
if (this.Handlers.TryRemove(workflowInstance.GetQualifiedName(), out var process)) await process.DisposeAsync().ConfigureAwait(false);
153-
var selectors = new LabelSelector[]
166+
try
154167
{
168+
await base.OnResourceDeletedAsync(workflowInstance, cancellationToken).ConfigureAwait(false);
169+
if (this.Handlers.TryRemove(workflowInstance.GetQualifiedName(), out var process)) await process.DisposeAsync().ConfigureAwait(false);
170+
var selectors = new LabelSelector[]
171+
{
155172
new(SynapseDefaults.Resources.Labels.WorkflowInstance, LabelSelectionOperator.Equals, workflowInstance.GetQualifiedName())
156-
};
157-
await foreach (var correlation in this.Repository.GetAllAsync<Correlation>(null, selectors, cancellationToken: cancellationToken))
173+
};
174+
await foreach (var correlation in this.Repository.GetAllAsync<Correlation>(null, selectors, cancellationToken: cancellationToken))
175+
{
176+
await this.Repository.RemoveAsync<Correlation>(correlation.GetName(), correlation.GetNamespace(), false, cancellationToken).ConfigureAwait(false);
177+
}
178+
if (workflowInstance.Status != null)
179+
{
180+
var documentReferences = new List<string>();
181+
if (!string.IsNullOrWhiteSpace(workflowInstance.Status.ContextReference)) documentReferences.Add(workflowInstance.Status.ContextReference);
182+
if (!string.IsNullOrWhiteSpace(workflowInstance.Status.OutputReference)) documentReferences.Add(workflowInstance.Status.OutputReference);
183+
if (workflowInstance.Status.Tasks != null)
184+
{
185+
foreach (var task in workflowInstance.Status.Tasks)
186+
{
187+
if (!string.IsNullOrWhiteSpace(task.ContextReference)) documentReferences.Add(task.ContextReference);
188+
if (!string.IsNullOrWhiteSpace(task.InputReference)) documentReferences.Add(task.InputReference);
189+
if (!string.IsNullOrWhiteSpace(task.OutputReference)) documentReferences.Add(task.OutputReference);
190+
}
191+
}
192+
foreach (var documentReference in documentReferences.Distinct()) await this.Documents.RemoveAsync(documentReference, cancellationToken).ConfigureAwait(false);
193+
}
194+
}
195+
catch(Exception ex)
158196
{
159-
await this.Repository.RemoveAsync<Correlation>(correlation.GetName(), correlation.GetNamespace(), false, cancellationToken).ConfigureAwait(false);
197+
this.Logger.LogError("An error occured while handling the deletion of workflow instance '{workflowInstance}': {ex}", workflowInstance.GetQualifiedName(), ex);
160198
}
161199
}
162200

‎src/operator/Synapse.Operator/appsettings.Development.json

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@
1616
},
1717
"Runtime": {
1818
"Native": {
19-
"Directory": "..\\..\\..\\..\\..\\runner\\Synapse.Runner\\bin\\Debug\\net8.0\\",
19+
"Directory": "..\\..\\..\\..\\..\\runner\\Synapse.Runner\\bin\\Debug\\net9.0\\",
2020
"Executable": "Synapse.Runner.exe"
2121
}
2222
}

‎src/runner/Synapse.Runner/Services/Executors/TryTaskExecutor.cs

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -158,6 +158,7 @@ protected virtual async Task OnTryFaultedAsync(ITaskExecutor executor, Exception
158158
protected virtual async Task OnTryCompletedAsync(ITaskExecutor executor, CancellationToken cancellationToken)
159159
{
160160
ArgumentNullException.ThrowIfNull(executor);
161+
if (this.Task.ContextData != executor.Task.ContextData) await this.Task.SetContextDataAsync(executor.Task.ContextData, cancellationToken).ConfigureAwait(false);
161162
var last = executor.Task.Instance;
162163
var output = executor.Task.Output!;
163164
this.Executors.Remove(executor);
@@ -188,6 +189,7 @@ protected virtual async Task OnHandlerFaultAsync(ITaskExecutor executor, Excepti
188189
protected virtual async Task OnHandlerCompletedAsync(ITaskExecutor executor, CancellationToken cancellationToken)
189190
{
190191
ArgumentNullException.ThrowIfNull(executor);
192+
if (this.Task.ContextData != executor.Task.ContextData) await this.Task.SetContextDataAsync(executor.Task.ContextData, cancellationToken).ConfigureAwait(false);
191193
var last = executor.Task.Instance;
192194
var output = executor.Task.Output!;
193195
this.Executors.Remove(executor);

‎tests/Synapse.UnitTests/Cases/Runtime/NativeRuntimeTests.cs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ protected override void ConfigureServices(IServiceCollection services)
3131
};
3232
options.Runtime.Native = new()
3333
{
34-
Directory = Path.Combine("..", "..", "..", "..", "..", "src", "runner", "Synapse.Runner", "bin", "Debug", "net8.0"),
34+
Directory = Path.Combine("..", "..", "..", "..", "..", "src", "runner", "Synapse.Runner", "bin", "Debug", "net9.0"),
3535
Executable = "Synapse.Runner"
3636
};
3737
});

0 commit comments

Comments
 (0)