WIP

Stidsborg · Stidsborg · commit 9e6fc6fe05d6 · 2025-06-05T11:45:12.000+02:00
diff --git a/Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/LeaseUpdaterTests/LeaseUpdaterTestFunctionStore.cs b/Core/Cleipnir.ResilientFunctions.Tests/InMemoryTests/LeaseUpdaterTests/LeaseUpdaterTestFunctionStore.cs
@@ -25,6 +25,7 @@ public class LeaseUpdaterTestFunctionStore : IFunctionStore
     public Utilities Utilities => _inner.Utilities;
     public IMigrator Migrator => _inner.Migrator;
     public ISemaphoreStore SemaphoreStore => _inner.SemaphoreStore;
+    public IReplicaStore ReplicaStore => _inner.ReplicaStore;
     public Task Initialize() => _inner.Initialize();
 
     public Task<bool> CreateFunction(
diff --git a/Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/ReplicaStoreTests.cs b/Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/ReplicaStoreTests.cs
@@ -0,0 +1,73 @@
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using Cleipnir.ResilientFunctions.Helpers;
+using Cleipnir.ResilientFunctions.Storage;
+using Cleipnir.ResilientFunctions.Tests.Utils;
+using Shouldly;
+
+namespace Cleipnir.ResilientFunctions.Tests.TestTemplates;
+
+public abstract class ReplicaStoreTests
+{
+    public abstract Task SunshineScenarioTest();
+    protected async Task SunshineScenarioTest(Task<IFunctionStore> storeTask)
+    {
+        var store = await storeTask.SelectAsync(s => s.ReplicaStore);
+        await store.GetAll().ShouldBeEmptyAsync();
+        var replicaId1 = Guid.NewGuid();        
+        var replicaId2 = Guid.NewGuid();
+        
+        {
+            await store.Insert(replicaId1);
+            var all = await store.GetAll();
+            all.Count.ShouldBe(1);
+            var stored = all.Single();
+            stored.ReplicaId.ShouldBe(replicaId1);
+            stored.Heartbeat.ShouldBe(0);            
+        }
+
+        {
+            await store.Insert(replicaId2);
+            var all = await store.GetAll();
+            all.Count.ShouldBe(2);
+            var stored = all.Single(id => id.ReplicaId == replicaId2);
+            stored.ReplicaId.ShouldBe(replicaId2);
+            stored.Heartbeat.ShouldBe(0);            
+        }
+
+        await store.UpdateHeartbeat(replicaId1);
+        {
+            var all = await store.GetAll();
+            all.Count.ShouldBe(2);
+            var stored1 = all.Single(r => r.ReplicaId == replicaId1);
+            stored1.Heartbeat.ShouldBe(1);
+            var stored2 = all.Single(r => r.ReplicaId == replicaId2);
+            stored2.Heartbeat.ShouldBe(0);
+        }
+        
+        await store.UpdateHeartbeat(replicaId2);
+        {
+            var all = await store.GetAll();
+            all.Count.ShouldBe(2);
+            var stored1 = all.Single(r => r.ReplicaId == replicaId1);
+            stored1.Heartbeat.ShouldBe(1);
+            var stored2 = all.Single(r => r.ReplicaId == replicaId2);
+            stored2.Heartbeat.ShouldBe(1);
+        }
+
+        await store.Delete(replicaId1);
+        {
+            var all = await store.GetAll();
+            all.Count.ShouldBe(1);
+            var stored2 = all.Single(r => r.ReplicaId == replicaId2);
+            stored2.Heartbeat.ShouldBe(1);
+        }
+        
+        await store.Delete(replicaId2);
+        {
+            var all = await store.GetAll();
+            all.ShouldBeEmpty();
+        }
+    }
+}
diff --git a/Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/WatchDogsTests/CrashableFunctionStore.cs b/Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/WatchDogsTests/CrashableFunctionStore.cs
@@ -25,6 +25,7 @@ public class CrashableFunctionStore : IFunctionStore
     public Utilities Utilities => _crashed ? throw new TimeoutException() : _inner.Utilities;
     public IMigrator Migrator => _crashed ? throw new TimeoutException() : _inner.Migrator;
     public ISemaphoreStore SemaphoreStore => _crashed ? throw new TimeoutException() : _inner.SemaphoreStore;
+    public IReplicaStore ReplicaStore => _crashed ? throw new TimeoutException() : _inner.ReplicaStore;
 
     public CrashableFunctionStore(IFunctionStore inner)
     {
diff --git a/Core/Cleipnir.ResilientFunctions/Storage/IFunctionStore.cs b/Core/Cleipnir.ResilientFunctions/Storage/IFunctionStore.cs
@@ -16,6 +16,7 @@ public interface IFunctionStore
     public Utilities Utilities { get; }
     public IMigrator Migrator { get; }
     public ISemaphoreStore SemaphoreStore { get; }
+    public IReplicaStore ReplicaStore { get; }
     public Task Initialize();
     
     Task<bool> CreateFunction(
diff --git a/Core/Cleipnir.ResilientFunctions/Storage/IReplicaStore.cs b/Core/Cleipnir.ResilientFunctions/Storage/IReplicaStore.cs
@@ -0,0 +1,14 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+
+namespace Cleipnir.ResilientFunctions.Storage;
+
+public interface IReplicaStore
+{
+    public Task Initialize();
+    public Task Insert(Guid replicaId);
+    public Task Delete(Guid replicaId);
+    public Task UpdateHeartbeat(Guid replicaId);
+    public Task<IReadOnlyList<StoredReplica>> GetAll();
+}
diff --git a/Core/Cleipnir.ResilientFunctions/Storage/InMemoryFunctionStore.cs b/Core/Cleipnir.ResilientFunctions/Storage/InMemoryFunctionStore.cs
@@ -1,4 +1,5 @@
-﻿using System.Collections.Generic;
+﻿using System;
+using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
@@ -30,6 +31,7 @@ public class InMemoryFunctionStore : IFunctionStore, IMessageStore
     
     public IMigrator Migrator { get; } = new InMemoryMigrator();
     public ISemaphoreStore SemaphoreStore { get; } = new InMemorySemaphoreStore();
+    public IReplicaStore ReplicaStore => throw new NotImplementedException(); //todo
 
     public Task Initialize() => Task.CompletedTask;
 
diff --git a/Core/Cleipnir.ResilientFunctions/Storage/Types.cs b/Core/Cleipnir.ResilientFunctions/Storage/Types.cs
@@ -137,4 +137,6 @@ public static class StoredEffectExtensions
 {
     public static StoredEffectChange ToStoredChange(this StoredEffect effect, StoredId storedId, CrudOperation operation) 
         => new(storedId, effect.StoredEffectId, operation, effect);
-}
+}
+
+public record StoredReplica(Guid ReplicaId, int Heartbeat);
diff --git a/Samples/Sample.ConsoleApp/Utils/CrashableFunctionStore.cs b/Samples/Sample.ConsoleApp/Utils/CrashableFunctionStore.cs
@@ -21,6 +21,7 @@ public class CrashableFunctionStore : IFunctionStore
     public Utilities Utilities => _inner.Utilities;
     public IMigrator Migrator => _inner.Migrator;
     public ISemaphoreStore SemaphoreStore => _inner.SemaphoreStore;
+    public IReplicaStore ReplicaStore => _inner.ReplicaStore;
 
     public CrashableFunctionStore(IFunctionStore inner) => _inner = inner;
 
diff --git a/Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB.Tests/ReplicaStoreTests.cs b/Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB.Tests/ReplicaStoreTests.cs
@@ -0,0 +1,11 @@
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+namespace Cleipnir.ResilientFunctions.MariaDb.Tests;
+
+[TestClass]
+public class ReplicaStoreTests : ResilientFunctions.Tests.TestTemplates.ReplicaStoreTests
+{
+    [TestMethod]
+    public override Task SunshineScenarioTest() 
+        => SunshineScenarioTest(FunctionStoreFactory.Create());
+}
diff --git a/Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB/MariaDbFunctionStore.cs b/Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB/MariaDbFunctionStore.cs
@@ -36,6 +36,9 @@ public class MariaDbFunctionStore : IFunctionStore
     private readonly MariaDbSemaphoreStore _semaphoreStore;
     public ISemaphoreStore SemaphoreStore => _semaphoreStore;
 
+    private readonly MariaDbReplicaStore _replicaStore;
+    public IReplicaStore ReplicaStore => _replicaStore;
+
     public Utilities Utilities { get; }
     private readonly MariaDbUnderlyingRegister _mariaDbUnderlyingRegister;
 
@@ -57,6 +60,7 @@ public MariaDbFunctionStore(string connectionString, string tablePrefix = "")
         _mariaDbUnderlyingRegister = new MariaDbUnderlyingRegister(connectionString, tablePrefix);
         _typeStore = new MariaDbTypeStore(connectionString, tablePrefix);
         _migrator  = new MariaDbMigrator(connectionString, tablePrefix);
+        _replicaStore = new MariaDbReplicaStore(connectionString, tablePrefix);
         
         Utilities = new Utilities(_mariaDbUnderlyingRegister);
     }
@@ -75,6 +79,7 @@ public async Task Initialize()
         await _semaphoreStore.Initialize();
         await TimeoutStore.Initialize();
         await _typeStore.Initialize();
+        await _replicaStore.Initialize();
         await using var conn = await CreateOpenConnection(_connectionString);
         _initializeSql ??= $@"
             CREATE TABLE IF NOT EXISTS {_tablePrefix} (
@@ -108,6 +113,7 @@ public async Task TruncateTables()
         await _correlationStore.Truncate();
         await _semaphoreStore.Truncate();
         await _typeStore.Truncate();
+        await _replicaStore.Truncate();
         
         await using var conn = await CreateOpenConnection(_connectionString);
         _truncateTablesSql ??= $"TRUNCATE TABLE {_tablePrefix}";
diff --git a/Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB/MariaDbReplicaStore.cs b/Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB/MariaDbReplicaStore.cs
@@ -0,0 +1,116 @@
+using Cleipnir.ResilientFunctions.Storage;
+using MySqlConnector;
+
+namespace Cleipnir.ResilientFunctions.MariaDb;
+
+public class MariaDbReplicaStore(string connectionString, string tablePrefix) : IReplicaStore
+{
+    private string? _initializeSql;
+    public async Task Initialize()
+    {
+        _initializeSql ??= $@"
+            CREATE TABLE IF NOT EXISTS {tablePrefix}_replicas (
+                id CHAR(32) PRIMARY KEY,
+                heartbeat INT
+            );";
+        await using var conn = await CreateConnection();
+        var command = new MySqlCommand(_initializeSql, conn);
+        await command.ExecuteNonQueryAsync();
+    }
+
+    private string? _insertSql;
+    public async Task Insert(Guid replicaId)
+    {
+        _insertSql ??= $@"
+            INSERT INTO {tablePrefix}_replicas
+                (id, heartbeat)
+            VALUES
+                (?, 0)";
+        
+        await using var conn = await CreateConnection();
+        await using var command = new MySqlCommand(_insertSql, conn)
+        {
+            Parameters =
+            {
+                new() {Value = replicaId.ToString("N")}
+            }
+        };
+
+        await command.ExecuteNonQueryAsync();
+    }
+
+    private string? _deleteSql;
+    public async Task Delete(Guid replicaId)
+    {
+        _deleteSql ??= $"DELETE FROM {tablePrefix}_replicas WHERE id = ?";
+        
+        await using var conn = await CreateConnection();
+        await using var command = new MySqlCommand(_deleteSql, conn)
+        {
+            Parameters =
+            {
+                new() {Value = replicaId.ToString("N")}
+            }
+        };
+
+        await command.ExecuteNonQueryAsync();
+    }
+
+    private string? _updateHeartbeatSql;
+    public async Task UpdateHeartbeat(Guid replicaId)
+    {
+        _updateHeartbeatSql ??= $@"
+            UPDATE {tablePrefix}_replicas
+            SET heartbeat = heartbeat + 1
+            WHERE id = ?";
+        
+        await using var conn = await CreateConnection();
+        await using var command = new MySqlCommand(_updateHeartbeatSql, conn)
+        {
+            Parameters =
+            {
+                new() {Value = replicaId.ToString("N")}
+            }
+        };
+
+        await command.ExecuteNonQueryAsync();
+    }
+
+    private string? _getAllSql;
+    public async Task<IReadOnlyList<StoredReplica>> GetAll()
+    {
+        _getAllSql ??= $"SELECT id, heartbeat FROM {tablePrefix}_replicas";
+        
+        await using var conn = await CreateConnection();
+        await using var command = new MySqlCommand(_getAllSql, conn);
+        var storedReplicas = new List<StoredReplica>();
+
+        await using var reader = await command.ExecuteReaderAsync();
+        while (await reader.ReadAsync())
+        {
+            var id = Guid.Parse(reader.GetString(0));
+            var heartbeat = reader.GetInt32(1);
+            storedReplicas.Add(new StoredReplica(id, heartbeat));
+        }
+
+        return storedReplicas;
+    }
+
+    private string? _truncateSql;
+    public async Task Truncate()
+    {
+        _truncateSql ??= $"TRUNCATE TABLE {tablePrefix}_replicas";
+
+        await using var conn = await CreateConnection();
+        await using var command = new MySqlCommand(_truncateSql, conn);
+
+        await command.ExecuteNonQueryAsync();
+    }
+    
+    private async Task<MySqlConnection> CreateConnection()
+    {
+        var conn = new MySqlConnection(connectionString);
+        await conn.OpenAsync();
+        return conn;
+    }
+}
diff --git a/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlFunctionStore.cs b/Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL/PostgreSqlFunctionStore.cs
@@ -34,6 +34,7 @@ public class PostgreSqlFunctionStore : IFunctionStore
     
     private readonly PostgreSqlSemaphoreStore _semaphoreStore;
     public ISemaphoreStore SemaphoreStore => _semaphoreStore;
+    public IReplicaStore ReplicaStore => throw new NotImplementedException();
 
     public Utilities Utilities { get; }
     public IMigrator Migrator => _migrator;
diff --git a/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer/SqlServerFunctionStore.cs b/Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer/SqlServerFunctionStore.cs
@@ -34,6 +34,7 @@ public class SqlServerFunctionStore : IFunctionStore
     public IMigrator Migrator => _migrator;
     private readonly SqlServerSemaphoreStore _semaphoreStore;
     public ISemaphoreStore SemaphoreStore => _semaphoreStore;
+    public IReplicaStore ReplicaStore => throw new NotImplementedException(); //todo
 
     private readonly SqlServerUnderlyingRegister _underlyingRegister;
     

Original file line number	Diff line number	Diff line change
`@@ -25,6 +25,7 @@ public class CrashableFunctionStore : IFunctionStore`
`25`	`25`	`public Utilities Utilities => _crashed ? throw new TimeoutException() : _inner.Utilities;`
`26`	`26`	`public IMigrator Migrator => _crashed ? throw new TimeoutException() : _inner.Migrator;`
`27`	`27`	`public ISemaphoreStore SemaphoreStore => _crashed ? throw new TimeoutException() : _inner.SemaphoreStore;`
	`28`	`+ public IReplicaStore ReplicaStore => _crashed ? throw new TimeoutException() : _inner.ReplicaStore;`
`28`	`29`
`29`	`30`	`public CrashableFunctionStore(IFunctionStore inner)`
`30`	`31`	`{`
Original file line number	Diff line number	Diff line change
`@@ -137,4 +137,6 @@ public static class StoredEffectExtensions`
`137`	`137`	`{`
`138`	`138`	`public static StoredEffectChange ToStoredChange(this StoredEffect effect, StoredId storedId, CrudOperation operation)`
`139`	`139`	`=> new(storedId, effect.StoredEffectId, operation, effect);`
`140`		`-}`
	`140`	`+}`
	`141`	`+`
	`142`	`+public record StoredReplica(Guid ReplicaId, int Heartbeat);`