Add ResetInterrupted method to function stores

Add ResetInterrupted(IReadOnlyList<StoredId>) to IFunctionStore and all
implementations to allow clearing the interrupted flag for a set of stored IDs.

Author: stidsborg

Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/WatchDogsTests/CrashableFunctionStore.cs

@@ -187,6 +187,8 @@ public Task<bool> Interrupt(StoredId storedId)
 
     public Task Interrupt(IReadOnlyList<StoredId> storedIds) => _inner.Interrupt(storedIds);
 
+    public Task ResetInterrupted(IReadOnlyList<StoredId> storedIds) => _inner.ResetInterrupted(storedIds);
+
     public Task<bool?> Interrupted(StoredId storedId) 
         => _crashed
             ? Task.FromException<bool?>(new TimeoutException())

Core/Cleipnir.ResilientFunctions/Storage/IFunctionStore.cs

@@ -104,7 +104,8 @@ Task<bool> SuspendFunction(
 
     Task<bool> Interrupt(StoredId storedId);
     Task Interrupt(IReadOnlyList<StoredId> storedIds);
-    Task<bool?> Interrupted(StoredId storedId); 
+    Task ResetInterrupted(IReadOnlyList<StoredId> storedIds);
+    Task<bool?> Interrupted(StoredId storedId);
 
     Task<Status?> GetFunctionStatus(StoredId storedId);
     Task<IReadOnlyList<StatusAndId>> GetFunctionsStatus(IEnumerable<StoredId> storedIds);

Core/Cleipnir.ResilientFunctions/Storage/InMemoryFunctionStore.cs

@@ -450,6 +450,18 @@ public async Task Interrupt(IReadOnlyList<StoredId> storedIds)
             await Interrupt(storedId);
     }
 
+    public Task ResetInterrupted(IReadOnlyList<StoredId> storedIds)
+    {
+        lock (_sync)
+        {
+            foreach (var storedId in storedIds)
+                if (_states.TryGetValue(storedId, out var state))
+                    state.Interrupted = false;
+        }
+
+        return Task.CompletedTask;
+    }
+
     public Task<bool?> Interrupted(StoredId storedId)
     {
         lock (_sync)

Samples/Sample.ConsoleApp/Utils/CrashableFunctionStore.cs

@@ -170,6 +170,11 @@ public Task Interrupt(IReadOnlyList<StoredId> storedIds)
             ? Task.FromException(new TimeoutException())
             : _inner.Interrupt(storedIds);
 
+    public Task ResetInterrupted(IReadOnlyList<StoredId> storedIds)
+        => _crashed
+            ? Task.FromException(new TimeoutException())
+            : _inner.ResetInterrupted(storedIds);
+
     public Task<bool?> Interrupted(StoredId storedId)
         => _crashed
             ? Task.FromException<bool?>(new TimeoutException())

Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB/MariaDbFunctionStore.cs

@@ -670,12 +670,22 @@ public async Task Interrupt(IReadOnlyList<StoredId> storedIds)
     {
         if (storedIds.Count == 0)
             return;
-        
+
         await using var conn = await CreateOpenConnection(_connectionString);
         await using var cmd = _sqlGenerator.Interrupt(storedIds).ToSqlCommand(conn);
         await cmd.ExecuteNonQueryAsync();
     }
 
+    public async Task ResetInterrupted(IReadOnlyList<StoredId> storedIds)
+    {
+        if (storedIds.Count == 0)
+            return;
+
+        await using var conn = await CreateOpenConnection(_connectionString);
+        await using var cmd = _sqlGenerator.ResetInterrupted(storedIds).ToSqlCommand(conn);
+        await cmd.ExecuteNonQueryAsync();
+    }
+
     private string? _setParametersSql;
     public async Task<bool> SetParameters(
         StoredId storedId,

Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB/SqlGenerator.cs

@@ -33,7 +33,17 @@ ELSE expires
 
         return StoreCommand.Create(sql);
     }
-    
+
+    public StoreCommand ResetInterrupted(IEnumerable<StoredId> storedIds)
+    {
+        var sql = @$"
+                UPDATE {tablePrefix}
+                SET interrupted = FALSE
+                WHERE Id IN ({storedIds.Select(id => $"'{id.AsGuid:N}'").StringJoin(", ")});";
+
+        return StoreCommand.Create(sql);
+    }
+
     private string? _getEffectResultsSql;
     public StoreCommand GetEffects(StoredId storedId)
     {

Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlFunctionStore.cs

@@ -668,12 +668,22 @@ public async Task Interrupt(IReadOnlyList<StoredId> storedIds)
     {
         if (storedIds.Count == 0)
             return;
-        
+
         await using var conn = await CreateConnection();
         await using var command = _sqlGenerator.Interrupt(storedIds).ToNpgsqlCommand(conn);
         await command.ExecuteNonQueryAsync();
     }
 
+    public async Task ResetInterrupted(IReadOnlyList<StoredId> storedIds)
+    {
+        if (storedIds.Count == 0)
+            return;
+
+        await using var conn = await CreateConnection();
+        await using var command = _sqlGenerator.ResetInterrupted(storedIds).ToNpgsqlCommand(conn);
+        await command.ExecuteNonQueryAsync();
+    }
+
     private string? _interruptedSql;
     public async Task<bool?> Interrupted(StoredId storedId)
     {

Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/SqlGenerator.cs

@@ -37,7 +37,17 @@ ELSE expires
 
         return StoreCommand.Create(sql, values: [ storedIds.Select(id => id.AsGuid).ToArray() ]);
     }
-    
+
+    public StoreCommand ResetInterrupted(IEnumerable<StoredId> storedIds)
+    {
+        var sql = @$"
+                UPDATE {tablePrefix}
+                SET interrupted = FALSE
+                WHERE Id = ANY($1)";
+
+        return StoreCommand.Create(sql, values: [ storedIds.Select(id => id.AsGuid).ToArray() ]);
+    }
+
     private string? _getEffectResultsSql;
     public StoreCommand GetEffects(StoredId storedId)
     {

Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer/SqlGenerator.cs

@@ -38,7 +38,17 @@ ELSE Expires
 
         return StoreCommand.Create(sql);
     }
-    
+
+    public StoreCommand ResetInterrupted(IEnumerable<StoredId> storedIds)
+    {
+        var sql = @$"
+                UPDATE {tablePrefix}
+                SET Interrupted = 0
+                WHERE Id IN ({storedIds.Select(id => $"'{id.AsGuid}'").StringJoin(", ")});";
+
+        return StoreCommand.Create(sql);
+    }
+
     public StoreCommand InsertEffects(StoredId storedId, IReadOnlyList<StoredEffectChange> changes, SnapshotStorageSession session, string paramPrefix)
     {
         foreach (var change in changes)

Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer/SqlServerFunctionStore.cs

@@ -730,12 +730,22 @@ public async Task Interrupt(IReadOnlyList<StoredId> storedIds)
     {
         if (storedIds.Count == 0)
             return;
-        
+
         await using var conn = await _connFunc();
         await using var cmd = _sqlGenerator.Interrupt(storedIds).ToSqlCommand(conn);
         await cmd.ExecuteNonQueryAsync();
     }
 
+    public async Task ResetInterrupted(IReadOnlyList<StoredId> storedIds)
+    {
+        if (storedIds.Count == 0)
+            return;
+
+        await using var conn = await _connFunc();
+        await using var cmd = _sqlGenerator.ResetInterrupted(storedIds).ToSqlCommand(conn);
+        await cmd.ExecuteNonQueryAsync();
+    }
+
     private string? _setParametersSql;
     public async Task<bool> SetParameters(
         StoredId storedId,