Merge branch 'master' into copilot/fix-1375984f-ce0b-4c23-bedb-b0cdd97306e2

danielgerlag · web-flow · commit b4660fbbf2b2 · 2025-09-17T09:52:31.000-07:00
diff --git a/src/WorkflowCore/Services/BackgroundTasks/QueueConsumer.cs b/src/WorkflowCore/Services/BackgroundTasks/QueueConsumer.cs
@@ -122,7 +122,7 @@ private async Task Execute()
                 catch (Exception ex)
                 {
                     Logger.LogError(ex, ex.Message);
-                    activity?.RecordException(ex);
+                    activity?.AddException(ex);
                 }
                 finally
                 {
@@ -158,7 +158,7 @@ private async Task ExecuteItem(string itemId, EventWaitHandle waitHandle, Activi
             catch (Exception ex)
             {
                 Logger.LogError(default(EventId), ex, $"Error executing item {itemId} - {ex.Message}");
-                activity?.RecordException(ex);
+                activity?.AddException(ex);
             }
             finally
             {
diff --git a/src/WorkflowCore/Services/BackgroundTasks/RunnablePoller.cs b/src/WorkflowCore/Services/BackgroundTasks/RunnablePoller.cs
@@ -86,7 +86,7 @@ await _persistenceStore.ScheduleCommand(new ScheduledCommand()
                                 catch (Exception ex)
                                 {
                                     _logger.LogError(ex, ex.Message);
-                                    activity?.RecordException(ex);
+                                    activity?.AddException(ex);
                                 }
                             }
                             if (_greylist.Contains($"wf:{item}"))
@@ -108,7 +108,7 @@ await _persistenceStore.ScheduleCommand(new ScheduledCommand()
             catch (Exception ex)
             {
                 _logger.LogError(ex, ex.Message);
-                activity?.RecordException(ex);
+                activity?.AddException(ex);
             }
             finally
             {
@@ -145,7 +145,7 @@ await _persistenceStore.ScheduleCommand(new ScheduledCommand()
                                 catch (Exception ex)
                                 {
                                     _logger.LogError(ex, ex.Message);
-                                    activity?.RecordException(ex);
+                                    activity?.AddException(ex);
                                 }
                             }
                             if (_greylist.Contains($"evt:{item}"))
@@ -167,7 +167,7 @@ await _persistenceStore.ScheduleCommand(new ScheduledCommand()
             catch (Exception ex)
             {
                 _logger.LogError(ex, ex.Message);
-                activity?.RecordException(ex);
+                activity?.AddException(ex);
             }
             finally
             {
@@ -210,7 +210,7 @@ await _persistenceStore.ProcessCommands(new DateTimeOffset(_dateTimeProvider.Utc
             catch (Exception ex)
             {
                 _logger.LogError(ex, ex.Message);
-                activity?.RecordException(ex);
+                activity?.AddException(ex);
             }
             finally
             {
diff --git a/src/WorkflowCore/Services/BackgroundTasks/WorkflowConsumer.cs b/src/WorkflowCore/Services/BackgroundTasks/WorkflowConsumer.cs
@@ -8,6 +8,11 @@
 
 namespace WorkflowCore.Services.BackgroundTasks
 {
+    /// <summary>
+    /// Background task responsible for consuming workflow items from the queue and processing them.
+    /// This consumer ensures that workflows are removed from the greylist after processing,
+    /// regardless of their status, to prevent workflows from getting stuck in "Pending" state.
+    /// </summary>
     internal class WorkflowConsumer : QueueConsumer, IBackgroundTask
     {
         private readonly IDistributedLockProvider _lockProvider;
@@ -57,12 +62,25 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance
                         WorkflowActivity.Enrich(result);
                         await _persistenceStore.PersistWorkflow(workflow, result?.Subscriptions, cancellationToken);
                         await QueueProvider.QueueWork(itemId, QueueType.Index);
-                        _greylist.Remove($"wf:{itemId}");
                     }
                 }
+                else
+                {
+                    Logger.LogDebug("Workflow {ItemId} is not runnable, status: {Status}", itemId, workflow.Status);
+                }
+            }
+            catch (Exception ex)
+            {
+                Logger.LogError(ex, "Error processing workflow {ItemId}", itemId);
+                throw;
             }
             finally
             {
+                // Always remove from greylist regardless of workflow status
+                // This prevents workflows from being stuck in greylist when they can't be processed
+                Logger.LogDebug("Removing workflow {ItemId} from greylist", itemId);
+                _greylist.Remove($"wf:{itemId}");
+                
                 await _lockProvider.ReleaseLock(itemId);
                 if ((workflow != null) && (result != null))
                 {
diff --git a/src/WorkflowCore/Services/WorkflowActivity.cs b/src/WorkflowCore/Services/WorkflowActivity.cs
@@ -83,7 +83,6 @@ internal static void Enrich(WorkflowExecutorResult result)
 
                 if (result?.Errors?.Count > 0)
                 {
-                    activity.SetStatus(Status.Error);
                     activity.SetStatus(ActivityStatusCode.Error);
                 }
             }
diff --git a/src/WorkflowCore/Services/WorkflowHost.cs b/src/WorkflowCore/Services/WorkflowHost.cs
@@ -105,7 +105,7 @@ public async Task StartAsync(CancellationToken cancellationToken)
             }
             catch (Exception ex)
             {
-                activity.RecordException(ex);
+                activity.AddException(ex);
                 throw;
             }
             finally
diff --git a/src/WorkflowCore/WorkflowCore.csproj b/src/WorkflowCore/WorkflowCore.csproj
@@ -26,8 +26,8 @@
     <PackageReference Include="System.Reflection.TypeExtensions" Version="4.7.0" />
     <PackageReference Include="System.Threading.Thread" Version="4.3.0" />
     <PackageReference Include="System.Linq.Queryable" Version="4.3.0" />
-    <PackageReference Include="OpenTelemetry.Api" Version="1.1.0" />
-    <PackageReference Include="System.Diagnostics.DiagnosticSource" Version="6.0.0" />
+    <PackageReference Include="OpenTelemetry.Api" Version="1.12.0" />
+    <PackageReference Include="System.Diagnostics.DiagnosticSource" Version="9.0.0" />
 	<AssemblyAttribute Include="System.Runtime.CompilerServices.InternalsVisibleTo">
 		<_Parameter1>WorkflowCore.IntegrationTests</_Parameter1>
 	</AssemblyAttribute>
diff --git a/test/WorkflowCore.Tests.MySQL/WorkflowCore.Tests.MySQL.csproj b/test/WorkflowCore.Tests.MySQL/WorkflowCore.Tests.MySQL.csproj
@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
-    <TargetFrameworks>net6.0</TargetFrameworks>
+    <TargetFrameworks>net6.0;net8.0</TargetFrameworks>
   </PropertyGroup>
 
   <ItemGroup>

Original file line number	Diff line number	Diff line change
`@@ -122,7 +122,7 @@ private async Task Execute()`
`122`	`122`	`catch (Exception ex)`
`123`	`123`	`{`
`124`	`124`	`Logger.LogError(ex, ex.Message);`
`125`		`- activity?.RecordException(ex);`
	`125`	`+ activity?.AddException(ex);`
`126`	`126`	`}`
`127`	`127`	`finally`
`128`	`128`	`{`
`@@ -158,7 +158,7 @@ private async Task ExecuteItem(string itemId, EventWaitHandle waitHandle, Activi`
`158`	`158`	`catch (Exception ex)`
`159`	`159`	`{`
`160`	`160`	`Logger.LogError(default(EventId), ex, $"Error executing item {itemId} - {ex.Message}");`
`161`		`- activity?.RecordException(ex);`
	`161`	`+ activity?.AddException(ex);`
`162`	`162`	`}`
`163`	`163`	`finally`
`164`	`164`	`{`
Original file line number	Diff line number	Diff line change
`@@ -86,7 +86,7 @@ await _persistenceStore.ScheduleCommand(new ScheduledCommand()`
`86`	`86`	`catch (Exception ex)`
`87`	`87`	`{`
`88`	`88`	`_logger.LogError(ex, ex.Message);`
`89`		`- activity?.RecordException(ex);`
	`89`	`+ activity?.AddException(ex);`
`90`	`90`	`}`
`91`	`91`	`}`
`92`	`92`	`if (_greylist.Contains($"wf:{item}"))`
`@@ -108,7 +108,7 @@ await _persistenceStore.ScheduleCommand(new ScheduledCommand()`
`108`	`108`	`catch (Exception ex)`
`109`	`109`	`{`
`110`	`110`	`_logger.LogError(ex, ex.Message);`
`111`		`- activity?.RecordException(ex);`
	`111`	`+ activity?.AddException(ex);`
`112`	`112`	`}`
`113`	`113`	`finally`
`114`	`114`	`{`
`@@ -145,7 +145,7 @@ await _persistenceStore.ScheduleCommand(new ScheduledCommand()`
`145`	`145`	`catch (Exception ex)`
`146`	`146`	`{`
`147`	`147`	`_logger.LogError(ex, ex.Message);`
`148`		`- activity?.RecordException(ex);`
	`148`	`+ activity?.AddException(ex);`
`149`	`149`	`}`
`150`	`150`	`}`
`151`	`151`	`if (_greylist.Contains($"evt:{item}"))`
`@@ -167,7 +167,7 @@ await _persistenceStore.ScheduleCommand(new ScheduledCommand()`
`167`	`167`	`catch (Exception ex)`
`168`	`168`	`{`
`169`	`169`	`_logger.LogError(ex, ex.Message);`
`170`		`- activity?.RecordException(ex);`
	`170`	`+ activity?.AddException(ex);`
`171`	`171`	`}`
`172`	`172`	`finally`
`173`	`173`	`{`
`@@ -210,7 +210,7 @@ await _persistenceStore.ProcessCommands(new DateTimeOffset(_dateTimeProvider.Utc`
`210`	`210`	`catch (Exception ex)`
`211`	`211`	`{`
`212`	`212`	`_logger.LogError(ex, ex.Message);`
`213`		`- activity?.RecordException(ex);`
	`213`	`+ activity?.AddException(ex);`
`214`	`214`	`}`
`215`	`215`	`finally`
`216`	`216`	`{`
Original file line number	Diff line number	Diff line change
`@@ -8,6 +8,11 @@`
`8`	`8`
`9`	`9`	`namespace WorkflowCore.Services.BackgroundTasks`
`10`	`10`	`{`
	`11`	`+ /// <summary>`
	`12`	`+ /// Background task responsible for consuming workflow items from the queue and processing them.`
	`13`	`+ /// This consumer ensures that workflows are removed from the greylist after processing,`
	`14`	`+ /// regardless of their status, to prevent workflows from getting stuck in "Pending" state.`
	`15`	`+ /// </summary>`
`11`	`16`	`internal class WorkflowConsumer : QueueConsumer, IBackgroundTask`
`12`	`17`	`{`
`13`	`18`	`private readonly IDistributedLockProvider _lockProvider;`
`@@ -57,12 +62,25 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance`
`57`	`62`	`WorkflowActivity.Enrich(result);`
`58`	`63`	`await _persistenceStore.PersistWorkflow(workflow, result?.Subscriptions, cancellationToken);`
`59`	`64`	`await QueueProvider.QueueWork(itemId, QueueType.Index);`
`60`		`- _greylist.Remove($"wf:{itemId}");`
`61`	`65`	`}`
`62`	`66`	`}`
	`67`	`+ else`
	`68`	`+ {`
	`69`	`+ Logger.LogDebug("Workflow {ItemId} is not runnable, status: {Status}", itemId, workflow.Status);`
	`70`	`+ }`
	`71`	`+ }`
	`72`	`+ catch (Exception ex)`
	`73`	`+ {`
	`74`	`+ Logger.LogError(ex, "Error processing workflow {ItemId}", itemId);`
	`75`	`+ throw;`
`63`	`76`	`}`
`64`	`77`	`finally`
`65`	`78`	`{`
	`79`	`+ // Always remove from greylist regardless of workflow status`
	`80`	`+ // This prevents workflows from being stuck in greylist when they can't be processed`
	`81`	`+ Logger.LogDebug("Removing workflow {ItemId} from greylist", itemId);`
	`82`	`+ _greylist.Remove($"wf:{itemId}");`
	`83`	`+`
`66`	`84`	`await _lockProvider.ReleaseLock(itemId);`
`67`	`85`	`if ((workflow != null) && (result != null))`
`68`	`86`	`{`
Original file line number	Diff line number	Diff line change
`@@ -83,7 +83,6 @@ internal static void Enrich(WorkflowExecutorResult result)`
`83`	`83`
`84`	`84`	`if (result?.Errors?.Count > 0)`
`85`	`85`	`{`
`86`		`- activity.SetStatus(Status.Error);`
`87`	`86`	`activity.SetStatus(ActivityStatusCode.Error);`
`88`	`87`	`}`
`89`	`88`	`}`
Original file line number	Diff line number	Diff line change
`@@ -105,7 +105,7 @@ public async Task StartAsync(CancellationToken cancellationToken)`
`105`	`105`	`}`
`106`	`106`	`catch (Exception ex)`
`107`	`107`	`{`
`108`		`- activity.RecordException(ex);`
	`108`	`+ activity.AddException(ex);`
`109`	`109`	`throw;`
`110`	`110`	`}`
`111`	`111`	`finally`