Fix synchronization of semaphore in retry delegating handler (#3135)

* Decouple client open-close semaphore from callback subscription semaphore * Cancel pending operations when CloseAsync() is invoked
Azure · Mar 7, 2023 · d297180 · d297180
1 parent 480a27b
commit d297180
Show file tree

Hide file tree

Showing 8 changed files with 318 additions and 138 deletions.
diff --git a/build.ps1 b/build.ps1
@@ -217,7 +217,7 @@ Function BuildPackage($path, $message)
         SignDotNetBinary $filesToSign
     }
 
-    & dotnet pack --verbosity $verbosity --configuration $configuration --no-build --include-symbols --include-source --output $localPackages
+    & dotnet pack --verbosity $verbosity --configuration $configuration --no-build --include-symbols --include-source --property:PackageOutputPath=$localPackages
 
     if ($LASTEXITCODE -ne 0)
     {

diff --git a/e2e/test/iothub/FileUploadE2ETests.cs b/e2e/test/iothub/FileUploadE2ETests.cs
@@ -114,6 +114,7 @@ public async Task FileUpload_SmallFile_Http_GranularSteps_x509()
         [TestMethod]
         [Timeout(TestTimeoutMilliseconds)]
         [TestCategory("LongRunning")]
+        [TestCategory("Proxy")]
         public async Task FileUpload_SmallFile_Http_GranularSteps_Proxy()
         {
             string filename = await GetTestFileNameAsync(FileSizeSmall).ConfigureAwait(false);

diff --git a/e2e/test/iothub/service/RegistryManagerE2ETests.cs b/e2e/test/iothub/service/RegistryManagerE2ETests.cs
@@ -328,6 +328,7 @@ public async Task RegistryManager_RemoveDevices2Async_Works()
 
         [TestMethod]
         [Timeout(TestTimeoutMilliseconds)]
+        [TestCategory("Proxy")]
         public async Task RegistryManager_AddDeviceWithProxy()
         {
             string deviceId = _idPrefix + Guid.NewGuid();

diff --git a/iothub/device/src/Pipeline/DefaultDelegatingHandler.cs b/iothub/device/src/Pipeline/DefaultDelegatingHandler.cs
@@ -11,8 +11,9 @@ namespace Microsoft.Azure.Devices.Client.Transport
 {
     internal abstract class DefaultDelegatingHandler : IDelegatingHandler
     {
-        private volatile IDelegatingHandler _innerHandler;
+        protected const string ClientDisposedMessage = "The client has been disposed and is no longer usable.";
         protected volatile bool _isDisposed;
+        private volatile IDelegatingHandler _innerHandler;
 
         protected DefaultDelegatingHandler(PipelineContext context, IDelegatingHandler innerHandler)
         {
@@ -209,7 +210,7 @@ protected void ThrowIfDisposed()
         {
             if (_isDisposed)
             {
-                throw new ObjectDisposedException("IoT hub client");
+                throw new ObjectDisposedException("IoT client", ClientDisposedMessage);
             }
         }
 

diff --git a/iothub/device/src/Pipeline/RetryDelegatingHandler.cs b/iothub/device/src/Pipeline/RetryDelegatingHandler.cs
diff --git a/iothub/device/tests/Pipeline/RetryDelegatingHandlerTests.cs b/iothub/device/tests/Pipeline/RetryDelegatingHandlerTests.cs
@@ -27,13 +27,12 @@ public async Task RetryDelegatingHandler_OpenAsyncRetries()
             // arrange
             int callCounter = 0;
 
-            var ct = CancellationToken.None;
             PipelineContext contextMock = Substitute.For<PipelineContext>();
             contextMock.ConnectionStatusChangesHandler = new ConnectionStatusChangesHandler(delegate (ConnectionStatus status, ConnectionStatusChangeReason reason) { });
             IDelegatingHandler innerHandlerMock = Substitute.For<IDelegatingHandler>();
 
             innerHandlerMock
-                .OpenAsync(ct)
+                .OpenAsync(Arg.Any<CancellationToken>())
                 .Returns(t =>
                     {
                         return ++callCounter == 1
@@ -45,7 +44,7 @@ public async Task RetryDelegatingHandler_OpenAsyncRetries()
             var retryDelegatingHandler = new RetryDelegatingHandler(contextMock, innerHandlerMock);
 
             // act
-            await retryDelegatingHandler.OpenAsync(ct).ConfigureAwait(false);
+            await retryDelegatingHandler.OpenAsync(CancellationToken.None).ConfigureAwait(false);
 
             // assert
             callCounter.Should().Be(2);
@@ -255,12 +254,12 @@ public async Task DeviceNotFoundExceptionReturnsDeviceDisabledStatus()
         public async Task RetryTransientErrorThrownAfterNumberOfRetriesThrows()
         {
             // arrange
-            using var cts = new CancellationTokenSource(100);
+            using var cts = new CancellationTokenSource(1000);
             var contextMock = Substitute.For<PipelineContext>();
             contextMock.ConnectionStatusChangesHandler = new ConnectionStatusChangesHandler(delegate (ConnectionStatus status, ConnectionStatusChangeReason reason) { });
             var innerHandlerMock = Substitute.For<IDelegatingHandler>();
             innerHandlerMock
-                .OpenAsync(cts.Token)
+                .OpenAsync(Arg.Any<CancellationToken>())
                 .Returns(t => throw new IotHubException(TestExceptionMessage, isTransient: true));
 
             var sut = new RetryDelegatingHandler(contextMock, innerHandlerMock);
@@ -352,12 +351,12 @@ public async Task RetrySetRetryPolicyVerifyInternalsSuccess()
                 delegate (ConnectionStatus status, ConnectionStatusChangeReason reason) { });
             var sut = new RetryDelegatingHandler(contextMock, innerHandlerMock);
 
-            var retryPolicy = new TestRetryPolicy();
+            var retryPolicy = new TestRetryPolicyRetryTwice();
             sut.SetRetryPolicy(retryPolicy);
 
             int innerHandlerCallCounter = 0;
 
-            innerHandlerMock.OpenAsync(CancellationToken.None).Returns(t =>
+            innerHandlerMock.OpenAsync(Arg.Any<CancellationToken>()).Returns(t =>
                {
                    innerHandlerCallCounter++;
                    throw new IotHubCommunicationException();
@@ -397,7 +396,7 @@ public async Task RetryCancellationTokenCanceledAbandon()
         {
             // arrange
             var innerHandlerMock = Substitute.For<IDelegatingHandler>();
-            innerHandlerMock.AbandonAsync(null, CancellationToken.None).ReturnsForAnyArgs(TaskHelpers.CompletedTask);
+            innerHandlerMock.AbandonAsync(null, Arg.Any<CancellationToken>()).ReturnsForAnyArgs(TaskHelpers.CompletedTask);
 
             var contextMock = Substitute.For<PipelineContext>();
             var sut = new RetryDelegatingHandler(contextMock, innerHandlerMock);
@@ -416,7 +415,7 @@ public async Task RetryCancellationTokenCanceledReject()
         {
             // arrange
             var innerHandlerMock = Substitute.For<IDelegatingHandler>();
-            innerHandlerMock.RejectAsync(null, CancellationToken.None).ReturnsForAnyArgs(TaskHelpers.CompletedTask);
+            innerHandlerMock.RejectAsync(null, Arg.Any<CancellationToken>()).ReturnsForAnyArgs(TaskHelpers.CompletedTask);
 
             var contextMock = Substitute.For<PipelineContext>();
             var sut = new RetryDelegatingHandler(contextMock, innerHandlerMock);
@@ -427,7 +426,7 @@ public async Task RetryCancellationTokenCanceledReject()
             await sut.RejectAsync(Arg.Any<string>(), cts.Token).ExpectedAsync<TaskCanceledException>().ConfigureAwait(false);
         }
 
-        private class TestRetryPolicy : IRetryPolicy
+        private class TestRetryPolicyRetryTwice : IRetryPolicy
         {
             public int Counter { get; private set; }
 

diff --git a/iothub/device/tests/Transport/Amqp/AmqpConnectionPoolTests.cs b/iothub/device/tests/Transport/Amqp/AmqpConnectionPoolTests.cs
@@ -1,4 +1,7 @@
-using System;
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using System;
 using System.Collections.Generic;
 using System.Linq;
 using System.Text;
@@ -13,6 +16,7 @@
 namespace Microsoft.Azure.Devices.Client.Tests.Amqp
 {
     [TestClass]
+    [TestCategory("Unit")]
     public class AmqpConnectionPoolTests
     {
         internal class AmqpConnectionPoolTest : AmqpConnectionPool

diff --git a/vsts/vsts.yaml b/vsts/vsts.yaml
@@ -548,11 +548,7 @@ jobs:
     pool:
       vmImage: windows-2022
     steps:
-      - script: |
-          rem Run dotnet first experience.
-          dotnet new
-          rem Start build
-          build.cmd -clean -build -configuration Debug -package
+      - powershell: .\build.ps1 -clean -build -configutaion Debug -package
         displayName: Build Package
 
       - task: ComponentGovernanceComponentDetection@0