Run request hooks on WebSocket upgrades; add free-form connection_init payload

CHANGELOG.md

@@ -14,10 +14,44 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 - `GraphClientOptions.SubscriptionMaxBufferedEvents` and `SubscriptionOverflowPolicy`
   (`Fault`, `DropOldest`, `KeepLatest`) control the buffer between the transport and
   the observer. Each subscriber gets its own buffer.
-- `GraphClientOptions.SubscriptionIdleTimeout` (SSE) fails a stalled stream.
+- `GraphClientOptions.SubscriptionIdleTimeout` (SSE) fails a stalled stream. The timer runs
+  only while a read waits for data, so time spent handling an event between reads never
+  counts as idle; the error states the configured timeout exactly (`750 ms`, `30 s`).
 - `GraphClientOptions.SubscriptionRetryAttempts` (WebSocket, default 5) bounds reconnects.
 - `QueryExecutor.ProcessResponse(JsonElement, ...)` and `ProcessResponse(byte[], ...)`
   overloads for streaming callers.
+- **WebSocket upgrades now run the request hooks** (`ConfigureRequest`,
+  `WithRequestConfiguration`, `WithAccessToken`): the headers they produce are copied onto
+  the upgrade request of the initial connect and of every reconnect, so a refreshed token
+  is picked up on reconnect. `HttpClient.DefaultRequestHeaders` are sent too (a hook header
+  of the same name wins), as they are for queries. Handshake-owned headers (`Host`,
+  `Sec-WebSocket-*`, ...) are skipped; on browsers, where custom WebSocket headers are
+  impossible, the hooks still run and the `Authorization` value is available for
+  `SendAuthorizationInConnectionInit` (#117). The hooks count against the connect
+  timeout, which is now 30 s (Websocket.Client's 5 s default covered only the socket
+  connect). An exception thrown by a hook (or by `ConnectionInitPayload`) reaches the
+  subscription's error path (`OnError`, `WhenConnected`) as itself, not wrapped in
+  Websocket.Client's `WebsocketException`; on a reconnect it is the inner exception of the
+  terminal `IOException`.
+- `GraphClientOptions.ConnectionInitPayload` (and the matching `GraphClient` property)
+  set the WebSocket `connection_init` payload. Unlike the old option it accepts any JSON
+  object, e.g. `new Dictionary<string, object> { ["Authorization"] = $"Bearer {token}" }`
+  (the graphql-ws/Apollo connection-parameter convention) or a server-specific
+  `authToken`. Called before every (re)connect, before the socket opens (within the connect
+  timeout), so a slow token fetch cannot hit the server's `connection_init` wait.
+- `GraphClientOptions.SendAuthorizationInConnectionInit` (default off) merges the
+  `Authorization` header produced by the request hooks into the `connection_init`
+  payload — how a per-user token reaches the server on Blazor WebAssembly, where the
+  browser does not allow headers on WebSocket requests. Wins on key conflicts with the
+  `ConnectionInitPayload`, which must serialize to a JSON object (dictionary, anonymous
+  object, POCO, `JsonObject`); any other shape throws.
+
+### Deprecated
+
+- `GraphClientOptions.WSConnectionInitPayload` / `GraphClient.WSConnectionInitPayload`: use
+  `ConnectionInitPayload`, which can return any JSON object, not just a `GraphQLRequest`.
+  The old option keeps working (and falls back) until removed; `ConnectionInitPayload`
+  wins when both are set.
 
 ### Changed
 
@@ -56,6 +90,10 @@ Behaviour changes that callers may notice. None of them changes a public signatu
 - **WebSocket: unknown message types are ignored** instead of terminating the
   subscription with `InvalidDataException`. This matches the SSE transport's tolerance for
   unknown event types and keeps forward compatibility with proxies/servers.
+- **A failed initial WebSocket connect or handshake no longer retries in the background.** The
+  exception surfaces from `StartAsync`; previously the transport's `Error` disconnect
+  also started the bounded reconnect loop, re-running request hooks up to five times and
+  reporting a second, delayed error for a failure the caller had already handled.
 - **WebSocket close codes follow the graphql-ws reference client.** Fatal: `4004`, `4005`,
   `4400`, `4401`, `4406`, `4409`, `4429`, `4500` and fatal internal `1xxx` codes.
   Retryable: `4403`, `4408`, `4499` and `4504`. Previously `4408` was treated as fatal.

README.md

@@ -235,7 +235,11 @@ var launches = await userClient.Query.Launches().Select().ExecuteAsync();
 ```
 
 `WithAccessToken` sets `Authorization: Bearer <token>` 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<TResult>`. 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<string, object> { ["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:
 

docs/Linq2GraphQL.Docs/Pages/Subscriptions.razor

@@ -72,10 +72,18 @@
     <CodeBlock Code="@Protocol"/>
     <p>
         Server-sent events go through the same HTTP pipeline as queries, so <code>ConfigureRequest</code>,
-        <code>WithAccessToken</code> and your <code>HttpClient</code> 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.
+        <code>WithAccessToken</code> and your <code>HttpClient</code> 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.
     </p>
+    <p>
+        Blazor WebAssembly is the exception: the browser does not allow custom headers on WebSocket requests.
+        Set <code>SendAuthorizationInConnectionInit = true</code> to send the hook's
+        <code>Authorization</code> header inside the <code>connection_init</code> payload instead — the
+        graphql-ws/Apollo convention of an <code>Authorization</code> connection parameter. For any other
+        connection parameters, use <code>ConnectionInitPayload</code>:
+    </p>
+    <CodeBlock Code="@ConnectionInit"/>
 </section>
 
 <section class="doc-section">
@@ -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<object>(
+                    new Dictionary<string, object>
+                    {
+                        ["authToken"] = tokenService.CurrentToken,
+                        ["clientVersion"] = "1.4",
+                    });
+            });
+        """;
+
     private const string WithErrors = """
         var events = await sampleClient
             .Subscription

src/Linq2GraphQL.Client.Subscriptions/IdleTimeoutStream.cs

@@ -2,15 +2,15 @@ namespace Linq2GraphQL.Client.Subscriptions;
 
 /// <summary>
 ///     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 <c>:</c> 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 <c>:</c> 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.
 /// </summary>
 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<int> ReadAsync(Memory<byte> 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<int> ReadInnerAsync(Memory<byte> 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)}.");
         }
     }
 
+    /// <summary>
+    ///     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 "&lt;1 ms", not "0 s". Expects a positive timeout, as
+    ///     the constructor enforces.
+    /// </summary>
+    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);
     }