Replace ReplicationProgressHandler with IProgress<T>

Documentation/CHANGELOG.md

@@ -27,6 +27,7 @@ All notable changes to this project will be documented in this file. The format
 
 ### Changed
 
+- **Breaking (`DSInternals.Replication` library):** Replaced the custom `ReplicationProgressHandler` delegate with the standard `IProgress<ReplicationProgress>` interface on `DirectoryReplicationClient.GetAccounts`, `ReplicateAllObjects`, and `FetchFullSchema`. The methods also accept an optional `CancellationToken` so long-running replication can be cooperatively cancelled between cycles. External consumers must migrate from the delegate to `IProgress<ReplicationProgress>`.
 - Renamed the `Get-ADDBDnsResourceRecord` cmdlet to [Get-ADDBDnsServerResourceRecord](PowerShell/Get-ADDBDnsServerResourceRecord.md#get-addbdnsserverresourcerecord), the `Save-DnsResourceRecord` cmdlet to [Save-DnsServerResourceRecord](PowerShell/Save-DnsServerResourceRecord.md#save-dnsserverresourcerecord), and the `Get-ADDBDnsZone` cmdlet to [Get-ADDBDnsServerZone](PowerShell/Get-ADDBDnsServerZone.md#get-addbdnsserverzone) for naming consistency with [Get-ADSIDnsServerResourceRecord](PowerShell/Get-ADSIDnsServerResourceRecord.md#get-adsidnsserverresourcerecord) and [Get-ADSIDnsServerZone](PowerShell/Get-ADSIDnsServerZone.md#get-adsidnsserverzone). The previous names are preserved as aliases.
 - Renamed the `Save-DPAPIBlob` cmdlet to [Save-DpapiBlob](PowerShell/Save-DpapiBlob.md#save-dpapiblob) for consistent casing with the other DPAPI cmdlets. PowerShell is case-insensitive, so existing scripts continue to work.
 - The `-Encoding` parameter on [Protect-DpapiNgData](PowerShell/Protect-DpapiNgData.md#protect-dpapingdata) and [Unprotect-DpapiNgData](PowerShell/Unprotect-DpapiNgData.md#unprotect-dpapingdata) now offers tab completion and accepts strings such as `UTF8`, `Unicode`, or `ASCII` in addition to `System.Text.Encoding` instances.

Src/DSInternals.PowerShell/Commands/Base/SynchronousProgress.cs

@@ -0,0 +1,22 @@
+namespace DSInternals.PowerShell.Commands;
+
+/// <summary>
+/// An <see cref="IProgress{T}"/> implementation that invokes the supplied handler synchronously on the calling thread.
+/// </summary>
+/// <remarks>
+/// PowerShell cmdlets must call <c>WriteProgress</c> from the pipeline thread. The BCL's <see cref="Progress{T}"/>
+/// dispatches asynchronously through the captured <see cref="SynchronizationContext"/> (or the thread pool when none
+/// is present), which would marshal the callback off the pipeline thread and cause <c>WriteProgress</c> to throw.
+/// </remarks>
+internal sealed class SynchronousProgress<T> : IProgress<T>
+{
+    private readonly Action<T> _handler;
+
+    public SynchronousProgress(Action<T> handler)
+    {
+        ArgumentNullException.ThrowIfNull(handler);
+        _handler = handler;
+    }
+
+    public void Report(T value) => _handler(value);
+}

Src/DSInternals.PowerShell/Commands/Replication/GetADReplAccountCommand.cs

@@ -2,7 +2,6 @@
 using System.Security.Principal;
 using DSInternals.Common.Data;
 using DSInternals.Replication;
-using DSInternals.Replication.Model;
 
 namespace DSInternals.PowerShell.Commands;
 
@@ -110,38 +109,67 @@ protected override void ProcessRecord()
             this.ReturnSingleAccount();
         }
     }
+
+    protected override void StopProcessing()
+    {
+        _cancellationTokenSource.Cancel();
+        base.StopProcessing();
+    }
+
+    protected override void Dispose(bool disposing)
+    {
+        if (disposing)
+        {
+            _cancellationTokenSource.Dispose();
+        }
+
+        base.Dispose(disposing);
+    }
     #endregion Cmdlet Overrides
 
+    #region Cancellation
+    private readonly CancellationTokenSource _cancellationTokenSource = new();
+    #endregion Cancellation
+
     #region Helper Methods
 
     protected void ReturnAllAccounts()
     {
         // Write the initial progress
-        var progress = new ProgressRecord(1, "Account Replication", "Replicating Active Directory objects.");
-        progress.PercentComplete = 0;
-        this.WriteProgress(progress);
+        var progressRecord = new ProgressRecord(1, "Account Replication", "Replicating Active Directory objects.");
+        progressRecord.PercentComplete = 0;
+        this.WriteProgress(progressRecord);
 
         // Update the progress after each replication cycle
-        ReplicationProgressHandler progressReporter = (ReplicationCookie cookie, int processedObjectCount, int totalObjectCount) =>
+        IProgress<ReplicationProgress> progress = new SynchronousProgress<ReplicationProgress>(report =>
         {
-            int percentComplete = (int)(((double)processedObjectCount / (double)totalObjectCount) * 100);
+            int percentComplete = (int)(((double)report.ProcessedObjectCount / (double)report.TotalObjectCount) * 100);
             // AD's object count estimate is sometimes lower than the actual count, so we cap the value to 100%.
-            progress.PercentComplete = Math.Min(percentComplete, 100);
-            this.WriteProgress(progress);
-        };
+            progressRecord.PercentComplete = Math.Min(percentComplete, 100);
+            this.WriteProgress(progressRecord);
+        });
 
         // Automatically infer domain name if no value is provided
         string domainNamingContext = this.NamingContext ?? this.ReplicationClient.DomainNamingContext;
 
-        // Replicate all accounts
-        foreach (var account in this.ReplicationClient.GetAccounts(domainNamingContext, progressReporter, this.Properties))
+        try
+        {
+            // Replicate all accounts
+            foreach (var account in this.ReplicationClient.GetAccounts(domainNamingContext, progress, this.Properties, _cancellationTokenSource.Token))
+            {
+                this.WriteObject(account);
+            }
+        }
+        catch (OperationCanceledException)
         {
-            this.WriteObject(account);
+            // The pipeline is stopping (e.g. Ctrl+C). Fall through to a clean progress completion.
+        }
+        finally
+        {
+            // Write progress completed
+            progressRecord.RecordType = ProgressRecordType.Completed;
+            this.WriteProgress(progressRecord);
         }
-
-        // Write progress completed
-        progress.RecordType = ProgressRecordType.Completed;
-        this.WriteProgress(progress);
     }
 
     protected void ReturnSingleAccount()
@@ -183,25 +211,34 @@ protected void ReturnSingleAccount()
     protected void FetchSchema()
     {
         // Write the initial progress
-        var progress = new ProgressRecord(2, "Schema Replication", "Replicating Active Directory schema.");
-        progress.PercentComplete = 0;
-        this.WriteProgress(progress);
+        var progressRecord = new ProgressRecord(2, "Schema Replication", "Replicating Active Directory schema.");
+        progressRecord.PercentComplete = 0;
+        this.WriteProgress(progressRecord);
 
         // Update the progress after each replication cycle
-        ReplicationProgressHandler progressReporter = (ReplicationCookie cookie, int processedObjectCount, int totalObjectCount) =>
+        IProgress<ReplicationProgress> progress = new SynchronousProgress<ReplicationProgress>(report =>
         {
-            int percentComplete = (int)(((double)processedObjectCount / (double)totalObjectCount) * 100);
+            int percentComplete = (int)(((double)report.ProcessedObjectCount / (double)report.TotalObjectCount) * 100);
             // AD's object count estimate is sometimes lower than the actual count, so we cap the value to 100%.
-            progress.PercentComplete = Math.Min(percentComplete, 100);
-            this.WriteProgress(progress);
-        };
-
-        // Replicate the schema partition
-        this.ReplicationClient.FetchFullSchema(progressReporter);
+            progressRecord.PercentComplete = Math.Min(percentComplete, 100);
+            this.WriteProgress(progressRecord);
+        });
 
-        // Write progress completed
-        progress.RecordType = ProgressRecordType.Completed;
-        this.WriteProgress(progress);
+        try
+        {
+            // Replicate the schema partition
+            this.ReplicationClient.FetchFullSchema(progress, _cancellationTokenSource.Token);
+        }
+        catch (OperationCanceledException)
+        {
+            // The pipeline is stopping (e.g. Ctrl+C). Fall through to a clean progress completion.
+        }
+        finally
+        {
+            // Write progress completed
+            progressRecord.RecordType = ProgressRecordType.Completed;
+            this.WriteProgress(progressRecord);
+        }
     }
 
     private new void WriteObject(object sendToPipeline)

Src/DSInternals.Replication/DirectoryReplicationClient.cs

@@ -150,14 +150,15 @@ public ReplicationCursor[] GetReplicationCursors(string namingContext)
     /// Retrieves all accounts from the specified domain partition.
     /// </summary>
     /// <param name="domainNamingContext">The distinguished name of the domain partition.</param>
-    /// <param name="progressReporter">The progress reporter to report replication progress.</param>
+    /// <param name="progress">Optional progress reporter invoked after each replication cycle.</param>
     /// <param name="propertySets">The set of properties to retrieve for each account.</param>
+    /// <param name="cancellationToken">Token used to cooperatively cancel the replication between cycles.</param>
     /// <returns>An enumerable collection of directory service accounts.</returns>
-    public IEnumerable<DSAccount> GetAccounts(string domainNamingContext, ReplicationProgressHandler progressReporter = null, AccountPropertySets propertySets = AccountPropertySets.All)
+    public IEnumerable<DSAccount> GetAccounts(string domainNamingContext, IProgress<ReplicationProgress> progress = null, AccountPropertySets propertySets = AccountPropertySets.All, CancellationToken cancellationToken = default)
     {
         ArgumentException.ThrowIfNullOrWhiteSpace(domainNamingContext);
 
-        return ReplicateAllObjects(domainNamingContext, progressReporter)
+        return ReplicateAllObjects(domainNamingContext, progress, cancellationToken)
             .Select(dsObject => AccountFactory.CreateAccount(dsObject, this.NetBIOSDomainName, _secretDecryptor.Value, _rootKeyResolver, propertySets))
             .Where(account => account != null); // CreateAccount returns null for other object types
     }
@@ -166,9 +167,10 @@ public IEnumerable<DSAccount> GetAccounts(string domainNamingContext, Replicatio
     /// Retrieves all directory objects from the specified naming context.
     /// </summary>
     /// <param name="namingContext">Partition to replicate.</param>
-    /// <param name="progressReporter">Progress reporter for replication progress.</param>
+    /// <param name="progress">Optional progress reporter invoked after each replication cycle.</param>
+    /// <param name="cancellationToken">Token used to cooperatively cancel the replication between cycles.</param>
     /// <returns>An enumerable collection of directory service objects.</returns>
-    public IEnumerable<ReplicaObject> ReplicateAllObjects(string namingContext, ReplicationProgressHandler progressReporter = null)
+    public IEnumerable<ReplicaObject> ReplicateAllObjects(string namingContext, IProgress<ReplicationProgress> progress = null, CancellationToken cancellationToken = default)
     {
         ArgumentException.ThrowIfNullOrWhiteSpace(namingContext);
         ReplicationCookie currentCookie = new(namingContext);
@@ -177,15 +179,15 @@ public IEnumerable<ReplicaObject> ReplicateAllObjects(string namingContext, Repl
 
         do
         {
+            // Check for cancellation between replication cycles (the native DRS call itself is not interruptible).
+            cancellationToken.ThrowIfCancellationRequested();
+
             // Perform one replication cycle
             result = this._drsConnection.ReplicateAllObjects(currentCookie);
 
             // Report replication progress
-            if (progressReporter != null)
-            {
-                processedObjectCount += result.Objects.Count;
-                progressReporter(result.Cookie, processedObjectCount, result.TotalObjectCount);
-            }
+            processedObjectCount += result.Objects.Count;
+            progress?.Report(new ReplicationProgress(result.Cookie, processedObjectCount, result.TotalObjectCount));
 
             // Pass-through the returned objects
             foreach (var obj in result.Objects)
@@ -451,8 +453,9 @@ public void AddSidHistory(
     /// <summary>
     /// Replicates the entire schema partition.
     /// </summary>
-    /// <param name="progressReporter">Replication progress reporter.</param>
-    public void FetchFullSchema(ReplicationProgressHandler progressReporter = null)
+    /// <param name="progress">Optional progress reporter invoked after each replication cycle.</param>
+    /// <param name="cancellationToken">Token used to cooperatively cancel the replication between cycles.</param>
+    public void FetchFullSchema(IProgress<ReplicationProgress> progress = null, CancellationToken cancellationToken = default)
     {
         if (_isFullSchemaLoaded)
         {
@@ -470,15 +473,15 @@ public void FetchFullSchema(ReplicationProgressHandler progressReporter = null)
 
         do
         {
+            // Check for cancellation between replication cycles (the native DRS call itself is not interruptible).
+            cancellationToken.ThrowIfCancellationRequested();
+
             // Perform one replication cycle
             result = this._drsConnection.ReplicateAllObjects(currentCookie);
 
             // Report replication progress
-            if (progressReporter != null)
-            {
-                processedObjectCount += result.Objects.Count;
-                progressReporter(result.Cookie, processedObjectCount, result.TotalObjectCount);
-            }
+            processedObjectCount += result.Objects.Count;
+            progress?.Report(new ReplicationProgress(result.Cookie, processedObjectCount, result.TotalObjectCount));
 
             // Merge the prefix tables
             if (result.PrefixTable != null)

Src/DSInternals.Replication/ReplicationProgress.cs

@@ -0,0 +1,14 @@
+using DSInternals.Replication.Model;
+
+namespace DSInternals.Replication;
+
+/// <summary>
+/// Progress information reported during a replication operation.
+/// </summary>
+/// <param name="Cookie">The current replication cookie, which captures the position of the replication cursor.</param>
+/// <param name="ProcessedObjectCount">The number of directory objects processed so far.</param>
+/// <param name="TotalObjectCount">The estimated total number of directory objects to be replicated, as reported by the source domain controller.</param>
+public readonly record struct ReplicationProgress(
+    ReplicationCookie Cookie,
+    int ProcessedObjectCount,
+    int TotalObjectCount);