fix(Runner): Fixed the ConnectedWorkflowExecutionContext to remove handled correlation contexts from the WorkflowInstance they are associated to

fix(Runner): Fixed the `ConnectedWorkflowExecutionContext` to populate `WorkfowInstance` correlation keys set by handled correlation contexts
fix(Runner): Fixed a feature-breaking bug with the `AsyncApiCallExecutor` disposing inline of the IAsyncApiSubscribeOperationResult, thus irreversibly breaking the message stream
fix(Runner): Fixed the `AsyncApiCallExecutor`, which was not properly evaluating `while` and `until` message consumption conditions
fix(Runner): Fixed the `AsyncApiCallExecutor`, which was not passing the `item` and `index` arguments to the `output.as` and `export.as` expressions

Signed-off-by: Charles d'Avernas <[email protected]>

Author: cdavernas

src/api/Synapse.Api.Http/Extensions/IServiceCollectionExtensions.cs

@@ -20,6 +20,7 @@
 using Synapse.Api.Http.Controllers;
 using Synapse.Core.Api.Services;
 using System.Text.Json;
+using System.Text.Json.Serialization;
 
 namespace Synapse.Api.Http;
 
@@ -44,6 +45,7 @@ public static IServiceCollection AddSynapseHttpApi(this IServiceCollection servi
             .AddJsonOptions(options =>
             {
                 options.JsonSerializerOptions.PropertyNamingPolicy = JsonNamingPolicy.CamelCase;
+                options.JsonSerializerOptions.DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull | JsonIgnoreCondition.WhenWritingDefault;
             })
             .AddApplicationPart(typeof(WorkflowsController).Assembly);
         services.AddIdentityServer(options =>

src/core/Synapse.Core/Resources/CorrelationSpec.cs

@@ -46,16 +46,22 @@ public record CorrelationSpec
     [DataMember(Name = "events", Order = 4), JsonPropertyName("events"), JsonPropertyOrder(4), YamlMember(Alias = "events", Order = 4)]
     public virtual EventConsumptionStrategyDefinition Events { get; set; } = null!;
 
+    /// <summary>
+    /// Gets/sets a key/value mapping, if any, of the keys to use to correlate events
+    /// </summary>
+    [DataMember(Name = "keys", Order = 5), JsonPropertyName("keys"), JsonPropertyOrder(5), YamlMember(Alias = "keys", Order = 5)]
+    public virtual EquatableDictionary<string, string>? Keys { get; set; }
+
     /// <summary>
     /// Gets/sets a boolean indicating whether or not to stream events. When enabled, each correlated event is atomically published to the subscriber immediately rather than waiting for the entire correlation to complete
     /// </summary>
-    [DataMember(Name = "stream", Order = 5), JsonPropertyName("stream"), JsonPropertyOrder(5), YamlMember(Alias = "stream", Order = 5)]
+    [DataMember(Name = "stream", Order = 6), JsonPropertyName("stream"), JsonPropertyOrder(6), YamlMember(Alias = "stream", Order = 6)]
     public virtual bool Stream { get; set; }
 
     /// <summary>
     /// Gets/sets an object used to configure the correlation's outcome
     /// </summary>
-    [DataMember(Name = "outcome", Order = 6), JsonPropertyName("outcome"), JsonPropertyOrder(6), YamlMember(Alias = "outcome", Order = 6)]
+    [DataMember(Name = "outcome", Order = 7), JsonPropertyName("outcome"), JsonPropertyOrder(7), YamlMember(Alias = "outcome", Order = 7)]
     public virtual CorrelationOutcomeDefinition Outcome { get; set; } = null!;
 
 }

src/correlator/Synapse.Correlator/Services/CorrelationHandler.cs

@@ -123,7 +123,7 @@ protected virtual async Task CorrelateEventAsync(CloudEvent e, CancellationToken
                         {
                             Id = Guid.NewGuid().ToString("N")[..12],
                             Events = [new(filter.Key, e)],
-                            Keys = CorrelationKeys == null ? new() : new(CorrelationKeys)
+                            Keys = CorrelationKeys == null ? this.Correlation.Resource.Spec.Keys ?? [] : new(CorrelationKeys)
                         };
                         this.Logger.LogInformation("Correlation context with id '{contextId}' successfully created", context.Id);
                         this.Logger.LogInformation("Event successfully correlated to context with id '{contextId}'", context.Id);
@@ -152,7 +152,7 @@ protected virtual async Task CorrelateEventAsync(CloudEvent e, CancellationToken
                         {
                             Id = Guid.NewGuid().ToString("N")[..12],
                             Events = [new(filter.Key, e)],
-                            Keys = CorrelationKeys == null ? new() : new(CorrelationKeys)
+                            Keys = CorrelationKeys == null ? this.Correlation.Resource.Spec.Keys ?? [] : new(CorrelationKeys)
                         };
                         await this.CreateOrUpdateContextAsync(context, cancellationToken).ConfigureAwait(false);
                         this.Logger.LogInformation("Correlation context with id '{contextId}' successfully created", context.Id);
@@ -289,7 +289,7 @@ protected virtual async Task<bool> TryFilterEventAsync(EventFilterDefinition fil
     protected virtual async Task<(bool Succeeded, IDictionary<string, string>? CorrelationKeys)> TryExtractCorrelationKeysAsync(CloudEvent e, IDictionary<string, CorrelationKeyDefinition>? keyDefinitions, CancellationToken cancellationToken)
     {
         ArgumentNullException.ThrowIfNull(e);
-        var correlationKeys = new Dictionary<string, string>();
+        var correlationKeys = this.Correlation.Resource.Spec.Keys ?? [];
         if (keyDefinitions == null || keyDefinitions.Count < 1) return (true, correlationKeys);
         foreach (var keyDefinition in keyDefinitions)
         {
@@ -305,6 +305,7 @@ protected virtual async Task<bool> TryFilterEventAsync(EventFilterDefinition fil
                 }
                 else if (!keyDefinition.Value.Expect.Equals(correlationTerm, StringComparison.OrdinalIgnoreCase)) return (false, null);
             }
+            if (correlationKeys.ContainsKey(keyDefinition.Key) && correlationTerm != correlationKeys[keyDefinition.Key]) return (false, null);
             correlationKeys[keyDefinition.Key] = correlationTerm;
         }
         return (true, correlationKeys);
@@ -373,6 +374,13 @@ protected virtual async Task CreateOrUpdateContextAsync(CorrelationContext conte
                         {
                             Definition = this.Correlation.Resource.Spec.Outcome.Start!.Workflow,
                             Input = input
+                        },
+                        Status = new()
+                        {
+                            Correlation = new()
+                            {
+                                Keys = context.Keys
+                            }
                         }
                     };
                     await this.Resources.AddAsync(workflowInstance, false, cancellationToken).ConfigureAwait(false);

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

@@ -449,6 +449,7 @@ public virtual async Task<CorrelationContext> CorrelateAsync(ITaskExecutionConte
                     Source = new ResourceReference<WorkflowInstance>(task.Workflow.Instance.GetName(), task.Workflow.Instance.GetNamespace()),
                     Lifetime = CorrelationLifetime.Ephemeral,
                     Events = listenTask.Listen.To,
+                    Keys = this.Instance.Status?.Correlation?.Keys,
                     Expressions = task.Workflow.Definition.Evaluate ?? new(),
                     Outcome = new()
                     {
@@ -511,6 +512,17 @@ public virtual async Task<CorrelationContext> CorrelateAsync(ITaskExecutionConte
                 CompletedAt = DateTimeOffset.Now
             }
         }, cancellationToken).ConfigureAwait(false);
+        using var @lock = await this.Lock.LockAsync(cancellationToken).ConfigureAwait(false);
+        this.Instance = await this.Api.WorkflowInstances.GetAsync(this.Instance.GetName(), this.Instance.GetNamespace()!, cancellationToken).ConfigureAwait(false);
+        var originalInstance = this.Instance.Clone();
+        foreach(var correlationKey in correlationContext.Keys)
+        {
+            this.Instance.Status!.Correlation!.Keys ??= [];
+            this.Instance.Status!.Correlation!.Keys[correlationKey.Key] = correlationKey.Value;
+        }
+        this.Instance.Status!.Correlation!.Contexts!.Remove(task.Instance.Reference.OriginalString);
+        var jsonPatch = JsonPatchUtility.CreateJsonPatchFromDiff(originalInstance, this.Instance);
+        this.Instance = await this.Api.WorkflowInstances.PatchStatusAsync(this.Instance.GetName(), this.Instance.GetNamespace()!, new Patch(PatchType.JsonPatch, jsonPatch), null, cancellationToken).ConfigureAwait(false);
         return correlationContext;
     }
 

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

@@ -18,6 +18,7 @@
 using Neuroglia.AsyncApi.IO;
 using Neuroglia.AsyncApi.v3;
 using Neuroglia.Data.Expressions;
+using System.Threading;
 
 namespace Synapse.Runner.Services.Executors;
 
@@ -94,6 +95,11 @@ public class AsyncApiCallExecutor(IServiceProvider serviceProvider, ILogger<Asyn
     /// </summary>
     protected uint? Offset { get; set; }
 
+    /// <summary>
+    /// Gets/sets a boolean indicating whether or not to keep consuming incoming messages 
+    /// </summary>
+    protected bool KeepConsume { get; set; } = true;
+
     /// <summary>
     /// Gets the path for the specified message
     /// </summary>
@@ -234,7 +240,7 @@ protected virtual async Task DoExecuteSubscribeOperationAsync(CancellationToken
         if (this.AsyncApi.Subscription == null) throw new NullReferenceException("The 'subscription' must be set when performing an AsyncAPI v3 subscribe operation");
         await using var asyncApiClient = this.AsyncApiClientFactory.CreateFor(this.Document);
         var parameters = new AsyncApiSubscribeOperationParameters(this.Operation.Key, this.AsyncApi.Server, this.AsyncApi.Protocol);
-        await using var result = await asyncApiClient.SubscribeAsync(parameters, cancellationToken).ConfigureAwait(false);
+        var result = await asyncApiClient.SubscribeAsync(parameters, cancellationToken).ConfigureAwait(false);
         if (!result.IsSuccessful) throw new Exception("Failed to execute the AsyncAPI subscribe operation");
         if(result.Messages == null)
         {
@@ -244,24 +250,24 @@ protected virtual async Task DoExecuteSubscribeOperationAsync(CancellationToken
         var observable = result.Messages;
         if (this.AsyncApi.Subscription.Consume.For != null) observable = observable.TakeUntil(Observable.Timer(this.AsyncApi.Subscription.Consume.For.ToTimeSpan()));
         if (this.AsyncApi.Subscription.Consume.Amount.HasValue) observable = observable.Take(this.AsyncApi.Subscription.Consume.Amount.Value);
-        else if (!string.IsNullOrWhiteSpace(this.AsyncApi.Subscription.Consume.While)) observable = observable.Select(message => Observable.FromAsync(async () =>
-        {
-            var keepGoing = await this.Task.Workflow.Expressions.EvaluateConditionAsync(this.AsyncApi.Subscription.Consume.While, this.Task.Input!,this.GetExpressionEvaluationArguments(),cancellationToken).ConfigureAwait(false);
-            return (message, keepGoing);
-        })).Concat().TakeWhile(i => i.keepGoing).Select(i => i.message);
-        else if (!string.IsNullOrWhiteSpace(this.AsyncApi.Subscription.Consume.Until)) observable = observable.Select(message => Observable.FromAsync(async () =>
-        {
-            var keepGoing = !(await this.Task.Workflow.Expressions.EvaluateConditionAsync(this.AsyncApi.Subscription.Consume.Until, this.Task.Input!, this.GetExpressionEvaluationArguments(), cancellationToken).ConfigureAwait(false));
-            return (message, keepGoing);
-        })).Concat().TakeWhile(i => i.keepGoing).Select(i => i.message);
         if (this.AsyncApi.Subscription.Foreach == null)
         {
-            var messages = await observable.ToAsyncEnumerable().ToListAsync(cancellationToken).ConfigureAwait(false);
+            var messages = await observable.ToAsyncEnumerable().TakeWhileAwait(async m =>
+            {
+                if (!string.IsNullOrWhiteSpace(this.AsyncApi.Subscription.Consume.While)) return await this.Task.Workflow.Expressions.EvaluateConditionAsync(this.AsyncApi.Subscription.Consume.While, this.Task.Input!, this.GetExpressionEvaluationArguments(), cancellationToken).ConfigureAwait(false);
+                if (!string.IsNullOrWhiteSpace(this.AsyncApi.Subscription.Consume.Until)) return !await this.Task.Workflow.Expressions.EvaluateConditionAsync(this.AsyncApi.Subscription.Consume.Until, this.Task.Input!, this.GetExpressionEvaluationArguments(), cancellationToken).ConfigureAwait(false);
+                return true;
+            }).ToListAsync(cancellationToken).ConfigureAwait(false);
             await this.SetResultAsync(messages, this.Task.Definition.Then, cancellationToken).ConfigureAwait(false);
         }
         else
         {
-            this.Subscription = observable.SubscribeAsync(OnStreamingMessageAsync, OnStreamingErrorAsync, OnStreamingCompletedAsync);
+            //todo: fix
+            this.Subscription = observable.TakeWhile(_ => this.KeepConsume).SelectMany(m =>
+            {
+                OnStreamingMessageAsync(m).GetAwaiter().GetResult();
+                return Observable.Return(m);
+            }).SubscribeAsync(_ => System.Threading.Tasks.Task.CompletedTask, OnStreamingErrorAsync, OnStreamingCompletedAsync);
         }
     }
 
@@ -274,6 +280,11 @@ protected virtual async Task OnStreamingMessageAsync(IAsyncApiMessage message)
     {
         if (this.AsyncApi == null || this.Document == null || this.Operation.Value == null) throw new InvalidOperationException("The executor must be initialized before execution");
         if (this.AsyncApi.Subscription == null) throw new NullReferenceException("The 'subscription' must be set when performing an AsyncAPI v3 subscribe operation");
+        if (!string.IsNullOrWhiteSpace(this.AsyncApi.Subscription.Consume.While) && !await this.Task.Workflow.Expressions.EvaluateConditionAsync(this.AsyncApi.Subscription.Consume.While, this.Task.Input!, this.GetExpressionEvaluationArguments(), this.CancellationTokenSource!.Token).ConfigureAwait(false))
+        {
+            this.KeepConsume = false;
+            return;
+        }
         if (this.AsyncApi.Subscription.Foreach?.Do != null)
         {
             var taskDefinition = new DoTaskDefinition()
@@ -284,30 +295,36 @@ protected virtual async Task OnStreamingMessageAsync(IAsyncApiMessage message)
                     new(SynapseDefaults.Tasks.Metadata.PathPrefix.Name, false)
                 ]
             };
-            var arguments = this.GetExpressionEvaluationArguments();
             var messageData = message as object;
+            var offset = this.Offset ?? 0;
+            if (!this.Offset.HasValue) this.Offset = 0;
+            var arguments = this.GetExpressionEvaluationArguments();
+            arguments ??= new Dictionary<string, object>();
+            arguments[this.AsyncApi.Subscription.Foreach.Item ?? RuntimeExpressions.Arguments.Each] = messageData!;
+            arguments[this.AsyncApi.Subscription.Foreach.At ?? RuntimeExpressions.Arguments.Index] = offset;
             if (this.AsyncApi.Subscription.Foreach.Output?.As is string fromExpression) messageData = await this.Task.Workflow.Expressions.EvaluateAsync<object>(fromExpression, messageData ?? new(), arguments, this.CancellationTokenSource!.Token).ConfigureAwait(false);
-            else if (this.AsyncApi.Subscription.Foreach.Output?.As != null) messageData = await this.Task.Workflow.Expressions.EvaluateAsync<object>(this.AsyncApi.Subscription.Foreach.Output.As, messageData ?? new(), arguments, this.CancellationTokenSource!.Token).ConfigureAwait(false);
+            else if (this.AsyncApi.Subscription.Foreach.Output?.As != null) messageData = await this.Task.Workflow.Expressions.EvaluateAsync<object>(this.AsyncApi.Subscription.Foreach.Output.As!, messageData ?? new(), arguments, this.CancellationTokenSource!.Token).ConfigureAwait(false);
             if (this.AsyncApi.Subscription.Foreach.Export?.As is string toExpression)
             {
                 var context = (await this.Task.Workflow.Expressions.EvaluateAsync<IDictionary<string, object>>(toExpression, messageData ?? new(), arguments, this.CancellationTokenSource!.Token).ConfigureAwait(false))!;
                 await this.Task.SetContextDataAsync(context, this.CancellationTokenSource!.Token).ConfigureAwait(false);
             }
             else if (this.AsyncApi.Subscription.Foreach.Export?.As != null)
             {
-                var context = (await this.Task.Workflow.Expressions.EvaluateAsync<IDictionary<string, object>>(this.AsyncApi.Subscription.Foreach.Export.As, messageData ?? new(), arguments, this.CancellationTokenSource!.Token).ConfigureAwait(false))!;
+                var context = (await this.Task.Workflow.Expressions.EvaluateAsync<IDictionary<string, object>>(this.AsyncApi.Subscription.Foreach.Export.As!, messageData ?? new(), arguments, this.CancellationTokenSource!.Token).ConfigureAwait(false))!;
                 await this.Task.SetContextDataAsync(context, this.CancellationTokenSource!.Token).ConfigureAwait(false);
             }
-            var offset = this.Offset ?? 0;
-            if (!this.Offset.HasValue) this.Offset = 0;
-            arguments ??= new Dictionary<string, object>();
-            arguments[this.AsyncApi.Subscription.Foreach.Item ?? RuntimeExpressions.Arguments.Each] = messageData!;
-            arguments[this.AsyncApi.Subscription.Foreach.At ?? RuntimeExpressions.Arguments.Index] = offset;
             var task = await this.Task.Workflow.CreateTaskAsync(taskDefinition, this.GetPathFor(offset), this.Task.Input, null, this.Task, false, this.CancellationTokenSource!.Token).ConfigureAwait(false);
             var taskExecutor = await this.CreateTaskExecutorAsync(task, taskDefinition, this.Task.ContextData, arguments, this.CancellationTokenSource!.Token).ConfigureAwait(false);
             await taskExecutor.ExecuteAsync(this.CancellationTokenSource!.Token).ConfigureAwait(false);
+            if (this.Task.ContextData != taskExecutor.Task.ContextData) await this.Task.SetContextDataAsync(taskExecutor.Task.ContextData, this.CancellationTokenSource!.Token).ConfigureAwait(false);
             this.Offset++;
         }
+        if (!string.IsNullOrWhiteSpace(this.AsyncApi.Subscription.Consume.Until) && await this.Task.Workflow.Expressions.EvaluateConditionAsync(this.AsyncApi.Subscription.Consume.Until, this.Task.Input!, this.GetExpressionEvaluationArguments(), this.CancellationTokenSource!.Token).ConfigureAwait(false))
+        {
+            this.KeepConsume = false;
+            return;
+        }
     }
 
     /// <summary>