From 5ff683a5ae851b1c424be45c527fc65e6fb30224 Mon Sep 17 00:00:00 2001 From: Rickard Redler Date: Fri, 16 Oct 2026 20:23:15 +0200 Subject: [PATCH 1/5] fix(subscriptions): run request hooks on WebSocket upgrades, free-form connection_init payload WSClient never called ConfigureRequestAsync, so WithAccessToken/ WithRequestConfiguration/ConfigureRequest reached queries and SSE but not the WebSocket subscription upgrade (issue #117). The connection factory is now async (Websocket.Client Func>) and runs the hooks on a dummy GET for the subscription URL before every connect - initial and each reconnect - copying the resulting headers onto the upgrade request (handshake-owned headers skipped). On browsers, where custom WebSocket headers are impossible, the hook's Authorization value is captured for the init payload instead. GraphClientOptions.ConnectionInitPayload (new) sets the connection_init payload to any JSON object (graphql-ws/Apollo connection-parameter convention, e.g. Authorization/authToken); it is called before every (re)connect. SendAuthorizationInConnectionInit (new) merges the hook's Authorization header into that payload for Blazor WebAssembly. WSConnectionInitPayload is obsolete (still honored as fallback). Tests: upgrade header assertions for both WS protocols (server records snapshotted upgrade headers), token refresh across reconnect, free-form and legacy payloads, merge behaviour, hook failure on initial connect. --- CHANGELOG.md | 23 ++ README.md | 12 +- .../Pages/Subscriptions.razor | 32 ++- .../WSClient.cs | 102 ++++++- src/Linq2GraphQL.Client/GraphClient.cs | 22 ++ src/Linq2GraphQL.Client/GraphClientOptions.cs | 32 ++- .../GraphClientOptionsTests.cs | 4 + .../TestSupport/GraphQLWsTestServer.cs | 10 +- test/Linq2GraphQL.Tests/WsClientTests.cs | 265 ++++++++++++++++++ 9 files changed, 486 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6262789e..481b2f69 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,29 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). - `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. 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). +- `GraphClientOptions.ConnectionInitPayload` sets the WebSocket `connection_init` payload. + Unlike the old option it accepts any JSON object, e.g. + `["Authorization"] = $"Bearer {token}"` (the graphql-ws/Apollo connection-parameter + convention) or a server-specific `authToken`. Called before every (re)connect. +- `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 a + `ConnectionInitPayload` dictionary. + +### 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 diff --git a/README.md b/README.md index 1b83d63b..21ed828c 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. 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 over (and wins on key conflicts with) a `ConnectionInitPayload` dictionary. +- WebSocket upgrade requests carry the headers produced by `ConfigureRequest`/`WithAccessToken` (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/WSClient.cs b/src/Linq2GraphQL.Client.Subscriptions/WSClient.cs index b784d801..bfb06d91 100644 --- a/src/Linq2GraphQL.Client.Subscriptions/WSClient.cs +++ b/src/Linq2GraphQL.Client.Subscriptions/WSClient.cs @@ -42,6 +42,22 @@ public class WSClient : IAsyncDisposable // reconnection is enabled, which this class disables (verified: no NoMessageReceived event). private Timer watchdog; private long lastMessageTicks; + // Authorization header produced by the request hooks of the CURRENT connection; read when + // building the connection_init payload (SendAuthorizationInConnectionInit). Written in the + // connection factory before SendConnectionInitAsync runs for the same generation. + private volatile string connectionAuthorization; + + // 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,14 +65,19 @@ 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()); + await ApplyRequestHooksAsync(socket.Options, cancellationToken).ConfigureAwait(false); + await socket.ConnectAsync(uri, cancellationToken).ConfigureAwait(false); return socket; }); - 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. @@ -68,6 +89,37 @@ public WSClient(GraphClient graphClient, GraphQLRequest payload) disconnectionSubscription = client.DisconnectionHappened.Subscribe(HandleDisconnection, ReportFailure); } + 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); + + connectionAuthorization = request.Headers.Authorization?.ToString(); + + // Browsers do not allow custom headers on WebSocket requests: the Authorization value + // captured above travels in the connection_init payload instead (Blazor WebAssembly). + if (OperatingSystem.IsBrowser()) + { + return; + } + + foreach (var (key, values) in request.Headers) + { + if (HandshakeOwnedHeaders.Contains(key)) + { + continue; + } + + var joined = string.Join(", ", values); + if (!string.IsNullOrEmpty(joined)) + { + socketOptions.SetRequestHeader(key, joined); + } + } + } + public IObservable Subscription => subscriptionSubject.AsObservable(); internal IObservable Payloads => payloadSubject.AsObservable(); public IObservable Errors => errorSubject.AsObservable(); @@ -214,11 +266,10 @@ private static TaskCompletionSource CreateAckSource() private async Task SendConnectionInitAsync() { - var initRequest = new WebsocketRequest(WebsocketRequestTypes.CONNECTION_INIT); - if (graphClient.WSConnectionInitPayload is not null) + var initRequest = new WebsocketRequest(WebsocketRequestTypes.CONNECTION_INIT) { - initRequest.Payload = await graphClient.WSConnectionInitPayload(graphClient).ConfigureAwait(false); - } + Payload = await BuildConnectionInitPayloadAsync(cancellationSource.Token).ConfigureAwait(false) + }; SendRequest(initRequest); lock (stateLock) @@ -231,6 +282,45 @@ private async Task SendConnectionInitAsync() SendSubscription(); } + private async ValueTask BuildConnectionInitPayloadAsync(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) + { + return payload; + } + + var authorization = connectionAuthorization; + if (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 Dictionary { ["Authorization"] = authorization }; + } + + if (payload is IDictionary mergeable) + { + var merged = new Dictionary(mergeable, StringComparer.Ordinal) + { + ["Authorization"] = authorization + }; + return merged; + } + + throw new InvalidOperationException( + "SendAuthorizationInConnectionInit requires the ConnectionInitPayload to be a dictionary " + + $"(or 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 diff --git a/src/Linq2GraphQL.Client/GraphClient.cs b/src/Linq2GraphQL.Client/GraphClient.cs index 43f0f094..fc16070a 100644 --- a/src/Linq2GraphQL.Client/GraphClient.cs +++ b/src/Linq2GraphQL.Client/GraphClient.cs @@ -99,8 +99,30 @@ public Task> ExecuteAsync(string query, IDictionary options.Value; + 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..a1304cee 100644 --- a/src/Linq2GraphQL.Client/GraphClientOptions.cs +++ b/src/Linq2GraphQL.Client/GraphClientOptions.cs @@ -20,13 +20,39 @@ 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, so a refreshed token is picked up. Overrides the obsolete + /// when set. + /// + 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 () 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. The header is merged over a + /// dictionary payload from and wins on key conflicts. + /// + 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 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; } 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/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..b3d0a9bd 100644 --- a/test/Linq2GraphQL.Tests/WsClientTests.cs +++ b/test/Linq2GraphQL.Tests/WsClientTests.cs @@ -586,4 +586,269 @@ 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); + } } From 6c8d980101e32f7a7509fa0cbd678fe9a15b4aa4 Mon Sep 17 00:00:00 2001 From: Rickard Redler Date: Fri, 16 Oct 2026 20:52:23 +0200 Subject: [PATCH 2/5] fix(subscriptions): review fixes for WS auth PR From four-axis review (standards, spec, graphql-architect, performance-engineer): - ConnectTimeout 30s (was Websocket.Client default 5s): request hooks (token fetch) run inside the connect budget and a slow IdP round-trip must not fail the connect. - Failed initial connect no longer retries in the background: InitializeAsync marks the failure terminal before rethrowing from Start(), and the retry loop re-checks the terminal flag after each backoff delay. Previously a deterministic failure (e.g. a throwing token hook) was retried 5x with backoff and surfaced a second, delayed error after Start() had already thrown. - SendAuthorizationInConnectionInit merge accepts any string-keyed dictionary shape (IEnumerable>), not just IDictionary; error message and docs state the contract. - Docs: fixed broken GraphClient.WithAccessToken cref (generated-client method), documented GraphClient.ConnectionInitPayload (and added it to the CHANGELOG), wss:// note for tokens in connection_init, connect- timeout budget note, CHANGELOG dictionary example wrapped correctly. - Tests: no-background-retry regression test, IReadOnlyDictionary merge test. --- CHANGELOG.md | 20 +++++--- .../WSClient.cs | 42 +++++++++++++--- src/Linq2GraphQL.Client/GraphClient.cs | 5 ++ src/Linq2GraphQL.Client/GraphClientOptions.cs | 12 +++-- test/Linq2GraphQL.Tests/WsClientTests.cs | 50 +++++++++++++++++++ 5 files changed, 112 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 481b2f69..08354cde 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,16 +24,20 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). is picked up on reconnect. 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). -- `GraphClientOptions.ConnectionInitPayload` sets the WebSocket `connection_init` payload. - Unlike the old option it accepts any JSON object, e.g. - `["Authorization"] = $"Bearer {token}"` (the graphql-ws/Apollo connection-parameter - convention) or a server-specific `authToken`. Called before every (re)connect. + `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). +- `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. - `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 a - `ConnectionInitPayload` dictionary. + `ConnectionInitPayload` dictionary (both `IDictionary` and `IReadOnlyDictionary` string-keyed + shapes merge; anything else throws). ### Deprecated @@ -79,6 +83,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 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/src/Linq2GraphQL.Client.Subscriptions/WSClient.cs b/src/Linq2GraphQL.Client.Subscriptions/WSClient.cs index bfb06d91..8dc1c3d7 100644 --- a/src/Linq2GraphQL.Client.Subscriptions/WSClient.cs +++ b/src/Linq2GraphQL.Client.Subscriptions/WSClient.cs @@ -83,6 +83,9 @@ public WSClient(GraphClient graphClient, GraphQLRequest payload) // 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); @@ -192,7 +195,19 @@ 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 client.StartOrFail().ConfigureAwait(false); + } + catch + { + // The caller sees this exception from Start(). The transport also raises a + // DisconnectionHappened(Error) for the same failure; without marking it terminal the + // retry loop would run the full bounded budget in the background and surface a + // second, delayed error for a failure the caller already handled. + Interlocked.Exchange(ref terminalFailure, 1); + throw; + } // Initial connect: a missing subprotocol is a protocol violation and stays terminal. var negotiated = client.NativeClient?.SubProtocol; @@ -301,24 +316,29 @@ private async ValueTask BuildConnectionInitPayloadAsync(CancellationToke // 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. + // "Authorization" connection parameter). The hook value wins on key conflicts. Merging + // needs string-keyed entries, so both dictionary interfaces are accepted here. if (payload is null) { return new Dictionary { ["Authorization"] = authorization }; } - if (payload is IDictionary mergeable) + if (payload is IEnumerable> entries) { - var merged = new Dictionary(mergeable, StringComparer.Ordinal) + var merged = new Dictionary(StringComparer.Ordinal); + foreach (var (key, value) in entries) { - ["Authorization"] = authorization - }; + merged[key] = value; + } + + merged["Authorization"] = authorization; return merged; } throw new InvalidOperationException( "SendAuthorizationInConnectionInit requires the ConnectionInitPayload to be a dictionary " - + $"(or null) so the Authorization connection parameter can be merged; it was {payload.GetType()}."); + + "with string keys (or null) so the Authorization connection parameter can be merged; " + + $"it was {payload.GetType()}."); } private string ValidateNegotiatedSubprotocol(string negotiatedProtocol) @@ -553,6 +573,14 @@ private async Task RunRetryLoopAsync(Exception lastException) var delay = BackoffDelay(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() diff --git a/src/Linq2GraphQL.Client/GraphClient.cs b/src/Linq2GraphQL.Client/GraphClient.cs index fc16070a..ddfc634d 100644 --- a/src/Linq2GraphQL.Client/GraphClient.cs +++ b/src/Linq2GraphQL.Client/GraphClient.cs @@ -99,6 +99,11 @@ 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; diff --git a/src/Linq2GraphQL.Client/GraphClientOptions.cs b/src/Linq2GraphQL.Client/GraphClientOptions.cs index a1304cee..26482bad 100644 --- a/src/Linq2GraphQL.Client/GraphClientOptions.cs +++ b/src/Linq2GraphQL.Client/GraphClientOptions.cs @@ -26,8 +26,10 @@ public class GraphClientOptions /// 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, so a refreshed token is picked up. Overrides the obsolete - /// when set. + /// (re)connect, so a refreshed token is picked up, and counted within the connect timeout + /// (30 s). Overrides the obsolete when set. When + /// is enabled the payload must be a dictionary + /// with string keys (or null) so the Authorization parameter can be merged. /// public Func> ConnectionInitPayload { get; set; } @@ -38,10 +40,12 @@ public class GraphClientOptions /// /// Copies the Authorization header produced by and - /// derived clients () into the connection_init + /// 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. The header is merged over a - /// dictionary payload from and wins on key conflicts. + /// dictionary payload 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; } diff --git a/test/Linq2GraphQL.Tests/WsClientTests.cs b/test/Linq2GraphQL.Tests/WsClientTests.cs index b3d0a9bd..14c6c5cf 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; @@ -851,4 +852,53 @@ public async Task HookThrowingOnInitialConnect_FailsStart() 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); + + 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(3)); + 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); + } } From fac210707c716d8e8a8c70a243243e01ea277995 Mon Sep 17 00:00:00 2001 From: Rickard Redler Date: Fri, 16 Oct 2026 21:02:25 +0200 Subject: [PATCH 3/5] fix(subscriptions): terminal handshake failures, doc precision, coverage Self-review loop round: InitializeAsync now marks ANY initial connect/handshake failure terminal (missing subprotocol, connection_init build throw, ack timeout), not only a StartOrFail transport error. Previously the server's later close (e.g. 4408 on its own ack timeout) started the bounded retry loop after Start() had already thrown, re-running hooks and reporting a second, delayed error. Also: ConnectionInitPayload doc and the merge InvalidOperationException state the exact dictionary shape (string keys, object values); CHANGELOG wording broadened to connect-or-handshake. Tests: non-dictionary payload + SendAuthorizationInConnectionInit throws with guidance; a request hook slower than Websocket.Client's 5 s default ConnectTimeout still connects (pins the raised 30 s budget). --- CHANGELOG.md | 2 +- .../WSClient.cs | 34 ++++++++------- src/Linq2GraphQL.Client/GraphClientOptions.cs | 3 +- test/Linq2GraphQL.Tests/WsClientTests.cs | 43 +++++++++++++++++++ 4 files changed, 64 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 08354cde..ae3727c9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -83,7 +83,7 @@ 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 no longer retries in the background.** The +- **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. diff --git a/src/Linq2GraphQL.Client.Subscriptions/WSClient.cs b/src/Linq2GraphQL.Client.Subscriptions/WSClient.cs index 8dc1c3d7..00b85dac 100644 --- a/src/Linq2GraphQL.Client.Subscriptions/WSClient.cs +++ b/src/Linq2GraphQL.Client.Subscriptions/WSClient.cs @@ -198,27 +198,29 @@ private async Task InitializeAsync() try { await client.StartOrFail().ConfigureAwait(false); + + // 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); + await SendConnectionInitAsync().ConfigureAwait(false); + await WaitForAckAsync().ConfigureAwait(false); } catch { - // The caller sees this exception from Start(). The transport also raises a - // DisconnectionHappened(Error) for the same failure; without marking it terminal the - // retry loop would run the full bounded budget in the background and surface a - // second, delayed error for a failure the caller already handled. + // 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; } - // 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); - await SendConnectionInitAsync().ConfigureAwait(false); - await WaitForAckAsync().ConfigureAwait(false); StartWatchdog(); } @@ -337,8 +339,8 @@ private async ValueTask BuildConnectionInitPayloadAsync(CancellationToke throw new InvalidOperationException( "SendAuthorizationInConnectionInit requires the ConnectionInitPayload to be a dictionary " - + "with string keys (or null) so the Authorization connection parameter can be merged; " - + $"it was {payload.GetType()}."); + + "with string keys and object values (or null) so the Authorization connection parameter " + + $"can be merged; it was {payload.GetType()}."); } private string ValidateNegotiatedSubprotocol(string negotiatedProtocol) diff --git a/src/Linq2GraphQL.Client/GraphClientOptions.cs b/src/Linq2GraphQL.Client/GraphClientOptions.cs index 26482bad..88898383 100644 --- a/src/Linq2GraphQL.Client/GraphClientOptions.cs +++ b/src/Linq2GraphQL.Client/GraphClientOptions.cs @@ -29,7 +29,8 @@ public class GraphClientOptions /// (re)connect, so a refreshed token is picked up, and counted within the connect timeout /// (30 s). Overrides the obsolete when set. When /// is enabled the payload must be a dictionary - /// with string keys (or null) so the Authorization parameter can be merged. + /// with string keys and object values ( of string to + /// object, or the read-only equivalent) so the Authorization parameter can be merged. /// public Func> ConnectionInitPayload { get; set; } diff --git a/test/Linq2GraphQL.Tests/WsClientTests.cs b/test/Linq2GraphQL.Tests/WsClientTests.cs index 14c6c5cf..4ea8caaf 100644 --- a/test/Linq2GraphQL.Tests/WsClientTests.cs +++ b/test/Linq2GraphQL.Tests/WsClientTests.cs @@ -901,4 +901,47 @@ public async Task SendAuthorizationInConnectionInit_MergesIntoReadOnlyDictionary Assert.Equal("Bearer wasm-token", payload.GetProperty("Authorization").GetString()); Assert.Empty(failures); } + + [Fact] + public async Task SendAuthorizationInConnectionInit_NonDictionaryPayload_Throws() + { + await using var server = await GraphQLWsTestServer.StartAsync(); + var graphClient = WsTestGraphClientFactory.Create(server, configureOptions: options => + { + options.SendAuthorizationInConnectionInit = true; + options.ConnectionInitPayload = (_, _) => ValueTask.FromResult(new { token = "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("dictionary", exception.Message); + Assert.Contains("Authorization", exception.Message); + } + + [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"]); + } } From 57510ece8f2e392758e508037eca2dca420ed2f4 Mon Sep 17 00:00:00 2001 From: Rickard Redler Date: Sat, 17 Oct 2026 16:55:33 +0200 Subject: [PATCH 4/5] fix(subscriptions): harden WebSocket auth (#117 review follow-up) - Resolve the connection_init payload in the connection factory, before the socket opens (within ConnectTimeout), so a slow token fetch cannot run into the server's connection_init wait (graphql-ws closes with 4408). - Send HttpClient.DefaultRequestHeaders on the upgrade request (hook headers win on a name conflict) and use DefaultRequestHeaders.Authorization as the SendAuthorizationInConnectionInit fallback, matching the query path. - Merge the Authorization parameter into any payload that serializes to a JSON object (anonymous objects, Dictionary, JsonObject, JsonElement, POCOs); other shapes still throw. - Surface exceptions from request hooks and the payload delegate as their original type instead of Websocket.Client's WebsocketException wrapper (inner exception of the terminal IOException on reconnect). Cancellation stays wrapped. - Make HookThrowingOnInitialConnect_DoesNotRetryInBackground deterministic via an internal RetryDelay seam instead of outwaiting the real backoff. Tests: 10 new cases; each fix verified by a mutation that the tests kill. --- CHANGELOG.md | 21 +- README.md | 6 +- .../WSClient.cs | 170 ++++++++++---- src/Linq2GraphQL.Client/GraphClientOptions.cs | 21 +- test/Linq2GraphQL.Tests/WsClientTests.cs | 209 +++++++++++++++++- 5 files changed, 355 insertions(+), 72 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ae3727c9..a1cf08bb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,23 +21,28 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). - **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. 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 + 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). + 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. + `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 a - `ConnectionInitPayload` dictionary (both `IDictionary` and `IReadOnlyDictionary` string-keyed - shapes merge; anything else throws). + 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 diff --git a/README.md b/README.md index 21ed828c..bf245945 100644 --- a/README.md +++ b/README.md @@ -406,9 +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. -- `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. 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 over (and wins on key conflicts with) a `ConnectionInitPayload` dictionary. -- WebSocket upgrade requests carry the headers produced by `ConfigureRequest`/`WithAccessToken` (except on browsers, where custom headers are impossible). +- `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/src/Linq2GraphQL.Client.Subscriptions/WSClient.cs b/src/Linq2GraphQL.Client.Subscriptions/WSClient.cs index 00b85dac..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,10 +44,15 @@ public class WSClient : IAsyncDisposable // reconnection is enabled, which this class disables (verified: no NoMessageReceived event). private Timer watchdog; private long lastMessageTicks; - // Authorization header produced by the request hooks of the CURRENT connection; read when - // building the connection_init payload (SendAuthorizationInConnectionInit). Written in the - // connection factory before SendConnectionInitAsync runs for the same generation. - private volatile string connectionAuthorization; + // 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) @@ -71,10 +78,33 @@ public WSClient(GraphClient graphClient, GraphQLRequest payload) var factory = new Func>(async (uri, cancellationToken) => { var socket = new ClientWebSocket(); - socket.Options.AddSubProtocol(subprotocol ?? GetSubprotocolString()); - await ApplyRequestHooksAsync(socket.Options, cancellationToken).ConfigureAwait(false); - await socket.ConnectAsync(uri, cancellationToken).ConfigureAwait(false); - 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), null, factory) @@ -92,35 +122,58 @@ public WSClient(GraphClient graphClient, GraphQLRequest payload) disconnectionSubscription = client.DisconnectionHappened.Subscribe(HandleDisconnection, ReportFailure); } - private async Task ApplyRequestHooksAsync(ClientWebSocketOptions socketOptions, CancellationToken cancellationToken) + /// + /// 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); - connectionAuthorization = request.Headers.Authorization?.ToString(); + // 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 - // captured above travels in the connection_init payload instead (Blazor WebAssembly). + // returned here travels in the connection_init payload instead (Blazor WebAssembly). if (OperatingSystem.IsBrowser()) { - return; + return authorization; } - foreach (var (key, values) in request.Headers) + foreach (var (name, value) in headers) { - if (HandshakeOwnedHeaders.Contains(key)) + if (!HandshakeOwnedHeaders.Contains(name) && !string.IsNullOrEmpty(value)) { - continue; - } - - var joined = string.Join(", ", values); - if (!string.IsNullOrEmpty(joined)) - { - socketOptions.SetRequestHeader(key, joined); + socketOptions.SetRequestHeader(name, value); } } + + return authorization; } public IObservable Subscription => subscriptionSubject.AsObservable(); @@ -134,6 +187,12 @@ private async Task ApplyRequestHooksAsync(ClientWebSocketOptions socketOptions, /// 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) @@ -197,7 +256,7 @@ private async Task InitializeAsync() ackSource = CreateAckSource(); try { - await client.StartOrFail().ConfigureAwait(false); + await StartTransportAsync().ConfigureAwait(false); // Initial connect: a missing subprotocol is a protocol violation and stays terminal. var negotiated = client.NativeClient?.SubProtocol; @@ -207,7 +266,7 @@ private async Task InitializeAsync() } subprotocol = ValidateNegotiatedSubprotocol(negotiated); - await SendConnectionInitAsync().ConfigureAwait(false); + SendConnectionInit(); await WaitForAckAsync().ConfigureAwait(false); } catch @@ -224,6 +283,26 @@ private async Task InitializeAsync() 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); @@ -281,11 +360,13 @@ private static TaskCompletionSource CreateAckSource() return new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); } - private async Task SendConnectionInitAsync() + private void SendConnectionInit() { + // 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) { - Payload = await BuildConnectionInitPayloadAsync(cancellationSource.Token).ConfigureAwait(false) + Payload = connectionInitPayload }; SendRequest(initRequest); @@ -299,48 +380,39 @@ private async Task SendConnectionInitAsync() SendSubscription(); } - private async ValueTask BuildConnectionInitPayloadAsync(CancellationToken cancellationToken) + 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) - { - return payload; - } - - var authorization = connectionAuthorization; - if (string.IsNullOrEmpty(authorization)) + 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. Merging - // needs string-keyed entries, so both dictionary interfaces are accepted here. + // "Authorization" connection parameter). The hook value wins on key conflicts. if (payload is null) { - return new Dictionary { ["Authorization"] = authorization }; + return new JsonObject { ["Authorization"] = authorization }; } - if (payload is IEnumerable> entries) + // 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) { - var merged = new Dictionary(StringComparer.Ordinal); - foreach (var (key, value) in entries) - { - merged[key] = value; - } - merged["Authorization"] = authorization; return merged; } throw new InvalidOperationException( - "SendAuthorizationInConnectionInit requires the ConnectionInitPayload to be a dictionary " - + "with string keys and object values (or null) so the Authorization connection parameter " - + $"can be merged; it was {payload.GetType()}."); + "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) @@ -572,7 +644,7 @@ 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 @@ -589,12 +661,12 @@ private async Task RunRetryLoopAsync(Exception lastException) // 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/GraphClientOptions.cs b/src/Linq2GraphQL.Client/GraphClientOptions.cs index 88898383..bfad78ad 100644 --- a/src/Linq2GraphQL.Client/GraphClientOptions.cs +++ b/src/Linq2GraphQL.Client/GraphClientOptions.cs @@ -26,11 +26,13 @@ public class GraphClientOptions /// 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, so a refreshed token is picked up, and counted within the connect timeout - /// (30 s). Overrides the obsolete when set. When - /// is enabled the payload must be a dictionary - /// with string keys and object values ( of string to - /// object, or the read-only equivalent) so the Authorization parameter can be merged. + /// (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; } @@ -43,8 +45,9 @@ public class GraphClientOptions /// 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. The header is merged over a - /// dictionary payload from and wins on key conflicts. Like + /// 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. /// @@ -53,7 +56,9 @@ public class GraphClientOptions /// /// 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 are copied to the upgrade request. Use it + /// 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 diff --git a/test/Linq2GraphQL.Tests/WsClientTests.cs b/test/Linq2GraphQL.Tests/WsClientTests.cs index 4ea8caaf..ee707ec2 100644 --- a/test/Linq2GraphQL.Tests/WsClientTests.cs +++ b/test/Linq2GraphQL.Tests/WsClientTests.cs @@ -865,6 +865,9 @@ public async Task HookThrowingOnInitialConnect_DoesNotRetryInBackground() 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))); @@ -872,7 +875,7 @@ public async Task HookThrowingOnInitialConnect_DoesNotRetryInBackground() // 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(3)); + await Task.Delay(TimeSpan.FromSeconds(2)); Assert.Equal(1, Volatile.Read(ref calls)); Assert.Empty(failures); Assert.Empty(server.UpgradeHeaders); @@ -902,14 +905,51 @@ public async Task SendAuthorizationInConnectionInit_MergesIntoReadOnlyDictionary 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_NonDictionaryPayload_Throws() + 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 { token = "s3cret" }); + options.ConnectionInitPayload = (_, _) => ValueTask.FromResult(new[] { "s3cret" }); }) .WithRequestConfiguration(RequestConfigurations.BearerToken(_ => ValueTask.FromResult("wasm-token"))); var (client, _, _, _) = CreateClient(graphClient); @@ -918,8 +958,169 @@ public async Task SendAuthorizationInConnectionInit_NonDictionaryPayload_Throws( var exception = await Assert.ThrowsAsync(() => client.Start().WaitAsync(TimeSpan.FromSeconds(10))); - Assert.Contains("dictionary", exception.Message); + 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] From ae08e55d9980fa510078ea671f7e2d198b6c5165 Mon Sep 17 00:00:00 2001 From: Rickard Redler Date: Sat, 17 Oct 2026 19:00:49 +0200 Subject: [PATCH 5/5] fix(subscriptions): SSE idle timer measures only time waiting for data IdleTimeoutStream armed one shared timer at the start of each read and never disarmed it, so time the consumer spent between reads (handling an event, or a continuation starved of a thread) counted as idle. A single stall longer than the window failed a healthy stream with "no data received" - the cause of the CI flake in SlowButRegularData_DoesNotTimeOut (one stalled continuation on a loaded runner exhausted the 750 ms window). - Each read gets its own CancellationTokenSource, linked to the caller's token, with CancelAfter(idleTimeout), disposed when the read completes. Consumer time between reads never counts, and a timer that fired during a read which still returned data cannot poison the next read. - An OperationCanceledException the inner stream raises on its own is no longer mislabelled as an idle timeout. - The timeout message states the configured value exactly ("750 ms", "30 s", "<1 ms") instead of rounding to whole seconds ("1s", "0s"). Tests: 7 new cases (consumer pause, delay before first read, fired timer with data returned, message format theory, inner cancellation), and the external cancel test now also asserts that the cancel propagates promptly. Each fix is verified by a mutation that the tests kill. --- CHANGELOG.md | 4 +- .../IdleTimeoutStream.cs | 62 ++++---- src/Linq2GraphQL.Client/GraphClientOptions.cs | 4 +- .../IdleTimeoutStreamTests.cs | 140 ++++++++++++++++++ 4 files changed, 180 insertions(+), 30 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a1cf08bb..518ee4f5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,7 +14,9 @@ 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. 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/GraphClientOptions.cs b/src/Linq2GraphQL.Client/GraphClientOptions.cs index bfad78ad..98bb0ca6 100644 --- a/src/Linq2GraphQL.Client/GraphClientOptions.cs +++ b/src/Linq2GraphQL.Client/GraphClientOptions.cs @@ -84,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/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;