PsQueue fix OnProcess throws exception to propagate

qdr.fnd.core.pqueue.test/Process/Mock/PsProcessProviderBaseMock.cs

@@ -25,7 +25,9 @@ namespace qdr.fnd.core.pqueue.test.Process.Mock
             {
                 var message = GetArgumentValue<string>(CS_ARG_EMIT_EXC);
                 Log.Log(LogSeverityEnum.Info, $"Emmiting exception '{message}'");
-                throw new Exception("Test exception");
+                //await Task.FromException(new Exception(message));
+                throw new Exception(message);
+               
             }
             else
             {

qdr.fnd.core.pqueue.test/Process/PsProcessProviderBaseTest.cs

@@ -188,6 +188,26 @@ namespace qdr.fnd.core.pqueue.test.Process
             Assert.That(_statusDonePool[process.ItemId].Any(x => x.Item1 == PsStatusEnum.DoneFailed), Is.True);
         }
 
+        [Test(TestOf=typeof(PsProcessProviderBase))]
+        public void ProcessFailedInsideOk()
+        {
+            var inProcessMessage = "Foo wants to live";
+            var args = PsProcessProviderBaseMock.CreateArguments().ToArray();
+            args.SetArgumentValue<TimeSpan>(PsProcessProviderBase.CS_ARG_PROCESS_TIMEOUT, TimeSpan.FromSeconds(500));
+            args.SetArgumentValue<string>(PsProcessProviderBaseMock.CS_ARG_EMIT_EXC, inProcessMessage);
+
+
+            var process = CreateProcess(args);
+            var dtStart = DateTime.Now;
+            process.Start();
+            process.WaitToDone();
+            TestContext.WriteLine($"#Process duration check: {(DateTime.Now - dtStart).ToReadableString()}");
+
+            Assert.That(process.IsRunning, Is.False);
+            Assert.That(_statusDonePool.ContainsKey(process.ItemId), Is.True);
+            Assert.That(_statusDonePool[process.ItemId].Any(x => x.Item1 == PsStatusEnum.DoneFailed && x.Item2.Contains(inProcessMessage)), Is.True);
+        }
+
         private PsProcessProviderBaseMock CreateProcess(TypedArgument[] args)
         {
             var itemId = "a";

qdr.fnd.core.pqueue/Process/PsProcessProviderBase.cs

@@ -92,7 +92,20 @@ namespace qdr.fnd.core.pqueue.Process
 
                         await OnProcess().ContinueWith(task =>
                         {
-                            ProcessDoneOk("Succesfuly done.");
+                            if (task.IsFaulted)
+                            {
+                                var e = task.Exception!;
+                                OnProcessError(e).ContinueWith(taskInner =>
+                                    {
+                                        if (taskInner.IsFaulted)
+                                            throw taskInner.Exception!;
+                                        ProcessDoneFailed(e.Message, e);
+                                    });
+                            }
+                            else
+                            {
+                                ProcessDoneOk("Succesfuly done.");
+                            }
                         });
                     }, timeout);
                 }
@@ -100,6 +113,8 @@ namespace qdr.fnd.core.pqueue.Process
                 {
                     await OnProcessError(e).ContinueWith(task =>
                     {
+                        if (task.IsFaulted)
+                            throw task.Exception!;
                         ProcessDoneFailed(e.Message, e);
                     });
                 }

qdr.fnd.core.pqueue/fnd.candidates/TimeoutUtils.cs

@@ -21,6 +21,9 @@
                     throw new TimeoutException($"The operation has timed out after {timeout.TotalSeconds} seconds.");
                 }
             }
+            catch(Exception){
+                throw;
+            }
             finally
             {
                 cts.Cancel(); // Ensure we always cancel the CancellationTokenSource