Expand obsolete usage add tests

Signed-off-by: Hal Spang <[email protected]>

Author: halspang

src/Worker/Core/DependencyInjection/DurableTaskWorkerBuilderExtensions.cs

@@ -116,7 +116,6 @@ public static IDurableTaskWorkerBuilder UseVersioning(this IDurableTaskWorkerBui
     /// <param name="builder">The builder to set the builder target for.</param>
     /// <typeparam name="TOrchestrationFilter">The implementation of a <see cref="IOrchestrationFilter"/> that will be bound.</typeparam>
     /// <returns>The same <see cref="IDurableTaskWorkerBuilder"/> instance, allowing for method chaining.</returns>
-
     [Obsolete("Experimental")]
     public static IDurableTaskWorkerBuilder UseOrchestrationFilter<TOrchestrationFilter>(this IDurableTaskWorkerBuilder builder) where TOrchestrationFilter : class, IOrchestrationFilter
     {

src/Worker/Core/DurableTaskWorkerOptions.cs

@@ -148,6 +148,7 @@ public DataConverter DataConverter
     /// <summary>
     /// Gets or sets a callback function that determines whether an orchestration should be accepted for work.
     /// </summary>
+    [Obsolete("Experimental")]
     public IOrchestrationFilter? OrchestrationFilter { get; set; }
 
     /// <summary>

src/Worker/Core/IOrchestrationFilter.cs

@@ -6,6 +6,7 @@ namespace Microsoft.DurableTask.Worker;
 /// <summary>
 /// Defines a filter for validating orchestrations.
 /// </summary>
+[Obsolete("Experimental")]
 public interface IOrchestrationFilter
 {
     /// <summary>
@@ -14,21 +15,21 @@ public interface IOrchestrationFilter
     /// <param name="info">The information on the orchestration to validate.</param>
     /// <param name="cancellationToken">The cancellation token for the request to timeout.</param>
     /// <returns><code>true</code> if the orchestration is valid <code>false</code> otherwise.</returns>
-    Task<bool> IsOrchestrationValidAsync(OrchestrationInfo info, CancellationToken cancellationToken = default);
+    ValueTask<bool> IsOrchestrationValidAsync(OrchestrationFilterParameters info, CancellationToken cancellationToken = default);
 }
 
 /// <summary>
 /// Struct representation of orchestration information.
 /// </summary>
-public struct OrchestrationInfo
+public struct OrchestrationFilterParameters
 {
     /// <summary>
     /// Gets the name of the orchestration.
     /// </summary>
-    public string Name { get; init; }
+    public string? Name { get; init; }
 
     /// <summary>
     /// Gets the tags associated with the orchestration.
     /// </summary>
-    public Dictionary<string, string> Tags { get; init; }
-}
+    public IReadOnlyDictionary<string, string>? Tags { get; init; }
+}

src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs

@@ -1,7 +1,6 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
-using System.Collections.Immutable;
 using System.Text;
 using DurableTask.Core;
 using DurableTask.Core.Entities;
@@ -31,7 +30,8 @@ class Processor
         readonly TaskHubSidecarServiceClient client;
         readonly DurableTaskShimFactory shimFactory;
         readonly GrpcDurableTaskWorkerOptions.InternalOptions internalOptions;
-        readonly IOrchestrationFilter? orchestrationFilter = null;
+        [Obsolete("Experimental")]
+        readonly IOrchestrationFilter? orchestrationFilter;
 
         public Processor(GrpcDurableTaskWorker worker, TaskHubSidecarServiceClient client, IOrchestrationFilter? orchestrationFilter = null)
         {
@@ -381,10 +381,10 @@ async Task OnRunOrchestratorAsync(
                 if (this.orchestrationFilter != null)
                 {
                     filterPassed = await this.orchestrationFilter.IsOrchestrationValidAsync(
-                        new OrchestrationInfo
+                        new OrchestrationFilterParameters
                         {
                             Name = runtimeState.Name,
-                            Tags = new Dictionary<string, string>(runtimeState.Tags ?? ImmutableDictionary<string, string>.Empty),
+                            Tags = runtimeState.Tags != null ? new Dictionary<string, string>(runtimeState.Tags) : null,
                         },
                         cancellationToken);
                 }

src/Worker/Grpc/GrpcDurableTaskWorker.cs

@@ -2,7 +2,6 @@
 // Licensed under the MIT License.
 
 using Microsoft.DurableTask.Worker.Hosting;
-using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 
@@ -18,6 +17,7 @@ sealed partial class GrpcDurableTaskWorker : DurableTaskWorker
     readonly IServiceProvider services;
     readonly ILoggerFactory loggerFactory;
     readonly ILogger logger;
+    readonly IOrchestrationFilter? orchestrationFilter;
 
     /// <summary>
     /// Initializes a new instance of the <see cref="GrpcDurableTaskWorker" /> class.
@@ -28,29 +28,31 @@ sealed partial class GrpcDurableTaskWorker : DurableTaskWorker
     /// <param name="workerOptions">The generic worker options.</param>
     /// <param name="services">The service provider.</param>
     /// <param name="loggerFactory">The logger.</param>
+    /// <param name="orchestrationFilter">The optional <see cref="IOrchestrationFilter"/> used to filter orchestration execution.</param>
     public GrpcDurableTaskWorker(
         string name,
         IDurableTaskFactory factory,
         IOptionsMonitor<GrpcDurableTaskWorkerOptions> grpcOptions,
         IOptionsMonitor<DurableTaskWorkerOptions> workerOptions,
         IServiceProvider services,
-        ILoggerFactory loggerFactory)
+        ILoggerFactory loggerFactory,
+        IOrchestrationFilter? orchestrationFilter = null)
         : base(name, factory)
     {
         this.grpcOptions = Check.NotNull(grpcOptions).Get(name);
         this.workerOptions = Check.NotNull(workerOptions).Get(name);
         this.services = Check.NotNull(services);
         this.loggerFactory = Check.NotNull(loggerFactory);
         this.logger = loggerFactory.CreateLogger("Microsoft.DurableTask"); // TODO: use better category name.
+        this.orchestrationFilter = orchestrationFilter;
     }
 
     /// <inheritdoc />
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
         await using AsyncDisposable disposable = this.GetCallInvoker(out CallInvoker callInvoker, out string address);
         this.logger.StartingTaskHubWorker(address);
-        IOrchestrationFilter? filter = this.services.GetService<IOrchestrationFilter>();
-        await new Processor(this, new(callInvoker), filter).ExecuteAsync(stoppingToken);
+        await new Processor(this, new(callInvoker), this.orchestrationFilter).ExecuteAsync(stoppingToken);
     }
 
 #if NET6_0_OR_GREATER

test/Grpc.IntegrationTests/OrchestrationPatterns.cs

@@ -1009,6 +1009,41 @@ public async Task FilterOrchestrationsByName()
         await Assert.ThrowsAsync<OperationCanceledException>(async () => await server.Client.WaitForInstanceCompletionAsync(instanceId, new CancellationTokenSource(TimeSpan.FromSeconds(5)).Token));
     }
 
+    [Obsolete("Experimental")]
+    [Fact]
+    public async Task FilterOrchestrationsByNamePassesWhenNotMatching()
+    {
+        // Setup a worker with an Orchestration Filter.
+        TaskName orchestratorName = nameof(EmptyOrchestration);
+        var orchestrationFilter = new OrchestrationFilter();
+        await using HostTestLifetime server = await this.StartWorkerAsync(b =>
+        {
+            b.AddTasks(tasks => tasks.AddOrchestratorFunc(orchestratorName, ctx => Task.FromResult<object?>(null)));
+            b.UseOrchestrationFilter(orchestrationFilter);
+        });
+
+        // Nothing in the filter set, the orchestration should complete.
+        string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName);
+        OrchestrationMetadata metadata = await server.Client.WaitForInstanceCompletionAsync(
+            instanceId, this.TimeoutToken);
+
+        Assert.NotNull(metadata);
+        Assert.Equal(instanceId, metadata.InstanceId);
+        Assert.Equal(OrchestrationRuntimeStatus.Completed, metadata.RuntimeStatus);
+
+        // Update the filter and re-enqueue. The name doesn't match so the filter should be OK.
+        orchestrationFilter.NameDenySet.Add($"not-{orchestratorName}");
+
+        // This should throw as the work is denied.
+        instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName);
+        metadata = await server.Client.WaitForInstanceCompletionAsync(
+            instanceId, this.TimeoutToken);
+
+        Assert.NotNull(metadata);
+        Assert.Equal(instanceId, metadata.InstanceId);
+        Assert.Equal(OrchestrationRuntimeStatus.Completed, metadata.RuntimeStatus);
+    }
+
     [Obsolete("Experimental")]
     [Fact]
     public async Task FilterOrchestrationsByTag()
@@ -1049,16 +1084,62 @@ public async Task FilterOrchestrationsByTag()
         await Assert.ThrowsAsync<OperationCanceledException>(async () => await server.Client.WaitForInstanceCompletionAsync(instanceId, new CancellationTokenSource(TimeSpan.FromSeconds(5)).Token));
     }
 
+    [Obsolete("Experimental")]
+    [Fact]
+    public async Task FilterOrchestrationsByTagPassesWithNoMatch()
+    {
+        // Setup a worker with an Orchestration Filter.
+        TaskName orchestratorName = nameof(EmptyOrchestration);
+        IReadOnlyDictionary<string, string> orchestratorTags = new Dictionary<string, string>
+        {
+            { "test", "true" }
+        };
+        var orchestrationFilter = new OrchestrationFilter();
+        await using HostTestLifetime server = await this.StartWorkerAsync(b =>
+        {
+            b.AddTasks(tasks => tasks.AddOrchestratorFunc(orchestratorName, ctx => Task.FromResult<object?>(null)));
+            b.UseOrchestrationFilter(orchestrationFilter);
+        });
+
+        // Nothing in the filter set, the orchestration should complete.
+        string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName, new StartOrchestrationOptions
+        {
+            Tags = orchestratorTags,
+        });
+        OrchestrationMetadata metadata = await server.Client.WaitForInstanceCompletionAsync(
+            instanceId, this.TimeoutToken);
+
+        Assert.NotNull(metadata);
+        Assert.Equal(instanceId, metadata.InstanceId);
+        Assert.Equal(OrchestrationRuntimeStatus.Completed, metadata.RuntimeStatus);
+
+        // Update the filter and re-enqueue. The tags don't match so the orchestration should be OK.
+        orchestrationFilter.TagDenyDict.Add("test", "false");
+
+        // This should throw as the work is denied.
+        instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync(orchestratorName, new StartOrchestrationOptions
+        {
+            Tags = orchestratorTags,
+        });
+        metadata = await server.Client.WaitForInstanceCompletionAsync(
+            instanceId, this.TimeoutToken);
+
+        Assert.NotNull(metadata);
+        Assert.Equal(instanceId, metadata.InstanceId);
+        Assert.Equal(OrchestrationRuntimeStatus.Completed, metadata.RuntimeStatus);
+    }
+
+    [Obsolete("Experimental")]
     class OrchestrationFilter : IOrchestrationFilter
     {
         public ISet<string> NameDenySet { get; set; } = new HashSet<string>();
         public IDictionary<string, string> TagDenyDict = new Dictionary<string, string>();
 
-        public Task<bool> IsOrchestrationValidAsync(OrchestrationInfo info, CancellationToken cancellationToken = default)
+        public ValueTask<bool> IsOrchestrationValidAsync(OrchestrationFilterParameters info, CancellationToken cancellationToken = default)
         {
-            return Task.FromResult(
+            return ValueTask.FromResult(
                 !this.NameDenySet.Contains(info.Name)
-                && !this.TagDenyDict.Any(kvp => info.Tags.ContainsKey(kvp.Key) && info.Tags[kvp.Key] == kvp.Value));
+                && !this.TagDenyDict.Any(kvp => info.Tags != null && info.Tags.ContainsKey(kvp.Key) && info.Tags[kvp.Key] == kvp.Value));
         }
     }