diff --git a/src/AdoNet/Orleans.Clustering.AdoNet/Messaging/AdoNetClusteringTable.cs b/src/AdoNet/Orleans.Clustering.AdoNet/Messaging/AdoNetClusteringTable.cs index a016054fe4c..7b3fb2ba96e 100644 --- a/src/AdoNet/Orleans.Clustering.AdoNet/Messaging/AdoNetClusteringTable.cs +++ b/src/AdoNet/Orleans.Clustering.AdoNet/Messaging/AdoNetClusteringTable.cs @@ -172,6 +172,12 @@ public async Task UpdateRowAsync(MembershipEntry entry, string etag, Table throw new ArgumentNullException(nameof(tableVersion)); } + // Relational reads use the table version as the row etag. + if (!string.Equals(etag, tableVersion.VersionEtag, StringComparison.Ordinal)) + { + return false; + } + try { return await orleansQueries.UpdateMembershipRowAsync(this.clusterId, entry, tableVersion.VersionEtag, cancellationToken); diff --git a/src/AdoNet/Orleans.Clustering.AdoNet/MySQL-Clustering.sql b/src/AdoNet/Orleans.Clustering.AdoNet/MySQL-Clustering.sql index 3016b97ad73..9d5f9f377c1 100644 --- a/src/AdoNet/Orleans.Clustering.AdoNet/MySQL-Clustering.sql +++ b/src/AdoNet/Orleans.Clustering.AdoNet/MySQL-Clustering.sql @@ -37,10 +37,10 @@ VALUES SET IAmAliveTime = @IAmAliveTime WHERE - DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL - AND Address = @Address AND @Address IS NOT NULL - AND Port = @Port AND @Port IS NOT NULL - AND Generation = @Generation AND @Generation IS NOT NULL; + DeploymentId = @DeploymentId + AND Address = @Address + AND Port = @Port + AND Generation = @Generation; '); INSERT INTO OrleansQuery(QueryKey, QueryText) @@ -89,7 +89,14 @@ CREATE PROCEDURE InsertMembershipKey( ) BEGIN DECLARE _ROWCOUNT INT; + DECLARE EXIT HANDLER FOR SQLEXCEPTION + BEGIN + ROLLBACK; + RESIGNAL; + END; START TRANSACTION; + + -- Preserve the row-then-version statement order used by cached legacy inserts. INSERT INTO OrleansMembershipTable ( DeploymentId, @@ -127,10 +134,8 @@ BEGIN ); UPDATE OrleansMembershipVersionTable - SET - Version = Version + 1 - WHERE - DeploymentId = _DeploymentId AND _DeploymentId IS NOT NULL + SET Version = Version + 1 + WHERE DeploymentId = _DeploymentId AND _DeploymentId IS NOT NULL AND Version = _Version AND _Version IS NOT NULL AND ROW_COUNT() > 0; @@ -151,29 +156,19 @@ INSERT INTO OrleansQuery(QueryKey, QueryText) VALUES ( 'UpdateMembershipKey',' - START TRANSACTION; - - UPDATE OrleansMembershipVersionTable - SET - Version = Version + 1 - WHERE - DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL - AND Version = @Version AND @Version IS NOT NULL; + UPDATE OrleansMembershipVersionTable v + INNER JOIN OrleansMembershipTable m ON m.DeploymentId = v.DeploymentId + SET v.Version = v.Version + 1, + m.Status = @Status, + m.SuspectTimes = @SuspectTimes, + m.IAmAliveTime = GREATEST(m.IAmAliveTime, @IAmAliveTime) + WHERE v.DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL + AND v.Version = @Version AND @Version IS NOT NULL + AND m.Address = @Address AND @Address IS NOT NULL + AND m.Port = @Port AND @Port IS NOT NULL + AND m.Generation = @Generation AND @Generation IS NOT NULL; - UPDATE OrleansMembershipTable - SET - Status = @Status, - SuspectTimes = @SuspectTimes, - IAmAliveTime = @IAmAliveTime - WHERE - DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL - AND Address = @Address AND @Address IS NOT NULL - AND Port = @Port AND @Port IS NOT NULL - AND Generation = @Generation AND @Generation IS NOT NULL - AND ROW_COUNT() > 0; - - SELECT ROW_COUNT(); - COMMIT; + SELECT ROW_COUNT() > 0; '); INSERT INTO OrleansQuery(QueryKey, QueryText) @@ -262,5 +257,18 @@ VALUES WHERE DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL AND IAmAliveTime < @IAmAliveTime - AND Status !=3; + AND StartTime < @IAmAliveTime + AND COALESCE(SuspectTimes, '''') = '''' + AND Status = 6; +'); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +VALUES +( + 'CleanupDefunctSiloEntryKey',' + DELETE FROM OrleansMembershipTable + WHERE DeploymentId = @DeploymentId AND Status = 6 + AND Address = @Address AND Port = @Port AND Generation = @Generation + AND IAmAliveTime = @IAmAliveTime AND StartTime = @StartTime + AND CAST(COALESCE(SuspectTimes, '''') AS BINARY) = CAST(COALESCE(@SuspectTimes, '''') AS BINARY); '); diff --git a/src/AdoNet/Orleans.Clustering.AdoNet/Oracle-Clustering.sql b/src/AdoNet/Orleans.Clustering.AdoNet/Oracle-Clustering.sql index 0eaa255a9b9..192141de010 100644 --- a/src/AdoNet/Orleans.Clustering.AdoNet/Oracle-Clustering.sql +++ b/src/AdoNet/Orleans.Clustering.AdoNet/Oracle-Clustering.sql @@ -36,6 +36,13 @@ CREATE OR REPLACE FUNCTION InsertMembership(PARAM_DEPLOYMENTID IN NVARCHAR2, PAR rowcount NUMBER; PRAGMA AUTONOMOUS_TRANSACTION; BEGIN + UPDATE OrleansMembershipVersionTable + SET Timestamp = sys_extract_utc(systimestamp), + Version = Version + 1 + WHERE DeploymentId = PARAM_DEPLOYMENTID AND PARAM_DEPLOYMENTID IS NOT NULL + AND Version = PARAM_VERSION AND PARAM_VERSION IS NOT NULL; + rowcount := SQL%ROWCOUNT; + INSERT INTO OrleansMembershipTable ( DeploymentId, @@ -60,7 +67,7 @@ CREATE OR REPLACE FUNCTION InsertMembership(PARAM_DEPLOYMENTID IN NVARCHAR2, PAR PARAM_PROXYPORT, PARAM_STARTTIME, PARAM_IAMALIVETIME - FROM DUAL WHERE NOT EXISTS + FROM DUAL WHERE rowcount > 0 AND NOT EXISTS ( SELECT 1 FROM OrleansMembershipTable WHERE DeploymentId = PARAM_DEPLOYMENTID AND PARAM_DEPLOYMENTID IS NOT NULL @@ -69,14 +76,6 @@ CREATE OR REPLACE FUNCTION InsertMembership(PARAM_DEPLOYMENTID IN NVARCHAR2, PAR AND Generation = PARAM_GENERATION AND PARAM_GENERATION IS NOT NULL ); rowcount := SQL%ROWCOUNT; - UPDATE OrleansMembershipVersionTable - SET Timestamp = sys_extract_utc(systimestamp), - Version = Version + 1 - WHERE - DeploymentId = PARAM_DEPLOYMENTID AND PARAM_DEPLOYMENTID IS NOT NULL - AND Version = PARAM_VERSION AND PARAM_VERSION IS NOT NULL - AND rowcount > 0; - rowcount := SQL%ROWCOUNT; IF rowcount = 0 THEN ROLLBACK; ELSE @@ -88,6 +87,9 @@ CREATE OR REPLACE FUNCTION InsertMembership(PARAM_DEPLOYMENTID IN NVARCHAR2, PAR ELSE RETURN(0); END IF; + EXCEPTION WHEN OTHERS THEN + ROLLBACK; + RAISE; END; / @@ -110,15 +112,22 @@ CREATE OR REPLACE FUNCTION UpdateMembership(PARAM_DEPLOYMENTID IN NVARCHAR2, PAR SET Status = PARAM_STATUS, SuspectTimes = PARAM_SUSPECTTIMES, - IAmAliveTime = PARAM_IAMALIVETIME + IAmAliveTime = GREATEST(IAmAliveTime, PARAM_IAMALIVETIME) WHERE DeploymentId = PARAM_DEPLOYMENTID AND PARAM_DEPLOYMENTID IS NOT NULL AND Address = PARAM_ADDRESS AND PARAM_ADDRESS IS NOT NULL AND Port = PARAM_PORT AND PARAM_PORT IS NOT NULL AND Generation = PARAM_GENERATION AND PARAM_GENERATION IS NOT NULL AND rowcount > 0; rowcount := SQL%ROWCOUNT; - COMMIT; + IF rowcount = 0 THEN + ROLLBACK; + ELSE + COMMIT; + END IF; RETURN(rowcount); + EXCEPTION WHEN OTHERS THEN + ROLLBACK; + RAISE; END; / @@ -153,10 +162,10 @@ BEGIN SET IAmAliveTime = PARAM_IAMALIVE WHERE - DeploymentId = PARAM_DEPLOYMENTID AND PARAM_DEPLOYMENTID IS NOT NULL - AND Address = PARAM_ADDRESS AND PARAM_ADDRESS IS NOT NULL - AND Port = PARAM_PORT AND PARAM_PORT IS NOT NULL - AND Generation = PARAM_GENERATION AND PARAM_GENERATION IS NOT NULL; + DeploymentId = PARAM_DEPLOYMENTID + AND Address = PARAM_ADDRESS + AND Port = PARAM_PORT + AND Generation = PARAM_GENERATION; COMMIT; RETURN(0); END; @@ -247,7 +256,9 @@ VALUES WHERE DeploymentId = :DeploymentId AND :DeploymentId IS NOT NULL AND IAmAliveTime < :IAmAliveTime - AND Status != 3; + AND StartTime < :IAmAliveTime + AND SuspectTimes IS NULL + AND Status = 6; END; '); / @@ -264,4 +275,16 @@ VALUES '); / +INSERT INTO OrleansQuery(QueryKey, QueryText) +VALUES +( + 'CleanupDefunctSiloEntryKey',' + DELETE FROM OrleansMembershipTable + WHERE DeploymentId = :DeploymentId AND Status = 6 + AND Address = :Address AND Port = :Port AND Generation = :Generation + AND IAmAliveTime = :IAmAliveTime AND StartTime = :StartTime + AND (SuspectTimes = :SuspectTimes OR (SuspectTimes IS NULL AND :SuspectTimes IS NULL)) +'); +/ + COMMIT; diff --git a/src/AdoNet/Orleans.Clustering.AdoNet/PostgreSQL-Clustering.sql b/src/AdoNet/Orleans.Clustering.AdoNet/PostgreSQL-Clustering.sql index 459e76b579d..e90af40c886 100644 --- a/src/AdoNet/Orleans.Clustering.AdoNet/PostgreSQL-Clustering.sql +++ b/src/AdoNet/Orleans.Clustering.AdoNet/PostgreSQL-Clustering.sql @@ -42,10 +42,10 @@ BEGIN SET IAmAliveTime = i_am_alive_time WHERE - d.DeploymentId = deployment_id AND deployment_id IS NOT NULL - AND d.Address = address_arg AND address_arg IS NOT NULL - AND d.Port = port_arg AND port_arg IS NOT NULL - AND d.Generation = generation_arg AND generation_arg IS NOT NULL; + d.DeploymentId = deployment_id + AND d.Address = address_arg + AND d.Port = port_arg + AND d.Generation = generation_arg; END $func$ LANGUAGE plpgsql; @@ -123,6 +123,16 @@ DECLARE BEGIN BEGIN + UPDATE OrleansMembershipVersionTable + SET + Timestamp = now(), + Version = Version + 1 + WHERE + DeploymentId = DeploymentIdArg AND DeploymentIdArg IS NOT NULL + AND Version = VersionArg AND VersionArg IS NOT NULL; + + GET DIAGNOSTICS RowCountVar = ROW_COUNT; + INSERT INTO OrleansMembershipTable ( DeploymentId, @@ -147,24 +157,16 @@ BEGIN ProxyPortArg, StartTimeArg, IAmAliveTimeArg + WHERE RowCountVar > 0 ON CONFLICT (DeploymentId, Address, Port, Generation) DO NOTHING; GET DIAGNOSTICS RowCountVar = ROW_COUNT; - UPDATE OrleansMembershipVersionTable - SET - Timestamp = now(), - Version = Version + 1 - WHERE - DeploymentId = DeploymentIdArg AND DeploymentIdArg IS NOT NULL - AND Version = VersionArg AND VersionArg IS NOT NULL - AND RowCountVar > 0; - - GET DIAGNOSTICS RowCountVar = ROW_COUNT; - - ASSERT RowCountVar <> 0, 'no rows affected, rollback'; + IF RowCountVar = 0 THEN + RAISE EXCEPTION 'no rows affected, rollback' USING ERRCODE = 'assert_failure'; + END IF; RETURN QUERY SELECT RowCountVar; @@ -228,7 +230,7 @@ BEGIN SET Status = StatusArg, SuspectTimes = SuspectTimesArg, - IAmAliveTime = IAmAliveTimeArg + IAmAliveTime = GREATEST(IAmAliveTime, IAmAliveTimeArg) WHERE DeploymentId = DeploymentIdArg AND DeploymentIdArg IS NOT NULL AND Address = AddressArg AND AddressArg IS NOT NULL @@ -239,7 +241,9 @@ BEGIN GET DIAGNOSTICS RowCountVar = ROW_COUNT; - ASSERT RowCountVar <> 0, 'no rows affected, rollback'; + IF RowCountVar = 0 THEN + RAISE EXCEPTION 'no rows affected, rollback' USING ERRCODE = 'assert_failure'; + END IF; RETURN QUERY SELECT RowCountVar; @@ -353,5 +357,18 @@ VALUES WHERE DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL AND IAmAliveTime < @IAmAliveTime - AND Status != 3; + AND StartTime < @IAmAliveTime + AND COALESCE(SuspectTimes, '''') = '''' + AND Status = 6; +'); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +VALUES +( + 'CleanupDefunctSiloEntryKey', + 'DELETE FROM OrleansMembershipTable + WHERE DeploymentId = @DeploymentId AND Status = 6 + AND Address = @Address AND Port = @Port AND Generation = @Generation + AND IAmAliveTime = @IAmAliveTime AND StartTime = @StartTime + AND COALESCE(SuspectTimes, '''') = COALESCE(@SuspectTimes, ''''); '); diff --git a/src/AdoNet/Orleans.Clustering.AdoNet/README.md b/src/AdoNet/Orleans.Clustering.AdoNet/README.md index 9a8d22cc2d1..00824420265 100644 --- a/src/AdoNet/Orleans.Clustering.AdoNet/README.md +++ b/src/AdoNet/Orleans.Clustering.AdoNet/README.md @@ -123,6 +123,20 @@ namespace ExampleGrains; - [PostgreSQL Scripts](https://github.com/dotnet/orleans/tree/main/src/AdoNet/Orleans.Clustering.AdoNet/PostgreSQL-Clustering.sql) - [Oracle Scripts](https://github.com/dotnet/orleans/tree/main/src/AdoNet/Orleans.Clustering.AdoNet/Oracle-Clustering.sql) +### Optional membership SQL enhancements + +Each silo owns its heartbeat. A heartbeat is one primary-key-targeted assignment to `IAmAliveTime`, sent as a single database command. Heartbeats preserve the logical membership row and table tokens, so a membership update can use tokens captured before a heartbeat. Canonical membership consistency is governed by those logical tokens; heartbeat timestamps are best-effort liveness observations. + +Upgrading silos or ADO.NET gateway-discovery clients preserves support for existing query installations. Database updates are optional. With an existing catalog, cleanup continues using its installed `CleanupDefunctSiloEntriesKey` query and behavior. + +The existing `*-Clustering.sql` installation scripts include improved conditional-write rollback and Dead-row cleanup. Their optional `CleanupDefunctSiloEntryKey` enables cleanup based on the latest start, heartbeat, and suspicion timestamps, with a captured-row condition protecting concurrent changes. Fresh installations receive these definitions through the usual database setup. + +SQL Server joined membership reads use statement snapshots provided by `READ_COMMITTED_SNAPSHOT`, which the existing `SQLServer-Main.sql` setup enables. + +Providers load and cache queries during initialization. Existing deployments continue using their installed definitions. Operators choosing to adopt revised definitions can use their normal database change-management process; reinitialize providers afterward to load the revised catalog. The original query keys and parameter sets remain supported. + +The SQL enhancements take effect per process as it adopts the updated queries. Until then, cached SQL Server/MySQL updates retain their previous missing-row version-increment behavior, and cached cleanup retains its previous status filter. The enhanced behavior applies cluster-wide once all membership writers use those definitions. SQL Server and MySQL inserts retain the original row-then-version statement order. + ## Documentation For more comprehensive documentation, please refer to: - [Microsoft Orleans Documentation](https://dotnet.github.io/orleans/docs/) diff --git a/src/AdoNet/Orleans.Clustering.AdoNet/SQLServer-Clustering.sql b/src/AdoNet/Orleans.Clustering.AdoNet/SQLServer-Clustering.sql index 695f6a32984..df27fbe2195 100644 --- a/src/AdoNet/Orleans.Clustering.AdoNet/SQLServer-Clustering.sql +++ b/src/AdoNet/Orleans.Clustering.AdoNet/SQLServer-Clustering.sql @@ -39,10 +39,10 @@ SELECT SET IAmAliveTime = @IAmAliveTime WHERE - DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL - AND Address = @Address AND @Address IS NOT NULL - AND Port = @Port AND @Port IS NOT NULL - AND Generation = @Generation AND @Generation IS NOT NULL; + DeploymentId = @DeploymentId + AND Address = @Address + AND Port = @Port + AND Generation = @Generation; ' WHERE NOT EXISTS ( @@ -84,6 +84,8 @@ SELECT 'SET XACT_ABORT, NOCOUNT ON; DECLARE @ROWCOUNT AS INT; BEGIN TRANSACTION; + + -- Preserve the row-then-version lock order used by cached legacy inserts. INSERT INTO OrleansMembershipTable ( DeploymentId, @@ -128,9 +130,9 @@ SELECT DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL AND Version = @Version AND @Version IS NOT NULL AND @@ROWCOUNT > 0; - + SET @ROWCOUNT = @@ROWCOUNT; - + IF @ROWCOUNT = 0 ROLLBACK TRANSACTION ELSE @@ -148,8 +150,9 @@ INSERT INTO OrleansQuery(QueryKey, QueryText) SELECT 'UpdateMembershipKey', 'SET XACT_ABORT, NOCOUNT ON; + DECLARE @ROWCOUNT AS INT; BEGIN TRANSACTION; - + UPDATE OrleansMembershipVersionTable SET Timestamp = GETUTCDATE(), @@ -157,21 +160,25 @@ SELECT WHERE DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL AND Version = @Version AND @Version IS NOT NULL; - + UPDATE OrleansMembershipTable SET Status = @Status, SuspectTimes = @SuspectTimes, - IAmAliveTime = @IAmAliveTime + IAmAliveTime = CASE WHEN IAmAliveTime > @IAmAliveTime THEN IAmAliveTime ELSE @IAmAliveTime END WHERE DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL AND Address = @Address AND @Address IS NOT NULL AND Port = @Port AND @Port IS NOT NULL AND Generation = @Generation AND @Generation IS NOT NULL AND @@ROWCOUNT > 0; - - SELECT @@ROWCOUNT; - COMMIT TRANSACTION; + + SET @ROWCOUNT = @@ROWCOUNT; + IF @ROWCOUNT = 0 + ROLLBACK TRANSACTION; + ELSE + COMMIT TRANSACTION; + SELECT @ROWCOUNT; ' WHERE NOT EXISTS ( @@ -285,7 +292,9 @@ SELECT WHERE DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL AND IAmAliveTime < @IAmAliveTime - AND Status != 3; + AND StartTime < @IAmAliveTime + AND COALESCE(SuspectTimes, '''') = '''' + AND Status = 6; ' WHERE NOT EXISTS ( @@ -293,3 +302,19 @@ WHERE NOT EXISTS FROM OrleansQuery oqt WHERE oqt.[QueryKey] = 'CleanupDefunctSiloEntriesKey' ); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +SELECT + 'CleanupDefunctSiloEntryKey', + 'DELETE FROM OrleansMembershipTable + WHERE DeploymentId = @DeploymentId AND Status = 6 + AND Address = @Address AND Port = @Port AND Generation = @Generation + AND IAmAliveTime = @IAmAliveTime AND StartTime = @StartTime + AND CONVERT(VARBINARY(8000), COALESCE(SuspectTimes, '''')) = CONVERT(VARBINARY(8000), CONVERT(VARCHAR(8000), COALESCE(@SuspectTimes, ''''))); + ' +WHERE NOT EXISTS +( + SELECT 1 + FROM OrleansQuery oqt + WHERE oqt.[QueryKey] = 'CleanupDefunctSiloEntryKey' +); diff --git a/src/AdoNet/Shared/Storage/DbStoredQueries.cs b/src/AdoNet/Shared/Storage/DbStoredQueries.cs index 0936ddc716e..57fcaa52f1b 100644 --- a/src/AdoNet/Shared/Storage/DbStoredQueries.cs +++ b/src/AdoNet/Shared/Storage/DbStoredQueries.cs @@ -99,6 +99,11 @@ internal DbStoredQueries(Dictionary queries) /// internal string CleanupDefunctSiloEntriesKey => queries[nameof(CleanupDefunctSiloEntriesKey)]; + /// + /// Gets the optional query for deleting a dead row whose captured activity values still match. + /// + internal string? GetCleanupDefunctSiloEntryQuery() => queries.GetValueOrDefault("CleanupDefunctSiloEntryKey"); + #endif #if REMINDERS_ADONET || TESTER_SQLUTILS || ORLEANS_REMINDERS_PROVIDER diff --git a/src/AdoNet/Shared/Storage/RelationalOrleansQueries.cs b/src/AdoNet/Shared/Storage/RelationalOrleansQueries.cs index 8e3d837c74b..f7279761e71 100644 --- a/src/AdoNet/Shared/Storage/RelationalOrleansQueries.cs +++ b/src/AdoNet/Shared/Storage/RelationalOrleansQueries.cs @@ -2,6 +2,7 @@ using System.Collections.Generic; using System.Data; using System.Data.Common; +using System.Globalization; using System.Linq; using System.Text; using System.Threading; @@ -294,17 +295,48 @@ internal Task DeleteMembershipTableEntriesAsync(string deploymentId, Cancellatio } /// - /// deletes all membership entries for inactive silos where the IAmAliveTime is before the beforeDate parameter - /// and the silo status is . + /// Runs the installed cleanup query, using captured-row eligibility when the optional query is available. /// /// /// /// The cancellation token. /// - internal Task CleanupDefunctSiloEntriesAsync(DateTimeOffset beforeDate, string deploymentId, CancellationToken cancellationToken = default) + internal async Task CleanupDefunctSiloEntriesAsync(DateTimeOffset beforeDate, string deploymentId, CancellationToken cancellationToken = default) { - return ExecuteAsync(dbStoredQueries.CleanupDefunctSiloEntriesKey, command => - new DbStoredQueries.Columns(command) { DeploymentId = deploymentId, IAmAliveTime = beforeDate.UtcDateTime }, cancellationToken); + cancellationToken.ThrowIfCancellationRequested(); + if (dbStoredQueries.GetCleanupDefunctSiloEntryQuery() is not { } cleanupQuery) + { + await ExecuteAsync(dbStoredQueries.CleanupDefunctSiloEntriesKey, command => + new DbStoredQueries.Columns(command) { DeploymentId = deploymentId, IAmAliveTime = beforeDate.UtcDateTime }, cancellationToken); + return; + } + + var table = await MembershipReadAllAsync(deploymentId, cancellationToken); + var cutoff = beforeDate.UtcDateTime; + foreach (var (entry, _) in table.Members) + { + if (entry.Status == SiloStatus.Dead + && entry.StartTime < cutoff + && entry.IAmAliveTime < cutoff + && entry.SuspectTimes?.Any(vote => vote.Item2 >= cutoff) != true) + { + // Evaluate the cutoff in .NET and match the original storage values in SQL. + await ExecuteAsync(cleanupQuery, command => + { + var columns = new DbStoredQueries.Columns(command) + { + DeploymentId = deploymentId, + SiloAddress = entry.SiloAddress, + SuspectTimes = entry.SuspectTimes + }; + // SQL Server's DateTime binding rounds values read from the DateTime2 columns. + var timestampType = storage.InvariantName == AdoNetInvariants.InvariantNameSqlServer ? DbType.DateTime2 : DbType.DateTime; + command.AddParameter(nameof(DbStoredQueries.Columns.StartTime), entry.StartTime, dbType: timestampType); + command.AddParameter(nameof(DbStoredQueries.Columns.IAmAliveTime), entry.IAmAliveTime, dbType: timestampType); + return columns; + }, cancellationToken); + } + } } /// @@ -349,6 +381,12 @@ internal Task InsertMembershipVersionRowAsync(string deploymentId, Cancell internal Task InsertMembershipRowAsync(string deploymentId, MembershipEntry membershipEntry, string etag, CancellationToken cancellationToken = default) { + cancellationToken.ThrowIfCancellationRequested(); + if (!IsMembershipVersionEtag(etag)) + { + return Task.FromResult(false); + } + return ReadAsync(dbStoredQueries.InsertMembershipKey, DbStoredQueries.Converters.GetSingleBooleanValue, command => new DbStoredQueries.Columns(command) { @@ -375,6 +413,12 @@ internal Task InsertMembershipRowAsync(string deploymentId, MembershipEntr internal Task UpdateMembershipRowAsync(string deploymentId, MembershipEntry membershipEntry, string etag, CancellationToken cancellationToken = default) { + cancellationToken.ThrowIfCancellationRequested(); + if (!IsMembershipVersionEtag(etag)) + { + return Task.FromResult(false); + } + return ReadAsync(dbStoredQueries.UpdateMembershipKey, DbStoredQueries.Converters.GetSingleBooleanValue, command => new DbStoredQueries.Columns(command) { @@ -387,16 +431,21 @@ internal Task UpdateMembershipRowAsync(string deploymentId, MembershipEntr }, ret => ret.First(), cancellationToken); } + private static bool IsMembershipVersionEtag(string etag) => + int.TryParse(etag, NumberStyles.None, CultureInfo.InvariantCulture, out var version) + && string.Equals(etag, version.ToString(CultureInfo.InvariantCulture), StringComparison.Ordinal); + private static MembershipTableData ConvertToMembershipTableData(IEnumerable> ret) { var retList = ret.ToList(); var tableVersionEtag = retList[0].Item2; + var etag = tableVersionEtag.ToString(CultureInfo.InvariantCulture); var membershipEntries = new List>(); if (retList[0].Item1 != null) { - membershipEntries.AddRange(retList.Select(i => new Tuple(i.Item1!, string.Empty))); + membershipEntries.AddRange(retList.Select(i => new Tuple(i.Item1!, etag))); } - return new MembershipTableData(membershipEntries, new TableVersion(tableVersionEtag, tableVersionEtag.ToString())); + return new MembershipTableData(membershipEntries, new TableVersion(tableVersionEtag, etag)); } #endif diff --git a/test/Extensions/Orleans.AdoNet.Tests/Orleans.AdoNet.Tests.csproj b/test/Extensions/Orleans.AdoNet.Tests/Orleans.AdoNet.Tests.csproj index ef9876f13f7..ac33b289d3a 100644 --- a/test/Extensions/Orleans.AdoNet.Tests/Orleans.AdoNet.Tests.csproj +++ b/test/Extensions/Orleans.AdoNet.Tests/Orleans.AdoNet.Tests.csproj @@ -43,6 +43,9 @@ + PreserveNewest diff --git a/test/Extensions/Orleans.AdoNet.Tests/RelationalUtilities/MySqlStorageForTesting.cs b/test/Extensions/Orleans.AdoNet.Tests/RelationalUtilities/MySqlStorageForTesting.cs index a7c0f5afa28..3b8408f50ff 100644 --- a/test/Extensions/Orleans.AdoNet.Tests/RelationalUtilities/MySqlStorageForTesting.cs +++ b/test/Extensions/Orleans.AdoNet.Tests/RelationalUtilities/MySqlStorageForTesting.cs @@ -18,7 +18,7 @@ public MySqlStorageForTesting(string connectionString) : base(AdoNetInvariants.I public IEnumerable SplitScript(string setupScript) { return setupScript.Replace("END$$", "END;") - .Split(new[] { "DELIMITER $$", "DELIMITER ;" }, StringSplitOptions.RemoveEmptyEntries); + .Split(new[] { "DELIMITER $$", "DELIMITER ;" }, StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries); } protected override string CreateDatabaseTemplate @@ -38,7 +38,7 @@ protected override string ExistsDatabaseTemplate protected override IEnumerable ConvertToExecutableBatches(string setupScript, string databaseName) { - var batches = setupScript.Replace("END$$", "END;").Split(new[] { "DELIMITER $$", "DELIMITER ;" }, StringSplitOptions.RemoveEmptyEntries).ToList(); + var batches = SplitScript(setupScript).ToList(); batches.Add(CreateStreamTestTable); return batches; diff --git a/test/Extensions/Orleans.AdoNet.Tests/RelationalUtilities/RelationalStorageForTesting.cs b/test/Extensions/Orleans.AdoNet.Tests/RelationalUtilities/RelationalStorageForTesting.cs index 60b6c2347c9..b58a51b12c7 100644 --- a/test/Extensions/Orleans.AdoNet.Tests/RelationalUtilities/RelationalStorageForTesting.cs +++ b/test/Extensions/Orleans.AdoNet.Tests/RelationalUtilities/RelationalStorageForTesting.cs @@ -54,6 +54,7 @@ public virtual string StreamTestInsert /// /// The script that creates Orleans schema in the database, usually CreateOrleansTables_xxxx.sql /// + // Replaying historical clustering migrations after a fresh install can replace current routines. protected string[] SetupSqlScriptFileNames => new[] { $"{this.ProviderMoniker}-Main.sql", $"{this.ProviderMoniker}-Clustering.sql", @@ -61,8 +62,7 @@ public virtual string StreamTestInsert $"{this.ProviderMoniker}-Reminders.sql", $"{this.ProviderMoniker}-Streaming.sql", $"{this.ProviderMoniker}-GrainDirectory.sql" - }.Concat(Directory.GetFiles(Environment.CurrentDirectory, $"{this.ProviderMoniker}-Clustering-*.sql") - .Concat(Directory.GetFiles(Environment.CurrentDirectory, $"{this.ProviderMoniker}-Persistence-*.sql")) + }.Concat(Directory.GetFiles(Environment.CurrentDirectory, $"{this.ProviderMoniker}-Persistence-*.sql") .Concat(Directory.GetFiles(Environment.CurrentDirectory, $"{this.ProviderMoniker}-Reminders-*.sql")) .Concat(Directory.GetFiles(Environment.CurrentDirectory, $"{this.ProviderMoniker}-Streaming-*.sql")) .Concat(Directory.GetFiles(Environment.CurrentDirectory, $"{this.ProviderMoniker}-GrainDirectory-*.sql")) @@ -109,7 +109,8 @@ public static async Task SetupInstance( string invariantName, string testDatabaseName, string? connectionString = null, - CancellationToken cancellationToken = default) + CancellationToken cancellationToken = default, + IEnumerable? setupSqlScriptFileNames = null) { CheckPreconditionsOrThrow(invariantName, connectionString); if (string.IsNullOrWhiteSpace(invariantName)) @@ -152,7 +153,7 @@ public static async Task SetupInstance( var setupScript = string.Empty; // Concatenate scripts - foreach (var fileName in testStorage.SetupSqlScriptFileNames) + foreach (var fileName in setupSqlScriptFileNames ?? testStorage.SetupSqlScriptFileNames) { setupScript += await File.ReadAllTextAsync(fileName, cancellationToken); diff --git a/test/Extensions/Orleans.AdoNet.Tests/StorageTests/Relational/AdoNetMembershipSqlTests.cs b/test/Extensions/Orleans.AdoNet.Tests/StorageTests/Relational/AdoNetMembershipSqlTests.cs new file mode 100644 index 00000000000..ce63574bc40 --- /dev/null +++ b/test/Extensions/Orleans.AdoNet.Tests/StorageTests/Relational/AdoNetMembershipSqlTests.cs @@ -0,0 +1,243 @@ +using System.Text.RegularExpressions; +using UnitTests.General; + +namespace UnitTests.StorageTests.Relational; + +[TestSuite("BVT")] +[TestProvider("None")] +[TestArea("Clustering")] +public sealed class AdoNetMembershipSqlTests +{ + public static TheoryData Engines => new() { "SQLServer", "PostgreSQL", "MySQL", "Oracle" }; + + [Fact] + public void MySqlSetup_SkipsWhitespaceOnlyDelimiterBatches() + { + var setup = new MySqlStorageForTesting("Server=localhost;"); + var batches = setup.SplitScript(" \r\nDELIMITER $$\r\nCREATE PROCEDURE p() BEGIN SELECT 1; END$$\r\nDELIMITER ;\r\n \t"); + + Assert.Equal(["CREATE PROCEDURE p() BEGIN SELECT 1; END;"], batches); + } + + [Fact] + public void MySqlSetup_FreshInstallationProducesExecutableBatches() + { + var setup = new MySqlStorageForTesting("Server=localhost;"); + var script = string.Join("\r\n", new[] { "Main", "Clustering", "Persistence", "Reminders", "Streaming", "GrainDirectory" } + .Select(name => File.ReadAllText(Path.Combine(AppContext.BaseDirectory, $"MySQL-{name}.sql")))); + var batches = setup.SplitScript(script).ToArray(); + + Assert.NotEmpty(batches); + Assert.All(batches, batch => Assert.False(string.IsNullOrWhiteSpace(batch))); + Assert.Contains(batches, batch => batch.Contains("CREATE PROCEDURE InsertMembershipKey(", StringComparison.Ordinal)); + } + + [Theory] + [MemberData(nameof(Engines))] + public void Cleanup_UsesCapturedValuesWithoutChangingVersion(string engine) + { + var query = Queries(ReadScript(engine))["CleanupDefunctSiloEntryKey"]; + + Assert.Equal( + ["Address", "DeploymentId", "Generation", "IAmAliveTime", "Port", "StartTime", "SuspectTimes"], + Parameters(query)); + Assert.StartsWith("DELETE FROM OrleansMembershipTable", query.Trim(), StringComparison.Ordinal); + Assert.Contains("Status = 6", query, StringComparison.Ordinal); + foreach (var column in new[] { "DeploymentId", "Address", "Port", "Generation", "StartTime", "IAmAliveTime" }) + { + Assert.Matches($@"\b{column} = [@:]{column}\b", query); + } + + Assert.Matches(@"SuspectTimes.*(?:IS NULL|COALESCE)", Regex.Replace(query, @"\s+", " ")); + Assert.DoesNotContain("Version", query, StringComparison.OrdinalIgnoreCase); + Assert.DoesNotContain("UPDATE", query, StringComparison.OrdinalIgnoreCase); + Assert.DoesNotContain("2147483647", query, StringComparison.Ordinal); + if (engine == "SQLServer") + { + Assert.Contains("CONVERT(VARCHAR(8000), COALESCE(@SuspectTimes, ''))", query, StringComparison.Ordinal); + } + } + + [Theory] + [MemberData(nameof(Engines))] + public void LegacyCleanup_PreservesParametersAndProtectsActivity(string engine) + { + var query = Queries(ReadScript(engine))["CleanupDefunctSiloEntriesKey"]; + + Assert.Equal(["DeploymentId", "IAmAliveTime"], Parameters(query)); + Assert.Contains("Status = 6", query, StringComparison.Ordinal); + Assert.Matches(@"IAmAliveTime < [@:]IAmAliveTime", query); + Assert.Matches(@"StartTime < [@:]IAmAliveTime", query); + Assert.Matches(@"SuspectTimes(?: IS NULL|, ''\) = '')", query); + Assert.DoesNotContain("Version", query, StringComparison.OrdinalIgnoreCase); + } + + [Theory] + [MemberData(nameof(Engines))] + public void MembershipReads_JoinRowsAndVersionInOneStatement(string engine) + { + var queries = Queries(ReadScript(engine)); + foreach (var key in new[] { "MembershipReadRowKey", "MembershipReadAllKey" }) + { + var query = queries[key]; + Assert.Single(Regex.Matches(query, @"\bSELECT\b", RegexOptions.IgnoreCase)); + Assert.Contains("v.Version", query, StringComparison.Ordinal); + Assert.Contains("LEFT OUTER JOIN OrleansMembershipTable m", query, StringComparison.Ordinal); + Assert.Contains("v.DeploymentId = m.DeploymentId", query, StringComparison.Ordinal); + if (engine == "SQLServer") + { + Assert.DoesNotContain("HOLDLOCK", query, StringComparison.OrdinalIgnoreCase); + var setup = File.ReadAllText(Path.Combine(AppContext.BaseDirectory, "SQLServer-Main.sql")); + Assert.Contains("SET READ_COMMITTED_SNAPSHOT ON", setup, StringComparison.Ordinal); + } + } + } + + [Theory] + [MemberData(nameof(Engines))] + public void Heartbeat_IsOneBlindColumnWriteByPrimaryKey(string engine) + { + var script = ReadScript(engine); + var queries = Queries(script); + var routines = Routines(script); + var heartbeat = engine switch + { + "PostgreSQL" => routines["update_i_am_alive_time"], + "Oracle" => routines["UpdateIAmAlivetime"], + _ => queries["UpdateIAmAlivetimeKey"], + }; + Assert.Single(Regex.Matches(heartbeat, @"\bUPDATE\b", RegexOptions.IgnoreCase)); + Assert.DoesNotMatch(@"(?i)\b(?:SELECT|JOIN|IF|CASE|GREATEST|Status|Version)\b", heartbeat); + var statement = Regex.Match(heartbeat, @"UPDATE OrleansMembershipTable.*?;", RegexOptions.Singleline).Value; + var expected = engine switch + { + "PostgreSQL" => "UPDATE OrleansMembershipTable as d SET IAmAliveTime = i_am_alive_time WHERE d.DeploymentId = deployment_id AND d.Address = address_arg AND d.Port = port_arg AND d.Generation = generation_arg;", + "Oracle" => "UPDATE OrleansMembershipTable SET IAmAliveTime = PARAM_IAMALIVE WHERE DeploymentId = PARAM_DEPLOYMENTID AND Address = PARAM_ADDRESS AND Port = PARAM_PORT AND Generation = PARAM_GENERATION;", + _ => "UPDATE OrleansMembershipTable SET IAmAliveTime = @IAmAliveTime WHERE DeploymentId = @DeploymentId AND Address = @Address AND Port = @Port AND Generation = @Generation;", + }; + Assert.Equal(expected, Regex.Replace(statement, @"\s+", " ")); + } + + [Theory] + [MemberData(nameof(Engines))] + public void FullMembershipUpdate_PreservesHeartbeatWithinExistingStatement(string engine) + { + var script = ReadScript(engine); + var update = UpdateBody(engine, Queries(script), Routines(script)); + if (engine == "SQLServer") + { + Assert.Contains("IAmAliveTime = CASE WHEN IAmAliveTime > @IAmAliveTime THEN IAmAliveTime ELSE @IAmAliveTime END", update, StringComparison.Ordinal); + } + else + { + Assert.Matches(@"IAmAliveTime = GREATEST\((?:[dm]\.)?IAmAliveTime, ", update); + } + } + + [Theory] + [MemberData(nameof(Engines))] + public void InsertMembership_AtomicallyChecksVersionAndRollsBackConflicts(string engine) + { + var script = ReadScript(engine); + var queries = Queries(script); + var routines = Routines(script); + var insert = engine switch + { + "PostgreSQL" => routines["insert_membership"], + "Oracle" => routines["InsertMembership"], + "MySQL" => routines["InsertMembershipKey"], + _ => queries["InsertMembershipKey"], + }; + + var versionWrite = insert.IndexOf("UPDATE OrleansMembershipVersionTable", StringComparison.Ordinal); + var rowWrite = insert.IndexOf("INSERT INTO OrleansMembershipTable", StringComparison.Ordinal); + if (engine is "SQLServer" or "MySQL") + { + Assert.True(rowWrite >= 0 && versionWrite > rowWrite, insert); + Assert.Contains(engine == "SQLServer" ? "AND @@ROWCOUNT > 0;" : "AND ROW_COUNT() > 0;", insert[versionWrite..], StringComparison.Ordinal); + } + else + { + Assert.True(versionWrite >= 0 && rowWrite > versionWrite, insert); + Assert.Matches(@"WHERE (?:_ROWCOUNT|RowCountVar|rowcount) > 0", insert); + } + + Assert.Matches(@"Version = (?:@Version|_Version|VersionArg|PARAM_VERSION)", insert); + AssertRollback(engine, insert); + } + + [Theory] + [MemberData(nameof(Engines))] + public void UpdateMembership_AtomicallyRejectsStaleVersionAndMissingRow(string engine) + { + var script = ReadScript(engine); + var update = UpdateBody(engine, Queries(script), Routines(script)); + + Assert.Matches(@"Version = (?:@Version|VersionArg|PARAM_VERSION)", update); + foreach (var column in new[] { "Address", "Port", "Generation" }) + { + Assert.Matches($@"\b{column} = (?:@{column}|{column}Arg|PARAM_{column.ToUpperInvariant()})", update); + } + + var rowConditions = update[update.LastIndexOf("WHERE", StringComparison.Ordinal)..]; + Assert.DoesNotContain("IAmAliveTime", rowConditions, StringComparison.OrdinalIgnoreCase); + Assert.DoesNotContain("SuspectTimes", rowConditions, StringComparison.OrdinalIgnoreCase); + Assert.DoesNotContain("StartTime", rowConditions, StringComparison.OrdinalIgnoreCase); + + if (engine == "MySQL") + { + Assert.Single(Regex.Matches(update, @"\bUPDATE\b")); + Assert.Contains("INNER JOIN OrleansMembershipTable m ON m.DeploymentId = v.DeploymentId", update, StringComparison.Ordinal); + Assert.Contains("SET v.Version = v.Version + 1,", update, StringComparison.Ordinal); + Assert.Contains("SELECT ROW_COUNT() > 0;", update, StringComparison.Ordinal); + } + else + { + var rowWrite = update.IndexOf("UPDATE OrleansMembershipTable", StringComparison.Ordinal); + Assert.True(rowWrite > update.IndexOf("UPDATE OrleansMembershipVersionTable", StringComparison.Ordinal), update); + AssertRollback(engine, update[rowWrite..]); + } + } + + private static void AssertRollback(string engine, string body) + { + if (engine == "PostgreSQL") + { + Assert.DoesNotMatch(@"\bASSERT\b", body); + Assert.Contains("IF RowCountVar = 0 THEN", body, StringComparison.Ordinal); + Assert.Contains("RAISE EXCEPTION 'no rows affected, rollback' USING ERRCODE = 'assert_failure';", body, StringComparison.Ordinal); + Assert.Contains("WHEN assert_failure THEN", body, StringComparison.Ordinal); + } + else + { + Assert.Matches(@"(?i)IF (?:@ROWCOUNT|_ROWCOUNT|rowcount) = 0\s*(?:THEN\s*)?ROLLBACK", body); + Assert.Matches(@"(?i)ELSE\s*COMMIT", body); + if (engine == "MySQL") + { + Assert.Contains("DECLARE EXIT HANDLER FOR SQLEXCEPTION", body, StringComparison.Ordinal); + Assert.Contains("RESIGNAL;", body, StringComparison.Ordinal); + } + } + } + + private static string UpdateBody(string engine, Dictionary queries, Dictionary routines) => engine switch + { + "PostgreSQL" => routines["update_membership"], + "Oracle" => routines["UpdateMembership"], + _ => queries["UpdateMembershipKey"], + }; + + private static string ReadScript(string engine) => + File.ReadAllText(Path.Combine(AppContext.BaseDirectory, $"{engine}-Clustering.sql")).Replace("\r\n", "\n"); + + private static string[] Parameters(string query) => + Regex.Matches(query, @"[@:](\w+)").Select(match => match.Groups[1].Value).Distinct().Order(StringComparer.Ordinal).ToArray(); + + private static Dictionary Queries(string script) => + Regex.Matches(script, @"'(?\w+Key)'\s*,\s*'(?(?:''|[^'])*)'") + .ToDictionary(match => match.Groups["key"].Value, match => match.Groups["text"].Value.Replace("''", "'")); + + private static Dictionary Routines(string script) => + Regex.Matches(script, @"CREATE (?:OR REPLACE )?(?:FUNCTION|PROCEDURE) (?\w+)\(.*?(?:\$func\$ LANGUAGE plpgsql;|END\$\$|END;\n/)", RegexOptions.Singleline) + .ToDictionary(match => match.Groups["name"].Value, match => match.Value.Replace("CREATE OR REPLACE FUNCTION", "CREATE FUNCTION")); +} diff --git a/test/Extensions/Orleans.AdoNet.Tests/StorageTests/Relational/AdoNetMembershipUpgradeTests.cs b/test/Extensions/Orleans.AdoNet.Tests/StorageTests/Relational/AdoNetMembershipUpgradeTests.cs new file mode 100644 index 00000000000..1620a6ecfde --- /dev/null +++ b/test/Extensions/Orleans.AdoNet.Tests/StorageTests/Relational/AdoNetMembershipUpgradeTests.cs @@ -0,0 +1,314 @@ +using System.Data.Common; +using System.Globalization; +using System.Net; +using System.Security.Cryptography; +using System.Text; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Npgsql; +using Orleans.Configuration; +using Orleans.Runtime; +using Orleans.Runtime.Membership; +using Orleans.Runtime.MembershipService; +using Orleans.Tests.SqlUtils; +using UnitTests.General; + +namespace UnitTests.StorageTests.Relational; + +[TestSuite("Functional")] +[TestArea("Membership")] +public sealed class AdoNetMembershipUpgradeTests +{ + private const string ClusterId = "membership-compatibility"; + private static readonly DateTime StartTime = new(2030, 1, 2, 3, 4, 5, DateTimeKind.Utc); + + // src/AdoNet/Orleans.Clustering.AdoNet/*-Clustering.sql from d515d75eaaa3a0390a74cb11c96aa758358b5a67, + // extracted with only trailing line spaces/tabs removed. Hashes also normalize checkout line endings. + [Theory] + [InlineData(false)] + [InlineData(true)] + [TestProvider("SqlServer")] + public Task SqlServer_NativeMembership_OriginalAndEnhancedCatalogs(bool enhanced) => + VerifyInstallationAsync("SQLServer", AdoNetInvariants.InvariantNameSqlServer, enhanced, + "FB94B7309CA2FDB11BDC8FDA44684DC2CA4636BDC570A491EBF91E9D4E579660"); + + [Theory] + [InlineData(false)] + [InlineData(true)] + [TestProvider("PostgreSql")] + public Task PostgreSql_NativeMembership_OriginalAndEnhancedCatalogs(bool enhanced) => + VerifyInstallationAsync("PostgreSQL", AdoNetInvariants.InvariantNamePostgreSql, enhanced, + "75682538351DC15AD74D86FAB14110CCA2603FD063D19CEBC76118FF84C88328"); + + [Theory] + [InlineData(false)] + [InlineData(true)] + [TestProvider("MySql")] + public Task MySql_NativeMembership_OriginalAndEnhancedCatalogs(bool enhanced) => + VerifyInstallationAsync("MySQL", AdoNetInvariants.InvariantNameMySql, enhanced, + "FD33B6AD41DDA95942F275857F27C7FD02A0DB433747DA4C3B92D677ACB6014E"); + + private static async Task VerifyInstallationAsync(string engine, string invariant, bool enhanced, string fixtureHash) + { + var cancellationToken = TestContext.Current.CancellationToken; + var fixturePath = Path.Combine(AppContext.BaseDirectory, "MembershipUpgradeFixtures", $"{engine}-Clustering.sql"); + var fixture = await File.ReadAllTextAsync(fixturePath, cancellationToken); + Assert.Equal(fixtureHash, Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(fixture.Replace("\r\n", "\n"))))); + var clusteringPath = enhanced ? Path.Combine(AppContext.BaseDirectory, $"{engine}-Clustering.sql") : fixturePath; + var database = await RelationalStorageForTesting.SetupInstance( + invariant, + $"MembershipCompatibility_{Guid.NewGuid():N}", + cancellationToken: cancellationToken, + setupSqlScriptFileNames: [Path.Combine(AppContext.BaseDirectory, $"{engine}-Main.sql"), clusteringPath]); + var storage = database.Storage; + var queries = await ReadQueriesAsync(storage, cancellationToken); + Assert.Equal(enhanced, queries.ContainsKey("CleanupDefunctSiloEntryKey")); + var connectionString = database.CurrentConnectionString; + if (enhanced && engine == "PostgreSQL") + { + // Enhanced rollback must not depend on the server's optional PL/pgSQL assertions. + connectionString = new NpgsqlConnectionStringBuilder(connectionString) { Options = "-c plpgsql.check_asserts=off" }.ConnectionString; + } + + using var services = new ServiceCollection().BuildServiceProvider(); + var clusterOptions = Options.Create(new ClusterOptions { ClusterId = ClusterId }); + var current = new AdoNetClusteringTable( + services, clusterOptions, + Options.Create(new AdoNetClusteringSiloOptions { Invariant = invariant, ConnectionString = connectionString }), + NullLogger.Instance); + var gateway = new AdoNetGatewayListProvider( + NullLogger.Instance, services, + Options.Create(new AdoNetClusteringClientOptions { Invariant = invariant, ConnectionString = connectionString }), + Options.Create(new GatewayOptions()), clusterOptions); + await current.InitializeMembershipTableAsync(true, cancellationToken); + var empty = await ReadSnapshotAsync(storage, cancellationToken); + AssertStored(empty, 0); + AssertReadMatches(empty, await current.ReadAllAsync(cancellationToken)); + + var cutoff = StartTime.AddHours(1); + var active = Entry(1, SiloStatus.Active); + active.ProxyPort = 30001; + var dead = Entry(2, SiloStatus.Dead); + var suspected = Entry(3, SiloStatus.Dead); + var joining = Entry(4, SiloStatus.Joining); + var startingAtCutoff = Entry(5, SiloStatus.Dead); + startingAtCutoff.StartTime = cutoff; + var heartbeatAtCutoff = Entry(6, SiloStatus.Dead); + heartbeatAtCutoff.IAmAliveTime = cutoff; + var oldSuspect = Entry(7, SiloStatus.Dead); + var entries = new[] { active, dead, suspected, joining, startingAtCutoff, heartbeatAtCutoff, oldSuspect }; + for (var version = 0; version < entries.Length; version++) + { + Assert.True(await current.InsertRowAsync(entries[version], new TableVersion(version + 1, version.ToString(CultureInfo.InvariantCulture)), cancellationToken)); + } + + suspected.SuspectTimes = [Tuple.Create(active.SiloAddress, cutoff)]; + oldSuspect.SuspectTimes = [Tuple.Create(active.SiloAddress, cutoff.AddSeconds(-1))]; + foreach (var entry in new[] { suspected, oldSuspect }) + { + var row = await current.ReadRowAsync(entry.SiloAddress, cancellationToken); + Assert.True(await current.UpdateRowAsync(entry, Assert.Single(row.Members).Item2, row.Version.Next(), cancellationToken)); + } + + // A populated original installation supports package upgrades without any catalog edits. + var populated = await ReadSnapshotAsync(storage, cancellationToken); + AssertStored(populated, 9, entries); + await current.InitializeMembershipTableAsync(true, cancellationToken); + await gateway.InitializeGatewayListProvider().WaitAsync(cancellationToken); + AssertReadMatches(populated, await current.ReadAllAsync(cancellationToken)); + Assert.Equal( + [SiloAddress.New(active.SiloAddress.Endpoint.Address, active.ProxyPort, active.SiloAddress.Generation).ToGatewayUri()], + await gateway.GetGateways().WaitAsync(cancellationToken)); + AssertUnchanged(populated, await ReadSnapshotAsync(storage, cancellationToken)); + Assert.Equal(queries.OrderBy(pair => pair.Key), (await ReadQueriesAsync(storage, cancellationToken)).OrderBy(pair => pair.Key)); + + var captured = await current.ReadRowAsync(active.SiloAddress, cancellationToken); + Assert.Equal(populated.Version, captured.Version.Version); + Assert.Equal(populated.Etag, captured.Version.VersionEtag); + var capturedRow = Assert.Single(captured.Members); + Assert.Equal(StoredMember.FromEntry(active), StoredMember.FromEntry(capturedRow.Item1)); + var originalRowToken = capturedRow.Item2; + Assert.Equal(populated.Etag, originalRowToken); + active.IAmAliveTime = StartTime.AddMinutes(10); + await current.UpdateIAmAliveAsync(active, cancellationToken); + var afterHeartbeat = await ReadSnapshotAsync(storage, cancellationToken); + AssertStored(afterHeartbeat, 9, entries); + Assert.Equal(populated.VersionTimestamp, afterHeartbeat.VersionTimestamp); + active.Status = SiloStatus.ShuttingDown; + active.SuspectTimes = [Tuple.Create(joining.SiloAddress, StartTime.AddMinutes(4))]; + active.IAmAliveTime = StartTime.AddMinutes(5); + Assert.True(await current.UpdateRowAsync(active, originalRowToken, captured.Version.Next(), cancellationToken)); + // Original SQL can overwrite a newer heartbeat; only enhanced full-row writes retain MAX. + active.IAmAliveTime = StartTime.AddMinutes(enhanced ? 10 : 5); + var afterUpdate = await ReadSnapshotAsync(storage, cancellationToken); + AssertStored(afterUpdate, 10, entries); + active.IAmAliveTime = StartTime.AddMinutes(3); + await current.UpdateIAmAliveAsync(active, cancellationToken); + var afterBlindHeartbeat = await ReadSnapshotAsync(storage, cancellationToken); + AssertStored(afterBlindHeartbeat, 10, entries); + Assert.Equal(afterUpdate.VersionTimestamp, afterBlindHeartbeat.VersionTimestamp); + active.IAmAliveTime = StartTime.AddMinutes(20); + await current.UpdateIAmAliveAsync(active, cancellationToken); + var stable = await ReadSnapshotAsync(storage, cancellationToken); + AssertStored(stable, 10, entries); + Assert.Equal(afterUpdate.VersionTimestamp, stable.VersionTimestamp); + await current.UpdateIAmAliveAsync(Entry(20, SiloStatus.Active), cancellationToken); + AssertUnchanged(stable, await ReadSnapshotAsync(storage, cancellationToken)); + + if (enhanced) + { + var absent = Entry(20, SiloStatus.Dead); + var changed = Entry(1, SiloStatus.Dead); + changed.IAmAliveTime = StartTime.AddDays(1); + changed.SuspectTimes = [Tuple.Create(joining.SiloAddress, StartTime.AddDays(1))]; + await AssertRejectedAsync(() => current.InsertRowAsync(absent, captured.Version.Next(), cancellationToken)); + await AssertRejectedAsync(() => current.UpdateRowAsync(changed, originalRowToken, captured.Version.Next(), cancellationToken)); + await AssertRejectedAsync(() => current.InsertRowAsync(changed, stable.NextVersion, cancellationToken)); + await AssertRejectedAsync(() => current.UpdateRowAsync(changed, originalRowToken, stable.NextVersion, cancellationToken)); + await AssertRejectedAsync(() => current.UpdateRowAsync(absent, stable.Etag, stable.NextVersion, cancellationToken)); + + absent.HostName = null!; + var error = await Assert.ThrowsAnyAsync(() => current.InsertRowAsync(absent, stable.NextVersion, cancellationToken)); + Assert.True(engine switch + { + "SQLServer" => error is Microsoft.Data.SqlClient.SqlException { Number: 515 }, + "MySQL" => error is MySql.Data.MySqlClient.MySqlException { Number: 1048 }, + "PostgreSQL" => error is PostgresException { SqlState: PostgresErrorCodes.NotNullViolation }, + _ => false + }, error.ToString()); + AssertUnchanged(stable, await ReadSnapshotAsync(storage, cancellationToken)); + using var canceled = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + canceled.Cancel(); + await Assert.ThrowsAnyAsync(() => current.InsertRowAsync(Entry(20, SiloStatus.Joining), stable.NextVersion, canceled.Token)); + AssertUnchanged(stable, await ReadSnapshotAsync(storage, cancellationToken)); + } + + // Original cleanup retains its original semantics, so use a safe no-op cutoff in that mode. + await current.CleanupDefunctSiloEntriesAsync(new DateTimeOffset(enhanced ? cutoff : StartTime), cancellationToken); + var afterCleanup = await ReadSnapshotAsync(storage, cancellationToken); + var survivors = enhanced ? entries.Where(entry => entry != dead && entry != oldSuspect).ToArray() : entries; + AssertStored(afterCleanup, 10, survivors); + Assert.Equal(stable.VersionTimestamp, afterCleanup.VersionTimestamp); + AssertReadMatches(afterCleanup, await current.ReadAllAsync(cancellationToken)); + + if (engine == "SQLServer") + { + Assert.True(Assert.Single(await storage.ReadAsync( + "SELECT is_read_committed_snapshot_on FROM sys.databases WHERE database_id = DB_ID()", + record => record.GetBoolean(0), null, cancellationToken))); + var overlappingEntry = Entry(12, SiloStatus.Joining); + var start = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var insertion = InsertAfterBarrierAsync(); + var reading = ReadAfterBarrierAsync(); + start.SetResult(); + await Task.WhenAll(insertion, reading); + Assert.True(await insertion); + var observed = await reading; + var afterOverlap = await ReadSnapshotAsync(storage, cancellationToken); + AssertStored(afterOverlap, 11, survivors.Append(overlappingEntry).ToArray()); + Assert.Equal(observed.Version.Version.ToString(CultureInfo.InvariantCulture), observed.Version.VersionEtag); + if (observed.Version.Version == afterCleanup.Version) + { + Assert.Empty(observed.Members); + } + else + { + Assert.Equal(afterOverlap.Version, observed.Version.Version); + var row = Assert.Single(observed.Members); + Assert.Equal(afterOverlap.Etag, row.Item2); + Assert.Equal(StoredMember.FromEntry(overlappingEntry), StoredMember.FromEntry(row.Item1)); + } + + async Task InsertAfterBarrierAsync() + { + await start.Task.WaitAsync(cancellationToken); + return await current.InsertRowAsync(overlappingEntry, afterCleanup.NextVersion, cancellationToken); + } + + async Task ReadAfterBarrierAsync() + { + await start.Task.WaitAsync(cancellationToken); + return await current.ReadRowAsync(overlappingEntry.SiloAddress, cancellationToken); + } + } + + Assert.Equal(queries.OrderBy(pair => pair.Key), (await ReadQueriesAsync(storage, cancellationToken)).OrderBy(pair => pair.Key)); + async Task AssertRejectedAsync(Func> write) + { + Assert.False(await write()); + AssertUnchanged(stable, await ReadSnapshotAsync(storage, cancellationToken)); + } + } + + private static MembershipEntry Entry(int id, SiloStatus status) => new() + { + SiloAddress = SiloAddress.New(IPAddress.Loopback, 11000 + id, 100 + id), + SiloName = $"silo-{id}", + HostName = $"host-{id}", + Status = status, + StartTime = StartTime, + IAmAliveTime = StartTime + }; + + private static async Task> ReadQueriesAsync(IRelationalStorage storage, CancellationToken cancellationToken) => + (await storage.ReadAsync(DbStoredQueries.GetQueriesKey, DbStoredQueries.Converters.GetQueryKeyAndValue, null, cancellationToken)) + .ToDictionary(pair => pair.Key, pair => pair.Value); + + private static async Task ReadSnapshotAsync(IRelationalStorage storage, CancellationToken cancellationToken) + { + var version = Assert.Single(await storage.ReadAsync( + "SELECT Version, Timestamp FROM OrleansMembershipVersionTable WHERE DeploymentId = @DeploymentId", + record => (Version: record.GetInt32(0), Timestamp: record.GetDateTime(1)), + command => _ = new DbStoredQueries.Columns(command) { DeploymentId = ClusterId }, cancellationToken)); + var members = await storage.ReadAsync( + """ + SELECT DeploymentId, Address, Port, Generation, SiloName, HostName, Status, ProxyPort, SuspectTimes, StartTime, IAmAliveTime + FROM OrleansMembershipTable WHERE DeploymentId = @DeploymentId ORDER BY Address, Port, Generation + """, + record => new StoredMember( + record.GetString(0), + SiloAddress.New(IPAddress.Parse(record.GetString(1)), record.GetInt32(2), record.GetInt32(3)), + record.GetString(4), record.GetString(5), (SiloStatus)record.GetInt32(6), record.GetInt32(7), + record.IsDBNull(8) ? null : record.GetString(8), record.GetDateTime(9), record.GetDateTime(10)), + command => _ = new DbStoredQueries.Columns(command) { DeploymentId = ClusterId }, cancellationToken); + return new Snapshot(version.Version, version.Timestamp, members.ToArray()); + } + + private static void AssertStored(Snapshot snapshot, int version, params MembershipEntry[] entries) + { + Assert.Equal(version, snapshot.Version); + Assert.Equal(entries.OrderBy(entry => entry.SiloAddress.Endpoint.Port).Select(StoredMember.FromEntry), snapshot.Members); + } + + private static void AssertUnchanged(Snapshot before, Snapshot after) + { + Assert.Equal(before.Version, after.Version); + Assert.Equal(before.VersionTimestamp, after.VersionTimestamp); + Assert.Equal(before.Members, after.Members); + } + + private static void AssertReadMatches(Snapshot expected, MembershipTableData actual) + { + Assert.Equal(expected.Version, actual.Version.Version); + Assert.Equal(expected.Etag, actual.Version.VersionEtag); + Assert.All(actual.Members, member => Assert.Equal(expected.Etag, member.Item2)); + Assert.Equal(expected.Members, actual.Members.OrderBy(member => member.Item1.SiloAddress.Endpoint.Port).Select(member => StoredMember.FromEntry(member.Item1))); + } + + private sealed record Snapshot(int Version, DateTime VersionTimestamp, StoredMember[] Members) + { + public string Etag => Version.ToString(CultureInfo.InvariantCulture); + public TableVersion NextVersion => new(Version + 1, Etag); + } + + private sealed record StoredMember( + string DeploymentId, SiloAddress SiloAddress, string SiloName, string HostName, SiloStatus Status, + int ProxyPort, string? SuspectTimes, DateTime StartTime, DateTime IAmAliveTime) + { + public static StoredMember FromEntry(MembershipEntry entry) => new( + ClusterId, entry.SiloAddress, entry.SiloName, entry.HostName, entry.Status, entry.ProxyPort, + entry.SuspectTimes is null ? null : string.Join("|", entry.SuspectTimes.Select(vote => $"{vote.Item1.ToParsableString()},{LogFormatter.PrintDate(vote.Item2)}")), + entry.StartTime, entry.IAmAliveTime); + } +} diff --git a/test/Extensions/Orleans.AdoNet.Tests/StorageTests/Relational/MembershipUpgradeFixtures/MySQL-Clustering.sql b/test/Extensions/Orleans.AdoNet.Tests/StorageTests/Relational/MembershipUpgradeFixtures/MySQL-Clustering.sql new file mode 100644 index 00000000000..3016b97ad73 --- /dev/null +++ b/test/Extensions/Orleans.AdoNet.Tests/StorageTests/Relational/MembershipUpgradeFixtures/MySQL-Clustering.sql @@ -0,0 +1,266 @@ +-- For each deployment, there will be only one (active) membership version table version column which will be updated periodically. +CREATE TABLE OrleansMembershipVersionTable +( + DeploymentId NVARCHAR(150) NOT NULL, + Timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + Version INT NOT NULL DEFAULT 0, + + CONSTRAINT PK_OrleansMembershipVersionTable_DeploymentId PRIMARY KEY(DeploymentId) +); + +-- Every silo instance has a row in the membership table. +CREATE TABLE OrleansMembershipTable +( + DeploymentId NVARCHAR(150) NOT NULL, + Address VARCHAR(45) NOT NULL, + Port INT NOT NULL, + Generation INT NOT NULL, + SiloName NVARCHAR(150) NOT NULL, + HostName NVARCHAR(150) NOT NULL, + Status INT NOT NULL, + ProxyPort INT NULL, + SuspectTimes VARCHAR(8000) NULL, + StartTime DATETIME NOT NULL, + IAmAliveTime DATETIME NOT NULL, + + CONSTRAINT PK_MembershipTable_DeploymentId PRIMARY KEY(DeploymentId, Address, Port, Generation), + CONSTRAINT FK_MembershipTable_MembershipVersionTable_DeploymentId FOREIGN KEY (DeploymentId) REFERENCES OrleansMembershipVersionTable (DeploymentId) +); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +VALUES +( + 'UpdateIAmAlivetimeKey',' + -- This is expected to never fail by Orleans, so return value + -- is not needed nor is it checked. + UPDATE OrleansMembershipTable + SET + IAmAliveTime = @IAmAliveTime + WHERE + DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL + AND Address = @Address AND @Address IS NOT NULL + AND Port = @Port AND @Port IS NOT NULL + AND Generation = @Generation AND @Generation IS NOT NULL; +'); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +VALUES +( + 'InsertMembershipVersionKey',' + INSERT INTO OrleansMembershipVersionTable + ( + DeploymentId + ) + SELECT * FROM ( SELECT @DeploymentId ) AS TMP + WHERE NOT EXISTS + ( + SELECT 1 + FROM + OrleansMembershipVersionTable + WHERE + DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL + ); + + SELECT ROW_COUNT(); +'); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +VALUES +( + 'InsertMembershipKey',' + call InsertMembershipKey(@DeploymentId, @Address, @Port, @Generation, + @Version, @SiloName, @HostName, @Status, @ProxyPort, @StartTime, @IAmAliveTime);' +); + +DELIMITER $$ + +CREATE PROCEDURE InsertMembershipKey( + in _DeploymentId NVARCHAR(150), + in _Address VARCHAR(45), + in _Port INT, + in _Generation INT, + in _Version INT, + in _SiloName NVARCHAR(150), + in _HostName NVARCHAR(150), + in _Status INT, + in _ProxyPort INT, + in _StartTime DATETIME, + in _IAmAliveTime DATETIME +) +BEGIN + DECLARE _ROWCOUNT INT; + START TRANSACTION; + INSERT INTO OrleansMembershipTable + ( + DeploymentId, + Address, + Port, + Generation, + SiloName, + HostName, + Status, + ProxyPort, + StartTime, + IAmAliveTime + ) + SELECT * FROM ( SELECT + _DeploymentId, + _Address, + _Port, + _Generation, + _SiloName, + _HostName, + _Status, + _ProxyPort, + _StartTime, + _IAmAliveTime) AS TMP + WHERE NOT EXISTS + ( + SELECT 1 + FROM + OrleansMembershipTable + WHERE + DeploymentId = _DeploymentId AND _DeploymentId IS NOT NULL + AND Address = _Address AND _Address IS NOT NULL + AND Port = _Port AND _Port IS NOT NULL + AND Generation = _Generation AND _Generation IS NOT NULL + ); + + UPDATE OrleansMembershipVersionTable + SET + Version = Version + 1 + WHERE + DeploymentId = _DeploymentId AND _DeploymentId IS NOT NULL + AND Version = _Version AND _Version IS NOT NULL + AND ROW_COUNT() > 0; + + SET _ROWCOUNT = ROW_COUNT(); + + IF _ROWCOUNT = 0 + THEN + ROLLBACK; + ELSE + COMMIT; + END IF; + SELECT _ROWCOUNT; +END$$ + +DELIMITER ; + +INSERT INTO OrleansQuery(QueryKey, QueryText) +VALUES +( + 'UpdateMembershipKey',' + START TRANSACTION; + + UPDATE OrleansMembershipVersionTable + SET + Version = Version + 1 + WHERE + DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL + AND Version = @Version AND @Version IS NOT NULL; + + UPDATE OrleansMembershipTable + SET + Status = @Status, + SuspectTimes = @SuspectTimes, + IAmAliveTime = @IAmAliveTime + WHERE + DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL + AND Address = @Address AND @Address IS NOT NULL + AND Port = @Port AND @Port IS NOT NULL + AND Generation = @Generation AND @Generation IS NOT NULL + AND ROW_COUNT() > 0; + + SELECT ROW_COUNT(); + COMMIT; +'); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +VALUES +( + 'GatewaysQueryKey',' + SELECT + Address, + ProxyPort, + Generation + FROM + OrleansMembershipTable + WHERE + DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL + AND Status = @Status AND @Status IS NOT NULL + AND ProxyPort > 0; +'); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +VALUES +( + 'MembershipReadRowKey',' + SELECT + v.DeploymentId, + m.Address, + m.Port, + m.Generation, + m.SiloName, + m.HostName, + m.Status, + m.ProxyPort, + m.SuspectTimes, + m.StartTime, + m.IAmAliveTime, + v.Version + FROM + OrleansMembershipVersionTable v + -- This ensures the version table will returned even if there is no matching membership row. + LEFT OUTER JOIN OrleansMembershipTable m ON v.DeploymentId = m.DeploymentId + AND Address = @Address AND @Address IS NOT NULL + AND Port = @Port AND @Port IS NOT NULL + AND Generation = @Generation AND @Generation IS NOT NULL + WHERE + v.DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL; +'); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +VALUES +( + 'MembershipReadAllKey',' + SELECT + v.DeploymentId, + m.Address, + m.Port, + m.Generation, + m.SiloName, + m.HostName, + m.Status, + m.ProxyPort, + m.SuspectTimes, + m.StartTime, + m.IAmAliveTime, + v.Version + FROM + OrleansMembershipVersionTable v LEFT OUTER JOIN OrleansMembershipTable m + ON v.DeploymentId = m.DeploymentId + WHERE + v.DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL; +'); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +VALUES +( + 'DeleteMembershipTableEntriesKey',' + DELETE FROM OrleansMembershipTable + WHERE DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL; + DELETE FROM OrleansMembershipVersionTable + WHERE DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL; +'); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +VALUES +( + 'CleanupDefunctSiloEntriesKey',' + DELETE FROM OrleansMembershipTable + WHERE DeploymentId = @DeploymentId + AND @DeploymentId IS NOT NULL + AND IAmAliveTime < @IAmAliveTime + AND Status !=3; +'); diff --git a/test/Extensions/Orleans.AdoNet.Tests/StorageTests/Relational/MembershipUpgradeFixtures/PostgreSQL-Clustering.sql b/test/Extensions/Orleans.AdoNet.Tests/StorageTests/Relational/MembershipUpgradeFixtures/PostgreSQL-Clustering.sql new file mode 100644 index 00000000000..459e76b579d --- /dev/null +++ b/test/Extensions/Orleans.AdoNet.Tests/StorageTests/Relational/MembershipUpgradeFixtures/PostgreSQL-Clustering.sql @@ -0,0 +1,357 @@ +-- For each deployment, there will be only one (active) membership version table version column which will be updated periodically. +CREATE TABLE OrleansMembershipVersionTable +( + DeploymentId varchar(150) NOT NULL, + Timestamp timestamptz(3) NOT NULL DEFAULT now(), + Version integer NOT NULL DEFAULT 0, + + CONSTRAINT PK_OrleansMembershipVersionTable_DeploymentId PRIMARY KEY(DeploymentId) +); + +-- Every silo instance has a row in the membership table. +CREATE TABLE OrleansMembershipTable +( + DeploymentId varchar(150) NOT NULL, + Address varchar(45) NOT NULL, + Port integer NOT NULL, + Generation integer NOT NULL, + SiloName varchar(150) NOT NULL, + HostName varchar(150) NOT NULL, + Status integer NOT NULL, + ProxyPort integer NULL, + SuspectTimes varchar(8000) NULL, + StartTime timestamptz(3) NOT NULL, + IAmAliveTime timestamptz(3) NOT NULL, + + CONSTRAINT PK_MembershipTable_DeploymentId PRIMARY KEY(DeploymentId, Address, Port, Generation), + CONSTRAINT FK_MembershipTable_MembershipVersionTable_DeploymentId FOREIGN KEY (DeploymentId) REFERENCES OrleansMembershipVersionTable (DeploymentId) +); + +CREATE FUNCTION update_i_am_alive_time( + deployment_id OrleansMembershipTable.DeploymentId%TYPE, + address_arg OrleansMembershipTable.Address%TYPE, + port_arg OrleansMembershipTable.Port%TYPE, + generation_arg OrleansMembershipTable.Generation%TYPE, + i_am_alive_time OrleansMembershipTable.IAmAliveTime%TYPE) + RETURNS void AS +$func$ +BEGIN + -- This is expected to never fail by Orleans, so return value + -- is not needed nor is it checked. + UPDATE OrleansMembershipTable as d + SET + IAmAliveTime = i_am_alive_time + WHERE + d.DeploymentId = deployment_id AND deployment_id IS NOT NULL + AND d.Address = address_arg AND address_arg IS NOT NULL + AND d.Port = port_arg AND port_arg IS NOT NULL + AND d.Generation = generation_arg AND generation_arg IS NOT NULL; +END +$func$ LANGUAGE plpgsql; + +INSERT INTO OrleansQuery(QueryKey, QueryText) +VALUES +( + 'UpdateIAmAlivetimeKey',' + -- This is expected to never fail by Orleans, so return value + -- is not needed nor is it checked. + SELECT * from update_i_am_alive_time( + @DeploymentId, + @Address, + @Port, + @Generation, + @IAmAliveTime + ); +'); + +CREATE FUNCTION insert_membership_version( + DeploymentIdArg OrleansMembershipTable.DeploymentId%TYPE +) + RETURNS TABLE(row_count integer) AS +$func$ +DECLARE + RowCountVar int := 0; +BEGIN + + BEGIN + + INSERT INTO OrleansMembershipVersionTable + ( + DeploymentId + ) + SELECT DeploymentIdArg + ON CONFLICT (DeploymentId) DO NOTHING; + + GET DIAGNOSTICS RowCountVar = ROW_COUNT; + + ASSERT RowCountVar <> 0, 'no rows affected, rollback'; + + RETURN QUERY SELECT RowCountVar; + EXCEPTION + WHEN assert_failure THEN + RETURN QUERY SELECT RowCountVar; + END; + +END +$func$ LANGUAGE plpgsql; + +INSERT INTO OrleansQuery(QueryKey, QueryText) +VALUES +( + 'InsertMembershipVersionKey',' + SELECT * FROM insert_membership_version( + @DeploymentId + ); +'); + +CREATE FUNCTION insert_membership( + DeploymentIdArg OrleansMembershipTable.DeploymentId%TYPE, + AddressArg OrleansMembershipTable.Address%TYPE, + PortArg OrleansMembershipTable.Port%TYPE, + GenerationArg OrleansMembershipTable.Generation%TYPE, + SiloNameArg OrleansMembershipTable.SiloName%TYPE, + HostNameArg OrleansMembershipTable.HostName%TYPE, + StatusArg OrleansMembershipTable.Status%TYPE, + ProxyPortArg OrleansMembershipTable.ProxyPort%TYPE, + StartTimeArg OrleansMembershipTable.StartTime%TYPE, + IAmAliveTimeArg OrleansMembershipTable.IAmAliveTime%TYPE, + VersionArg OrleansMembershipVersionTable.Version%TYPE) + RETURNS TABLE(row_count integer) AS +$func$ +DECLARE + RowCountVar int := 0; +BEGIN + + BEGIN + INSERT INTO OrleansMembershipTable + ( + DeploymentId, + Address, + Port, + Generation, + SiloName, + HostName, + Status, + ProxyPort, + StartTime, + IAmAliveTime + ) + SELECT + DeploymentIdArg, + AddressArg, + PortArg, + GenerationArg, + SiloNameArg, + HostNameArg, + StatusArg, + ProxyPortArg, + StartTimeArg, + IAmAliveTimeArg + ON CONFLICT (DeploymentId, Address, Port, Generation) DO + NOTHING; + + + GET DIAGNOSTICS RowCountVar = ROW_COUNT; + + UPDATE OrleansMembershipVersionTable + SET + Timestamp = now(), + Version = Version + 1 + WHERE + DeploymentId = DeploymentIdArg AND DeploymentIdArg IS NOT NULL + AND Version = VersionArg AND VersionArg IS NOT NULL + AND RowCountVar > 0; + + GET DIAGNOSTICS RowCountVar = ROW_COUNT; + + ASSERT RowCountVar <> 0, 'no rows affected, rollback'; + + + RETURN QUERY SELECT RowCountVar; + EXCEPTION + WHEN assert_failure THEN + RETURN QUERY SELECT RowCountVar; + END; + +END +$func$ LANGUAGE plpgsql; + +INSERT INTO OrleansQuery(QueryKey, QueryText) +VALUES +( + 'InsertMembershipKey',' + SELECT * FROM insert_membership( + @DeploymentId, + @Address, + @Port, + @Generation, + @SiloName, + @HostName, + @Status, + @ProxyPort, + @StartTime, + @IAmAliveTime, + @Version + ); +'); + +CREATE FUNCTION update_membership( + DeploymentIdArg OrleansMembershipTable.DeploymentId%TYPE, + AddressArg OrleansMembershipTable.Address%TYPE, + PortArg OrleansMembershipTable.Port%TYPE, + GenerationArg OrleansMembershipTable.Generation%TYPE, + StatusArg OrleansMembershipTable.Status%TYPE, + SuspectTimesArg OrleansMembershipTable.SuspectTimes%TYPE, + IAmAliveTimeArg OrleansMembershipTable.IAmAliveTime%TYPE, + VersionArg OrleansMembershipVersionTable.Version%TYPE + ) + RETURNS TABLE(row_count integer) AS +$func$ +DECLARE + RowCountVar int := 0; +BEGIN + + BEGIN + + UPDATE OrleansMembershipVersionTable + SET + Timestamp = now(), + Version = Version + 1 + WHERE + DeploymentId = DeploymentIdArg AND DeploymentIdArg IS NOT NULL + AND Version = VersionArg AND VersionArg IS NOT NULL; + + + GET DIAGNOSTICS RowCountVar = ROW_COUNT; + + UPDATE OrleansMembershipTable + SET + Status = StatusArg, + SuspectTimes = SuspectTimesArg, + IAmAliveTime = IAmAliveTimeArg + WHERE + DeploymentId = DeploymentIdArg AND DeploymentIdArg IS NOT NULL + AND Address = AddressArg AND AddressArg IS NOT NULL + AND Port = PortArg AND PortArg IS NOT NULL + AND Generation = GenerationArg AND GenerationArg IS NOT NULL + AND RowCountVar > 0; + + + GET DIAGNOSTICS RowCountVar = ROW_COUNT; + + ASSERT RowCountVar <> 0, 'no rows affected, rollback'; + + + RETURN QUERY SELECT RowCountVar; + EXCEPTION + WHEN assert_failure THEN + RETURN QUERY SELECT RowCountVar; + END; + +END +$func$ LANGUAGE plpgsql; + +INSERT INTO OrleansQuery(QueryKey, QueryText) +VALUES +( + 'UpdateMembershipKey',' + SELECT * FROM update_membership( + @DeploymentId, + @Address, + @Port, + @Generation, + @Status, + @SuspectTimes, + @IAmAliveTime, + @Version + ); +'); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +VALUES +( + 'MembershipReadRowKey',' + SELECT + v.DeploymentId, + m.Address, + m.Port, + m.Generation, + m.SiloName, + m.HostName, + m.Status, + m.ProxyPort, + m.SuspectTimes, + m.StartTime, + m.IAmAliveTime, + v.Version + FROM + OrleansMembershipVersionTable v + -- This ensures the version table will returned even if there is no matching membership row. + LEFT OUTER JOIN OrleansMembershipTable m ON v.DeploymentId = m.DeploymentId + AND Address = @Address AND @Address IS NOT NULL + AND Port = @Port AND @Port IS NOT NULL + AND Generation = @Generation AND @Generation IS NOT NULL + WHERE + v.DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL; +'); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +VALUES +( + 'MembershipReadAllKey',' + SELECT + v.DeploymentId, + m.Address, + m.Port, + m.Generation, + m.SiloName, + m.HostName, + m.Status, + m.ProxyPort, + m.SuspectTimes, + m.StartTime, + m.IAmAliveTime, + v.Version + FROM + OrleansMembershipVersionTable v LEFT OUTER JOIN OrleansMembershipTable m + ON v.DeploymentId = m.DeploymentId + WHERE + v.DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL; +'); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +VALUES +( + 'DeleteMembershipTableEntriesKey',' + DELETE FROM OrleansMembershipTable + WHERE DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL; + DELETE FROM OrleansMembershipVersionTable + WHERE DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL; +'); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +VALUES +( + 'GatewaysQueryKey',' + SELECT + Address, + ProxyPort, + Generation + FROM + OrleansMembershipTable + WHERE + DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL + AND Status = @Status AND @Status IS NOT NULL + AND ProxyPort > 0; +'); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +VALUES +( + 'CleanupDefunctSiloEntriesKey', + 'DELETE FROM OrleansMembershipTable + WHERE DeploymentId = @DeploymentId + AND @DeploymentId IS NOT NULL + AND IAmAliveTime < @IAmAliveTime + AND Status != 3; +'); diff --git a/test/Extensions/Orleans.AdoNet.Tests/StorageTests/Relational/MembershipUpgradeFixtures/SQLServer-Clustering.sql b/test/Extensions/Orleans.AdoNet.Tests/StorageTests/Relational/MembershipUpgradeFixtures/SQLServer-Clustering.sql new file mode 100644 index 00000000000..b89d81e9946 --- /dev/null +++ b/test/Extensions/Orleans.AdoNet.Tests/StorageTests/Relational/MembershipUpgradeFixtures/SQLServer-Clustering.sql @@ -0,0 +1,295 @@ +-- For each deployment, there will be only one (active) membership version table version column which will be updated periodically. +IF OBJECT_ID(N'[OrleansMembershipVersionTable]', 'U') IS NULL +CREATE TABLE OrleansMembershipVersionTable +( + DeploymentId NVARCHAR(150) NOT NULL, + Timestamp DATETIME2(3) NOT NULL DEFAULT GETUTCDATE(), + Version INT NOT NULL DEFAULT 0, + + CONSTRAINT PK_OrleansMembershipVersionTable_DeploymentId PRIMARY KEY(DeploymentId) +); + +-- Every silo instance has a row in the membership table. +IF OBJECT_ID(N'[OrleansMembershipTable]', 'U') IS NULL +CREATE TABLE OrleansMembershipTable +( + DeploymentId NVARCHAR(150) NOT NULL, + Address VARCHAR(45) NOT NULL, + Port INT NOT NULL, + Generation INT NOT NULL, + SiloName NVARCHAR(150) NOT NULL, + HostName NVARCHAR(150) NOT NULL, + Status INT NOT NULL, + ProxyPort INT NULL, + SuspectTimes VARCHAR(8000) NULL, + StartTime DATETIME2(3) NOT NULL, + IAmAliveTime DATETIME2(3) NOT NULL, + + CONSTRAINT PK_MembershipTable_DeploymentId PRIMARY KEY(DeploymentId, Address, Port, Generation), + CONSTRAINT FK_MembershipTable_MembershipVersionTable_DeploymentId FOREIGN KEY (DeploymentId) REFERENCES OrleansMembershipVersionTable (DeploymentId) +); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +SELECT + 'UpdateIAmAlivetimeKey', + '-- This is expected to never fail by Orleans, so return value + -- is not needed nor is it checked. + SET NOCOUNT ON; + UPDATE OrleansMembershipTable + SET + IAmAliveTime = @IAmAliveTime + WHERE + DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL + AND Address = @Address AND @Address IS NOT NULL + AND Port = @Port AND @Port IS NOT NULL + AND Generation = @Generation AND @Generation IS NOT NULL; + ' +WHERE NOT EXISTS +( + SELECT 1 + FROM OrleansQuery oqt + WHERE oqt.[QueryKey] = 'UpdateIAmAlivetimeKey' +); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +SELECT + 'InsertMembershipVersionKey', + 'SET NOCOUNT ON; + INSERT INTO OrleansMembershipVersionTable + ( + DeploymentId + ) + SELECT @DeploymentId + WHERE NOT EXISTS + ( + SELECT 1 + FROM + OrleansMembershipVersionTable WITH(HOLDLOCK, XLOCK, ROWLOCK) + WHERE + DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL + ); + + SELECT @@ROWCOUNT; + ' +WHERE NOT EXISTS +( + SELECT 1 + FROM OrleansQuery oqt + WHERE oqt.[QueryKey] = 'InsertMembershipVersionKey' +); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +SELECT + 'InsertMembershipKey', + 'SET XACT_ABORT, NOCOUNT ON; + DECLARE @ROWCOUNT AS INT; + BEGIN TRANSACTION; + INSERT INTO OrleansMembershipTable + ( + DeploymentId, + Address, + Port, + Generation, + SiloName, + HostName, + Status, + ProxyPort, + StartTime, + IAmAliveTime + ) + SELECT + @DeploymentId, + @Address, + @Port, + @Generation, + @SiloName, + @HostName, + @Status, + @ProxyPort, + @StartTime, + @IAmAliveTime + WHERE NOT EXISTS + ( + SELECT 1 + FROM + OrleansMembershipTable WITH(HOLDLOCK, XLOCK, ROWLOCK) + WHERE + DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL + AND Address = @Address AND @Address IS NOT NULL + AND Port = @Port AND @Port IS NOT NULL + AND Generation = @Generation AND @Generation IS NOT NULL + ); + + UPDATE OrleansMembershipVersionTable + SET + Timestamp = GETUTCDATE(), + Version = Version + 1 + WHERE + DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL + AND Version = @Version AND @Version IS NOT NULL + AND @@ROWCOUNT > 0; + + SET @ROWCOUNT = @@ROWCOUNT; + + IF @ROWCOUNT = 0 + ROLLBACK TRANSACTION + ELSE + COMMIT TRANSACTION + SELECT @ROWCOUNT; + ' +WHERE NOT EXISTS +( + SELECT 1 + FROM OrleansQuery oqt + WHERE oqt.[QueryKey] = 'InsertMembershipKey' +); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +SELECT + 'UpdateMembershipKey', + 'SET XACT_ABORT, NOCOUNT ON; + BEGIN TRANSACTION; + + UPDATE OrleansMembershipVersionTable + SET + Timestamp = GETUTCDATE(), + Version = Version + 1 + WHERE + DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL + AND Version = @Version AND @Version IS NOT NULL; + + UPDATE OrleansMembershipTable + SET + Status = @Status, + SuspectTimes = @SuspectTimes, + IAmAliveTime = @IAmAliveTime + WHERE + DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL + AND Address = @Address AND @Address IS NOT NULL + AND Port = @Port AND @Port IS NOT NULL + AND Generation = @Generation AND @Generation IS NOT NULL + AND @@ROWCOUNT > 0; + + SELECT @@ROWCOUNT; + COMMIT TRANSACTION; + ' +WHERE NOT EXISTS +( + SELECT 1 + FROM OrleansQuery oqt + WHERE oqt.[QueryKey] = 'UpdateMembershipKey' +); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +SELECT + 'GatewaysQueryKey', + 'SELECT + Address, + ProxyPort, + Generation + FROM + OrleansMembershipTable + WHERE + DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL + AND Status = @Status AND @Status IS NOT NULL + AND ProxyPort > 0; + ' +WHERE NOT EXISTS +( + SELECT 1 + FROM OrleansQuery oqt + WHERE oqt.[QueryKey] = 'GatewaysQueryKey' +); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +SELECT + 'MembershipReadRowKey', + 'SELECT + v.DeploymentId, + m.Address, + m.Port, + m.Generation, + m.SiloName, + m.HostName, + m.Status, + m.ProxyPort, + m.SuspectTimes, + m.StartTime, + m.IAmAliveTime, + v.Version + FROM + OrleansMembershipVersionTable v + -- This ensures the version table will returned even if there is no matching membership row. + LEFT OUTER JOIN OrleansMembershipTable m ON v.DeploymentId = m.DeploymentId + AND Address = @Address AND @Address IS NOT NULL + AND Port = @Port AND @Port IS NOT NULL + AND Generation = @Generation AND @Generation IS NOT NULL + WHERE + v.DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL; + ' +WHERE NOT EXISTS +( + SELECT 1 + FROM OrleansQuery oqt + WHERE oqt.[QueryKey] = 'MembershipReadRowKey' +); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +SELECT + 'MembershipReadAllKey', + 'SELECT + v.DeploymentId, + m.Address, + m.Port, + m.Generation, + m.SiloName, + m.HostName, + m.Status, + m.ProxyPort, + m.SuspectTimes, + m.StartTime, + m.IAmAliveTime, + v.Version + FROM + OrleansMembershipVersionTable v LEFT OUTER JOIN OrleansMembershipTable m + ON v.DeploymentId = m.DeploymentId + WHERE + v.DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL; + ' +WHERE NOT EXISTS +( + SELECT 1 + FROM OrleansQuery oqt + WHERE oqt.[QueryKey] = 'MembershipReadAllKey' +); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +SELECT + 'DeleteMembershipTableEntriesKey', + 'DELETE FROM OrleansMembershipTable + WHERE DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL; + DELETE FROM OrleansMembershipVersionTable + WHERE DeploymentId = @DeploymentId AND @DeploymentId IS NOT NULL; + ' +WHERE NOT EXISTS +( + SELECT 1 + FROM OrleansQuery oqt + WHERE oqt.[QueryKey] = 'DeleteMembershipTableEntriesKey' +); + +INSERT INTO OrleansQuery(QueryKey, QueryText) +SELECT + 'CleanupDefunctSiloEntriesKey', + 'DELETE FROM OrleansMembershipTable + WHERE DeploymentId = @DeploymentId + AND @DeploymentId IS NOT NULL + AND IAmAliveTime < @IAmAliveTime + AND Status != 3; + ' +WHERE NOT EXISTS +( + SELECT 1 + FROM OrleansQuery oqt + WHERE oqt.[QueryKey] = 'CleanupDefunctSiloEntriesKey' +); diff --git a/test/Extensions/Orleans.AdoNet.Tests/StorageTests/Relational/RelationalOrleansQueriesUnitTests.cs b/test/Extensions/Orleans.AdoNet.Tests/StorageTests/Relational/RelationalOrleansQueriesUnitTests.cs index 248a14d42cd..a34c24de9c9 100644 --- a/test/Extensions/Orleans.AdoNet.Tests/StorageTests/Relational/RelationalOrleansQueriesUnitTests.cs +++ b/test/Extensions/Orleans.AdoNet.Tests/StorageTests/Relational/RelationalOrleansQueriesUnitTests.cs @@ -1,8 +1,13 @@ using System.Data; using System.Net; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; using Orleans; +using Orleans.Configuration; using Orleans.GrainDirectory.AdoNet; using Orleans.Runtime; +using Orleans.Runtime.MembershipService; using Orleans.Streaming.AdoNet; using UnitTests.StorageTests.Relational.Fakes; using ClusteringQueries = Orleans.Clustering.AdoNet.Storage.RelationalOrleansQueries; @@ -42,6 +47,7 @@ public sealed class RelationalOrleansQueriesUnitTests "UpdateMembershipKey", "DeleteMembershipTableEntriesKey", "CleanupDefunctSiloEntriesKey", + "CleanupDefunctSiloEntryKey", ]; private static readonly string[] StreamingQueryKeys = @@ -271,6 +277,7 @@ public async Task ReadAllMembershipRows_ConvertsRowsAndVersion() Assert.Null(member.SuspectTimes); Assert.Equal(42, result.Version.Version); Assert.Equal("42", result.Version.VersionEtag); + Assert.Equal("42", Assert.Single(result.Members).Item2); AssertOperationCall(storage, Sql("MembershipReadAllKey")); storage.VerifyComplete(); } @@ -951,7 +958,7 @@ public async Task MembershipReadRowAsync_ReturnsEntryAndCapturesParameters() var memberAndEtag = Assert.Single(result.Members); var member = memberAndEtag.Item1; - Assert.Equal(string.Empty, memberAndEtag.Item2); + Assert.Equal("106", memberAndEtag.Item2); Assert.Equal(address, member.SiloAddress); Assert.Equal("silo-read-row", member.SiloName); Assert.Equal("host-read-row", member.HostName); @@ -1188,13 +1195,15 @@ await queries.EvictStreamDeadLettersAsync( storage.VerifyComplete(); } - [Fact] - public async Task UpdateIAmAliveTimeAsync_ExecutesSentinelQueryAndCapturesUtcValue() + [Theory] + [InlineData(0)] + [InlineData(1)] + public async Task UpdateIAmAliveTimeAsync_ExecutesOneBlindWriteWithKeyAndTimestamp(int affectedRows) { var address = SiloAddress.New(IPAddress.Parse("10.50.60.70"), 12_345, 9); var aliveTime = new DateTime(2026, 8, 27, 22, 31, 45, DateTimeKind.Utc); var storage = ExpectQueryLoad(new ScriptedRelationalStorage(), MembershipQueryKeys) - .ExpectExecute(Sql("UpdateIAmAlivetimeKey"), affectedRows: 1); + .ExpectExecute(Sql("UpdateIAmAlivetimeKey"), affectedRows); var queries = await ClusteringQueries.CreateInstance(storage, TestContext.Current.CancellationToken); await queries.UpdateIAmAliveTimeAsync("cluster-alive", address, aliveTime, TestContext.Current.CancellationToken); @@ -1241,22 +1250,293 @@ public async Task DeleteMembershipTableEntriesAsync_ExecutesSentinelQueryAndCapt storage.VerifyComplete(); } - [Fact] - public async Task CleanupDefunctSiloEntriesAsync_ExecutesSentinelQueryAndCapturesUtcDateTime() + [Theory] + [InlineData("start")] + [InlineData("heartbeat")] + [InlineData("vote")] + public async Task CleanupDefunctSiloEntriesAsync_UsesExactExclusiveCutoffAndCapturedValues(string latest) { - var beforeDate = new DateTimeOffset(2026, 8, 28, 4, 15, 30, TimeSpan.FromHours(5.5)); + var boundary = new DateTime(2026, 8, 27, 22, 45, 30, 123, DateTimeKind.Utc); + var start = latest == "start" ? boundary : boundary.AddMinutes(-3); + var heartbeat = latest == "heartbeat" ? boundary : boundary.AddMinutes(-2); + var vote = latest == "vote" ? boundary : boundary.AddMinutes(-1); + var suspectTimes = $"10.4.5.8:11224@16,{LogFormatter.PrintDate(vote.AddMinutes(-1))}|10.4.5.7:11223@15,{LogFormatter.PrintDate(vote)}"; var storage = ExpectQueryLoad(new ScriptedRelationalStorage(), MembershipQueryKeys) - .ExpectExecute(Sql("CleanupDefunctSiloEntriesKey"), affectedRows: 5); + .ExpectRead(Sql("MembershipReadAllKey"), CreateMembershipCleanupRow(8, SiloStatus.Dead, start, heartbeat, suspectTimes)) + .ExpectRead(Sql("MembershipReadAllKey"), CreateMembershipCleanupRow(8, SiloStatus.Dead, start, heartbeat, suspectTimes)) + .ExpectExecute(Sql("CleanupDefunctSiloEntryKey"), affectedRows: 1); var queries = await ClusteringQueries.CreateInstance(storage, TestContext.Current.CancellationToken); + var cutoff = new DateTimeOffset(boundary).ToOffset(TimeSpan.FromHours(5.5)); - await queries.CleanupDefunctSiloEntriesAsync(beforeDate, "cluster-cleanup", TestContext.Current.CancellationToken); + await queries.CleanupDefunctSiloEntriesAsync(cutoff, "cluster-cleanup", TestContext.Current.CancellationToken); + Assert.Equal(2, storage.Calls.Count); + await queries.CleanupDefunctSiloEntriesAsync(cutoff.AddTicks(1), "cluster-cleanup", TestContext.Current.CancellationToken); - var call = AssertOperationCall(storage, Sql("CleanupDefunctSiloEntriesKey"), ExpectedCallKind.Execute); + Assert.Equal( + [GetQueriesSql, Sql("MembershipReadAllKey"), Sql("MembershipReadAllKey"), Sql("CleanupDefunctSiloEntryKey")], + storage.Calls.Select(call => call.Query)); + var deletion = Assert.Single(storage.Calls, call => call.Query == Sql("CleanupDefunctSiloEntryKey")); + Assert.Equal(ExpectedCallKind.Execute, deletion.Kind); AssertParameters( - call, + deletion, ("DeploymentId", "cluster-cleanup"), - ("IAmAliveTime", new DateTime(2026, 8, 27, 22, 45, 30, DateTimeKind.Utc))); - Assert.Equal(DateTimeKind.Utc, Assert.IsType(Parameter(call, "IAmAliveTime").Value).Kind); + ("Address", "10.4.5.6"), + ("Port", 11_222), + ("Generation", 14), + ("StartTime", start), + ("IAmAliveTime", heartbeat), + ("SuspectTimes", suspectTimes)); + Assert.Equal(TestContext.Current.CancellationToken, deletion.CancellationToken); + storage.VerifyComplete(); + } + + [Fact] + public async Task CleanupDefunctSiloEntriesAsync_PreservesEveryNonDeadStatus() + { + var time = new DateTime(2026, 8, 27, 22, 45, 30, DateTimeKind.Utc); + foreach (var status in Enum.GetValues().Where(status => status != SiloStatus.Dead)) + { + var storage = ExpectQueryLoad(new ScriptedRelationalStorage(), MembershipQueryKeys) + .ExpectRead(Sql("MembershipReadAllKey"), CreateMembershipCleanupRow(8, status, time, time)); + var queries = await ClusteringQueries.CreateInstance(storage, TestContext.Current.CancellationToken); + + await queries.CleanupDefunctSiloEntriesAsync(time.AddDays(1), "cluster-cleanup", TestContext.Current.CancellationToken); + + AssertOperationCall(storage, Sql("MembershipReadAllKey")); + storage.VerifyComplete(); + } + } + + [Fact] + public async Task CleanupDefunctSiloEntriesAsync_MaxVersionStillPrunesDeadRows() + { + var time = new DateTime(2026, 8, 27, 22, 45, 30, DateTimeKind.Utc); + var storage = ExpectQueryLoad(new ScriptedRelationalStorage(), MembershipQueryKeys) + .ExpectRead(Sql("MembershipReadAllKey"), CreateMembershipCleanupRow(int.MaxValue, SiloStatus.Dead, time, time)) + .ExpectExecute(Sql("CleanupDefunctSiloEntryKey"), affectedRows: 1); + var queries = await ClusteringQueries.CreateInstance(storage, TestContext.Current.CancellationToken); + + await queries.CleanupDefunctSiloEntriesAsync(time.AddTicks(1), "cluster-cleanup", TestContext.Current.CancellationToken); + + Assert.Equal(3, storage.Calls.Count); + AssertParameters(storage.Calls[2], + ("DeploymentId", "cluster-cleanup"), ("Address", "10.4.5.6"), ("Port", 11_222), ("Generation", 14), + ("StartTime", time), ("IAmAliveTime", time), ("SuspectTimes", DBNull.Value)); + storage.VerifyComplete(); + } + + [Fact] + public async Task CleanupDefunctSiloEntriesAsync_ContinuesAfterConcurrentRowChange() + { + var time = new DateTime(2026, 8, 27, 22, 45, 30, DateTimeKind.Utc); + var rows = CreateMembershipCleanupRow(8, SiloStatus.Dead, time, time); + var second = rows.NewRow(); + second.ItemArray = rows.Rows[0].ItemArray; + second["Generation"] = 15; + rows.Rows.Add(second); + var storage = ExpectQueryLoad(new ScriptedRelationalStorage(), MembershipQueryKeys) + .ExpectRead(Sql("MembershipReadAllKey"), rows) + .ExpectExecute(Sql("CleanupDefunctSiloEntryKey"), affectedRows: 0) + .ExpectExecute(Sql("CleanupDefunctSiloEntryKey"), affectedRows: 1); + var queries = await ClusteringQueries.CreateInstance(storage, TestContext.Current.CancellationToken); + + await queries.CleanupDefunctSiloEntriesAsync(time.AddTicks(1), "cluster-cleanup", TestContext.Current.CancellationToken); + + Assert.Equal( + [GetQueriesSql, Sql("MembershipReadAllKey"), Sql("CleanupDefunctSiloEntryKey"), Sql("CleanupDefunctSiloEntryKey")], + storage.Calls.Select(call => call.Query)); + Assert.Equal([14, 15], storage.Calls.Skip(2).Select(call => Assert.IsType(Parameter(call, "Generation").Value)).Order()); + storage.VerifyComplete(); + } + + [Theory] + [InlineData("Microsoft.Data.SqlClient", DbType.DateTime2)] + [InlineData("Npgsql", DbType.DateTime)] + [InlineData("MySql.Data.MySqlClient", DbType.DateTime)] + [InlineData("Oracle.ManagedDataAccess.Client", DbType.DateTime)] + public async Task CleanupDefunctSiloEntriesAsync_PreservesCapturedTimestampPrecision(string invariant, DbType expectedType) + { + var time = new DateTime(2026, 8, 27, 22, 45, 30, 123, DateTimeKind.Utc).AddTicks(4560); + var storage = ExpectQueryLoad(new ScriptedRelationalStorage(invariant), MembershipQueryKeys) + .ExpectRead(Sql("MembershipReadAllKey"), CreateMembershipCleanupRow(8, SiloStatus.Dead, time, time)) + .ExpectExecute(Sql("CleanupDefunctSiloEntryKey"), affectedRows: 1); + var queries = await ClusteringQueries.CreateInstance(storage, TestContext.Current.CancellationToken); + + await queries.CleanupDefunctSiloEntriesAsync(time.AddTicks(1), "cluster-cleanup", TestContext.Current.CancellationToken); + + Assert.Equal(3, storage.Calls.Count); + foreach (var name in new[] { "StartTime", "IAmAliveTime" }) + { + var parameter = Parameter(storage.Calls[2], name); + Assert.Equal(expectedType, parameter.DbType); + Assert.Equal(time, parameter.Value); + } + + storage.VerifyComplete(); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task CleanupDefunctSiloEntriesAsync_PropagatesStorageFailureWithoutRetry(bool canceled) + { + var time = new DateTime(2026, 8, 27, 22, 45, 30, DateTimeKind.Utc); + Exception failure = canceled ? new OperationCanceledException() : new InvalidOperationException("Storage unavailable."); + var storage = ExpectQueryLoad(new ScriptedRelationalStorage(), MembershipQueryKeys) + .ExpectRead(Sql("MembershipReadAllKey"), CreateMembershipCleanupRow(8, SiloStatus.Dead, time, time)) + .ExpectExecuteException(Sql("CleanupDefunctSiloEntryKey"), failure); + var queries = await ClusteringQueries.CreateInstance(storage, TestContext.Current.CancellationToken); + + Assert.Same(failure, await Record.ExceptionAsync( + () => queries.CleanupDefunctSiloEntriesAsync(time.AddTicks(1), "cluster-cleanup", TestContext.Current.CancellationToken))); + + Assert.Equal(3, storage.Calls.Count); + Assert.Equal(Sql("CleanupDefunctSiloEntryKey"), storage.Calls[2].Query); + storage.VerifyComplete(); + } + + [Fact] + public async Task CleanupDefunctSiloEntriesAsync_CancellationAfterReadPreventsDelete() + { + var time = new DateTime(2026, 8, 27, 22, 45, 30, DateTimeKind.Utc); + using var cancellation = new CancellationTokenSource(); + var storage = ExpectQueryLoad(new ScriptedRelationalStorage(), MembershipQueryKeys) + .ExpectRead(Sql("MembershipReadAllKey"), CreateMembershipCleanupRow(8, SiloStatus.Dead, time, time)); + var queries = await ClusteringQueries.CreateInstance(storage, TestContext.Current.CancellationToken); + storage.BeforeSelector = cancellation.Cancel; + + await Assert.ThrowsAnyAsync( + () => queries.CleanupDefunctSiloEntriesAsync(time.AddTicks(1), "cluster-cleanup", cancellation.Token)); + + AssertOperationCall(storage, Sql("MembershipReadAllKey")); + storage.VerifyComplete(); + } + + [Fact] + public async Task CleanupDefunctSiloEntriesAsync_UsesExistingQueryWithoutSqlUpdate() + { + var cutoff = new DateTimeOffset(2026, 8, 28, 4, 15, 30, TimeSpan.FromHours(5.5)); + var storage = ExpectQueryLoad(new ScriptedRelationalStorage(), + MembershipQueryKeys.Where(key => key != "CleanupDefunctSiloEntryKey").ToArray()) + .ExpectExecute(Sql("CleanupDefunctSiloEntriesKey"), affectedRows: 1); + var queries = await ClusteringQueries.CreateInstance(storage, TestContext.Current.CancellationToken); + + await queries.CleanupDefunctSiloEntriesAsync(cutoff, "cluster-cleanup", TestContext.Current.CancellationToken); + + var call = AssertOperationCall(storage, Sql("CleanupDefunctSiloEntriesKey"), ExpectedCallKind.Execute); + AssertParameters(call, ("DeploymentId", "cluster-cleanup"), ("IAmAliveTime", cutoff.UtcDateTime)); + Assert.Equal(TestContext.Current.CancellationToken, call.CancellationToken); + storage.VerifyComplete(); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task CleanupDefunctSiloEntriesAsync_ExistingQueryPropagatesNativeFailure(bool canceled) + { + Exception failure = canceled ? new OperationCanceledException() : new InvalidOperationException("Storage unavailable."); + var storage = ExpectQueryLoad(new ScriptedRelationalStorage(), + MembershipQueryKeys.Where(key => key != "CleanupDefunctSiloEntryKey").ToArray()) + .ExpectExecuteException(Sql("CleanupDefunctSiloEntriesKey"), failure); + var queries = await ClusteringQueries.CreateInstance(storage, TestContext.Current.CancellationToken); + + Assert.Same(failure, await Record.ExceptionAsync(() => + queries.CleanupDefunctSiloEntriesAsync(new DateTimeOffset(2026, 8, 28, 0, 0, 0, TimeSpan.Zero), "cluster-cleanup", TestContext.Current.CancellationToken))); + + AssertOperationCall(storage, Sql("CleanupDefunctSiloEntriesKey"), ExpectedCallKind.Execute); + storage.VerifyComplete(); + } + + [Fact] + public async Task CreateInstance_StillRequiresExistingMembershipQueries() + { + var storage = ExpectQueryLoad(new ScriptedRelationalStorage(), + MembershipQueryKeys.Where(key => key != "UpdateMembershipKey").ToArray()); + + var failure = await Assert.ThrowsAsync( + () => ClusteringQueries.CreateInstance(storage, TestContext.Current.CancellationToken)); + + Assert.Contains("UpdateMembershipKey", failure.Message, StringComparison.Ordinal); + AssertOnlyQueryLoadCall(storage); + storage.VerifyComplete(); + } + + [Theory] + [InlineData("")] + [InlineData(" 7")] + [InlineData("7 ")] + [InlineData("+7")] + [InlineData("07")] + [InlineData("-1")] + [InlineData("2147483648")] + [InlineData("opaque")] + [InlineData(null)] + public async Task MembershipWrites_RejectNoncanonicalTokensWithoutStorageEffects(string? etag) + { + var storage = ExpectQueryLoad(new ScriptedRelationalStorage(), MembershipQueryKeys); + var queries = await ClusteringQueries.CreateInstance(storage, TestContext.Current.CancellationToken); + + Assert.False(await queries.InsertMembershipRowAsync("cluster", new MembershipEntry(), etag!, TestContext.Current.CancellationToken)); + Assert.False(await queries.UpdateMembershipRowAsync("cluster", new MembershipEntry(), etag!, TestContext.Current.CancellationToken)); + + AssertOnlyQueryLoadCall(storage); + storage.VerifyComplete(); + } + + [Fact] + public async Task MembershipWrites_ObserveCancellationBeforeTokenComparison() + { + var storage = ExpectQueryLoad(new ScriptedRelationalStorage(), MembershipQueryKeys); + var queries = await ClusteringQueries.CreateInstance(storage, TestContext.Current.CancellationToken); + using var cancellation = new CancellationTokenSource(); + cancellation.Cancel(); + + await Assert.ThrowsAnyAsync( + () => queries.InsertMembershipRowAsync("cluster", new MembershipEntry(), "invalid", cancellation.Token)); + await Assert.ThrowsAnyAsync( + () => queries.UpdateMembershipRowAsync("cluster", new MembershipEntry(), "invalid", cancellation.Token)); + + AssertOnlyQueryLoadCall(storage); + storage.VerifyComplete(); + } + + [Theory] + [InlineData("")] + [InlineData("6")] + [InlineData("07")] + [InlineData("7 ")] + [InlineData(null)] + public async Task UpdateRowAsync_RejectsMismatchedRowTokenBeforeStorage(string? etag) + { + using var services = new ServiceCollection().BuildServiceProvider(); + var table = new AdoNetClusteringTable(services, Options.Create(new ClusterOptions()), + Options.Create(new AdoNetClusteringSiloOptions()), NullLogger.Instance); + + Assert.False(await table.UpdateRowAsync(new MembershipEntry(), etag!, new TableVersion(8, "7"), TestContext.Current.CancellationToken)); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task MembershipWrites_PropagateConditionalFailure(bool update) + { + var key = update ? "UpdateMembershipKey" : "InsertMembershipKey"; + var storage = ExpectQueryLoad(new ScriptedRelationalStorage(), MembershipQueryKeys) + .ExpectRead(Sql(key), CreateTable([("Result", typeof(int))], [0])); + var queries = await ClusteringQueries.CreateInstance(storage, TestContext.Current.CancellationToken); + var entry = new MembershipEntry + { + SiloAddress = SiloAddress.New(IPAddress.Loopback, 11_111, 1), + SiloName = "silo", + HostName = "host", + }; + + var result = update + ? await queries.UpdateMembershipRowAsync("cluster", entry, "7", TestContext.Current.CancellationToken) + : await queries.InsertMembershipRowAsync("cluster", entry, "7", TestContext.Current.CancellationToken); + + Assert.False(result); + Assert.Equal(7, Parameter(AssertOperationCall(storage, Sql(key)), "Version").Value); storage.VerifyComplete(); } @@ -1289,6 +1569,16 @@ public void GetQueryKeyAndValue_WithDuplicateColumns_ReturnsExpectedPair() Assert.False(reader.Read()); } + private static DataTable CreateMembershipCleanupRow(int version, SiloStatus status, DateTime start, DateTime heartbeat, string? suspectTimes = null) => + CreateTable( + [ + ("StartTime", typeof(DateTime)), ("Port", typeof(int)), ("Generation", typeof(int)), + ("Address", typeof(string)), ("SiloName", typeof(string)), ("HostName", typeof(string)), + ("Status", typeof(int)), ("ProxyPort", typeof(int)), ("IAmAliveTime", typeof(DateTime)), + ("SuspectTimes", typeof(string)), ("Version", typeof(long)), + ], + [start, 11_222, 14, "10.4.5.6", "silo-cleanup", "host-cleanup", (int)status, 30_004, heartbeat, suspectTimes, (long)version]); + private static readonly (string Name, Type Type)[] StreamMessageColumns = [ ("ServiceId", typeof(string)), diff --git a/test/Orleans.Runtime.Internal.Tests/MembershipTests/MembershipTableTestsBase.cs b/test/Orleans.Runtime.Internal.Tests/MembershipTests/MembershipTableTestsBase.cs index e99d16a1c5e..0ad65748915 100644 --- a/test/Orleans.Runtime.Internal.Tests/MembershipTests/MembershipTableTestsBase.cs +++ b/test/Orleans.Runtime.Internal.Tests/MembershipTests/MembershipTableTestsBase.cs @@ -460,17 +460,18 @@ protected async Task MembershipTable_CleanupDefunctSiloEntries(bool extendedProt Assert.True(ok, "InsertRow Dead failed"); newTableVersion = table.Version.Next(); - var oldEntryJoining = CreateMembershipEntryForTest(); - oldEntryJoining.IAmAliveTime = oldEntryJoining.IAmAliveTime.AddDays(-10); - oldEntryJoining.StartTime = oldEntryJoining.StartTime.AddDays(-10); - oldEntryJoining.Status = SiloStatus.Joining; - ok = await membershipTable.InsertRowAsync(oldEntryJoining, newTableVersion, cancellationToken); + var oldEntryActive = CreateMembershipEntryForTest(); + oldEntryActive.IAmAliveTime = oldEntryActive.IAmAliveTime.AddDays(-10); + oldEntryActive.StartTime = oldEntryActive.StartTime.AddDays(-10); + oldEntryActive.Status = SiloStatus.Active; + ok = await membershipTable.InsertRowAsync(oldEntryActive, newTableVersion, cancellationToken); table = await membershipTable.ReadAllAsync(cancellationToken); - Assert.True(ok, "InsertRow Joining failed"); + Assert.True(ok, "InsertRow Active failed"); newTableVersion = table.Version.Next(); var newEntry = CreateMembershipEntryForTest(); + newEntry.Status = SiloStatus.Active; ok = await membershipTable.InsertRowAsync(newEntry, newTableVersion, cancellationToken); Assert.True(ok, "InsertRow failed"); @@ -481,8 +482,9 @@ protected async Task MembershipTable_CleanupDefunctSiloEntries(bool extendedProt Assert.Equal(3, data.Members.Count); - // Every status other than Active should get cleared out if old - foreach (var siloStatus in Enum.GetValues()) + var activeAddresses = new List { oldEntryActive.SiloAddress, newEntry.SiloAddress }; + var deadAddresses = new List { oldEntryDead.SiloAddress }; + foreach (var siloStatus in new[] { SiloStatus.Active, SiloStatus.Dead }) { var oldEntry = CreateMembershipEntryForTest(); oldEntry.IAmAliveTime = oldEntry.IAmAliveTime.AddDays(-10); @@ -493,6 +495,7 @@ protected async Task MembershipTable_CleanupDefunctSiloEntries(bool extendedProt Assert.True(ok, "InsertRow failed"); + (siloStatus == SiloStatus.Active ? activeAddresses : deadAddresses).Add(oldEntry.SiloAddress); newTableVersion = table.Version.Next(); } @@ -501,7 +504,10 @@ protected async Task MembershipTable_CleanupDefunctSiloEntries(bool extendedProt data = await membershipTable.ReadAllAsync(cancellationToken); logger.LogInformation("Membership.ReadAll returned TableVersion={TableVersion} Data={Data}", data.Version, data); - Assert.Equal(2, data.Members.Count); + Assert.Equal(3, data.Members.Count); + Assert.All(data.Members, row => Assert.Equal(SiloStatus.Active, row.Item1.Status)); + Assert.All(activeAddresses, address => Assert.NotNull(data.TryGet(address))); + Assert.All(deadAddresses, address => Assert.Null(data.TryGet(address))); } // Utility methods