diff --git a/CHANGELOG.md b/CHANGELOG.md index 6262789e..518ee4f5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,10 +14,44 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). - `GraphClientOptions.SubscriptionMaxBufferedEvents` and `SubscriptionOverflowPolicy` (`Fault`, `DropOldest`, `KeepLatest`) control the buffer between the transport and the observer. Each subscriber gets its own buffer. -- `GraphClientOptions.SubscriptionIdleTimeout` (SSE) fails a stalled stream. +- `GraphClientOptions.SubscriptionIdleTimeout` (SSE) fails a stalled stream. The timer runs + only while a read waits for data, so time spent handling an event between reads never + counts as idle; the error states the configured timeout exactly (`750 ms`, `30 s`). - `GraphClientOptions.SubscriptionRetryAttempts` (WebSocket, default 5) bounds reconnects. - `QueryExecutor.ProcessResponse(JsonElement, ...)` and `ProcessResponse(byte[], ...)` overloads for streaming callers. +- **WebSocket upgrades now run the request hooks** (`ConfigureRequest`, + `WithRequestConfiguration`, `WithAccessToken`): the headers they produce are copied onto + the upgrade request of the initial connect and of every reconnect, so a refreshed token + is picked up on reconnect. `HttpClient.DefaultRequestHeaders` are sent too (a hook header + of the same name wins), as they are for queries. Handshake-owned headers (`Host`, + `Sec-WebSocket-*`, ...) are skipped; on browsers, where custom WebSocket headers are + impossible, the hooks still run and the `Authorization` value is available for + `SendAuthorizationInConnectionInit` (#117). The hooks count against the connect + timeout, which is now 30 s (Websocket.Client's 5 s default covered only the socket + connect). An exception thrown by a hook (or by `ConnectionInitPayload`) reaches the + subscription's error path (`OnError`, `WhenConnected`) as itself, not wrapped in + Websocket.Client's `WebsocketException`; on a reconnect it is the inner exception of the + terminal `IOException`. +- `GraphClientOptions.ConnectionInitPayload` (and the matching `GraphClient` property) + set the WebSocket `connection_init` payload. Unlike the old option it accepts any JSON + object, e.g. `new Dictionary { ["Authorization"] = $"Bearer {token}" }` + (the graphql-ws/Apollo connection-parameter convention) or a server-specific + `authToken`. Called before every (re)connect, before the socket opens (within the connect + timeout), so a slow token fetch cannot hit the server's `connection_init` wait. +- `GraphClientOptions.SendAuthorizationInConnectionInit` (default off) merges the + `Authorization` header produced by the request hooks into the `connection_init` + payload — how a per-user token reaches the server on Blazor WebAssembly, where the + browser does not allow headers on WebSocket requests. Wins on key conflicts with the + `ConnectionInitPayload`, which must serialize to a JSON object (dictionary, anonymous + object, POCO, `JsonObject`); any other shape throws. + +### Deprecated + +- `GraphClientOptions.WSConnectionInitPayload` / `GraphClient.WSConnectionInitPayload`: use + `ConnectionInitPayload`, which can return any JSON object, not just a `GraphQLRequest`. + The old option keeps working (and falls back) until removed; `ConnectionInitPayload` + wins when both are set. ### Changed @@ -56,6 +90,10 @@ Behaviour changes that callers may notice. None of them changes a public signatu - **WebSocket: unknown message types are ignored** instead of terminating the subscription with `InvalidDataException`. This matches the SSE transport's tolerance for unknown event types and keeps forward compatibility with proxies/servers. +- **A failed initial WebSocket connect or handshake no longer retries in the background.** The + exception surfaces from `StartAsync`; previously the transport's `Error` disconnect + also started the bounded reconnect loop, re-running request hooks up to five times and + reporting a second, delayed error for a failure the caller had already handled. - **WebSocket close codes follow the graphql-ws reference client.** Fatal: `4004`, `4005`, `4400`, `4401`, `4406`, `4409`, `4429`, `4500` and fatal internal `1xxx` codes. Retryable: `4403`, `4408`, `4499` and `4504`. Previously `4408` was treated as fatal. diff --git a/README.md b/README.md index 1b83d63b..bf245945 100644 --- a/README.md +++ b/README.md @@ -235,7 +235,11 @@ var launches = await userClient.Query.Launches().Select().ExecuteAsync(); ``` `WithAccessToken` sets `Authorization: Bearer ` and asks the delegate for the token on every request; -`WithRequestConfiguration` takes any hook. +`WithRequestConfiguration` takes any hook. The hooks also apply to subscriptions: server-sent events through +the HTTP pipeline, WebSocket upgrades by copying the headers onto the upgrade request (initial connect and +each reconnect — a refreshed token is picked up automatically). On Blazor WebAssembly, where the browser +forbids headers on WebSocket requests, set `GraphClientOptions.SendAuthorizationInConnectionInit` to send +the token in the `connection_init` payload instead. ## Agent mode (preview) @@ -393,7 +397,7 @@ await foreach (var orders in pager.GetAllPagesAsync(maxPages: 50)) `StartAsync` returns a cold `IObservable`. Three transports are supported via `GraphClientOptions.SubscriptionProtocol`: -- `GraphQLWebSocket` (default) — `graphql-transport-ws` protocol: sends `connection_init` (with `WSConnectionInitPayload` if configured), waits up to 10 s for `connection_ack`, then subscribes. Transient disconnects (connection lost, server-initiated close, or no frame at all for 30 s) reconnect and **resubscribe** on each new session; events during the gap are not replayed. Reconnection is bounded: up to `SubscriptionRetryAttempts` attempts with exponential backoff (1 s, 2 s, 4 s ... capped at 30 s) plus 300-3000 ms jitter; a successful `connection_ack` resets the budget. When the budget is spent the stream fails once with an `IOException` whose inner exception is the last failure. Protocol-level close codes `4004`, `4005`, `4400`, `4401`, `4406`, `4409`, `4429`, `4500` (and fatal internal `1xxx` codes) are fatal and never retried, per the graphql-ws reference client. `4403`, `4408`, `4499` and `4504` are retried. +- `GraphQLWebSocket` (default) — `graphql-transport-ws` protocol: sends `connection_init` (with `ConnectionInitPayload` if configured), waits up to 10 s for `connection_ack`, then subscribes. Transient disconnects (connection lost, server-initiated close, or no frame at all for 30 s) reconnect and **resubscribe** on each new session; events during the gap are not replayed. Reconnection is bounded: up to `SubscriptionRetryAttempts` attempts with exponential backoff (1 s, 2 s, 4 s ... capped at 30 s) plus 300-3000 ms jitter; a successful `connection_ack` resets the budget. When the budget is spent the stream fails once with an `IOException` whose inner exception is the last failure. Protocol-level close codes `4004`, `4005`, `4400`, `4401`, `4406`, `4409`, `4429`, `4500` (and fatal internal `1xxx` codes) are fatal and never retried, per the graphql-ws reference client. `4403`, `4408`, `4499` and `4504` are retried. - `ApolloWebSocket` — the legacy `graphql-ws` (Apollo) subprotocol: `connection_init` / `start` / `stop`, with `data` messages instead of `next`. - `ServerSentEvents` — POSTs the subscription and reads the `text/event-stream` response (GraphQL-over-SSE, distinct-connections mode). Per-event `error` frames throw; a `complete` frame completes the stream. Unknown event types (e.g. `ka` keep-alives) are ignored per the WHATWG SSE spec. @@ -402,7 +406,9 @@ Options: - `SubscriptionMaxBufferedEvents` (default 128) and `SubscriptionOverflowPolicy` (`Fault` default, `DropOldest`, `KeepLatest`) control the buffer between transport and observer. - `SubscriptionRetryAttempts` (WebSocket, default 5) bounds reconnect attempts after a dropped connection; `0` fails on the first drop. - `SubscriptionIdleTimeout` (SSE only, default off) fails the subscription when no bytes arrive — including keep-alive comments — within the window. -- `WSConnectionInitPayload` sets the `connection_init` payload for the WebSocket transports. +- `ConnectionInitPayload` sets the `connection_init` payload for the WebSocket transports. Return any JSON object the server expects as connection parameters, e.g. `new Dictionary { ["Authorization"] = $"Bearer {token}" }` (the graphql-ws/Apollo convention). Called before every (re)connect, before the socket opens. Supersedes the obsolete `WSConnectionInitPayload`, which is limited to a `GraphQLRequest` payload. +- `SendAuthorizationInConnectionInit` (default off) copies the `Authorization` header produced by `ConfigureRequest`/`WithAccessToken` into the `connection_init` payload — how a per-user token reaches the server on Blazor WebAssembly, where the browser does not allow headers on WebSocket requests. It merges into (and wins on key conflicts with) a `ConnectionInitPayload` that serializes to a JSON object. +- WebSocket upgrade requests carry the headers produced by `ConfigureRequest`/`WithAccessToken` plus `HttpClient.DefaultRequestHeaders` (a hook header wins on a name clash), except on browsers, where custom headers are impossible. Register them through DI like any other options: diff --git a/docs/Linq2GraphQL.Docs/Pages/Subscriptions.razor b/docs/Linq2GraphQL.Docs/Pages/Subscriptions.razor index 4d81c76d..6ef29e2e 100644 --- a/docs/Linq2GraphQL.Docs/Pages/Subscriptions.razor +++ b/docs/Linq2GraphQL.Docs/Pages/Subscriptions.razor @@ -72,10 +72,18 @@

Server-sent events go through the same HTTP pipeline as queries, so ConfigureRequest, - WithAccessToken and your HttpClient handlers apply to them. WebSocket connections - don't send those headers; when a subscription needs a per-user token, server-sent events are the simpler - choice. + WithAccessToken and your HttpClient handlers apply to them. The same hooks also + run for the WebSocket transports: the headers they produce are copied onto every upgrade request + (initial connect and each reconnect), so a refreshed token is picked up automatically.

+

+ Blazor WebAssembly is the exception: the browser does not allow custom headers on WebSocket requests. + Set SendAuthorizationInConnectionInit = true to send the hook's + Authorization header inside the connection_init payload instead — the + graphql-ws/Apollo convention of an Authorization connection parameter. For any other + connection parameters, use ConnectionInitPayload: +

+
@@ -168,6 +176,24 @@ .WithHttpClient(httpClient => httpClient.BaseAddress = new Uri("https://example.com/graphql")); """; + private const string ConnectionInit = """ + builder.Services + .AddSampleClient(options => + { + // Blazor WebAssembly: headers are not allowed on WebSocket requests, + // so the token travels in the connection_init payload. + options.SendAuthorizationInConnectionInit = true; + + // Any other connection parameters your server expects: + options.ConnectionInitPayload = (client, cancellationToken) => ValueTask.FromResult( + new Dictionary + { + ["authToken"] = tokenService.CurrentToken, + ["clientVersion"] = "1.4", + }); + }); + """; + private const string WithErrors = """ var events = await sampleClient .Subscription diff --git a/src/Linq2GraphQL.Client.Subscriptions/IdleTimeoutStream.cs b/src/Linq2GraphQL.Client.Subscriptions/IdleTimeoutStream.cs index 4f1298c3..0d435042 100644 --- a/src/Linq2GraphQL.Client.Subscriptions/IdleTimeoutStream.cs +++ b/src/Linq2GraphQL.Client.Subscriptions/IdleTimeoutStream.cs @@ -2,15 +2,15 @@ namespace Linq2GraphQL.Client.Subscriptions; /// /// Stream decorator that fails a read with an IOException once no bytes have -/// arrived within the idle timeout. The timer is re-armed before every read, so -/// keep-alives (including WHATWG : comments, which the SSE parser consumes -/// invisibly) keep the stream alive. +/// arrived within the idle timeout. Each read gets its own timer that runs only while +/// that read waits for data, so keep-alives (including WHATWG : comments, which +/// the SSE parser consumes invisibly) keep the stream alive, and time the consumer spends +/// between reads (processing an event, waiting for a thread) never counts as idle. /// internal sealed class IdleTimeoutStream : Stream { private readonly Stream inner; private readonly TimeSpan idleTimeout; - private readonly CancellationTokenSource idleCts = new(); private bool disposed; internal IdleTimeoutStream(Stream inner, TimeSpan idleTimeout) @@ -22,7 +22,6 @@ internal IdleTimeoutStream(Stream inner, TimeSpan idleTimeout) this.inner = inner ?? throw new ArgumentNullException(nameof(inner)); this.idleTimeout = idleTimeout; - idleCts.CancelAfter(idleTimeout); } public override bool CanRead => inner.CanRead; @@ -35,36 +34,45 @@ public override async ValueTask ReadAsync(Memory buffer, Cancellation { ObjectDisposedException.ThrowIf(disposed, this); - if (!cancellationToken.CanBeCanceled) - { - return await ReadInnerAsync(buffer, idleCts.Token, external: default).ConfigureAwait(false); - } - - using var linked = CancellationTokenSource.CreateLinkedTokenSource(idleCts.Token, cancellationToken); - return await ReadInnerAsync(buffer, linked.Token, external: cancellationToken).ConfigureAwait(false); - } - - private async ValueTask ReadInnerAsync(Memory buffer, CancellationToken readToken, - CancellationToken external) - { - // Re-arm for this read: any bytes (data line, event boundary, comment) reset it. - idleCts.CancelAfter(idleTimeout); + // A fresh timer per read, disposed when the read completes: it runs only while waiting + // for data, and a timer that fired during a read which still returned bytes cannot leak + // into the next read (a CancellationTokenSource cannot be reset once cancelled). + // CreateLinkedTokenSource with a non-cancelable token is equivalent to a plain source. + using var readCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + readCts.CancelAfter(idleTimeout); try { - return await inner.ReadAsync(buffer, readToken).ConfigureAwait(false); + return await inner.ReadAsync(buffer, readCts.Token).ConfigureAwait(false); } catch (OperationCanceledException) when ( - idleCts.IsCancellationRequested && - (!external.CanBeCanceled || !external.IsCancellationRequested)) + readCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested) { - // The idle timer fired and the caller did not cancel: the connection is dead. - // The linked readToken cannot be used to distinguish the sources — it is also - // cancelled by the idle timer — so the caller's own token is checked here. - throw new IOException($"SSE idle timeout: no data received within {idleTimeout.TotalSeconds:F0}s."); + // Only the idle timer or the caller can cancel readCts; it fired and the caller did + // not cancel, so the timer fired with no bytes received: the connection is dead. An + // OperationCanceledException the inner stream raised on its own propagates unchanged. + throw new IOException($"SSE idle timeout: no data received within {FormatTimeout(idleTimeout)}."); } } + /// + /// Whole seconds as "N s", anything else as "N ms" - never rounds up (750 ms is not "1s"); + /// a positive timeout under 1 ms is "<1 ms", not "0 s". Expects a positive timeout, as + /// the constructor enforces. + /// + internal static string FormatTimeout(TimeSpan timeout) + { + var milliseconds = (long)timeout.TotalMilliseconds; + if (milliseconds == 0) + { + return "<1 ms"; + } + + return milliseconds % 1000 == 0 + ? FormattableString.Invariant($"{milliseconds / 1000} s") + : FormattableString.Invariant($"{milliseconds} ms"); + } + public override void Flush() => throw new NotSupportedException(); public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException(); public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); @@ -80,7 +88,6 @@ protected override void Dispose(bool disposing) if (disposing) { - idleCts.Dispose(); inner.Dispose(); } @@ -94,7 +101,6 @@ public override async ValueTask DisposeAsync() return; } - idleCts.Dispose(); await inner.DisposeAsync().ConfigureAwait(false); GC.SuppressFinalize(this); } diff --git a/src/Linq2GraphQL.Client.Subscriptions/WSClient.cs b/src/Linq2GraphQL.Client.Subscriptions/WSClient.cs index b784d801..0ca5b93e 100644 --- a/src/Linq2GraphQL.Client.Subscriptions/WSClient.cs +++ b/src/Linq2GraphQL.Client.Subscriptions/WSClient.cs @@ -1,7 +1,9 @@ using System.Net.WebSockets; using System.Reactive.Linq; using System.Reactive.Subjects; +using System.Runtime.ExceptionServices; using System.Text.Json; +using System.Text.Json.Nodes; using Websocket.Client; namespace Linq2GraphQL.Client.Subscriptions; @@ -42,6 +44,27 @@ public class WSClient : IAsyncDisposable // reconnection is enabled, which this class disables (verified: no NoMessageReceived event). private Timer watchdog; private long lastMessageTicks; + // connection_init payload of the CURRENT connection. Resolved in the connection factory, + // before the socket connects, so the payload delegate runs inside ConnectTimeout and cannot + // hold an open socket past the server's connection_init wait (graphql-ws closes with 4408). + // Read by SendConnectionInit, which runs after the factory for the same generation. + private volatile object connectionInitPayload; + // The last failure of the pre-connect preparation (request hooks, payload delegate). + // Websocket.Client wraps any factory exception in its own WebsocketException; this lets + // Start() and the retry loop surface the caller's original exception instead. + private volatile ExceptionDispatchInfo preparationFailure; + + // Headers the WebSocket handshake owns; a hook setting them must not break the upgrade. + private static readonly HashSet HandshakeOwnedHeaders = new(StringComparer.OrdinalIgnoreCase) + { + "Host", + "Connection", + "Upgrade", + "Sec-WebSocket-Key", + "Sec-WebSocket-Version", + "Sec-WebSocket-Protocol", + "Sec-WebSocket-Extensions", + }; public WSClient(GraphClient graphClient, GraphQLRequest payload) { @@ -49,25 +72,110 @@ public WSClient(GraphClient graphClient, GraphQLRequest payload) this.payload = payload; maxRetryAttempts = Math.Max(0, graphClient.ClientOptions.SubscriptionRetryAttempts); - var factory = new Func(() => + // The async connection factory runs for EVERY connect, including each reconnect attempt + // inside the retry loop — so request hooks (WithAccessToken etc.) run per connection and + // a refreshed token is picked up on reconnect. It must return an already-connected socket. + var factory = new Func>(async (uri, cancellationToken) => { var socket = new ClientWebSocket(); - socket.Options.AddSubProtocol(subprotocol ?? GetSubprotocolString()); - return socket; + try + { + try + { + socket.Options.AddSubProtocol(subprotocol ?? GetSubprotocolString()); + var authorization = await ApplyRequestHooksAsync(socket.Options, cancellationToken) + .ConfigureAwait(false); + connectionInitPayload = await BuildConnectionInitPayloadAsync(authorization, cancellationToken) + .ConfigureAwait(false); + preparationFailure = null; + } + catch (Exception exception) when (exception is not OperationCanceledException) + { + // Cancellation (ConnectTimeout elapsed, disposal) stays wrapped: a bare + // OperationCanceledException from Start() would read as the caller's own cancel. + preparationFailure = ExceptionDispatchInfo.Capture(exception); + throw; + } + + await socket.ConnectAsync(uri, cancellationToken).ConfigureAwait(false); + return socket; + } + catch + { + socket.Dispose(); + throw; + } }); - client = new WebsocketClient(new Uri(graphClient.SubscriptionUrl), factory) + client = new WebsocketClient(new Uri(graphClient.SubscriptionUrl), null, factory) { // This class owns reconnection (bounded, with backoff) and silence detection. The // library's own loop retries forever and goes silent after an Error. ReconnectTimeout = null, IsReconnectionEnabled = false, + // The default 5 s budget covers only the socket connect; the request hooks (token + // fetch, e.g. an IdP round-trip) run inside the same window, so give them room. + ConnectTimeout = TimeSpan.FromSeconds(30), }; messageSubscription = client.MessageReceived.Subscribe(HandleMessage, ReportFailure); disconnectionSubscription = client.DisconnectionHappened.Subscribe(HandleDisconnection, ReportFailure); } + /// + /// Runs the request hooks for one upgrade request and copies the resulting headers onto the + /// socket. Returns the effective Authorization value (for + /// ), or null. + /// + private async Task ApplyRequestHooksAsync(ClientWebSocketOptions socketOptions, + CancellationToken cancellationToken) + { + // The hooks are HttpRequestMessage-based (same shape as queries and SSE); run them on a + // dummy request for the subscription URL, then copy the resulting headers onto the socket. + using var request = new HttpRequestMessage(HttpMethod.Get, new Uri(graphClient.SubscriptionUrl)); + await graphClient.ConfigureRequestAsync(request, cancellationToken).ConfigureAwait(false); + + // Queries get HttpClient.DefaultRequestHeaders from HttpClient.SendAsync; the upgrade does + // not go through HttpClient, so merge them here. Same precedence as SendAsync: a header + // already on the request (set by a hook) wins over the default of the same name. + var defaults = graphClient.HttpClient?.DefaultRequestHeaders; + var headers = new List<(string Name, string Value)>(); + if (defaults is not null) + { + foreach (var (key, values) in defaults) + { + if (!request.Headers.Contains(key)) + { + headers.Add((key, string.Join(", ", values))); + } + } + } + + foreach (var (key, values) in request.Headers) + { + headers.Add((key, string.Join(", ", values))); + } + + var authorization = request.Headers.Authorization?.ToString() ?? defaults?.Authorization?.ToString(); + + // Browsers do not allow custom headers on WebSocket requests: the Authorization value + // returned here travels in the connection_init payload instead (Blazor WebAssembly). + if (OperatingSystem.IsBrowser()) + { + return authorization; + } + + foreach (var (name, value) in headers) + { + if (!HandshakeOwnedHeaders.Contains(name) && !string.IsNullOrEmpty(value)) + { + socketOptions.SetRequestHeader(name, value); + } + } + + return authorization; + } + public IObservable Subscription => subscriptionSubject.AsObservable(); internal IObservable Payloads => payloadSubject.AsObservable(); public IObservable Errors => errorSubject.AsObservable(); @@ -79,6 +187,12 @@ public WSClient(GraphClient graphClient, GraphQLRequest payload) /// internal TimeSpan NoMessageTimeout { get; set; } = TimeSpan.FromSeconds(30); + /// + /// Delay before reconnect attempt n (0-based). Tests shorten it so "no background retry" + /// assertions do not depend on outwaiting the real backoff. + /// + internal Func RetryDelay { get; set; } = BackoffDelay; + public async ValueTask DisposeAsync() { lock (stateLock) @@ -140,21 +254,55 @@ private async Task InitializeAsync() // Created before StartOrFail so an eager ack arriving during the handshake // completes this generation's source instead of racing a null field (L2). ackSource = CreateAckSource(); - await client.StartOrFail().ConfigureAwait(false); + try + { + await StartTransportAsync().ConfigureAwait(false); - // Initial connect: a missing subprotocol is a protocol violation and stays terminal. - var negotiated = client.NativeClient?.SubProtocol; - if (string.IsNullOrEmpty(negotiated)) + // Initial connect: a missing subprotocol is a protocol violation and stays terminal. + var negotiated = client.NativeClient?.SubProtocol; + if (string.IsNullOrEmpty(negotiated)) + { + throw new InvalidDataException("WebSocket server did not negotiate a GraphQL subprotocol."); + } + + subprotocol = ValidateNegotiatedSubprotocol(negotiated); + SendConnectionInit(); + await WaitForAckAsync().ConfigureAwait(false); + } + catch { - throw new InvalidDataException("WebSocket server did not negotiate a GraphQL subprotocol."); + // Any failure of the initial connect/handshake propagates to the caller from Start(). + // The transport may still surface the same failure later (e.g. the server closing on + // its own ack timeout); without marking it terminal the retry loop would run the + // full bounded budget in the background and report a second, delayed error for a + // failure the caller already handled. + Interlocked.Exchange(ref terminalFailure, 1); + throw; } - subprotocol = ValidateNegotiatedSubprotocol(negotiated); - await SendConnectionInitAsync().ConfigureAwait(false); - await WaitForAckAsync().ConfigureAwait(false); StartWatchdog(); } + /// + /// , but a failure of our own pre-connect step + /// (a request hook or the connection_init payload delegate) is rethrown as the original + /// exception, with its stack, instead of Websocket.Client's WebsocketException + /// wrapper - so callers can catch their own types (e.g. a token service's exception). + /// + private async Task StartTransportAsync() + { + preparationFailure = null; + try + { + await client.StartOrFail().ConfigureAwait(false); + } + catch (Exception exception) when (preparationFailure is { } failure + && ReferenceEquals(exception.InnerException, failure.SourceException)) + { + failure.Throw(); + } + } + private void StartWatchdog() { Volatile.Write(ref lastMessageTicks, Environment.TickCount64); @@ -212,13 +360,14 @@ private static TaskCompletionSource CreateAckSource() return new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); } - private async Task SendConnectionInitAsync() + private void SendConnectionInit() { - var initRequest = new WebsocketRequest(WebsocketRequestTypes.CONNECTION_INIT); - if (graphClient.WSConnectionInitPayload is not null) + // The payload was resolved by the connection factory for this connection (before the + // socket opened), so nothing slow runs between the open socket and connection_init. + var initRequest = new WebsocketRequest(WebsocketRequestTypes.CONNECTION_INIT) { - initRequest.Payload = await graphClient.WSConnectionInitPayload(graphClient).ConfigureAwait(false); - } + Payload = connectionInitPayload + }; SendRequest(initRequest); lock (stateLock) @@ -231,6 +380,41 @@ private async Task SendConnectionInitAsync() SendSubscription(); } + private async ValueTask BuildConnectionInitPayloadAsync(string authorization, + CancellationToken cancellationToken) + { + // New option wins; the obsolete WSConnectionInitPayload is the fallback (resolved inside + // GraphClient so the deprecation pragma lives in exactly one place). + var payload = await graphClient.GetConnectionInitPayloadAsync(cancellationToken).ConfigureAwait(false); + + if (!graphClient.ClientOptions.SendAuthorizationInConnectionInit || string.IsNullOrEmpty(authorization)) + { + return payload; + } + + // Blazor WebAssembly path: the browser drops headers on WebSocket requests, so the token + // produced by the request hooks travels inside connection_init (graphql-ws/Apollo style + // "Authorization" connection parameter). The hook value wins on key conflicts. + if (payload is null) + { + return new JsonObject { ["Authorization"] = authorization }; + } + + // Merge on the JSON the payload serializes to - with the same options it would be sent + // with - so any shape that is a JSON object merges: dictionaries of any value type, + // anonymous objects, POCOs ([JsonPropertyName] honored), JsonObject and JsonElement. + if (JsonSerializer.SerializeToNode(payload, payload.GetType(), jsonOptions) is JsonObject merged) + { + merged["Authorization"] = authorization; + return merged; + } + + throw new InvalidOperationException( + "SendAuthorizationInConnectionInit requires the ConnectionInitPayload to serialize to a JSON " + + "object (or be null) so the Authorization connection parameter can be merged; " + + $"it was {payload.GetType()}."); + } + private string ValidateNegotiatedSubprotocol(string negotiatedProtocol) { // Null on reconnect can mean the handshake has not completed yet. Skipping @@ -460,21 +644,29 @@ private async Task RunRetryLoopAsync(Exception lastException) return; } - var delay = BackoffDelay(attempt); + var delay = RetryDelay(attempt); await Task.Delay(delay, cancellationSource.Token).ConfigureAwait(false); + // Re-check after the backoff: a terminal failure (e.g. the initial connect's + // Start() already threw) can be flagged while this loop is sleeping, and the + // attempt below must not run hooks for a client that already reported failure. + if (Volatile.Read(ref terminalFailure) != 0) + { + return; + } + try { // Websocket.Client keeps a failed/aborted socket as its current one: a bare StartOrFail() // after a failed attempt returns at once with State=Aborted and no connection (verified). // Stop() resets it so every attempt opens a real socket. await client.Stop(WebSocketCloseStatus.NormalClosure, string.Empty).ConfigureAwait(false); - await client.StartOrFail().ConfigureAwait(false); + await StartTransportAsync().ConfigureAwait(false); // A fresh ack source per handshake: the previous one is already completed and would // make WaitForAckAsync return without waiting for this connection's connection_ack. subprotocol = ValidateNegotiatedSubprotocol(client.NativeClient?.SubProtocol); ackSource = CreateAckSource(); - await SendConnectionInitAsync().ConfigureAwait(false); + SendConnectionInit(); await WaitForAckAsync().ConfigureAwait(false); Volatile.Write(ref lastMessageTicks, Environment.TickCount64); return; diff --git a/src/Linq2GraphQL.Client/GraphClient.cs b/src/Linq2GraphQL.Client/GraphClient.cs index 43f0f094..ddfc634d 100644 --- a/src/Linq2GraphQL.Client/GraphClient.cs +++ b/src/Linq2GraphQL.Client/GraphClient.cs @@ -99,8 +99,35 @@ public Task> ExecuteAsync(string query, IDictionary options.Value; + /// + /// The connection_init payload delegate from , + /// exposed for callers that build transports by hand. The obsolete + /// is the fallback when this is null. + /// + public Func> ConnectionInitPayload => + options.Value.ConnectionInitPayload; + + [Obsolete("Use ConnectionInitPayload instead: it can return any JSON object (an authorization payload, " + + "for example), not just a GraphQLRequest.")] public Func> WSConnectionInitPayload => options.Value.WSConnectionInitPayload; + internal async ValueTask GetConnectionInitPayloadAsync(CancellationToken cancellationToken) + { + if (options.Value.ConnectionInitPayload is { } payload) + { + return await payload(this, cancellationToken).ConfigureAwait(false); + } + +#pragma warning disable CS0618 // WSConnectionInitPayload is the legacy shape, kept for source compatibility. + if (options.Value.WSConnectionInitPayload is { } legacy) + { + return await legacy(this).ConfigureAwait(false); + } +#pragma warning restore CS0618 + + return null; + } + internal async ValueTask ConfigureRequestAsync(HttpRequestMessage request, CancellationToken cancellationToken) { if (options.Value.ConfigureRequest is { } configureRequest) diff --git a/src/Linq2GraphQL.Client/GraphClientOptions.cs b/src/Linq2GraphQL.Client/GraphClientOptions.cs index 86870740..98bb0ca6 100644 --- a/src/Linq2GraphQL.Client/GraphClientOptions.cs +++ b/src/Linq2GraphQL.Client/GraphClientOptions.cs @@ -20,13 +20,49 @@ public class GraphClientOptions { public bool UseSafeMode { get; set; } = false; public SubscriptionProtocol SubscriptionProtocol { get; set; } = default; + + /// + /// Sets the connection_init payload of the WebSocket subscription transports. Return any + /// JSON object the server expects as connection parameters, for example + /// new Dictionary<string, object> { ["Authorization"] = $"Bearer {token}" } (the + /// graphql-ws and Apollo convention) or { { "token", token } }. Called before every + /// (re)connect, before the socket opens, so a refreshed token is picked up and a slow token + /// fetch cannot run into the server's connection_init wait; it counts within the connect + /// timeout (30 s) together with the request hooks. Overrides the obsolete + /// when set. When + /// is enabled the payload must serialize to a + /// JSON object (a dictionary, an anonymous object, a POCO, a JsonObject...) so the + /// Authorization parameter can be merged. + /// + public Func> ConnectionInitPayload { get; set; } + + [Obsolete("Use ConnectionInitPayload instead: it can return any JSON object (an authorization payload, " + + "for example), not just a GraphQLRequest. Superseded when ConnectionInitPayload is set.")] public Func> WSConnectionInitPayload { get; set; } = _ => Task.FromResult(null!); /// - /// Called for every HTTP request the client sends (queries, mutations, the safe-mode schema query and - /// server-sent-event subscriptions) just before it is sent. Use it to set per-request headers, such as an - /// access token, without a . + /// Copies the Authorization header produced by and + /// derived clients (WithAccessToken on a generated client) into the connection_init + /// payload. This is how a per-user token reaches the server on Blazor WebAssembly, where the + /// browser does not allow headers on the WebSocket upgrade request. An Authorization in + /// is used when no hook sets one. The header is + /// merged into the JSON object from and wins on key conflicts. Like + /// any connection_init payload, the token is only as protected as the transport: prefer + /// wss:// so it is not sent in clear text. + /// + public bool SendAuthorizationInConnectionInit { get; set; } + + /// + /// Called for every request the client sends (queries, mutations, the safe-mode schema query, + /// server-sent-event subscriptions and the WebSocket subscription upgrade) just before it is + /// sent. On the WebSocket upgrade the hook's headers, together with the + /// (a hook header wins on a name conflict), are + /// copied to the upgrade request. Use it + /// to set per-request headers, such as an access token, without a + /// . On Blazor WebAssembly the browser ignores headers on + /// WebSocket requests; set to send the token + /// in the connection_init payload instead. /// public Func ConfigureRequest { get; set; } @@ -48,7 +84,9 @@ public class GraphClientOptions /// /// Optional idle timeout for SSE subscriptions. If no bytes (including keep-alive - /// comments) arrive within this window, the subscription fails. Null disables it. + /// comments) arrive within this window, the subscription fails. The window is measured + /// per read while waiting for data, so time spent handling an event never counts as idle. + /// Null disables it. /// public TimeSpan? SubscriptionIdleTimeout { get; set; } diff --git a/test/Linq2GraphQL.Tests/GraphClientOptionsTests.cs b/test/Linq2GraphQL.Tests/GraphClientOptionsTests.cs index 9860958a..d8e63645 100644 --- a/test/Linq2GraphQL.Tests/GraphClientOptionsTests.cs +++ b/test/Linq2GraphQL.Tests/GraphClientOptionsTests.cs @@ -158,6 +158,10 @@ public void Defaults_ArePinned() Assert.Equal(128, options.SubscriptionMaxBufferedEvents); Assert.Equal(SubscriptionOverflowPolicy.Fault, options.SubscriptionOverflowPolicy); Assert.Null(options.SubscriptionIdleTimeout); + Assert.Null(options.ConnectionInitPayload); + Assert.False(options.SendAuthorizationInConnectionInit); +#pragma warning disable CS0618 // The legacy default is pinned deliberately: non-null delegate returning null. Assert.NotNull(options.WSConnectionInitPayload); +#pragma warning restore CS0618 } } diff --git a/test/Linq2GraphQL.Tests/IdleTimeoutStreamTests.cs b/test/Linq2GraphQL.Tests/IdleTimeoutStreamTests.cs index 5ae78827..8a879248 100644 --- a/test/Linq2GraphQL.Tests/IdleTimeoutStreamTests.cs +++ b/test/Linq2GraphQL.Tests/IdleTimeoutStreamTests.cs @@ -60,17 +60,105 @@ public async Task SlowButRegularData_DoesNotTimeOut() Assert.Equal(20, total); } + [Fact] + public async Task ConsumerPauseBetweenReads_IsNotIdle() + { + // The timer measures waiting for data only: a consumer that is busy (or starved of a + // thread) between reads longer than the timeout must not fail a stream with data ready. + // Regression for the CI flake of SlowButRegularData_DoesNotTimeOut (one stalled + // continuation between reads exhausted the 750 ms window). + using var stream = new IdleTimeoutStream(new MemoryStream(new byte[3]), TimeSpan.FromMilliseconds(100)); + var buffer = new byte[1]; + + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + + Assert.Equal(1, await stream.ReadAsync(buffer)); + await Task.Delay(400); + Assert.Equal(1, await stream.ReadAsync(buffer)); + await Task.Delay(400); + Assert.Equal(1, await stream.ReadAsync(buffer, cts.Token)); + } + + [Fact] + public async Task InnerStreamOwnCancellation_IsNotReportedAsIdleTimeout() + { + // An OperationCanceledException the inner stream raises itself (neither the idle timer + // nor the caller fired) propagates unchanged rather than being mislabelled as idle. + using var stream = new IdleTimeoutStream(new SelfCancellingStream(), TimeSpan.FromSeconds(30)); + + await Assert.ThrowsAsync(async () => +#pragma warning disable CA2022 // The read is expected to throw; the byte count is irrelevant. + await stream.ReadAsync(new byte[1])); +#pragma warning restore CA2022 + } + + [Fact] + public async Task DelayBeforeFirstRead_IsNotIdle() + { + using var stream = new IdleTimeoutStream(new MemoryStream(new byte[1]), TimeSpan.FromMilliseconds(100)); + await Task.Delay(400); + + Assert.Equal(1, await stream.ReadAsync(new byte[1])); + } + + [Fact] + public async Task TimerFiredDuringReadThatStillReturnedData_NextReadIsNotFailed() + { + // The inner stream ignores cancellation on the first read and returns data after the + // timeout fired. Bytes arrived, so the connection is alive: the fired timer must not + // poison the next read (a CancellationTokenSource cannot be un-cancelled). + // 1 s timeout: the second read's fresh timer must not fire before the inner stream checks + // its token, which leaves ~1 s of tolerance for a stalled thread on a loaded CI runner. + using var stream = new IdleTimeoutStream(new LateFirstByteStream(TimeSpan.FromMilliseconds(1500)), + TimeSpan.FromSeconds(1)); + var buffer = new byte[1]; + + Assert.Equal(1, await stream.ReadAsync(buffer)); + Assert.Equal(1, await stream.ReadAsync(buffer)); + } + + [Theory] + // endToEnd: also wait out the real timeout (only where that is cheap); the formatter is + // checked directly for every case. + [InlineData(0.5, "<1 ms", false)] + [InlineData(300, "300 ms", true)] + [InlineData(750, "750 ms", true)] + [InlineData(1500, "1500 ms", false)] + [InlineData(2000, "2 s", false)] + [InlineData(30000, "30 s", false)] + public async Task TimeoutMessage_StatesTheConfiguredTimeoutExactly(double milliseconds, string expected, + bool endToEnd) + { + using var stream = new IdleTimeoutStream(new NeverendingStream(), TimeSpan.FromMilliseconds(milliseconds)); + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + if (endToEnd) + { + var exception = await Assert.ThrowsAsync(async () => +#pragma warning disable CA2022 // The read is expected to throw (idle timeout); the byte count is irrelevant. + await stream.ReadAsync(new byte[1], cts.Token)); +#pragma warning restore CA2022 + Assert.Equal($"SSE idle timeout: no data received within {expected}.", exception.Message); + } + + Assert.Equal(expected, IdleTimeoutStream.FormatTimeout(TimeSpan.FromMilliseconds(milliseconds))); + } + [Fact] public async Task ExternalCancel_WinsOverIdleTimeout() { using var stream = new IdleTimeoutStream(new NeverendingStream(), TimeSpan.FromSeconds(30)); using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(100)); + var stopwatch = System.Diagnostics.Stopwatch.StartNew(); // TaskCanceledException derives from OperationCanceledException. await Assert.ThrowsAsync(async () => #pragma warning disable CA2022 // The read is expected to throw (external cancel); the byte count is irrelevant. await stream.ReadAsync(new byte[16], cts.Token)); #pragma warning restore CA2022 + + // The caller's cancel reaches the inner read promptly. Without this bound a token that is + // not passed through would only show up after the 30 s idle timer, as a wrong exception type. + Assert.True(stopwatch.Elapsed < TimeSpan.FromSeconds(10), $"Cancel took {stopwatch.Elapsed}."); } [Fact] @@ -184,6 +272,58 @@ public override ValueTask DisposeAsync() } } + private sealed class SelfCancellingStream : Stream + { + public override bool CanRead => true; + public override bool CanSeek => false; + public override bool CanWrite => false; + public override long Length => throw new NotSupportedException(); + public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); } + + public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) => + ValueTask.FromException(new OperationCanceledException("inner stream's own cancellation")); + + public override void Flush() => throw new NotSupportedException(); + public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + public override void SetLength(long value) => throw new NotSupportedException(); + public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + } + + private sealed class LateFirstByteStream(TimeSpan firstDelay) : Stream + { + private bool first = true; + + public override bool CanRead => true; + public override bool CanSeek => false; + public override bool CanWrite => false; + public override long Length => throw new NotSupportedException(); + public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); } + + public override async ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) + { + if (first) + { + // Ignores the token, like a stream that does not observe cancellation mid-read. + first = false; + await Task.Delay(firstDelay, CancellationToken.None); + } + else + { + cancellationToken.ThrowIfCancellationRequested(); + } + + buffer.Span[0] = (byte)'x'; + return 1; + } + + public override void Flush() => throw new NotSupportedException(); + public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + public override void SetLength(long value) => throw new NotSupportedException(); + public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + } + private sealed class NeverendingStream : Stream { public override bool CanRead => true; diff --git a/test/Linq2GraphQL.Tests/TestSupport/GraphQLWsTestServer.cs b/test/Linq2GraphQL.Tests/TestSupport/GraphQLWsTestServer.cs index 2b05658f..16e0229c 100644 --- a/test/Linq2GraphQL.Tests/TestSupport/GraphQLWsTestServer.cs +++ b/test/Linq2GraphQL.Tests/TestSupport/GraphQLWsTestServer.cs @@ -92,6 +92,10 @@ internal sealed class GraphQLWsTestServer : IAsyncDisposable public ConcurrentQueue Incoming { get; } = new(); public List Sessions { get; } = new(); + /// Headers of every accepted upgrade request, in arrival order (auth tests). + /// Snapshot copies: Kestrel recycles the live header collection once the request ends. + public ConcurrentQueue> UpgradeHeaders { get; } = new(); + /// Set when a connection_init arrives; payload is the raw message text. public TaskCompletionSource InitReceived { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); @@ -134,6 +138,8 @@ private async Task HandleAsync(HttpContext context) } Interlocked.Exchange(ref RejectUpgrades, 0); + UpgradeHeaders.Enqueue(context.Request.Headers.ToDictionary( + header => header.Key, header => header.Value.ToString(), StringComparer.OrdinalIgnoreCase)); var subProtocol = string.IsNullOrEmpty(SubProtocol) ? null : SubProtocol; using var socket = await context.WebSockets.AcceptWebSocketAsync(subProtocol); var session = new WsTestSession(socket); @@ -214,13 +220,15 @@ internal static class WsTestGraphClientFactory { /// A GraphClient (safe mode OFF) whose SubscriptionUrl points at the test server. public static GraphClient Create(GraphQLWsTestServer server, - SubscriptionProtocol protocol = SubscriptionProtocol.GraphQLWebSocket) + SubscriptionProtocol protocol = SubscriptionProtocol.GraphQLWebSocket, + Action? configureOptions = null) { var options = new GraphClientOptions { SubscriptionProtocol = protocol, UseSafeMode = false, }; + configureOptions?.Invoke(options); var httpClient = new HttpClient { BaseAddress = new Uri(server.HttpBaseAddress) }; return new GraphClient(httpClient, Options.Create(options), new ServiceCollection().BuildServiceProvider()); } diff --git a/test/Linq2GraphQL.Tests/WsClientTests.cs b/test/Linq2GraphQL.Tests/WsClientTests.cs index 0fc2e2d9..ee707ec2 100644 --- a/test/Linq2GraphQL.Tests/WsClientTests.cs +++ b/test/Linq2GraphQL.Tests/WsClientTests.cs @@ -1,5 +1,6 @@ using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Options; +using System.Collections.ObjectModel; using System.Text; using System.Text.Json; using Linq2GraphQL.Client; @@ -586,4 +587,562 @@ public async Task MissingSubprotocol_FailsInitialConnect() Assert.Contains("subprotocol", exception.Message, StringComparison.OrdinalIgnoreCase); } + + // WebSocket upgrade auth (issue #117): request hooks and derived clients (WithAccessToken / + // WithRequestConfiguration + RequestConfigurations.BearerToken) must reach the upgrade request. + + [Fact] + public async Task BearerTokenHook_SentOnWebSocketUpgrade() + { + await using var server = await GraphQLWsTestServer.StartAsync(); + var graphClient = WsTestGraphClientFactory.Create(server) + .WithRequestConfiguration(RequestConfigurations.BearerToken(_ => ValueTask.FromResult("user-token"))); + var (client, _, _, failures) = CreateClient(graphClient); + + await using var _ = client; + await client.Start().WaitAsync(TimeSpan.FromSeconds(10)); + await WaitForIncomingAsync(server, t => t.Contains("connection_init"), TimeSpan.FromSeconds(10)); + + var headers = Assert.Single(server.UpgradeHeaders); + Assert.Equal("Bearer user-token", headers["Authorization"]); + Assert.Empty(failures); + } + + [Fact] + public async Task BearerTokenHook_SentOnApolloUpgrade() + { + await using var server = await GraphQLWsTestServer.StartAsync(); + server.SubProtocol = "graphql-ws"; + var graphClient = WsTestGraphClientFactory.Create(server, SubscriptionProtocol.ApolloWebSocket) + .WithRequestConfiguration(RequestConfigurations.BearerToken(_ => ValueTask.FromResult("apollo-token"))); + var (client, _, _, failures) = CreateClient(graphClient); + + await using var _ = client; + await client.Start().WaitAsync(TimeSpan.FromSeconds(10)); + await WaitForIncomingAsync(server, t => t.Contains("connection_init"), TimeSpan.FromSeconds(10)); + + var headers = Assert.Single(server.UpgradeHeaders); + Assert.Equal("Bearer apollo-token", headers["Authorization"]); + Assert.Empty(failures); + } + + [Fact] + public async Task ConfigureRequestOption_ReachesWebSocketUpgrade() + { + await using var server = await GraphQLWsTestServer.StartAsync(); + var graphClient = WsTestGraphClientFactory.Create(server, configureOptions: options => + options.ConfigureRequest = (request, _) => + { + request.Headers.Add("X-Correlation", "ws-42"); + return ValueTask.CompletedTask; + }); + var (client, _, _, failures) = CreateClient(graphClient); + + await using var _ = client; + await client.Start().WaitAsync(TimeSpan.FromSeconds(10)); + await WaitForIncomingAsync(server, t => t.Contains("connection_init"), TimeSpan.FromSeconds(10)); + + var headers = Assert.Single(server.UpgradeHeaders); + Assert.Equal("ws-42", headers["X-Correlation"].ToString()); + Assert.Empty(failures); + } + + [Fact] + public async Task Reconnect_AskedForFreshTokenOnEveryUpgrade() + { + await using var server = await GraphQLWsTestServer.StartAsync(); + var calls = 0; + var graphClient = WsTestGraphClientFactory.Create(server) + .WithRequestConfiguration(RequestConfigurations.BearerToken( + _ => ValueTask.FromResult($"token-{Interlocked.Increment(ref calls)}"))); + var (client, payloads, _, failures) = CreateClient(graphClient); + var subscribeCount = 0; + + server.OnMessage = async (session, text) => + { + if (!text.Contains("subscribe")) + { + return; + } + + var id = GetSubscriptionId(text); + if (Interlocked.Increment(ref subscribeCount) == 1) + { + await Task.Delay(100); + session.Abort(); + return; + } + + await session.SendNext(id, """{"customerId":"88888888-8888-8888-8888-888888888888"}"""); + }; + + await using var _ = client; + await client.Start().WaitAsync(TimeSpan.FromSeconds(10)); + await WaitUntilAsync(() => payloads.Count == 1 || failures.Count > 0, 30); + + Assert.Empty(failures); + await WaitUntilAsync(() => server.UpgradeHeaders.Count >= 2, 10); + Assert.Equal("Bearer token-1", server.UpgradeHeaders.First()["Authorization"]); + Assert.Equal("Bearer token-2", server.UpgradeHeaders.Last()["Authorization"]); + } + + // connection_init payload (issue #117): free-form object, legacy option, Blazor WASM merge. + + private static JsonElement GetInitPayload(string initMessage) + { + using var document = JsonDocument.Parse(initMessage); + return document.RootElement.GetProperty("payload").Clone(); + } + + [Fact] + public async Task ConnectionInitPayload_FreeFormObject_ArrivesVerbatim() + { + await using var server = await GraphQLWsTestServer.StartAsync(); + var graphClient = WsTestGraphClientFactory.Create(server, configureOptions: options => + options.ConnectionInitPayload = (_, _) => ValueTask.FromResult( + new Dictionary { ["authToken"] = "s3cret", ["reconnect"] = true })); + var (client, _, _, failures) = CreateClient(graphClient); + + await using var _ = client; + await client.Start().WaitAsync(TimeSpan.FromSeconds(10)); + var init = await server.InitReceived.Task.WaitAsync(TimeSpan.FromSeconds(10)); + + var payload = GetInitPayload(init); + Assert.Equal("s3cret", payload.GetProperty("authToken").GetString()); + Assert.True(payload.GetProperty("reconnect").GetBoolean()); + Assert.Empty(failures); + } + + [Fact] + public async Task ConnectionInitPayload_IsAskedAgainOnReconnect() + { + await using var server = await GraphQLWsTestServer.StartAsync(); + var calls = 0; + var graphClient = WsTestGraphClientFactory.Create(server, configureOptions: options => + options.ConnectionInitPayload = (_, _) => ValueTask.FromResult( + new Dictionary { ["generation"] = Interlocked.Increment(ref calls) })); + var (client, payloads, _, failures) = CreateClient(graphClient); + var subscribeCount = 0; + + server.OnMessage = async (session, text) => + { + if (!text.Contains("subscribe")) + { + return; + } + + var id = GetSubscriptionId(text); + if (Interlocked.Increment(ref subscribeCount) == 1) + { + await Task.Delay(100); + session.Abort(); + return; + } + + await session.SendNext(id, """{"customerId":"99999999-9999-9999-9999-999999999999"}"""); + }; + + await using var _ = client; + await client.Start().WaitAsync(TimeSpan.FromSeconds(10)); + await WaitUntilAsync(() => payloads.Count == 1 || failures.Count > 0, 30); + + Assert.Empty(failures); + var inits = server.Incoming.Where(t => t.Contains("connection_init")).ToList(); + Assert.True(inits.Count >= 2, $"expected >= 2 connection_init messages, saw {inits.Count}"); + Assert.Equal(1, GetInitPayload(inits[0]).GetProperty("generation").GetInt32()); + Assert.Equal(2, GetInitPayload(inits[^1]).GetProperty("generation").GetInt32()); + } + + [Fact] + public async Task ConnectionInitPayload_OverridesLegacyOption() + { + await using var server = await GraphQLWsTestServer.StartAsync(); + var graphClient = WsTestGraphClientFactory.Create(server, configureOptions: options => + { + options.ConnectionInitPayload = (_, _) => ValueTask.FromResult( + new Dictionary { ["authToken"] = "new-shape" }); +#pragma warning disable CS0618 // Deliberately sets the legacy option to prove precedence. + options.WSConnectionInitPayload = _ => Task.FromResult(new GraphQLRequest { Query = "legacy" }); +#pragma warning restore CS0618 + }); + var (client, _, _, failures) = CreateClient(graphClient); + + await using var _ = client; + await client.Start().WaitAsync(TimeSpan.FromSeconds(10)); + var init = await server.InitReceived.Task.WaitAsync(TimeSpan.FromSeconds(10)); + + var payload = GetInitPayload(init); + Assert.Equal("new-shape", payload.GetProperty("authToken").GetString()); + Assert.False(payload.TryGetProperty("query", out var _)); + Assert.Empty(failures); + } + + [Fact] + public async Task LegacyWSConnectionInitPayload_StillApplied() + { + await using var server = await GraphQLWsTestServer.StartAsync(); + var graphClient = WsTestGraphClientFactory.Create(server, configureOptions: options => +#pragma warning disable CS0618 // The legacy option keeps working (with a warning) after deprecation. + options.WSConnectionInitPayload = _ => Task.FromResult(new GraphQLRequest { Query = "legacy-init" })); +#pragma warning restore CS0618 + var (client, _, _, failures) = CreateClient(graphClient); + + await using var _ = client; + await client.Start().WaitAsync(TimeSpan.FromSeconds(10)); + var init = await server.InitReceived.Task.WaitAsync(TimeSpan.FromSeconds(10)); + + Assert.Equal("legacy-init", GetInitPayload(init).GetProperty("query").GetString()); + Assert.Empty(failures); + } + + [Fact] + public async Task SendAuthorizationInConnectionInit_MergesTokenIntoPayload() + { + await using var server = await GraphQLWsTestServer.StartAsync(); + var graphClient = WsTestGraphClientFactory.Create(server, configureOptions: options => + { + options.ConnectionInitPayload = (_, _) => ValueTask.FromResult( + new Dictionary { ["authToken"] = "s3cret" }); + options.SendAuthorizationInConnectionInit = true; + }) + .WithRequestConfiguration(RequestConfigurations.BearerToken(_ => ValueTask.FromResult("wasm-token"))); + var (client, _, _, failures) = CreateClient(graphClient); + + await using var _ = client; + await client.Start().WaitAsync(TimeSpan.FromSeconds(10)); + var init = await server.InitReceived.Task.WaitAsync(TimeSpan.FromSeconds(10)); + + var payload = GetInitPayload(init); + Assert.Equal("s3cret", payload.GetProperty("authToken").GetString()); + Assert.Equal("Bearer wasm-token", payload.GetProperty("Authorization").GetString()); + Assert.Empty(failures); + } + + [Fact] + public async Task SendAuthorizationInConnectionInit_CreatesPayloadWhenNoneConfigured() + { + await using var server = await GraphQLWsTestServer.StartAsync(); + var graphClient = WsTestGraphClientFactory.Create(server, configureOptions: options => + options.SendAuthorizationInConnectionInit = true) + .WithRequestConfiguration(RequestConfigurations.BearerToken(_ => ValueTask.FromResult("only-token"))); + var (client, _, _, failures) = CreateClient(graphClient); + + await using var _ = client; + await client.Start().WaitAsync(TimeSpan.FromSeconds(10)); + var init = await server.InitReceived.Task.WaitAsync(TimeSpan.FromSeconds(10)); + + var payload = GetInitPayload(init); + var property = Assert.Single(payload.EnumerateObject()); + Assert.Equal("Authorization", property.Name); + Assert.Equal("Bearer only-token", property.Value.GetString()); + Assert.Empty(failures); + } + + [Fact] + public async Task HookThrowingOnInitialConnect_FailsStart() + { + await using var server = await GraphQLWsTestServer.StartAsync(); + var graphClient = WsTestGraphClientFactory.Create(server) + .WithRequestConfiguration((_, _) => throw new InvalidOperationException("token-service-down")); + var (client, _, _, _) = CreateClient(graphClient); + + await using var _ = client; + var exception = await Assert.ThrowsAnyAsync(() => + client.Start().WaitAsync(TimeSpan.FromSeconds(10))); + + Assert.Contains("token-service-down", exception.Message); + } + + [Fact] + public async Task HookThrowingOnInitialConnect_DoesNotRetryInBackground() + { + await using var server = await GraphQLWsTestServer.StartAsync(); + var calls = 0; + var graphClient = WsTestGraphClientFactory.Create(server) + .WithRequestConfiguration((_, _) => + { + Interlocked.Increment(ref calls); + throw new InvalidOperationException("token-service-down"); + }); + var (client, _, _, failures) = CreateClient(graphClient); + // A short, fixed backoff: if the background retry loop ran at all, it would re-run the hook + // well inside the wait below (the real 1.3-4 s first backoff could outlast a fixed wait). + client.RetryDelay = _ => TimeSpan.FromMilliseconds(200); + + await using var _ = client; + await Assert.ThrowsAnyAsync(() => client.Start().WaitAsync(TimeSpan.FromSeconds(10))); + + // No background retry loop: the hook ran once (the initial connect), not again after + // the retry backoff, and no second failure is reported beyond Start's own exception. + // The hook throws before the socket connects, so no upgrade request ever left the client. + await Task.Delay(TimeSpan.FromSeconds(2)); + Assert.Equal(1, Volatile.Read(ref calls)); + Assert.Empty(failures); + Assert.Empty(server.UpgradeHeaders); + } + + [Fact] + public async Task SendAuthorizationInConnectionInit_MergesIntoReadOnlyDictionaryPayload() + { + await using var server = await GraphQLWsTestServer.StartAsync(); + var graphClient = WsTestGraphClientFactory.Create(server, configureOptions: options => + { + options.SendAuthorizationInConnectionInit = true; + options.ConnectionInitPayload = (_, _) => ValueTask.FromResult( + new ReadOnlyDictionary( + new Dictionary { ["authToken"] = "s3cret" })); + }) + .WithRequestConfiguration(RequestConfigurations.BearerToken(_ => ValueTask.FromResult("wasm-token"))); + var (client, _, _, failures) = CreateClient(graphClient); + + await using var _ = client; + await client.Start().WaitAsync(TimeSpan.FromSeconds(10)); + var init = await server.InitReceived.Task.WaitAsync(TimeSpan.FromSeconds(10)); + + var payload = GetInitPayload(init); + Assert.Equal("s3cret", payload.GetProperty("authToken").GetString()); + Assert.Equal("Bearer wasm-token", payload.GetProperty("Authorization").GetString()); + Assert.Empty(failures); + } + + public static TheoryData ObjectPayloadShapes => new() + { + "anonymous", "string-dictionary", "json-object", "json-element", + }; + + private static object CreatePayload(string shape) => shape switch + { + "anonymous" => new { authToken = "s3cret" }, + "string-dictionary" => new Dictionary { ["authToken"] = "s3cret" }, + "json-object" => new System.Text.Json.Nodes.JsonObject { ["authToken"] = "s3cret" }, + "json-element" => JsonDocument.Parse("""{"authToken":"s3cret"}""").RootElement.Clone(), + _ => throw new ArgumentOutOfRangeException(nameof(shape)), + }; + + [Theory] + [MemberData(nameof(ObjectPayloadShapes))] + public async Task SendAuthorizationInConnectionInit_MergesIntoAnyJsonObjectPayload(string shape) + { + await using var server = await GraphQLWsTestServer.StartAsync(); + var graphClient = WsTestGraphClientFactory.Create(server, configureOptions: options => + { + options.SendAuthorizationInConnectionInit = true; + options.ConnectionInitPayload = (_, _) => ValueTask.FromResult(CreatePayload(shape)); + }) + .WithRequestConfiguration(RequestConfigurations.BearerToken(_ => ValueTask.FromResult("wasm-token"))); + var (client, _, _, failures) = CreateClient(graphClient); + + await using var _ = client; + await client.Start().WaitAsync(TimeSpan.FromSeconds(10)); + var init = await server.InitReceived.Task.WaitAsync(TimeSpan.FromSeconds(10)); + + var payload = GetInitPayload(init); + Assert.Equal("s3cret", payload.GetProperty("authToken").GetString()); + Assert.Equal("Bearer wasm-token", payload.GetProperty("Authorization").GetString()); + Assert.Empty(failures); + } + + [Fact] + public async Task SendAuthorizationInConnectionInit_NonObjectPayload_Throws() + { + await using var server = await GraphQLWsTestServer.StartAsync(); + var graphClient = WsTestGraphClientFactory.Create(server, configureOptions: options => + { + options.SendAuthorizationInConnectionInit = true; + options.ConnectionInitPayload = (_, _) => ValueTask.FromResult(new[] { "s3cret" }); + }) + .WithRequestConfiguration(RequestConfigurations.BearerToken(_ => ValueTask.FromResult("wasm-token"))); + var (client, _, _, _) = CreateClient(graphClient); + + await using var _ = client; + var exception = await Assert.ThrowsAsync(() => + client.Start().WaitAsync(TimeSpan.FromSeconds(10))); + + Assert.Contains("JSON object", exception.Message); + Assert.Contains("Authorization", exception.Message); + // Rejected before connecting: no socket was opened for a payload that cannot be sent. + Assert.Empty(server.UpgradeHeaders); + } + + [Fact] + public async Task ConnectionInitPayload_ResolvedBeforeSocketOpens_OnConnectAndReconnect() + { + // The payload delegate (e.g. a token fetch) must run before the upgrade, not while an open + // socket waits for connection_init (the server's init timeout would close it, 4408). + await using var server = await GraphQLWsTestServer.StartAsync(); + var upgradesSeenByDelegate = new List(); + var graphClient = WsTestGraphClientFactory.Create(server, configureOptions: options => + options.ConnectionInitPayload = (_, _) => + { + lock (upgradesSeenByDelegate) + { + upgradesSeenByDelegate.Add(server.UpgradeHeaders.Count); + } + + return ValueTask.FromResult(new Dictionary { ["authToken"] = "s3cret" }); + }); + var (client, payloads, _, failures) = CreateClient(graphClient); + var subscribeCount = 0; + + server.OnMessage = async (session, text) => + { + if (!text.Contains("subscribe")) + { + return; + } + + var id = GetSubscriptionId(text); + if (Interlocked.Increment(ref subscribeCount) == 1) + { + await Task.Delay(100); + session.Abort(); + return; + } + + await session.SendNext(id, """{"customerId":"77777777-7777-7777-7777-777777777777"}"""); + }; + + await using var _ = client; + await client.Start().WaitAsync(TimeSpan.FromSeconds(10)); + await WaitUntilAsync(() => payloads.Count == 1 || failures.Count > 0, 30); + + Assert.Empty(failures); + lock (upgradesSeenByDelegate) + { + // Connect n's delegate ran when only the n-1 earlier upgrades had reached the server. + Assert.Equal(new[] { 0, 1 }, upgradesSeenByDelegate.Take(2)); + } + } + + [Fact] + public async Task HttpClientDefaultRequestHeaders_SentOnUpgrade_HookWinsOnConflict() + { + await using var server = await GraphQLWsTestServer.StartAsync(); + var graphClient = WsTestGraphClientFactory.Create(server, configureOptions: options => + options.ConfigureRequest = (request, _) => + { + request.Headers.Remove("X-Tenant"); + request.Headers.Add("X-Tenant", "from-hook"); + return ValueTask.CompletedTask; + }); + graphClient.HttpClient.DefaultRequestHeaders.Authorization = new("Bearer", "default-token"); + graphClient.HttpClient.DefaultRequestHeaders.Add("X-Tenant", "from-default"); + var (client, _, _, failures) = CreateClient(graphClient); + + await using var _ = client; + await client.Start().WaitAsync(TimeSpan.FromSeconds(10)); + + var headers = Assert.Single(server.UpgradeHeaders); + Assert.Equal("Bearer default-token", headers["Authorization"]); + Assert.Equal("from-hook", headers["X-Tenant"]); + Assert.Empty(failures); + } + + [Fact] + public async Task SendAuthorizationInConnectionInit_FallsBackToDefaultRequestHeaders() + { + await using var server = await GraphQLWsTestServer.StartAsync(); + var graphClient = WsTestGraphClientFactory.Create(server, configureOptions: options => + options.SendAuthorizationInConnectionInit = true); + graphClient.HttpClient.DefaultRequestHeaders.Authorization = new("Bearer", "default-token"); + var (client, _, _, failures) = CreateClient(graphClient); + + await using var _ = client; + await client.Start().WaitAsync(TimeSpan.FromSeconds(10)); + var init = await server.InitReceived.Task.WaitAsync(TimeSpan.FromSeconds(10)); + + Assert.Equal("Bearer default-token", GetInitPayload(init).GetProperty("Authorization").GetString()); + Assert.Empty(failures); + } + + private sealed class TokenServiceException(string message) : Exception(message); + + [Fact] + public async Task HookThrowingOnInitialConnect_SurfacesOriginalExceptionType() + { + await using var server = await GraphQLWsTestServer.StartAsync(); + var graphClient = WsTestGraphClientFactory.Create(server) + .WithRequestConfiguration((_, _) => throw new TokenServiceException("token-service-down")); + var (client, _, _, _) = CreateClient(graphClient); + + await using var _ = client; + // Exact type: not Websocket.Client's WebsocketException wrapper. + var exception = await Assert.ThrowsAsync(() => + client.Start().WaitAsync(TimeSpan.FromSeconds(10))); + Assert.Equal("token-service-down", exception.Message); + } + + [Fact] + public async Task HookThrowingOnReconnect_OriginalExceptionIsInnerOfTerminalFailure() + { + await using var server = await GraphQLWsTestServer.StartAsync(); + var calls = 0; + var graphClient = WsTestGraphClientFactory.Create(server, configureOptions: options => + options.SubscriptionRetryAttempts = 1) + .WithRequestConfiguration((request, _) => + { + // First connect succeeds; the reconnect's token fetch fails. + if (Interlocked.Increment(ref calls) > 1) + { + throw new TokenServiceException("refresh-failed"); + } + + return ValueTask.CompletedTask; + }); + var (client, _, _, failures) = CreateClient(graphClient); + client.RetryDelay = _ => TimeSpan.FromMilliseconds(50); + + server.OnMessage = async (session, text) => + { + if (text.Contains("subscribe")) + { + await Task.Delay(100); + session.Abort(); + } + }; + + await using var _ = client; + await client.Start().WaitAsync(TimeSpan.FromSeconds(10)); + await WaitUntilAsync(() => failures.Count > 0, 15); + + var failure = Assert.IsType(Assert.Single(failures)); + var inner = Assert.IsType(failure.InnerException); + Assert.Equal("refresh-failed", inner.Message); + } + + [Fact] + public async Task ConnectionInitPayloadThrowing_SurfacesOriginalExceptionType() + { + await using var server = await GraphQLWsTestServer.StartAsync(); + var graphClient = WsTestGraphClientFactory.Create(server, configureOptions: options => + options.ConnectionInitPayload = (_, _) => throw new TokenServiceException("payload-down")); + var (client, _, _, _) = CreateClient(graphClient); + + await using var _ = client; + await Assert.ThrowsAsync(() => client.Start().WaitAsync(TimeSpan.FromSeconds(10))); + Assert.Empty(server.UpgradeHeaders); + } + + [Fact] + public async Task SlowRequestHook_BeyondDefaultConnectTimeout_StillConnects() + { + // Websocket.Client's default ConnectTimeout is 5 s and covers the whole connection + // factory, request hooks included. WSClient raises it to 30 s so a token hook with a + // slow identity-provider round-trip does not fail the connect; this pins that. + await using var server = await GraphQLWsTestServer.StartAsync(); + var graphClient = WsTestGraphClientFactory.Create(server) + .WithRequestConfiguration(async (request, _) => + { + await Task.Delay(TimeSpan.FromSeconds(6)); + request.Headers.Authorization = new("Bearer", "slow-token"); + }); + var (client, _, _, failures) = CreateClient(graphClient); + + await using var _ = client; + await client.Start().WaitAsync(TimeSpan.FromSeconds(20)); + + Assert.Empty(failures); + var headers = Assert.Single(server.UpgradeHeaders); + Assert.Equal("Bearer slow-token", headers["Authorization"]); + } }