Adding Timeout Property to DurableHttpRequest

src/WebJobs.Extensions.DurableTask/ContextImplementations/DurableOrchestrationContext.cs

@@ -299,7 +299,8 @@ private DurableHttpRequest CreateLocationPollRequest(DurableHttpRequest durableH
                 method: HttpMethod.Get,
                 uri: new Uri(locationUri),
                 headers: durableHttpRequest.Headers,
-                tokenSource: durableHttpRequest.TokenSource);
+                tokenSource: durableHttpRequest.TokenSource,
+                timeout: durableHttpRequest.Timeout);
 
             // Do not copy over the x-functions-key header, as in many cases, the
             // functions key used for the initial request will be a Function-level key
@@ -683,6 +684,13 @@ string IDurableOrchestrationContext.StartNewOrchestration(string functionName, o
             }
             catch (TaskFailedException e)
             {
+                // Check to see if CallHttpAsync() threw a TimeoutException
+                // In this case, we want to throw a TimeoutException instead of a FunctionFailedException
+                if (functionName.Equals(HttpOptions.HttpTaskActivityReservedName) && e.InnerException is TimeoutException)
+                {
+                    throw e.InnerException;
+                }
+
                 exception = e;
                 string message = string.Format(
                     "The {0} function '{1}' failed: \"{2}\". See the function execution logs for additional details.",

src/WebJobs.Extensions.DurableTask/DurableHttpRequest.cs

@@ -26,20 +26,23 @@ public class DurableHttpRequest
         /// <param name="content">Content added to the body of the HTTP request.</param>
         /// <param name="tokenSource">AAD authentication attached to the HTTP request.</param>
         /// <param name="asynchronousPatternEnabled">Specifies whether the DurableHttpRequest should handle the asynchronous pattern.</param>
+        /// <param name="timeout">TimeSpan used for HTTP request timeout.</param>
         public DurableHttpRequest(
             HttpMethod method,
             Uri uri,
             IDictionary<string, StringValues> headers = null,
             string content = null,
             ITokenSource tokenSource = null,
-            bool asynchronousPatternEnabled = true)
+            bool asynchronousPatternEnabled = true,
+            TimeSpan? timeout = null)
         {
             this.Method = method;
             this.Uri = uri;
             this.Headers = HttpHeadersConverter.CreateCopy(headers);
             this.Content = content;
             this.TokenSource = tokenSource;
             this.AsynchronousPatternEnabled = asynchronousPatternEnabled;
+            this.Timeout = timeout;
         }
 
         /// <summary>
@@ -82,6 +85,13 @@ public class DurableHttpRequest
         [JsonProperty("asynchronousPatternEnabled")]
         public bool AsynchronousPatternEnabled { get; }
 
+        /// <summary>
+        /// The total timeout for the original HTTP request and any
+        /// asynchronous polling.
+        /// </summary>
+        [JsonProperty("timeout")]
+        public TimeSpan? Timeout { get; }
+
         private class HttpMethodConverter : JsonConverter
         {
             public override bool CanConvert(Type objectType)

src/WebJobs.Extensions.DurableTask/Listener/TaskHttpActivityShim.cs

@@ -7,8 +7,11 @@
 using System.Net.Http;
 using System.Net.Http.Headers;
 using System.Text;
+using System.Threading;
 using System.Threading.Tasks;
 using DurableTask.Core;
+using DurableTask.Core.Common;
+using DurableTask.Core.Exceptions;
 using Microsoft.Extensions.Primitives;
 using Newtonsoft.Json;
 
@@ -35,20 +38,49 @@ public override string Run(TaskContext context, string input)
 
         public async override Task<string> RunAsync(TaskContext context, string rawInput)
         {
-            HttpRequestMessage requestMessage = await this.ReconstructHttpRequestMessage(rawInput);
-            HttpResponseMessage response = await this.httpClient.SendAsync(requestMessage);
+            DurableHttpRequest durableHttpRequest = ReconstructDurableHttpRequest(rawInput);
+            HttpRequestMessage requestMessage = await this.ConvertToHttpRequestMessage(durableHttpRequest);
+
+            HttpResponseMessage response;
+            if (durableHttpRequest.Timeout == null)
+            {
+                response = await this.httpClient.SendAsync(requestMessage);
+            }
+            else
+            {
+                try
+                {
+                    using (CancellationTokenSource cts = new CancellationTokenSource())
+                    {
+                        cts.CancelAfter(durableHttpRequest.Timeout.Value);
+                        response = await this.httpClient.SendAsync(requestMessage, cts.Token);
+                    }
+                }
+                catch (OperationCanceledException ex)
+                {
+                    TimeoutException e = new TimeoutException(ex.Message + $" Reached user specified timeout: {durableHttpRequest.Timeout.Value}.");
+
+                    string details = Utils.SerializeCause(e, this.config.ErrorDataConverter);
+                    throw new TaskFailureException(e.Message, e, details);
+                }
+            }
+
             DurableHttpResponse durableHttpResponse = await DurableHttpResponse.CreateDurableHttpResponseWithHttpResponseMessage(response);
 
             return JsonConvert.SerializeObject(durableHttpResponse);
         }
 
-        private async Task<HttpRequestMessage> ReconstructHttpRequestMessage(string serializedRequest)
+        private static DurableHttpRequest ReconstructDurableHttpRequest(string serializedRequest)
         {
             // DeserializeObject deserializes into a List and then the first element
             // of that list is the DurableHttpRequest
             IList<DurableHttpRequest> input = JsonConvert.DeserializeObject<IList<DurableHttpRequest>>(serializedRequest);
             DurableHttpRequest durableHttpRequest = input.First();
+            return durableHttpRequest;
+        }
 
+        private async Task<HttpRequestMessage> ConvertToHttpRequestMessage(DurableHttpRequest durableHttpRequest)
+        {
             string contentType = "";
             HttpRequestMessage requestMessage = new HttpRequestMessage(durableHttpRequest.Method, durableHttpRequest.Uri);
             if (durableHttpRequest.Headers != null)

test/Common/DurableHttpTests.cs

@@ -4,6 +4,7 @@
 using System;
 using System.Collections.Generic;
 using System.Diagnostics;
+using System.IO;
 using System.Linq;
 using System.Net;
 using System.Net.Http;
@@ -144,7 +145,8 @@ public void SerializeManagedIdentityOptions()
       ""tenantid"": ""tenant_id""
     }
   },
-  ""AsynchronousPatternEnabled"": true
+  ""AsynchronousPatternEnabled"": true,
+  ""Timeout"": null
 }";
 
             Dictionary<string, string> headers = new Dictionary<string, string>();
@@ -182,7 +184,8 @@ public void SerializeManagedIdentityOptions()
       ""tenantid"": ""tenant_id""
     }
    },
-  ""asynchronousPatternEnabled"": true
+  ""asynchronousPatternEnabled"": true,
+  ""timeout"": null
 }";
             ManagedIdentityTokenSource managedIdentityTokenSource = new ManagedIdentityTokenSource("dummy url", options);
             TestDurableHttpRequest testDurableHttpRequest = new TestDurableHttpRequest(
@@ -212,7 +215,8 @@ public void SerializeDurableHttpRequestWithoutManagedIdentityOptions()
     ""kind"": ""AzureManagedIdentity"",
     ""resource"": ""dummy url""
   },
-  ""asynchronousPatternEnabled"": true
+  ""asynchronousPatternEnabled"": true,
+  ""timeout"": null
 }";
 
             Dictionary<string, string> headers = new Dictionary<string, string>();
@@ -310,6 +314,86 @@ public async Task DurableHttpAsync_SynchronousAPI_Returns200(string storageProvi
             }
         }
 
+        /// <summary>
+        /// End-to-end test which checks if the CallHttpAsync Orchestrator returns an OK (200) status code
+        /// when a DurableHttpRequest timeout value is set and the request completes within the timeout.
+        /// </summary>
+        [Theory]
+        [Trait("Category", PlatformSpecificHelpers.TestCategory)]
+        [MemberData(nameof(TestDataGenerator.GetFullFeaturedStorageProviderOptions), MemberType = typeof(TestDataGenerator))]
+        public async Task DurableHttpAsync_Synchronous_TimeoutNotReached(string storageProvider)
+        {
+            HttpResponseMessage testHttpResponseMessage = CreateTestHttpResponseMessage(HttpStatusCode.OK);
+            HttpMessageHandler httpMessageHandler = MockSynchronousHttpMessageHandlerWithTimeout(testHttpResponseMessage, TimeSpan.FromMilliseconds(2000));
+
+            using (ITestHost host = TestHelpers.GetJobHost(
+                this.loggerProvider,
+                nameof(this.DurableHttpAsync_Synchronous_TimeoutNotReached),
+                enableExtendedSessions: false,
+                storageProviderType: storageProvider,
+                durableHttpMessageHandler: new DurableHttpMessageHandlerFactory(httpMessageHandler)))
+            {
+                await host.StartAsync();
+
+                Dictionary<string, string> headers = new Dictionary<string, string>();
+                headers.Add("Accept", "application/json");
+                TestDurableHttpRequest testRequest = new TestDurableHttpRequest(
+                    httpMethod: HttpMethod.Get,
+                    headers: headers,
+                    timeout: TimeSpan.FromMilliseconds(5000));
+
+                string functionName = nameof(TestOrchestrations.CallHttpAsyncOrchestrator);
+                var client = await host.StartOrchestratorAsync(functionName, testRequest, this.output);
+                var status = await client.WaitForCompletionAsync(this.output, timeout: TimeSpan.FromSeconds(400));
+
+                var output = status?.Output;
+                DurableHttpResponse response = output.ToObject<DurableHttpResponse>();
+                Assert.Equal(HttpStatusCode.OK, response.StatusCode);
+
+                await host.StopAsync();
+            }
+        }
+
+        /// <summary>
+        /// End-to-end test which checks if the CallHttpAsync Orchestrator fails  when the
+        /// HTTP request times out and the CallHttpAsync API throws a TimeoutException.
+        /// </summary>
+        [Theory]
+        [Trait("Category", PlatformSpecificHelpers.TestCategory)]
+        [MemberData(nameof(TestDataGenerator.GetFullFeaturedStorageProviderOptions), MemberType = typeof(TestDataGenerator))]
+        public async Task DurableHttpAsync_Synchronous_TimeoutException(string storageProvider)
+        {
+            HttpResponseMessage testHttpResponseMessage = CreateTestHttpResponseMessage(HttpStatusCode.OK);
+            HttpMessageHandler httpMessageHandler = MockSynchronousHttpMessageHandlerWithTimeoutException(TimeSpan.FromMilliseconds(10000));
+
+            using (ITestHost host = TestHelpers.GetJobHost(
+                this.loggerProvider,
+                nameof(this.DurableHttpAsync_Synchronous_TimeoutException),
+                enableExtendedSessions: false,
+                storageProviderType: storageProvider,
+                durableHttpMessageHandler: new DurableHttpMessageHandlerFactory(httpMessageHandler)))
+            {
+                await host.StartAsync();
+
+                Dictionary<string, string> headers = new Dictionary<string, string>();
+                headers.Add("Accept", "application/json");
+                TestDurableHttpRequest testRequest = new TestDurableHttpRequest(
+                    httpMethod: HttpMethod.Get,
+                    headers: headers,
+                    timeout: TimeSpan.FromMilliseconds(5000));
+
+                string functionName = nameof(TestOrchestrations.CallHttpAsyncOrchestrator);
+                var client = await host.StartOrchestratorAsync(functionName, testRequest, this.output);
+                var status = await client.WaitForCompletionAsync(this.output, timeout: TimeSpan.FromSeconds(400));
+
+                var output = status?.Output;
+                Assert.Contains("Orchestrator function 'CallHttpAsyncOrchestrator' failed: The operation was canceled. Reached user specified timeout: 00:00:05", output.ToString());
+                Assert.Equal(OrchestrationRuntimeStatus.Failed, status?.RuntimeStatus);
+
+                await host.StopAsync();
+            }
+        }
+
         /// <summary>
         /// End-to-end test which checks if the UserAgent header is set in the HttpResponseMessage.
         /// </summary>
@@ -1423,6 +1507,40 @@ private static HttpMessageHandler MockSynchronousHttpMessageHandler(HttpResponse
             return handlerMock.Object;
         }
 
+        private static HttpMessageHandler MockSynchronousHttpMessageHandlerWithTimeout(HttpResponseMessage httpResponseMessage, TimeSpan timeoutTimespan)
+        {
+            var handlerMock = new Mock<HttpMessageHandler>(MockBehavior.Strict);
+            handlerMock
+               .Protected()
+               .Setup<Task<HttpResponseMessage>>("SendAsync", ItExpr.IsAny<HttpRequestMessage>(), ItExpr.IsAny<CancellationToken>())
+               .Returns(async () =>
+               {
+                   await Task.Delay(timeoutTimespan);
+                   return httpResponseMessage;
+               });
+
+            return handlerMock.Object;
+        }
+
+        private static HttpMessageHandler MockSynchronousHttpMessageHandlerWithTimeoutException(TimeSpan timeoutTimespan)
+        {
+            HttpResponseMessage httpResponseMessage = CreateTestHttpResponseMessage(HttpStatusCode.OK);
+
+            httpResponseMessage.Content = new ExceptionThrowingContent(new OperationCanceledException());
+
+            var handlerMock = new Mock<HttpMessageHandler>(MockBehavior.Strict);
+            handlerMock
+               .Protected()
+               .Setup<Task<HttpResponseMessage>>("SendAsync", ItExpr.IsAny<HttpRequestMessage>(), ItExpr.IsAny<CancellationToken>())
+               .Returns(async () =>
+               {
+                   await Task.Delay(timeoutTimespan);
+                   return httpResponseMessage;
+               });
+
+            return handlerMock.Object;
+        }
+
         private static HttpMessageHandler MockHttpMessageHandlerCheckUserAgent()
         {
             HttpResponseMessage okHttpResponseMessage = CreateTestHttpResponseMessage(HttpStatusCode.OK);
@@ -1612,5 +1730,26 @@ public ManagedIdentityOptions GetOptions()
                 return this.options;
             }
         }
+
+        private class ExceptionThrowingContent : HttpContent
+        {
+            private readonly Exception exception;
+
+            public ExceptionThrowingContent(Exception exception)
+            {
+                this.exception = exception;
+            }
+
+            protected override Task SerializeToStreamAsync(Stream stream, TransportContext context)
+            {
+                return Task.FromException(this.exception);
+            }
+
+            protected override bool TryComputeLength(out long length)
+            {
+                length = 0L;
+                return false;
+            }
+        }
     }
 }