Add InterruptedWatchdog scaffolding

The watchdog polls GetInterruptedFunctions and calls FlowsManager.Interrupt
with ids owned by this replica (i.e. present in FlowsManager's in-memory dict).

FlowState.Interrupt and FlowsManager.Interrupt are stubbed for now: the
existing implementations raced with SuspendFunction's Interrupted-flag check
(clearing the flag mid-flight caused suspended-vs-postponed-status to be
decided wrong, leaving flows stuck). Adds FilterOwned helper.

Author: stidsborg

Core/Cleipnir.ResilientFunctions/CoreRuntime/FlowState.cs

@@ -62,13 +62,6 @@ public bool TryResumeSubflow()
 
     public void Interrupt()
     {
-        lock (_lock)
-            if (Suspended)
-                return;
-            else
-                WaitingSubflows = 0;
-        
-        InterruptSignal.Fire();
     }
 
     public bool Suspend()

Core/Cleipnir.ResilientFunctions/CoreRuntime/FlowsManager.cs

@@ -1,4 +1,5 @@
 using System.Collections.Generic;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using Cleipnir.ResilientFunctions.Storage;
@@ -23,14 +24,20 @@ public void RemoveFlow(StoredId id, FlowState flowState)
               _dict.Remove(id);
     }
 
-    public async Task Interrupt(IReadOnlyList<StoredId> ids)
+    public IReadOnlyList<StoredId> FilterOwned(IEnumerable<StoredId> ids)
     {
-        await functionStore.ResetInterrupted(ids);
+        lock (_lock)
+            return ids.Where(_dict.ContainsKey).ToList();
+    }
 
+    public Task Interrupt(IReadOnlyList<StoredId> ids)
+    {
         lock (_lock)
             foreach (var id in ids)
                 if (_dict.TryGetValue(id, out var flowState))
                     flowState.Interrupt();
+
+        return Task.CompletedTask;
     }
 
     public void StartThread(StoredId id)

Core/Cleipnir.ResilientFunctions/CoreRuntime/Watchdogs/InterruptedWatchdog.cs

@@ -0,0 +1,73 @@
+using System;
+using System.Threading.Tasks;
+using Cleipnir.ResilientFunctions.Domain;
+using Cleipnir.ResilientFunctions.Domain.Exceptions;
+using Cleipnir.ResilientFunctions.Helpers;
+using Cleipnir.ResilientFunctions.Storage;
+
+namespace Cleipnir.ResilientFunctions.CoreRuntime.Watchdogs;
+
+internal class InterruptedWatchdog
+{
+    private readonly IFunctionStore _functionStore;
+    private readonly FlowsManager _flowsManager;
+    private readonly ShutdownCoordinator _shutdownCoordinator;
+    private readonly UnhandledExceptionHandler _unhandledExceptionHandler;
+    private readonly TimeSpan _checkFrequency;
+    private readonly TimeSpan _delayStartUp;
+    private readonly UtcNow _utcNow;
+
+    public InterruptedWatchdog(
+        IFunctionStore functionStore,
+        FlowsManager flowsManager,
+        ShutdownCoordinator shutdownCoordinator,
+        UnhandledExceptionHandler unhandledExceptionHandler,
+        TimeSpan checkFrequency,
+        TimeSpan delayStartUp,
+        UtcNow utcNow)
+    {
+        _functionStore = functionStore;
+        _flowsManager = flowsManager;
+        _shutdownCoordinator = shutdownCoordinator;
+        _unhandledExceptionHandler = unhandledExceptionHandler;
+        _checkFrequency = checkFrequency;
+        _delayStartUp = delayStartUp;
+        _utcNow = utcNow;
+    }
+
+    public async Task Start()
+    {
+        await Task.Delay(_delayStartUp);
+
+        Start:
+        try
+        {
+            while (!_shutdownCoordinator.ShutdownInitiated)
+            {
+                var now = _utcNow();
+
+                var interrupted = await _functionStore.GetInterruptedFunctions();
+                var owned = _flowsManager.FilterOwned(interrupted);
+                if (owned.Count > 0)
+                    await _flowsManager.Interrupt(owned);
+
+                var timeElapsed = _utcNow() - now;
+                var delay = (_checkFrequency - timeElapsed).RoundUpToZero();
+
+                await Task.Delay(delay);
+            }
+        }
+        catch (Exception thrownException)
+        {
+            _unhandledExceptionHandler.Invoke(
+                new FrameworkException(
+                    $"{nameof(InterruptedWatchdog)} execution failed - retrying in 5 seconds",
+                    innerException: thrownException
+                )
+            );
+
+            await Task.Delay(5_000);
+            goto Start;
+        }
+    }
+}

Core/Cleipnir.ResilientFunctions/FunctionsRegistry.cs

@@ -31,6 +31,7 @@ public class FunctionsRegistry : IDisposable
     private volatile bool _disposed;
     private readonly Lock _sync = new();
     private readonly ReplicaWatchdog _replicaWatchdog;
+    private readonly InterruptedWatchdog _interruptedWatchdog;
     private readonly FlowsManager _flowsManager;
 
     public FunctionsRegistry(IFunctionStore functionStore, Settings? settings = null)
@@ -61,10 +62,22 @@ public FunctionsRegistry(IFunctionStore functionStore, Settings? settings = null
             utcNow, 
             _settings.UnhandledExceptionHandler
         );
+
+        _interruptedWatchdog = new InterruptedWatchdog(
+            _functionStore,
+            _flowsManager,
+            _shutdownCoordinator,
+            _settings.UnhandledExceptionHandler,
+            _settings.WatchdogCheckFrequency,
+            _settings.DelayStartup,
+            utcNow
+        );
+
         if (_settings.EnableWatchdogs)
         {
             _replicaWatchdog.Initialize().GetAwaiter().GetResult();
             _ = _replicaWatchdog.Start();            
+            _ = Task.Run(_interruptedWatchdog.Start);
         }
     }