Added flow state parameter to FlowsManager's RemoveFlow-method

Author: stidsborg

Core/Cleipnir.ResilientFunctions/CoreRuntime/FlowsManager.cs

@@ -50,10 +50,11 @@ public FlowState AddFlow(StoredId id, QueueManager queueManager, FlowTimeouts ti
             return _dict[id] = new FlowState(id, queueManager, threads: 1, waitingThreads: 0, timeouts);
     }
 
-    public void RemoveFlow(StoredId id)
+    public void RemoveFlow(StoredId id, FlowState flowState)
     {
         lock (_lock)
-            _dict.Remove(id);
+            if (_dict.TryGetValue(id, out var existingState) && flowState == existingState)
+              _dict.Remove(id);
     }
 
     public async Task Interrupt(IReadOnlyList<StoredId> ids)

Core/Cleipnir.ResilientFunctions/CoreRuntime/Invocation/Invoker.cs

@@ -118,7 +118,7 @@ await PersistResultAndEnsureSuccess(
             }
             finally
             {
-                _flowsManager.RemoveFlow(storedId);
+                _flowsManager.RemoveFlow(storedId, flowState);
             }
         });
 
@@ -208,7 +208,7 @@ public async Task<InnerScheduled<TReturn>> ScheduleRestart(StoredId storedId)
             }
             finally
             {
-                _flowsManager.RemoveFlow(storedId);
+                _flowsManager.RemoveFlow(storedId, flowState);
             }
         });
 
@@ -250,7 +250,7 @@ internal async Task ScheduleRestart(StoredId storedId, RestartedFunction rf, Act
                     tcs.TrySetCanceled();
             }
             catch (Exception exception) { _unhandledExceptionHandler.Invoke(_flowType, exception); tcs.TrySetException(exception); }
-            finally{ _flowsManager.RemoveFlow(storedId); onCompletion(); }
+            finally{ _flowsManager.RemoveFlow(storedId, flowState); onCompletion(); }
         });
     }