From 9bd0d71d063f4d92d516a5267978e653683986bc Mon Sep 17 00:00:00 2001 From: Whit Waldo Date: Tue, 29 Sep 2026 11:30:23 -0500 Subject: [PATCH 1/2] Iterates on #1816 implementation to both fix the possible reflection lookup stalls, but also improve on the source generator to better identify if methods are supported, unsupported or unknown based on protos at build time. At runtime then, it prefers to use reflection to see what's there, fails back to the source generator registry with failovers based on that information that'll try through each method until one works. Added in-memory cache so if a failover is required for a given method (and the response succeeds), it's persistently used until the app restarts. Signed-off-by: Whit Waldo --- .../DaprVersionAwareGenerator.cs | 7 +- .../Emission/WrapperCodeEmitter.cs | 230 +++++++--- src/Dapr.Common/DaprRuntimeCapabilities.cs | 189 ++++++--- src/Dapr.Common/DaprRuntimeSupport.cs | 37 ++ src/Dapr.Common/IDaprRuntimeCapabilities.cs | 14 +- .../StateTestClient.cs | 8 +- .../BulkPublishEventApiTest.cs | 145 ++++++- test/Dapr.Client.Test/MockClient.cs | 8 +- .../VersionAwareDaprClientFallbackTests.cs | 240 +++++++++++ .../Emission/WrapperCodeEmitterTests.cs | 102 +++-- test/Dapr.Common.Test/Dapr.Common.Test.csproj | 2 + .../DaprRuntimeCapabilitiesTests.cs | 399 ++++++++++++++++++ ...rPublishSubscribeGrpcClientPublishTests.cs | 8 +- test/Shared/TestClient.cs | 8 +- 14 files changed, 1220 insertions(+), 177 deletions(-) create mode 100644 src/Dapr.Common/DaprRuntimeSupport.cs create mode 100644 test/Dapr.Client.Test/VersionAwareDaprClientFallbackTests.cs create mode 100644 test/Dapr.Common.Test/DaprRuntimeCapabilitiesTests.cs 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..a9688f391 --- /dev/null +++ b/test/Dapr.Common.Test/DaprRuntimeCapabilitiesTests.cs @@ -0,0 +1,399 @@ +// ------------------------------------------------------------------------ +// 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(ReflectionTimeout); + + var calls = Enumerable.Range(0, 20) + .Select(index => capabilities.GetMethodSupportAsync( + $"dapr.proto.runtime.v1.Dapr/BulkPublishEvent{index}", + testCancellationToken)) + .ToArray(); + + var results = await Task.WhenAll(calls).WaitAsync(TimeSpan.FromSeconds(2), 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) From 69b07098ce57f37a16ecb10cd69eaa6681affa53 Mon Sep 17 00:00:00 2001 From: Whit Waldo Date: Tue, 29 Sep 2026 11:54:04 -0500 Subject: [PATCH 2/2] Addressing too short of a timeout, especially during integration tests Signed-off-by: Whit Waldo --- test/Dapr.Common.Test/DaprRuntimeCapabilitiesTests.cs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/test/Dapr.Common.Test/DaprRuntimeCapabilitiesTests.cs b/test/Dapr.Common.Test/DaprRuntimeCapabilitiesTests.cs index a9688f391..f0fe500de 100644 --- a/test/Dapr.Common.Test/DaprRuntimeCapabilitiesTests.cs +++ b/test/Dapr.Common.Test/DaprRuntimeCapabilitiesTests.cs @@ -60,7 +60,7 @@ public async Task GetMethodSupportAsync_WhenReflectionNeverCompletes_ConcurrentC var testCancellationToken = TestContext.Current.CancellationToken; var reflection = ReflectionService.Hanging(); await using var server = await ReflectionTestServer.StartAsync(reflection, testCancellationToken); - var capabilities = server.CreateCapabilities(ReflectionTimeout); + var capabilities = server.CreateCapabilities(TimeSpan.FromSeconds(2)); var calls = Enumerable.Range(0, 20) .Select(index => capabilities.GetMethodSupportAsync( @@ -68,7 +68,8 @@ public async Task GetMethodSupportAsync_WhenReflectionNeverCompletes_ConcurrentC testCancellationToken)) .ToArray(); - var results = await Task.WhenAll(calls).WaitAsync(TimeSpan.FromSeconds(2), testCancellationToken); + 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);