Added Define.TaskSeqField with streaming of IAsyncEnumerable results

Packages.props

@@ -21,6 +21,8 @@
     <PackageReference Update="FsToolkit.ErrorHandling" Version="$(FsToolkitVersion)" />
     <PackageReference Update="FsToolkit.ErrorHandling.TaskResult" Version="$(FsToolkitVersion)" />
     <PackageReference Update="Giraffe" Version="7.*" />
+    <PackageReference Update="IcedTasks" Version="0.11.*" />
+    <PackageReference Update="Microsoft.Bcl.AsyncInterfaces" Version="$(SystemVersion)" />
     <PackageReference Update="Microsoft.Extensions.Http" Version="$(MicrosoftExtensionsVersion)" />
     <PackageReference Update="Microsoft.Extensions.Logging.Abstractions" Version="$(MicrosoftExtensionsVersion)" />
     <PackageReference Update="Microsoft.NETCore.Platforms" Version="$(SystemVersion)" />
@@ -67,9 +69,11 @@
     <PackageReference Update="xunit.runner.visualstudio" Version="3.1.4" />
   </ItemGroup>
   <ItemGroup Label="Tests and Samples">
+    <PackageReference Update="Azure.Core" Version="1.*" />
     <PackageReference Update="CommandLineParser" Version="2.9.*" />
     <PackageReference Update="Donald" Version="10.1.0" />
     <PackageReference Update="EntityFramework" Version="1.*" />
+    <PackageReference Update="FSharp.Control.TaskSeq" Version="1.*" />
     <PackageReference Update="FSharp.Data.TypeProviders" Version="1.*" />
     <PackageReference Update="GraphQL.Server.Ui.Altair" Version="8.*" />
     <PackageReference Update="GraphQL.Server.Ui.GraphiQL" Version="8.*" />

README.md

@@ -111,7 +111,7 @@ This boilerplate code can be easily reduced with a built-in implementation:
 
 ```fsharp
 let streamOptions =
-    { Interval = Some 2000; PreferredBatchSize = None }
+    { Interval = ValueSome 2000; PreferredBatchSize = ValueNone }
 let schemaConfig =
     SchemaConfig.DefaultWithBufferedStream(streamOptions)
 ```

RELEASE_NOTES.md

@@ -288,6 +288,29 @@
 
 * **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** `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`
 * Added case-insensitive string comparison support to `ObjectListFilter`, including comparer-aware filter cases and GraphQL filter suffix handling
 * Improved Relay XML documentation comments
 * Changed query planning to throw `MalformedGQLQueryException` for invalid queries, `NotSupportedException` for unsupported type definition implementations and `InvalidOperationException` for internal planning errors instead of `System.Exception`, with messages naming the affected field, type and execution kind
+* Added `Define.TaskSeqField` for list fields resolved from `IAsyncEnumerable<'T>`, such as `taskSeq { }` or Azure SDK `AsyncPageable<T>`. Without directives the sequence is enumerated into a list, `@defer` delivers the whole list, and `@stream` delivers every item as soon as it is produced and its fields are resolved
+* Added cancellation of a streamed `Define.TaskSeqField` enumeration when the client unsubscribes, and delivery of a failure raised acquiring the sequence's enumerator, while enumerating, or disposing it, as a deferred error for the field, after every item already pulled has been resolved and delivered, so a slower item can never be overtaken by the error that follows it; an item resolution that throws stops the enumeration and is delivered the same way, while an item whose own fields fail is delivered as that item's deferred errors and streaming continues, exactly as for `@stream` on an ordinary list; a concurrency slot is never leaked even if delivering an item's result fails
+* Added `maxConcurrency` to `Define.TaskSeqField`, bounding how many items of a streamed sequence are pulled and resolved at the same time; defaults to `Environment.ProcessorCount`
+* 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`
+* 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

docs/type-system.md

@@ -78,6 +78,52 @@ let rec Person = Define.Object(name = "Person", fieldsFn = fun () -> [
 
 As you may see, we defined Person object definition using *rec* keyword and instead of defining fields as a list and we used a lazily evaluated function instead.
 
+### Defining fields backed by asynchronous sequences
+
+When a list comes from an asynchronous source, such as a database cursor or a paged SDK client, use `Define.TaskSeqField`. Its resolver returns `IAsyncEnumerable<'T>`, which is what the `taskSeq { }` computation expression from [FSharp.Control.TaskSeq](https://github.com/fsprojects/FSharp.Control.TaskSeq) and C# async iterators produce.
+
+```fsharp
+let getOrders (customerId : int) = taskSeq {
+    for page in 0 .. 10 do
+        let! orders = db.GetOrdersPageAsync (customerId, page)
+        yield! orders
+}
+
+Define.TaskSeqField("orders", ListOf Order, fun _ customer -> getOrders customer.Id)
+```
+
+How the sequence is delivered depends on the query:
+
+- Without directives the sequence is enumerated completely and returned as a regular list.
+- 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.
+
+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
+Define.TaskSeqField("orders", ListOf Order, (fun _ customer -> getOrders customer.Id), batching = StreamBatching.Fixed 50)
+
+Define.TaskSeqField(
+    "blobs",
+    ListOf Blob,
+    (fun _ container -> listBlobs container),
+    batching = StreamBatching.FromSource (function
+        | :? PagedSequence<BlobItem> as paged -> ValueSome paged.PageSize
+        | _ -> ValueNone))
+```
+
+Azure SDK `AsyncPageable<T>` does not expose its page size, because the size is only a hint passed to `AsPages`. To batch its items by pages, keep the hint in your own type, for example a subclass of `AsyncPageable<T>` or a wrapper, and read it in `StreamBatching.FromSource`.
+
+With `@stream`, at most `maxConcurrency` items are pulled from the sequence and resolved at the same time; enumeration waits for one of them to complete before pulling the next, so a fast or infinite source cannot outrun resolution. It defaults to `Environment.ProcessorCount`.
+
+```fsharp
+Define.TaskSeqField("orders", ListOf Order, (fun _ customer -> getOrders customer.Id), maxConcurrency = 4)
+```
+
+An error raised while enumerating the source is delivered after every item already pulled has been resolved and delivered, so a slow item can never be overtaken by a failure that follows it. An item whose own fields fail is delivered as that item's deferred errors, and the following items are still streamed, exactly as for `@stream` on an ordinary list; only an exception that escapes the item's resolution, or the source itself, ends the stream.
+
+Resolvers are captured as F# quotations. A `taskSeq { }` block that uses `let!` or `yield!` cannot be written inline in the resolver lambda, so define it in a separate function as shown above. Fields defined this way do not support `WithResolveMiddleware`.
+
 ## Defining an Interface 
 
 GraphQL interfaces are so called abstract types (along with unions). This means, that they can be used as part of the query, however query materialization must always be bound to some concrete Object type definition.

samples/star-wars-api/Schema.fs

@@ -2,7 +2,9 @@ namespace FSharp.Data.GraphQL.Samples.StarWarsApi
 
 open System.Linq
 open System.Text.Json.Serialization
+open System.Threading.Tasks
 open Microsoft.FSharp.Reflection
+open FSharp.Control
 open FSharp.Data.GraphQL
 open FSharp.Data.GraphQL.Types
 open FSharp.Data.GraphQL.Server.Relay
@@ -141,6 +143,17 @@ module Schema =
 
     let getCharacter id = characters |> List.tryFind (matchesId id)
 
+    /// Produces friends one by one with a delay, which demonstrates the @stream directive.
+    /// TaskSeq functions are used instead of a taskSeq block, because a taskSeq block compiled
+    /// without optimizations does not resume correctly after an await.
+    let getFriendsStream (friendIds : string list) =
+        friendIds
+        |> TaskSeq.ofList
+        |> TaskSeq.chooseAsync (fun id -> task {
+            do! Task.Delay 500
+            return getCharacter id
+        })
+
     let EpisodeType =
         Define.Enum (
             name = "Episode",
@@ -226,6 +239,12 @@ module Schema =
                             con
                     )
                     Define.Field ("appearsIn", ListOf EpisodeType, "Which movies they appear in.", (fun _ (h : Human) -> h.AppearsIn))
+                    Define.TaskSeqField (
+                        "friendsStream",
+                        ListOf CharacterType,
+                        "The friends of the human produced one by one. Request the field with @stream to receive each friend as soon as it is available.",
+                        fun _ (h : Human) -> getFriendsStream h.Friends
+                    )
                     Define.Field ("homePlanet", Nullable StringType, "The home planet of the human, or null if unknown.", (fun _ h -> h.HomePlanet))
                 ]
         )

samples/star-wars-api/star-wars-api.fsproj

@@ -7,6 +7,7 @@
   </PropertyGroup>
 
   <ItemGroup Label="PackageReferences">
+    <PackageReference Include="FSharp.Control.TaskSeq" />
     <PackageReference Include="FsToolkit.ErrorHandling.TaskResult" />
     <PackageReference Include="GraphQL.Server.Ui.Altair" />
     <PackageReference Include="GraphQL.Server.Ui.GraphiQL" />

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

@@ -11,6 +11,10 @@
     <FrameworkReference Include="Microsoft.AspNetCore.App" />
   </ItemGroup>
 
+  <ItemGroup Label="InternalsVisibleTo">
+    <InternalsVisibleTo Include="FSharp.Data.GraphQL.Tests" />
+  </ItemGroup>
+
   <ItemGroup>
     <Compile Include="Helpers.fs" />
     <Compile Include="RequestExecutionContext.fs" />

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

@@ -57,7 +57,7 @@ and [<AbstractClass>] GraphQLRequestHandler<'Root>
             if logger.IsEnabled LogLevel.Trace then
                 logger.LogTrace ("GraphQL response data:\n{data}", serializeIndented data)
 
-            GQLResponse.Direct (documentId, data, errs)
+            GQLResponse.Direct (documentId, data |> ValueOption.toObj, errs)
         | Deferred (data, errs, deferred) ->
             logger.LogDebug ("Produced deferred GraphQL response with documentId = '{documentId}' and metadata:\n{metadata}", documentId, metadata)
 
@@ -69,12 +69,12 @@ and [<AbstractClass>] GraphQLRequestHandler<'Root>
 
                         if logger.IsEnabled LogLevel.Trace then
                             logger.LogTrace ("GraphQL deferred data:\n{data}", serializeIndented data)
-                    | DeferredErrors (null, errors, path) ->
+                    | DeferredErrors (ValueNone, 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 (data, errors, path) ->
+                    | DeferredErrors (ValueSome data, errors, path) ->
                         logger.LogDebug (
                             "Produced GraphQL deferred result with errors for path: {path}",
                             path |> Seq.map string |> Seq.toArray |> Path.Join
@@ -96,12 +96,12 @@ and [<AbstractClass>] GraphQLRequestHandler<'Root>
 
                         if logger.IsEnabled LogLevel.Trace then
                             logger.LogTrace ("GraphQL subscription data:\n{data}", serializeIndented data)
-                    | SubscriptionErrors (null, errors) ->
+                    | SubscriptionErrors (ValueNone, errors) ->
                         logger.LogDebug ("Produced GraphQL subscription errors")
 
                         if logger.IsEnabled LogLevel.Trace then
                             logger.LogTrace ("GraphQL subscription errors:\n{errors}", errors)
-                    | SubscriptionErrors (data, errors) ->
+                    | SubscriptionErrors (ValueSome data, errors) ->
                         logger.LogDebug ("Produced GraphQL subscription result with errors")
 
                         if logger.IsEnabled LogLevel.Trace then

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

@@ -6,9 +6,9 @@ let addSubscription
     (id : SubscriptionId, unsubscriber : SubscriptionUnsubscriber, onUnsubscribe : OnUnsubscribeAction)
     (subscriptions : SubscriptionsDict)
     =
-    subscriptions.Add (id, (unsubscriber, onUnsubscribe))
+    lock subscriptions (fun () -> subscriptions.Add (id, (unsubscriber, onUnsubscribe)))
 
-let isIdTaken (id : SubscriptionId) (subscriptions : SubscriptionsDict) = subscriptions.ContainsKey (id)
+let isIdTaken (id : SubscriptionId) (subscriptions : SubscriptionsDict) = lock subscriptions (fun () -> subscriptions.ContainsKey (id))
 
 let executeOnUnsubscribeAndDispose (id : SubscriptionId) (subscription : SubscriptionUnsubscriber * OnUnsubscribeAction) =
     match subscription with
@@ -19,15 +19,28 @@ let executeOnUnsubscribeAndDispose (id : SubscriptionId) (subscription : Subscri
             unsubscriber.Dispose ()
 
 let removeSubscription (id : SubscriptionId) (subscriptions : SubscriptionsDict) =
-    match subscriptions.TryGetValue id with
-    | true, sub ->
-        sub |> executeOnUnsubscribeAndDispose id
-        subscriptions.Remove (id) |> ignore
-    | false, _ -> ()
+    let subscription =
+        lock subscriptions (fun () ->
+            match subscriptions.TryGetValue id with
+            | true, sub ->
+                subscriptions.Remove (id) |> ignore
+                ValueSome sub
+            | false, _ -> ValueNone)
+
+    match subscription with
+    | ValueSome sub -> sub |> executeOnUnsubscribeAndDispose id
+    | ValueNone -> ()
 
 let removeAllSubscriptions (subscriptions : SubscriptionsDict) =
-    subscriptions
-    |> Seq.iter (fun subscription ->
-        subscription.Value
-        |> executeOnUnsubscribeAndDispose subscription.Key)
-    subscriptions.Clear ()
+    let subscriptionsToDispose =
+        lock subscriptions (fun () ->
+            let snapshot =
+                subscriptions
+                |> Seq.map (fun subscription -> struct (subscription.Key, subscription.Value))
+                |> Seq.toArray
+
+            subscriptions.Clear ()
+            snapshot)
+
+    subscriptionsToDispose
+    |> Seq.iter (fun struct (id, subscription) -> subscription |> executeOnUnsubscribeAndDispose id)