fix(Operator): Fixed the WorkflowController to label correlations created by a workflow schedule

fix(Operator): Fixed the WorkflowController/WorkflowInstanceController to delete correlations owned by a deleted workflow/workflow instance
fix(Runner): Fixed the WorkflowExecutionContext to label correlations created  by a workflow instance

Author: cdavernas

src/core/Synapse.Core/SynapseDefaults.cs

@@ -360,6 +360,11 @@ public static class Labels
             /// Gets the label used by Synapse to indicate the version of the workflow used by a workflow instance
             /// </summary>
             public const string WorkflowVersion = Prefix + "workflow/version";
+            /// <summary>
+            /// Gets the label used by Synapse to indicate the qualified name of the workflow instance that owns a concept, such as a correlation
+            /// </summary>
+            public const string WorkflowInstance = Prefix + "workflow-instance";
+
         }
 
     }

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

@@ -163,10 +163,18 @@ protected override async Task OnResourceDeletedAsync(Workflow workflow, Cancella
     {
         await base.OnResourceDeletedAsync(workflow, cancellationToken).ConfigureAwait(false);
         if (this.Schedulers.TryRemove(workflow.GetQualifiedName(), out var scheduler)) await scheduler.DisposeAsync().ConfigureAwait(false);
-        await foreach(var instance in this.Repository.GetAllAsync<WorkflowInstance>(cancellationToken: cancellationToken))
+        var selectors = new LabelSelector[]
+        {
+            new(SynapseDefaults.Resources.Labels.Workflow, LabelSelectionOperator.Equals, workflow.GetQualifiedName())
+        };
+        await foreach(var instance in this.Repository.GetAllAsync<WorkflowInstance>(null, selectors, cancellationToken: cancellationToken))
         {
             await this.Repository.RemoveAsync<WorkflowInstance>(instance.GetName(), instance.GetNamespace(), false, cancellationToken).ConfigureAwait(false);
         }
+        await foreach (var correlation in this.Repository.GetAllAsync<Correlation>(null, selectors, cancellationToken: cancellationToken))
+        {
+            await this.Repository.RemoveAsync<Correlation>(correlation.GetName(), correlation.GetNamespace(), false, cancellationToken).ConfigureAwait(false);
+        }
     }
 
     /// <summary>

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

@@ -150,6 +150,14 @@ protected override async Task OnResourceDeletedAsync(WorkflowInstance workflowIn
     {
         await base.OnResourceDeletedAsync(workflowInstance, cancellationToken).ConfigureAwait(false);
         if (this.Handlers.TryRemove(workflowInstance.GetQualifiedName(), out var process)) await process.DisposeAsync().ConfigureAwait(false);
+        var selectors = new LabelSelector[]
+        {
+            new(SynapseDefaults.Resources.Labels.WorkflowInstance, LabelSelectionOperator.Equals, workflowInstance.GetQualifiedName())
+        };
+        await foreach (var correlation in this.Repository.GetAllAsync<Correlation>(null, selectors, cancellationToken: cancellationToken))
+        {
+            await this.Repository.RemoveAsync<Correlation>(correlation.GetName(), correlation.GetNamespace(), false, cancellationToken).ConfigureAwait(false);
+        }
     }
 
     /// <summary>

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

@@ -78,6 +78,10 @@ public virtual async Task ScheduleAsync(CancellationToken cancellationToken = de
                 {
                     Namespace = this.Workflow.Resource.GetNamespace(),
                     Name = correlationName,
+                    Labels = new Dictionary<string, string>()
+                    {
+                        { SynapseDefaults.Resources.Labels.Workflow, this.Workflow.Resource.GetQualifiedName() }
+                    }
                 },
                 Spec = new()
                 {

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

@@ -235,7 +235,11 @@ public virtual async Task<CorrelationContext> CorrelateAsync(ITaskExecutionConte
                 Metadata = new()
                 {
                     Namespace = @namespace,
-                    Name = name
+                    Name = name,
+                    Labels = new Dictionary<string, string>()
+                    {
+                        { SynapseDefaults.Resources.Labels.WorkflowInstance, this.Instance.GetQualifiedName() }
+                    }
                 },
                 Spec = new()
                 {