From 1eb0bfd69b7fd184c92b47e0542ef75d23ee8af2 Mon Sep 17 00:00:00 2001 From: Adolfo Marinucci Date: Wed, 5 Aug 2026 16:46:35 +0200 Subject: [PATCH] SqlServerCT: surface stale sync anchors and make change retention configurable Clients whose anchor fell out of the change tracking retention window stopped uploading updates without reporting an error. Update/Delete statements pass the target anchor into CHANGETABLE(CHANGES , @last_sync_version); nothing guarded that on the upload path and the resulting error was swallowed by T-SQL, so the batch reported zero affected rows and the item was misread as a write conflict and dropped. - Add SyncAnchorTooOldException (TableName, RequestedVersion, MinValidVersion) so callers can tell "reinitialize this client" from "transient, retry". It derives from InvalidOperationException, which keeps GetChangesAsync's existing contract while giving that path the same typed signal. It propagates unwrapped through SyncErrorException and SyncAgent. - Guard ApplyChangesAsync against an anchor below CHANGE_TRACKING_MIN_VALID_VERSION, only for tables carrying Update/Delete items so first-time inserts still work, and also at the point an insert is converted into an update. - Return SyncAnchor.Null rather than version 0 for a missing remote anchor row. Zero is not a sentinel: it is a real-looking version permanently below the retention floor. Applied to SqlServerCT and, for consistency with the other providers, to SqlServer (where 0 and -1 are equivalent since __CORE_SYNC_CT.ID is IDENTITY(1,1), so this is a naming fix rather than a behaviour change). - Drop the BEGIN TRY/CATCH PRINT ERROR_MESSAGE() wrappers so a zero rowcount means only a genuine conflict. Those swallowed errors were load-bearing, so the two cases that relied on them are now handled explicitly: a duplicate key routes into the existing exists-check and retries as an update, and a constraint violation (547) defers the item to the next pass, preserving the out-of-order insert retry. When the retry loop stops making progress with a real error in hand it now throws instead of dropping the items with a warning. - Log the exception and the accumulated messages on failure, and clear the message log per item so a failure no longer replays every message since the call began. - Reconcile CHANGE_RETENTION on provisioning instead of only setting it when change tracking is first enabled, which is why .ChangeRetention(n) was silently ignored on every database that already had it on. Adds a ChangeRetentionUnit overload, validates the value, reports ALTER DATABASE permission failures clearly, and warns when the window is shorter than seven days. A database that already has change tracking on is only touched when the retention was configured explicitly. - Switch the per-item ExecuteNonQuery to ExecuteNonQueryAsync, replace the UPDATE-then-INSERT anchor writes with MERGE to close the concurrent-sync race, and pass the anchor ID as SqlDbType.UniqueIdentifier throughout. - Cover all of the above in SqlServerCTResilienceTests, and correct the README's change tracking example, which referenced builder methods that do not exist. --- README.md | 31 +- src/CoreSync.SqlServer/SqlSyncProvider.cs | 5 +- .../ChangeRetentionUnit.cs | 28 ++ .../ChangeTrackingDatabaseOptions.cs | 23 + .../SqlConnectionExtensions.cs | 58 ++- .../SqlServerCTProvider.cs | 266 +++++++++-- .../SqlServerCTSyncConfiguration.cs | 56 ++- .../SqlServerCTSyncConfigurationBuilder.cs | 43 +- .../SqlServerCTSyncTable.cs | 33 +- .../SqlServerCTResilienceTests.cs | 424 ++++++++++++++++++ src/CoreSync/SyncAgent.cs | 7 + src/CoreSync/SyncAnchorTooOldException.cs | 85 ++++ 12 files changed, 994 insertions(+), 65 deletions(-) create mode 100644 src/CoreSync.SqlServerCT/ChangeRetentionUnit.cs create mode 100644 src/CoreSync.SqlServerCT/ChangeTrackingDatabaseOptions.cs create mode 100644 src/CoreSync.Tests/SqlServerCTResilienceTests.cs create mode 100644 src/CoreSync/SyncAnchorTooOldException.cs diff --git a/README.md b/README.md index 718a286..235c12e 100644 --- a/README.md +++ b/README.md @@ -194,8 +194,7 @@ If you prefer SQL Server's built-in Change Tracking over custom triggers: ```csharp var config = new SqlServerCTSyncConfigurationBuilder(connectionString) - .ChangeRetentionDays(7) // how long change history is kept - .AutoCleanup(true) + .ChangeRetention(days: 7, autoCleanup: true) // how long change history is kept .Table("Users") .Table("Posts") .Build(); @@ -204,6 +203,34 @@ var provider = new SqlServerCTProvider(config); await provider.ApplyProvisionAsync(); ``` +The retention can also be expressed in hours or minutes: + +```csharp +.ChangeRetention(90, ChangeRetentionUnit.Minutes) +``` + +`ApplyProvisionAsync` reconciles the setting: when the retention is configured explicitly it is applied +even to a database that already has change tracking enabled. Leave `ChangeRetention` unset to keep +whatever the database is already configured with. + +Retention matters. A client that stays offline longer than the window can no longer resume an +incremental sync: both `GetChangesAsync` and `ApplyChangesAsync` then fail with +`SyncAnchorTooOldException`, which is permanent for that client. Catch it to trigger a +reinitialization from a fresh snapshot rather than retrying: + +```csharp +try +{ + await agent.SynchronizeAsync(); +} +catch (SyncAnchorTooOldException ex) +{ + // ex.TableName / ex.RequestedVersion / ex.MinValidVersion + // Retrying can never succeed - drop the local copy and take a fresh initial snapshot. + await ReinitializeClientAsync(); +} +``` + ### Managing Change Tracking ```csharp diff --git a/src/CoreSync.SqlServer/SqlSyncProvider.cs b/src/CoreSync.SqlServer/SqlSyncProvider.cs index 4ba75fd..5f24f56 100644 --- a/src/CoreSync.SqlServer/SqlSyncProvider.cs +++ b/src/CoreSync.SqlServer/SqlSyncProvider.cs @@ -343,8 +343,11 @@ private async Task GetLastRemoteAnchorForStoreAsync(Guid otherStoreI var version = await cmd.ExecuteScalarAsync(cancellationToken); + // Report a missing anchor row as the null sentinel, matching + // GetLastLocalAnchorForStoreAsync and the other providers. Version 0 is not a sentinel: it + // is a real-looking version that callers cannot tell apart from "synchronized up to 0". if (version == null || version == DBNull.Value) - return new SyncAnchor(otherStoreId, 0); + return SyncAnchor.Null; return new SyncAnchor(otherStoreId, (long)version); } diff --git a/src/CoreSync.SqlServerCT/ChangeRetentionUnit.cs b/src/CoreSync.SqlServerCT/ChangeRetentionUnit.cs new file mode 100644 index 0000000..7d5d6e9 --- /dev/null +++ b/src/CoreSync.SqlServerCT/ChangeRetentionUnit.cs @@ -0,0 +1,28 @@ +namespace CoreSync.SqlServerCT +{ + /// + /// The time unit a SQL Server Change Tracking retention period is expressed in. + /// + /// + /// The numeric values match the retention_period_units column of + /// sys.change_tracking_databases, so a value read from that catalog view can be cast + /// directly to this enum. + /// + public enum ChangeRetentionUnit + { + /// + /// The retention period is expressed in minutes (CHANGE_RETENTION = n MINUTES). + /// + Minutes = 1, + + /// + /// The retention period is expressed in hours (CHANGE_RETENTION = n HOURS). + /// + Hours = 2, + + /// + /// The retention period is expressed in days (CHANGE_RETENTION = n DAYS). + /// + Days = 3 + } +} diff --git a/src/CoreSync.SqlServerCT/ChangeTrackingDatabaseOptions.cs b/src/CoreSync.SqlServerCT/ChangeTrackingDatabaseOptions.cs new file mode 100644 index 0000000..d2b464e --- /dev/null +++ b/src/CoreSync.SqlServerCT/ChangeTrackingDatabaseOptions.cs @@ -0,0 +1,23 @@ +namespace CoreSync.SqlServerCT +{ + /// + /// The database-level change tracking settings as reported by sys.change_tracking_databases. + /// + internal sealed class ChangeTrackingDatabaseOptions + { + public ChangeTrackingDatabaseOptions(int retentionPeriod, ChangeRetentionUnit retentionPeriodUnit, bool autoCleanup) + { + RetentionPeriod = retentionPeriod; + RetentionPeriodUnit = retentionPeriodUnit; + AutoCleanup = autoCleanup; + } + + public int RetentionPeriod { get; } + + public ChangeRetentionUnit RetentionPeriodUnit { get; } + + public bool AutoCleanup { get; } + + public override string ToString() => $"CHANGE_RETENTION = {RetentionPeriod} {RetentionPeriodUnit}, AUTO_CLEANUP = {(AutoCleanup ? "ON" : "OFF")}"; + } +} diff --git a/src/CoreSync.SqlServerCT/SqlConnectionExtensions.cs b/src/CoreSync.SqlServerCT/SqlConnectionExtensions.cs index 5e89c28..9f5a7ca 100644 --- a/src/CoreSync.SqlServerCT/SqlConnectionExtensions.cs +++ b/src/CoreSync.SqlServerCT/SqlConnectionExtensions.cs @@ -16,13 +16,67 @@ public static async Task GetIsChangeTrackingEnabledAsync(this SqlConnectio return ((int)await cmd.ExecuteScalarAsync(cancellationToken)) == 1; } - public static async Task EnableChangeTrackingAsync(this SqlConnection connection, int days = 7, bool autoCleanup = true, CancellationToken cancellationToken = default) + /// + /// Reads the change tracking retention settings currently in effect for the connected database, + /// or null when change tracking is not enabled on it. + /// + public static async Task GetChangeTrackingOptionsAsync(this SqlConnection connection, CancellationToken cancellationToken) { - var cmdText = $@"ALTER DATABASE CURRENT SET CHANGE_TRACKING = ON (CHANGE_RETENTION = {days} DAYS, AUTO_CLEANUP = {(autoCleanup ? "ON" : "OFF")})"; + var cmdText = @"SELECT retention_period, retention_period_units, is_auto_cleanup_on +FROM sys.change_tracking_databases +WHERE database_id = DB_ID()"; + using var cmd = new SqlCommand(cmdText, connection); + using var reader = await cmd.ExecuteReaderAsync(cancellationToken); + + if (!await reader.ReadAsync(cancellationToken)) + return null; + + // sys.change_tracking_databases reports retention_period_units as tinyint and + // is_auto_cleanup_on as tinyint too, so read them through Convert rather than the typed + // getters. + return new ChangeTrackingDatabaseOptions( + Convert.ToInt32(reader.GetValue(0)), + (ChangeRetentionUnit)Convert.ToInt32(reader.GetValue(1)), + Convert.ToBoolean(reader.GetValue(2))); + } + + /// + /// Turns change tracking on for the connected database. Only valid when change tracking is off: + /// the = ON (...) form is rejected once it is already enabled - use + /// to change the settings of a database that + /// already has it on. + /// + public static async Task EnableChangeTrackingAsync(this SqlConnection connection, int retentionPeriod = 7, ChangeRetentionUnit retentionPeriodUnit = ChangeRetentionUnit.Days, bool autoCleanup = true, CancellationToken cancellationToken = default) + { + var cmdText = $@"ALTER DATABASE CURRENT SET CHANGE_TRACKING = ON ({FormatChangeTrackingOptions(retentionPeriod, retentionPeriodUnit, autoCleanup)})"; + using var cmd = new SqlCommand(cmdText, connection); + await cmd.ExecuteNonQueryAsync(cancellationToken); + } + + /// + /// Changes the retention settings of a database that already has change tracking enabled. + /// + public static async Task AlterChangeTrackingRetentionAsync(this SqlConnection connection, int retentionPeriod, ChangeRetentionUnit retentionPeriodUnit, bool autoCleanup, CancellationToken cancellationToken) + { + var cmdText = $@"ALTER DATABASE CURRENT SET CHANGE_TRACKING ({FormatChangeTrackingOptions(retentionPeriod, retentionPeriodUnit, autoCleanup)})"; using var cmd = new SqlCommand(cmdText, connection); await cmd.ExecuteNonQueryAsync(cancellationToken); } + private static string FormatChangeTrackingOptions(int retentionPeriod, ChangeRetentionUnit retentionPeriodUnit, bool autoCleanup) + => $"CHANGE_RETENTION = {retentionPeriod} {ToSqlKeyword(retentionPeriodUnit)}, AUTO_CLEANUP = {(autoCleanup ? "ON" : "OFF")}"; + + private static string ToSqlKeyword(ChangeRetentionUnit unit) + { + switch (unit) + { + case ChangeRetentionUnit.Minutes: return "MINUTES"; + case ChangeRetentionUnit.Hours: return "HOURS"; + case ChangeRetentionUnit.Days: return "DAYS"; + default: throw new ArgumentOutOfRangeException(nameof(unit), unit, "Unsupported change retention unit"); + } + } + public static async Task DisableChangeTrackingAsync(this SqlConnection connection, CancellationToken cancellationToken) { var cmdText = @"ALTER DATABASE CURRENT SET CHANGE_TRACKING = OFF"; diff --git a/src/CoreSync.SqlServerCT/SqlServerCTProvider.cs b/src/CoreSync.SqlServerCT/SqlServerCTProvider.cs index fa41de7..feffa28 100644 --- a/src/CoreSync.SqlServerCT/SqlServerCTProvider.cs +++ b/src/CoreSync.SqlServerCT/SqlServerCTProvider.cs @@ -63,6 +63,25 @@ public async Task ApplyChangesAsync([NotNull] SyncChangeSet changeSe cmd.CommandText = "SELECT CHANGE_TRACKING_CURRENT_VERSION()"; var version = await cmd.ExecuteLongScalarAsync(cancellationToken); + // Update and Delete resolve conflicts through CHANGETABLE(CHANGES
, + // @last_sync_version), which SQL Server rejects outright once the requested version + // falls out of the table's retention window. Validate the anchor up front so a stale + // client gets a typed "reinitialize me" signal instead of a masked failure. + // Inserts never touch CHANGETABLE, so a first-time sync - which legitimately carries + // a null target anchor and nothing but inserts - must not be validated here. + var minValidVersions = new Dictionary(StringComparer.OrdinalIgnoreCase); + foreach (var item in changeSet.Items) + { + if (item.ChangeType != ChangeType.Update && item.ChangeType != ChangeType.Delete) + continue; + + var tableToValidate = (SqlServerCTSyncTable?)Configuration.Tables.FirstOrDefault(_ => _.Name == item.TableName); + if (tableToValidate == null) + continue; + + await EnsureAnchorNotTooOldAsync(cmd, tableToValidate, changeSet.TargetAnchor, minValidVersions, cancellationToken); + } + var remainingItems = changeSet.Items.ToList(); int pass = 0; while (remainingItems.Count > 0) @@ -70,6 +89,7 @@ public async Task ApplyChangesAsync([NotNull] SyncChangeSet changeSe pass++; var failedItems = new List(); int appliedInPass = 0; + SqlException? lastFailure = null; foreach (var item in remainingItems) { @@ -83,10 +103,15 @@ public async Task ApplyChangesAsync([NotNull] SyncChangeSet changeSe bool syncForceWrite = false; var itemChangeType = item.ChangeType; bool itemHandled = false; + SqlException? itemFailure = null; retryWrite: cmd.Parameters.Clear(); + // messageLog accumulates for the lifetime of the connection; reset it so a + // failure reports only the messages this statement produced. + messageLog.Clear(); + table.SetupCommand(cmd, itemChangeType, item.Values); cmd.Parameters.Add(new SqlParameter("@last_sync_version", changeSet.TargetAnchor.Version)); @@ -100,16 +125,38 @@ public async Task ApplyChangesAsync([NotNull] SyncChangeSet changeSe try { - affectedRows = cmd.ExecuteNonQuery(); + affectedRows = await cmd.ExecuteNonQueryAsync(cancellationToken); if (affectedRows > 0) { _logger?.Trace($"[{_storeId}] Successfully applied {item}"); } } + catch (SqlException ex) when (itemChangeType == ChangeType.Insert && IsDuplicateKeyViolation(ex)) + { + // The row is already there. Report zero rows affected and let the handling + // below confirm it exists and retry the item as an update - the behaviour + // the swallowed T-SQL error used to produce, now driven by the real error. + _logger?.Trace($"[{_storeId}] {item} already exists on table {table}: {ex.Message}"); + itemFailure = ex; + affectedRows = 0; + } + catch (SqlException ex) when (ex.Number == SqlConstraintViolationErrorNumber) + { + // A referenced row may simply not have been applied yet (the change set + // is not dependency ordered). Defer the item and let the retry loop have + // another go once the rest of the pass has landed. A constraint violation + // is statement-scoped, so the transaction stays usable. + // If the loop stops making progress the error is rethrown below rather + // than dropped. + _logger?.Trace($"[{_storeId}] Deferring {itemChangeType} of {item} on table {table} (pass {pass}): {ex.Message}"); + lastFailure = ex; + failedItems.Add(item); + continue; + } catch (Exception ex) { - _logger?.Error($"Unable to {itemChangeType} item {item} to store for table {table}.{Environment.NewLine}Generated SQL:{Environment.NewLine}{cmd.CommandText}"); + _logger?.Error(DescribeItemFailure(itemChangeType, item, table, cmd.CommandText, messageLog, ex)); throw new SynchronizationException($"Unable to {itemChangeType} item {item} to store for table {table}", ex); } @@ -127,6 +174,11 @@ public async Task ApplyChangesAsync([NotNull] SyncChangeSet changeSe if (1 == (int)await cmd.ExecuteScalarAsync(cancellationToken) && !syncForceWrite) { + // The row is already there, so this insert becomes an update - + // which does query CHANGETABLE. Validate the anchor before the + // statement runs, since the up-front pass skipped this table. + await EnsureAnchorNotTooOldAsync(cmd, table, changeSet.TargetAnchor, minValidVersions, cancellationToken); + itemChangeType = ChangeType.Update; goto retryWrite; } @@ -179,6 +231,11 @@ public async Task ApplyChangesAsync([NotNull] SyncChangeSet changeSe if (!itemHandled) { failedItems.Add(item); + + if (itemFailure != null) + { + lastFailure = itemFailure; + } } } @@ -186,7 +243,18 @@ public async Task ApplyChangesAsync([NotNull] SyncChangeSet changeSe { if (failedItems.Count > 0) { - _logger?.Warning($"[{_storeId}] {failedItems.Count} item(s) could not be applied after {pass} pass(es) (possible unresolvable foreign key constraint)"); + if (lastFailure != null) + { + // The retry loop has stopped making progress and the remaining items + // still fail against the database. Surface it: dropping them here + // would be silent data loss. + _logger?.Error($"[{_storeId}] {failedItems.Count} item(s) could not be applied after {pass} pass(es): {lastFailure.Message}"); + throw new SynchronizationException( + $"Unable to apply {failedItems.Count} item(s) after {pass} pass(es); the change set does not resolve the last failure", + lastFailure); + } + + _logger?.Warning($"[{_storeId}] {failedItems.Count} item(s) could not be applied after {pass} pass(es)"); } break; } @@ -195,16 +263,11 @@ public async Task ApplyChangesAsync([NotNull] SyncChangeSet changeSe remainingItems = failedItems; } - cmd.CommandText = $"UPDATE [__CORE_SYNC_CT_REMOTE_ANCHOR] SET [REMOTE_VERSION] = @version WHERE [ID] = @id"; + cmd.CommandText = UpsertRemoteVersionCommandText; cmd.Parameters.Clear(); - cmd.Parameters.AddWithValue("@id", changeSet.SourceAnchor.StoreId.ToString()); - cmd.Parameters.AddWithValue("@version", changeSet.SourceAnchor.Version); - - if (0 == await cmd.ExecuteNonQueryAsync(cancellationToken)) - { - cmd.CommandText = "INSERT INTO [__CORE_SYNC_CT_REMOTE_ANCHOR] ([ID], [REMOTE_VERSION]) VALUES (@id, @version)"; - await cmd.ExecuteNonQueryAsync(cancellationToken); - } + cmd.Parameters.Add(new SqlParameter("@id", SqlDbType.UniqueIdentifier) { Value = changeSet.SourceAnchor.StoreId }); + cmd.Parameters.Add(new SqlParameter("@version", SqlDbType.BigInt) { Value = changeSet.SourceAnchor.Version }); + await cmd.ExecuteNonQueryAsync(cancellationToken); tr.Commit(); @@ -220,13 +283,78 @@ public async Task ApplyChangesAsync([NotNull] SyncChangeSet changeSe throw; } } + catch (SyncAnchorTooOldException) + { + // Deliberately not wrapped: callers have to be able to tell "reinitialize this client" + // apart from a transient error worth retrying. + throw; + } catch (Exception ex) { - var exceptionMessage = $"An exception occurred during synchronization:{Environment.NewLine}Errors:{Environment.NewLine}{string.Join(Environment.NewLine, messageLog.SelectMany(_ => _.Errors.Cast()))}{Environment.NewLine}Messages:{Environment.NewLine}{string.Join(Environment.NewLine, messageLog.Select(_ => _.Message))}"; + var exceptionMessage = $"An exception occurred during synchronization: {ex.Message}{Environment.NewLine}Errors:{Environment.NewLine}{string.Join(Environment.NewLine, messageLog.SelectMany(_ => _.Errors.Cast()))}{Environment.NewLine}Messages:{Environment.NewLine}{string.Join(Environment.NewLine, messageLog.Select(_ => _.Message))}"; throw new SyncErrorException(exceptionMessage, ex); } } + /// + /// SQL Server error raised when a statement conflicts with a FOREIGN KEY or CHECK constraint. + /// + private const int SqlConstraintViolationErrorNumber = 547; + + /// + /// Returns whether reports an attempt to insert a row that is already + /// there - either a PRIMARY KEY / UNIQUE constraint or a unique index violation. + /// + private static bool IsDuplicateKeyViolation(SqlException ex) + => ex.Number == 2627 || ex.Number == 2601; + + /// + /// Retention windows shorter than this are legal but risky: a client offline for longer can no + /// longer resume an incremental sync. + /// + private const int ShortChangeRetentionWarningThresholdInMinutes = 7 * 1440; + + private const string UpsertRemoteVersionCommandText = @"MERGE [dbo].[__CORE_SYNC_CT_REMOTE_ANCHOR] WITH (HOLDLOCK) AS T +USING (SELECT @id AS [ID]) AS S ON T.[ID] = S.[ID] +WHEN MATCHED THEN UPDATE SET [REMOTE_VERSION] = @version +WHEN NOT MATCHED THEN INSERT ([ID], [REMOTE_VERSION]) VALUES (@id, @version);"; + + private const string UpsertLocalVersionCommandText = @"MERGE [dbo].[__CORE_SYNC_CT_REMOTE_ANCHOR] WITH (HOLDLOCK) AS T +USING (SELECT @id AS [ID]) AS S ON T.[ID] = S.[ID] +WHEN MATCHED THEN UPDATE SET [LOCAL_VERSION] = @version +WHEN NOT MATCHED THEN INSERT ([ID], [LOCAL_VERSION]) VALUES (@id, @version);"; + + /// + /// Reads (and memoizes) the minimum version can still resolve changes + /// from, and throws when the anchor predates it. + /// + private static async Task EnsureAnchorNotTooOldAsync(SqlCommand cmd, SqlServerCTSyncTable table, SyncAnchor anchor, IDictionary minValidVersions, CancellationToken cancellationToken) + { + if (!minValidVersions.TryGetValue(table.NameWithSchema, out var minValidVersion)) + { + cmd.CommandText = $"SELECT CHANGE_TRACKING_MIN_VALID_VERSION(OBJECT_ID('{table.NameWithSchema}'))"; + cmd.Parameters.Clear(); + minValidVersion = await cmd.ExecuteLongScalarAsync(cancellationToken); + minValidVersions[table.NameWithSchema] = minValidVersion; + } + + // A null anchor (version -1) means no anchor was ever recorded for the peer store, which is + // just as unrecoverable as one that fell out of the retention window. + if (anchor.Version < minValidVersion) + { + throw new SyncAnchorTooOldException(table.NameWithSchema, anchor.Version, minValidVersion); + } + } + + private static string DescribeItemFailure(ChangeType changeType, SyncItem item, SqlServerCTSyncTable table, string commandText, List messageLog, Exception ex) + { + return $"Unable to {changeType} item {item} to store for table {table}: {ex.Message}" + + $"{Environment.NewLine}Errors:{Environment.NewLine}{string.Join(Environment.NewLine, messageLog.SelectMany(_ => _.Errors.Cast()))}" + + $"{Environment.NewLine}Messages:{Environment.NewLine}{string.Join(Environment.NewLine, messageLog.Select(_ => _.Message))}" + + $"{Environment.NewLine}Generated SQL:{Environment.NewLine}{commandText}" + + $"{Environment.NewLine}Exception:{Environment.NewLine}{ex}"; + } + public async Task SaveVersionForStoreAsync(Guid otherStoreId, long version, CancellationToken cancellationToken = default) { using var c = new SqlConnection(Configuration.ConnectionString); @@ -238,15 +366,10 @@ public async Task SaveVersionForStoreAsync(Guid otherStoreId, long version, Canc cmd.Transaction = tr; try { - cmd.CommandText = $"UPDATE [__CORE_SYNC_CT_REMOTE_ANCHOR] SET [LOCAL_VERSION] = @version WHERE [ID] = @id"; - cmd.Parameters.AddWithValue("@id", otherStoreId.ToString()); - cmd.Parameters.AddWithValue("@version", version); - - if (0 == await cmd.ExecuteNonQueryAsync(cancellationToken)) - { - cmd.CommandText = "INSERT INTO [__CORE_SYNC_CT_REMOTE_ANCHOR] ([ID], [LOCAL_VERSION]) VALUES (@id, @version)"; - await cmd.ExecuteNonQueryAsync(cancellationToken); - } + cmd.CommandText = UpsertLocalVersionCommandText; + cmd.Parameters.Add(new SqlParameter("@id", SqlDbType.UniqueIdentifier) { Value = otherStoreId }); + cmd.Parameters.Add(new SqlParameter("@version", SqlDbType.BigInt) { Value = version }); + await cmd.ExecuteNonQueryAsync(cancellationToken); tr.Commit(); @@ -268,7 +391,7 @@ private async Task GetLastLocalAnchorForStoreAsync(Guid otherStoreId using var cmd = c.CreateCommand(); cmd.CommandText = "SELECT [LOCAL_VERSION] FROM [dbo].[__CORE_SYNC_CT_REMOTE_ANCHOR] WHERE [ID] = @storeId"; - cmd.Parameters.AddWithValue("@storeId", otherStoreId); + cmd.Parameters.Add(new SqlParameter("@storeId", SqlDbType.UniqueIdentifier) { Value = otherStoreId }); var version = await cmd.ExecuteScalarAsync(cancellationToken); @@ -284,13 +407,17 @@ private async Task GetLastRemoteAnchorForStoreAsync(Guid otherStoreI await c.OpenAsync(cancellationToken); using var cmd = c.CreateCommand(); - cmd.CommandText = "SELECT [REMOTE_VERSION] FROM [__CORE_SYNC_CT_REMOTE_ANCHOR] WHERE [ID] = @storeId"; - cmd.Parameters.AddWithValue("@storeId", otherStoreId.ToString()); + cmd.CommandText = "SELECT [REMOTE_VERSION] FROM [dbo].[__CORE_SYNC_CT_REMOTE_ANCHOR] WHERE [ID] = @storeId"; + cmd.Parameters.Add(new SqlParameter("@storeId", SqlDbType.UniqueIdentifier) { Value = otherStoreId }); var version = await cmd.ExecuteScalarAsync(cancellationToken); + // Version 0 would not be a sentinel: it is a real-looking version permanently below the + // change tracking retention floor, so every Update/Delete sent against it is doomed to fail. + // Report the missing anchor as null and let ApplyChangesAsync turn it into a typed + // "reinitialize this client" signal. if (version == null || version == DBNull.Value) - return new SyncAnchor(otherStoreId, 0); + return SyncAnchor.Null; return new SyncAnchor(otherStoreId, (long)version); } @@ -439,11 +566,7 @@ public async Task ApplyProvisionAsync(CancellationToken cancellationToken = defa using var connection = new SqlConnection(Configuration.ConnectionString); await connection.OpenAsync(cancellationToken); - // Enable change tracking at database level if not already enabled - if (!await connection.GetIsChangeTrackingEnabledAsync(cancellationToken)) - { - await connection.EnableChangeTrackingAsync(Configuration.ChangeRetentionDays, Configuration.AutoCleanup, cancellationToken); - } + await ReconcileChangeTrackingRetentionAsync(connection, cancellationToken); // Enable change tracking per table foreach (SqlServerCTSyncTable table in Configuration.Tables.Cast()) @@ -468,6 +591,85 @@ public async Task ApplyProvisionAsync(CancellationToken cancellationToken = defa } } + /// + /// Brings the database-level change tracking retention in line with the configuration. + /// + /// + /// Enabling change tracking is only half of it: on a database that already has it on, the + /// = ON (...) form is rejected, so the configured retention used to be silently ignored + /// forever. Here the current settings are read back from sys.change_tracking_databases + /// and, when they differ, applied through the modify form of the statement. + /// + /// A database that already has change tracking enabled is only touched when the retention was + /// set explicitly on the builder, so provisioning with a low privilege account against a + /// pre-configured database keeps working. + /// + /// + private async Task ReconcileChangeTrackingRetentionAsync(SqlConnection connection, CancellationToken cancellationToken) + { + if (Configuration.ChangeRetentionInMinutes < ShortChangeRetentionWarningThresholdInMinutes) + { + _logger?.Warning($"[{_storeId}] Change tracking retention is configured to {Configuration.ChangeRetention} {Configuration.ChangeRetentionUnit}, " + + $"which is shorter than {ShortChangeRetentionWarningThresholdInMinutes / 1440} days. Any client that stays offline longer than the retention window " + + "cannot resume an incremental sync and has to be reinitialized from a fresh snapshot."); + } + + var isEnabled = await connection.GetIsChangeTrackingEnabledAsync(cancellationToken); + + try + { + if (!isEnabled) + { + await connection.EnableChangeTrackingAsync(Configuration.ChangeRetention, Configuration.ChangeRetentionUnit, Configuration.AutoCleanup, cancellationToken); + return; + } + + if (!Configuration.IsChangeRetentionConfigured) + { + return; + } + + var current = await connection.GetChangeTrackingOptionsAsync(cancellationToken); + + if (current != null && + current.RetentionPeriod == Configuration.ChangeRetention && + current.RetentionPeriodUnit == Configuration.ChangeRetentionUnit && + current.AutoCleanup == Configuration.AutoCleanup) + { + return; + } + + _logger?.Info($"[{_storeId}] Change tracking retention is '{current?.ToString() ?? "unknown"}', applying configured " + + $"'CHANGE_RETENTION = {Configuration.ChangeRetention} {Configuration.ChangeRetentionUnit}, AUTO_CLEANUP = {(Configuration.AutoCleanup ? "ON" : "OFF")}'"); + + await connection.AlterChangeTrackingRetentionAsync(Configuration.ChangeRetention, Configuration.ChangeRetentionUnit, Configuration.AutoCleanup, cancellationToken); + } + catch (SqlException ex) + { + var action = isEnabled ? "update the change tracking retention policy of" : "enable change tracking on"; + var hint = IsPermissionError(ex) + ? " The account used by the connection string needs the ALTER DATABASE permission (or membership of db_owner)." + : string.Empty; + + throw new InvalidOperationException( + $"Unable to {action} database '{connection.Database}': {ex.Message}.{hint}", ex); + } + } + + private static bool IsPermissionError(SqlException ex) + { + switch (ex.Number) + { + case 262: // ALTER DATABASE permission denied in database '%.*ls' + case 300: // VIEW SERVER STATE / generic permission denied on object + case 5011: // User does not have permission to alter database '%.*ls' + case 15151: // Cannot alter the database '%.*ls', because it does not exist or you do not have permission + return true; + default: + return false; + } + } + public async Task RemoveProvisionAsync(CancellationToken cancellationToken = default) { var connStringBuilder = new SqlConnectionStringBuilder(Configuration.ConnectionString); @@ -556,7 +758,7 @@ public async Task GetChangesAsync(Guid otherStoreId, SyncFilterPa var minVersion = await cmd.ExecuteLongScalarAsync(cancellationToken); if (fromAnchor.Version < minVersion) - throw new InvalidOperationException($"Unable to get changes for table '{table.NameWithSchema}', version of data requested ({fromAnchor.Version}) is too old (min valid version {minVersion})"); + throw new SyncAnchorTooOldException(table.NameWithSchema, fromAnchor.Version, minVersion); } } diff --git a/src/CoreSync.SqlServerCT/SqlServerCTSyncConfiguration.cs b/src/CoreSync.SqlServerCT/SqlServerCTSyncConfiguration.cs index 1a84dd2..3e93210 100644 --- a/src/CoreSync.SqlServerCT/SqlServerCTSyncConfiguration.cs +++ b/src/CoreSync.SqlServerCT/SqlServerCTSyncConfiguration.cs @@ -1,4 +1,5 @@ using JetBrains.Annotations; +using System; namespace CoreSync.SqlServerCT { @@ -6,23 +7,70 @@ public class SqlServerCTSyncConfiguration : SyncConfiguration { public string ConnectionString { get; } - public int ChangeRetentionDays { get; } + /// + /// Gets the configured change tracking retention period, expressed in . + /// + public int ChangeRetention { get; } + + /// + /// Gets the unit is expressed in. + /// + public ChangeRetentionUnit ChangeRetentionUnit { get; } + + /// + /// Gets the configured retention period expressed in minutes. + /// + public long ChangeRetentionInMinutes + { + get + { + switch (ChangeRetentionUnit) + { + case ChangeRetentionUnit.Minutes: return ChangeRetention; + case ChangeRetentionUnit.Hours: return ChangeRetention * 60L; + default: return ChangeRetention * 1440L; + } + } + } + + /// + /// Gets the configured retention period expressed in whole days, rounded up. + /// + /// + /// Retained for backward compatibility with configurations built through the + /// ChangeRetention(int days, bool autoCleanup) overload. When the retention is configured + /// in hours or minutes this reports the equivalent rounded up to the next whole day, so prefer + /// together with for an exact value. + /// + public int ChangeRetentionDays => ChangeRetentionUnit == ChangeRetentionUnit.Days + ? ChangeRetention + : (int)Math.Ceiling(ChangeRetentionInMinutes / 1440.0); public bool AutoCleanup { get; } + /// + /// Gets a value indicating whether the retention policy was set explicitly on the builder. + /// Provisioning only reconciles the retention of an already change-tracked database when it was. + /// + internal bool IsChangeRetentionConfigured { get; } + internal SqlServerCTSyncConfiguration( [NotNull] string connectionString, [NotNull] SqlServerCTSyncTable[] tables, - int changeRetentionDays = 7, - bool autoCleanup = true) + int changeRetention = 7, + ChangeRetentionUnit changeRetentionUnit = ChangeRetentionUnit.Days, + bool autoCleanup = true, + bool isChangeRetentionConfigured = false) : base(tables) { Validate.NotNullOrEmptyOrWhiteSpace(connectionString, nameof(connectionString)); Validate.NotNullOrEmptyArray(tables, nameof(tables)); ConnectionString = connectionString; - ChangeRetentionDays = changeRetentionDays; + ChangeRetention = changeRetention; + ChangeRetentionUnit = changeRetentionUnit; AutoCleanup = autoCleanup; + IsChangeRetentionConfigured = isChangeRetentionConfigured; } } } diff --git a/src/CoreSync.SqlServerCT/SqlServerCTSyncConfigurationBuilder.cs b/src/CoreSync.SqlServerCT/SqlServerCTSyncConfigurationBuilder.cs index 397acfd..cabc835 100644 --- a/src/CoreSync.SqlServerCT/SqlServerCTSyncConfigurationBuilder.cs +++ b/src/CoreSync.SqlServerCT/SqlServerCTSyncConfigurationBuilder.cs @@ -32,8 +32,10 @@ public class SqlServerCTSyncConfigurationBuilder private readonly string _connectionString; private readonly List _tables = new List(); private string _schema = "dbo"; - private int _changeRetentionDays = 7; + private int _changeRetention = 7; + private ChangeRetentionUnit _changeRetentionUnit = ChangeRetentionUnit.Days; private bool _autoCleanup = true; + private bool _changeRetentionConfigured; /// /// Initializes a new instance of the class. @@ -67,10 +69,45 @@ public SqlServerCTSyncConfigurationBuilder Schema(string schema) /// Defaults to true. /// /// This builder instance for method chaining. + /// is not greater than zero. public SqlServerCTSyncConfigurationBuilder ChangeRetention(int days, bool autoCleanup = true) + => ChangeRetention(days, ChangeRetentionUnit.Days, autoCleanup); + + /// + /// Configures the SQL Server Change Tracking retention policy using an explicit time unit. + /// This controls how long change history is retained before being automatically cleaned up. + /// + /// The retention period, expressed in . + /// The unit is expressed in. + /// + /// When true, SQL Server automatically removes expired change tracking data. + /// Defaults to true. + /// + /// This builder instance for method chaining. + /// + /// Provisioning reconciles the setting: calling this makes + /// apply the retention even to a database + /// that already has change tracking enabled. A retention shorter than seven days is applied but + /// logged as a warning, because clients that stay offline longer than the window have to be + /// reinitialized from a fresh snapshot. + /// + /// is not greater than zero. + public SqlServerCTSyncConfigurationBuilder ChangeRetention(int value, ChangeRetentionUnit unit, bool autoCleanup = true) { - _changeRetentionDays = days; + if (value <= 0) + { + throw new ArgumentOutOfRangeException(nameof(value), value, "Change retention must be greater than zero"); + } + + if (!Enum.IsDefined(typeof(ChangeRetentionUnit), unit)) + { + throw new ArgumentOutOfRangeException(nameof(unit), unit, "Unsupported change retention unit"); + } + + _changeRetention = value; + _changeRetentionUnit = unit; _autoCleanup = autoCleanup; + _changeRetentionConfigured = true; return this; } @@ -257,6 +294,6 @@ public SqlServerCTSyncConfigurationBuilder IdentityInsert(IdentityInsertMode mod /// Builds the from the registered tables and change tracking settings. /// /// A new instance. - public SqlServerCTSyncConfiguration Build() => new SqlServerCTSyncConfiguration(_connectionString, _tables.ToArray(), _changeRetentionDays, _autoCleanup); + public SqlServerCTSyncConfiguration Build() => new SqlServerCTSyncConfiguration(_connectionString, _tables.ToArray(), _changeRetention, _changeRetentionUnit, _autoCleanup, _changeRetentionConfigured); } } diff --git a/src/CoreSync.SqlServerCT/SqlServerCTSyncTable.cs b/src/CoreSync.SqlServerCT/SqlServerCTSyncTable.cs index 37f8b94..2b1cf31 100644 --- a/src/CoreSync.SqlServerCT/SqlServerCTSyncTable.cs +++ b/src/CoreSync.SqlServerCT/SqlServerCTSyncTable.cs @@ -85,7 +85,13 @@ internal void SetupCommand(SqlCommand cmd, ChangeType itemChangeType, Dictionary var allSyncItemsExceptPrimaryKey = allSyncItems.Where(_ => !PrimaryKeyColumns.Any(kc => kc == _.Key)).ToArray(); - // WITH CHANGE_TRACKING_CONTEXT(@sync_client_id) must immediately precede the DML statement + // WITH CHANGE_TRACKING_CONTEXT(@sync_client_id) must immediately precede the DML statement, + // and the leading semicolon terminates whatever came before it (an empty statement when the + // batch starts here). + // + // These statements are deliberately not wrapped in BEGIN TRY/CATCH: a swallowed error makes + // the batch report zero affected rows, which the caller cannot tell apart from a genuine + // write conflict. Letting the SqlException surface keeps "zero rows" meaning "conflict". const string ctContext = ";WITH CHANGE_TRACKING_CONTEXT(@sync_client_id) "; switch (itemChangeType) @@ -109,13 +115,8 @@ internal void SetupCommand(SqlCommand cmd, ChangeType itemChangeType, Dictionary identityInsertCommand = $"SET IDENTITY_INSERT {NameWithSchema} OFF\n"; } - cmd.CommandText = $@"{identityInsertCommand}BEGIN TRY -{ctContext}INSERT INTO {NameWithSchema} ({string.Join(", ", allSyncItems.Select(_ => "[" + _.Key + "]"))}) + cmd.CommandText = $@"{identityInsertCommand}{ctContext}INSERT INTO {NameWithSchema} ({string.Join(", ", allSyncItems.Select(_ => "[" + _.Key + "]"))}) VALUES ({string.Join(", ", allSyncItems.Select((_, index) => $"@p{index}"))}); -END TRY -BEGIN CATCH -PRINT ERROR_MESSAGE() -END CATCH "; int pIndex = 0; @@ -130,19 +131,14 @@ END CATCH case ChangeType.Update: { - cmd.CommandText = $@"BEGIN TRY -{ctContext}UPDATE {NameWithSchema} + cmd.CommandText = $@"{ctContext}UPDATE {NameWithSchema} SET {string.Join(", ", allSyncItemsExceptPrimaryKey.Select((_, index) => $"[{_.Key}] = @p{index}"))} WHERE {NameWithSchema}.[{PrimaryColumnName}] = @PrimaryColumnParameter AND (@sync_force_write = 1 OR NOT EXISTS ( SELECT 1 FROM CHANGETABLE(CHANGES {NameWithSchema}, @last_sync_version) AS CT WHERE CT.[{PrimaryColumnName}] = @PrimaryColumnParameter AND (CT.SYS_CHANGE_CONTEXT IS NULL OR CT.SYS_CHANGE_CONTEXT <> @sync_client_id) -)) -END TRY -BEGIN CATCH -PRINT ERROR_MESSAGE() -END CATCH"; +))"; cmd.Parameters.Add(Columns[PrimaryColumnName].CreateParameter("@PrimaryColumnParameter", syncItemValues[PrimaryColumnName])); @@ -158,18 +154,13 @@ PRINT ERROR_MESSAGE() case ChangeType.Delete: { - cmd.CommandText = $@"BEGIN TRY -{ctContext}DELETE FROM {NameWithSchema} + cmd.CommandText = $@"{ctContext}DELETE FROM {NameWithSchema} WHERE {NameWithSchema}.[{PrimaryColumnName}] = @PrimaryColumnParameter AND (@sync_force_write = 1 OR NOT EXISTS ( SELECT 1 FROM CHANGETABLE(CHANGES {NameWithSchema}, @last_sync_version) AS CT WHERE CT.[{PrimaryColumnName}] = @PrimaryColumnParameter AND (CT.SYS_CHANGE_CONTEXT IS NULL OR CT.SYS_CHANGE_CONTEXT <> @sync_client_id) -)) -END TRY -BEGIN CATCH -PRINT ERROR_MESSAGE() -END CATCH"; +))"; cmd.Parameters.Add(Columns[PrimaryColumnName] .CreateParameter("@PrimaryColumnParameter", syncItemValues[PrimaryColumnName])); diff --git a/src/CoreSync.Tests/SqlServerCTResilienceTests.cs b/src/CoreSync.Tests/SqlServerCTResilienceTests.cs new file mode 100644 index 0000000..ce48da8 --- /dev/null +++ b/src/CoreSync.Tests/SqlServerCTResilienceTests.cs @@ -0,0 +1,424 @@ +using CoreSync.SqlServerCT; +using Microsoft.Data.SqlClient; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using Shouldly; +using System; +using System.Collections.Generic; +using System.Threading.Tasks; + +namespace CoreSync.Tests; + +/// +/// Covers the failure modes behind the "clients silently stop uploading updates" incident on the +/// SQL Server Change Tracking provider: +/// +/// an upload whose anchor predates the table's change tracking retention window must +/// fail with a typed, non-retryable rather than be masked; +/// a missing anchor row must read back as , not as version 0 +/// (which looks valid but sits permanently below the retention floor); +/// a genuine constraint violation must surface as an exception instead of being +/// misreported as a write conflict and dropped; +/// the configured change retention must be applied when provisioning a database that +/// already has change tracking enabled. +/// +/// +[TestClass] +public class SqlServerCTResilienceTests +{ + private static string SqlServerConnectionString => Environment.GetEnvironmentVariable("CORE-SYNC_CONNECTION_STRING") ?? + "Server=localhost;User Id=sa;Password=CoreSync_Test123!;TrustServerCertificate=True"; + + #region Helpers + + private static async Task CreateDatabase(string dbName) + { + using var conn = new SqlConnection(SqlServerConnectionString + ";Initial Catalog=master"); + await conn.OpenAsync(); + using var cmd = conn.CreateCommand(); + cmd.CommandText = $@" + IF DB_ID('{dbName}') IS NOT NULL + BEGIN + ALTER DATABASE [{dbName}] SET SINGLE_USER WITH ROLLBACK IMMEDIATE; + DROP DATABASE [{dbName}]; + END + CREATE DATABASE [{dbName}];"; + await cmd.ExecuteNonQueryAsync(); + } + + private static async Task DropDatabase(string dbName) + { + using var conn = new SqlConnection(SqlServerConnectionString + ";Initial Catalog=master"); + await conn.OpenAsync(); + using var cmd = conn.CreateCommand(); + cmd.CommandText = $@" + IF DB_ID('{dbName}') IS NOT NULL + BEGIN + ALTER DATABASE [{dbName}] SET SINGLE_USER WITH ROLLBACK IMMEDIATE; + DROP DATABASE [{dbName}]; + END"; + await cmd.ExecuteNonQueryAsync(); + } + + /// + /// Authors/Articles, with Articles.AuthorId a real foreign key so constraint violations can be + /// provoked on purpose. + /// + private static async Task CreateSchema(string connectionString) + { + await ExecuteNonQuery(connectionString, @" + CREATE TABLE [dbo].[Authors] ( + [Id] INT NOT NULL PRIMARY KEY, + [Name] NVARCHAR(100) NOT NULL + ); + CREATE TABLE [dbo].[Articles] ( + [Id] INT NOT NULL PRIMARY KEY, + [Title] NVARCHAR(200) NOT NULL, + [AuthorId] INT NULL REFERENCES [dbo].[Authors]([Id]) + );"); + } + + private static async Task ExecuteNonQuery(string connectionString, string commandText) + { + using var conn = new SqlConnection(connectionString); + await conn.OpenAsync(); + using var cmd = conn.CreateCommand(); + cmd.CommandText = commandText; + await cmd.ExecuteNonQueryAsync(); + } + + private static async Task ExecuteScalar(string connectionString, string commandText) + { + using var conn = new SqlConnection(connectionString); + await conn.OpenAsync(); + using var cmd = conn.CreateCommand(); + cmd.CommandText = commandText; + var res = await cmd.ExecuteScalarAsync(); + return res == DBNull.Value ? null : res; + } + + private static async Task GetCurrentVersion(string connectionString) + => Convert.ToInt64(await ExecuteScalar(connectionString, "SELECT CHANGE_TRACKING_CURRENT_VERSION()")); + + private static async Task GetMinValidVersion(string connectionString, string tableNameWithSchema) + => Convert.ToInt64(await ExecuteScalar(connectionString, $"SELECT CHANGE_TRACKING_MIN_VALID_VERSION(OBJECT_ID('{tableNameWithSchema}'))")); + + /// + /// Pushes a table's minimum valid version up to the current database version, which is what the + /// change tracking auto-cleanup task does once the retention window elapses - without having to + /// wait for it. + /// + private static Task ExpireChangeHistory(string connectionString, string tableNameWithSchema) + => ExecuteNonQuery(connectionString, $@" + ALTER TABLE {tableNameWithSchema} DISABLE CHANGE_TRACKING; + ALTER TABLE {tableNameWithSchema} ENABLE CHANGE_TRACKING WITH (TRACK_COLUMNS_UPDATED = ON);"); + + private static async Task<(int retentionPeriod, int retentionPeriodUnits, bool autoCleanup)> GetChangeTrackingSettings(string connectionString) + { + using var conn = new SqlConnection(connectionString); + await conn.OpenAsync(); + using var cmd = conn.CreateCommand(); + cmd.CommandText = @"SELECT retention_period, retention_period_units, is_auto_cleanup_on +FROM sys.change_tracking_databases WHERE database_id = DB_ID()"; + using var reader = await cmd.ExecuteReaderAsync(); + (await reader.ReadAsync()).ShouldBeTrue("change tracking should be enabled on the database"); + // retention_period_units and is_auto_cleanup_on are tinyint, not int/bit. + return (Convert.ToInt32(reader.GetValue(0)), Convert.ToInt32(reader.GetValue(1)), Convert.ToBoolean(reader.GetValue(2))); + } + + private static SyncItem Item(string tableName, ChangeType changeType, Dictionary values) + => new(tableName, changeType, values); + + private static SqlException? FindSqlException(Exception? ex) + { + while (ex != null) + { + if (ex is SqlException sqlException) + return sqlException; + ex = ex.InnerException; + } + return null; + } + + private static async Task RunInDatabase(string dbName, Func body) + { + try + { + await CreateDatabase(dbName); + await CreateSchema(SqlServerConnectionString + $";Initial Catalog={dbName}"); + await body(SqlServerConnectionString + $";Initial Catalog={dbName}"); + } + finally + { + SqlConnection.ClearAllPools(); + await DropDatabase(dbName); + } + } + + private static SqlServerCTProvider CreateProvider(string connectionString, string label = "SUT") + { + var config = new SqlServerCTSyncConfigurationBuilder(connectionString) + .Table("Authors") + .Table("Articles") + .Build(); + + return new SqlServerCTProvider(config, logger: new ConsoleLogger(label)); + } + + #endregion + + /// + /// A store that holds no anchor row for a peer must report the null sentinel. Version 0 is not a + /// sentinel: it is a real-looking version permanently below the retention floor, which is what made + /// every subsequent update from such a peer fail. + /// + [TestMethod] + public async Task GetChanges_WithNoAnchorRowForPeer_ReportsNullTargetAnchor() + { + await RunInDatabase("CoreSyncCT_NullAnchor", async connStr => + { + var provider = CreateProvider(connStr); + await provider.ApplyProvisionAsync(); + + var changeSet = await provider.GetChangesAsync(Guid.NewGuid()); + + changeSet.TargetAnchor.IsNull().ShouldBeTrue("a missing anchor row must read back as the null sentinel"); + changeSet.TargetAnchor.Version.ShouldBe(-1); + }); + } + + /// + /// The end of the chain the previous test starts: an update sent with no recorded anchor cannot be + /// conflict-checked at all, so it must be rejected as "reinitialize this client" rather than be + /// pushed into CHANGETABLE with a doomed version. + /// + [TestMethod] + public async Task ApplyChanges_UpdateWithNullTargetAnchor_ThrowsSyncAnchorTooOld() + { + await RunInDatabase("CoreSyncCT_UpdateNullAnchor", async connStr => + { + var provider = CreateProvider(connStr); + await provider.ApplyProvisionAsync(); + + await ExecuteNonQuery(connStr, "INSERT INTO [dbo].[Authors] ([Id], [Name]) VALUES (1, 'Original')"); + + var changeSet = new SyncChangeSet( + new SyncAnchor(Guid.NewGuid(), 1), + SyncAnchor.Null, + new[] + { + Item("Authors", ChangeType.Update, new Dictionary { ["Id"] = 1, ["Name"] = "Updated" }) + }); + + var ex = await Should.ThrowAsync(() => provider.ApplyChangesAsync(changeSet)); + + ex.TableName.ShouldContain("Authors"); + ex.RequestedVersion.ShouldBe(-1); + ex.MinValidVersion.ShouldBeGreaterThanOrEqualTo(0); + + // Nothing was written. + (await ExecuteScalar(connStr, "SELECT [Name] FROM [dbo].[Authors] WHERE [Id] = 1")).ShouldBe("Original"); + }); + } + + /// + /// The production failure itself: an anchor that is a perfectly ordinary looking version number but + /// has fallen out of the retention window. It used to be handed straight to + /// CHANGETABLE(CHANGES ..., @last_sync_version) and the resulting error was swallowed. + /// + [TestMethod] + public async Task ApplyChanges_UpdateWithAnchorBelowMinValidVersion_ThrowsSyncAnchorTooOld() + { + await RunInDatabase("CoreSyncCT_StaleAnchor", async connStr => + { + var provider = CreateProvider(connStr); + await provider.ApplyProvisionAsync(); + + await ExecuteNonQuery(connStr, "INSERT INTO [dbo].[Authors] ([Id], [Name]) VALUES (1, 'Original')"); + await ExecuteNonQuery(connStr, "UPDATE [dbo].[Authors] SET [Name] = 'Second' WHERE [Id] = 1"); + + var staleVersion = await GetCurrentVersion(connStr); + staleVersion.ShouldBeGreaterThan(0); + + await ExpireChangeHistory(connStr, "[dbo].[Authors]"); + (await GetMinValidVersion(connStr, "[dbo].[Authors]")).ShouldBeGreaterThan(0); + + // The peer is still asking for changes relative to a version the store no longer retains. + var changeSet = new SyncChangeSet( + new SyncAnchor(Guid.NewGuid(), 1), + new SyncAnchor(await provider.GetStoreIdAsync(), 0), + new[] + { + Item("Authors", ChangeType.Update, new Dictionary { ["Id"] = 1, ["Name"] = "Updated" }) + }); + + var ex = await Should.ThrowAsync(() => provider.ApplyChangesAsync(changeSet)); + + ex.TableName.ShouldContain("Authors"); + ex.RequestedVersion.ShouldBe(0); + ex.MinValidVersion.ShouldBeGreaterThan(0); + + (await ExecuteScalar(connStr, "SELECT [Name] FROM [dbo].[Authors] WHERE [Id] = 1")).ShouldBe("Second"); + }); + } + + /// + /// Negative control for the guard above: inserts never touch CHANGETABLE, so a first-time sync - + /// which legitimately carries a null target anchor and nothing but inserts - must still go through. + /// + [TestMethod] + public async Task ApplyChanges_InsertOnlyChangeSetWithNullTargetAnchor_IsApplied() + { + await RunInDatabase("CoreSyncCT_FirstSync", async connStr => + { + var provider = CreateProvider(connStr); + await provider.ApplyProvisionAsync(); + + var changeSet = new SyncChangeSet( + new SyncAnchor(Guid.NewGuid(), 1), + SyncAnchor.Null, + new[] + { + Item("Authors", ChangeType.Insert, new Dictionary { ["Id"] = 1, ["Name"] = "First" }), + Item("Articles", ChangeType.Insert, new Dictionary { ["Id"] = 10, ["Title"] = "Hello", ["AuthorId"] = 1 }) + }); + + await provider.ApplyChangesAsync(changeSet); + + (await ExecuteScalar(connStr, "SELECT [Name] FROM [dbo].[Authors] WHERE [Id] = 1")).ShouldBe("First"); + (await ExecuteScalar(connStr, "SELECT [Title] FROM [dbo].[Articles] WHERE [Id] = 10")).ShouldBe("Hello"); + }); + } + + /// + /// A foreign key violation on an update used to be swallowed by the T-SQL CATCH; the batch then + /// reported zero affected rows, which the provider read as a write conflict and - with the default + /// Skip resolution - dropped the item. That is silent data loss. It must throw instead, and the + /// conflict callback must never see the item. + /// + [TestMethod] + public async Task ApplyChanges_UpdateViolatingForeignKey_ThrowsAndIsNotReportedAsConflict() + { + await RunInDatabase("CoreSyncCT_FkViolation", async connStr => + { + var provider = CreateProvider(connStr); + await provider.ApplyProvisionAsync(); + + await ExecuteNonQuery(connStr, @" + INSERT INTO [dbo].[Authors] ([Id], [Name]) VALUES (1, 'Author'); + INSERT INTO [dbo].[Articles] ([Id], [Title], [AuthorId]) VALUES (10, 'Title', 1);"); + + // Anchor at the current version so the conflict check passes and the UPDATE actually runs. + var currentVersion = await GetCurrentVersion(connStr); + + var conflicts = new List(); + + var changeSet = new SyncChangeSet( + new SyncAnchor(Guid.NewGuid(), 1), + new SyncAnchor(await provider.GetStoreIdAsync(), currentVersion), + new[] + { + // Author 999 does not exist. + Item("Articles", ChangeType.Update, new Dictionary { ["Id"] = 10, ["Title"] = "Retitled", ["AuthorId"] = 999 }) + }); + + var ex = await Should.ThrowAsync(() => provider.ApplyChangesAsync(changeSet, item => + { + conflicts.Add(item); + return ConflictResolution.Skip; + })); + + conflicts.ShouldBeEmpty("a constraint violation is not a write conflict and must not reach the conflict handler"); + + var sqlException = FindSqlException(ex); + sqlException.ShouldNotBeNull("the underlying SqlException must survive in the exception chain"); + sqlException.Number.ShouldBe(547, "547 is the FOREIGN KEY / CHECK constraint violation"); + + // The whole change set was rolled back. + (await ExecuteScalar(connStr, "SELECT [Title] FROM [dbo].[Articles] WHERE [Id] = 10")).ShouldBe("Title"); + (await ExecuteScalar(connStr, "SELECT [AuthorId] FROM [dbo].[Articles] WHERE [Id] = 10")).ShouldBe(1); + }); + } + + /// + /// The main regression: ChangeRetention used to reach the database only through the + /// SET CHANGE_TRACKING = ON (...) statement, which provisioning skips whenever change tracking + /// is already enabled - so on every real deployment the setting was silently ignored forever. + /// Provisioning must reconcile it instead. + /// + [TestMethod] + public async Task ApplyProvision_AppliesChangeRetention_WhenChangeTrackingIsAlreadyEnabled() + { + await RunInDatabase("CoreSyncCT_Retention", async connStr => + { + // First provisioning turns change tracking on and sets the retention. + var initial = new SqlServerCTProvider( + new SqlServerCTSyncConfigurationBuilder(connStr).Table("Authors").Table("Articles").ChangeRetention(2).Build(), + logger: new ConsoleLogger("P1")); + await initial.ApplyProvisionAsync(); + + (await GetChangeTrackingSettings(connStr)).ShouldBe((2, (int)ChangeRetentionUnit.Days, true)); + + // Second provisioning runs against a database that already has change tracking enabled. + var reconfigured = new SqlServerCTProvider( + new SqlServerCTSyncConfigurationBuilder(connStr).Table("Authors").Table("Articles").ChangeRetention(90, autoCleanup: false).Build(), + logger: new ConsoleLogger("P2")); + await reconfigured.ApplyProvisionAsync(); + + (await GetChangeTrackingSettings(connStr)).ShouldBe((90, (int)ChangeRetentionUnit.Days, false), + "the configured retention must be applied even when change tracking is already on"); + }); + } + + /// + /// Retention is also configurable in units other than days, and a database that already has change + /// tracking enabled is left alone when the caller never asked for a particular retention. + /// + [TestMethod] + public async Task ApplyProvision_AppliesChangeRetentionWithExplicitUnit_AndLeavesItAloneWhenUnconfigured() + { + await RunInDatabase("CoreSyncCT_RetentionUnit", async connStr => + { + var configured = new SqlServerCTProvider( + new SqlServerCTSyncConfigurationBuilder(connStr).Table("Authors").Table("Articles") + .ChangeRetention(45, ChangeRetentionUnit.Minutes).Build(), + logger: new ConsoleLogger("P1")); + await configured.ApplyProvisionAsync(); + + (await GetChangeTrackingSettings(connStr)).ShouldBe((45, (int)ChangeRetentionUnit.Minutes, true)); + + // A provider that never called ChangeRetention must not silently reset the database to the + // 7 day default. + var unconfigured = new SqlServerCTProvider( + new SqlServerCTSyncConfigurationBuilder(connStr).Table("Authors").Table("Articles").Build(), + logger: new ConsoleLogger("P2")); + await unconfigured.ApplyProvisionAsync(); + + (await GetChangeTrackingSettings(connStr)).ShouldBe((45, (int)ChangeRetentionUnit.Minutes, true), + "an unconfigured retention must leave the existing setting untouched"); + }); + } + + /// + /// The builder rejects a non-positive retention rather than emitting invalid T-SQL. + /// + [TestMethod] + public void ChangeRetention_RejectsNonPositiveValues() + { + var builder = new SqlServerCTSyncConfigurationBuilder("Server=.;Initial Catalog=x;Integrated Security=true").Table("Authors"); + + Should.Throw(() => builder.ChangeRetention(0)); + Should.Throw(() => builder.ChangeRetention(-1, ChangeRetentionUnit.Hours)); + + // The days-only overload keeps working and is expressed in days. + var config = builder.ChangeRetention(14).Build(); + config.ChangeRetention.ShouldBe(14); + config.ChangeRetentionUnit.ShouldBe(ChangeRetentionUnit.Days); + config.ChangeRetentionDays.ShouldBe(14); + config.AutoCleanup.ShouldBeTrue(); + + var inMinutes = new SqlServerCTSyncConfigurationBuilder("Server=.;Initial Catalog=x;Integrated Security=true") + .Table("Authors").ChangeRetention(90, ChangeRetentionUnit.Minutes).Build(); + inMinutes.ChangeRetentionInMinutes.ShouldBe(90); + inMinutes.ChangeRetentionDays.ShouldBe(1, "rounded up to whole days for the legacy property"); + } +} diff --git a/src/CoreSync/SyncAgent.cs b/src/CoreSync/SyncAgent.cs index 252e23a..1994a2c 100644 --- a/src/CoreSync/SyncAgent.cs +++ b/src/CoreSync/SyncAgent.cs @@ -88,6 +88,13 @@ public async Task SynchronizeAsync( //await LocalSyncProvider.ApplyProvisionAsync(cancellationToken: cancellationToken); //await RemoteSyncProvider.ApplyProvisionAsync(cancellationToken: cancellationToken); } + catch (SyncAnchorTooOldException) + { + // Deliberately not wrapped: this one is not retryable, the client has to be + // reinitialized from a fresh snapshot, and callers need to see that without digging + // through inner exceptions. + throw; + } catch (Exception ex) { throw new SynchronizationException("Unable to synchronize stores", ex); diff --git a/src/CoreSync/SyncAnchorTooOldException.cs b/src/CoreSync/SyncAnchorTooOldException.cs new file mode 100644 index 0000000..2fe64f2 --- /dev/null +++ b/src/CoreSync/SyncAnchorTooOldException.cs @@ -0,0 +1,85 @@ +using System; + +namespace CoreSync +{ + /// + /// Raised when a store is asked to resolve changes relative to a version whose change history + /// has already been discarded by the store's change retention policy. + /// + /// + /// This is a permanent condition for the store involved: retrying the same operation with + /// the same anchor can never succeed, because the change information the anchor refers to no + /// longer exists (for example it was removed by the SQL Server Change Tracking auto-cleanup task). + /// + /// Callers should treat it as a signal to reinitialize the affected client from scratch - drop the + /// local copy and take a fresh initial snapshot - as opposed to a transient failure worth retrying. + /// Retrying instead leaves the client permanently unable to upload or download changes. + /// + /// + /// It derives from so that callers written against the + /// previous, untyped behaviour keep working. + /// + /// + public class SyncAnchorTooOldException : InvalidOperationException + { + /// + /// Initializes a new instance of the class. + /// + /// The table whose change history no longer covers the requested version. + /// + /// The version the operation asked for. A negative value means no anchor was recorded at all + /// (see ), which is equally unrecoverable. + /// + /// The oldest version the store can still resolve changes from. + public SyncAnchorTooOldException(string tableName, long requestedVersion, long minValidVersion) + : base(BuildMessage(tableName, requestedVersion, minValidVersion)) + { + TableName = tableName; + RequestedVersion = requestedVersion; + MinValidVersion = minValidVersion; + } + + /// + /// Initializes a new instance of the class with an + /// inner exception. + /// + /// The table whose change history no longer covers the requested version. + /// The version the operation asked for. + /// The oldest version the store can still resolve changes from. + /// The underlying error, when the condition was detected through one. + public SyncAnchorTooOldException(string tableName, long requestedVersion, long minValidVersion, Exception innerException) + : base(BuildMessage(tableName, requestedVersion, minValidVersion), innerException) + { + TableName = tableName; + RequestedVersion = requestedVersion; + MinValidVersion = minValidVersion; + } + + /// + /// Gets the name of the table whose change history no longer covers . + /// + public string TableName { get; } + + /// + /// Gets the version the operation asked for. A negative value means the store holds no anchor + /// at all for the peer. + /// + public long RequestedVersion { get; } + + /// + /// Gets the oldest version the store can still resolve changes from. + /// + public long MinValidVersion { get; } + + private static string BuildMessage(string tableName, long requestedVersion, long minValidVersion) + { + var requested = requestedVersion < 0 + ? "no anchor is recorded for the peer store" + : $"version {requestedVersion} is older than the retained history"; + + return $"Unable to resolve changes for table '{tableName}': {requested} " + + $"(requested version {requestedVersion}, minimum valid version {minValidVersion}). " + + "The peer store must be reinitialized from a fresh snapshot; retrying will not help."; + } + } +}