diff --git a/src/Dapr.Common.Generators/DaprVersionAwareGenerator.cs b/src/Dapr.Common.Generators/DaprVersionAwareGenerator.cs index 23d322d24..fdeeeb89c 100644 --- a/src/Dapr.Common.Generators/DaprVersionAwareGenerator.cs +++ b/src/Dapr.Common.Generators/DaprVersionAwareGenerator.cs @@ -1,4 +1,4 @@ -// ------------------------------------------------------------------------ +// ------------------------------------------------------------------------ // Copyright 2025 The Dapr Authors // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -27,8 +27,9 @@ namespace Dapr.Common.Generators; /// The generator scans the compilation for the Dapr gRPC autogenerated client /// (Dapr.Client.Autogen.Grpc.v1.Dapr.DaprClient), groups its async unary /// methods by base name, classifies each group as PassThrough / AutoCompatible / -/// SchemaDivergent, and emits a wrapper that automatically selects the -/// highest-maturity method variant supported by the connected Dapr runtime. +/// SchemaDivergent, and emits an ordered build-time method catalog. The generated +/// wrapper combines that catalog with runtime capability discovery and safely +/// negotiates compatible variants when discovery is unavailable. /// [Generator] public sealed class DaprVersionAwareGenerator : IIncrementalGenerator diff --git a/src/Dapr.Common.Generators/Emission/WrapperCodeEmitter.cs b/src/Dapr.Common.Generators/Emission/WrapperCodeEmitter.cs index 88b8bb210..7f69dcea8 100644 --- a/src/Dapr.Common.Generators/Emission/WrapperCodeEmitter.cs +++ b/src/Dapr.Common.Generators/Emission/WrapperCodeEmitter.cs @@ -1,4 +1,4 @@ -// ------------------------------------------------------------------------ +// ------------------------------------------------------------------------ // Copyright 2025 The Dapr Authors // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -85,7 +85,7 @@ internal static string EmitClass(IReadOnlyList groups) sb.AppendLine("/// "); sb.AppendLine("/// Version-aware gRPC wrapper that automatically selects the highest-supported"); sb.AppendLine("/// Dapr runtime API variant for each operation and falls back to older variants"); - sb.AppendLine("/// when the connected runtime does not yet support the newest API."); + sb.AppendLine("/// using an ordered method catalog generated from the build-time protobuf API."); sb.AppendLine("/// "); sb.AppendLine($"internal sealed partial class {ClassName} : {InterfaceName}"); sb.AppendLine("{"); @@ -95,6 +95,16 @@ internal static string EmitClass(IReadOnlyList groups) sb.AppendLine($" private readonly global::Dapr.Common.IDaprRuntimeCapabilities {CapabilitiesFieldName};"); sb.AppendLine(); + foreach (var group in groups) + { + EmitMethodCatalog(sb, group); + if (group.Classification == MethodClassification.AutoCompatible) + { + sb.AppendLine($" private int {GetSelectedVariantFieldName(group)} = -1;"); + sb.AppendLine(); + } + } + // Constructor sb.AppendLine($" public {ClassName}("); sb.AppendLine(" global::Dapr.Client.Autogen.Grpc.v1.Dapr.DaprClient inner,"); @@ -111,6 +121,11 @@ internal static string EmitClass(IReadOnlyList groups) EmitMethod(sb, group); } + if (groups.Any(group => group.Fallbacks.Count > 0)) + { + EmitMethodUnavailableHelper(sb); + } + sb.AppendLine("}"); return sb.ToString(); } @@ -157,58 +172,55 @@ private static void EmitAutoCompatibleMethod(StringBuilder sb, MethodGroup group { var mostRecent = group.MostRecent; var (reqFqn, respFqn) = GetRequestResponseFqns(mostRecent); + var variants = GetVariants(group); + var catalogFieldName = GetCatalogFieldName(group); + var selectedVariantFieldName = GetSelectedVariantFieldName(group); sb.AppendLine($" /// "); sb.Append($" public async global::System.Threading.Tasks.Task<{respFqn}> {group.BaseName}Async("); sb.AppendLine($"{reqFqn} request, global::Grpc.Core.CallOptions options)"); sb.AppendLine(" {"); sb.AppendLine(" var __ct = options.CancellationToken;"); - sb.AppendLine(); - - // Most-recent variant: check capability then call, catching Unimplemented so that a method - // defined in the runtime's proto but not yet backed by a handler falls through to the older variant. - // Also catch Unknown with the Dapr proxy-routing error that older runtimes emit when they cannot - // match the method internally and attempt (and fail) to forward it as a service invocation. - sb.AppendLine($" if (await {CapabilitiesFieldName}.SupportsMethodAsync(\"{mostRecent.FullyQualifiedMethodName}\", __ct).ConfigureAwait(false))"); + sb.AppendLine($" switch (global::System.Threading.Volatile.Read(ref {selectedVariantFieldName}))"); sb.AppendLine(" {"); - sb.AppendLine(" try"); - sb.AppendLine(" {"); - sb.AppendLine($" return await {InnerFieldName}.{mostRecent.CSharpMethodName}(request, options).ResponseAsync.ConfigureAwait(false);"); - sb.AppendLine(" }"); - sb.AppendLine(" catch (global::Grpc.Core.RpcException __implEx) when ("); - sb.AppendLine(" __implEx.StatusCode == global::Grpc.Core.StatusCode.Unimplemented ||"); - sb.AppendLine(" (__implEx.StatusCode == global::Grpc.Core.StatusCode.Unknown &&"); - sb.AppendLine(" __implEx.Status.Detail.Contains(\"dapr-callee-app-id or dapr-app-id not found\")))"); - sb.AppendLine(" {"); - sb.AppendLine(" // Method is in the runtime proto but not yet implemented, or the runtime"); - sb.AppendLine(" // does not recognise the method and attempted to proxy it as a service invocation."); - sb.AppendLine(" // Either way, fall through to the older variant."); - sb.AppendLine(" }"); + for (var index = 0; index < variants.Count; index++) + { + sb.AppendLine($" case {index}:"); + sb.AppendLine(" {"); + EmitVariantInvocation( + sb, + mostRecent, + variants[index], + index, + " ", + selectedVariantFieldName: null); + sb.AppendLine(" }"); + } sb.AppendLine(" }"); sb.AppendLine(); - // Fallback variants - foreach (var fallback in group.Fallbacks) + for (var index = 0; index < variants.Count; index++) { - var sameRequest = SymbolEqualityComparer.Default.Equals(mostRecent.RequestType, fallback.RequestType); - var sameResponse = SymbolEqualityComparer.Default.Equals(mostRecent.ResponseType, fallback.ResponseType); - - sb.AppendLine($" if (await {CapabilitiesFieldName}.SupportsMethodAsync(\"{fallback.FullyQualifiedMethodName}\", __ct).ConfigureAwait(false))"); - - if (sameRequest && sameResponse) - { - // No type conversion needed — compact one-liner - sb.AppendLine($" return await {InnerFieldName}.{fallback.CSharpMethodName}(request, options).ResponseAsync.ConfigureAwait(false);"); - } - else - { - sb.AppendLine(" {"); - EmitRequestConversion(sb, mostRecent, fallback, " "); - sb.AppendLine($" var __fallbackResponse = await {InnerFieldName}.{fallback.CSharpMethodName}(__fallbackRequest, options).ResponseAsync.ConfigureAwait(false);"); - EmitResponseConversion(sb, mostRecent, fallback, " "); - sb.AppendLine(" }"); - } - + var variant = variants[index]; + sb.AppendLine($" var __support{index} = await {CapabilitiesFieldName}.GetMethodSupportAsync({catalogFieldName}[{index}], __ct).ConfigureAwait(false);"); + sb.AppendLine($" if (__support{index} != global::Dapr.Common.DaprRuntimeSupport.Unsupported)"); + sb.AppendLine(" {"); + sb.AppendLine(" try"); + sb.AppendLine(" {"); + EmitVariantInvocation( + sb, + mostRecent, + variant, + index, + " ", + selectedVariantFieldName); + + sb.AppendLine(" }"); + sb.AppendLine(" catch (global::Grpc.Core.RpcException __implEx) when (IsMethodUnavailable(__implEx))"); + sb.AppendLine(" {"); + sb.AppendLine(" // Continue to the next older compatible method in the generated catalog."); + sb.AppendLine(" }"); + sb.AppendLine(" }"); sb.AppendLine(); } @@ -227,6 +239,7 @@ private static void EmitSchemaDivergentMethod(StringBuilder sb, MethodGroup grou { var mostRecent = group.MostRecent; var (reqFqn, respFqn) = GetRequestResponseFqns(mostRecent); + var catalogFieldName = GetCatalogFieldName(group); sb.AppendLine($" /// "); sb.Append($" public async global::System.Threading.Tasks.Task<{respFqn}> {group.BaseName}Async("); @@ -235,36 +248,36 @@ private static void EmitSchemaDivergentMethod(StringBuilder sb, MethodGroup grou sb.AppendLine(" var __ct = options.CancellationToken;"); sb.AppendLine(); - // Most-recent variant: catch Unimplemented so a proto-defined-but-not-yet-handled method - // falls through to the schema-divergent NotSupportedException path rather than surfacing a raw RpcException. - // Also catch Unknown with the Dapr proxy-routing error that older runtimes emit when they cannot - // match the method internally and attempt (and fail) to forward it as a service invocation. - sb.AppendLine($" if (await {CapabilitiesFieldName}.SupportsMethodAsync(\"{mostRecent.FullyQualifiedMethodName}\", __ct).ConfigureAwait(false))"); + sb.AppendLine($" var __support0 = await {CapabilitiesFieldName}.GetMethodSupportAsync({catalogFieldName}[0], __ct).ConfigureAwait(false);"); + sb.AppendLine(" if (__support0 != global::Dapr.Common.DaprRuntimeSupport.Unsupported)"); sb.AppendLine(" {"); sb.AppendLine(" try"); sb.AppendLine(" {"); sb.AppendLine($" return await {InnerFieldName}.{mostRecent.CSharpMethodName}(request, options).ResponseAsync.ConfigureAwait(false);"); sb.AppendLine(" }"); - sb.AppendLine(" catch (global::Grpc.Core.RpcException __implEx) when ("); - sb.AppendLine(" __implEx.StatusCode == global::Grpc.Core.StatusCode.Unimplemented ||"); - sb.AppendLine(" (__implEx.StatusCode == global::Grpc.Core.StatusCode.Unknown &&"); - sb.AppendLine(" __implEx.Status.Detail.Contains(\"dapr-callee-app-id or dapr-app-id not found\")))"); + sb.AppendLine(" catch (global::Grpc.Core.RpcException __implEx) when (IsMethodUnavailable(__implEx))"); sb.AppendLine(" {"); - sb.AppendLine(" // Method is in the runtime proto but not yet implemented, or the runtime"); - sb.AppendLine(" // does not recognise the method and attempted to proxy it as a service invocation."); - sb.AppendLine(" // Either way, fall through to the older variant."); + sb.AppendLine(" // The newest method is unavailable; inspect the generated catalog for older variants."); sb.AppendLine(" }"); sb.AppendLine(" }"); sb.AppendLine(); // Older, incompatible variants → NotSupportedException - foreach (var fallback in group.Fallbacks) + for (var index = 0; index < group.Fallbacks.Count; index++) { - sb.AppendLine($" if (await {CapabilitiesFieldName}.SupportsMethodAsync(\"{fallback.FullyQualifiedMethodName}\", __ct).ConfigureAwait(false))"); + var fallback = group.Fallbacks[index]; + var catalogIndex = index + 1; + sb.AppendLine($" var __support{catalogIndex} = await {CapabilitiesFieldName}.GetMethodSupportAsync({catalogFieldName}[{catalogIndex}], __ct).ConfigureAwait(false);"); + sb.AppendLine($" if (__support{catalogIndex} == global::Dapr.Common.DaprRuntimeSupport.Supported)"); sb.AppendLine(" throw new global::System.NotSupportedException("); sb.AppendLine($" \"The '{group.BaseName}' operation cannot automatically fall back from '{mostRecent.GrpcMethodName}' to '{fallback.GrpcMethodName}' \" +"); sb.AppendLine($" \"because the schemas are incompatible. Provide a partial-class override of {ClassName} to handle this older runtime version.\");"); sb.AppendLine(); + sb.AppendLine($" if (__support{catalogIndex} == global::Dapr.Common.DaprRuntimeSupport.Unknown)"); + sb.AppendLine(" throw new global::System.NotSupportedException("); + sb.AppendLine($" \"The '{group.BaseName}' operation could not use '{mostRecent.GrpcMethodName}', and runtime method discovery is unavailable. \" +"); + sb.AppendLine($" \"The SDK cannot safely determine whether the schema-incompatible fallback '{fallback.GrpcMethodName}' is supported.\");"); + sb.AppendLine(); } EmitFeatureNotAvailableThrow(sb, group, " "); @@ -272,6 +285,84 @@ private static void EmitSchemaDivergentMethod(StringBuilder sb, MethodGroup grou sb.AppendLine(); } + private static void EmitMethodCatalog(StringBuilder sb, MethodGroup group) + { + sb.AppendLine($" private static readonly string[] {GetCatalogFieldName(group)} ="); + sb.AppendLine(" ["); + foreach (var variant in GetVariants(group)) + { + sb.AppendLine($" \"{variant.FullyQualifiedMethodName}\","); + } + sb.AppendLine(" ];"); + sb.AppendLine(); + } + + private static void EmitMethodUnavailableHelper(StringBuilder sb) + { + sb.AppendLine(" private static bool IsMethodUnavailable(global::Grpc.Core.RpcException exception)"); + sb.AppendLine(" {"); + sb.AppendLine(" if (exception.StatusCode == global::Grpc.Core.StatusCode.Unimplemented)"); + sb.AppendLine(" {"); + sb.AppendLine(" return true;"); + sb.AppendLine(" }"); + sb.AppendLine(); + sb.AppendLine(" if (exception.StatusCode != global::Grpc.Core.StatusCode.Unknown)"); + sb.AppendLine(" {"); + sb.AppendLine(" return false;"); + sb.AppendLine(" }"); + sb.AppendLine(); + sb.AppendLine(" return exception.Status.Detail.Contains("); + sb.AppendLine(" \"dapr-callee-app-id or dapr-app-id not found\","); + sb.AppendLine(" global::System.StringComparison.Ordinal) ||"); + sb.AppendLine(" exception.Status.Detail.Contains("); + sb.AppendLine(" \"missing dapr-callee-app-id or dapr-app-id metadata\","); + sb.AppendLine(" global::System.StringComparison.Ordinal);"); + sb.AppendLine(" }"); + sb.AppendLine(); + } + + private static void EmitVariantInvocation( + StringBuilder sb, + MethodVariant mostRecent, + MethodVariant variant, + int variantIndex, + string indent, + string? selectedVariantFieldName) + { + var sameRequest = SymbolEqualityComparer.Default.Equals(mostRecent.RequestType, variant.RequestType); + var sameResponse = SymbolEqualityComparer.Default.Equals(mostRecent.ResponseType, variant.ResponseType); + var requestVariable = "request"; + + if (!sameRequest) + { + EmitRequestConversion(sb, mostRecent, variant, indent); + requestVariable = "__fallbackRequest"; + } + + sb.AppendLine($"{indent}var __response{variantIndex} = await {InnerFieldName}.{variant.CSharpMethodName}({requestVariable}, options).ResponseAsync.ConfigureAwait(false);"); + var cacheSelectionStatement = selectedVariantFieldName is null + ? null + : $"global::System.Threading.Volatile.Write(ref {selectedVariantFieldName}, {variantIndex});"; + + if (sameResponse) + { + if (cacheSelectionStatement is not null) + { + sb.AppendLine($"{indent}{cacheSelectionStatement}"); + } + sb.AppendLine($"{indent}return __response{variantIndex};"); + return; + } + + EmitResponseConversion( + sb, + mostRecent, + variant, + indent, + $"__response{variantIndex}", + cacheSelectionStatement); + } + // ------------------------------------------------------------------------- // Type-mapping helpers // ------------------------------------------------------------------------- @@ -313,11 +404,17 @@ private static void EmitResponseConversion( StringBuilder sb, MethodVariant mostRecent, MethodVariant fallback, - string indent) + string indent, + string fallbackResponseVariable = "__fallbackResponse", + string? beforeReturnStatement = null) { if (SymbolEqualityComparer.Default.Equals(mostRecent.ResponseType, fallback.ResponseType)) { - sb.AppendLine($"{indent}return __fallbackResponse;"); + if (beforeReturnStatement is not null) + { + sb.AppendLine($"{indent}{beforeReturnStatement}"); + } + sb.AppendLine($"{indent}return {fallbackResponseVariable};"); return; } @@ -329,9 +426,13 @@ private static void EmitResponseConversion( // the newer type may have additional fields which remain at their defaults. foreach (var prop in DaprClientAnalyzer.GetUserInstanceProperties(fallback.ResponseType)) { - EmitPropertyCopy(sb, prop, "__fallbackResponse", "__convertedResponse", indent); + EmitPropertyCopy(sb, prop, fallbackResponseVariable, "__convertedResponse", indent); } + if (beforeReturnStatement is not null) + { + sb.AppendLine($"{indent}{beforeReturnStatement}"); + } sb.AppendLine($"{indent}return __convertedResponse;"); } @@ -372,6 +473,15 @@ private static void EmitPropertyCopy( // Shared helpers // ------------------------------------------------------------------------- + private static IReadOnlyList GetVariants(MethodGroup group) => + new[] { group.MostRecent }.Concat(group.Fallbacks).ToList(); + + private static string GetCatalogFieldName(MethodGroup group) => + $"_{char.ToLowerInvariant(group.BaseName[0])}{group.BaseName.Substring(1)}MethodCatalog"; + + private static string GetSelectedVariantFieldName(MethodGroup group) => + $"_{char.ToLowerInvariant(group.BaseName[0])}{group.BaseName.Substring(1)}SelectedVariant"; + private static void EmitFeatureNotAvailableThrow(StringBuilder sb, MethodGroup group, string indent) { var allVariants = new[] { group.MostRecent }.Concat(group.Fallbacks); diff --git a/src/Dapr.Common/DaprRuntimeCapabilities.cs b/src/Dapr.Common/DaprRuntimeCapabilities.cs index 74141e22f..3132090c7 100644 --- a/src/Dapr.Common/DaprRuntimeCapabilities.cs +++ b/src/Dapr.Common/DaprRuntimeCapabilities.cs @@ -1,4 +1,5 @@ -using Google.Protobuf.Reflection; +using System.Collections.Concurrent; +using Google.Protobuf.Reflection; using Grpc.Core; using Grpc.Net.Client; using Grpc.Reflection.V1; @@ -8,18 +9,46 @@ namespace Dapr.Common; /// /// Used to determine Dapr runtime capability for fallback purposes by the SDKs. /// -/// The to validate with. -internal sealed class DaprRuntimeCapabilities(GrpcChannel channel) : IDaprRuntimeCapabilities, IDisposable +internal sealed class DaprRuntimeCapabilities : IDaprRuntimeCapabilities { - private readonly ServerReflection.ServerReflectionClient _reflectionClient = new(channel); - private readonly SemaphoreSlim _gate = new(1, 1); - private HashSet? _cachedServices = null; - private readonly Dictionary> _cachedMethodsByService = []; + private static readonly TimeSpan DefaultReflectionTimeout = TimeSpan.FromSeconds(5); + + private readonly ServerReflection.ServerReflectionClient _reflectionClient; + private readonly TimeSpan _reflectionTimeout; + private readonly Lazy?>> _servicesLookup; + private readonly ConcurrentDictionary?>>> _methodLookups = + new(StringComparer.Ordinal); + private int _reflectionUnavailable; public const string Namespace = "dapr.proto.runtime.v1.Dapr"; - + + /// + /// Creates a runtime capability reader for the provided channel. + /// + /// The to validate with. + public DaprRuntimeCapabilities(GrpcChannel channel) + : this(new ServerReflection.ServerReflectionClient(channel), DefaultReflectionTimeout) + { + } + + internal DaprRuntimeCapabilities( + ServerReflection.ServerReflectionClient reflectionClient, + TimeSpan reflectionTimeout) + { + ArgumentNullException.ThrowIfNull(reflectionClient); + ArgumentOutOfRangeException.ThrowIfLessThanOrEqual(reflectionTimeout, TimeSpan.Zero); + + this._reflectionClient = reflectionClient; + this._reflectionTimeout = reflectionTimeout; + this._servicesLookup = new Lazy?>>( + QueryServicesWithFallbackAsync, + LazyThreadSafetyMode.ExecutionAndPublication); + } + /// - public async Task SupportsMethodAsync(string fullyQualifiedMethodName, CancellationToken cancellationToken = default) + public async Task GetMethodSupportAsync( + string fullyQualifiedMethodName, + CancellationToken cancellationToken = default) { var slash = fullyQualifiedMethodName.LastIndexOf('/'); if (slash <= 0) @@ -30,46 +59,59 @@ public async Task SupportsMethodAsync(string fullyQualifiedMethodName, Can var service = fullyQualifiedMethodName[..slash]; var method = fullyQualifiedMethodName[(slash + 1)..]; - try - { - var methods = await GetMethodsForServiceAsync(service, cancellationToken).ConfigureAwait(false); - return methods.Contains(method); - } - catch (RpcException) - { - // gRPC reflection is unavailable or returned an error for this service. - // Return true (optimistic) so the version-aware caller will attempt the method - // and handle StatusCode.Unimplemented as the runtime-version fallback signal. - return true; - } - catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) - { - // The HTTP/2 stream underlying the reflection call was internally cancelled - // (e.g. connection reset, keep-alive expiry, or stream setup race on .NET 9+). - // This is NOT a caller cancellation — treat it the same as an unavailable - // reflection service and proceed optimistically. - return true; - } + var methods = await GetMethodsForServiceAsync(service, cancellationToken).ConfigureAwait(false); + return methods is null + ? DaprRuntimeSupport.Unknown + : methods.Contains(method) + ? DaprRuntimeSupport.Supported + : DaprRuntimeSupport.Unsupported; } /// - public async Task SupportsServiceAsync(string serviceName, CancellationToken cancellationToken = default) + public async Task GetServiceSupportAsync( + string serviceName, + CancellationToken cancellationToken = default) { var services = await GetServicesAsync(cancellationToken).ConfigureAwait(false); - return services.Contains(serviceName); + return services is null + ? DaprRuntimeSupport.Unknown + : services.Contains(serviceName) + ? DaprRuntimeSupport.Supported + : DaprRuntimeSupport.Unsupported; } - private async Task> GetServicesAsync(CancellationToken cancellationToken) + private Task?> GetServicesAsync(CancellationToken cancellationToken) { - if (_cachedServices is not null) - return _cachedServices; + if (Volatile.Read(ref _reflectionUnavailable) != 0) + { + return Task.FromResult?>(null); + } - await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); - try + return _servicesLookup.Value.WaitAsync(cancellationToken); + } + + private Task?> GetMethodsForServiceAsync(string serviceName, + CancellationToken cancellationToken) + { + if (Volatile.Read(ref _reflectionUnavailable) != 0) { - if (_cachedServices is not null) - return _cachedServices; + return Task.FromResult?>(null); + } + + var lookup = _methodLookups.GetOrAdd( + serviceName, + static (name, capabilities) => new Lazy?>>( + () => capabilities.QueryMethodsWithFallbackAsync(name), + LazyThreadSafetyMode.ExecutionAndPublication), + this); + + return lookup.Value.WaitAsync(cancellationToken); + } + private Task?> QueryServicesWithFallbackAsync() + { + return ExecuteReflectionQueryAsync(async cancellationToken => + { using var call = _reflectionClient.ServerReflectionInfo(cancellationToken: cancellationToken); await call.RequestStream.WriteAsync(new ServerReflectionRequest { ListServices = "" }, cancellationToken) .ConfigureAwait(false); @@ -81,36 +123,25 @@ await call.RequestStream.WriteAsync(new ServerReflectionRequest { ListServices = if (response.MessageResponseCase == ServerReflectionResponse.MessageResponseOneofCase.ListServicesResponse) { - foreach (var s in response.ListServicesResponse.Service) + foreach (var service in response.ListServicesResponse.Service) { - set.Add(s.Name); + set.Add(service.Name); } } } - _cachedServices = set; return set; - } - finally - { - _gate.Release(); - } + }); } - private async Task> GetMethodsForServiceAsync(string serviceName, - CancellationToken cancellationToken) + private Task?> QueryMethodsWithFallbackAsync(string serviceName) { - if (_cachedMethodsByService.TryGetValue(serviceName, out var existing)) - return existing; - - await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); - try + return ExecuteReflectionQueryAsync(async cancellationToken => { - if (_cachedMethodsByService.TryGetValue(serviceName, out existing)) - return existing; - using var call = _reflectionClient.ServerReflectionInfo(cancellationToken: cancellationToken); - await call.RequestStream.WriteAsync(new ServerReflectionRequest { FileContainingSymbol = serviceName }, cancellationToken) + await call.RequestStream.WriteAsync( + new ServerReflectionRequest { FileContainingSymbol = serviceName }, + cancellationToken) .ConfigureAwait(false); await call.RequestStream.CompleteAsync().ConfigureAwait(false); @@ -119,7 +150,9 @@ await call.RequestStream.WriteAsync(new ServerReflectionRequest { FileContaining { if (response.MessageResponseCase != ServerReflectionResponse.MessageResponseOneofCase.FileDescriptorResponse) + { continue; + } foreach (var raw in response.FileDescriptorResponse.FileDescriptorProto) { @@ -128,7 +161,10 @@ await call.RequestStream.WriteAsync(new ServerReflectionRequest { FileContaining { var fqn = string.IsNullOrEmpty(fd.Package) ? svc.Name : $"{fd.Package}.{svc.Name}"; if (fqn != serviceName) + { continue; + } + foreach (var m in svc.Method) { set.Add(m.Name); @@ -137,14 +173,45 @@ await call.RequestStream.WriteAsync(new ServerReflectionRequest { FileContaining } } - _cachedMethodsByService[serviceName] = set; return set; + }); + } + + private async Task?> ExecuteReflectionQueryAsync( + Func>> query) + { + using var timeoutSource = new CancellationTokenSource(_reflectionTimeout); + var queryTask = query(timeoutSource.Token); + + try + { + return await queryTask.WaitAsync(_reflectionTimeout, timeoutSource.Token).ConfigureAwait(false); } - finally + catch (RpcException) { - _gate.Release(); + Volatile.Write(ref _reflectionUnavailable, 1); + return null; + } + catch (OperationCanceledException) + { + Volatile.Write(ref _reflectionUnavailable, 1); + return null; + } + catch (TimeoutException) + { + await timeoutSource.CancelAsync(); + ObserveFault(queryTask); + Volatile.Write(ref _reflectionUnavailable, 1); + return null; } } - public void Dispose() => _gate.Dispose(); + private static void ObserveFault(Task task) + { + _ = task.ContinueWith( + static completedTask => _ = completedTask.Exception, + CancellationToken.None, + TaskContinuationOptions.ExecuteSynchronously | TaskContinuationOptions.OnlyOnFaulted, + TaskScheduler.Default); + } } diff --git a/src/Dapr.Common/DaprRuntimeSupport.cs b/src/Dapr.Common/DaprRuntimeSupport.cs new file mode 100644 index 000000000..5796d1c9a --- /dev/null +++ b/src/Dapr.Common/DaprRuntimeSupport.cs @@ -0,0 +1,37 @@ +// ------------------------------------------------------------------------ +// Copyright 2026 The Dapr Authors +// 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 Dapr.Common; + +/// +/// Describes whether the connected Dapr runtime exposes a gRPC capability. +/// +internal enum DaprRuntimeSupport +{ + /// + /// Runtime discovery could not determine whether the capability is available. + /// + Unknown, + + /// + /// Runtime discovery confirmed that the capability is unavailable. + /// + Unsupported, + + /// + /// Runtime discovery confirmed that the capability is available. + /// + Supported, +} diff --git a/src/Dapr.Common/IDaprRuntimeCapabilities.cs b/src/Dapr.Common/IDaprRuntimeCapabilities.cs index dc5362eaf..cebab4ee8 100644 --- a/src/Dapr.Common/IDaprRuntimeCapabilities.cs +++ b/src/Dapr.Common/IDaprRuntimeCapabilities.cs @@ -21,18 +21,18 @@ namespace Dapr.Common; internal interface IDaprRuntimeCapabilities { /// - /// Validates whether the connected Dapr runtime supports the specified fully-qualified gRPC method. + /// Gets whether the connected Dapr runtime supports the specified fully-qualified gRPC method. /// /// The name of the fully-qualified gRPC method. /// Cancellation token. - /// Returns true if the runtime exposes the given fully-qualified gRPC method (e.g. dapr.proto.runtime.v1.Dapr/ScheduleJob - Task SupportsMethodAsync(string fullyQualifiedMethodName, CancellationToken cancellationToken = default); - + /// The runtime support state for the method. + Task GetMethodSupportAsync(string fullyQualifiedMethodName, CancellationToken cancellationToken = default); + /// - /// Validates whether the connected Dapr runtime supports the specified service. + /// Gets whether the connected Dapr runtime supports the specified service. /// /// The name of the service to validate. /// Cancellation token. - /// Returns true if the runtime exposes the given service (e.g. dapr.proto.runtime.v1.Dapr - Task SupportsServiceAsync(string serviceName, CancellationToken cancellationToken = default); + /// The runtime support state for the service. + Task GetServiceSupportAsync(string serviceName, CancellationToken cancellationToken = default); } diff --git a/test/Dapr.AspNetCore.IntegrationTest/StateTestClient.cs b/test/Dapr.AspNetCore.IntegrationTest/StateTestClient.cs index c4475b309..722579fdf 100644 --- a/test/Dapr.AspNetCore.IntegrationTest/StateTestClient.cs +++ b/test/Dapr.AspNetCore.IntegrationTest/StateTestClient.cs @@ -37,11 +37,11 @@ internal StateTestClient() private sealed class AlwaysAvailableCapabilities : IDaprRuntimeCapabilities { - public Task SupportsMethodAsync(string fullyQualifiedMethodName, CancellationToken cancellationToken = default) - => Task.FromResult(true); + public Task GetMethodSupportAsync(string fullyQualifiedMethodName, CancellationToken cancellationToken = default) + => Task.FromResult(DaprRuntimeSupport.Supported); - public Task SupportsServiceAsync(string serviceName, CancellationToken cancellationToken = default) - => Task.FromResult(true); + public Task GetServiceSupportAsync(string serviceName, CancellationToken cancellationToken = default) + => Task.FromResult(DaprRuntimeSupport.Supported); } public override Task GetStateAsync(string storeName, string key, ConsistencyMode? consistencyMode = default, IReadOnlyDictionary metadata = default, CancellationToken cancellationToken = default) diff --git a/test/Dapr.Client.Test/BulkPublishEventApiTest.cs b/test/Dapr.Client.Test/BulkPublishEventApiTest.cs index 746138456..929a11887 100644 --- a/test/Dapr.Client.Test/BulkPublishEventApiTest.cs +++ b/test/Dapr.Client.Test/BulkPublishEventApiTest.cs @@ -15,13 +15,17 @@ namespace Dapr.Client.Test; using System; using System.Collections.Generic; +using System.Net.Http; using System.Text.Json; using System.Threading; using System.Threading.Tasks; using Autogenerated = Dapr.Client.Autogen.Grpc.v1; -using Shouldly; +using Dapr.Common; using Grpc.Core; +using Grpc.Net.Client; +using Grpc.Reflection.V1; using Moq; +using Shouldly; using Xunit; public class BulkPublishEventApiTest @@ -285,7 +289,93 @@ await client.InnerClient.BulkPublishEventAsync(TestPubsubName, TestTopicName, bu null, cancellationToken: cts.Token); }); } - + + [Fact] + public async Task BulkPublishEventAsync_WhenReflectionNeverCompletes_ReachesBulkPublishRpc() + { + var testCancellationToken = TestContext.Current.CancellationToken; + var (capabilities, getReflectionCallCount) = CreateHangingReflectionCapabilities(); + var daprGrpcClient = new Mock(MockBehavior.Strict); + daprGrpcClient + .Setup(client => client.BulkPublishEventAsync( + It.IsAny(), + It.IsAny())) + .Returns(CreateSuccessfulBulkPublishCall()); + + using var channel = GrpcChannel.ForAddress("http://localhost"); + using var httpClient = new HttpClient(); + using var daprClient = new DaprClientGrpc( + channel, + daprGrpcClient.Object, + httpClient, + new Uri("http://localhost:3500"), + new JsonSerializerOptions(), + default, + capabilities); + + var response = await daprClient + .BulkPublishEventAsync( + TestPubsubName, + TestTopicName, + bulkPublishData, + cancellationToken: testCancellationToken) + .WaitAsync(TimeSpan.FromSeconds(2), testCancellationToken); + + response.FailedEntries.ShouldBeEmpty(); + getReflectionCallCount().ShouldBe(1); + daprGrpcClient.Verify(client => client.BulkPublishEventAsync( + It.IsAny(), + It.IsAny()), Times.Once); + } + +#pragma warning disable CS0612 // The generated selector intentionally invokes the obsolete runtime fallback. + [Fact] + public async Task BulkPublishEventAsync_WhenReflectionNeverCompletesAndStableMethodIsUnavailable_UsesGeneratedFallback() + { + var testCancellationToken = TestContext.Current.CancellationToken; + var (capabilities, getReflectionCallCount) = CreateHangingReflectionCapabilities(); + var daprGrpcClient = new Mock(MockBehavior.Strict); + daprGrpcClient + .Setup(client => client.BulkPublishEventAsync( + It.IsAny(), + It.IsAny())) + .Throws(new RpcException(new Status(StatusCode.Unimplemented, "Method unavailable."))); + daprGrpcClient + .Setup(client => client.BulkPublishEventAlpha1Async( + It.IsAny(), + It.IsAny())) + .Returns(CreateSuccessfulBulkPublishCall()); + + using var channel = GrpcChannel.ForAddress("http://localhost"); + using var httpClient = new HttpClient(); + using var daprClient = new DaprClientGrpc( + channel, + daprGrpcClient.Object, + httpClient, + new Uri("http://localhost:3500"), + new JsonSerializerOptions(), + default, + capabilities); + + var response = await daprClient + .BulkPublishEventAsync( + TestPubsubName, + TestTopicName, + bulkPublishData, + cancellationToken: testCancellationToken) + .WaitAsync(TimeSpan.FromSeconds(2), testCancellationToken); + + response.FailedEntries.ShouldBeEmpty(); + getReflectionCallCount().ShouldBe(1); + daprGrpcClient.Verify(client => client.BulkPublishEventAsync( + It.IsAny(), + It.IsAny()), Times.Once); + daprGrpcClient.Verify(client => client.BulkPublishEventAlpha1Async( + It.IsAny(), + It.IsAny()), Times.Once); + } +#pragma warning restore CS0612 + [Fact] public async Task BulkPublishEventAsync_WrapsRpcException() { @@ -397,6 +487,57 @@ public async Task BulkPublishEventAsync_CanPublishTopicWithByteArrayEvents() bulkPublishResponse.FailedEntries.Count.ShouldBe(0); } + private static (DaprRuntimeCapabilities capabilities, Func getCallCount) + CreateHangingReflectionCapabilities() + { + var requestStream = new Mock>(); + requestStream + .Setup(stream => stream.WriteAsync( + It.IsAny(), + It.IsAny())) + .Returns(Task.CompletedTask); + requestStream + .Setup(stream => stream.CompleteAsync()) + .Returns(Task.CompletedTask); + + var responseStream = new Mock>(); + var neverCompletes = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + responseStream + .Setup(stream => stream.MoveNext(It.IsAny())) + .Returns(neverCompletes.Task); + + var reflectionCallCount = 0; + var callInvoker = new Mock(); + callInvoker + .Setup(invoker => invoker.AsyncDuplexStreamingCall( + It.IsAny>(), + It.IsAny(), + It.IsAny())) + .Callback(() => Interlocked.Increment(ref reflectionCallCount)) + .Returns(new AsyncDuplexStreamingCall( + requestStream.Object, + responseStream.Object, + Task.FromResult(new Metadata()), + () => Status.DefaultSuccess, + () => new Metadata(), + () => { })); + + var capabilities = new DaprRuntimeCapabilities( + new ServerReflection.ServerReflectionClient(callInvoker.Object), + TimeSpan.FromMilliseconds(100)); + return (capabilities, () => Volatile.Read(ref reflectionCallCount)); + } + + private static AsyncUnaryCall CreateSuccessfulBulkPublishCall() + { + return new AsyncUnaryCall( + Task.FromResult(new Autogenerated.BulkPublishResponse()), + Task.FromResult(new Metadata()), + () => Status.DefaultSuccess, + () => new Metadata(), + () => { }); + } + private class Widget { public string Size { get; set; } diff --git a/test/Dapr.Client.Test/MockClient.cs b/test/Dapr.Client.Test/MockClient.cs index 249d88bee..3897c7f9b 100644 --- a/test/Dapr.Client.Test/MockClient.cs +++ b/test/Dapr.Client.Test/MockClient.cs @@ -35,11 +35,11 @@ public MockClient() // are not required to handle the gRPC reflection round-trip. private sealed class AlwaysAvailableCapabilities : IDaprRuntimeCapabilities { - public Task SupportsMethodAsync(string fullyQualifiedMethodName, CancellationToken cancellationToken = default) - => Task.FromResult(true); + public Task GetMethodSupportAsync(string fullyQualifiedMethodName, CancellationToken cancellationToken = default) + => Task.FromResult(DaprRuntimeSupport.Supported); - public Task SupportsServiceAsync(string serviceName, CancellationToken cancellationToken = default) - => Task.FromResult(true); + public Task GetServiceSupportAsync(string serviceName, CancellationToken cancellationToken = default) + => Task.FromResult(DaprRuntimeSupport.Supported); } public Mock Mock { get; } diff --git a/test/Dapr.Client.Test/VersionAwareDaprClientFallbackTests.cs b/test/Dapr.Client.Test/VersionAwareDaprClientFallbackTests.cs new file mode 100644 index 000000000..c19cc4e04 --- /dev/null +++ b/test/Dapr.Client.Test/VersionAwareDaprClientFallbackTests.cs @@ -0,0 +1,240 @@ +// ------------------------------------------------------------------------ +// Copyright 2026 The Dapr Authors +// 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. +// ------------------------------------------------------------------------ + +using System; +using System.Threading; +using System.Threading.Tasks; +using Dapr.Common; +using Grpc.Core; +using Moq; +using Xunit; +using Autogenerated = Dapr.Client.Autogen.Grpc.v1; + +namespace Dapr.Client.Test; + +public sealed class VersionAwareDaprClientFallbackTests +{ + [Fact] + public async Task StartWorkflowAsync_WhenDiscoveryIsUnknownAndBetaIsUnavailable_FallsBackToAlpha() + { + var cancellationToken = TestContext.Current.CancellationToken; + var capabilities = CreateCapabilities(DaprRuntimeSupport.Unknown); + var inner = new Mock(MockBehavior.Strict); + + inner + .Setup(client => client.StartWorkflowBeta1Async( + It.IsAny(), + It.IsAny())) + .Throws(new RpcException(new Status(StatusCode.Unimplemented, "Beta method unavailable."))); + +#pragma warning disable CS0612 // The generated selector intentionally invokes the obsolete runtime fallback. + inner + .Setup(client => client.StartWorkflowAlpha1Async( + It.IsAny(), + It.IsAny())) + .Returns(CreateUnaryCall()); +#pragma warning restore CS0612 + + var client = new VersionAwareDaprClient(inner.Object, capabilities.Object); + + var firstResponse = await client.StartWorkflowAsync( + new Autogenerated.StartWorkflowRequest(), + new CallOptions(cancellationToken: cancellationToken)); + var secondResponse = await client.StartWorkflowAsync( + new Autogenerated.StartWorkflowRequest(), + new CallOptions(cancellationToken: cancellationToken)); + + Assert.NotNull(firstResponse); + Assert.NotNull(secondResponse); + capabilities.Verify(capability => capability.GetMethodSupportAsync( + "dapr.proto.runtime.v1.Dapr/StartWorkflowBeta1", + cancellationToken), Times.Once); + capabilities.Verify(capability => capability.GetMethodSupportAsync( + "dapr.proto.runtime.v1.Dapr/StartWorkflowAlpha1", + cancellationToken), Times.Once); + inner.Verify(client => client.StartWorkflowBeta1Async( + It.IsAny(), + It.IsAny()), Times.Once); +#pragma warning disable CS0612 // Verify that the cached obsolete fallback is invoked directly. + inner.Verify(client => client.StartWorkflowAlpha1Async( + It.IsAny(), + It.IsAny()), Times.Exactly(2)); +#pragma warning restore CS0612 + } + + [Fact] + public async Task StartWorkflowAsync_WhenCatalogConfirmsBetaUnsupported_CallsAlphaDirectly() + { + var cancellationToken = TestContext.Current.CancellationToken; + var capabilities = new Mock(MockBehavior.Strict); + capabilities + .Setup(capability => capability.GetMethodSupportAsync( + "dapr.proto.runtime.v1.Dapr/StartWorkflowBeta1", + cancellationToken)) + .ReturnsAsync(DaprRuntimeSupport.Unsupported); + capabilities + .Setup(capability => capability.GetMethodSupportAsync( + "dapr.proto.runtime.v1.Dapr/StartWorkflowAlpha1", + cancellationToken)) + .ReturnsAsync(DaprRuntimeSupport.Supported); + + var inner = new Mock(MockBehavior.Strict); +#pragma warning disable CS0612 // The generated selector intentionally invokes the obsolete runtime fallback. + inner + .Setup(client => client.StartWorkflowAlpha1Async( + It.IsAny(), + It.IsAny())) + .Returns(CreateUnaryCall()); +#pragma warning restore CS0612 + + var client = new VersionAwareDaprClient(inner.Object, capabilities.Object); + + var response = await client.StartWorkflowAsync( + new Autogenerated.StartWorkflowRequest(), + new CallOptions(cancellationToken: cancellationToken)); + + Assert.NotNull(response); + inner.Verify(client => client.StartWorkflowBeta1Async( + It.IsAny(), + It.IsAny()), Times.Never); + } + + [Fact] + public async Task StartWorkflowAsync_WhenBetaReturnsDaprProxyError_FallsBackToAlpha() + { + var cancellationToken = TestContext.Current.CancellationToken; + var capabilities = CreateCapabilities(DaprRuntimeSupport.Unknown); + var inner = new Mock(MockBehavior.Strict); + inner + .Setup(client => client.StartWorkflowBeta1Async( + It.IsAny(), + It.IsAny())) + .Throws(new RpcException(new Status( + StatusCode.Unknown, + "cannot proxy request: missing dapr-callee-app-id or dapr-app-id metadata"))); + +#pragma warning disable CS0612 // The generated selector intentionally invokes the obsolete runtime fallback. + inner + .Setup(client => client.StartWorkflowAlpha1Async( + It.IsAny(), + It.IsAny())) + .Returns(CreateUnaryCall()); +#pragma warning restore CS0612 + + var client = new VersionAwareDaprClient(inner.Object, capabilities.Object); + + var response = await client.StartWorkflowAsync( + new Autogenerated.StartWorkflowRequest(), + new CallOptions(cancellationToken: cancellationToken)); + + Assert.NotNull(response); + } + + [Fact] + public async Task ScheduleJobAsync_WhenDiscoveryIsUnknownAndStableIsUnavailable_FallsBackToAlpha() + { + var cancellationToken = TestContext.Current.CancellationToken; + var capabilities = CreateCapabilities(DaprRuntimeSupport.Unknown); + var inner = new Mock(MockBehavior.Strict); + inner + .Setup(client => client.ScheduleJobAsync( + It.IsAny(), + It.IsAny())) + .Throws(new RpcException(new Status(StatusCode.Unimplemented, "Stable method unavailable."))); + +#pragma warning disable CS0612 // The generated selector intentionally invokes the obsolete runtime fallback. + inner + .Setup(client => client.ScheduleJobAlpha1Async( + It.IsAny(), + It.IsAny())) + .Returns(CreateUnaryCall()); +#pragma warning restore CS0612 + + var client = new VersionAwareDaprClient(inner.Object, capabilities.Object); + + var firstResponse = await client.ScheduleJobAsync( + new Autogenerated.ScheduleJobRequest(), + new CallOptions(cancellationToken: cancellationToken)); + var secondResponse = await client.ScheduleJobAsync( + new Autogenerated.ScheduleJobRequest(), + new CallOptions(cancellationToken: cancellationToken)); + + Assert.NotNull(firstResponse); + Assert.NotNull(secondResponse); + capabilities.Verify(capability => capability.GetMethodSupportAsync( + "dapr.proto.runtime.v1.Dapr/ScheduleJob", + cancellationToken), Times.Once); + capabilities.Verify(capability => capability.GetMethodSupportAsync( + "dapr.proto.runtime.v1.Dapr/ScheduleJobAlpha1", + cancellationToken), Times.Once); + inner.Verify(client => client.ScheduleJobAsync( + It.IsAny(), + It.IsAny()), Times.Once); +#pragma warning disable CS0612 // Verify that the cached obsolete fallback is invoked directly. + inner.Verify(client => client.ScheduleJobAlpha1Async( + It.IsAny(), + It.IsAny()), Times.Exactly(2)); +#pragma warning restore CS0612 + } + + [Fact] + public async Task ConverseAsync_WhenDiscoveryIsUnknownAndAlpha2IsUnavailable_DoesNotGuessIncompatibleAlpha1() + { + var cancellationToken = TestContext.Current.CancellationToken; + var capabilities = CreateCapabilities(DaprRuntimeSupport.Unknown); + var inner = new Mock(MockBehavior.Strict); + inner + .Setup(client => client.ConverseAlpha2Async( + It.IsAny(), + It.IsAny())) + .Throws(new RpcException(new Status(StatusCode.Unimplemented, "Alpha2 method unavailable."))); + + var client = new VersionAwareDaprClient(inner.Object, capabilities.Object); + + var exception = await Assert.ThrowsAsync(() => client.ConverseAsync( + new Autogenerated.ConversationRequestAlpha2(), + new CallOptions(cancellationToken: cancellationToken))); + + Assert.Contains("runtime method discovery is unavailable", exception.Message); + Assert.Contains("ConverseAlpha1", exception.Message); +#pragma warning disable CS0612 // Verify that the obsolete incompatible fallback was not invoked. + inner.Verify(client => client.ConverseAlpha1Async( + It.IsAny(), + It.IsAny()), Times.Never); +#pragma warning restore CS0612 + } + + private static Mock CreateCapabilities(DaprRuntimeSupport support) + { + var capabilities = new Mock(MockBehavior.Strict); + capabilities + .Setup(capability => capability.GetMethodSupportAsync( + It.IsAny(), + It.IsAny())) + .ReturnsAsync(support); + return capabilities; + } + + private static AsyncUnaryCall CreateUnaryCall() + where TResponse : new() + { + return new AsyncUnaryCall( + Task.FromResult(new TResponse()), + Task.FromResult(new Metadata()), + () => Status.DefaultSuccess, + () => new Metadata(), + () => { }); + } +} diff --git a/test/Dapr.Common.Generators.Tests/Emission/WrapperCodeEmitterTests.cs b/test/Dapr.Common.Generators.Tests/Emission/WrapperCodeEmitterTests.cs index 4c93f4a1e..d31a978c3 100644 --- a/test/Dapr.Common.Generators.Tests/Emission/WrapperCodeEmitterTests.cs +++ b/test/Dapr.Common.Generators.Tests/Emission/WrapperCodeEmitterTests.cs @@ -1,4 +1,4 @@ -// ------------------------------------------------------------------------ +// ------------------------------------------------------------------------ // Copyright 2025 The Dapr Authors // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -77,9 +77,11 @@ public void EmitClass_AutoCompatible_IdenticalTypes_ContainsCapabilityChecks() var groups = Analyze(StubCompilation.WithIdenticalTypeVariants()); var source = WrapperCodeEmitter.EmitClass(groups!); - // Both variants must be checked - Assert.Contains("SupportsMethodAsync(\"dapr.proto.runtime.v1.Dapr/Foo\"", source); - Assert.Contains("SupportsMethodAsync(\"dapr.proto.runtime.v1.Dapr/FooAlpha1\"", source); + // Both variants must be present in the generated catalog and checked at runtime. + Assert.Contains("\"dapr.proto.runtime.v1.Dapr/Foo\"", source); + Assert.Contains("\"dapr.proto.runtime.v1.Dapr/FooAlpha1\"", source); + Assert.Contains("GetMethodSupportAsync(_fooMethodCatalog[0]", source); + Assert.Contains("GetMethodSupportAsync(_fooMethodCatalog[1]", source); } [Fact] @@ -101,6 +103,27 @@ public void EmitClass_AutoCompatible_IdenticalTypes_ThrowsFeatureNotAvailable() Assert.Contains("\"Foo\"", source); } + [Fact] + public void EmitClass_AutoCompatible_EmitsSuccessfulVariantCache() + { + var groups = Analyze(StubCompilation.WithIdenticalTypeVariants()); + var source = WrapperCodeEmitter.EmitClass(groups!); + + Assert.Contains("private int _fooSelectedVariant = -1;", source); + Assert.Contains("Volatile.Read(ref _fooSelectedVariant)", source); + Assert.Contains("Volatile.Write(ref _fooSelectedVariant, 0)", source); + Assert.Contains("Volatile.Write(ref _fooSelectedVariant, 1)", source); + } + + [Fact] + public void EmitClass_SchemaDivergent_DoesNotCacheIncompatibleVariantSelection() + { + var groups = Analyze(StubCompilation.WithIncompatibleAlphaVariants()); + var source = WrapperCodeEmitter.EmitClass(groups!); + + Assert.DoesNotContain("_bazSelectedVariant", source); + } + // ------------------------------------------------------------------------- // Class emission – SchemaDivergent // ------------------------------------------------------------------------- @@ -121,14 +144,12 @@ public void EmitClass_SchemaDivergent_MostRecentVariantIsCheckedFirst() var groups = Analyze(StubCompilation.WithIncompatibleAlphaVariants()); var source = WrapperCodeEmitter.EmitClass(groups!); - // The capability-check strings should appear in order: Alpha2 before Alpha1. - // For SchemaDivergent the fallback variant is not called (NotSupportedException is thrown), - // so we look at the SupportsMethodAsync capability-check strings. + // The generated catalog must preserve the build-time order: Alpha2 before Alpha1. var alpha2Pos = source.IndexOf("Dapr/BazAlpha2\"", StringComparison.Ordinal); var alpha1Pos = source.IndexOf("Dapr/BazAlpha1\"", StringComparison.Ordinal); - Assert.True(alpha2Pos >= 0, "SupportsMethodAsync check for BazAlpha2 should be present"); - Assert.True(alpha1Pos >= 0, "SupportsMethodAsync check for BazAlpha1 should be present"); + Assert.True(alpha2Pos >= 0, "Catalog entry for BazAlpha2 should be present"); + Assert.True(alpha1Pos >= 0, "Catalog entry for BazAlpha1 should be present"); Assert.True(alpha2Pos < alpha1Pos, "Alpha2 (most recent) capability check should appear before Alpha1"); } @@ -148,6 +169,19 @@ public void EmitClass_AutoCompatible_MostRecentVariant_ContainsUnimplementedCatc Assert.Contains("catch (global::Grpc.Core.RpcException __implEx)", source); } + [Fact] + public void EmitClass_AutoCompatible_EveryCatalogVariant_ContainsUnavailableCatch() + { + var groups = Analyze(StubCompilation.WithMultipleFallbacks())!; + var source = WrapperCodeEmitter.EmitClass(groups); + + Assert.Equal( + 3, + source.Split( + "catch (global::Grpc.Core.RpcException __implEx) when (IsMethodUnavailable(__implEx))", + StringSplitOptions.None).Length - 1); + } + [Fact] public void EmitClass_SchemaDivergent_MostRecentVariant_ContainsUnimplementedCatch() { @@ -178,6 +212,7 @@ public void EmitClass_AutoCompatible_MostRecentVariant_ContainsUnknownProxyError Assert.Contains("StatusCode.Unknown", source); Assert.Contains("dapr-callee-app-id or dapr-app-id not found", source); + Assert.Contains("missing dapr-callee-app-id or dapr-app-id metadata", source); } [Fact] @@ -189,6 +224,18 @@ public void EmitClass_SchemaDivergent_MostRecentVariant_ContainsUnknownProxyErro Assert.Contains("StatusCode.Unknown", source); Assert.Contains("dapr-callee-app-id or dapr-app-id not found", source); + Assert.Contains("missing dapr-callee-app-id or dapr-app-id metadata", source); + } + + [Fact] + public void EmitClass_SchemaDivergent_UnknownDiscoveryDoesNotAssumeFallbackSupport() + { + var groups = Analyze(StubCompilation.WithIncompatibleAlphaVariants()); + var source = WrapperCodeEmitter.EmitClass(groups!); + + Assert.Contains("DaprRuntimeSupport.Unknown", source); + Assert.Contains("runtime method discovery is unavailable", source); + Assert.Contains("cannot safely determine", source); } // ------------------------------------------------------------------------- @@ -214,8 +261,10 @@ public void EmitClass_ObsoleteAlpha_GeneratesAutoCompatibleFallback() var groups = Analyze(StubCompilation.WithObsoleteAlphaVariant()); var source = WrapperCodeEmitter.EmitClass(groups!); - Assert.Contains("SupportsMethodAsync(\"dapr.proto.runtime.v1.Dapr/Corf\"", source); - Assert.Contains("SupportsMethodAsync(\"dapr.proto.runtime.v1.Dapr/CorfAlpha1\"", source); + Assert.Contains("\"dapr.proto.runtime.v1.Dapr/Corf\"", source); + Assert.Contains("\"dapr.proto.runtime.v1.Dapr/CorfAlpha1\"", source); + Assert.Contains("GetMethodSupportAsync(_corfMethodCatalog[0]", source); + Assert.Contains("GetMethodSupportAsync(_corfMethodCatalog[1]", source); Assert.Contains("CorfAlpha1Async", source); Assert.Contains("async global::System.Threading.Tasks.Task", source); } @@ -388,7 +437,7 @@ public void EmitClass_AutoCompatible_DifferentTypes_EmitsScalarPropertyCopyForRe var groups = Analyze(StubCompilation.WithCompatibleDifferentTypes())!; var source = WrapperCodeEmitter.EmitClass(groups); - Assert.Contains("__convertedResponse.Result = __fallbackResponse.Result;", source); + Assert.Contains("__convertedResponse.Result = __response1.Result;", source); } [Fact] @@ -405,20 +454,17 @@ public void EmitClass_AutoCompatible_DifferentTypes_UsesFullFallbackBlock() } // ------------------------------------------------------------------------- - // EmitRequestConversion – same-type inner short-circuit - // (sameRequest=true, sameResponse=false → else block runs, but request is not converted) + // Same request, different response // ------------------------------------------------------------------------- [Fact] - public void EmitClass_SameRequestDifferentResponse_EmitsFallbackRequestEqualsRequest() + public void EmitClass_SameRequestDifferentResponse_UsesOriginalRequest() { - // When mostRecent and fallback share the same request type, EmitRequestConversion - // must emit the short-circuit form `var __fallbackRequest = request;` rather than - // creating a new object. var groups = Analyze(StubCompilation.WithSameRequestDifferentResponse())!; var source = WrapperCodeEmitter.EmitClass(groups); - Assert.Contains("var __fallbackRequest = request;", source); + Assert.Contains("XyzAlpha1Async(request, options)", source); + Assert.DoesNotContain("var __fallbackRequest = request;", source); } [Fact] @@ -440,13 +486,12 @@ public void EmitClass_SameRequestDifferentResponse_EmitsNewResponseObjectAndProp var source = WrapperCodeEmitter.EmitClass(groups); Assert.Contains("new global::Dapr.Client.Autogen.Grpc.v1.XyzResponse()", source); - Assert.Contains("__convertedResponse.Result = __fallbackResponse.Result;", source); + Assert.Contains("__convertedResponse.Result = __response1.Result;", source); Assert.Contains("return __convertedResponse;", source); } // ------------------------------------------------------------------------- - // EmitResponseConversion – same-type inner short-circuit - // (sameRequest=false, sameResponse=true → else block runs, but response is not converted) + // Different request, same response // ------------------------------------------------------------------------- [Fact] @@ -463,12 +508,10 @@ public void EmitClass_DifferentRequestSameResponse_EmitsNewFallbackRequestObject [Fact] public void EmitClass_DifferentRequestSameResponse_EmitsReturnFallbackResponse() { - // When mostRecent and fallback share the same response type, EmitResponseConversion - // must emit the short-circuit `return __fallbackResponse;` rather than a conversion. var groups = Analyze(StubCompilation.WithDifferentRequestSameResponse())!; var source = WrapperCodeEmitter.EmitClass(groups); - Assert.Contains("return __fallbackResponse;", source); + Assert.Contains("return __response1;", source); } [Fact] @@ -551,9 +594,12 @@ public void EmitClass_MultipleFallbacks_EmitsAllCapabilityChecks() var groups = Analyze(StubCompilation.WithMultipleFallbacks())!; var source = WrapperCodeEmitter.EmitClass(groups); - Assert.Contains("SupportsMethodAsync(\"dapr.proto.runtime.v1.Dapr/Grault\"", source); - Assert.Contains("SupportsMethodAsync(\"dapr.proto.runtime.v1.Dapr/GraultAlpha2\"", source); - Assert.Contains("SupportsMethodAsync(\"dapr.proto.runtime.v1.Dapr/GraultAlpha1\"", source); + Assert.Contains("\"dapr.proto.runtime.v1.Dapr/Grault\"", source); + Assert.Contains("\"dapr.proto.runtime.v1.Dapr/GraultAlpha2\"", source); + Assert.Contains("\"dapr.proto.runtime.v1.Dapr/GraultAlpha1\"", source); + Assert.Contains("GetMethodSupportAsync(_graultMethodCatalog[0]", source); + Assert.Contains("GetMethodSupportAsync(_graultMethodCatalog[1]", source); + Assert.Contains("GetMethodSupportAsync(_graultMethodCatalog[2]", source); } [Fact] diff --git a/test/Dapr.Common.Test/Dapr.Common.Test.csproj b/test/Dapr.Common.Test/Dapr.Common.Test.csproj index e73e6bc80..804fbff4b 100644 --- a/test/Dapr.Common.Test/Dapr.Common.Test.csproj +++ b/test/Dapr.Common.Test/Dapr.Common.Test.csproj @@ -6,6 +6,8 @@ all + + diff --git a/test/Dapr.Common.Test/DaprRuntimeCapabilitiesTests.cs b/test/Dapr.Common.Test/DaprRuntimeCapabilitiesTests.cs new file mode 100644 index 000000000..f0fe500de --- /dev/null +++ b/test/Dapr.Common.Test/DaprRuntimeCapabilitiesTests.cs @@ -0,0 +1,400 @@ +// ------------------------------------------------------------------------ +// Copyright 2026 The Dapr Authors +// 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. +// ------------------------------------------------------------------------ + +using System; +using System.Diagnostics; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Google.Protobuf; +using Google.Protobuf.Reflection; +using Grpc.Core; +using Grpc.Net.Client; +using Grpc.Reflection.V1; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.TestHost; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Xunit; + +namespace Dapr.Common.Test; + +public sealed class DaprRuntimeCapabilitiesTests +{ + private static readonly TimeSpan ReflectionTimeout = TimeSpan.FromMilliseconds(200); + + [Fact] + public async Task GetMethodSupportAsync_WhenReflectionNeverCompletes_ReturnsUnknown() + { + var testCancellationToken = TestContext.Current.CancellationToken; + var reflection = ReflectionService.Hanging(); + await using var server = await ReflectionTestServer.StartAsync(reflection, testCancellationToken); + var capabilities = server.CreateCapabilities(ReflectionTimeout); + var stopwatch = Stopwatch.StartNew(); + + var support = await capabilities.GetMethodSupportAsync( + "dapr.proto.runtime.v1.Dapr/BulkPublishEvent", + testCancellationToken); + + Assert.Equal(DaprRuntimeSupport.Unknown, support); + Assert.Equal(1, reflection.CallCount); + Assert.InRange(stopwatch.Elapsed, TimeSpan.Zero, TimeSpan.FromSeconds(2)); + } + + [Fact] + public async Task GetMethodSupportAsync_WhenReflectionNeverCompletes_ConcurrentCallersShareOneBoundedLookup() + { + var testCancellationToken = TestContext.Current.CancellationToken; + var reflection = ReflectionService.Hanging(); + await using var server = await ReflectionTestServer.StartAsync(reflection, testCancellationToken); + var capabilities = server.CreateCapabilities(TimeSpan.FromSeconds(2)); + + var calls = Enumerable.Range(0, 20) + .Select(index => capabilities.GetMethodSupportAsync( + $"dapr.proto.runtime.v1.Dapr/BulkPublishEvent{index}", + testCancellationToken)) + .ToArray(); + + await reflection.RequestReceived.WaitAsync(TimeSpan.FromSeconds(1), testCancellationToken); + var results = await Task.WhenAll(calls).WaitAsync(TimeSpan.FromSeconds(3), testCancellationToken); + + Assert.All(results, result => Assert.Equal(DaprRuntimeSupport.Unknown, result)); + Assert.Equal(1, reflection.CallCount); + } + + [Fact] + public async Task GetMethodSupportAsync_WhenOneServiceHangs_DoesNotBlockAnotherServiceLookup() + { + var testCancellationToken = TestContext.Current.CancellationToken; + const string hangingService = "dapr.proto.runtime.v1.Dapr"; + const string availableService = "example.v1.Available"; + var reflection = ReflectionService.HangingForService( + hangingService, + CreateDescriptor(availableService, "AvailableMethod")); + await using var server = await ReflectionTestServer.StartAsync(reflection, testCancellationToken); + var capabilities = server.CreateCapabilities(TimeSpan.FromSeconds(2)); + + var hangingLookup = capabilities.GetMethodSupportAsync( + $"{hangingService}/BulkPublishEvent", + testCancellationToken); + await reflection.HangingRequestReceived.WaitAsync(TimeSpan.FromSeconds(1), testCancellationToken); + + var availableLookup = capabilities.GetMethodSupportAsync( + $"{availableService}/AvailableMethod", + testCancellationToken); + + await reflection.AvailableRequestReceived.WaitAsync(TimeSpan.FromSeconds(1), testCancellationToken); + Assert.Equal(DaprRuntimeSupport.Supported, await availableLookup); + Assert.False(hangingLookup.IsCompleted); + Assert.Equal( + DaprRuntimeSupport.Unknown, + await hangingLookup.WaitAsync(TimeSpan.FromSeconds(3), testCancellationToken)); + Assert.Equal(2, reflection.CallCount); + } + + [Fact] + public async Task GetMethodSupportAsync_WhenCallerCancels_DoesNotCancelOrPoisonSharedLookup() + { + var testCancellationToken = TestContext.Current.CancellationToken; + const string serviceName = "dapr.proto.runtime.v1.Dapr"; + var reflection = ReflectionService.BlockedUntilReleased( + CreateDescriptor(serviceName, "BulkPublishEvent")); + await using var server = await ReflectionTestServer.StartAsync(reflection, testCancellationToken); + var capabilities = server.CreateCapabilities(TimeSpan.FromSeconds(2)); + using var callerCancellation = new CancellationTokenSource(); + + var cancelledLookup = capabilities.GetMethodSupportAsync( + $"{serviceName}/BulkPublishEvent", + callerCancellation.Token); + await reflection.RequestReceived.WaitAsync(TimeSpan.FromSeconds(1), testCancellationToken); + callerCancellation.Cancel(); + + await Assert.ThrowsAnyAsync(() => cancelledLookup); + + reflection.Release(); + + Assert.Equal( + DaprRuntimeSupport.Supported, + await capabilities.GetMethodSupportAsync( + $"{serviceName}/BulkPublishEvent", + testCancellationToken)); + Assert.Equal(1, reflection.CallCount); + } + + [Theory] + [InlineData(StatusCode.Unknown)] + [InlineData(StatusCode.Unimplemented)] + [InlineData(StatusCode.Cancelled)] + [InlineData(StatusCode.DeadlineExceeded)] + public async Task GetMethodSupportAsync_WhenReflectionReturnsRpcError_CachesUnknownResult( + StatusCode statusCode) + { + var testCancellationToken = TestContext.Current.CancellationToken; + var reflection = ReflectionService.Failing(statusCode); + await using var server = await ReflectionTestServer.StartAsync(reflection, testCancellationToken); + var capabilities = server.CreateCapabilities(ReflectionTimeout); + + var first = await capabilities.GetMethodSupportAsync( + "dapr.proto.runtime.v1.Dapr/BulkPublishEvent", + testCancellationToken); + var second = await capabilities.GetMethodSupportAsync( + "another.Service/PublishEvent", + testCancellationToken); + + Assert.Equal(DaprRuntimeSupport.Unknown, first); + Assert.Equal(DaprRuntimeSupport.Unknown, second); + Assert.Equal(1, reflection.CallCount); + } + + [Fact] + public async Task GetMethodSupportAsync_WhenReflectionSucceeds_CachesAndDistinguishesMethods() + { + var testCancellationToken = TestContext.Current.CancellationToken; + const string serviceName = "dapr.proto.runtime.v1.Dapr"; + var reflection = ReflectionService.Successful( + CreateDescriptor(serviceName, "BulkPublishEvent", "PublishEvent")); + await using var server = await ReflectionTestServer.StartAsync(reflection, testCancellationToken); + var capabilities = server.CreateCapabilities(ReflectionTimeout); + + var bulkPublishSupport = await capabilities.GetMethodSupportAsync( + $"{serviceName}/BulkPublishEvent", + testCancellationToken); + var publishSupport = await capabilities.GetMethodSupportAsync( + $"{serviceName}/PublishEvent", + testCancellationToken); + var missingSupport = await capabilities.GetMethodSupportAsync( + $"{serviceName}/MissingMethod", + testCancellationToken); + + Assert.Equal(DaprRuntimeSupport.Supported, bulkPublishSupport); + Assert.Equal(DaprRuntimeSupport.Supported, publishSupport); + Assert.Equal(DaprRuntimeSupport.Unsupported, missingSupport); + Assert.Equal(1, reflection.CallCount); + } + + [Fact] + public async Task GetServiceSupportAsync_WhenReflectionUnavailable_ReturnsUnknownAndCaches() + { + var testCancellationToken = TestContext.Current.CancellationToken; + var reflection = ReflectionService.Hanging(); + await using var server = await ReflectionTestServer.StartAsync(reflection, testCancellationToken); + var capabilities = server.CreateCapabilities(ReflectionTimeout); + + var first = await capabilities.GetServiceSupportAsync( + "dapr.proto.runtime.v1.Dapr", + testCancellationToken); + var second = await capabilities.GetServiceSupportAsync( + "another.Service", + testCancellationToken); + + Assert.Equal(DaprRuntimeSupport.Unknown, first); + Assert.Equal(DaprRuntimeSupport.Unknown, second); + Assert.Equal(1, reflection.CallCount); + } + + private static FileDescriptorProto CreateDescriptor(string fullyQualifiedServiceName, params string[] methods) + { + var separator = fullyQualifiedServiceName.LastIndexOf('.'); + var package = separator < 0 ? string.Empty : fullyQualifiedServiceName[..separator]; + var serviceName = separator < 0 + ? fullyQualifiedServiceName + : fullyQualifiedServiceName[(separator + 1)..]; + var service = new ServiceDescriptorProto { Name = serviceName }; + service.Method.Add(methods.Select(method => new MethodDescriptorProto { Name = method })); + + var descriptor = new FileDescriptorProto + { + Name = $"{serviceName.ToLowerInvariant()}.proto", + Package = package, + }; + descriptor.Service.Add(service); + return descriptor; + } + + private sealed class ReflectionTestServer : IAsyncDisposable + { + private readonly IHost host; + + private ReflectionTestServer(IHost host, GrpcChannel channel) + { + this.host = host; + Channel = channel; + } + + private GrpcChannel Channel { get; } + + public static async Task StartAsync( + ReflectionService reflection, + CancellationToken cancellationToken) + { + var host = await new HostBuilder() + .ConfigureWebHost(webHost => webHost + .UseTestServer() + .ConfigureServices(services => + { + services.AddGrpc(); + services.AddSingleton(reflection); + }) + .Configure(app => + { + app.UseRouting(); + app.UseEndpoints(endpoints => endpoints.MapGrpcService()); + })) + .StartAsync(cancellationToken); + + var client = host.GetTestClient(); + client.DefaultRequestVersion = new Version(2, 0); + var channel = GrpcChannel.ForAddress( + client.BaseAddress!, + new GrpcChannelOptions { HttpClient = client }); + return new ReflectionTestServer(host, channel); + } + + public DaprRuntimeCapabilities CreateCapabilities(TimeSpan timeout) + { + return new DaprRuntimeCapabilities( + new ServerReflection.ServerReflectionClient(Channel), + timeout); + } + + public async ValueTask DisposeAsync() + { + Channel.Dispose(); + await host.StopAsync(TestContext.Current.CancellationToken); + host.Dispose(); + } + } + + private sealed class ReflectionService : ServerReflection.ServerReflectionBase + { + private readonly Func, + ServerCallContext, Task> handleRequest; + private int callCount; + + private ReflectionService( + Func, + ServerCallContext, Task> handleRequest) + { + this.handleRequest = handleRequest; + } + + public int CallCount => Volatile.Read(ref callCount); + + public Task RequestReceived => requestReceived.Task; + + public Task HangingRequestReceived => hangingRequestReceived.Task; + + public Task AvailableRequestReceived => availableRequestReceived.Task; + + private readonly TaskCompletionSource requestReceived = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource hangingRequestReceived = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource availableRequestReceived = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private TaskCompletionSource release = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + public static ReflectionService Hanging() + { + return new ReflectionService(async (_, _, context) => + { + await Task.Delay(Timeout.InfiniteTimeSpan, context.CancellationToken); + }); + } + + public static ReflectionService HangingForService( + string hangingService, + FileDescriptorProto availableDescriptor) + { + return new ReflectionService(async (request, responseStream, context) => + { + if (request.FileContainingSymbol == hangingService) + { + await Task.Delay(Timeout.InfiniteTimeSpan, context.CancellationToken); + return; + } + + await WriteDescriptorAsync(responseStream, availableDescriptor); + }); + } + + public static ReflectionService BlockedUntilReleased(FileDescriptorProto descriptor) + { + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var service = new ReflectionService(async (_, responseStream, context) => + { + await release.Task.WaitAsync(context.CancellationToken); + await WriteDescriptorAsync(responseStream, descriptor); + }); + service.release = release; + return service; + } + + public static ReflectionService Failing(StatusCode statusCode) + { + return new ReflectionService((_, _, _) => + throw new RpcException(new Status(statusCode, "Reflection unavailable."))); + } + + public static ReflectionService Successful(FileDescriptorProto descriptor) + { + return new ReflectionService((_, responseStream, _) => + WriteDescriptorAsync(responseStream, descriptor)); + } + + public void Release() + { + release.TrySetResult(); + } + + public override async Task ServerReflectionInfo( + IAsyncStreamReader requestStream, + IServerStreamWriter responseStream, + ServerCallContext context) + { + Interlocked.Increment(ref callCount); + + while (await requestStream.MoveNext(context.CancellationToken)) + { + var request = requestStream.Current; + requestReceived.TrySetResult(); + if (request.FileContainingSymbol == "dapr.proto.runtime.v1.Dapr") + { + hangingRequestReceived.TrySetResult(); + } + else + { + availableRequestReceived.TrySetResult(); + } + + await handleRequest(request, responseStream, context); + } + } + + private static Task WriteDescriptorAsync( + IServerStreamWriter responseStream, + FileDescriptorProto descriptor) + { + var response = new ServerReflectionResponse + { + FileDescriptorResponse = new FileDescriptorResponse(), + }; + response.FileDescriptorResponse.FileDescriptorProto.Add(descriptor.ToByteString()); + return responseStream.WriteAsync(response); + } + } +} diff --git a/test/Dapr.Messaging.Test/PublishSubscribe/DaprPublishSubscribeGrpcClientPublishTests.cs b/test/Dapr.Messaging.Test/PublishSubscribe/DaprPublishSubscribeGrpcClientPublishTests.cs index 1ecc652d5..88d748b02 100644 --- a/test/Dapr.Messaging.Test/PublishSubscribe/DaprPublishSubscribeGrpcClientPublishTests.cs +++ b/test/Dapr.Messaging.Test/PublishSubscribe/DaprPublishSubscribeGrpcClientPublishTests.cs @@ -55,8 +55,8 @@ private static (Mock Mock, DaprPublishSubscribeGrpcClient Cli { var mock = new Mock(); var caps = new Mock(); - caps.Setup(c => c.SupportsMethodAsync(It.IsAny(), It.IsAny())) - .ReturnsAsync(true); + caps.Setup(c => c.GetMethodSupportAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(DaprRuntimeSupport.Supported); var client = new DaprPublishSubscribeGrpcClient( mock.Object, new HttpClient(), new System.Text.Json.JsonSerializerOptions(), caps.Object); return (mock, client); @@ -430,8 +430,8 @@ public async Task PublishEventAsync_WithApiToken_ForwardsDaprApiTokenHeader() { var mock = new Mock(); var caps = new Mock(); - caps.Setup(c => c.SupportsMethodAsync(It.IsAny(), It.IsAny())) - .ReturnsAsync(true); + caps.Setup(c => c.GetMethodSupportAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(DaprRuntimeSupport.Supported); var client = new DaprPublishSubscribeGrpcClient( mock.Object, new HttpClient(), new System.Text.Json.JsonSerializerOptions(), caps.Object, daprApiToken: "secret-token"); diff --git a/test/Shared/TestClient.cs b/test/Shared/TestClient.cs index e1bc5cc7c..e08e5928f 100644 --- a/test/Shared/TestClient.cs +++ b/test/Shared/TestClient.cs @@ -77,11 +77,11 @@ public static TestClient CreateForDaprClient(Action private sealed class AlwaysAvailableRuntimeCapabilities : IDaprRuntimeCapabilities { - public Task SupportsMethodAsync(string fullyQualifiedMethodName, CancellationToken cancellationToken = default) - => Task.FromResult(true); + public Task GetMethodSupportAsync(string fullyQualifiedMethodName, CancellationToken cancellationToken = default) + => Task.FromResult(DaprRuntimeSupport.Supported); - public Task SupportsServiceAsync(string serviceName, CancellationToken cancellationToken = default) - => Task.FromResult(true); + public Task GetServiceSupportAsync(string serviceName, CancellationToken cancellationToken = default) + => Task.FromResult(DaprRuntimeSupport.Supported); } private static async Task WithTimeout(Task task, TimeSpan timeout, string message)