Imported sidecar for tests

Author: torosent

Directory.Packages.props

@@ -30,7 +30,6 @@
   <ItemGroup>
     <PackageVersion Include="Microsoft.Azure.DurableTask.Core" Version="3.0.0" />
     <PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask" Version="1.2.2" />
-    <PackageVersion Include="Microsoft.DurableTask.Sidecar" Version="1.1.2" />
   </ItemGroup>
 
   <!-- Grpc / Protobuf Packages -->

src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs

@@ -11,7 +11,6 @@
 
   <ItemGroup>
     <PackageReference Include="Grpc.AspNetCore.Server" />
-    <PackageReference Include="Microsoft.DurableTask.Sidecar" />
   </ItemGroup>
 
 </Project>

test/Grpc.IntegrationTests/Grpc.IntegrationTests.csproj

@@ -0,0 +1,58 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+namespace Microsoft.DurableTask.Sidecar;
+
+class AsyncManualResetEvent
+{
+    readonly object mutex = new();
+    TaskCompletionSource tcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+    public AsyncManualResetEvent(bool isSignaled)
+    {
+        if (isSignaled)
+        {
+            this.tcs.TrySetCanceled();
+        }
+    }
+
+    public async Task<bool> WaitAsync(TimeSpan timeout, CancellationToken cancellationToken)
+    {
+        Task delayTask = Task.Delay(timeout, cancellationToken);
+        Task waitTask = this.tcs.Task;
+
+        Task winner = await Task.WhenAny(waitTask, delayTask);
+
+        // Await ensures we get a TaskCancelledException if there was a cancellation.
+        await winner;
+
+        return winner == waitTask;
+    }
+
+    public bool IsSignaled => this.tcs.Task.IsCompleted;
+
+    /// <summary>
+    /// Puts the event in the signaled state, unblocking any waiting threads.
+    /// </summary>
+    public bool Set()
+    {
+        lock (this.mutex)
+        {
+            return this.tcs.TrySetResult();
+        }
+    }
+
+    /// <summary>
+    /// Puts this event into the unsignaled state, causing threads to block.
+    /// </summary>
+    public void Reset()
+    {
+        lock (this.mutex)
+        {
+            if (this.tcs.Task.IsCompleted)
+            {
+                this.tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+            }
+        }
+    }
+}

test/Grpc.IntegrationTests/GrpcSidecar/AsyncManualResetEvent.cs

@@ -0,0 +1,35 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using DurableTask.Core;
+using DurableTask.Core.History;
+
+namespace Microsoft.DurableTask.Sidecar.Dispatcher;
+
+interface ITaskExecutor
+{
+    /// <summary>
+    /// When implemented by a concrete type, executes an orchestrator and returns the next set of orchestrator actions.
+    /// </summary>
+    /// <param name="instance">The instance ID information of the orchestrator to execute.</param>
+    /// <param name="pastEvents">The history events for previous executions of this orchestration instance.</param>
+    /// <param name="newEvents">The history events that have not yet been executed by this orchestration instance.</param>
+    /// <returns>
+    /// Returns a task containing the result of the orchestrator execution. These are effectively the side-effects of the
+    /// orchestrator code, such as calling activities, scheduling timers, etc.
+    /// </returns>
+    Task<OrchestratorExecutionResult> ExecuteOrchestrator(
+        OrchestrationInstance instance,
+        IEnumerable<HistoryEvent> pastEvents,
+        IEnumerable<HistoryEvent> newEvents);
+
+    /// <summary>
+    /// When implemented by a concreate type, executes an activity task and returns its results.
+    /// </summary>
+    /// <param name="instance">The instance ID information of the orchestration that scheduled this activity task.</param>
+    /// <param name="activityEvent">The metadata of the activity task execution, including the activity name and input.</param>
+    /// <returns>Returns a task that contains the execution result of the activity.</returns>
+    Task<ActivityExecutionResult> ExecuteActivity(
+        OrchestrationInstance instance,
+        TaskScheduledEvent activityEvent);
+}

test/Grpc.IntegrationTests/GrpcSidecar/Dispatcher/ITaskExecutor.cs

@@ -0,0 +1,29 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+namespace Microsoft.DurableTask.Sidecar.Dispatcher;
+
+/// <summary>
+/// A simple primitive that can be used to block logical threads until some condition occurs.
+/// </summary>
+interface ITrafficSignal
+{
+    /// <summary>
+    /// Provides a human-friendly reason for why the signal is in the "wait" state.
+    /// </summary>
+    string WaitReason { get; }
+
+    /// <summary>
+    /// Blocks the caller until the <see cref="Set"/> method is called.
+    /// </summary>
+    /// <param name="waitTime">The amount of time to wait until the signal is unblocked.</param>
+    /// <param name="cancellationToken">A cancellation token that can be used to interrupt a waiting caller.</param>
+    /// <returns>
+    /// Returns <c>true</c> if the traffic signal is all-clear; <c>false</c> if we timed-out waiting for the signal to clear.
+    /// </returns>
+    /// <exception cref="OperationCanceledException">
+    /// Thrown if <paramref name="cancellationToken"/> is triggered while waiting.
+    /// </exception>
+    Task<bool> WaitAsync(TimeSpan waitTime, CancellationToken cancellationToken);
+}
+

test/Grpc.IntegrationTests/GrpcSidecar/Dispatcher/ITrafficSignal.cs

@@ -0,0 +1,58 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using DurableTask.Core;
+using DurableTask.Core.History;
+using Microsoft.Extensions.Logging;
+
+namespace Microsoft.DurableTask.Sidecar.Dispatcher;
+
+class TaskActivityDispatcher : WorkItemDispatcher<TaskActivityWorkItem>
+{
+    readonly IOrchestrationService service;
+    readonly ITaskExecutor taskExecutor;
+
+    public TaskActivityDispatcher(ILogger log, ITrafficSignal trafficSignal, IOrchestrationService service, ITaskExecutor taskExecutor)
+        : base(log, trafficSignal)
+    {
+        this.service = service;
+        this.taskExecutor = taskExecutor;
+    }
+
+    public override int MaxWorkItems => this.service.MaxConcurrentTaskActivityWorkItems;
+
+    public override Task AbandonWorkItemAsync(TaskActivityWorkItem workItem) =>
+        this.service.AbandonTaskActivityWorkItemAsync(workItem);
+
+    public override Task<TaskActivityWorkItem?> FetchWorkItemAsync(TimeSpan timeout, CancellationToken cancellationToken) =>
+        this.service.LockNextTaskActivityWorkItem(timeout, cancellationToken);
+
+    protected override async Task ExecuteWorkItemAsync(TaskActivityWorkItem workItem)
+    {
+        TaskScheduledEvent scheduledEvent = (TaskScheduledEvent)workItem.TaskMessage.Event;
+
+        // TODO: Error handling for internal errors (user code exceptions are handled by the executor).
+        ActivityExecutionResult result = await this.taskExecutor.ExecuteActivity(
+            instance: workItem.TaskMessage.OrchestrationInstance,
+            activityEvent: scheduledEvent);
+
+        TaskMessage responseMessage = new()
+        {
+            Event = result.ResponseEvent,
+            OrchestrationInstance = workItem.TaskMessage.OrchestrationInstance,
+        };
+
+        await this.service.CompleteTaskActivityWorkItemAsync(workItem, responseMessage);
+    }
+
+    public override int GetDelayInSecondsOnFetchException(Exception ex) =>
+        this.service.GetDelayInSecondsAfterOnFetchException(ex);
+
+    public override string GetWorkItemId(TaskActivityWorkItem workItem) => workItem.Id;
+
+    // No-op
+    public override Task ReleaseWorkItemAsync(TaskActivityWorkItem workItem) => Task.CompletedTask;
+
+    public override Task<TaskActivityWorkItem> RenewWorkItemAsync(TaskActivityWorkItem workItem) =>
+        this.service.RenewTaskActivityWorkItemLockAsync(workItem);
+}

test/Grpc.IntegrationTests/GrpcSidecar/Dispatcher/TaskActivityDispatcher.cs

@@ -0,0 +1,51 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using DurableTask.Core;
+using Microsoft.Extensions.Logging;
+
+namespace Microsoft.DurableTask.Sidecar.Dispatcher;
+
+class TaskHubDispatcherHost
+{
+    readonly TaskOrchestrationDispatcher orchestrationDispatcher;
+    readonly TaskActivityDispatcher activityDispatcher;
+
+    readonly IOrchestrationService orchestrationService;
+    readonly ILogger log;
+
+    public TaskHubDispatcherHost(
+        ILoggerFactory loggerFactory,
+        ITrafficSignal trafficSignal,
+        IOrchestrationService orchestrationService,
+        ITaskExecutor taskExecutor)
+    {
+        this.orchestrationService = orchestrationService ?? throw new ArgumentNullException(nameof(orchestrationService));
+        this.log = loggerFactory.CreateLogger("Microsoft.DurableTask.Sidecar");
+
+        this.orchestrationDispatcher = new TaskOrchestrationDispatcher(log, trafficSignal, orchestrationService, taskExecutor);
+        this.activityDispatcher = new TaskActivityDispatcher(log, trafficSignal, orchestrationService, taskExecutor);
+    }
+
+    public async Task StartAsync(CancellationToken cancellationToken)
+    {
+        // Start any background processing in the orchestration service
+        await this.orchestrationService.StartAsync();
+
+        // Start the dispatchers, which will allow orchestrations/activities to execute
+        await Task.WhenAll(
+            this.orchestrationDispatcher.StartAsync(cancellationToken),
+            this.activityDispatcher.StartAsync(cancellationToken));
+    }
+
+    public async Task StopAsync(CancellationToken cancellationToken)
+    {
+        // Stop the dispatchers from polling the orchestration service
+        await Task.WhenAll(
+            this.orchestrationDispatcher.StopAsync(cancellationToken),
+            this.activityDispatcher.StopAsync(cancellationToken));
+
+        // Tell the storage provider to stop doing any background work.
+        await this.orchestrationService.StopAsync();
+    }
+}