Production-ready subscriptions (SSE + graphql-ws) and a single UTF-8 envelope parser

CHANGELOG.md

@@ -0,0 +1,94 @@
+# Changelog
+
+All notable changes to the Linq2GraphQL client packages are documented here.
+The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
+
+## [Unreleased]
+
+### Added
+
+- `GraphSubscriptionExecute.StartWithErrorsAsync()` returns `IObservable<GraphResult<T>>`.
+  Each event carries its data AND its GraphQL `errors`. A `next` event with field errors
+  no longer terminates the subscription: partial data and later events keep flowing.
+  `StartAsync()` is unchanged and still ends on any event error (`OnError`).
+- `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.SubscriptionRetryAttempts` (WebSocket, default 5) bounds reconnects.
+- `QueryExecutor.ProcessResponse(JsonElement, ...)` and `ProcessResponse(byte[], ...)`
+  overloads for streaming callers.
+
+### Changed
+
+Behaviour changes that callers may notice. None of them changes a public signature.
+
+- **`GraphClient.SerializerOptions` is now a shared, frozen instance.** Every client in
+  the process reuses one `JsonSerializerOptions`; post-construction mutation of the
+  returned options throws `InvalidOperationException`. `GraphClientOptions` does not
+  expose serializer settings, so there is no supported way to customise them. If you
+  relied on mutating `SerializerOptions`, open an issue describing the setting you need.
+- **Extension-field reads use the shared client options.** `GetMethodValue` /
+  `GetFirstMethodValue` deserialize with `ClientJson.Default` instead of .NET's default
+  options. Only read-side behaviour is shared; `WhenWritingNull` does not affect reads.
+- **`__AdditionalProperties` is null until first used.** `GraphQLTypeBase` no longer
+  eagerly allocates the extension-data dictionary for every entity; guard with `?.` or
+  assign before use when reading extension fields.
+- **WebSocket handshake is no longer awaited by `StartAsync`.** Connection/ack failures
+  surface as `OnError` on the subscription stream (previously thrown from `StartAsync`).
+  Per-message parse errors also terminate the stream via `OnError` (previously silently
+  swallowed). Applies to the WS transport; SSE already behaved this way.
+- **An empty HTTP 200 body now yields `Data = default` instead of throwing.** The body
+  used to be parsed with `JsonDocument.Parse`, which threw on an empty response. Callers
+  that relied on the exception to detect a broken response must check
+  `GraphResult.HasData` / `HasErrors` explicitly.
+- **`QueryExecutor.ProcessResponseFull(string, name)` / `ProcessResponse(string, ...)`: an
+  empty string no longer throws.** They used to parse with `JsonDocument.Parse`, which threw
+  `JsonException` on `""`; they now return a result with `Data = default` and no errors, the
+  same as an empty HTTP body. A `null` string throws `ArgumentNullException`; malformed or
+  non-object JSON still throws `JsonException`.
+- **One envelope parser for every response path.** Typed queries, `GraphClient.ExecuteAsync`
+  (GraphQL text), the public `ProcessResponse` / `ProcessResponseFull` overloads (`string`,
+  `byte[]`, `JsonElement`) and subscription payloads all parse `data` / `errors` /
+  `extensions` with the internal `GraphEnvelopeReader` (`Utf8JsonReader`, no
+  `JsonDocument`), so they can no longer disagree. The `JsonElement` overloads now reject
+  a non-object root with `JsonException` instead of `InvalidOperationException`.
+- **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.
+- **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.
+  After a fatal close the transport no longer reconnects in the background.
+- **WebSocket reconnect is owned by the subscription** (not by Websocket.Client's own
+  loop, which retries forever and goes silent after an error). A dropped connection
+  retries up to `SubscriptionRetryAttempts` times with exponential backoff plus
+  300-3000 ms jitter (graphql-ws default). A successful `connection_ack` resets the
+  budget. When the budget is spent, the subscription fails once with an `IOException`
+  whose inner exception is the last failure. Set to 0 to fail on the first drop. Every
+  attempt opens a fresh socket.
+- A WebSocket connection that delivers no frame for 30 s is treated as lost and
+  reconnected (the same window as the previous `ReconnectTimeout`, now enforced by the
+  subscription itself).
+
+### Fixed
+
+- A server `complete` frame now completes the typed subscription observable. Previously
+  the observable stayed open after the server finished the stream.
+- A `Fault` overflow in one subscriber no longer starves other subscribers of the same
+  observable.
+- `WSConnectionInitPayload` default returns a completed task instead of a null task.
+- `SSEClient` dispose race: the response handed off under the lock is now used via the
+  local, so `Dispose()` during connect can no longer cause an NRE or a spurious
+  failure report.
+
+### Performance
+
+- Responses are read as a stream and parsed directly from UTF-8 bytes
+  (`GraphEnvelopeReader`): no `JsonDocument`, no UTF-16 round trip, sibling root fields
+  are skipped without materializing them, and the buffered body is parsed in place
+  (no second copy).
+- One shared frozen `JsonSerializerOptions` per process instead of per-client metadata
+  rebuilds.
+- SSE events are delivered to subscribers as UTF-8 bytes; string decoding only happens
+  for listeners actually attached to `Subscription`.

CLAUDE.md

@@ -47,13 +47,14 @@ The query pipeline is a chain of three stages; understanding it is usually the w
 
 1. **Build** — generated `QueryMethods`/`MutationMethods` return `GraphQuery<T>` (`GraphBase<T,TGraph>`), seeded with a root `QueryNode` carrying the field name and `ArgumentValue`s.
 2. **Parse** — `Include(...)` and `Select(...)` lambdas go through `Utilities.ParseExpression` → `Visitors/QueryExpressionVisitor`. This is the heart of the library. It has two modes: `ResolvePath` handles expressions that *name* a field (member chains, `[GraphQLMember]` method calls, LINQ operators from `Visitors/LinqOperator`) and returns/creates the node; anything else (object initializers, comparisons, string calls) is walked as a plain `ExpressionVisitor` so every field mentioned still lands in the query. Lambda parameters are bound to nodes in a `scopes` dictionary **keyed by ParameterExpression reference**, not name, so nested lambdas reusing a name stay distinct.
-3. **Execute** — `GraphBaseExecute` lazily initializes (`SetAllUniqueVariableNames`, `AddPrimitiveChildren`), renders the query text from the `QueryNode` tree, and `QueryExecutor<T>` POSTs it and unwraps `data`/`errors`/`extensions`.
+3. **Execute** — `GraphBaseExecute` lazily initializes (`SetAllUniqueVariableNames`, `AddPrimitiveChildren`), renders the query text from the `QueryNode` tree, and `QueryExecutor<T>` POSTs it and unwraps `data`/`errors`/`extensions` with `Common/GraphEnvelopeReader`.
 
 Key details that bite:
 
 - **`QueryNode`** (`QueryNode.cs`) is the whole intermediate representation. When a field is requested more than once with different arguments, `Utilities.GetArgumentsId` derives a hash suffix used as a GraphQL **alias** — the same computation must reproduce the alias when deserializing the response, so changing argument hashing breaks reads as well as writes.
 - **Safe mode** (`GraphClientOptions.UseSafeMode`) makes the client run an introspection query once (cached in `IMemoryCache`, keyed by base address) so auto-included primitive fields can be validated against the real schema. Off by default; the test fixtures turn it **on**, which is why tests require `IMemoryCache` in the container.
-- **Errors** have two surfaces: `ExecuteAsync` throws `GraphQueryExecutionException`; `ExecuteWithResultAsync` returns `GraphResult<T>` with `Errors`/`Extensions`. `QueryExecutor.ProcessResponseFull` is the single place both go through — keep them in sync.
+- **Errors** have two surfaces: `ExecuteAsync` throws `GraphQueryExecutionException`; `ExecuteWithResultAsync` returns `GraphResult<T>` with `Errors`/`Extensions`. Both go through the same parser.
+- **`Common/GraphEnvelopeReader` is the single parser for the response envelope** (`data`/`errors`/`extensions`) and every path goes through it: typed queries (`QueryExecutor.ExecuteRawAsync` streams the body into a UTF-8 buffer), `GraphClient.ExecuteAsync` for GraphQL text (`ReadDocument` returns the whole `data` as a `JsonElement`), the public `QueryExecutor.ProcessResponse`/`ProcessResponseFull` overloads (`string` is transcoded to UTF-8, `JsonElement` hands over its raw UTF-8 text) and subscription payloads. It reads with `Utf8JsonReader` and deserializes only the requested root field, skipping siblings. Do not add a second envelope parser (no `JsonDocument.Parse` of a response); `EnvelopeAndResponseTests.AllOverloads_ProduceTheSameResult` guards it. An empty body yields `Data = default` with no errors rather than throwing.
 - **Attributes** (`Attributes/`) are the contract between generated code and the runtime: `GraphQLMemberAttribute` maps a CLR member to its GraphQL name and marks interface/extension members, `GraphQLArgumentAttribute` carries the GraphQL type string used to emit variable declarations.
 - **Subscriptions** live in a separate package with two transports (`WSClient` for graphql-ws, `SSEClient` for server-sent events). Only SSE works under the test host.
 

README.md

@@ -389,6 +389,56 @@ await foreach (var orders in pager.GetAllPagesAsync(maxPages: 50))
 
 `GetAllPagesWithResultAsync` yields each page's `GraphResult<T>` and stops after the first page with errors.
 
+# Subscriptions
+
+`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.
+- `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.
+
+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.
+
+Register them through DI like any other options:
+
+```cs
+services.AddSampleClient(options =>
+{
+    options.SubscriptionProtocol = SubscriptionProtocol.ServerSentEvents;
+    options.SubscriptionMaxBufferedEvents = 256;
+    options.SubscriptionOverflowPolicy = SubscriptionOverflowPolicy.DropOldest;
+    options.SubscriptionIdleTimeout = TimeSpan.FromSeconds(30);
+});
+```
+
+Guarantees:
+
+- Events are delivered in order, at most one message at a time (no concurrent observer callbacks).
+- Each subscriber gets its own buffer, so one slow or overflowing observer does not affect another.
+- A bounded buffer decouples transport from the observer; with the default `Fault` policy an overflow faults the subscription with `InvalidOperationException` instead of dropping events silently.
+- The first terminal transport error (initial connection failure, server `connection_error`, fatal close code, or an exhausted reconnect budget) terminates the stream via `OnError`.
+- A malformed GraphQL response for a single event also terminates the stream.
+
+Not supported: replay of missed events, incremental delivery (`@defer`/`@stream` partial payloads).
+
+```cs
+var subscription = await sampleClient
+    .Subscription
+    .CustomerAdded()
+    .Select()
+    .StartAsync();
+
+subscription.Subscribe(
+    customer => Console.WriteLine(customer.CustomerName),
+    error => Console.WriteLine(error.Message),
+    () => Console.WriteLine("completed"));
+```
+
 # Acknowledgments
 
 Linq2GraphQL is inspired by [GraphQLinq](https://github.com/Giorgi/GraphQLinq) , thank

src/Linq2GraphQL.Client.Subscriptions/Assembly.cs

@@ -0,0 +1,3 @@
+using System.Runtime.CompilerServices;
+
+[assembly: InternalsVisibleTo("Linq2GraphQL.Tests")]