From 63be65f2f674b9905d032d77d282de0cdd1d8897 Mon Sep 17 00:00:00 2001 From: Steven Tran Date: Mon, 20 Jul 2026 11:18:36 +1000 Subject: [PATCH 1/3] feat(AzureStorage): support orchestrator-only and activity-only workers via WorkerDispatchMode Add a WorkerDispatchMode setting (Both/Orchestrator/Activity) so a worker can run only orchestrations+entities or only activities. Activity mode skips control-queue partition leasing; Orchestrator mode skips the work-item queue. TaskHubWorker starts only the enabled dispatchers via the existing dispatcher-count capabilities. Default Both preserves current behavior. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6d5f4a2b-07b2-4c3e-a250-378aa68aa065 --- .../DecoupledWorkerDispatchTests.cs | 187 ++++++++++++++++++ .../AzureStorageOrchestrationService.cs | 48 ++++- ...zureStorageOrchestrationServiceSettings.cs | 7 + .../WorkerDispatchMode.cs | 44 +++++ src/DurableTask.Core/TaskHubWorker.cs | 73 ++++--- 5 files changed, 332 insertions(+), 27 deletions(-) create mode 100644 Test/DurableTask.AzureStorage.Tests/DecoupledWorkerDispatchTests.cs create mode 100644 src/DurableTask.AzureStorage/WorkerDispatchMode.cs diff --git a/Test/DurableTask.AzureStorage.Tests/DecoupledWorkerDispatchTests.cs b/Test/DurableTask.AzureStorage.Tests/DecoupledWorkerDispatchTests.cs new file mode 100644 index 000000000..3752eee17 --- /dev/null +++ b/Test/DurableTask.AzureStorage.Tests/DecoupledWorkerDispatchTests.cs @@ -0,0 +1,187 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- + +namespace DurableTask.AzureStorage.Tests +{ + using DurableTask.AzureStorage; + using DurableTask.Core; + using Microsoft.VisualStudio.TestTools.UnitTesting; + using System; + using System.Linq; + using System.Threading.Tasks; + + [TestClass] + public class DecoupledWorkerDispatchTests + { + const string TestConnectionString = "UseDevelopmentStorage=true"; + + // --------------------------------------------------------------------------------------- + // Unit tests: the WorkerDispatchMode setting maps onto the IOrchestrationService dispatcher + // counts, which is the signal TaskHubWorker uses to gate each dispatcher. No storage needed. + // --------------------------------------------------------------------------------------- + + [TestMethod] + public void DefaultMode_IsBoth_AndRunsBothDispatchers() + { + var service = CreateService("DispatchModeDefault", mode: null); + + Assert.AreEqual(WorkerDispatchMode.Both, new AzureStorageOrchestrationServiceSettings().WorkerDispatchMode); + Assert.AreEqual(1, service.TaskOrchestrationDispatcherCount); + Assert.AreEqual(1, service.TaskActivityDispatcherCount); + } + + [TestMethod] + public void BothMode_RunsBothDispatchers() + { + var service = CreateService("DispatchModeBoth", WorkerDispatchMode.Both); + + Assert.AreEqual(1, service.TaskOrchestrationDispatcherCount); + Assert.AreEqual(1, service.TaskActivityDispatcherCount); + } + + [TestMethod] + public void OrchestratorMode_DisablesActivityDispatcher() + { + var service = CreateService("DispatchModeOrch", WorkerDispatchMode.Orchestrator); + + Assert.AreEqual(1, service.TaskOrchestrationDispatcherCount); + Assert.AreEqual(0, service.TaskActivityDispatcherCount); + } + + [TestMethod] + public void ActivityMode_DisablesOrchestrationDispatcher() + { + var service = CreateService("DispatchModeAct", WorkerDispatchMode.Activity); + + Assert.AreEqual(0, service.TaskOrchestrationDispatcherCount); + Assert.AreEqual(1, service.TaskActivityDispatcherCount); + } + + static AzureStorageOrchestrationService CreateService(string taskHub, WorkerDispatchMode? mode) + { + var settings = new AzureStorageOrchestrationServiceSettings + { + TaskHubName = taskHub, + StorageAccountClientProvider = new StorageAccountClientProvider(TestConnectionString), + }; + + if (mode.HasValue) + { + settings.WorkerDispatchMode = mode.Value; + } + + return new AzureStorageOrchestrationService(settings); + } + + // --------------------------------------------------------------------------------------- + // Integration test (Azurite): two workers on the SAME task hub, one Orchestrator-only and + // one Activity-only. The orchestration (which calls an activity) must run to completion, and + // the roles must be decoupled: the orchestrator worker never dequeues the work-item queue and + // the activity worker never leases a control-queue partition. + // --------------------------------------------------------------------------------------- + + [TestMethod] + public async Task DecoupledWorkers_RunOrchestrationToCompletion() + { + const string taskHub = "DecoupledDispatch"; + const string input = "world"; + + var orchestratorSettings = new AzureStorageOrchestrationServiceSettings + { + TaskHubName = taskHub, + StorageAccountClientProvider = new StorageAccountClientProvider(TestConnectionString), + WorkerDispatchMode = WorkerDispatchMode.Orchestrator, + WorkerId = "orchestrator-worker", + }; + + var activitySettings = new AzureStorageOrchestrationServiceSettings + { + TaskHubName = taskHub, + StorageAccountClientProvider = new StorageAccountClientProvider(TestConnectionString), + WorkerDispatchMode = WorkerDispatchMode.Activity, + WorkerId = "activity-worker", + }; + + var orchestratorService = new AzureStorageOrchestrationService(orchestratorSettings); + var activityService = new AzureStorageOrchestrationService(activitySettings); + + // Sanity check on the mode gating before starting anything. + Assert.AreEqual(1, orchestratorService.TaskOrchestrationDispatcherCount); + Assert.AreEqual(0, orchestratorService.TaskActivityDispatcherCount); + Assert.AreEqual(0, activityService.TaskOrchestrationDispatcherCount); + Assert.AreEqual(1, activityService.TaskActivityDispatcherCount); + + // Both deployments ship the same code, so both workers register the same orchestration and + // activity. Each worker only dispatches the kind of work allowed by its mode. + var orchestratorWorker = new TaskHubWorker(orchestratorService); + orchestratorWorker.AddTaskOrchestrations(typeof(DecoupledHelloOrchestrator)); + orchestratorWorker.AddTaskActivities(typeof(DecoupledHello)); + + var activityWorker = new TaskHubWorker(activityService); + activityWorker.AddTaskOrchestrations(typeof(DecoupledHelloOrchestrator)); + activityWorker.AddTaskActivities(typeof(DecoupledHello)); + + var client = new TaskHubClient(orchestratorService); + + await orchestratorWorker.StartAsync(); + await activityWorker.StartAsync(); + + try + { + OrchestrationInstance instance = await client.CreateOrchestrationInstanceAsync( + typeof(DecoupledHelloOrchestrator), input); + + OrchestrationState state = await client.WaitForOrchestrationAsync(instance, TimeSpan.FromSeconds(90)); + + // The orchestrator-only worker cannot dequeue activities and the activity-only worker + // cannot dequeue orchestrations, so completion proves the two roles cooperated. + Assert.IsNotNull(state); + Assert.AreEqual(OrchestrationStatus.Completed, state.OrchestrationStatus); + Assert.AreEqual($"\"Hello, {input}!\"", state.Output); + + // The activity worker must never have leased a control-queue partition... + Assert.AreEqual(0, activityService.OwnedControlQueues.Count(), "Activity-only worker should not lease any control-queue partition."); + // ...while the orchestrator worker owns at least one partition. + await TestHelpers.WaitFor(() => orchestratorService.OwnedControlQueues.Any(), TimeSpan.FromSeconds(30)); + Assert.IsTrue(orchestratorService.OwnedControlQueues.Any(), "Orchestrator-only worker should lease control-queue partitions."); + } + finally + { + await orchestratorWorker.StopAsync(isForced: true); + await activityWorker.StopAsync(isForced: true); + await orchestratorService.DeleteAsync(); + } + } + + internal class DecoupledHelloOrchestrator : TaskOrchestration + { + public override Task RunTask(OrchestrationContext context, string input) + { + return context.ScheduleTask(typeof(DecoupledHello), input); + } + } + + internal class DecoupledHello : TaskActivity + { + protected override string Execute(TaskContext context, string input) + { + if (string.IsNullOrEmpty(input)) + { + throw new ArgumentNullException(nameof(input)); + } + + return $"Hello, {input}!"; + } + } + } +} diff --git a/src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs b/src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs index 74798cb45..8274f171c 100644 --- a/src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs +++ b/src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs @@ -276,11 +276,13 @@ public BehaviorOnContinueAsNew EventBehaviourForContinueAsNew } // We always leave the dispatcher counts at one unless we can find a customer workload that requires more. + // A count of zero disables the corresponding dispatcher entirely, which is how a worker is restricted + // to only orchestrations or only activities via . /// - public int TaskActivityDispatcherCount { get; } = 1; + public int TaskActivityDispatcherCount => this.settings.WorkerDispatchMode == WorkerDispatchMode.Orchestrator ? 0 : 1; /// - public int TaskOrchestrationDispatcherCount { get; } = 1; + public int TaskOrchestrationDispatcherCount => this.settings.WorkerDispatchMode == WorkerDispatchMode.Activity ? 0 : 1; #region IEntityOrchestrationService @@ -462,7 +464,12 @@ public async Task StartAsync() this.shutdownSource = new CancellationTokenSource(); this.statsLoop = Task.Run(() => this.ReportStatsLoop(this.shutdownSource.Token)); - await this.appLeaseManager.StartAsync(); + // Activity-only workers never lease a control-queue partition, so the partition manager + // (started via the app lease manager) is skipped entirely. + if (this.settings.WorkerDispatchMode != WorkerDispatchMode.Activity) + { + await this.appLeaseManager.StartAsync(); + } this.isStarted = true; } @@ -487,7 +494,12 @@ public async Task StopAsync(bool isForced) this.orchestrationSessionManager.AbortAllSessions(); } - await this.appLeaseManager.StopAsync(); + // Symmetric with StartAsync: activity-only workers never started the partition manager. + if (this.settings.WorkerDispatchMode != WorkerDispatchMode.Activity) + { + await this.appLeaseManager.StopAsync(); + } + this.isStarted = false; } @@ -690,6 +702,20 @@ static TaskHubInfo GetTaskHubInfo(string taskHub, int partitionCount) #endregion + // When a dispatcher is disabled for this worker's WorkerDispatchMode, its fetch loop (if it runs at all) + // should idle rather than spin. This waits for the normal polling interval and swallows cancellation. + async Task BackoffAsync(CancellationToken cancellationToken) + { + try + { + await Task.Delay(this.settings.MaxQueuePollingInterval, cancellationToken); + } + catch (OperationCanceledException) + { + // Shutting down; fall through to return no work item. + } + } + #region Orchestration Work Item Methods /// public Task LockNextTaskOrchestrationWorkItemAsync( @@ -705,6 +731,13 @@ public Task LockNextTaskOrchestrationWorkItemAsync( async Task LockNextTaskOrchestrationWorkItemAsync(bool entitiesOnly, CancellationToken cancellationToken) { + // Safety net: an activity-only worker must never dispatch orchestrations or entities. + if (this.settings.WorkerDispatchMode == WorkerDispatchMode.Activity) + { + await this.BackoffAsync(cancellationToken); + return null; + } + Guid traceActivityId = StartNewLogicalTraceScope(useExisting: true); await this.EnsureTaskHubAsync(); @@ -1540,6 +1573,13 @@ public async Task LockNextTaskActivityWorkItem( TimeSpan receiveTimeout, CancellationToken cancellationToken) { + // Safety net: an orchestrator-only worker must never dequeue the work-item queue. + if (this.settings.WorkerDispatchMode == WorkerDispatchMode.Orchestrator) + { + await this.BackoffAsync(cancellationToken); + return null; + } + await this.EnsureTaskHubAsync(); using (var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, this.shutdownSource.Token)) diff --git a/src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs b/src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs index e614da92f..97d6f17f0 100644 --- a/src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs +++ b/src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs @@ -295,6 +295,13 @@ internal LogHelper Logger /// public QueueClientMessageEncoding QueueClientMessageEncoding { get; set; } = QueueClientMessageEncoding.UTF8; + /// + /// Gets or sets which kinds of work this worker instance dispatches, allowing orchestrations + /// and activities to be scaled independently across separate deployments on the same task hub. + /// The default is , which preserves the historical behavior. + /// + public WorkerDispatchMode WorkerDispatchMode { get; set; } = WorkerDispatchMode.Both; + /// /// When true, an etag is used when attempting to make instance table updates upon completing an orchestration work item. /// diff --git a/src/DurableTask.AzureStorage/WorkerDispatchMode.cs b/src/DurableTask.AzureStorage/WorkerDispatchMode.cs new file mode 100644 index 000000000..d1a9d665b --- /dev/null +++ b/src/DurableTask.AzureStorage/WorkerDispatchMode.cs @@ -0,0 +1,44 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- + +namespace DurableTask.AzureStorage +{ + /// + /// Specifies which kinds of work a single worker instance dispatches, enabling orchestrations + /// and activities to be scaled independently across separate worker deployments on the same task hub. + /// + /// + /// This is a single, self-contained value so it can be surfaced from a single environment variable + /// or configuration setting without any code changes. + /// + public enum WorkerDispatchMode + { + /// + /// The worker dispatches both orchestrations (including entities) and activities. + /// This is the default and matches the historical behavior. + /// + Both = 0, + + /// + /// The worker leases control-queue partitions and dispatches only orchestrations and entities. + /// It does not dequeue the work-item queue, so it runs no activities. + /// + Orchestrator = 1, + + /// + /// The worker dispatches only activities from the work-item queue. + /// It does not lease any control-queue partition, so it runs no orchestrations or entities. + /// + Activity = 2, + } +} diff --git a/src/DurableTask.Core/TaskHubWorker.cs b/src/DurableTask.Core/TaskHubWorker.cs index 65dfbb47e..be5699d89 100644 --- a/src/DurableTask.Core/TaskHubWorker.cs +++ b/src/DurableTask.Core/TaskHubWorker.cs @@ -44,6 +44,9 @@ public sealed class TaskHubWorker : IDisposable readonly DispatchMiddlewarePipeline activityDispatchPipeline = new DispatchMiddlewarePipeline(); readonly bool dispatchEntitiesSeparately; + readonly bool dispatchEntities; + readonly bool dispatchOrchestrations; + readonly bool dispatchActivities; readonly SemaphoreSlim slimLock = new SemaphoreSlim(1, 1); readonly LogHelper logHelper; @@ -216,7 +219,16 @@ public TaskHubWorker( this.entityManager = entityObjectManager ?? throw new ArgumentException("entityObjectManager"); this.orchestrationService = orchestrationService ?? throw new ArgumentException("orchestrationService"); this.logHelper = new LogHelper(loggerFactory?.CreateLogger("DurableTask.Core")); + + // A dispatcher count of zero means the backend has disabled that kind of dispatch for this worker + // (e.g. a worker restricted to only orchestrations or only activities). Entities are dispatched + // alongside orchestrations, so they only run when orchestrations do. Note that dispatchEntitiesSeparately + // keeps its original meaning (backend capability) so that entity registration still succeeds regardless + // of mode; dispatchEntities controls whether the entity dispatcher actually runs on this worker. + this.dispatchOrchestrations = orchestrationService.TaskOrchestrationDispatcherCount > 0; + this.dispatchActivities = orchestrationService.TaskActivityDispatcherCount > 0; this.dispatchEntitiesSeparately = (orchestrationService as IEntityOrchestrationService)?.EntityBackendProperties?.UseSeparateQueueForEntityWorkItems ?? false; + this.dispatchEntities = this.dispatchEntitiesSeparately && this.dispatchOrchestrations; this.versioningSettings = versioningSettings; } @@ -296,23 +308,30 @@ public async Task StartAsync() this.logHelper.TaskHubWorkerStarting(); var sw = Stopwatch.StartNew(); - this.orchestrationDispatcher = new TaskOrchestrationDispatcher( - this.orchestrationService, - this.orchestrationManager, - this.orchestrationDispatchPipeline, - this.logHelper, - this.ErrorPropagationMode, - this.versioningSettings, - this.ExceptionPropertiesProvider); - this.activityDispatcher = new TaskActivityDispatcher( - this.orchestrationService, - this.activityManager, - this.activityDispatchPipeline, - this.logHelper, - this.ErrorPropagationMode, - this.ExceptionPropertiesProvider); - - if (this.dispatchEntitiesSeparately) + if (this.dispatchOrchestrations) + { + this.orchestrationDispatcher = new TaskOrchestrationDispatcher( + this.orchestrationService, + this.orchestrationManager, + this.orchestrationDispatchPipeline, + this.logHelper, + this.ErrorPropagationMode, + this.versioningSettings, + this.ExceptionPropertiesProvider); + } + + if (this.dispatchActivities) + { + this.activityDispatcher = new TaskActivityDispatcher( + this.orchestrationService, + this.activityManager, + this.activityDispatchPipeline, + this.logHelper, + this.ErrorPropagationMode, + this.ExceptionPropertiesProvider); + } + + if (this.dispatchEntities) { this.entityDispatcher = new TaskEntityDispatcher( this.orchestrationService, @@ -324,10 +343,18 @@ public async Task StartAsync() } await this.orchestrationService.StartAsync(); - await this.orchestrationDispatcher.StartAsync(); - await this.activityDispatcher.StartAsync(); - if (this.dispatchEntitiesSeparately) + if (this.dispatchOrchestrations) + { + await this.orchestrationDispatcher.StartAsync(); + } + + if (this.dispatchActivities) + { + await this.activityDispatcher.StartAsync(); + } + + if (this.dispatchEntities) { await this.entityDispatcher.StartAsync(); } @@ -367,9 +394,9 @@ public async Task StopAsync(bool isForced) var dispatcherShutdowns = new Task[] { - this.orchestrationDispatcher.StopAsync(isForced), - this.activityDispatcher.StopAsync(isForced), - this.dispatchEntitiesSeparately ? this.entityDispatcher.StopAsync(isForced) : Task.CompletedTask, + this.dispatchOrchestrations ? this.orchestrationDispatcher.StopAsync(isForced) : Task.CompletedTask, + this.dispatchActivities ? this.activityDispatcher.StopAsync(isForced) : Task.CompletedTask, + this.dispatchEntities ? this.entityDispatcher.StopAsync(isForced) : Task.CompletedTask, }; await Task.WhenAll(dispatcherShutdowns); From 8e6740d33044f8bc1972863c2d6239a118d07faa Mon Sep 17 00:00:00 2001 From: Steven Tran Date: Mon, 20 Jul 2026 11:53:55 +1000 Subject: [PATCH 2/3] fix(AzureStorage): address review feedback on decoupled worker dispatch - Move DecoupledWorkerDispatchTests.cs from Test/ to test/ so the solution's test project compiles and runs it in CI. - Use TestHelpers.GetTestStorageAccountConnectionString() and GetTestTaskHubName() instead of a hard-coded UseDevelopmentStorage=true connection string and a fixed task hub name, honoring existing DurableTaskTest* configuration. - TaskHubWorker: always construct the orchestration/activity dispatchers and only gate their StartAsync() calls, so the public dispatcher properties stay non-null for callers that configure them post-start (e.g. IncludeDetails); a disabled dispatcher (count 0) still starts zero fetch loops. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6d5f4a2b-07b2-4c3e-a250-378aa68aa065 --- src/DurableTask.Core/TaskHubWorker.cs | 51 +++++++++---------- .../DecoupledWorkerDispatchTests.cs | 6 +-- 2 files changed, 27 insertions(+), 30 deletions(-) rename {Test => test}/DurableTask.AzureStorage.Tests/DecoupledWorkerDispatchTests.cs (97%) diff --git a/src/DurableTask.Core/TaskHubWorker.cs b/src/DurableTask.Core/TaskHubWorker.cs index be5699d89..6f5f02a1b 100644 --- a/src/DurableTask.Core/TaskHubWorker.cs +++ b/src/DurableTask.Core/TaskHubWorker.cs @@ -308,30 +308,27 @@ public async Task StartAsync() this.logHelper.TaskHubWorkerStarting(); var sw = Stopwatch.StartNew(); - if (this.dispatchOrchestrations) - { - this.orchestrationDispatcher = new TaskOrchestrationDispatcher( - this.orchestrationService, - this.orchestrationManager, - this.orchestrationDispatchPipeline, - this.logHelper, - this.ErrorPropagationMode, - this.versioningSettings, - this.ExceptionPropertiesProvider); - } - - if (this.dispatchActivities) - { - this.activityDispatcher = new TaskActivityDispatcher( - this.orchestrationService, - this.activityManager, - this.activityDispatchPipeline, - this.logHelper, - this.ErrorPropagationMode, - this.ExceptionPropertiesProvider); - } - - if (this.dispatchEntities) + // Dispatcher objects are always constructed so the public TaskOrchestrationDispatcher/ + // TaskActivityDispatcher properties stay non-null for callers that configure them after + // StartAsync (e.g. IncludeDetails). Whether each one actually polls is gated below on + // StartAsync, and a disabled dispatcher (dispatcher count 0) starts zero fetch loops. + this.orchestrationDispatcher = new TaskOrchestrationDispatcher( + this.orchestrationService, + this.orchestrationManager, + this.orchestrationDispatchPipeline, + this.logHelper, + this.ErrorPropagationMode, + this.versioningSettings, + this.ExceptionPropertiesProvider); + this.activityDispatcher = new TaskActivityDispatcher( + this.orchestrationService, + this.activityManager, + this.activityDispatchPipeline, + this.logHelper, + this.ErrorPropagationMode, + this.ExceptionPropertiesProvider); + + if (this.dispatchEntitiesSeparately) { this.entityDispatcher = new TaskEntityDispatcher( this.orchestrationService, @@ -394,9 +391,9 @@ public async Task StopAsync(bool isForced) var dispatcherShutdowns = new Task[] { - this.dispatchOrchestrations ? this.orchestrationDispatcher.StopAsync(isForced) : Task.CompletedTask, - this.dispatchActivities ? this.activityDispatcher.StopAsync(isForced) : Task.CompletedTask, - this.dispatchEntities ? this.entityDispatcher.StopAsync(isForced) : Task.CompletedTask, + this.orchestrationDispatcher.StopAsync(isForced), + this.activityDispatcher.StopAsync(isForced), + this.dispatchEntitiesSeparately ? this.entityDispatcher.StopAsync(isForced) : Task.CompletedTask, }; await Task.WhenAll(dispatcherShutdowns); diff --git a/Test/DurableTask.AzureStorage.Tests/DecoupledWorkerDispatchTests.cs b/test/DurableTask.AzureStorage.Tests/DecoupledWorkerDispatchTests.cs similarity index 97% rename from Test/DurableTask.AzureStorage.Tests/DecoupledWorkerDispatchTests.cs rename to test/DurableTask.AzureStorage.Tests/DecoupledWorkerDispatchTests.cs index 3752eee17..c8f9d8cdd 100644 --- a/Test/DurableTask.AzureStorage.Tests/DecoupledWorkerDispatchTests.cs +++ b/test/DurableTask.AzureStorage.Tests/DecoupledWorkerDispatchTests.cs @@ -23,7 +23,7 @@ namespace DurableTask.AzureStorage.Tests [TestClass] public class DecoupledWorkerDispatchTests { - const string TestConnectionString = "UseDevelopmentStorage=true"; + static readonly string TestConnectionString = TestHelpers.GetTestStorageAccountConnectionString(); // --------------------------------------------------------------------------------------- // Unit tests: the WorkerDispatchMode setting maps onto the IOrchestrationService dispatcher @@ -71,7 +71,7 @@ static AzureStorageOrchestrationService CreateService(string taskHub, WorkerDisp { var settings = new AzureStorageOrchestrationServiceSettings { - TaskHubName = taskHub, + TaskHubName = TestHelpers.GetTestTaskHubName() + taskHub, StorageAccountClientProvider = new StorageAccountClientProvider(TestConnectionString), }; @@ -93,7 +93,7 @@ static AzureStorageOrchestrationService CreateService(string taskHub, WorkerDisp [TestMethod] public async Task DecoupledWorkers_RunOrchestrationToCompletion() { - const string taskHub = "DecoupledDispatch"; + string taskHub = TestHelpers.GetTestTaskHubName() + "Decoupled"; const string input = "world"; var orchestratorSettings = new AzureStorageOrchestrationServiceSettings From 49de02ab40ae08dc0dc6aa6d952cea87f0c6642f Mon Sep 17 00:00:00 2001 From: Steven Tran Date: Mon, 20 Jul 2026 13:16:53 +1000 Subject: [PATCH 3/3] refactor(Core): evaluate dispatcher counts in StartAsync instead of caching in ctor Addresses review feedback: TaskHubWorker previously cached TaskOrchestrationDispatcherCount/TaskActivityDispatcherCount in the constructor, which could go stale if a backend derives those counts from mutable settings changed between construction and StartAsync. The counts are now read inside StartAsync just before starting the dispatchers, so a non-zero count is always honored. Restores the constructor to its original form. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6d5f4a2b-07b2-4c3e-a250-378aa68aa065 --- src/DurableTask.Core/TaskHubWorker.cs | 26 +++++++++++--------------- 1 file changed, 11 insertions(+), 15 deletions(-) diff --git a/src/DurableTask.Core/TaskHubWorker.cs b/src/DurableTask.Core/TaskHubWorker.cs index 6f5f02a1b..ba507e54a 100644 --- a/src/DurableTask.Core/TaskHubWorker.cs +++ b/src/DurableTask.Core/TaskHubWorker.cs @@ -44,9 +44,6 @@ public sealed class TaskHubWorker : IDisposable readonly DispatchMiddlewarePipeline activityDispatchPipeline = new DispatchMiddlewarePipeline(); readonly bool dispatchEntitiesSeparately; - readonly bool dispatchEntities; - readonly bool dispatchOrchestrations; - readonly bool dispatchActivities; readonly SemaphoreSlim slimLock = new SemaphoreSlim(1, 1); readonly LogHelper logHelper; @@ -219,16 +216,7 @@ public TaskHubWorker( this.entityManager = entityObjectManager ?? throw new ArgumentException("entityObjectManager"); this.orchestrationService = orchestrationService ?? throw new ArgumentException("orchestrationService"); this.logHelper = new LogHelper(loggerFactory?.CreateLogger("DurableTask.Core")); - - // A dispatcher count of zero means the backend has disabled that kind of dispatch for this worker - // (e.g. a worker restricted to only orchestrations or only activities). Entities are dispatched - // alongside orchestrations, so they only run when orchestrations do. Note that dispatchEntitiesSeparately - // keeps its original meaning (backend capability) so that entity registration still succeeds regardless - // of mode; dispatchEntities controls whether the entity dispatcher actually runs on this worker. - this.dispatchOrchestrations = orchestrationService.TaskOrchestrationDispatcherCount > 0; - this.dispatchActivities = orchestrationService.TaskActivityDispatcherCount > 0; this.dispatchEntitiesSeparately = (orchestrationService as IEntityOrchestrationService)?.EntityBackendProperties?.UseSeparateQueueForEntityWorkItems ?? false; - this.dispatchEntities = this.dispatchEntitiesSeparately && this.dispatchOrchestrations; this.versioningSettings = versioningSettings; } @@ -308,6 +296,14 @@ public async Task StartAsync() this.logHelper.TaskHubWorkerStarting(); var sw = Stopwatch.StartNew(); + + // Read the dispatcher counts here (rather than caching them in the constructor) so a backend + // that derives them from mutable settings is honored if they changed after construction. + // A count of zero disables that dispatcher for this worker; entities ride with orchestrations. + bool dispatchOrchestrations = this.orchestrationService.TaskOrchestrationDispatcherCount > 0; + bool dispatchActivities = this.orchestrationService.TaskActivityDispatcherCount > 0; + bool dispatchEntities = this.dispatchEntitiesSeparately && dispatchOrchestrations; + // Dispatcher objects are always constructed so the public TaskOrchestrationDispatcher/ // TaskActivityDispatcher properties stay non-null for callers that configure them after // StartAsync (e.g. IncludeDetails). Whether each one actually polls is gated below on @@ -341,17 +337,17 @@ public async Task StartAsync() await this.orchestrationService.StartAsync(); - if (this.dispatchOrchestrations) + if (dispatchOrchestrations) { await this.orchestrationDispatcher.StartAsync(); } - if (this.dispatchActivities) + if (dispatchActivities) { await this.activityDispatcher.StartAsync(); } - if (this.dispatchEntities) + if (dispatchEntities) { await this.entityDispatcher.StartAsync(); }