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."; + } + } +}