Adopt pending/incremental/completed delivery format

.github/copilot-instructions.md

@@ -151,6 +151,18 @@ For F# work, prefer FsLangMCP over `rg`/plain text search whenever the task depe
   ```
 
 * Every public API must have XML documentation.
+* The primary constructor of an F# class is the parenthesized parameter list that follows the type name, and it is documented as a member of its own: the type's `<summary>` stays above `type`, and the constructor's `<param>` lines go on their own lines between the type name and the opening parenthesis, indented like the parameter list. Never put the constructor's `<param>` lines into the type's comment:
+
+  ```fsharp
+  /// <summary>
+  /// Translates the events of one subscription into payloads.
+  /// </summary>
+  type internal SubscriptionPayloads
+      /// <param name="logger">The logger of the connection.</param>
+      /// <param name="options">The options of the middleware.</param>
+      (logger : ILogger, options : Options) =
+  ```
+
 * On an explicit interface implementation (`interface X with member _.M (...) = ...`), write `/// <inheritdoc />` alone instead of restating the interface member's documentation, unless this implementation has behavior worth calling out beyond what the interface already documents – write a normal `<summary>`/`<remarks>` there instead.
 * Refer to types and members through `<see cref="Type.Member"/>`, never through `<c>` or plain text. `<c>` is for literal values only (JSON, GraphQL, setting names). Refer to language keywords through `<see langword="null"/>`.
 * Split multi-paragraph documentation into `<para>` elements inside `<summary>` – bare line breaks are collapsed by documentation renderers.

RELEASE_NOTES.md

@@ -288,7 +288,10 @@
 
 * **Breaking Change** Migrated to .NET 10
 * **Breaking Change** Made Relay `Edge` a read-only struct
-* **Breaking Change** `SubscriptionExecutionResult.Data` is now `obj voption Skippable`, and the record has new `Path` and `HasNext` fields for incremental delivery
+* **Breaking Change** `SubscriptionExecutionResult.Data` is now `Skippable<obj voption>` (absent, `null`, or an object), `SubscriptionExecutionResult.Errors` is now `GQLProblemDetails list Skippable`, `SubscriptionExecutionResult.Path` was removed, and the record has new `Pending`, `Incremental`, `Completed` and `HasNext` fields for incremental delivery; `IncrementalResult.Data` follows the same `Skippable<obj voption>` contract and `IncrementalResult` gained `SubPath`
+* **Breaking Change** Removed `SubscriptionsDict`, `SubscriptionUnsubscriber` and `OnUnsubscribeAction` from `FSharp.Data.GraphQL.Shared.WebSockets`: the `graphql-transport-ws` middleware keeps its subscriptions in a per-connection registry owned by a single loop
+* **Breaking Change** `@defer` and `@stream` now declare the arguments the incremental delivery specification requires: `if: Boolean = true` and `label: String` on both, `initialCount: Int = 0` on `@stream`. `@stream` is allowed on `FIELD` only and `@defer` on `FIELD`, `FRAGMENT_SPREAD` and `INLINE_FRAGMENT`, no longer on `FRAGMENT_DEFINITION`. `if: false`, literal or through a variable, executes the field inline; `initialCount` delivers the first items with the initial payload and streams the rest; `label` is carried by the `pending` entry announcing the field
+* **Breaking Change** `GQLDeferredResponseContent.DeferredPending` gained `InitialCount`, the number of items of a streamed field delivered with the initial payload, so that the `graphql-transport-ws` translator expects the streamed items from that index
 * **Breaking Change** `BufferedStreamOptions.Interval` and `BufferedStreamOptions.PreferredBatchSize` are now `int voption`
 * **Breaking Change** `ServerMessage.Error` and `ServerRawPayload.ErrorMessages` now carry `GQLProblemDetails list` instead of `NameValueLookup list`, so an `error` message's `payload` is a standard GraphQL error array as the `graphql-transport-ws` protocol requires
 * **Breaking Change** A query or mutation whose non-null root field fails during execution now produces a `Direct` (execution) result with `null` data instead of a `RequestError`, which is now only ever produced for a request rejected before execution (validation, planning, variable or inline argument coercion, a middleware, or the executor itself failing); HTTP and `graphql-transport-ws` responses for such a failure now carry `data: null` as the spec requires, instead of omitting `data` entirely. This also changes the public `GQLResponse.Data`, `GQLResponseContent.Direct.Data`, `DeferredErrors.Data`, and `SubscriptionErrors.Data` signatures to use `voption`
@@ -301,16 +304,22 @@
 * Added `StreamBatching` to group streamed items of a `Define.TaskSeqField` into batches of a fixed size or of a size computed from the sequence, such as a page size kept with a paged SDK sequence. The `preferredBatchSize` argument of `@stream` takes precedence, and the batching function itself is evaluated lazily, only for a `@stream` query that does not supply its own `preferredBatchSize`
 * Added `Microsoft.Bcl.AsyncInterfaces` dependency of `FSharp.Data.GraphQL.Shared` for `netstandard2.0`
 * Added `Human.friendsStream` field to the Star Wars sample to demonstrate `@stream`
-* Fixed a query or mutation whose root field has an invalid inline (literal) argument, such as a custom input object validator failing, being reported as a `Direct` result with `null` data instead of a `RequestError`; inline argument coercion is now checked for every root field before any of them execute, the same as variable coercion, so a mutation no longer executes earlier root fields before rejecting the request over a later one's invalid argument
-* Fixed `Define.TaskSeqField` streaming retaining a task for every item already delivered until the sequence ends
-* Fixed `graphql-transport-ws` delivery of `@defer` and `@stream` results, which are now sent as soon as they are produced with `path` and `hasNext` instead of after a fixed 5 second delay, followed by a final payload with `hasNext: false`
+* Added the validation rules of incremental delivery: `@stream` only on list fields, no `@defer` or `@stream` in a subscription operation or on a mutation root field unless disabled with `if: false`, and labels must be string literals unique within each operation, counting the fragments it spreads
+* Added `GraphQLTransportWS.SubProtocol`, the `graphql-transport-ws` sub-protocol name
+* Fixed `graphql-transport-ws` delivery of `@defer` and `@stream` results, which are now sent as soon as they are produced instead of after a fixed 5 second delay, followed by a final payload with `hasNext: false`
+* Changed `graphql-transport-ws` incremental delivery of `@defer` and `@stream` results to the `pending`/`incremental`/`completed`/`hasNext` wire format used by graphql-js 17 and Apollo Client's `GraphQL17Alpha9Handler`, superseding the previous `data`/`path`/`hasNext` shape. Every deferred or streamed field is announced once, in a `pending` entry, and identified afterwards by a short id instead of its path. A deferred field is announced in the same payload as its own value, while a streamed field is announced as soon as the payload exposing its containing data is sent. A `@stream` field's items are always delivered to the client in list order, buffering an item that arrives out of turn until the item before it fills the gap, and a batch of items (grouped by `preferredBatchSize` or `StreamBatching`) is delivered as the `items` of a single `incremental` entry addressed by that id, rather than one payload per item
+* Added a completion signal to the engine's deferred/streamed event stream (`DeferredCompleted`), fired once after a `@defer` field's own payload and once after all of a `@stream` field's items, whether they succeeded or the source failed; used to build the `completed` entries of the new wire format
+* Changed `graphql-transport-ws` delivery of a deferred field to the addressing the specification and graphql-js/Apollo clients expect: the `pending` entry names the object containing the field and the `incremental` entry carries an object map of that one field; a field whose announcement the client never received, because the payload that should have exposed it resolved to `null` there, is no longer completed
+* Changed the `graphql-transport-ws` middleware to single-owner loops communicating through `System.Threading.Channels`: one reader of the socket, one control loop owning the subscription registry, one sender owning every write and the close of the socket, and one worker per subscription. No lock, no thread blocked on a send, a graceful close on application shutdown; a slow client's messages now queue in memory for the life of the connection instead of blocking the subscription's source
+* Changed the execution engine to work without locks: the announcements of nested deferred and streamed fields are carried as data instead of being captured under a lock at subscription time, and a streamed `Define.TaskSeqField` delivers its results through a single-reader channel, with `SemaphoreSlim` only bounding `maxConcurrency`
+* Changed `graphql-transport-ws` result payloads to omit `errors` when there are none, as the GraphQL response format requires, instead of sending an empty array
 * Fixed `graphql-transport-ws` failure on deferred and streamed results that are not objects, such as streamed list items and scalars
 * Fixed `graphql-transport-ws` dropping errors of the initial payload of a deferred result together with all its deferred results
 * Fixed `graphql-transport-ws` discarding the partial `data` of a subscription result that also had field errors, sending `null` instead
 * Fixed `graphql-transport-ws` discarding the field errors of a `Direct` (non-subscription) result, sending an empty error list instead
 * Fixed `graphql-transport-ws` stranding a subscription id forever when its deferred result completed synchronously, before it was registered
 * Fixed `graphql-transport-ws` leaving a subscription id occupied when subscribing to its result failed synchronously
-* Fixed `graphql-transport-ws` addressing a batch of streamed items (grouped by `preferredBatchSize` or `StreamBatching`) with a `path` ending in the list of the batch's own indices, such as `["numbers", [0, 1]]`, which no client can merge into the response tree; a batch is now sent as one independently addressed payload per item instead, in the batch's own order
 * Fixed `graphql-transport-ws` never sending `complete` after the `next` of a query or mutation result, as the protocol requires
 * Fixed `graphql-transport-ws` sending a request error (rejected before execution: validation, planning, variable coercion, a middleware, or the executor itself failing) as a `next` result followed by `complete`, instead of the terminal `error` message the protocol requires for it; a query or mutation whose non-null root field fails during execution still gets `next` + `complete`, since it is a result, not a request error
 * Fixed `graphql-transport-ws` throwing while serializing an `error` message or a `pong` carrying a payload, since neither was written under the `payload` property name `Utf8JsonWriter` requires
+* Removed the internal `Observable.withCompletionMarker`

docs/type-system.md

@@ -98,6 +98,8 @@ How the sequence is delivered depends on the query:
 - With `@defer` on a `Nullable (ListOf ...)` field the complete list is delivered in one deferred payload.
 - With `@stream` every item is delivered as soon as the sequence produces it and its fields are resolved. The enumeration is cancelled when the client unsubscribes.
 
+Over `graphql-transport-ws`, a `@defer` or `@stream` field is delivered using the `pending`/`incremental`/`completed`/`hasNext` format used by graphql-js 17 and Apollo Client's `GraphQL17Alpha9Handler`. Each field is announced once, in a `pending` entry, and identified afterwards by a short id rather than its path. A deferred field is announced at the path of the object containing it, in the same payload as its own value, and that value is delivered as an object map of the one field, which the client merges into the announced object; a streamed field is announced at its own path as soon as the payload exposing its containing data is sent, and its items are delivered as the `items` of `incremental` entries, always in list order, a batch of items in one entry. The `label` of `@defer` or `@stream` surfaces as `pending.label`. Both directives take `if: Boolean = true`, which executes the field inline when false, and `@stream` takes `initialCount: Int = 0`, the number of items delivered with the initial payload before the rest is streamed. A payload that carries only GraphQL errors omits the top-level `data` property instead of sending `data: null`, matching the existing request-error contract used elsewhere in the transport.
+
 Streamed items can be grouped into batches. The `preferredBatchSize` argument of `@stream`, available with `SchemaConfig.DefaultWithBufferedStream`, has priority. Otherwise the `batching` parameter of the field applies. It is either a fixed size or a function that reads the size from the source, such as the page size of a paged SDK sequence. The function is evaluated lazily: only for a `@stream` query that does not itself specify `preferredBatchSize`, so it never runs for an ordinary or `@defer` query.
 
 ```fsharp

src/FSharp.Data.GraphQL.Server.AspNetCore/FSharp.Data.GraphQL.Server.AspNetCore.fsproj

@@ -19,7 +19,13 @@
     <Compile Include="Helpers.fs" />
     <Compile Include="RequestExecutionContext.fs" />
     <Compile Include="GraphQLOptions.fs" />
-    <Compile Include="GraphQLSubscriptionsManagement.fs" />
+    <Compile Include="IncrementalDelivery.fs" />
+    <Compile Include="WebSocketErrors.fs" />
+    <Compile Include="WebSocketMessaging.fs" />
+    <Compile Include="WebSocketTransport.fs" />
+    <Compile Include="SubscriptionPayloads.fs" />
+    <Compile Include="SubscriptionWorker.fs" />
+    <Compile Include="WebSocketConnection.fs" />
     <Compile Include="GraphQLWebsocketMiddleware.fs" />
     <Compile Include="Parser.fs" />
     <Compile Include="HttpContext.fs" />

src/FSharp.Data.GraphQL.Server.AspNetCore/GraphQLRequestHandler.fs

@@ -64,24 +64,24 @@ and [<AbstractClass>] GraphQLRequestHandler<'Root>
             if logger.IsEnabled LogLevel.Debug then
                 deferred
                 |> Observable.add (function
+                    | DeferredPending (path, label, isStream, _) ->
+                        let fieldKind = if isStream then "streamed" else "deferred"
+                        logger.LogDebug ("Announced GraphQL deferred field at path: {path}", path |> Seq.map string |> Seq.toArray |> Path.Join)
+                        match label with
+                        | ValueSome label -> logger.LogDebug ("Deferred field label: {label}; kind: {kind}", label, fieldKind)
+                        | ValueNone -> logger.LogDebug ("Deferred field kind: {kind}", fieldKind)
                     | DeferredResult (data, path) ->
                         logger.LogDebug ("Produced GraphQL deferred result for path: {path}", path |> Seq.map string |> Seq.toArray |> Path.Join)
 
                         if logger.IsEnabled LogLevel.Trace then
                             logger.LogTrace ("GraphQL deferred data:\n{data}", serializeIndented data)
-                    | DeferredErrors (ValueNone, errors, path) ->
+                    | DeferredErrors (data, errors, path) ->
                         logger.LogDebug ("Produced GraphQL deferred errors for path: {path}", path |> Seq.map string |> Seq.toArray |> Path.Join)
 
                         if logger.IsEnabled LogLevel.Trace then
-                            logger.LogTrace ("GraphQL deferred errors:\n{errors}", errors)
-                    | DeferredErrors (ValueSome data, errors, path) ->
-                        logger.LogDebug (
-                            "Produced GraphQL deferred result with errors for path: {path}",
-                            path |> Seq.map string |> Seq.toArray |> Path.Join
-                        )
-
-                        if logger.IsEnabled LogLevel.Trace then
-                            logger.LogTrace ("GraphQL deferred errors:\n{errors}\nGraphQL deferred data:\n{data}", errors, serializeIndented data))
+                            logger.LogTrace ("GraphQL deferred errors:\n{errors}\nGraphQL deferred data:\n{data}", errors, serializeIndented data)
+                    | DeferredCompleted path ->
+                        logger.LogDebug ("Completed GraphQL deferred field at path: {path}", path |> Seq.map string |> Seq.toArray |> Path.Join))
 
             GQLResponse.Direct (documentId, data, errs)