diff --git a/CHANGELOG.md b/CHANGELOG.md index 37b2a12..51fdaa2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,25 @@ ## [Unreleased] +## [0.7.10] - 2026-07-22 + +### Added + +- `SharpLinkMultiClusterClientBuilder`, `ISharpLinkMultiClusterClient`, and `SharpLinkClusterKey` coordinate multiple isolated child clients while routing each generated contract to exactly one cluster slot. +- `[assembly: SharpLinkClusterContractAssembly(cluster, typeof(Marker))]` generates a deterministic, weak-catalogued static route manifest. The generator rejects invalid cluster keys, missing generated manifests, and contradictory contract-assembly routes. +- Dynamic registration, unregister, and replacement now have explicit multi-cluster overloads. Contract-owning assemblies remain exclusive to one slot; dependency-only assemblies can remain independently owned by more than one slot. +- `AddSharpLinkMultiClusterClient` adds hosted startup, shutdown, and `ISharpLinkMultiClusterClientAccessor` without exposing child `ISharpLinkClient` instances through DI. + +### Changed + +- `SharpClientBuilder.Build()` keeps its existing complete generated-manifest snapshot behavior. The multi-cluster builder uses an internal filtered build context, so ordinary clients and the fixed-endpoint RPC hot path do not perform coordinator lookups or acquire new locks. +- Cluster selection is performed only by `Get()`; generated proxies call their selected child channel directly and Protocol v2 frames, handshake capabilities, headers, and metadata do not carry a cluster key. + +### Compatibility + +- Ordinary single-client applications require no migration. Endpoint-cluster retry, admission, circuit-breaker, resolver, transport, authentication, and connection-pool behavior stays owned by each child client. +- Static NativeAOT routing remains manifest based. Runtime assembly registration continues to return the existing structured platform-not-supported result when unavailable. + ## [0.7.9] - 2026-07-21 ### 收敛 diff --git a/Directory.Build.props b/Directory.Build.props index cf8e406..6173531 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -7,7 +7,7 @@ - 0.7.9 + 0.7.10 sunsi MIT false diff --git a/README.md b/README.md index f6c8afc..81d4a30 100644 --- a/README.md +++ b/README.md @@ -86,6 +86,47 @@ dotnet run --project demo/SeparatedClient/SeparatedClient.csproj [assembly: SharpLinkRpcContracts(typeof(MyContract1), typeof(MyContract2))] ``` +## Multi-cluster clients + +`SharpLinkMultiClusterClientBuilder` owns several isolated child clients. A contract is mapped to a +single child while its proxy is created; RPC calls made through that proxy do not consult a +coordinator, cluster name, or per-call routing context. + +Declare static contract-assembly routes in the application or host assembly, not in a reusable +contract package: + +```csharp +[assembly: SharpLinkClusterContractAssembly("orders", typeof(OrderContractsMarker))] +[assembly: SharpLinkClusterContractAssembly("payments", typeof(PaymentContractsMarker))] +``` + +Configure each slot with the existing child builder API. The `UseCluster` method inside the delegate +still configures endpoint topology for that one slot; it is not the multi-cluster coordinator API. + +```csharp +var client = SharpLinkMultiClusterClientBuilder.Create() + .AddCluster("orders", child => child.UseTcp("127.0.0.1", 5101)) + .AddCluster("payments", child => child.UseTcp("127.0.0.1", 5102)) + .Build(); + +await client.ConnectAsync(); +var orders = client.Get(); +var payments = client.Get(); +``` + +Every slot is required by default. A slot intentionally reserved for plugins must explicitly opt in: + +```csharp +.AddCluster("plugins", child => child.UseTcp("127.0.0.1", 5103), + slot => slot.AllowDynamicContracts = true) +``` + +Dynamic contracts are registered against an explicit slot with +`RegisterAssembly(cluster, assembly)`, `UnregisterAssemblyAsync(cluster, assembly, timeout)`, and +`ReplaceAssemblyAsync(cluster, oldAssembly, newAssembly, timeout)`. There is no default cluster, +per-call cluster override, cross-cluster retry, or cluster identifier on the wire. See +`doc/architecture-0.7.10.md` and `doc/migration-0.7.10.md` for lifecycle and migration details. + ## 契约 Manifest 与兼容性基线 `SharpLink.Sdk` 包会把当前契约写到 `obj///SharpLink.Contracts.sharplink.json`。JSON 按 Contract、Method、DTO member、enum、union 与 Service route 的稳定 ID 排序，不包含时间戳或源码路径；`schemaFingerprint` 覆盖规范化后的完整内容，可直接作为 CI 构建产物保存。 diff --git a/Sharplink.slnx b/Sharplink.slnx index d519d07..23e6ffd 100644 --- a/Sharplink.slnx +++ b/Sharplink.slnx @@ -29,5 +29,6 @@ + diff --git a/demo/SeparatedClient/Program.cs b/demo/SeparatedClient/Program.cs index 8650e79..a99803b 100644 --- a/demo/SeparatedClient/Program.cs +++ b/demo/SeparatedClient/Program.cs @@ -4,6 +4,7 @@ using SharpLink.Sdk; [assembly: SharpLinkRpcContracts(typeof(IGreetingService))] +[assembly: SharpLinkClusterContractAssembly("greetings", typeof(IGreetingService))] const int port = 19110; diff --git a/doc/architecture-0.7.10.md b/doc/architecture-0.7.10.md new file mode 100644 index 0000000..8389990 --- /dev/null +++ b/doc/architecture-0.7.10.md @@ -0,0 +1,36 @@ +# SharpLink 0.7.10 multi-cluster architecture + +## Ownership + +`SharpLinkMultiClusterClient` is a coordinator, not a flattened endpoint topology. Each slot owns an ordinary `SharpLinkClient`, including endpoint selection, retry exclusion, circuit breaker, admission, resolver, transport factories, connections, pending calls, and dynamic modules. + +``` +application + | + +-- Get() -> orders slot -> SharpLinkClient -> endpoint topology + +-- Get() -> payments slot -> SharpLinkClient -> endpoint topology +``` + +The coordinator stores immutable `FrozenDictionary` snapshots. `Get` does one type lookup and calls the selected child `Get`. The generated proxy retains that child channel, so an RPC call does not read a route table, parse a key, allocate a routing object, use `AsyncLocal`, or add a field to Protocol v2 frames. + +## Static routes + +The generator emits `ISharpLinkGeneratedClusterRouteManifest` from assembly-level `SharpLinkClusterContractAssembly` attributes. Its catalog keeps weak references and unregisters collectible AssemblyLoadContexts while unloading. At build time the coordinator validates limits, snapshots only declared routes, validates unique contract ownership, expands generated dependency closures inside the same slot, and passes the immutable manifest list to the child builder's internal build context. A dependency closure contributes its generated codecs and assembly identity, but its proxy descriptors are hidden unless that contract-owning assembly is explicitly routed to the slot. + +An ordinary `SharpClientBuilder.Build()` remains unchanged and still snapshots the complete generated assembly catalog. A multi-cluster child never implicitly exposes unrelated process-wide manifests. + +## Lifecycle and locking + +`ConnectAsync` is shared and starts all required child slots with bounded parallelism. Initial failure cancels unscheduled work, stops every started child, and transitions the coordinator to `Faulted`. After a successful start, a non-ready child makes the aggregate state `Degraded`; routes remain pinned to their original slot and no fallback is attempted. `StopAsync` first transitions to `Draining`, then stops every child even when another child fails cleanup, releases route snapshots, and reaches `Stopped`. + +Lock order is coordinator gate, then child gate. The coordinator never awaits while holding its gate, and child code never calls back into the coordinator while holding a child gate. Route reads are lock-free. User-provided factories, resolver, selector, admission, retry, interceptor, authenticator, and service-provider callbacks are not invoked while the coordinator gate is held. + +## Dynamic assemblies + +`RegisterAssembly(cluster, assembly)` validates the target slot and global route ownership, then delegates manifest, codec, dependency, and module validation to the selected child. Only after child registration succeeds does the coordinator publish one new route snapshot. A publication failure immediately starts a zero-wait child unregister. + +Unregister removes new-proxy visibility from the coordinator before using the child's existing module drain. Existing proxies retain their child module lease semantics. Replacement is limited to the same slot and identical `ContractId` sets; new `Get` calls observe the child replacement after it is published. Dynamic ownership is retained until the child reports that framework references have been released, preserving collectible ALC behavior. + +## AOT and observability + +Static routing uses generated manifests and module initializers only. It does not scan loaded assemblies. Dynamic registration retains the established structured failure on platforms where it is unsupported. Cluster names are not added to default per-call metrics. diff --git a/doc/migration-0.7.10.md b/doc/migration-0.7.10.md new file mode 100644 index 0000000..69ff9fa --- /dev/null +++ b/doc/migration-0.7.10.md @@ -0,0 +1,38 @@ +# Migrating to SharpLink 0.7.10 + +## Existing clients + +No migration is needed for `SharpClientBuilder`, `ISharpLinkClient`, endpoint clusters, Protocol v2, or existing generated contract assemblies. A normal client preserves its existing manifest discovery and fixed-endpoint fast path. + +## Static multi-cluster clients + +1. Keep each reusable contract package independent of deployment topology. +2. In the application or hosting assembly, declare one route for each contract-owning assembly: + +```csharp +[assembly: SharpLinkClusterContractAssembly("orders", typeof(OrderContractsMarker))] +[assembly: SharpLinkClusterContractAssembly("payments", typeof(PaymentContractsMarker))] +``` + +3. Configure matching slots with `SharpLinkMultiClusterClientBuilder`. +4. Call `Get()` without a cluster parameter. + +The generator rejects a malformed key or one contract assembly assigned to two clusters. Routes whose slots are not configured on a coordinator are ignored. For configured slots, Build rejects duplicate ContractType/ContractId ownership, missing generated dependencies, an empty slot without explicit dynamic opt-in, or a connection budget overrun. + +## Runtime plugins + +Use the explicit-slot APIs: + +```csharp +client.RegisterAssembly("plugins", pluginAssembly); +await client.UnregisterAssemblyAsync("plugins", pluginAssembly, TimeSpan.FromSeconds(10)); +await client.ReplaceAssemblyAsync("plugins", oldAssembly, newAssembly, TimeSpan.FromSeconds(10)); +``` + +Contract-owning assemblies can belong to exactly one slot. Dependency-only assemblies may be owned by several slots, but their generated dependencies must already be present in the same child slot. Replacement cannot add, remove, or move contracts; use unregister followed by register when the contract set changes. + +## Intentional limits + +0.7.10 has no `Get(cluster)`, default cluster, cross-cluster retry/failover, runtime route move, cluster list discovery, or cluster metadata in requests. Every configured slot participates in `ConnectAsync`; a plugin-only slot must set `AllowDynamicContracts = true`. + +For hosted applications use `services.AddSharpLinkMultiClusterClient(...)` and obtain only `ISharpLinkMultiClusterClientAccessor`. Child clients are intentionally not registered in DI. diff --git a/doc/performance-0.7.10.md b/doc/performance-0.7.10.md new file mode 100644 index 0000000..848e179 --- /dev/null +++ b/doc/performance-0.7.10.md @@ -0,0 +1,35 @@ +# SharpLink 0.7.10 performance evidence + +## Scope + +The coordinator is constructed only by `SharpLinkMultiClusterClientBuilder`; normal `SharpClientBuilder` construction and all single-client RPC invocation code are unchanged. The multi-cluster read path is limited to proxy construction. Calls made through an existing proxy use the identical child `IRpcChannel` path. + +## Reproducible gate + +Baseline is the implementation branch point `bfb26f0ba40c980a59e621d0a549399d7c444fba` (0.7.9). Candidate code is `08f83c1e76b4e5f4a9354527bb2d8d0a10100572`. Use Release builds with no debugger or profiler, an unchanged power mode, identical concurrency and payloads, and alternate baseline and candidate for five warmed runs. + +``` +dotnet build Sharplink.slnx -c Release +dotnet run --project test/SharpLink.Benchmarks/SharpLink.Benchmarks.csproj -c Release -- --filter *UnaryBenchmarks.Rpc_Add* +``` + +Record raw QPS, P50, P99, error count, process allocations, and BenchmarkDotNet bytes/op for every run. P1 passes only when ordinary fixed TCP QPS median is at least 99% of baseline, P99 is at most 105%, and bytes/op do not increase. Repeat for UDS, Named Pipe and Shared Memory where supported, at c1/c8/c32/c128. + +P2 compares a direct child proxy with a proxy acquired once from the multi-cluster coordinator; the measurement starts after `Get`. It requires QPS >=99%, P99 <=105%, identical bytes/op, and a route lookup counter that does not increase during calls. P3 runs resolver or dynamic-registration writes in slot A while slot B invokes a static unary method; slot B requires QPS >=97%, P99 <=105%, and zero unexpected failures. P4 records build/connect/stop time, configured and actual connection count, and coordinator memory for 2, 8, and 16 slots. + +## Local P1 sample + +The following alternating runs used macOS 26.4.1 arm64, .NET 10.0.2, 10 logical processors, Server GC disabled, fixed TCP `add`, c32, 2-second warmup, and 5-second measurement. Every run completed with zero errors. Raw JSON is retained locally under `artifacts/performance/issue-10/p1/`. + +| round | baseline QPS | candidate QPS | baseline P99 (us) | candidate P99 (us) | +| --- | ---: | ---: | ---: | ---: | +| 1 | 80,621 | 54,618 | 7,561 | 9,229 | +| 2 | 419,848 | 421,875 | 137 | 130 | +| 3 | 366,690 | 187,101 | 169 | 1,585 | +| 4 | 413,036 | 418,630 | 138 | 137 | +| 5 | 486,636 | 402,208 | 121 | 146 | +| median | 413,036 | 402,208 | 138 | 146 | + +The local machine showed substantial short-run scheduling variation. The observed medians are 97.38% QPS and 105.80% P99 relative to baseline, so this sample does **not** clear the strict P1 release gate. It is diagnostic evidence only, not a release approval. + +P1 still needs the full transport/concurrency matrix and BenchmarkDotNet allocation comparison on a stable release host. P2-P4 need their corresponding direct-child comparison, cross-slot writer-pressure, and 2/8/16-slot resource runs before a release can claim all Issue 10 performance gates. Functional Release build, Unit, Generator, Integration, PackageSmoke, 120-second Chaos, and NativeAOT smoke are verified separately. diff --git a/src/SharpLink.Abstractions/ISharpLinkMultiClusterClient.cs b/src/SharpLink.Abstractions/ISharpLinkMultiClusterClient.cs new file mode 100644 index 0000000..657bcbe --- /dev/null +++ b/src/SharpLink.Abstractions/ISharpLinkMultiClusterClient.cs @@ -0,0 +1,48 @@ +using System.Reflection; + +namespace SharpLink.Abstractions; + +/// +/// Coordinates isolated SharpLink clients and routes a contract once while its proxy is created. +/// Subsequent RPC calls execute directly through the selected child client. +/// +public interface ISharpLinkMultiClusterClient : IAsyncDisposable +{ + /// Gets the aggregate coordinator lifecycle state. + SharpLinkMultiClusterState State { get; } + + /// Connects every required cluster slot. + ValueTask ConnectAsync(CancellationToken cancellationToken = default); + + /// Stops every cluster slot and releases coordinator-owned state. + ValueTask StopAsync(CancellationToken cancellationToken = default); + + /// Creates a proxy by looking up the contract's cluster route exactly once. + TContract Get() where TContract : IService; + + /// Gets the lifecycle state of one configured cluster slot. + SharpLinkConnectionState GetClusterState(SharpLinkClusterKey cluster); + + /// Runs a health check only against the specified cluster slot. + ValueTask CheckHealthAsync( + SharpLinkClusterKey cluster, + CancellationToken cancellationToken = default); + + /// Registers an assembly's generated artifacts in the specified cluster slot. + SharpLinkAssemblyRegistrationResult RegisterAssembly(SharpLinkClusterKey cluster, Assembly assembly); + + /// Drains and unregisters an assembly from its specified cluster slot. + ValueTask UnregisterAssemblyAsync( + SharpLinkClusterKey cluster, + Assembly assembly, + TimeSpan gracefulTimeout, + CancellationToken cancellationToken = default); + + /// Replaces an assembly registration inside its specified cluster slot. + ValueTask ReplaceAssemblyAsync( + SharpLinkClusterKey cluster, + Assembly oldAssembly, + Assembly newAssembly, + TimeSpan gracefulTimeout, + CancellationToken cancellationToken = default); +} diff --git a/src/SharpLink.Abstractions/SharpLinkClusterKey.cs b/src/SharpLink.Abstractions/SharpLinkClusterKey.cs new file mode 100644 index 0000000..b019aa8 --- /dev/null +++ b/src/SharpLink.Abstractions/SharpLinkClusterKey.cs @@ -0,0 +1,47 @@ +namespace SharpLink.Abstractions; + +/// Identifies one case-sensitive logical cluster in a multi-cluster client. +public readonly record struct SharpLinkClusterKey +{ + /// Creates a validated cluster key. + /// A one to 64 character ASCII cluster key. + public SharpLinkClusterKey(string value) + { + if (!IsValid(value)) + { + throw new ArgumentException( + "Cluster keys must contain 1 to 64 ASCII characters, start with a letter or digit, and then contain only letters, digits, '.', '_', or '-'.", + nameof(value)); + } + + Value = value; + } + + /// Gets the original, case-sensitive key value. + public string Value { get; } + + /// Returns whether a value satisfies the cluster-key grammar without normalization. + public static bool IsValid(string? value) + { + if (string.IsNullOrEmpty(value) || value.Length > 64 || !IsAlphaNumeric(value[0])) + return false; + + for (var index = 1; index < value.Length; index++) + { + var character = value[index]; + if (!IsAlphaNumeric(character) && character is not '.' and not '_' and not '-') + return false; + } + + return true; + } + + /// Returns the original key value. + public override string ToString() => Value ?? string.Empty; + + /// Creates a validated cluster key from a string. + public static implicit operator SharpLinkClusterKey(string value) => new(value); + + private static bool IsAlphaNumeric(char value) + => value is >= 'a' and <= 'z' or >= 'A' and <= 'Z' or >= '0' and <= '9'; +} diff --git a/src/SharpLink.Abstractions/SharpLinkGeneratedClusterRouteManifest.cs b/src/SharpLink.Abstractions/SharpLinkGeneratedClusterRouteManifest.cs new file mode 100644 index 0000000..5f0b23f --- /dev/null +++ b/src/SharpLink.Abstractions/SharpLinkGeneratedClusterRouteManifest.cs @@ -0,0 +1,94 @@ +using System.Reflection; +using System.Runtime.Loader; + +namespace SharpLink.Abstractions; + +/// Maps one contract-owning assembly to a multi-cluster slot. +public sealed record SharpLinkGeneratedClusterAssemblyRoute( + SharpLinkClusterKey Cluster, + Assembly ContractAssembly, + string ContractAssemblyIdentity); + +/// Provides source-generated static contract routes owned by an application assembly. +public interface ISharpLinkGeneratedClusterRouteManifest +{ + /// Gets the host assembly that declared the route attributes. + Assembly OwnerAssembly { get; } + + /// Gets deterministically ordered routes declared by the host assembly. + IReadOnlyList Routes { get; } +} + +/// +/// Stores bounded weak references to generated cluster route manifests without keeping collectible +/// AssemblyLoadContexts alive. +/// +public static class SharpLinkGeneratedClusterRouteCatalog +{ + private const int MaximumEntries = 16_384; + private static readonly Lock Gate = new(); + private static readonly List> Entries = []; + + /// Adds a generated route manifest without taking process-lifetime ownership of it. + public static void Register(ISharpLinkGeneratedClusterRouteManifest manifest) + { + ArgumentNullException.ThrowIfNull(manifest); + lock (Gate) + { + for (var index = Entries.Count - 1; index >= 0; index--) + { + if (!Entries[index].TryGetTarget(out var existing)) + { + Entries.RemoveAt(index); + continue; + } + + if (ReferenceEquals(existing, manifest)) + return; + } + + if (Entries.Count >= MaximumEntries) + { + throw new InvalidOperationException( + $"The generated cluster route catalog reached its safety limit of {MaximumEntries} live entries."); + } + + var weakManifest = new WeakReference(manifest); + Entries.Add(weakManifest); + var loadContext = AssemblyLoadContext.GetLoadContext(manifest.OwnerAssembly); + if (loadContext?.IsCollectible == true) + loadContext.Unloading += _ => Remove(weakManifest); + } + } + + /// Creates a strong point-in-time route-manifest snapshot for one coordinator. + public static IReadOnlyList CreateSnapshot() + { + lock (Gate) + { + var snapshot = new List(Entries.Count); + for (var index = Entries.Count - 1; index >= 0; index--) + { + if (Entries[index].TryGetTarget(out var manifest)) + snapshot.Add(manifest); + else + Entries.RemoveAt(index); + } + + return snapshot; + } + } + + private static void Remove(WeakReference manifest) + { + lock (Gate) + { + Entries.Remove(manifest); + for (var index = Entries.Count - 1; index >= 0; index--) + { + if (!Entries[index].TryGetTarget(out _)) + Entries.RemoveAt(index); + } + } + } +} diff --git a/src/SharpLink.Abstractions/SharpLinkMultiClusterState.cs b/src/SharpLink.Abstractions/SharpLinkMultiClusterState.cs new file mode 100644 index 0000000..41b5da0 --- /dev/null +++ b/src/SharpLink.Abstractions/SharpLinkMultiClusterState.cs @@ -0,0 +1,26 @@ +namespace SharpLink.Abstractions; + +/// Describes the aggregate lifecycle state of a multi-cluster client. +public enum SharpLinkMultiClusterState +{ + /// The client has been built but has not begun connecting. + Created, + + /// All required cluster slots are being connected. + Connecting, + + /// Every configured cluster slot is ready. + Ready, + + /// At least one slot is unavailable after a successful initial connection. + Degraded, + + /// Stop has begun and new dynamic registrations are rejected. + Draining, + + /// All owned cluster slots have stopped. + Stopped, + + /// An initial connection or coordinator operation failed irrecoverably. + Faulted +} diff --git a/src/SharpLink.Client/SharpClientBuilder.cs b/src/SharpLink.Client/SharpClientBuilder.cs index e6fad22..afe12fc 100644 --- a/src/SharpLink.Client/SharpClientBuilder.cs +++ b/src/SharpLink.Client/SharpClientBuilder.cs @@ -7,6 +7,7 @@ public class SharpClientBuilder private IClientTransportFactory? _transport; private IEnumerable? _endpoints; + private SharpLinkEndpoint[]? _preflightEndpointSnapshot; private SharpLinkEndpointTransportFactory? _endpointTransportFactory; private ISharpLinkEndpointResolver? _endpointResolver; private SharpLinkEndpointTransportFactory? _resolverTransportFactory; @@ -185,6 +186,7 @@ public SharpClientBuilder UseEndpoint( ArgumentNullException.ThrowIfNull(endpoint); ArgumentNullException.ThrowIfNull(transportFactory); _endpoints = [endpoint]; + _preflightEndpointSnapshot = null; _endpointTransportFactory = transportFactory; return this; } @@ -197,6 +199,7 @@ public SharpClientBuilder UseEndpoints( SharpLinkEndpointTransportFactory transportFactory) { _endpoints = endpoints ?? throw new ArgumentNullException(nameof(endpoints)); + _preflightEndpointSnapshot = null; _endpointTransportFactory = transportFactory ?? throw new ArgumentNullException(nameof(transportFactory)); return this; } @@ -332,15 +335,45 @@ public SharpClientBuilder UseCircuitBreaker(ActionBuilds a normal client using the complete generated-manifest catalog. + public ISharpLinkClient Build() => BuildCore(staticManifests: null); + + internal int GetConfiguredMaximumConnections() { + if (_endpointResolver is not null) + return _cluster.MaxConnections; + if (_endpoints is not null) + { + var endpoints = CreateEndpointSnapshot(_endpoints, allowEmpty: false); + _preflightEndpointSnapshot = endpoints; + if (endpoints.Length == 1) + return GetFixedConnectionBudget(); + + var cluster = _cluster.CloneValidated(endpoints.Length); + return Math.Min(cluster.MaxConnections, + checked(endpoints.Length * cluster.MaxConnectionsPerEndpoint)); + } + return GetFixedConnectionBudget(); + } + + // Multi-cluster construction supplies a filtered immutable manifest snapshot here. Keeping this + // decision at construction time preserves the ordinary client's hot path unchanged. + internal ISharpLinkClient BuildCore(IReadOnlyList? staticManifests) + { + // Multi-cluster preflight enumerates a static endpoint source to calculate its exact + // connection budget. Consume that one build-local snapshot, then clear it so later builds + // retain the normal builder behavior of taking a fresh topology snapshot. + var preflightEndpoints = staticManifests is null ? null : _preflightEndpointSnapshot; + _preflightEndpointSnapshot = null; var modeCount = (_transport is null ? 0 : 1) + (_endpoints is null ? 0 : 1) + (_endpointResolver is null ? 0 : 1); if (modeCount > 1) throw new InvalidOperationException("UseTransport, UseEndpoint(s), and UseEndpointResolver are mutually exclusive."); if (modeCount == 0) throw new InvalidOperationException("Transport, endpoint(s), or an endpoint resolver must be set before building the client."); - var runtimeContext = _runtimeContextBuilder.Build(); + var runtimeContext = staticManifests is null + ? _runtimeContextBuilder.Build() + : _runtimeContextBuilder.Build(staticManifests); var protocolOptions = runtimeContext.Protocol; if (_endpointResolver is not null) { @@ -352,12 +385,13 @@ public ISharpLinkClient Build() _resolverTransportFactory!, cluster, runtimeContext, - protocolOptions); + protocolOptions, + staticManifests); } if (_endpoints is not null) { - var endpoints = CreateEndpointSnapshot(_endpoints, allowEmpty: false); + var endpoints = preflightEndpoints ?? CreateEndpointSnapshot(_endpoints, allowEmpty: false); if (endpoints.Length == 1) { if (_clusterConfigured) @@ -376,7 +410,8 @@ public ISharpLinkClient Build() runtimeContext, protocolOptions, singleEndpointPool, - fixedEndpoint: endpoints[0]); + fixedEndpoint: endpoints[0], + staticManifests: staticManifests); } catch { @@ -410,7 +445,7 @@ public ISharpLinkClient Build() endpoints[index], factory); } - return CreateClusterClient(configurations, cluster, runtimeContext, protocolOptions); + return CreateClusterClient(configurations, cluster, runtimeContext, protocolOptions, staticManifests); } catch (Exception buildException) { @@ -434,7 +469,8 @@ public ISharpLinkClient Build() if (fixedTransport is AnonymousPipeClientTransportFactory && connectionPool.MaxConnections != 1) throw new InvalidOperationException("Anonymous-pipe handle offers support exactly one client connection."); - return CreateFixedClient(fixedTransport, runtimeContext, protocolOptions, connectionPool); + return CreateFixedClient(fixedTransport, runtimeContext, protocolOptions, connectionPool, + staticManifests: staticManifests); } private ISharpLinkClient CreateFixedClient( @@ -442,7 +478,8 @@ private ISharpLinkClient CreateFixedClient( SharpLinkRuntimeContext runtimeContext, SharpLinkProtocolOptions protocolOptions, SharpLinkConnectionPoolOptions? connectionPool = null, - SharpLinkEndpoint? fixedEndpoint = null) + SharpLinkEndpoint? fixedEndpoint = null, + IReadOnlyList? staticManifests = null) { return new SharpLinkClient( transport, @@ -459,7 +496,8 @@ private ISharpLinkClient CreateFixedClient( fixedEndpoint: fixedEndpoint, retryOptions: CreateRetryOptions(), retryPolicy: _retryPolicy, - endpointAdmissionPolicy: CreateEndpointAdmissionPolicy() + endpointAdmissionPolicy: CreateEndpointAdmissionPolicy(), + staticManifests: staticManifests ); } @@ -467,7 +505,8 @@ private ISharpLinkClient CreateClusterClient( StaticEndpointConfiguration[] configurations, SharpLinkClusterOptions cluster, SharpLinkRuntimeContext runtimeContext, - SharpLinkProtocolOptions protocolOptions) + SharpLinkProtocolOptions protocolOptions, + IReadOnlyList? staticManifests) => new SharpLinkClient( configurations[0].TransportFactory, _heartbeatInterval, @@ -486,14 +525,16 @@ private ISharpLinkClient CreateClusterClient( _endpointSelector, retryOptions: CreateRetryOptions(), retryPolicy: _retryPolicy, - endpointAdmissionPolicy: CreateEndpointAdmissionPolicy()); + endpointAdmissionPolicy: CreateEndpointAdmissionPolicy(), + staticManifests: staticManifests); private ISharpLinkClient CreateDynamicClusterClient( ISharpLinkEndpointResolver resolver, SharpLinkEndpointTransportFactory transportFactory, SharpLinkClusterOptions cluster, SharpLinkRuntimeContext runtimeContext, - SharpLinkProtocolOptions protocolOptions) + SharpLinkProtocolOptions protocolOptions, + IReadOnlyList? staticManifests) => new SharpLinkClient( DynamicClusterTransportPlaceholder.Instance, _heartbeatInterval, @@ -513,7 +554,8 @@ private ISharpLinkClient CreateDynamicClusterClient( endpointSelector: _endpointSelector, retryOptions: CreateRetryOptions(), retryPolicy: _retryPolicy, - endpointAdmissionPolicy: CreateEndpointAdmissionPolicy()); + endpointAdmissionPolicy: CreateEndpointAdmissionPolicy(), + staticManifests: staticManifests); private SharpLinkRetryOptions? CreateRetryOptions() => _retryConfigured ? _retry.CloneValidated() : null; @@ -602,4 +644,15 @@ private SharpLinkConnectionPoolOptions CreateConnectionPoolSnapshot(SharpLinkRun MaxConnections = Math.Max(1, maxConnections) }.CloneValidated(); } + + private int GetFixedConnectionBudget() + { + if (_connectionPoolConfigured) + return _connectionPool.CloneValidated().MaxConnections; + + var runtimeContext = _runtimeContextBuilder.Build(includeGeneratedAssemblyCatalog: false); + return runtimeContext.Options.PerformanceProfile == SharpLinkPerformanceProfile.Throughput + ? Math.Max(1, Math.Min(Environment.ProcessorCount, 4)) + : 1; + } } diff --git a/src/SharpLink.Client/SharpLinkClient.Assemblies.cs b/src/SharpLink.Client/SharpLinkClient.Assemblies.cs index 1160da5..4d62b8c 100644 --- a/src/SharpLink.Client/SharpLinkClient.Assemblies.cs +++ b/src/SharpLink.Client/SharpLinkClient.Assemblies.cs @@ -483,6 +483,15 @@ private static ValueTask WaitForUnregisterAsy ? new ValueTask(operation.WaitAsync(cancellationToken)) : new ValueTask(operation); + internal bool IsDynamicAssemblyRegistered(Assembly assembly) + { + lock (_registryGate) + return _dynamicModules.ContainsKey(assembly); + } + + bool IDynamicAssemblyRegistrationInspector.IsDynamicAssemblyRegistered(Assembly assembly) + => IsDynamicAssemblyRegistered(assembly); + private IEnumerable EnumerateRegisteredManifests(SharpLinkDynamicModule[] modules) { for (var index = 0; index < _staticManifests.Count; index++) diff --git a/src/SharpLink.Client/SharpLinkClient.cs b/src/SharpLink.Client/SharpLinkClient.cs index 83edc99..c3991cb 100644 --- a/src/SharpLink.Client/SharpLinkClient.cs +++ b/src/SharpLink.Client/SharpLinkClient.cs @@ -3,15 +3,16 @@ namespace SharpLink.Client; -internal sealed partial class SharpLinkClient : IRpcChannel, ISharpLinkClient +internal sealed partial class SharpLinkClient : IRpcChannel, ISharpLinkClient, IDynamicAssemblyRegistrationInspector { private readonly IClientTransportFactory transportFactory; private readonly IEndpointClusterRuntime? _cluster; // Retained for endpoint-aware diagnostics without routing fixed calls through cluster selection. private readonly SharpLinkEndpoint? _fixedEndpoint; - private readonly SharpLinkRuntimeContext _runtimeContext = new SharpLinkRuntimeContextBuilder().Build(); - private readonly IReadOnlyList _staticManifests = - SharpLinkGeneratedAssemblyCatalog.CreateSnapshot(); + // A filtered multi-cluster child supplies its own context after construction. Do not snapshot + // the process-wide manifest catalog before that context is applied. + private readonly SharpLinkRuntimeContext _runtimeContext = SharpLinkRuntimeContext.Default; + private readonly IReadOnlyList _staticManifests; private FrozenDictionary _proxies = FrozenDictionary.Empty; private readonly Lock _registryGate = new(); @@ -61,9 +62,11 @@ private SharpLinkClient( SharpLinkEndpointTransportFactory? dynamicTransportFactory = null, SharpLinkRetryOptions? retryOptions = null, ISharpLinkRetryPolicy? retryPolicy = null, - ISharpLinkEndpointAdmissionPolicy? endpointAdmissionPolicy = null) + ISharpLinkEndpointAdmissionPolicy? endpointAdmissionPolicy = null, + IReadOnlyList? staticManifests = null) { this.transportFactory = transportFactory ?? throw new ArgumentNullException(nameof(transportFactory)); + _staticManifests = staticManifests ?? SharpLinkGeneratedAssemblyCatalog.CreateSnapshot(); _fixedEndpoint = fixedEndpoint; _retryOptions = retryOptions; _retryPolicy = retryPolicy; @@ -111,9 +114,10 @@ public SharpLinkClient( SharpLinkEndpointTransportFactory? dynamicTransportFactory = null, SharpLinkRetryOptions? retryOptions = null, ISharpLinkRetryPolicy? retryPolicy = null, - ISharpLinkEndpointAdmissionPolicy? endpointAdmissionPolicy = null) + ISharpLinkEndpointAdmissionPolicy? endpointAdmissionPolicy = null, + IReadOnlyList? staticManifests = null) : this(transportFactory, staticEndpoints, clusterOptions, loadBalancingStrategy, endpointSelector, fixedEndpoint, - dynamicResolver, dynamicTransportFactory, retryOptions, retryPolicy, endpointAdmissionPolicy) + dynamicResolver, dynamicTransportFactory, retryOptions, retryPolicy, endpointAdmissionPolicy, staticManifests) { ArgumentOutOfRangeException.ThrowIfLessThanOrEqual(heartbeatInterval, TimeSpan.Zero); ArgumentOutOfRangeException.ThrowIfLessThanOrEqual(heartbeatTimeout, TimeSpan.Zero); @@ -129,7 +133,7 @@ public SharpLinkClient( _heartbeatInterval = heartbeatInterval; _heartbeatTimeout = heartbeatTimeout; _authenticator = authenticator; - _runtimeContext = runtimeContext ?? new SharpLinkRuntimeContextBuilder().Build(); + _runtimeContext = runtimeContext ?? new SharpLinkRuntimeContextBuilder().Build(_staticManifests); _protocolOptions = (protocolOptions ?? _runtimeContext.Protocol).CloneValidated(); _rpcSessionFlushOptions = rpcSessionFlushOptions; _connectionPoolOptions = (connectionPoolOptions ?? new SharpLinkConnectionPoolOptions()).CloneValidated(); @@ -158,11 +162,12 @@ public SharpLinkClient( SharpLinkEndpointTransportFactory? dynamicTransportFactory = null, SharpLinkRetryOptions? retryOptions = null, ISharpLinkRetryPolicy? retryPolicy = null, - ISharpLinkEndpointAdmissionPolicy? endpointAdmissionPolicy = null) + ISharpLinkEndpointAdmissionPolicy? endpointAdmissionPolicy = null, + IReadOnlyList? staticManifests = null) : this(transportFactory, heartbeatInterval, heartbeatTimeout, requestTimeout, authenticator, protocolOptions, runtimeContext, rpcSessionFlushOptions, connectionPoolOptions, clientInterceptors, staticEndpoints, clusterOptions, loadBalancingStrategy, endpointSelector, fixedEndpoint, dynamicResolver, dynamicTransportFactory, - retryOptions, retryPolicy, endpointAdmissionPolicy) + retryOptions, retryPolicy, endpointAdmissionPolicy, staticManifests) { ArgumentNullException.ThrowIfNull(loggerFactory); _logger = loggerFactory.CreateLogger(); diff --git a/src/SharpLink.Client/SharpLinkMultiClusterClient.cs b/src/SharpLink.Client/SharpLinkMultiClusterClient.cs new file mode 100644 index 0000000..c99794b --- /dev/null +++ b/src/SharpLink.Client/SharpLinkMultiClusterClient.cs @@ -0,0 +1,560 @@ +using System.Reflection; +using System.Runtime.ExceptionServices; + +namespace SharpLink.Client; + +internal sealed class SharpLinkMultiClusterClient : ISharpLinkMultiClusterClient +{ + private readonly SharpLinkMultiClusterOptions _options; + private readonly Lock _gate = new(); + private readonly CancellationTokenSource _shutdown = new(); + private FrozenDictionary _clusters; + private FrozenDictionary _routes; + private readonly List _dynamicRegistrations = []; + private Task? _connectTask; + private Task? _stopTask; + private int _state = (int)SharpLinkMultiClusterState.Created; + + internal SharpLinkMultiClusterClient( + SharpLinkMultiClusterOptions options, + FrozenDictionary clusters, + FrozenDictionary routes, + IReadOnlyList routeManifestSnapshot) + { + _ = routeManifestSnapshot; + _options = options; + _clusters = clusters; + _routes = routes; + } + + public SharpLinkMultiClusterState State + { + get + { + var state = (SharpLinkMultiClusterState)Volatile.Read(ref _state); + if (state is not SharpLinkMultiClusterState.Ready and not SharpLinkMultiClusterState.Degraded) + return state; + + var slots = Volatile.Read(ref _clusters); + if (slots.Count == 0) + return state; + var ready = slots.Values.Count(static slot => slot.Client.State == SharpLinkConnectionState.Ready); + return ready == slots.Count ? SharpLinkMultiClusterState.Ready : SharpLinkMultiClusterState.Degraded; + } + } + + public ValueTask ConnectAsync(CancellationToken cancellationToken = default) + { + Task operation; + lock (_gate) + { + var state = (SharpLinkMultiClusterState)_state; + if (state == SharpLinkMultiClusterState.Ready) + return ValueTask.CompletedTask; + if (state is SharpLinkMultiClusterState.Draining or SharpLinkMultiClusterState.Stopped or SharpLinkMultiClusterState.Faulted) + return ValueTask.FromException(new InvalidOperationException($"Multi-cluster client state '{state}' cannot connect.")); + + _connectTask ??= ConnectCoreAsync(); + operation = _connectTask; + } + + return cancellationToken.CanBeCanceled + ? new ValueTask(operation.WaitAsync(cancellationToken)) + : new ValueTask(operation); + } + + public ValueTask StopAsync(CancellationToken cancellationToken = default) + { + Task operation; + lock (_gate) + { + _stopTask ??= StopCoreAsync(); + operation = _stopTask; + } + return cancellationToken.CanBeCanceled + ? new ValueTask(operation.WaitAsync(cancellationToken)) + : new ValueTask(operation); + } + + public ValueTask DisposeAsync() => StopAsync(); + + public TContract Get() where TContract : IService + { + var state = State; + if (state is SharpLinkMultiClusterState.Draining or SharpLinkMultiClusterState.Stopped or SharpLinkMultiClusterState.Faulted) + throw new InvalidOperationException($"Multi-cluster client state '{state}' does not create proxies."); + + if (Volatile.Read(ref _routes).TryGetValue(typeof(TContract), out var route)) + return route.Slot.Client.Get(); + + throw new InvalidOperationException($"Proxy for service interface {typeof(TContract).FullName} is not routed to a cluster."); + } + + public SharpLinkConnectionState GetClusterState(SharpLinkClusterKey cluster) + => GetSlot(cluster).Client.State; + + public ValueTask CheckHealthAsync( + SharpLinkClusterKey cluster, + CancellationToken cancellationToken = default) + => GetSlot(cluster).Client.CheckHealthAsync(cancellationToken); + + public SharpLinkAssemblyRegistrationResult RegisterAssembly(SharpLinkClusterKey cluster, Assembly assembly) + { + if (assembly is null) + return SharpLinkAssemblyManifestLoader.TryLoad(null, out _); + + SharpLinkClusterSlot slot; + lock (_gate) + { + var state = (SharpLinkMultiClusterState)_state; + if (state is SharpLinkMultiClusterState.Draining or SharpLinkMultiClusterState.Stopped or SharpLinkMultiClusterState.Faulted) + { + return Failure(SharpLinkAssemblyRegistrationErrorCode.InvalidObjectState, + $"Multi-cluster client state '{state}' does not accept runtime assembly registration.", assembly); + } + + slot = GetSlot(cluster); + } + if (!slot.AllowDynamicContracts) + return Failure(SharpLinkAssemblyRegistrationErrorCode.InvalidObjectState, + $"Cluster '{cluster}' does not allow dynamic contract registration.", assembly); + + var loaded = SharpLinkAssemblyManifestLoader.TryLoad(assembly, out var manifest); + if (!loaded.Succeeded) + return loaded; + + lock (_gate) + { + var state = (SharpLinkMultiClusterState)_state; + if (state is SharpLinkMultiClusterState.Draining or SharpLinkMultiClusterState.Stopped or SharpLinkMultiClusterState.Faulted) + { + return Failure(SharpLinkAssemblyRegistrationErrorCode.InvalidObjectState, + $"Multi-cluster client state '{state}' does not accept runtime assembly registration.", assembly); + } + if (_dynamicRegistrations.Any(registration => ReferenceEquals(registration.Assembly, assembly) && + ReferenceEquals(registration.Slot, slot))) + { + return Failure(SharpLinkAssemblyRegistrationErrorCode.DuplicateAssembly, + $"Assembly '{assembly.FullName}' is already registered in cluster '{cluster}'.", assembly); + } + if (manifest!.Contracts.Count > 0 && _dynamicRegistrations.Any(registration => + ReferenceEquals(registration.Assembly, assembly) && registration.Manifest.Contracts.Count > 0)) + { + return Failure(SharpLinkAssemblyRegistrationErrorCode.ContractConflict, + $"Contract-owning assembly '{assembly.FullName}' is already routed to another cluster.", assembly); + } + + var currentRoutes = Volatile.Read(ref _routes); + foreach (var contract in manifest.Contracts) + { + if (currentRoutes.ContainsKey(contract.ContractType) || + currentRoutes.Values.Any(route => route.ContractId == contract.ContractId) || + _dynamicRegistrations.Any(registration => registration.Manifest.Contracts.Any( + existingContract => existingContract.ContractId == contract.ContractId))) + { + return Failure(SharpLinkAssemblyRegistrationErrorCode.ContractConflict, + $"Contract '{contract.ContractName}' ({contract.ContractId}) is already routed to another assembly or cluster.", assembly); + } + } + + var childResult = slot.Client.RegisterAssembly(assembly); + if (!childResult.Succeeded) + return childResult; + + try + { + var nextRoutes = currentRoutes.ToDictionary(static pair => pair.Key, static pair => pair.Value); + foreach (var contract in manifest.Contracts) + { + nextRoutes.Add(contract.ContractType, new SharpLinkClusterRouteRegistration( + contract.ContractType, contract.ContractId, contract.Fingerprint, slot, assembly)); + } + Volatile.Write(ref _routes, nextRoutes.ToFrozenDictionary()); + _dynamicRegistrations.Add(new DynamicAssemblyRegistration(slot, assembly, manifest)); + return SharpLinkAssemblyRegistrationResult.Success(); + } + catch (Exception exception) when (exception is not OutOfMemoryException and not StackOverflowException) + { + _ = slot.Client.UnregisterAssemblyAsync(assembly, TimeSpan.Zero); + return Failure(SharpLinkAssemblyRegistrationErrorCode.InvalidManifest, + $"Cluster route publication failed after child registration: {exception.GetType().Name}: {exception.Message}", assembly); + } + } + } + + public ValueTask UnregisterAssemblyAsync( + SharpLinkClusterKey cluster, + Assembly assembly, + TimeSpan gracefulTimeout, + CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(assembly); + ArgumentOutOfRangeException.ThrowIfLessThan(gracefulTimeout, TimeSpan.Zero); + SharpLinkClusterSlot slot; + DynamicAssemblyRegistration? registration; + lock (_gate) + { + var state = (SharpLinkMultiClusterState)_state; + if (state is SharpLinkMultiClusterState.Draining or SharpLinkMultiClusterState.Stopped or SharpLinkMultiClusterState.Faulted) + return ValueTask.FromResult(new SharpLinkAssemblyUnregisterResult { ReferencesReleased = false }); + + slot = GetSlot(cluster); + registration = _dynamicRegistrations.FirstOrDefault(candidate => + ReferenceEquals(candidate.Slot, slot) && ReferenceEquals(candidate.Assembly, assembly)); + if (registration is null) + return ValueTask.FromResult(new SharpLinkAssemblyUnregisterResult { ReferencesReleased = false }); + + var nextRoutes = Volatile.Read(ref _routes) + .Where(pair => !ReferenceEquals(pair.Value.OwnerAssembly, assembly)) + .ToDictionary(static pair => pair.Key, static pair => pair.Value) + .ToFrozenDictionary(); + Volatile.Write(ref _routes, nextRoutes); + } + + var operation = CompleteUnregisterAsync(slot, registration!, gracefulTimeout); + ObserveBackgroundFailure(operation); + return WaitForOperationAsync(operation, cancellationToken); + } + + public ValueTask ReplaceAssemblyAsync( + SharpLinkClusterKey cluster, + Assembly oldAssembly, + Assembly newAssembly, + TimeSpan gracefulTimeout, + CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(oldAssembly); + ArgumentNullException.ThrowIfNull(newAssembly); + ArgumentOutOfRangeException.ThrowIfLessThan(gracefulTimeout, TimeSpan.Zero); + SharpLinkClusterSlot slot; + lock (_gate) + { + var state = (SharpLinkMultiClusterState)_state; + if (state is SharpLinkMultiClusterState.Draining or SharpLinkMultiClusterState.Stopped or SharpLinkMultiClusterState.Faulted) + { + return ValueTask.FromResult(SharpLinkAssemblyReplacementResult.Failure(Error( + SharpLinkAssemblyRegistrationErrorCode.InvalidObjectState, + $"Multi-cluster client state '{state}' does not accept runtime assembly replacement.", newAssembly))); + } + + slot = GetSlot(cluster); + } + var loaded = SharpLinkAssemblyManifestLoader.TryLoad(newAssembly, out var newManifest); + if (!loaded.Succeeded) + return ValueTask.FromResult(SharpLinkAssemblyReplacementResult.Failure(loaded.Error!)); + + DynamicAssemblyRegistration? registration; + lock (_gate) + { + var state = (SharpLinkMultiClusterState)_state; + if (state is SharpLinkMultiClusterState.Draining or SharpLinkMultiClusterState.Stopped or SharpLinkMultiClusterState.Faulted) + { + return ValueTask.FromResult(SharpLinkAssemblyReplacementResult.Failure(Error( + SharpLinkAssemblyRegistrationErrorCode.InvalidObjectState, + $"Multi-cluster client state '{state}' does not accept runtime assembly replacement.", newAssembly))); + } + + registration = _dynamicRegistrations.FirstOrDefault(candidate => + ReferenceEquals(candidate.Slot, slot) && ReferenceEquals(candidate.Assembly, oldAssembly)); + if (registration is null) + { + return ValueTask.FromResult(SharpLinkAssemblyReplacementResult.Failure(Error( + SharpLinkAssemblyRegistrationErrorCode.InvalidObjectState, + $"Assembly '{oldAssembly.FullName}' is not registered in cluster '{cluster}'.", newAssembly))); + } + + var oldIds = registration.Manifest.Contracts.Select(static contract => contract.ContractId).Order().ToArray(); + var newIds = newManifest!.Contracts.Select(static contract => contract.ContractId).Order().ToArray(); + if (!oldIds.SequenceEqual(newIds)) + { + return ValueTask.FromResult(SharpLinkAssemblyReplacementResult.Failure(Error( + SharpLinkAssemblyRegistrationErrorCode.ContractConflict, + "Replacement assemblies must preserve the exact ContractId set within the same cluster.", newAssembly))); + } + + } + + var operation = CompleteReplacementAsync( + slot, registration!, newAssembly, newManifest!, gracefulTimeout); + ObserveBackgroundFailure(operation); + return WaitForOperationAsync(operation, cancellationToken); + } + + private async Task ConnectCoreAsync() + { + Volatile.Write(ref _state, (int)SharpLinkMultiClusterState.Connecting); + using var attempts = CancellationTokenSource.CreateLinkedTokenSource(_shutdown.Token); + try + { + await Parallel.ForEachAsync( + Volatile.Read(ref _clusters).Values, + new ParallelOptions { CancellationToken = attempts.Token, MaxDegreeOfParallelism = _options.MaxConcurrentClusterConnects }, + static async (slot, token) => await slot.Client.ConnectAsync(token).ConfigureAwait(false)).ConfigureAwait(false); + _ = Interlocked.CompareExchange( + ref _state, + (int)SharpLinkMultiClusterState.Ready, + (int)SharpLinkMultiClusterState.Connecting); + } + catch (Exception connectException) + { + attempts.Cancel(); + var failures = new List { connectException }; + await StopSlotsAsync(Volatile.Read(ref _clusters).Values, failures).ConfigureAwait(false); + // StopAsync owns the terminal transition. A connect completion may only replace the + // original Connecting state, never Draining or Stopped. + _ = Interlocked.CompareExchange( + ref _state, + (int)SharpLinkMultiClusterState.Faulted, + (int)SharpLinkMultiClusterState.Connecting); + if (failures.Count == 1) + ExceptionDispatchInfo.Capture(connectException).Throw(); + throw new AggregateException(failures); + } + } + + private async Task StopCoreAsync() + { + Volatile.Write(ref _state, (int)SharpLinkMultiClusterState.Draining); + _shutdown.Cancel(); + var slots = Volatile.Read(ref _clusters).Values.ToArray(); + var failures = new List(); + await StopSlotsAsync(slots, failures).ConfigureAwait(false); + lock (_gate) + { + Volatile.Write(ref _routes, FrozenDictionary.Empty); + Volatile.Write(ref _clusters, FrozenDictionary.Empty); + _dynamicRegistrations.Clear(); + } + _shutdown.Dispose(); + Volatile.Write(ref _state, (int)SharpLinkMultiClusterState.Stopped); + if (failures.Count == 1) + ExceptionDispatchInfo.Capture(failures[0]).Throw(); + if (failures.Count > 1) + throw new AggregateException(failures); + } + + private static async Task StopSlotsAsync(IEnumerable slots, List failures) + { + foreach (var slot in slots) + { + try { await slot.Client.StopAsync().ConfigureAwait(false); } + catch (Exception exception) { failures.Add(exception); } + } + } + + private async Task CompleteUnregisterAsync( + SharpLinkClusterSlot slot, + DynamicAssemblyRegistration registration, + TimeSpan gracefulTimeout) + { + SharpLinkAssemblyUnregisterResult result; + try + { + result = await slot.Client.UnregisterAssemblyAsync( + registration.Assembly, gracefulTimeout).ConfigureAwait(false); + } + catch + { + RestoreRoutesAfterRejectedUnregister(registration); + throw; + } + if (result.ReferencesReleased) + { + lock (_gate) + _dynamicRegistrations.Remove(registration); + } + else + { + ObserveBackgroundFailure(CompleteDeferredUnregisterAsync(slot, registration)); + } + return result; + } + + private void RestoreRoutesAfterRejectedUnregister(DynamicAssemblyRegistration registration) + { + if (registration.Slot.Client is not IDynamicAssemblyRegistrationInspector inspector || + !inspector.IsDynamicAssemblyRegistered(registration.Assembly)) + { + return; + } + + lock (_gate) + { + var state = (SharpLinkMultiClusterState)_state; + if (state is SharpLinkMultiClusterState.Draining or SharpLinkMultiClusterState.Stopped || + !_dynamicRegistrations.Contains(registration)) + { + return; + } + + var nextRoutes = Volatile.Read(ref _routes) + .ToDictionary(static pair => pair.Key, static pair => pair.Value); + var routeIds = nextRoutes.Values + .Select(static route => route.ContractId) + .ToHashSet(); + foreach (var contract in registration.Manifest.Contracts) + { + if (nextRoutes.ContainsKey(contract.ContractType) || !routeIds.Add(contract.ContractId)) + { + throw new InvalidOperationException( + $"Cannot restore contract '{contract.ContractName}' ({contract.ContractId}) because its route is already owned by another assembly or cluster."); + } + nextRoutes.Add(contract.ContractType, new SharpLinkClusterRouteRegistration( + contract.ContractType, + contract.ContractId, + contract.Fingerprint, + registration.Slot, + registration.Assembly)); + } + Volatile.Write(ref _routes, nextRoutes.ToFrozenDictionary()); + } + } + + private async Task CompleteDeferredUnregisterAsync( + SharpLinkClusterSlot slot, + DynamicAssemblyRegistration registration) + { + while ((SharpLinkMultiClusterState)Volatile.Read(ref _state) is not SharpLinkMultiClusterState.Stopped) + { + await Task.Delay(TimeSpan.FromMilliseconds(100)).ConfigureAwait(false); + if (slot.Client is IDynamicAssemblyRegistrationInspector inspector) + { + if (!inspector.IsDynamicAssemblyRegistered(registration.Assembly)) + { + lock (_gate) + _dynamicRegistrations.Remove(registration); + return; + } + continue; + } + + try + { + var result = await slot.Client.UnregisterAssemblyAsync( + registration.Assembly, TimeSpan.Zero).ConfigureAwait(false); + if (!result.ReferencesReleased && IsDynamicAssemblyStillRegistered(slot, registration.Assembly)) + continue; + lock (_gate) + _dynamicRegistrations.Remove(registration); + return; + } + catch + { + // The owning child performs its own final module cleanup during StopAsync. + return; + } + } + } + + private async Task CompleteReplacementAsync( + SharpLinkClusterSlot slot, + DynamicAssemblyRegistration registration, + Assembly newAssembly, + ISharpLinkGeneratedAssemblyManifest newManifest, + TimeSpan gracefulTimeout) + { + var childOperation = slot.Client.ReplaceAssemblyAsync( + registration.Assembly, newAssembly, gracefulTimeout); + + if (childOperation.IsCompleted) + { + var completedResult = await childOperation.ConfigureAwait(false); + if (completedResult.Succeeded) + PublishReplacement(registration, newAssembly, newManifest); + return completedResult; + } + + // SharpLinkClient publishes the replacement before returning its pending drain operation. + // Keep the coordinator route in the same state while old calls drain. + PublishReplacement(registration, newAssembly, newManifest); + return await childOperation.ConfigureAwait(false); + } + + private void PublishReplacement( + DynamicAssemblyRegistration registration, + Assembly newAssembly, + ISharpLinkGeneratedAssemblyManifest newManifest) + { + lock (_gate) + { + // A successful child replacement may race coordinator shutdown. StopAsync owns the + // child cleanup in that case; do not republish routes or retain the new assembly. + var state = (SharpLinkMultiClusterState)_state; + if (state is SharpLinkMultiClusterState.Draining or SharpLinkMultiClusterState.Stopped) + return; + + _dynamicRegistrations.Remove(registration); + _dynamicRegistrations.Add(new DynamicAssemblyRegistration(registration.Slot, newAssembly, newManifest)); + var nextRoutes = Volatile.Read(ref _routes).ToDictionary(static pair => pair.Key, static pair => pair.Value); + foreach (var contract in registration.Manifest.Contracts) + nextRoutes.Remove(contract.ContractType); + foreach (var contract in newManifest.Contracts) + { + nextRoutes[contract.ContractType] = new SharpLinkClusterRouteRegistration( + contract.ContractType, contract.ContractId, contract.Fingerprint, registration.Slot, newAssembly); + } + Volatile.Write(ref _routes, nextRoutes.ToFrozenDictionary()); + } + } + + private static bool IsDynamicAssemblyStillRegistered(SharpLinkClusterSlot slot, Assembly assembly) + => slot.Client is not IDynamicAssemblyRegistrationInspector client || + client.IsDynamicAssemblyRegistered(assembly); + + private static ValueTask WaitForOperationAsync(Task operation, CancellationToken cancellationToken) + => cancellationToken.CanBeCanceled + ? new ValueTask(operation.WaitAsync(cancellationToken)) + : new ValueTask(operation); + + private static void ObserveBackgroundFailure(Task task) + { + _ = task.ContinueWith( + static completedTask => _ = completedTask.Exception, + CancellationToken.None, + TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default); + } + + private SharpLinkClusterSlot GetSlot(SharpLinkClusterKey cluster) + { + if (!SharpLinkClusterKey.IsValid(cluster.Value)) + throw new ArgumentException("A valid non-default SharpLinkClusterKey is required.", nameof(cluster)); + if (Volatile.Read(ref _clusters).TryGetValue(cluster, out var slot)) + return slot; + throw new ArgumentException($"Cluster '{cluster}' is not configured.", nameof(cluster)); + } + + private static SharpLinkAssemblyRegistrationResult Failure( + SharpLinkAssemblyRegistrationErrorCode code, + string message, + Assembly assembly) + => SharpLinkAssemblyRegistrationResult.Failure(Error(code, message, assembly)); + + private static SharpLinkAssemblyRegistrationError Error( + SharpLinkAssemblyRegistrationErrorCode code, + string message, + Assembly assembly) + => new(code, message, IncomingAssembly: assembly.FullName); +} + +internal sealed record SharpLinkClusterSlot( + SharpLinkClusterKey Key, + ISharpLinkClient Client, + bool AllowDynamicContracts); + +internal sealed record SharpLinkClusterRouteRegistration( + Type ContractType, + long ContractId, + string Fingerprint, + SharpLinkClusterSlot Slot, + Assembly OwnerAssembly); + +internal sealed record DynamicAssemblyRegistration( + SharpLinkClusterSlot Slot, + Assembly Assembly, + ISharpLinkGeneratedAssemblyManifest Manifest); + +internal interface IDynamicAssemblyRegistrationInspector +{ + bool IsDynamicAssemblyRegistered(Assembly assembly); +} diff --git a/src/SharpLink.Client/SharpLinkMultiClusterClientBuilder.cs b/src/SharpLink.Client/SharpLinkMultiClusterClientBuilder.cs new file mode 100644 index 0000000..bf8463e --- /dev/null +++ b/src/SharpLink.Client/SharpLinkMultiClusterClientBuilder.cs @@ -0,0 +1,308 @@ +using System.Reflection; +using System.Runtime.CompilerServices; +using System.Runtime.Loader; + +namespace SharpLink.Client; + +/// Builds a coordinator that routes generated contracts to isolated child clients. +public sealed class SharpLinkMultiClusterClientBuilder +{ + private readonly SharpLinkMultiClusterOptions _options = new(); + private readonly Dictionary _clusters = []; + + /// Creates a multi-cluster client builder. + public static SharpLinkMultiClusterClientBuilder Create() => new(); + + /// Configures global multi-cluster limits. + public SharpLinkMultiClusterClientBuilder Configure(Action configure) + { + ArgumentNullException.ThrowIfNull(configure); + configure(_options); + return this; + } + + /// Adds a cluster slot that must have at least one static contract route. + public SharpLinkMultiClusterClientBuilder AddCluster( + SharpLinkClusterKey cluster, + Action configure) + => AddCluster(cluster, configure, configureSlot: null); + + /// Adds a cluster slot and configures whether it can accept dynamic-only contracts. + public SharpLinkMultiClusterClientBuilder AddCluster( + SharpLinkClusterKey cluster, + Action configure, + Action? configureSlot) + { + ValidateCluster(cluster); + ArgumentNullException.ThrowIfNull(configure); + if (_clusters.ContainsKey(cluster)) + throw new InvalidOperationException($"Cluster '{cluster}' has already been configured."); + + var slotOptions = new SharpLinkMultiClusterSlotOptions(); + configureSlot?.Invoke(slotOptions); + var child = SharpClientBuilder.Create(); + configure(child); + _clusters.Add(cluster, new ClusterConfiguration(cluster, child, slotOptions.AllowDynamicContracts)); + return this; + } + + /// Builds the coordinator without opening network connections. + public ISharpLinkMultiClusterClient Build() + { + var options = _options.CloneValidated(); + if (_clusters.Count == 0) + throw new InvalidOperationException("At least one cluster slot must be configured."); + if (_clusters.Count > options.MaxClusters) + throw new InvalidOperationException($"Configured cluster count exceeds MaxClusters ({options.MaxClusters})."); + + var configuredConnections = 0; + foreach (var configuration in _clusters.Values) + { + configuredConnections = checked(configuredConnections + configuration.Builder.GetConfiguredMaximumConnections()); + } + if (configuredConnections > options.MaxTotalConfiguredConnections) + { + throw new InvalidOperationException( + $"Configured child connection budget ({configuredConnections}) exceeds MaxTotalConfiguredConnections ({options.MaxTotalConfiguredConnections})."); + } + + var routeManifestSnapshot = SharpLinkGeneratedClusterRouteCatalog.CreateSnapshot(); + var configuredRoutes = routeManifestSnapshot + .SelectMany(static manifest => manifest.Routes) + .Where(route => _clusters.ContainsKey(route.Cluster)) + .ToArray(); + var routedAssemblies = new HashSet(ReferenceEqualityComparer.Instance); + foreach (var route in configuredRoutes) + routedAssemblies.Add(route.ContractAssembly); + var manifestByAssembly = LoadRoutedManifestGraph(routedAssemblies); + + var manifestsByCluster = _clusters.Keys.ToDictionary( + static key => key, + static _ => new Dictionary(ReferenceEqualityComparer.Instance)); + var assemblyOwners = new Dictionary(ReferenceEqualityComparer.Instance); + foreach (var route in configuredRoutes) + { + if (!manifestByAssembly.TryGetValue(route.ContractAssembly, out var contractManifest)) + { + throw new InvalidOperationException( + $"Static route '{route.ContractAssemblyIdentity}' does not reference a compatible generated contract manifest."); + } + if (contractManifest.Contracts.Count == 0) + { + throw new InvalidOperationException( + $"Static route '{route.ContractAssemblyIdentity}' must reference an assembly that owns at least one generated contract."); + } + if (assemblyOwners.TryGetValue(route.ContractAssembly, out var existingCluster)) + { + if (existingCluster != route.Cluster) + { + throw new InvalidOperationException( + $"Contract assembly '{route.ContractAssemblyIdentity}' is routed to both '{existingCluster}' and '{route.Cluster}'."); + } + continue; + } + + assemblyOwners.Add(route.ContractAssembly, route.Cluster); + AddManifestClosure(contractManifest, route.Cluster, manifestsByCluster, manifestByAssembly); + } + + foreach (var configuration in _clusters.Values) + { + if (manifestsByCluster[configuration.Key].Values.All(static manifest => manifest.Contracts.Count == 0) && + !configuration.AllowDynamicContracts) + { + throw new InvalidOperationException( + $"Cluster '{configuration.Key}' has no static contract route. Configure AllowDynamicContracts to create a dynamic-only slot."); + } + } + + var createdSlots = new List(_clusters.Count); + try + { + foreach (var configuration in _clusters.Values) + { + var staticManifests = manifestsByCluster[configuration.Key].Values + .OrderBy(static manifest => manifest.OwnerAssembly.FullName, StringComparer.Ordinal) + .Select(manifest => IsRoutedToCluster(manifest, configuration.Key, assemblyOwners) + ? manifest + : new DependencyManifestView(manifest)) + .ToArray(); + var child = configuration.Builder.BuildCore(staticManifests); + createdSlots.Add(new SharpLinkClusterSlot(configuration.Key, child, configuration.AllowDynamicContracts)); + } + + var slots = createdSlots.ToFrozenDictionary(static slot => slot.Key); + var routes = BuildStaticRoutes(slots, assemblyOwners, manifestByAssembly); + return new SharpLinkMultiClusterClient(options, slots, routes, routeManifestSnapshot); + } + catch (Exception buildException) + { + var cleanupFailures = new List(); + for (var index = createdSlots.Count - 1; index >= 0; index--) + { + try { createdSlots[index].Client.DisposeAsync().AsTask().GetAwaiter().GetResult(); } + catch (Exception cleanupException) { cleanupFailures.Add(cleanupException); } + } + if (cleanupFailures.Count == 0) + throw; + cleanupFailures.Insert(0, buildException); + throw new AggregateException(cleanupFailures); + } + } + + /// Applies a logger factory to child builders that do not already have one. + public void UseLoggerFactoryIfUnset(ILoggerFactory loggerFactory) + { + ArgumentNullException.ThrowIfNull(loggerFactory); + foreach (var configuration in _clusters.Values) + configuration.Builder.UseLoggerFactoryIfUnset(loggerFactory); + } + + private static FrozenDictionary BuildStaticRoutes( + FrozenDictionary slots, + IReadOnlyDictionary assemblyOwners, + IReadOnlyDictionary manifestsByAssembly) + { + var routes = new Dictionary(); + var routesById = new Dictionary(); + foreach (var pair in assemblyOwners) + { + var slot = slots[pair.Value]; + var manifest = manifestsByAssembly[pair.Key]; + foreach (var contract in manifest.Contracts) + { + var registration = new SharpLinkClusterRouteRegistration( + contract.ContractType, + contract.ContractId, + contract.Fingerprint, + slot, + manifest.OwnerAssembly); + if (!routes.TryAdd(contract.ContractType, registration) || + !routesById.TryAdd(contract.ContractId, registration)) + { + throw new InvalidOperationException( + $"Contract '{contract.ContractName}' ({contract.ContractId}) is exposed by more than one multi-cluster slot."); + } + } + } + + return routes.ToFrozenDictionary(); + } + + private static Dictionary LoadRoutedManifestGraph( + IEnumerable routedAssemblies) + { + var manifestsByAssembly = new Dictionary(ReferenceEqualityComparer.Instance); + var pendingAssemblies = new Queue(routedAssemblies); + while (pendingAssemblies.TryDequeue(out var assembly)) + { + if (manifestsByAssembly.ContainsKey(assembly)) + continue; + + RuntimeHelpers.RunModuleConstructor(assembly.ManifestModule.ModuleHandle); + if (!TryGetRegisteredManifest(assembly, out var manifest)) + continue; + + SharpLinkClient.ValidateStaticManifestCompatibility(manifest); + manifestsByAssembly.Add(assembly, manifest); + foreach (var dependencyIdentity in manifest.Dependencies) + { + var dependencyAssembly = ResolveDependencyAssembly(assembly, dependencyIdentity); + if (dependencyAssembly is not null) + pendingAssemblies.Enqueue(dependencyAssembly); + } + } + + return manifestsByAssembly; + } + + private static bool TryGetRegisteredManifest( + Assembly assembly, + out ISharpLinkGeneratedAssemblyManifest manifest) + { + foreach (var candidate in SharpLinkGeneratedAssemblyCatalog.CreateSnapshot()) + { + if (ReferenceEquals(candidate.OwnerAssembly, assembly)) + { + manifest = candidate; + return true; + } + } + + manifest = null!; + return false; + } + + private static Assembly? ResolveDependencyAssembly(Assembly ownerAssembly, string dependencyIdentity) + { + var loadContext = AssemblyLoadContext.GetLoadContext(ownerAssembly) ?? AssemblyLoadContext.Default; + var loaded = loadContext.Assemblies.FirstOrDefault(assembly => + string.Equals(assembly.FullName, dependencyIdentity, StringComparison.Ordinal)); + if (loaded is not null) + return loaded; + + try + { + return loadContext.LoadFromAssemblyName(new AssemblyName(dependencyIdentity)); + } + catch (Exception exception) when (exception is not OutOfMemoryException and not StackOverflowException) + { + return null; + } + } + + private static void AddManifestClosure( + ISharpLinkGeneratedAssemblyManifest manifest, + SharpLinkClusterKey cluster, + IReadOnlyDictionary> manifestsByCluster, + IReadOnlyDictionary manifestsByAssembly) + { + var destination = manifestsByCluster[cluster]; + if (!destination.TryAdd(manifest.OwnerAssembly, manifest)) + return; + + foreach (var dependencyIdentity in manifest.Dependencies) + { + var dependencyAssembly = ResolveDependencyAssembly(manifest.OwnerAssembly, dependencyIdentity); + if (dependencyAssembly is null || !manifestsByAssembly.TryGetValue(dependencyAssembly, out var dependency)) + { + throw new InvalidOperationException( + $"Static route for '{manifest.OwnerAssembly.FullName}' is missing generated dependency '{dependencyIdentity}' in cluster '{cluster}'."); + } + AddManifestClosure(dependency, cluster, manifestsByCluster, manifestsByAssembly); + } + } + + private static void ValidateCluster(SharpLinkClusterKey cluster) + { + if (!SharpLinkClusterKey.IsValid(cluster.Value)) + throw new ArgumentException("A valid non-default SharpLinkClusterKey is required.", nameof(cluster)); + } + + private static bool IsRoutedToCluster( + ISharpLinkGeneratedAssemblyManifest manifest, + SharpLinkClusterKey cluster, + IReadOnlyDictionary assemblyOwners) + => assemblyOwners.TryGetValue(manifest.OwnerAssembly, out var owner) && owner == cluster; + + // A dependency can contribute codecs to multiple slots, but proxy descriptors become + // visible only when its contract-owning assembly is explicitly routed to that slot. + private sealed class DependencyManifestView(ISharpLinkGeneratedAssemblyManifest source) + : ISharpLinkGeneratedAssemblyManifest + { + public int ApiVersion => source.ApiVersion; + public int ProtocolVersion => source.ProtocolVersion; + public string GeneratorVersion => source.GeneratorVersion; + public Assembly OwnerAssembly => source.OwnerAssembly; + public string CompileTimeDescriptor => source.CompileTimeDescriptor; + public IReadOnlyList Contracts => []; + public IReadOnlyList Services => []; + public IReadOnlyList Codecs => source.Codecs; + public IReadOnlyList Dependencies => source.Dependencies; + } + + private sealed record ClusterConfiguration( + SharpLinkClusterKey Key, + SharpClientBuilder Builder, + bool AllowDynamicContracts); +} diff --git a/src/SharpLink.Client/SharpLinkMultiClusterOptions.cs b/src/SharpLink.Client/SharpLinkMultiClusterOptions.cs new file mode 100644 index 0000000..6ac8de0 --- /dev/null +++ b/src/SharpLink.Client/SharpLinkMultiClusterOptions.cs @@ -0,0 +1,38 @@ +namespace SharpLink.Client; + +/// Configures global resource limits for a multi-cluster client. +public sealed class SharpLinkMultiClusterOptions +{ + /// Gets or sets the maximum number of configured cluster slots. + public int MaxClusters { get; set; } = 16; + + /// Gets or sets the total configured connection budget across all slots. + public int MaxTotalConfiguredConnections { get; set; } = 64; + + /// Gets or sets the maximum number of slot connection attempts running concurrently. + public int MaxConcurrentClusterConnects { get; set; } = 4; + + internal SharpLinkMultiClusterOptions CloneValidated() + { + if (MaxClusters is < 1 or > 256) + throw new ArgumentOutOfRangeException(nameof(MaxClusters)); + if (MaxTotalConfiguredConnections is < 1 or > 16_384) + throw new ArgumentOutOfRangeException(nameof(MaxTotalConfiguredConnections)); + if (MaxConcurrentClusterConnects is < 1 or > 64) + throw new ArgumentOutOfRangeException(nameof(MaxConcurrentClusterConnects)); + + return new SharpLinkMultiClusterOptions + { + MaxClusters = MaxClusters, + MaxTotalConfiguredConnections = MaxTotalConfiguredConnections, + MaxConcurrentClusterConnects = MaxConcurrentClusterConnects + }; + } +} + +/// Configures one cluster slot in a multi-cluster client. +public sealed class SharpLinkMultiClusterSlotOptions +{ + /// Gets or sets whether a slot without a static contract route may accept dynamic contracts. + public bool AllowDynamicContracts { get; set; } +} diff --git a/src/SharpLink.Generator/AnalyzerReleases.Unshipped.md b/src/SharpLink.Generator/AnalyzerReleases.Unshipped.md index 4e1d6c0..00e35ad 100644 --- a/src/SharpLink.Generator/AnalyzerReleases.Unshipped.md +++ b/src/SharpLink.Generator/AnalyzerReleases.Unshipped.md @@ -39,3 +39,7 @@ SHARPLINK035 | SharpLink.Compatibility | Error | Existing RPC contract was removed SHARPLINK036 | SharpLink.Generator | Error | Contract Manifest output could not be written SHARPLINK037 | SharpLink.Compatibility | Error | Existing service route was removed + SHARPLINK038 | SharpLink.Generator | Error | Multi-cluster key is invalid + SHARPLINK039 | SharpLink.Generator | Error | Contract assembly has conflicting cluster routes + SHARPLINK040 | SharpLink.Generator | Error | Cluster route marker lacks generated manifest + SHARPLINK041 | SharpLink.Generator | Error | Multi-cluster route attribute is invalid diff --git a/src/SharpLink.Generator/RpcGenerator.ClusterRouteAnalysis.cs b/src/SharpLink.Generator/RpcGenerator.ClusterRouteAnalysis.cs new file mode 100644 index 0000000..492f96c --- /dev/null +++ b/src/SharpLink.Generator/RpcGenerator.ClusterRouteAnalysis.cs @@ -0,0 +1,131 @@ +namespace SharpLink.Generator; + +public partial class RpcGenerator +{ + private static ClusterRouteAnalysis AnalyzeClusterRoutes(Compilation compilation, CancellationToken cancellationToken) + { + var routes = ImmutableArray.CreateBuilder(); + var diagnostics = ImmutableArray.CreateBuilder(); + var owners = new Dictionary(SymbolEqualityComparer.Default); + + foreach (var attribute in compilation.Assembly.GetAttributes()) + { + cancellationToken.ThrowIfCancellationRequested(); + if (attribute.AttributeClass?.ToDisplayString() != ClusterContractAssemblyAttributeMetadataName) + continue; + + var location = attribute.ApplicationSyntaxReference?.GetSyntax(cancellationToken).GetLocation() ?? Location.None; + if (attribute.ConstructorArguments.Length != 2 || + attribute.ConstructorArguments[0].Value is not string cluster || + attribute.ConstructorArguments[1].Value is not INamedTypeSymbol marker) + { + diagnostics.Add(new ClusterRouteDiagnostic(InvalidClusterRouteAttributeRule, location, [])); + continue; + } + if (!IsValidClusterKey(cluster)) + { + diagnostics.Add(new ClusterRouteDiagnostic(InvalidClusterKeyRule, location, [cluster])); + continue; + } + if (!HasGeneratedManifest(marker.ContainingAssembly, compilation.Assembly)) + { + diagnostics.Add(new ClusterRouteDiagnostic( + MissingClusterRouteManifestRule, + location, + [marker.ContainingAssembly.Identity.ToString()])); + continue; + } + + var route = new ClusterRouteModel( + cluster, + marker.ContainingAssembly.Identity.ToString(), + marker.ToDisplayString(SymbolDisplayFormat.FullyQualifiedFormat)); + if (owners.TryGetValue(marker.ContainingAssembly, out var existing)) + { + if (!string.Equals(existing.Cluster, route.Cluster, StringComparison.Ordinal)) + { + diagnostics.Add(new ClusterRouteDiagnostic( + ConflictingClusterRouteRule, + location, + [route.AssemblyIdentity, existing.Cluster, route.Cluster])); + } + continue; + } + + owners.Add(marker.ContainingAssembly, route); + routes.Add(route); + } + + return new ClusterRouteAnalysis( + routes.OrderBy(static route => route.Cluster, StringComparer.Ordinal) + .ThenBy(static route => route.AssemblyIdentity, StringComparer.Ordinal) + .ToImmutableArray(), + diagnostics.ToImmutable()); + } + + private static bool HasGeneratedManifest(IAssemblySymbol assembly, IAssemblySymbol currentAssembly) + { + if (assembly.GetAttributes().Any(static attribute => + attribute.AttributeClass?.ToDisplayString() == + "SharpLink.Abstractions.SharpLinkGeneratedAssemblyManifestAttribute")) + { + return true; + } + + return SymbolEqualityComparer.Default.Equals(assembly, currentAssembly) && + AssemblyContainsRpcContract(assembly.GlobalNamespace); + } + + private static bool AssemblyContainsRpcContract(INamespaceSymbol @namespace) + { + foreach (var type in @namespace.GetTypeMembers()) + { + if (ContainsRpcContract(type)) + return true; + } + foreach (var child in @namespace.GetNamespaceMembers()) + { + if (AssemblyContainsRpcContract(child)) + return true; + } + return false; + } + + private static bool ContainsRpcContract(INamedTypeSymbol type) + { + if (type.TypeKind == TypeKind.Interface && HasRpcContractAttribute(type)) + return true; + return type.GetTypeMembers().Any(ContainsRpcContract); + } + + // Kept byte-for-byte equivalent to SharpLinkClusterKey.IsValid so generator diagnostics match runtime validation. + private static bool IsValidClusterKey(string value) + { + if (string.IsNullOrEmpty(value) || value.Length > 64 || !IsAsciiAlphaNumeric(value[0])) + return false; + for (var index = 1; index < value.Length; index++) + { + var character = value[index]; + if (!IsAsciiAlphaNumeric(character) && character is not '.' and not '_' and not '-') + return false; + } + return true; + } + + private static bool IsAsciiAlphaNumeric(char value) + => value is >= 'a' and <= 'z' or >= 'A' and <= 'Z' or >= '0' and <= '9'; + + private readonly record struct ClusterRouteModel( + string Cluster, + string AssemblyIdentity, + string MarkerTypeName); + + private readonly record struct ClusterRouteDiagnostic( + DiagnosticDescriptor Rule, + Location Location, + object[] Arguments); + + private readonly record struct ClusterRouteAnalysis( + ImmutableArray Routes, + ImmutableArray Diagnostics); +} diff --git a/src/SharpLink.Generator/RpcGenerator.ClusterRouteEmitter.cs b/src/SharpLink.Generator/RpcGenerator.ClusterRouteEmitter.cs new file mode 100644 index 0000000..f2d114f --- /dev/null +++ b/src/SharpLink.Generator/RpcGenerator.ClusterRouteEmitter.cs @@ -0,0 +1,47 @@ +namespace SharpLink.Generator; + +public partial class RpcGenerator +{ + private static string GenerateClusterRouteManifest(ImmutableArray routes) + { + var owner = string.Join("|", routes.Select(static route => + route.Cluster + ":" + route.AssemblyIdentity)); + var typeName = "__SharpLinkGeneratedClusterRouteManifest_" + Hashing.GetSha256(owner).Substring(0, 16); + var sb = new StringBuilder(); + sb.AppendLine("// "); + sb.AppendLine("#nullable enable"); + sb.AppendLine("using System.Collections.Generic;"); + sb.AppendLine("using System.Reflection;"); + sb.AppendLine("using System.Runtime.CompilerServices;"); + sb.AppendLine("using SharpLink.Abstractions;"); + sb.AppendLine(); + sb.AppendLine("namespace SharpLink.Generated;"); + sb.AppendLine(); + sb.AppendLine("[System.ComponentModel.EditorBrowsable(System.ComponentModel.EditorBrowsableState.Never)]"); + sb.AppendLine($"public sealed class {typeName} : ISharpLinkGeneratedClusterRouteManifest"); + sb.AppendLine("{"); + sb.AppendLine($" public static readonly {typeName} Instance = new();"); + sb.AppendLine($" public {typeName}() {{ }}"); + sb.AppendLine($" public Assembly OwnerAssembly => typeof({typeName}).Assembly;"); + sb.AppendLine(" public IReadOnlyList Routes => __routes;"); + sb.AppendLine(" private static readonly SharpLinkGeneratedClusterAssemblyRoute[] __routes = new SharpLinkGeneratedClusterAssemblyRoute[]"); + sb.AppendLine(" {"); + foreach (var route in routes) + { + sb.AppendLine(" new SharpLinkGeneratedClusterAssemblyRoute("); + sb.AppendLine($" new SharpLinkClusterKey(\"{EscapeString(route.Cluster)}\"),"); + sb.AppendLine($" typeof({route.MarkerTypeName}).Assembly,"); + sb.AppendLine($" \"{EscapeString(route.AssemblyIdentity)}\"),"); + } + sb.AppendLine(" };"); + sb.AppendLine("}"); + sb.AppendLine(); + sb.AppendLine($"internal static class {typeName}Initializer"); + sb.AppendLine("{"); + sb.AppendLine(" [ModuleInitializer]"); + sb.AppendLine(" internal static void Register()"); + sb.AppendLine($" => SharpLinkGeneratedClusterRouteCatalog.Register({typeName}.Instance);"); + sb.AppendLine("}"); + return sb.ToString(); + } +} diff --git a/src/SharpLink.Generator/RpcGenerator.Diagnostics.cs b/src/SharpLink.Generator/RpcGenerator.Diagnostics.cs index 2686e04..169c8b3 100644 --- a/src/SharpLink.Generator/RpcGenerator.Diagnostics.cs +++ b/src/SharpLink.Generator/RpcGenerator.Diagnostics.cs @@ -235,6 +235,38 @@ public partial class RpcGenerator private static readonly DiagnosticDescriptor ServiceRouteRemovedCompatibilityRule = CompatibilityRule( "SHARPLINK037", "Existing Service Route Was Removed"); + private static readonly DiagnosticDescriptor InvalidClusterKeyRule = new( + id: "SHARPLINK038", + title: "Multi-Cluster Key Is Invalid", + messageFormat: "Cluster key '{0}' must contain 1 to 64 ASCII characters, start with a letter or digit, and then contain only letters, digits, '.', '_', or '-'", + category: "SharpLink.Generator", + defaultSeverity: DiagnosticSeverity.Error, + isEnabledByDefault: true); + + private static readonly DiagnosticDescriptor ConflictingClusterRouteRule = new( + id: "SHARPLINK039", + title: "Contract Assembly Has Conflicting Cluster Routes", + messageFormat: "Contract assembly '{0}' is routed to both cluster '{1}' and cluster '{2}'", + category: "SharpLink.Generator", + defaultSeverity: DiagnosticSeverity.Error, + isEnabledByDefault: true); + + private static readonly DiagnosticDescriptor MissingClusterRouteManifestRule = new( + id: "SHARPLINK040", + title: "Cluster Route Marker Lacks Generated Manifest", + messageFormat: "Cluster route marker assembly '{0}' does not expose a compatible generated SharpLink manifest", + category: "SharpLink.Generator", + defaultSeverity: DiagnosticSeverity.Error, + isEnabledByDefault: true); + + private static readonly DiagnosticDescriptor InvalidClusterRouteAttributeRule = new( + id: "SHARPLINK041", + title: "Multi-Cluster Route Attribute Is Invalid", + messageFormat: "SharpLinkClusterContractAssembly requires a literal cluster key and a concrete marker type", + category: "SharpLink.Generator", + defaultSeverity: DiagnosticSeverity.Error, + isEnabledByDefault: true); + private static DiagnosticDescriptor CompatibilityRule(string id, string title) => new( id: id, diff --git a/src/SharpLink.Generator/RpcGenerator.cs b/src/SharpLink.Generator/RpcGenerator.cs index 62ed7dc..8d46639 100644 --- a/src/SharpLink.Generator/RpcGenerator.cs +++ b/src/SharpLink.Generator/RpcGenerator.cs @@ -6,6 +6,8 @@ public partial class RpcGenerator : IIncrementalGenerator private static readonly CultureInfo InvariantCulture = CultureInfo.InvariantCulture; private const string RpcContractAttributeMetadataName = "SharpLink.Sdk.RpcContractAttribute"; private const string RpcServiceAttributeMetadataName = "SharpLink.Sdk.RpcServiceAttribute"; + private const string ClusterContractAssemblyAttributeMetadataName = + "SharpLink.Sdk.SharpLinkClusterContractAssemblyAttribute"; public void Initialize(IncrementalGeneratorInitializationContext context) { @@ -36,6 +38,8 @@ public void Initialize(IncrementalGeneratorInitializationContext context) .Where(static diagnostic => diagnostic is not null); var staticRouteConflicts = context.CompilationProvider.Select(static (compilation, ct) => AnalyzeStaticRouteConflicts(compilation, ct)); + var clusterRoutes = context.CompilationProvider.Select(static (compilation, ct) => + AnalyzeClusterRoutes(compilation, ct)); var invalidMethods = context.SyntaxProvider.ForAttributeWithMetadataName( RpcContractAttributeMetadataName, @@ -213,6 +217,19 @@ public void Initialize(IncrementalGeneratorInitializationContext context) } }); + context.RegisterSourceOutput(clusterRoutes, static (spc, analysis) => + { + foreach (var diagnostic in analysis.Diagnostics) + spc.ReportDiagnostic(Diagnostic.Create(diagnostic.Rule, diagnostic.Location, diagnostic.Arguments)); + + if (!analysis.Routes.IsDefaultOrEmpty) + { + spc.AddSource( + "SharpLink.GeneratedClusterRouteManifest.g.cs", + SourceText.From(GenerateClusterRouteManifest(analysis.Routes), Encoding.UTF8)); + } + }); + context.RegisterSourceOutput(interfaces, (spc, model) => { var proxy = GenerateProxy(model!); diff --git a/src/SharpLink.Hosting/HostExtensions.cs b/src/SharpLink.Hosting/HostExtensions.cs index cc3e3d1..d53548c 100644 --- a/src/SharpLink.Hosting/HostExtensions.cs +++ b/src/SharpLink.Hosting/HostExtensions.cs @@ -35,5 +35,19 @@ public SharpClientBuilder AddSharpLinkClient(Action? configu services.AddHostedService(); return builder; } + + /// Adds one hosted multi-cluster coordinator without exposing individual child clients to DI. + public SharpLinkMultiClusterClientBuilder AddSharpLinkMultiClusterClient( + Action? configure = null) + { + var builder = SharpLinkMultiClusterClientBuilder.Create(); + configure?.Invoke(builder); + services.TryAddSingleton(builder); + services.TryAddSingleton(); + services.TryAddSingleton(static provider => + provider.GetRequiredService()); + services.AddHostedService(); + return builder; + } } } diff --git a/src/SharpLink.Hosting/ISharpLinkMultiClusterClientAccessor.cs b/src/SharpLink.Hosting/ISharpLinkMultiClusterClientAccessor.cs new file mode 100644 index 0000000..ebafd32 --- /dev/null +++ b/src/SharpLink.Hosting/ISharpLinkMultiClusterClientAccessor.cs @@ -0,0 +1,8 @@ +namespace SharpLink.Hosting; + +/// Provides the hosted multi-cluster client only after every required slot is ready. +public interface ISharpLinkMultiClusterClientAccessor +{ + /// Gets the published coordinator or waits for hosted startup to finish. + ValueTask GetClientAsync(CancellationToken cancellationToken = default); +} diff --git a/src/SharpLink.Hosting/SharpLinkMultiClusterClientAccessor.cs b/src/SharpLink.Hosting/SharpLinkMultiClusterClientAccessor.cs new file mode 100644 index 0000000..606eabe --- /dev/null +++ b/src/SharpLink.Hosting/SharpLinkMultiClusterClientAccessor.cs @@ -0,0 +1,67 @@ +namespace SharpLink.Hosting; + +internal sealed class SharpLinkMultiClusterClientAccessor : ISharpLinkMultiClusterClientAccessor +{ + private readonly Lock _gate = new(); + private readonly TaskCompletionSource _ready = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private ISharpLinkMultiClusterClient? _client; + private bool _stopped; + + public ValueTask GetClientAsync(CancellationToken cancellationToken = default) + { + Task task; + lock (_gate) + { + if (_stopped) + return ValueTask.FromException(Unavailable()); + + if (_client is { } client) + return ValueTask.FromResult(client); + + task = _ready.Task; + } + return cancellationToken.CanBeCanceled + ? new ValueTask(task.WaitAsync(cancellationToken)) + : new ValueTask(task); + } + + internal void SetClient(ISharpLinkMultiClusterClient client) + { + ArgumentNullException.ThrowIfNull(client); + lock (_gate) + { + if (_stopped) + throw Unavailable(); + if (_client is not null) + throw new InvalidOperationException("SharpLink multi-cluster client has already been published."); + + _client = client; + _ready.TrySetResult(client); + } + } + + internal void Fail(Exception exception) + { + ArgumentNullException.ThrowIfNull(exception); + lock (_gate) + { + _stopped = true; + _client = null; + _ready.TrySetException(exception); + } + } + + internal void Stop() + { + lock (_gate) + { + _stopped = true; + _client = null; + _ready.TrySetException(Unavailable()); + } + } + + private static InvalidOperationException Unavailable() + => new("SharpLink multi-cluster client is not available because the host has already stopped."); +} diff --git a/src/SharpLink.Hosting/SharpLinkMultiClusterClientHostedService.cs b/src/SharpLink.Hosting/SharpLinkMultiClusterClientHostedService.cs new file mode 100644 index 0000000..ee032d5 --- /dev/null +++ b/src/SharpLink.Hosting/SharpLinkMultiClusterClientHostedService.cs @@ -0,0 +1,41 @@ +namespace SharpLink.Hosting; + +internal sealed class SharpLinkMultiClusterClientHostedService( + SharpLinkMultiClusterClientBuilder builder, + SharpLinkMultiClusterClientAccessor accessor, + ILoggerFactory loggerFactory) : IHostedService, IAsyncDisposable +{ + private ISharpLinkMultiClusterClient? _client; + + public async Task StartAsync(CancellationToken cancellationToken) + { + try + { + builder.UseLoggerFactoryIfUnset(loggerFactory); + _client = builder.Build(); + await _client.ConnectAsync(cancellationToken).ConfigureAwait(false); + accessor.SetClient(_client); + } + catch (Exception exception) + { + accessor.Fail(exception); + await DisposeAsync().ConfigureAwait(false); + throw; + } + } + + public async Task StopAsync(CancellationToken cancellationToken) + { + accessor.Stop(); + var client = Interlocked.Exchange(ref _client, null); + if (client is not null) + await client.StopAsync(cancellationToken).ConfigureAwait(false); + } + + public async ValueTask DisposeAsync() + { + var client = Interlocked.Exchange(ref _client, null); + if (client is not null) + await client.DisposeAsync().ConfigureAwait(false); + } +} diff --git a/src/SharpLink.Runtime/SharpLinkRuntimeContext.cs b/src/SharpLink.Runtime/SharpLinkRuntimeContext.cs index 25754c5..88d8385 100644 --- a/src/SharpLink.Runtime/SharpLinkRuntimeContext.cs +++ b/src/SharpLink.Runtime/SharpLinkRuntimeContext.cs @@ -11,16 +11,13 @@ internal SharpLinkRuntimeContext( BufferWriterPoolOptions bufferPool, Func? resolver, IReadOnlyDictionary codecs, - bool includeGeneratedAssemblyCatalog) + IReadOnlyList generatedManifests) { _options = options.CloneValidated(); Concurrency = concurrency.CloneValidated(); var generatedFactories = new Dictionary( RpcGeneratedCodecRegistry.CreateSnapshot()); - var manifests = includeGeneratedAssemblyCatalog - ? SharpLinkGeneratedAssemblyCatalog.CreateSnapshot() - : []; - foreach (var manifest in manifests) + foreach (var manifest in generatedManifests) { foreach (var factory in manifest.Codecs) { @@ -130,14 +127,20 @@ public SharpLinkRuntimeContextBuilder AddCodec(IRpcCodec codec) /// Validates and freezes a new context. public SharpLinkRuntimeContext Build() - => Build(includeGeneratedAssemblyCatalog: true); + => Build(SharpLinkGeneratedAssemblyCatalog.CreateSnapshot()); internal SharpLinkRuntimeContext Build(bool includeGeneratedAssemblyCatalog) + => Build(includeGeneratedAssemblyCatalog + ? SharpLinkGeneratedAssemblyCatalog.CreateSnapshot() + : []); + + internal SharpLinkRuntimeContext Build(IReadOnlyList generatedManifests) { + ArgumentNullException.ThrowIfNull(generatedManifests); var options = _options.CloneValidated(); var concurrency = _concurrency.CloneValidated(); var bufferPool = _bufferPool.CloneValidated(); return new SharpLinkRuntimeContext(options, concurrency, bufferPool, _resolver, - new Dictionary(_codecs), includeGeneratedAssemblyCatalog); + new Dictionary(_codecs), generatedManifests); } } diff --git a/src/SharpLink.Sdk/SharpLinkClusterContractAssemblyAttribute.cs b/src/SharpLink.Sdk/SharpLinkClusterContractAssemblyAttribute.cs new file mode 100644 index 0000000..f022104 --- /dev/null +++ b/src/SharpLink.Sdk/SharpLinkClusterContractAssemblyAttribute.cs @@ -0,0 +1,21 @@ +namespace SharpLink.Sdk; + +/// Assigns all generated contracts in a marker assembly to one multi-cluster slot. +[AttributeUsage(AttributeTargets.Assembly, AllowMultiple = true)] +public sealed class SharpLinkClusterContractAssemblyAttribute : Attribute +{ + /// Creates a static contract-assembly route declaration. + /// The target case-sensitive cluster key. + /// A type in the contract-owning assembly. + public SharpLinkClusterContractAssemblyAttribute(string cluster, Type assemblyMarker) + { + Cluster = cluster ?? throw new ArgumentNullException(nameof(cluster)); + AssemblyMarker = assemblyMarker ?? throw new ArgumentNullException(nameof(assemblyMarker)); + } + + /// Gets the target cluster key. + public string Cluster { get; } + + /// Gets the type used only to locate the contract-owning assembly. + public Type AssemblyMarker { get; } +} diff --git a/test/SharpLink.AotContracts/SecondAotContract.cs b/test/SharpLink.AotContracts/SecondAotContract.cs new file mode 100644 index 0000000..2bc12ee --- /dev/null +++ b/test/SharpLink.AotContracts/SecondAotContract.cs @@ -0,0 +1,12 @@ +using System.Threading.Tasks; +using SharpLink.Abstractions; +using SharpLink.Sdk; + +namespace SharpLink.AotContracts; + +[RpcContract] +public interface ISecondAotService : IService +{ + [NonCancellable] + ValueTask MultiplyAsync(int value); +} diff --git a/test/SharpLink.AotContracts/SharpLink.AotContracts.csproj b/test/SharpLink.AotContracts/SharpLink.AotContracts.csproj new file mode 100644 index 0000000..a04c93e --- /dev/null +++ b/test/SharpLink.AotContracts/SharpLink.AotContracts.csproj @@ -0,0 +1,14 @@ + + + net10.0 + + + + + + + + + diff --git a/test/SharpLink.AotSmoke/Program.cs b/test/SharpLink.AotSmoke/Program.cs index bb6dab6..9446ad3 100644 --- a/test/SharpLink.AotSmoke/Program.cs +++ b/test/SharpLink.AotSmoke/Program.cs @@ -6,11 +6,15 @@ using System.Threading; using System.Threading.Tasks; using SharpLink.Abstractions; +using SharpLink.AotContracts; using SharpLink.Client; using SharpLink.Runtime; using SharpLink.Sdk; using SharpLink.Server; +[assembly: SharpLinkClusterContractAssembly("orders", typeof(SharpLink.AotSmoke.IAotService))] +[assembly: SharpLinkClusterContractAssembly("payments", typeof(ISecondAotService))] + namespace SharpLink.AotSmoke; public static class Program @@ -97,6 +101,8 @@ public static async Task Main(string[] args) try { await VerifyClientAsync(client, runToken).ConfigureAwait(false); + await using var multiClusterClient = CreateMultiClusterClient(useSharedMemory, sharedMemoryName, port); + await VerifyMultiClusterClientAsync(multiClusterClient, runToken).ConfigureAwait(false); Console.WriteLine($"AOT_SMOKE_PASS transport={(useSharedMemory ? "sharedmemory" : "tcp")}"); return 0; @@ -202,6 +208,52 @@ private static async Task VerifyClientAsync(ISharpLinkClient client, Cancellatio throw new Exception("unexpected pair result"); } + private static ISharpLinkMultiClusterClient CreateMultiClusterClient( + bool useSharedMemory, + string sharedMemoryName, + int port) + => SharpLinkMultiClusterClientBuilder.Create() + .AddCluster( + "orders", + child => ConfigureClientTransport(child, useSharedMemory, sharedMemoryName, port), + slot => slot.AllowDynamicContracts = true) + .AddCluster("payments", child => ConfigureClientTransport(child, useSharedMemory, sharedMemoryName, port)) + .Build(); + + private static void ConfigureClientTransport( + SharpClientBuilder builder, + bool useSharedMemory, + string sharedMemoryName, + int port) + { + builder.UseRuntime(ConfigureCompression); + if (useSharedMemory) + builder.UseSharedMemory(sharedMemoryName); + else + builder.UseTcp(IPAddress.Loopback.ToString(), port); + } + + private static async Task VerifyMultiClusterClientAsync( + ISharpLinkMultiClusterClient client, + CancellationToken cancellationToken) + { + VerifyRuntimeAssemblyBoundary(client); + await client.ConnectAsync(cancellationToken).ConfigureAwait(false); + + var ordersHealth = await client.CheckHealthAsync("orders", cancellationToken).ConfigureAwait(false); + var paymentsHealth = await client.CheckHealthAsync("payments", cancellationToken).ConfigureAwait(false); + if (ordersHealth.Status != SharpLinkHealthStatus.Ready || paymentsHealth.Status != SharpLinkHealthStatus.Ready) + throw new Exception("static multi-cluster health did not reach Ready"); + + var orders = client.Get(); + if (await orders.PingAsync().ConfigureAwait(false) != "pong") + throw new Exception("unexpected orders multi-cluster result"); + + var payments = client.Get(); + if (await payments.MultiplyAsync(21).ConfigureAwait(false) != 42) + throw new Exception("unexpected payments multi-cluster result"); + } + private static void VerifyRuntimeAssemblyBoundary(ISharpLinkClient client) { if (RuntimeFeature.IsDynamicCodeSupported) @@ -211,6 +263,15 @@ private static void VerifyRuntimeAssemblyBoundary(ISharpLinkClient client) throw new Exception($"unexpected NativeAOT client registration result: {result.Error}"); } + private static void VerifyRuntimeAssemblyBoundary(ISharpLinkMultiClusterClient client) + { + if (RuntimeFeature.IsDynamicCodeSupported) + return; + var result = client.RegisterAssembly("orders", typeof(Program).Assembly); + if (result.Succeeded || result.Error?.Code != SharpLinkAssemblyRegistrationErrorCode.PlatformNotSupported) + throw new Exception($"unexpected NativeAOT multi-cluster registration result: {result.Error}"); + } + private static void VerifyRuntimeAssemblyBoundary(ISharpLinkServer server) { if (RuntimeFeature.IsDynamicCodeSupported) @@ -272,6 +333,12 @@ public ValueTask EchoPairAsync(AotPair value) } } +[RpcService] +public sealed class SecondAotService : ISecondAotService +{ + public ValueTask MultiplyAsync(int value) => ValueTask.FromResult(value * 2); +} + public sealed class UserProfile { public string Name { get; set; } = string.Empty; diff --git a/test/SharpLink.AotSmoke/SharpLink.AotSmoke.csproj b/test/SharpLink.AotSmoke/SharpLink.AotSmoke.csproj index 34611ef..d0b9ae8 100644 --- a/test/SharpLink.AotSmoke/SharpLink.AotSmoke.csproj +++ b/test/SharpLink.AotSmoke/SharpLink.AotSmoke.csproj @@ -11,6 +11,7 @@ + Volatile.Read(ref _synchronousBlockStarted).Task; + public static Task RejectResponseStarted => Volatile.Read(ref _rejectResponseStarted).Task; + public static void Reset() { Volatile.Write(ref _blockStarted, NewSignal()); Volatile.Write(ref _blockRelease, NewSignal()); Volatile.Write(ref _synchronousBlockStarted, NewSignal()); Volatile.Write(ref _synchronousBlockRelease, NewSignal()); + Volatile.Write(ref _rejectResponseStarted, NewSignal()); + Volatile.Write(ref _rejectResponseRelease, NewSignal()); Volatile.Write(ref _created, 0); Volatile.Write(ref _disposed, 0); Volatile.Write(ref _notifications, 0); @@ -46,6 +52,9 @@ public static void Reset() public static void ReleaseSynchronousBlock() => Volatile.Read(ref _synchronousBlockRelease).TrySetResult(); + public static void ReleaseRejectResponse() + => Volatile.Read(ref _rejectResponseRelease).TrySetResult(); + public ValueTask UnaryAsync(int value, CancellationToken cancellationToken) => ValueTask.FromResult(value + 1); @@ -65,13 +74,15 @@ public async ValueTask ClientStreamAsync( return sum; } - public ValueTask RejectClientStreamAsync( + public async ValueTask RejectClientStreamAsync( IAsyncEnumerable values, CancellationToken cancellationToken) { _ = values; _ = cancellationToken; - return ValueTask.FromResult(-1); + Volatile.Read(ref _rejectResponseStarted).TrySetResult(); + await Volatile.Read(ref _rejectResponseRelease).Task.ConfigureAwait(false); + return -1; } public async IAsyncEnumerable ServerStreamAsync( diff --git a/test/SharpLink.Generator.Tests/RpcAnalyzerTests.cs b/test/SharpLink.Generator.Tests/RpcAnalyzerTests.cs index 307f864..9bafbaf 100644 --- a/test/SharpLink.Generator.Tests/RpcAnalyzerTests.cs +++ b/test/SharpLink.Generator.Tests/RpcAnalyzerTests.cs @@ -731,6 +731,102 @@ public SecondMarkedService(string ignored) { } return Task.CompletedTask; } + [Test] + public Task ClusterRouteShouldGenerateDeterministicSeparateManifest() + { + var source = BuildSource(""" +[SharpLink.Sdk.RpcContract] +public interface IOrdersService : SharpLink.Sdk.IService +{ + ValueTask GetAsync(int value, CancellationToken cancellationToken); +} +"""); + source = AddAssemblyAttribute( + source, + "[assembly: SharpLink.Sdk.SharpLinkClusterContractAssembly(\"orders\", typeof(IOrdersService))]"); + + var generated = RunGeneratorAndGetSources(source); + var route = generated.Single(text => text.Contains("GeneratedClusterRouteManifest", StringComparison.Ordinal)); + Ensure(route.Contains("new SharpLinkClusterKey(\"orders\")", StringComparison.Ordinal), + "cluster route should preserve the declared key"); + Ensure(route.Contains("SharpLinkGeneratedClusterRouteCatalog.Register", StringComparison.Ordinal), + "cluster route manifest should register from a module initializer"); + return Task.CompletedTask; + } + + [Test] + public Task InvalidClusterRouteKeyShouldReportSharplink038() + { + var source = BuildSource(""" +[SharpLink.Sdk.RpcContract] +public interface IOrdersService : SharpLink.Sdk.IService +{ + ValueTask GetAsync(int value, CancellationToken cancellationToken); +} +"""); + source = AddAssemblyAttribute( + source, + "[assembly: SharpLink.Sdk.SharpLinkClusterContractAssembly(\"bad key\", typeof(IOrdersService))]"); + + EnsureHasRule(source, "SHARPLINK038"); + return Task.CompletedTask; + } + + [Test] + public Task ConflictingClusterRouteShouldReportSharplink039() + { + var source = BuildSource(""" +[SharpLink.Sdk.RpcContract] +public interface IOrdersService : SharpLink.Sdk.IService +{ + ValueTask GetAsync(int value, CancellationToken cancellationToken); +} +"""); + source = AddAssemblyAttribute( + source, + "[assembly: SharpLink.Sdk.SharpLinkClusterContractAssembly(\"orders\", typeof(IOrdersService))]\n" + + "[assembly: SharpLink.Sdk.SharpLinkClusterContractAssembly(\"payments\", typeof(IOrdersService))]"); + + EnsureHasRule(source, "SHARPLINK039"); + return Task.CompletedTask; + } + + [Test] + public Task RouteMarkerWithoutGeneratedManifestShouldReportSharplink040() + { + var source = BuildSource(""" +[SharpLink.Sdk.RpcContract] +public interface IOrdersService : SharpLink.Sdk.IService +{ + ValueTask GetAsync(int value, CancellationToken cancellationToken); +} +"""); + source = AddAssemblyAttribute( + source, + "[assembly: SharpLink.Sdk.SharpLinkClusterContractAssembly(\"orders\", typeof(string))]"); + + EnsureHasRule(source, "SHARPLINK040"); + return Task.CompletedTask; + } + + [Test] + public Task NullRouteMarkerShouldReportSharplink041() + { + var source = BuildSource(""" +[SharpLink.Sdk.RpcContract] +public interface IOrdersService : SharpLink.Sdk.IService +{ + ValueTask GetAsync(int value, CancellationToken cancellationToken); +} +"""); + source = AddAssemblyAttribute( + source, + "[assembly: SharpLink.Sdk.SharpLinkClusterContractAssembly(\"orders\", null)]"); + + EnsureHasRule(source, "SHARPLINK041"); + return Task.CompletedTask; + } + private static string BuildSource(string contract) { return $$""" @@ -750,6 +846,14 @@ public sealed class RpcContractAttribute : Attribute { } + [AttributeUsage(AttributeTargets.Assembly, AllowMultiple = true)] + public sealed class SharpLinkClusterContractAssemblyAttribute : Attribute + { + public SharpLinkClusterContractAssemblyAttribute(string cluster, Type assemblyMarker) + { + } + } + [AttributeUsage(AttributeTargets.Method)] public sealed class TimeoutAttribute : Attribute { @@ -817,6 +921,9 @@ public RpcExternalCodecAttribute(Type type) { } """; } + private static string AddAssemblyAttribute(string source, string attribute) + => source.Replace("namespace SharpLink.Sdk", attribute + "\n\nnamespace SharpLink.Sdk", StringComparison.Ordinal); + private static void EnsureHasRule(string source, string ruleId) { var diagnostics = RunGenerator(source); diff --git a/test/SharpLink.IntegrationTests/RuntimeAssemblyIntegrationTests.cs b/test/SharpLink.IntegrationTests/RuntimeAssemblyIntegrationTests.cs index 565398c..ec47ebf 100644 --- a/test/SharpLink.IntegrationTests/RuntimeAssemblyIntegrationTests.cs +++ b/test/SharpLink.IntegrationTests/RuntimeAssemblyIntegrationTests.cs @@ -1,11 +1,299 @@ using System.Reflection; using System.Runtime.Loader; +using System.Collections.Frozen; using Microsoft.Extensions.DependencyInjection; namespace SharpLink.IntegrationTests; public sealed class RuntimeAssemblyIntegrationTests { + [Test] + [NotInParallel] + public async Task MultiClusterDynamicRegistrationShouldRouteToOneExplicitSlot() + { + await using var client = SharpLinkMultiClusterClientBuilder.Create() + .AddCluster("plugins", child => child.UseTcp(IPAddress.Loopback.ToString(), 1), + slot => slot.AllowDynamicContracts = true) + .AddCluster("other", child => child.UseTcp(IPAddress.Loopback.ToString(), 2), + slot => slot.AllowDynamicContracts = true) + .Build(); + using var plugin = PluginBundle.Load("multi-cluster-dynamic-registration", loadService: false); + + var first = client.RegisterAssembly("plugins", plugin.ContractAssembly); + Ensure(first.Succeeded, $"multi-cluster plugin registration: {first.Error}"); + + var proxy = GetMultiClusterProxy(client, plugin.ContractType); + Ensure(proxy is not null, "multi-cluster Get should create the dynamically routed proxy"); + + var second = client.RegisterAssembly("other", plugin.ContractAssembly); + Ensure(!second.Succeeded, "contract-owning assembly must not register in a second cluster"); + Ensure(second.Error?.Code == SharpLinkAssemblyRegistrationErrorCode.ContractConflict, + "second cluster should return a structured contract conflict"); + + var drained = await client.UnregisterAssemblyAsync( + "plugins", plugin.ContractAssembly, TimeSpan.FromSeconds(2)); + Ensure(drained.ReferencesReleased, "multi-cluster plugin unregister should release the child module"); + } + + [Test] + [NotInParallel] + public async Task MultiClusterSharedConnectShouldSurviveFirstWaiterCancellation() + { + var child = new BlockingConnectClient(); + var slot = new SharpLinkClusterSlot("plugins", child, AllowDynamicContracts: true); + await using var client = new SharpLinkMultiClusterClient( + new SharpLinkMultiClusterOptions(), + new[] { slot }.ToFrozenDictionary(static candidate => candidate.Key), + FrozenDictionary.Empty, + []); + + using var cancellation = new CancellationTokenSource(); + var cancelledWaiter = client.ConnectAsync(cancellation.Token).AsTask(); + await child.ConnectStarted.Task.WaitAsync(TimeSpan.FromSeconds(2)); + var survivingWaiter = client.ConnectAsync().AsTask(); + + cancellation.Cancel(); + await EnsureCancelledAsync(cancelledWaiter, "first shared connect waiter"); + Ensure(!survivingWaiter.IsCompleted, + "another connect waiter remains attached to the shared operation"); + + child.ReleaseConnect(); + await survivingWaiter.WaitAsync(TimeSpan.FromSeconds(2)); + Ensure(client.State == SharpLinkMultiClusterState.Ready, + "shared connect reaches ready after the first caller cancels its wait"); + } + + [Test] + [NotInParallel] + public async Task MultiClusterStopShouldWinWhenChildConnectCompletesAfterShutdown() + { + var child = new BlockingConnectClient(releaseWhenStopped: false, ignoreCancellation: true); + var slot = new SharpLinkClusterSlot("plugins", child, AllowDynamicContracts: true); + await using var client = new SharpLinkMultiClusterClient( + new SharpLinkMultiClusterOptions(), + new[] { slot }.ToFrozenDictionary(static candidate => candidate.Key), + FrozenDictionary.Empty, + []); + + var connecting = client.ConnectAsync().AsTask(); + await child.ConnectStarted.Task.WaitAsync(TimeSpan.FromSeconds(2)); + await client.StopAsync(); + + child.ReleaseConnect(); + await EnsureCancelledAsync(connecting.WaitAsync(TimeSpan.FromSeconds(2)), + "connect that was cancelled by coordinator shutdown"); + Ensure(client.State == SharpLinkMultiClusterState.Stopped, + "a post-stop child connect completion must not overwrite the coordinator terminal state"); + + try + { + await client.ConnectAsync(); + throw new Exception("assert failed: stopped coordinator must reject later connect attempts"); + } + catch (InvalidOperationException) + { + } + } + + [Test] + [NotInParallel] + public async Task MultiClusterCancelledUnregisterShouldStillReleaseCoordinatorRegistration() + { + await using var harness = await DynamicHarness.CreateAsync(); + await using var client = await CreateDynamicMultiClusterClientAsync(harness.Port); + using var plugin = PluginBundle.Load("multi-cluster-cancelled-unregister"); + plugin.ResetServiceState(); + RegisterMultiClusterPlugin(harness, client, plugin); + + var proxy = GetMultiClusterProxy(client, plugin.ContractType) + ?? throw new InvalidOperationException("Multi-cluster proxy factory returned null."); + var activeCall = InvokeValueTaskAsync( + proxy, plugin.ContractType, "BlockAsync", CancellationToken.None).AsTask(); + await plugin.GetStaticTask("BlockStarted").WaitAsync(TimeSpan.FromSeconds(2)); + + using var cancellation = new CancellationTokenSource(); + var unregister = client.UnregisterAssemblyAsync( + "plugins", plugin.ContractAssembly, TimeSpan.FromSeconds(2), cancellation.Token).AsTask(); + cancellation.Cancel(); + await EnsureCancelledAsync(unregister, "multi-cluster unregister wait"); + + plugin.ReleaseBlock(); + Ensure(await activeCall.WaitAsync(TimeSpan.FromSeconds(2)) == 42, + "the admitted call should complete before the child unregister releases its module"); + await WaitUntilAsync(() => client.RegisterAssembly("plugins", plugin.ContractAssembly).Succeeded); + + Ensure((await client.UnregisterAssemblyAsync( + "plugins", plugin.ContractAssembly, TimeSpan.FromSeconds(2))).ReferencesReleased, + "the re-registered coordinator module should release"); + await UnregisterMultiClusterPluginAsync(harness, plugin); + } + + [Test] + [NotInParallel] + public async Task MultiClusterDeferredUnregisterShouldRemoveARegistrationReleasedByItsChild() + { + using var plugin = PluginBundle.Load("multi-cluster-deferred-unregister", loadService: false); + await using var registrationSource = SharpClientBuilder.Create() + .UseTcp(IPAddress.Loopback.ToString(), 1) + .Build(); + var registrationResult = registrationSource.RegisterAssembly(plugin.ContractAssembly); + Ensure(registrationResult.Succeeded, "controlled child registration result"); + + var child = new ControlledDynamicAssemblyClient(registrationResult); + var slot = new SharpLinkClusterSlot("plugins", child, AllowDynamicContracts: true); + await using var client = new SharpLinkMultiClusterClient( + new SharpLinkMultiClusterOptions(), + new[] { slot }.ToFrozenDictionary(static candidate => candidate.Key), + FrozenDictionary.Empty, + []); + Ensure(client.RegisterAssembly("plugins", plugin.ContractAssembly).Succeeded, + "multi-cluster controlled registration"); + + var unregister = client.UnregisterAssemblyAsync( + "plugins", plugin.ContractAssembly, TimeSpan.Zero).AsTask(); + await child.FirstUnregisterStarted.Task.WaitAsync(TimeSpan.FromSeconds(2)); + child.CompleteTimedOutUnregister(); + Ensure(!(await unregister).ReferencesReleased, + "the child unregister should defer coordinator cleanup"); + + child.ReleaseAssembly(plugin.ContractAssembly); + await WaitUntilAsync(() => client.RegisterAssembly("plugins", plugin.ContractAssembly).Succeeded); + Ensure(child.UnregisterCalls == 1, + "deferred coordinator cleanup should poll child registration without starting another unregister"); + } + + [Test] + [NotInParallel] + public async Task MultiClusterRejectedUnregisterShouldRestoreCoordinatorRoute() + { + using var plugin = PluginBundle.Load("multi-cluster-rejected-unregister", loadService: false); + await using var registrationSource = SharpClientBuilder.Create() + .UseTcp(IPAddress.Loopback.ToString(), 1) + .Build(); + var registrationResult = registrationSource.RegisterAssembly(plugin.ContractAssembly); + Ensure(registrationResult.Succeeded, "controlled child registration result"); + + var child = new ControlledDynamicAssemblyClient(registrationResult); + var slot = new SharpLinkClusterSlot("plugins", child, AllowDynamicContracts: true); + await using var client = new SharpLinkMultiClusterClient( + new SharpLinkMultiClusterOptions(), + new[] { slot }.ToFrozenDictionary(static candidate => candidate.Key), + FrozenDictionary.Empty, + []); + Ensure(client.RegisterAssembly("plugins", plugin.ContractAssembly).Succeeded, + "multi-cluster controlled registration"); + + child.RejectNextUnregister(); + try + { + _ = await client.UnregisterAssemblyAsync( + "plugins", plugin.ContractAssembly, TimeSpan.Zero); + throw new Exception("assert failed: child unregister rejection must reach the caller"); + } + catch (InvalidOperationException exception) + { + Ensure(exception.Message.Contains("rejected", StringComparison.Ordinal), + "child unregister rejection is preserved"); + } + + _ = GetMultiClusterProxy(client, plugin.ContractType); + Ensure(child.IsDynamicAssemblyRegistered(plugin.ContractAssembly), + "child retains the rejected dynamic assembly"); + } + + [Test] + [NotInParallel] + public async Task MultiClusterRejectedUnregisterShouldReserveContractIdsUntilRoutesAreRestored() + { + using var originalPlugin = PluginBundle.Load("multi-cluster-rejected-unregister-original", loadService: false); + using var reloadedPlugin = PluginBundle.Load("multi-cluster-rejected-unregister-reloaded", loadService: false); + await using var registrationSource = SharpClientBuilder.Create() + .UseTcp(IPAddress.Loopback.ToString(), 1) + .Build(); + var registrationResult = registrationSource.RegisterAssembly(originalPlugin.ContractAssembly); + Ensure(registrationResult.Succeeded, "controlled child registration result"); + + var originalChild = new ControlledDynamicAssemblyClient(registrationResult); + var reloadedChild = new ControlledDynamicAssemblyClient(registrationResult); + var originalSlot = new SharpLinkClusterSlot("original", originalChild, AllowDynamicContracts: true); + var reloadedSlot = new SharpLinkClusterSlot("reloaded", reloadedChild, AllowDynamicContracts: true); + await using var client = new SharpLinkMultiClusterClient( + new SharpLinkMultiClusterOptions(), + new[] { originalSlot, reloadedSlot }.ToFrozenDictionary(static candidate => candidate.Key), + FrozenDictionary.Empty, + []); + Ensure(client.RegisterAssembly("original", originalPlugin.ContractAssembly).Succeeded, + "initial contract registration"); + + originalChild.BlockAndRejectNextUnregister(); + var unregister = client.UnregisterAssemblyAsync( + "original", originalPlugin.ContractAssembly, TimeSpan.Zero).AsTask(); + await originalChild.RejectedUnregisterStarted.Task.WaitAsync(TimeSpan.FromSeconds(2)); + + var conflictingRegistration = client.RegisterAssembly("reloaded", reloadedPlugin.ContractAssembly); + Ensure(!conflictingRegistration.Succeeded, + "an active unregister must continue reserving its ContractIds"); + Ensure(conflictingRegistration.Error?.Code == SharpLinkAssemblyRegistrationErrorCode.ContractConflict, + "the ContractId reservation should return a structured conflict"); + + originalChild.CompleteRejectedUnregister(); + try + { + await unregister; + throw new Exception("assert failed: the controlled child rejection must reach the caller"); + } + catch (InvalidOperationException exception) + { + Ensure(exception.Message.Contains("rejected", StringComparison.Ordinal), + "the controlled child rejection is preserved"); + } + _ = GetMultiClusterProxy(client, originalPlugin.ContractType); + } + + [Test] + [NotInParallel] + public async Task MultiClusterReplacementShouldPublishCoordinatorRoutesBeforeOldDrainAndAfterCallerCancellation() + { + await using var harness = await DynamicHarness.CreateAsync(); + await using var client = await CreateDynamicMultiClusterClientAsync(harness.Port); + using var oldPlugin = PluginBundle.Load("multi-cluster-cancelled-replacement-old"); + using var newPlugin = PluginBundle.Load("multi-cluster-cancelled-replacement-new"); + oldPlugin.ResetServiceState(); + RegisterMultiClusterPlugin(harness, client, oldPlugin); + + var proxy = GetMultiClusterProxy(client, oldPlugin.ContractType) + ?? throw new InvalidOperationException("Multi-cluster proxy factory returned null."); + var activeCall = InvokeValueTaskAsync( + proxy, oldPlugin.ContractType, "BlockAsync", CancellationToken.None).AsTask(); + await oldPlugin.GetStaticTask("BlockStarted").WaitAsync(TimeSpan.FromSeconds(2)); + + using var cancellation = new CancellationTokenSource(); + var replacement = client.ReplaceAssemblyAsync( + "plugins", + oldPlugin.ContractAssembly, + newPlugin.ContractAssembly, + TimeSpan.FromSeconds(2), + cancellation.Token).AsTask(); + + var newProxy = GetMultiClusterProxy(client, newPlugin.ContractType) + ?? throw new InvalidOperationException("Multi-cluster replacement proxy factory returned null."); + Ensure(await InvokeValueTaskAsync( + newProxy, newPlugin.ContractType, "UnaryAsync", 1, CancellationToken.None) == 2, + "replacement routes should publish while the old call is draining"); + + cancellation.Cancel(); + await EnsureCancelledAsync(replacement, "multi-cluster replacement wait"); + + oldPlugin.ReleaseBlock(); + Ensure(await activeCall.WaitAsync(TimeSpan.FromSeconds(2)) == 42, + "the admitted old call should complete before replacement cleanup"); + var released = await UnregisterWhenReplacementPublishesAsync(client, newPlugin.ContractAssembly); + Ensure(released.ReferencesReleased, + "the replacement assembly should become the coordinator registration after caller cancellation"); + + await UnregisterMultiClusterPluginAsync(harness, oldPlugin); + } + [Test] [NotInParallel] public async Task DynamicServiceRegistrationShouldRejectMissingProviderDependenciesTransactionally() @@ -184,6 +472,7 @@ public async Task EarlyServerResponseShouldRetainOnlyTheActiveClientStreamProduc await using var harness = await DynamicHarness.CreateAsync(); using var plugin = PluginBundle.Load("dynamic-early-client-stream-response"); RegisterAll(harness, plugin); + plugin.ResetServiceState(); object? proxy = GetProxy(harness.Client, plugin.ContractType); var producerStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); @@ -197,6 +486,8 @@ public async Task EarlyServerResponseShouldRetainOnlyTheActiveClientStreamProduc BlockingValues(producerStarted, producerRelease.Task), CancellationToken.None); await producerStarted.Task.WaitAsync(TimeSpan.FromSeconds(2)); + await plugin.GetStaticTask("RejectResponseStarted").WaitAsync(TimeSpan.FromSeconds(2)); + plugin.ReleaseRejectResponse(); Ensure(await response == -1, "server may return without consuming the request stream"); var serverService = await harness.Server.UnregisterAssemblyAsync( @@ -217,6 +508,7 @@ public async Task EarlyServerResponseShouldRetainOnlyTheActiveClientStreamProduc } finally { + plugin.ReleaseRejectResponse(); producerRelease.TrySetResult(); } @@ -1032,12 +1324,81 @@ private static async Task WaitUntilAsync(Func condition) } } + private static async Task CreateDynamicMultiClusterClientAsync(int port) + { + var client = SharpLinkMultiClusterClientBuilder.Create() + .AddCluster("plugins", child => child.UseTcp(IPAddress.Loopback.ToString(), port), + slot => slot.AllowDynamicContracts = true) + .Build(); + await client.ConnectAsync(); + return client; + } + + private static void RegisterMultiClusterPlugin( + DynamicHarness harness, + ISharpLinkMultiClusterClient client, + PluginBundle plugin) + { + Ensure(harness.Server.RegisterAssembly(plugin.ContractAssembly).Succeeded, + "multi-cluster server contract registration"); + Ensure(harness.Server.RegisterAssembly(plugin.ServiceAssembly).Succeeded, + "multi-cluster server service registration"); + Ensure(client.RegisterAssembly("plugins", plugin.ContractAssembly).Succeeded, + "multi-cluster client contract registration"); + } + + private static async Task UnregisterMultiClusterPluginAsync(DynamicHarness harness, PluginBundle plugin) + { + Ensure((await harness.Server.UnregisterAssemblyAsync( + plugin.ServiceAssembly, TimeSpan.FromSeconds(2))).ReferencesReleased, + "multi-cluster server service release"); + Ensure((await harness.Server.UnregisterAssemblyAsync( + plugin.ContractAssembly, TimeSpan.FromSeconds(2))).ReferencesReleased, + "multi-cluster server contract release"); + } + + private static async Task EnsureCancelledAsync(Task task, string name) + { + try + { + await task; + } + catch (OperationCanceledException) + { + return; + } + throw new Exception($"assert failed: {name} should observe caller cancellation"); + } + + private static async Task UnregisterWhenReplacementPublishesAsync( + ISharpLinkMultiClusterClient client, + Assembly assembly) + { + var deadline = Stopwatch.GetTimestamp() + (long)(Stopwatch.Frequency * 3d); + while (true) + { + try + { + return await client.UnregisterAssemblyAsync("plugins", assembly, TimeSpan.FromSeconds(2)); + } + catch (InvalidOperationException) when (Stopwatch.GetTimestamp() < deadline) + { + await Task.Delay(10); + } + } + } + private static void Ensure(bool condition, string message) { if (!condition) throw new Exception($"assert failed: {message}"); } + private static object? GetMultiClusterProxy(ISharpLinkMultiClusterClient client, Type contractType) + => typeof(ISharpLinkMultiClusterClient).GetMethod(nameof(ISharpLinkMultiClusterClient.Get))! + .MakeGenericMethod(contractType) + .Invoke(client, null); + private sealed class PluginBundle : IDisposable { private PluginLoadContext? _context; @@ -1089,6 +1450,8 @@ internal static PluginBundle Load(string contextName, bool loadService = true) internal void ReleaseSynchronousBlock() => InvokeStatic("ReleaseSynchronousBlock"); + internal void ReleaseRejectResponse() => InvokeStatic("ReleaseRejectResponse"); + internal int GetStaticInt(string propertyName) => (int)(ServiceType!.GetProperty(propertyName)!.GetValue(null) ?? -1); @@ -1166,6 +1529,180 @@ private sealed class PluginLoadContext(string name, string directory) } } + private sealed class ControlledDynamicAssemblyClient : ISharpLinkClient, IDynamicAssemblyRegistrationInspector + { + private readonly Lock _gate = new(); + private readonly HashSet _registeredAssemblies = new(ReferenceEqualityComparer.Instance); + private readonly SharpLinkAssemblyRegistrationResult _registrationResult; + private int _unregisterCalls; + private int _rejectNextUnregister; + private int _blockNextUnregisterRejection; + + internal ControlledDynamicAssemblyClient(SharpLinkAssemblyRegistrationResult registrationResult) + { + _registrationResult = registrationResult; + } + + internal TaskCompletionSource FirstUnregisterStarted { get; } = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + internal TaskCompletionSource RejectedUnregisterStarted { get; } = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + private TaskCompletionSource FirstUnregisterCompletion { get; } = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + private TaskCompletionSource RejectedUnregisterCompletion { get; } = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + public SharpLinkConnectionState State => SharpLinkConnectionState.Ready; + + public SharpLinkAssemblyRegistrationResult RegisterAssembly(Assembly assembly) + { + lock (_gate) + _registeredAssemblies.Add(assembly); + return _registrationResult; + } + + public ValueTask UnregisterAssemblyAsync( + Assembly assembly, + TimeSpan gracefulTimeout, + CancellationToken cancellationToken = default) + { + _ = assembly; + _ = gracefulTimeout; + _ = cancellationToken; + if (Interlocked.Exchange(ref _rejectNextUnregister, 0) != 0) + { + return ValueTask.FromException( + new InvalidOperationException("controlled child unregister rejected")); + } + if (Interlocked.Exchange(ref _blockNextUnregisterRejection, 0) != 0) + { + RejectedUnregisterStarted.TrySetResult(true); + return new ValueTask(RejectedUnregisterCompletion.Task); + } + if (Interlocked.Increment(ref _unregisterCalls) == 1) + { + FirstUnregisterStarted.TrySetResult(true); + return new ValueTask(FirstUnregisterCompletion.Task); + } + return ValueTask.FromResult(new SharpLinkAssemblyUnregisterResult { ReferencesReleased = false }); + } + + public ValueTask ReplaceAssemblyAsync( + Assembly oldAssembly, + Assembly newAssembly, + TimeSpan gracefulTimeout, + CancellationToken cancellationToken = default) + => throw new NotSupportedException(); + + public ValueTask ConnectAsync(CancellationToken cancellationToken = default) => ValueTask.CompletedTask; + + public ValueTask StopAsync(CancellationToken cancellationToken = default) => ValueTask.CompletedTask; + + public ValueTask CheckHealthAsync(CancellationToken cancellationToken = default) + => ValueTask.FromResult(new SharpLinkHealthCheckResult(SharpLinkHealthStatus.Ready)); + + public TContract Get() where TContract : IService + => default!; + + public ValueTask DisposeAsync() => ValueTask.CompletedTask; + + public bool IsDynamicAssemblyRegistered(Assembly assembly) + { + lock (_gate) + return _registeredAssemblies.Contains(assembly); + } + + internal void CompleteTimedOutUnregister() + => FirstUnregisterCompletion.TrySetResult(new SharpLinkAssemblyUnregisterResult + { + ReferencesReleased = false, + RemainingCalls = 1 + }); + + internal void ReleaseAssembly(Assembly assembly) + { + lock (_gate) + _registeredAssemblies.Remove(assembly); + } + + internal void RejectNextUnregister() => Volatile.Write(ref _rejectNextUnregister, 1); + + internal void BlockAndRejectNextUnregister() => Volatile.Write(ref _blockNextUnregisterRejection, 1); + + internal void CompleteRejectedUnregister() + => RejectedUnregisterCompletion.TrySetException( + new InvalidOperationException("controlled child unregister rejected")); + + internal int UnregisterCalls => Volatile.Read(ref _unregisterCalls); + } + + private sealed class BlockingConnectClient : ISharpLinkClient + { + private readonly bool _releaseWhenStopped; + private readonly bool _ignoreCancellation; + private readonly TaskCompletionSource _connectRelease = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _state = (int)SharpLinkConnectionState.Created; + + internal BlockingConnectClient(bool releaseWhenStopped = true, bool ignoreCancellation = false) + { + _releaseWhenStopped = releaseWhenStopped; + _ignoreCancellation = ignoreCancellation; + } + + internal TaskCompletionSource ConnectStarted { get; } = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + public SharpLinkConnectionState State => (SharpLinkConnectionState)Volatile.Read(ref _state); + + public async ValueTask ConnectAsync(CancellationToken cancellationToken = default) + { + ConnectStarted.TrySetResult(); + if (_ignoreCancellation) + await _connectRelease.Task.ConfigureAwait(false); + else + await _connectRelease.Task.WaitAsync(cancellationToken).ConfigureAwait(false); + Volatile.Write(ref _state, (int)SharpLinkConnectionState.Ready); + } + + public ValueTask StopAsync(CancellationToken cancellationToken = default) + { + _ = cancellationToken; + if (_releaseWhenStopped) + _connectRelease.TrySetResult(); + Volatile.Write(ref _state, (int)SharpLinkConnectionState.Stopped); + return ValueTask.CompletedTask; + } + + public ValueTask CheckHealthAsync(CancellationToken cancellationToken = default) + => ValueTask.FromResult(new SharpLinkHealthCheckResult(SharpLinkHealthStatus.Ready)); + + public TContract Get() where TContract : IService => throw new NotSupportedException(); + + public SharpLinkAssemblyRegistrationResult RegisterAssembly(Assembly assembly) + => throw new NotSupportedException(); + + public ValueTask UnregisterAssemblyAsync( + Assembly assembly, + TimeSpan gracefulTimeout, + CancellationToken cancellationToken = default) + => throw new NotSupportedException(); + + public ValueTask ReplaceAssemblyAsync( + Assembly oldAssembly, + Assembly newAssembly, + TimeSpan gracefulTimeout, + CancellationToken cancellationToken = default) + => throw new NotSupportedException(); + + public ValueTask DisposeAsync() => StopAsync(); + + internal void ReleaseConnect() => _connectRelease.TrySetResult(); + } + private sealed class DynamicHarness : IAsyncDisposable { private readonly CancellationTokenSource _serverCancellation; @@ -1175,12 +1712,14 @@ private sealed class DynamicHarness : IAsyncDisposable private DynamicHarness( ISharpLinkServer server, ISharpLinkClient client, + int port, CancellationTokenSource serverCancellation, Task serverTask, ServiceProvider serviceProvider) { Server = server; Client = client; + Port = port; _serverCancellation = serverCancellation; _serverTask = serverTask; _serviceProvider = serviceProvider; @@ -1188,6 +1727,7 @@ private DynamicHarness( internal ISharpLinkServer Server { get; } internal ISharpLinkClient Client { get; } + internal int Port { get; } internal static async Task CreateAsync( bool registerDynamicServiceDependencies = true) @@ -1209,7 +1749,7 @@ internal static async Task CreateAsync( .UseHeartbeat(TimeSpan.FromMilliseconds(250), TimeSpan.FromSeconds(5)) .Build(); await client.ConnectAsync(); - return new DynamicHarness(server, client, serverCancellation, serverTask, serviceProvider); + return new DynamicHarness(server, client, port, serverCancellation, serverTask, serviceProvider); } public async ValueTask DisposeAsync() diff --git a/test/SharpLink.UnitTests/Client/SharpLinkMultiClusterClientTests.cs b/test/SharpLink.UnitTests/Client/SharpLinkMultiClusterClientTests.cs new file mode 100644 index 0000000..5ec2dc9 --- /dev/null +++ b/test/SharpLink.UnitTests/Client/SharpLinkMultiClusterClientTests.cs @@ -0,0 +1,454 @@ +using System.Reflection; +using System.Reflection.Emit; +using System.Collections.Generic; +using System.Runtime.CompilerServices; +using System.Threading; +using SharpLink.Client; +using SharpLink.Sdk; + +namespace SharpLink.UnitTests.Client; + +public sealed class SharpLinkMultiClusterClientTests +{ + private static readonly Assembly TestManifestAssembly = CreateTestManifestAssembly(); + + [Test] + public async Task StaticRouteShouldCreateTheTargetChildProxyAndConnectEverySlot() + { + SharpLinkGeneratedAssemblyCatalog.Register(Manifest.Instance); + SharpLinkGeneratedClusterRouteCatalog.Register(RouteManifest.Instance); + var ordersTransport = new TestClientTransportFactory(); + var paymentsTransport = new TestClientTransportFactory(); + + await using var client = SharpLinkMultiClusterClientBuilder.Create() + .AddCluster("orders", child => child.UseTransport(ordersTransport)) + .AddCluster("payments", child => child.UseTransport(paymentsTransport), + slot => slot.AllowDynamicContracts = true) + .Build(); + + var proxy = client.Get(); + Ensure(proxy is OrdersProxy, "Get should create the proxy directly from the routed child client"); + await client.ConnectAsync(); + + Ensure(client.State == SharpLinkMultiClusterState.Ready, "all slots should be ready after shared connect"); + Ensure(ordersTransport.ConnectCount == 1, "orders child should connect once"); + Ensure(paymentsTransport.ConnectCount == 1, "payments child should connect once"); + Ensure(client.GetClusterState("orders") == SharpLinkConnectionState.Ready, "orders slot state"); + } + + [Test] + [NotInParallel] + public async Task FilteredStaticRoutesShouldIgnoreUnrelatedGlobalManifests() + { + SharpLinkGeneratedAssemblyCatalog.Register(Manifest.Instance); + SharpLinkGeneratedClusterRouteCatalog.Register(RouteManifest.Instance); + ISharpLinkGeneratedAssemblyManifest? unrelatedManifest = new ThrowingCodecManifest(); + SharpLinkGeneratedAssemblyCatalog.Register(unrelatedManifest); + try + { + await using var client = SharpLinkMultiClusterClientBuilder.Create() + .AddCluster("orders", child => child.UseTransport(new TestClientTransportFactory())) + .Build(); + + Ensure(client.Get() is OrdersProxy, + "a filtered child should build without reading an unrelated global manifest"); + } + finally + { + unrelatedManifest = null; + CollectWeakCatalogEntries(); + } + } + + [Test] + [NotInParallel] + public async Task BuildShouldIgnoreRoutesForUnconfiguredClusters() + { + SharpLinkGeneratedAssemblyCatalog.Register(Manifest.Instance); + SharpLinkGeneratedClusterRouteCatalog.Register(RouteManifest.Instance); + ISharpLinkGeneratedClusterRouteManifest? unrelatedRoute = new UnconfiguredRouteManifest(); + SharpLinkGeneratedClusterRouteCatalog.Register(unrelatedRoute); + try + { + await using var client = SharpLinkMultiClusterClientBuilder.Create() + .AddCluster("orders", child => child.UseTransport(new TestClientTransportFactory())) + .Build(); + + Ensure(client.Get() is OrdersProxy, + "unconfigured route manifests must not block a coordinator's configured routes"); + } + finally + { + unrelatedRoute = null; + CollectWeakCatalogEntries(); + } + } + + [Test] + [NotInParallel] + public async Task FilteredStaticRoutesShouldNotRetainUnconfiguredRouteManifests() + { + SharpLinkGeneratedAssemblyCatalog.Register(Manifest.Instance); + SharpLinkGeneratedClusterRouteCatalog.Register(RouteManifest.Instance); + var unrelatedRoute = RegisterUnconfiguredRouteManifest(); + + await using (var client = SharpLinkMultiClusterClientBuilder.Create() + .AddCluster("orders", child => child.UseTransport(new TestClientTransportFactory())) + .Build()) + { + Ensure(client.Get() is OrdersProxy, + "the configured route must build without retaining unrelated route manifests"); + } + + CollectWeakCatalogEntries(); + Ensure(!unrelatedRoute.IsAlive, + "a coordinator must not retain a collectible route manifest that contributes no configured route"); + } + + [Test] + public async Task DynamicRegistrationShouldPreserveStructuredNullAndMissingUnregisterResults() + { + await using var client = SharpLinkMultiClusterClientBuilder.Create() + .AddCluster("plugins", child => child.UseTransport(new TestClientTransportFactory()), + slot => slot.AllowDynamicContracts = true) + .Build(); + + var nullRegistration = client.RegisterAssembly("plugins", null!); + Ensure(!nullRegistration.Succeeded && + nullRegistration.Error?.Code == SharpLinkAssemblyRegistrationErrorCode.InvalidArgument, + "null dynamic registration must return the shared structured invalid-argument result"); + + var missingUnregister = await client.UnregisterAssemblyAsync( + "plugins", typeof(string).Assembly, TimeSpan.Zero); + Ensure(!missingUnregister.ReferencesReleased, + "unregistering an assembly that is not registered must match child false-result semantics"); + } + + [Test] + public async Task DynamicRegistrationShouldReturnStructuredFailureAfterStop() + { + await using var client = SharpLinkMultiClusterClientBuilder.Create() + .AddCluster("plugins", child => child.UseTransport(new TestClientTransportFactory()), + slot => slot.AllowDynamicContracts = true) + .Build(); + + await client.StopAsync(); + var registration = client.RegisterAssembly("plugins", typeof(string).Assembly); + Ensure(!registration.Succeeded && + registration.Error?.Code == SharpLinkAssemblyRegistrationErrorCode.InvalidObjectState, + "registration after shutdown must return the structured terminal-state failure before cluster lookup"); + } + + [Test] + public async Task DynamicReplacementShouldReturnStructuredFailureAfterStop() + { + await using var client = SharpLinkMultiClusterClientBuilder.Create() + .AddCluster("plugins", child => child.UseTransport(new TestClientTransportFactory()), + slot => slot.AllowDynamicContracts = true) + .Build(); + + await client.StopAsync(); + var replacement = await client.ReplaceAssemblyAsync( + "plugins", typeof(string).Assembly, typeof(int).Assembly, TimeSpan.Zero); + Ensure(!replacement.Succeeded && + replacement.Error?.Code == SharpLinkAssemblyRegistrationErrorCode.InvalidObjectState, + "replacement after shutdown must return the structured terminal-state failure before cluster lookup"); + } + + [Test] + public async Task DynamicUnregisterShouldReturnFalseAfterStop() + { + await using var client = SharpLinkMultiClusterClientBuilder.Create() + .AddCluster("plugins", child => child.UseTransport(new TestClientTransportFactory()), + slot => slot.AllowDynamicContracts = true) + .Build(); + + await client.StopAsync(); + var unregister = await client.UnregisterAssemblyAsync( + "plugins", typeof(string).Assembly, TimeSpan.Zero); + Ensure(!unregister.ReferencesReleased, + "unregistration after shutdown must return the child-compatible false result before cluster lookup"); + } + + [Test] + public Task EmptySlotShouldRequireExplicitDynamicOptIn() + { + var builder = SharpLinkMultiClusterClientBuilder.Create() + .AddCluster("dynamic", child => child.UseTransport(new TestClientTransportFactory())); + + return EnsureThrows(() => + { + _ = builder.Build(); + return Task.CompletedTask; + }); + } + + [Test] + public async Task UnknownContractShouldFailWithoutSelectingAnotherCluster() + { + SharpLinkGeneratedAssemblyCatalog.Register(Manifest.Instance); + SharpLinkGeneratedClusterRouteCatalog.Register(RouteManifest.Instance); + await using var client = SharpLinkMultiClusterClientBuilder.Create() + .AddCluster("orders", child => child.UseTransport(new TestClientTransportFactory())) + .Build(); + + await EnsureThrows(() => + { + _ = client.Get(); + return Task.CompletedTask; + }); + } + + [Test] + public async Task BuildShouldRejectZeroClustersAndConnectionBudgetOverflow() + { + await EnsureThrows(() => + { + _ = SharpLinkMultiClusterClientBuilder.Create().Build(); + return Task.CompletedTask; + }); + + SharpLinkGeneratedAssemblyCatalog.Register(Manifest.Instance); + SharpLinkGeneratedClusterRouteCatalog.Register(RouteManifest.Instance); + await EnsureThrows(() => + { + _ = SharpLinkMultiClusterClientBuilder.Create() + .Configure(options => options.MaxTotalConfiguredConnections = 1) + .AddCluster("orders", child => child.UseTransport(new TestClientTransportFactory())) + .AddCluster("plugins", child => child.UseTransport(new TestClientTransportFactory()), + slot => slot.AllowDynamicContracts = true) + .Build(); + return Task.CompletedTask; + }); + } + + [Test] + public async Task SingleEndpointSlotsShouldUseTheirFixedConnectionBudget() + { + SharpLinkGeneratedAssemblyCatalog.Register(Manifest.Instance); + SharpLinkGeneratedClusterRouteCatalog.Register(RouteManifest.Instance); + await using var client = SharpLinkMultiClusterClientBuilder.Create() + .Configure(options => options.MaxTotalConfiguredConnections = 2) + .AddCluster("orders", child => child.UseEndpoint( + Endpoint("orders", 5001), + static _ => new TestClientTransportFactory())) + .AddCluster("plugins", child => child.UseEndpoint( + Endpoint("plugins", 5002), + static _ => new TestClientTransportFactory()), + slot => slot.AllowDynamicContracts = true) + .Build(); + + Ensure(client.GetClusterState("orders") == SharpLinkConnectionState.Created, + "single-endpoint slots fit their configured fixed-client budget"); + } + + [Test] + public async Task SingleEndpointCollectionsShouldUseTheirFixedConnectionBudget() + { + SharpLinkGeneratedAssemblyCatalog.Register(Manifest.Instance); + SharpLinkGeneratedClusterRouteCatalog.Register(RouteManifest.Instance); + await using var client = SharpLinkMultiClusterClientBuilder.Create() + .Configure(options => options.MaxTotalConfiguredConnections = 2) + .AddCluster("orders", child => child.UseEndpoints( + new OneShotEndpointEnumerable(Endpoint("orders", 5001)), + static _ => new TestClientTransportFactory())) + .AddCluster("plugins", child => child.UseEndpoints( + new OneShotEndpointEnumerable(Endpoint("plugins", 5002)), + static _ => new TestClientTransportFactory()), + slot => slot.AllowDynamicContracts = true) + .Build(); + + Ensure(client.GetClusterState("orders") == SharpLinkConnectionState.Created, + "one-endpoint collections must use their fixed-client budget without a second enumeration"); + } + + [Test] + public async Task StaticEndpointClustersShouldUseTheirEffectiveConnectionBudget() + { + SharpLinkGeneratedAssemblyCatalog.Register(Manifest.Instance); + SharpLinkGeneratedClusterRouteCatalog.Register(RouteManifest.Instance); + await using var client = SharpLinkMultiClusterClientBuilder.Create() + .Configure(options => options.MaxTotalConfiguredConnections = 2) + .AddCluster("orders", child => child + .UseEndpoints( + [Endpoint("orders-a", 5001), Endpoint("orders-b", 5002)], + static _ => new TestClientTransportFactory()) + .UseCluster(static options => + { + options.MaxConnections = 4; + options.MaxConnectionsPerEndpoint = 1; + })) + .Build(); + + Ensure(client.GetClusterState("orders") == SharpLinkConnectionState.Created, + "a static cluster must count its endpoint-capped connection capacity during coordinator preflight"); + } + + [Test] + public async Task StopDuringInitialConnectShouldRemainStoppedAfterSharedConnectFaults() + { + SharpLinkGeneratedAssemblyCatalog.Register(Manifest.Instance); + SharpLinkGeneratedClusterRouteCatalog.Register(RouteManifest.Instance); + var blocked = new BlockingTransportFactory(); + await using var client = SharpLinkMultiClusterClientBuilder.Create() + .AddCluster("orders", child => child.UseTransport(blocked)) + .Build(); + + var connecting = client.ConnectAsync().AsTask(); + await blocked.ConnectStarted.Task.WaitAsync(TimeSpan.FromSeconds(2)); + await client.StopAsync(); + await EnsureThrows(async () => await connecting); + + Ensure(client.State == SharpLinkMultiClusterState.Stopped, + "shutdown must own the terminal state when it races the initial shared connect"); + await client.StopAsync(); + } + + private static async Task EnsureThrows(Func action) where TException : Exception + { + try + { + await action(); + } + catch (TException) + { + return; + } + throw new Exception($"Expected {typeof(TException).Name}."); + } + + private static void Ensure(bool condition, string message) + { + if (!condition) + throw new Exception(message); + } + + private static void CollectWeakCatalogEntries() + { + GC.Collect(); + GC.WaitForPendingFinalizers(); + GC.Collect(); + _ = SharpLinkGeneratedAssemblyCatalog.CreateSnapshot(); + _ = SharpLinkGeneratedClusterRouteCatalog.CreateSnapshot(); + } + + [MethodImpl(MethodImplOptions.NoInlining)] + private static WeakReference RegisterUnconfiguredRouteManifest() + { + ISharpLinkGeneratedClusterRouteManifest manifest = new UnconfiguredRouteManifest(); + SharpLinkGeneratedClusterRouteCatalog.Register(manifest); + return new WeakReference(manifest); + } + + private static SharpLinkEndpoint Endpoint(string id, int port) + => new() + { + Id = id, + Address = new SharpLinkTcpAddress("127.0.0.1", port) + }; + + private static Assembly CreateTestManifestAssembly() + => AssemblyBuilder.DefineDynamicAssembly( + new AssemblyName("SharpLink.MultiClusterClientTests.Manifest"), + AssemblyBuilderAccess.Run); + + private interface IOrdersContract : IService; + private interface IUnroutedContract : IService; + private sealed class OrdersProxy : IOrdersContract; + + private sealed class Manifest : ISharpLinkGeneratedAssemblyManifest + { + public static readonly Manifest Instance = new(); + public int ApiVersion => SharpLinkGeneratedManifestVersions.Api; + public int ProtocolVersion => SharpLinkGeneratedManifestVersions.Protocol; + public string GeneratorVersion => "test"; + public Assembly OwnerAssembly => TestManifestAssembly; + public string CompileTimeDescriptor => "multi-cluster-test"; + public IReadOnlyList Contracts { get; } = + [ + new SharpLinkGeneratedContractDescriptor( + typeof(IOrdersContract), + typeof(IOrdersContract).FullName!, + 8_101, + "orders-v1", + [], + static _ => new OrdersProxy(), + static () => throw new NotSupportedException()) + ]; + public IReadOnlyList Services { get; } = []; + public IReadOnlyList Codecs { get; } = []; + public IReadOnlyList Dependencies { get; } = []; + } + + private sealed class RouteManifest : ISharpLinkGeneratedClusterRouteManifest + { + public static readonly RouteManifest Instance = new(); + public Assembly OwnerAssembly => TestManifestAssembly; + public IReadOnlyList Routes { get; } = + [ + new SharpLinkGeneratedClusterAssemblyRoute( + "orders", + TestManifestAssembly, + TestManifestAssembly.FullName!) + ]; + } + + private sealed class ThrowingCodecManifest : ISharpLinkGeneratedAssemblyManifest + { + public int ApiVersion => SharpLinkGeneratedManifestVersions.Api; + public int ProtocolVersion => SharpLinkGeneratedManifestVersions.Protocol; + public string GeneratorVersion => "test"; + public Assembly OwnerAssembly => typeof(string).Assembly; + public string CompileTimeDescriptor => "unrelated-manifest"; + public IReadOnlyList Contracts => []; + public IReadOnlyList Services => []; + public IReadOnlyList Codecs + => throw new InvalidOperationException("Unrelated manifests must not be read by a filtered child."); + public IReadOnlyList Dependencies => []; + } + + private sealed class UnconfiguredRouteManifest : ISharpLinkGeneratedClusterRouteManifest + { + public Assembly OwnerAssembly => typeof(SharpLinkMultiClusterClientTests).Assembly; + public IReadOnlyList Routes { get; } = + [ + new SharpLinkGeneratedClusterAssemblyRoute( + "unconfigured", + typeof(string).Assembly, + typeof(string).Assembly.FullName!) + ]; + } + + private sealed class BlockingTransportFactory : IClientTransportFactory + { + internal TaskCompletionSource ConnectStarted { get; } = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + public async ValueTask ConnectAsync(CancellationToken cancellationToken = default) + { + ConnectStarted.TrySetResult(true); + await Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken); + throw new InvalidOperationException("The cancelled connect should not continue."); + } + + public ValueTask DisposeAsync() => ValueTask.CompletedTask; + } + + private sealed class OneShotEndpointEnumerable : IEnumerable + { + private readonly SharpLinkEndpoint _endpoint; + private int _enumerationCount; + + public OneShotEndpointEnumerable(SharpLinkEndpoint endpoint) => _endpoint = endpoint; + + public IEnumerator GetEnumerator() + { + if (Interlocked.Increment(ref _enumerationCount) != 1) + throw new InvalidOperationException("Endpoint source must be enumerated only once."); + + yield return _endpoint; + } + + System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator() => GetEnumerator(); + } +} diff --git a/test/SharpLink.UnitTests/Client/StaticEndpointBuilderTests.cs b/test/SharpLink.UnitTests/Client/StaticEndpointBuilderTests.cs index fd3c4ed..1b24885 100644 --- a/test/SharpLink.UnitTests/Client/StaticEndpointBuilderTests.cs +++ b/test/SharpLink.UnitTests/Client/StaticEndpointBuilderTests.cs @@ -127,6 +127,33 @@ public async Task StaticClusterShouldOwnEveryFactoryExactlyOnce() Ensure(second.DisposeCount == 1, "second cluster factory disposal count"); } + [Test] + public async Task BuilderShouldTakeFreshEndpointSnapshotsAfterPreflightBuilds() + { + var endpoints = new List { Endpoint("first", 5001) }; + var createdEndpointIds = new List(); + var builder = SharpClientBuilder.Create() + .UseEndpoints(endpoints, endpoint => + { + createdEndpointIds.Add(endpoint.Id); + return new TrackingFactory(); + }); + + Ensure(builder.GetConfiguredMaximumConnections() == 1, + "one endpoint should reserve the fixed-client connection budget"); + await using (var first = builder.Build()) + { + } + + endpoints[0] = Endpoint("second", 5002); + await using (var second = builder.Build()) + { + } + + Ensure(createdEndpointIds.SequenceEqual(["first", "second"]), + "a reused builder must take a fresh endpoint snapshot for each build"); + } + [Test] public async Task ClusterBuildCleanupShouldReleaseEveryFactoryWhenOneDisposalFails() { diff --git a/test/SharpLink.UnitTests/Hosting/SharpLinkMultiClusterClientAccessorTests.cs b/test/SharpLink.UnitTests/Hosting/SharpLinkMultiClusterClientAccessorTests.cs new file mode 100644 index 0000000..78c2031 --- /dev/null +++ b/test/SharpLink.UnitTests/Hosting/SharpLinkMultiClusterClientAccessorTests.cs @@ -0,0 +1,79 @@ +using System.Reflection; +using System.Threading; +using SharpLink.Hosting; +using SharpLink.Sdk; + +namespace SharpLink.UnitTests.Hosting; + +public sealed class SharpLinkMultiClusterClientAccessorTests +{ + [Test] + public async Task StopShouldRejectLaterPublicationAndReads() + { + var accessor = new SharpLinkMultiClusterClientAccessor(); + accessor.Stop(); + + await EnsureThrows(async () => await accessor.GetClientAsync()); + await EnsureThrows(() => + { + accessor.SetClient(new FakeMultiClusterClient()); + return Task.CompletedTask; + }); + } + + private static async Task EnsureThrows(Func action) where TException : Exception + { + try + { + await action(); + } + catch (TException) + { + return; + } + + throw new Exception($"Expected {typeof(TException).Name}."); + } + + private sealed class FakeMultiClusterClient : ISharpLinkMultiClusterClient + { + public SharpLinkMultiClusterState State => SharpLinkMultiClusterState.Ready; + + public ValueTask ConnectAsync(CancellationToken cancellationToken = default) => ValueTask.CompletedTask; + + public ValueTask StopAsync(CancellationToken cancellationToken = default) => ValueTask.CompletedTask; + + public TContract Get() where TContract : IService => throw new NotSupportedException(); + + public SharpLinkConnectionState GetClusterState(SharpLinkClusterKey cluster) => SharpLinkConnectionState.Ready; + + public ValueTask CheckHealthAsync( + SharpLinkClusterKey cluster, + CancellationToken cancellationToken = default) + => ValueTask.FromResult(new SharpLinkHealthCheckResult(SharpLinkHealthStatus.Ready)); + + public SharpLinkAssemblyRegistrationResult RegisterAssembly(SharpLinkClusterKey cluster, Assembly assembly) + => SharpLinkAssemblyRegistrationResult.Success(); + + public ValueTask UnregisterAssemblyAsync( + SharpLinkClusterKey cluster, + Assembly assembly, + TimeSpan gracefulTimeout, + CancellationToken cancellationToken = default) + => ValueTask.FromResult(new SharpLinkAssemblyUnregisterResult { ReferencesReleased = true }); + + public ValueTask ReplaceAssemblyAsync( + SharpLinkClusterKey cluster, + Assembly oldAssembly, + Assembly newAssembly, + TimeSpan gracefulTimeout, + CancellationToken cancellationToken = default) + => ValueTask.FromResult(new SharpLinkAssemblyReplacementResult + { + Succeeded = true, + ReferencesReleased = true + }); + + public ValueTask DisposeAsync() => ValueTask.CompletedTask; + } +}