diff --git a/Packages.props b/Packages.props index 6ab045fc..1f810695 100644 --- a/Packages.props +++ b/Packages.props @@ -21,6 +21,8 @@ + + @@ -67,9 +69,11 @@ + + diff --git a/README.md b/README.md index d194bfed..1c5b35bf 100644 --- a/README.md +++ b/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) ``` diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index f7009e03..bd4f20dd 100644 --- a/RELEASE_NOTES.md +++ b/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 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 * 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`. 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 `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 `Define.TaskSeqField` streaming retaining a task for every item already delivered until the sequence ends +* 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 +* 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 diff --git a/docs/type-system.md b/docs/type-system.md index ff8517e1..470e3de3 100644 --- a/docs/type-system.md +++ b/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 as paged -> ValueSome paged.PageSize + | _ -> ValueNone)) +``` + +Azure SDK `AsyncPageable` 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` 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. diff --git a/samples/star-wars-api/Schema.fs b/samples/star-wars-api/Schema.fs index 11d2d296..4c20f239 100644 --- a/samples/star-wars-api/Schema.fs +++ b/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)) ] ) diff --git a/samples/star-wars-api/star-wars-api.fsproj b/samples/star-wars-api/star-wars-api.fsproj index 1937930c..20556189 100644 --- a/samples/star-wars-api/star-wars-api.fsproj +++ b/samples/star-wars-api/star-wars-api.fsproj @@ -7,6 +7,7 @@ + diff --git a/src/FSharp.Data.GraphQL.Server.AspNetCore/FSharp.Data.GraphQL.Server.AspNetCore.fsproj b/src/FSharp.Data.GraphQL.Server.AspNetCore/FSharp.Data.GraphQL.Server.AspNetCore.fsproj index afe1b41e..7dee1155 100644 --- a/src/FSharp.Data.GraphQL.Server.AspNetCore/FSharp.Data.GraphQL.Server.AspNetCore.fsproj +++ b/src/FSharp.Data.GraphQL.Server.AspNetCore/FSharp.Data.GraphQL.Server.AspNetCore.fsproj @@ -36,4 +36,8 @@ + + + + diff --git a/src/FSharp.Data.GraphQL.Server.AspNetCore/GraphQLWebsocketMiddleware.fs b/src/FSharp.Data.GraphQL.Server.AspNetCore/GraphQLWebsocketMiddleware.fs index 59855f63..36efe7b5 100644 --- a/src/FSharp.Data.GraphQL.Server.AspNetCore/GraphQLWebsocketMiddleware.fs +++ b/src/FSharp.Data.GraphQL.Server.AspNetCore/GraphQLWebsocketMiddleware.fs @@ -23,6 +23,53 @@ open FSharp.Data.GraphQL open FSharp.Data.GraphQL.Execution open FSharp.Data.GraphQL.Shared.WebSockets +/// +/// Splits a batched deferred or streamed payload, addressed by a path whose last segment is the list of indices of +/// the items in the batch, into one independently addressed payload per item. +/// +/// +/// groups several streamed items produced together (by the +/// field's batching policy or the query's preferredBatchSize) into a single deferred payload, addressed by a +/// path ending in the list of the batch's item indices, such as ["items"; [2; 1]]. A graphql-transport-ws +/// client cannot merge that into the response tree: no single index identifies where the payload belongs. Splitting +/// it here, at the transport, keeps the engine's batching (still one buffered event, still one merge of concurrently +/// resolved items) while addressing every item the same way a field that streams one item at a time already does: a +/// one-element data array at a path ending in that item's own index. +/// +module internal IncrementalPayloadSplitting = + + // Written as `obj list`, not the (internal, and here inaccessible) `FieldPath` abbreviation it stands for: + // a type abbreviation is erased, so this is the exact same type and unifies fine with FieldPath-typed values. + let pathStartsWith (prefix : obj list) (path : obj list) = + let prefixLength = List.length prefix + List.length path >= prefixLength && List.truncate prefixLength path = prefix + + /// Matches a path ending in a list of indices, such as the path of a batched deferred payload, returning the + /// path of the batch's own field and the indices of its items. + let (|BatchPath|_|) (path : obj list) = + match List.rev path with + | (:? (obj list) as indices) :: fieldPathRev -> Some (List.rev fieldPathRev, indices) + | _ -> None + + /// Splits a batch's data (an array with one element per index, in the same order) and errors (each carrying the + /// full path of the item it belongs to, since every error of a batch originates from resolving one specific + /// item) into one (data, errors, path) triple per item, addressed at that item's own path. + let splitBatch (fieldPath : obj list) (indices : obj list) (data : obj) (errors : GQLProblemDetails list) = + let items = data :?> obj[] + (indices, List.ofArray items) + ||> List.map2 (fun index item -> + let itemPath = [ yield! fieldPath; yield index ] + let itemErrors = + errors + |> List.filter (fun error -> + error.Path + |> Skippable.toValueOption + |> ValueOption.map (pathStartsWith itemPath) + |> ValueOption.defaultValue false) + box [| item |], itemErrors, itemPath) + +open IncrementalPayloadSplitting + type GraphQLWebSocketMiddleware<'Root> ( next : RequestDelegate, // must be kept for middleware signature compatibility @@ -145,10 +192,21 @@ type GraphQLWebSocketMiddleware<'Root> |> GraphQLSubscriptionsManagement.removeSubscription (id)) ) - let unsubscriber = streamSource.Subscribe (observer) + // Registered before subscribing, so a stream that completes synchronously (from inside Subscribe) still + // finds the id when its onCompleted callback above runs; only then is it safe to remove and dispose it. + // Assigning Disposable on an already-disposed SingleAssignmentDisposable disposes the assigned value too. + let placeholder = new System.Reactive.Disposables.SingleAssignmentDisposable () subscriptions - |> GraphQLSubscriptionsManagement.addSubscription (id, unsubscriber, (fun _ -> ())) + |> GraphQLSubscriptionsManagement.addSubscription (id, placeholder, (fun _ -> ())) + + try + placeholder.Disposable <- streamSource.Subscribe (observer) + with _ -> + // Nothing will ever complete this subscription now, so the id is freed here instead; a no-op if the + // synchronous completion above already removed it. Rethrown for the caller to report the failure. + subscriptions |> GraphQLSubscriptionsManagement.removeSubscription id + reraise () let tryToGracefullyCloseSocket (code, message) theSocket = if theSocket |> canCloseSocket then @@ -174,26 +232,39 @@ type GraphQLWebSocketMiddleware<'Root> let sendSubscriptionResponseOutput id subscriptionResult = match subscriptionResult with - | SubscriptionResult output -> { Data = ValueSome output; Errors = [] } |> sendOutput id + | SubscriptionResult output -> SubscriptionExecutionResult.Create (output, []) |> sendOutput id | SubscriptionErrors (output, errors) -> logger.LogWarning ("Subscription errors: {subscriptionErrors}", (String.Join ('\n', errors |> Seq.map (fun x -> $"- %s{x.Message}")))) - { Data = ValueNone; Errors = errors } |> sendOutput id - - let sendDeferredResponseOutput id deferredResult = + // The executor may still have resolved partial data alongside the field errors; forward it as-is + match output with + | null -> SubscriptionExecutionResult.CreateErrors errors |> sendOutput id + | output -> SubscriptionExecutionResult.Create (output, errors) |> sendOutput id + + // Incremental payloads are sent as soon as they are produced, with their path inside the initial result, + // so a client can merge them. The completion marker becomes a final payload with hasNext set to false. + // A batched payload (path ending in a list of indices) is split into one payload per item first, since a + // client cannot merge a payload that isn't addressed by a single index. + let sendDeferredResponseOutput id deferredResult : Task = task { match deferredResult with - | DeferredResult (obj, path) -> - let output = obj :?> Dictionary - { Data = ValueSome output; Errors = [] } |> sendOutput id - | DeferredErrors (obj, errors, _) -> + | ValueSome (DeferredResult (data, BatchPath (fieldPath, indices))) -> + for itemData, _, itemPath in splitBatch fieldPath indices data [] do + do! SubscriptionExecutionResult.CreateIncremental (itemData, [], itemPath) |> sendOutput id + | ValueSome (DeferredResult (data, path)) -> + do! SubscriptionExecutionResult.CreateIncremental (data, [], path) |> sendOutput id + | ValueSome (DeferredErrors (data, errors, BatchPath (fieldPath, indices))) -> logger.LogWarning ( "Deferred response errors: {deferredErrors}", (String.Join ('\n', errors |> Seq.map (fun x -> $"- %s{x.Message}"))) ) - { Data = ValueNone; Errors = errors } |> sendOutput id - - let sendDeferredResultDelayedBy (ct : CancellationToken) (ms : int) id deferredResult : Task = task { - do! Task.Delay (ms, ct) - do! deferredResult |> sendDeferredResponseOutput id + for itemData, itemErrors, itemPath in splitBatch fieldPath indices data errors do + do! SubscriptionExecutionResult.CreateIncremental (itemData, itemErrors, itemPath) |> sendOutput id + | ValueSome (DeferredErrors (data, errors, path)) -> + logger.LogWarning ( + "Deferred response errors: {deferredErrors}", + (String.Join ('\n', errors |> Seq.map (fun x -> $"- %s{x.Message}"))) + ) + do! SubscriptionExecutionResult.CreateIncremental (data, errors, path) |> sendOutput id + | ValueNone -> do! SubscriptionExecutionResult.CreateCompleted () |> sendOutput id } let applyPlanExecutionResult (id : SubscriptionId) (socket) (executionResult : GQLExecutionResult) : Task = task { @@ -202,16 +273,24 @@ type GraphQLWebSocketMiddleware<'Root> (subscriptions, socket, observableOutput, serializerOptions) |> addClientSubscription id sendSubscriptionResponseOutput | Deferred (data, errors, observableOutput) -> - do! { Data = ValueSome data; Errors = [] } |> sendOutput id - if errors.IsEmpty then - (subscriptions, socket, observableOutput, serializerOptions) - |> addClientSubscription id (sendDeferredResultDelayedBy cancellationToken 5000) - else - () - | Direct (data, _) -> do! { Data = ValueSome data; Errors = [] } |> sendOutput id + do! SubscriptionExecutionResult.CreateInitial (data, errors) |> sendOutput id + (subscriptions, socket, observableOutput |> Observable.withCompletionMarker, serializerOptions) + |> addClientSubscription id sendDeferredResponseOutput + | Direct (data, errors) -> + // An execution result, whose data is null when a non-null root field failed during execution; + // still a result, so it is sent as Next + Complete like any other, not as the terminal Error + // message below + if not errors.IsEmpty then + logger.LogWarning ("Execution errors:\n{errors}", errors) + do! SubscriptionExecutionResult.Create (data, errors) |> sendOutput id + // The graphql-transport-ws protocol requires Complete after the single Next of a query or mutation + do! sendMsg (Complete id) | RequestError problemDetails -> logger.LogWarning("Request errors:\n{errors}", problemDetails) - do! { Data = ValueNone; Errors = problemDetails } |> sendOutput id + // The request was rejected before execution, so it is not a result: the protocol requires it to be + // sent as the terminal Error message instead of a Next followed by Complete, or a client would + // read it as a successful result with null data + do! sendMsg (Error (id, problemDetails)) } let logMsgReceivedWithOptionalPayload optionalPayload (msgAsStr : string) = @@ -279,7 +358,7 @@ type GraphQLWebSocketMiddleware<'Root> do! planExecutionResult |> applyPlanExecutionResult id socket with ex -> logger.LogError (ex, "Unexpected error during subscription with id '{id}'", id) - do! sendMsg (Error (id, [NameValueLookup([ ("subscription", "Unexpected error during subscription" :> obj) ])])) + do! sendMsg (Error (id, [ GQLProblemDetails.Create "Unexpected error during subscription" ])) | ClientComplete id -> "ClientComplete" |> logMsgWithIdReceived id subscriptions diff --git a/src/FSharp.Data.GraphQL.Server/ErrorMessages.fs b/src/FSharp.Data.GraphQL.Server/ErrorMessages.fs index 8d59797f..6a6530f4 100644 --- a/src/FSharp.Data.GraphQL.Server/ErrorMessages.fs +++ b/src/FSharp.Data.GraphQL.Server/ErrorMessages.fs @@ -6,4 +6,4 @@ open System let variableNotFound variableName = $"A variable '$%s{variableName}' was not provided" -let expectedEnumerableValue indetifier ``type`` = $"Expected to have enumerable value in field '%s{indetifier}' but got '%O{(``type``:Type)}'" +let expectedEnumerableValue indetifier ``type`` = $"Expected to have enumerable or asynchronous enumerable value in field '%s{indetifier}' but got '%O{(``type``:Type)}'" diff --git a/src/FSharp.Data.GraphQL.Server/Execution.fs b/src/FSharp.Data.GraphQL.Server/Execution.fs index af1a06d2..164b5c64 100644 --- a/src/FSharp.Data.GraphQL.Server/Execution.fs +++ b/src/FSharp.Data.GraphQL.Server/Execution.fs @@ -131,6 +131,13 @@ type StreamOutput = | NonBufferedList of int * (KeyValuePair * GQLProblemDetails list) | BufferedList of int list * (KeyValuePair * GQLProblemDetails list) list +/// An event of a streamed list: a resolved item with its index in the source, +/// or a failure raised while enumerating an asynchronous source. +[] +type private StreamEvent = + | StreamedItem of index : int * result : ResolverResult> + | StreamFailure of error : exn + let private raiseErrors errs = AsyncVal.wrap <| Error errs /// Given an error e, call ParseError in the given context's Schema to convert it into @@ -210,14 +217,33 @@ let rec private direct (returnDef : OutputDef) (inputContext : InputExecutionCon | kind -> failwithf "Unexpected value of ctx.ExecutionPlan.Kind: %A" kind let resolveItem index item = executeResolvers inputContext innerCtx (box index :: path) value (toOption item |> AsyncVal.wrap) + let resolveItems (items : obj[]) = + items + |> Array.mapi resolveItem + |> collectFields Parallel + |> AsyncVal.map(ResolverResult.mapValue(fun items -> KeyValuePair(name, items |> Array.map(fun d -> d.Value) |> box))) match value with + | :? IAsyncEnumerableFieldValue as fieldValue -> + async { + // The sequence is drained first, the same way a lazy seq is materialized below. + // Enumeration errors are caught inside the computation, because resolveWith only catches synchronous exceptions. + let! drained = async { + try + let! items = AsyncEnumerableExtensions.toArrayAsync fieldValue.Items + return Ok items + with e -> + return Error (resolverError path ctx e) + } + match drained with + | Error errs -> return Error errs + | Ok items -> return! resolveItems items + } + |> AsyncVal.ofAsync | :? System.Collections.IEnumerable as enumerable -> enumerable |> Seq.cast |> Seq.toArray - |> Array.mapi resolveItem - |> collectFields Parallel - |> AsyncVal.map(ResolverResult.mapValue(fun items -> KeyValuePair(name, items |> Array.map(fun d -> d.Value) |> box))) + |> resolveItems | _ -> raise <| GQLMessageException (ErrorMessages.expectedEnumerableValue ctx.ExecutionInfo.Identifier (value.GetType())) | Nullable (Output innerDef) -> @@ -267,7 +293,16 @@ and private streamed (options : BufferedStreamOptions) (innerDef : OutputDef) (i | ResolveCollection innerPlan -> { ctx with ExecutionInfo = innerPlan } | kind -> failwithf "Unexpected value of ctx.ExecutionPlan.Kind: %A" kind - let collectBuffered : (int * ResolverResult>) list -> IObservable = function + // A batch size requested by the @stream directive takes precedence over the batching policy declared on the field. + // The policy is evaluated here, lazily, so it never runs for an ordinary or deferred query, and only once per + // streamed query even when the query itself supplies a batch size. + let options = + match options.PreferredBatchSize, value with + | ValueNone, (:? IAsyncEnumerableFieldValue as fieldValue) -> + { options with PreferredBatchSize = fieldValue.GetPreferredBatchSize () } + | _ -> options + + let collectItems : (int * ResolverResult>) list -> IObservable = function | [] -> Observable.empty | [(index, result)] -> result @@ -284,13 +319,30 @@ and private streamed (options : BufferedStreamOptions) (innerDef : OutputDef) (i let (_, indicies, deferred, errs) = List.foldBack merge chunk (chunk.Length - 1, [], None, []) deferResults (box indicies :: path) (Ok (box data, deferred, errs)) - let buffer (items : IObservable>>) : IObservable = + let collectBuffered (events : StreamEvent list) : IObservable = + let items = + events + |> List.vchoose (function + | StreamedItem (index, result) -> ValueSome (index, result) + | StreamFailure _ -> ValueNone) + // An enumeration failure is delivered as a value after the items of the same buffer, + // so it neither loses buffered items nor terminates sibling deferred streams + let failures = + events + |> List.vchoose (function + | StreamFailure error -> ValueSome (DeferredErrors (null, resolverError path ctx error, normalizeErrorPath path)) + | StreamedItem _ -> ValueNone) + match failures with + | [] -> collectItems items + | failures -> collectItems items |> Observable.concat (Observable.ofSeq failures) + + let buffer (events : IObservable) : IObservable = let buffered = match options.Interval, options.PreferredBatchSize with - | Some i, None -> Observable.bufferMilliseconds i items |> Observable.map List.ofSeq - | None, Some c -> Observable.bufferCount c items |> Observable.map List.ofSeq - | Some i, Some c -> Observable.bufferMillisecondsCount i c items |> Observable.map List.ofSeq - | None, None -> Observable.map(List.singleton) items + | ValueSome i, ValueNone -> Observable.bufferMilliseconds i events |> Observable.map List.ofSeq + | ValueNone, ValueSome c -> Observable.bufferCount c events |> Observable.map List.ofSeq + | ValueSome i, ValueSome c -> Observable.bufferMillisecondsCount i c events |> Observable.map List.ofSeq + | ValueNone, ValueNone -> Observable.map(List.singleton) events buffered |> Observable.bind collectBuffered @@ -300,6 +352,15 @@ and private streamed (options : BufferedStreamOptions) (innerDef : OutputDef) (i } match value with + | :? IAsyncEnumerableFieldValue as fieldValue -> + let resolveStreamedItem index item = resolveItem index item |> AsyncVal.map StreamedItem + let stream : IObservable = + fieldValue.Items + // At most fieldValue.MaxConcurrency items are pulled from the source and resolved at the same time, + // each emitted as soon as it is resolved; a failure of the source itself is emitted last + |> Observable.ofAsyncEnumerableResolved fieldValue.MaxConcurrency resolveStreamedItem StreamFailure + |> buffer + ResolverResult.defered (KeyValuePair (name, box [])) stream |> AsyncVal.wrap | :? System.Collections.IEnumerable as enumerable -> let stream : IObservable = enumerable @@ -307,8 +368,9 @@ and private streamed (options : BufferedStreamOptions) (innerDef : OutputDef) (i |> Seq.toArray |> Array.mapi resolveItem |> Observable.ofAsyncValSeq + |> Observable.map StreamedItem |> buffer - ResolverResult.defered (KeyValuePair (info.Identifier, box [])) stream |> AsyncVal.wrap + ResolverResult.defered (KeyValuePair (name, box [])) stream |> AsyncVal.wrap | _ -> raise <| GQLMessageException (ErrorMessages.expectedEnumerableValue ctx.ExecutionInfo.Identifier (value.GetType())) and private live (inputContext : InputExecutionContextProvider) (ctx : ResolveFieldContext) (path : FieldPath) (parent : obj) (value : obj) = @@ -439,6 +501,10 @@ let internal compileField (fieldDef: FieldDef) : ExecuteField = fun resolveFieldCtx value -> asyncVal { return! resolve resolveFieldCtx value } + | Resolve.BoxedTaskSeq(_, _, resolve) -> + fun resolveFieldCtx value -> + try resolve resolveFieldCtx value |> AsyncVal.wrap + with e -> AsyncVal.Failure(e) | Resolve.BoxedExpr (resolve) -> fun resolveFieldCtx value -> downcast resolve resolveFieldCtx value | _ -> @@ -450,44 +516,60 @@ let private (|String|Other|) (o : obj) = | _ -> Other let private executeQueryOrMutation (resultSet: (string * ExecutionInfo) []) (ctx: ExecutionContext) (objDef: ObjectDef) (rootValue : obj) : AsyncVal = - let executeRootOperation (name, info) = + let executeRootOperation (name, info) (args : Map) = let fDef = info.Definition - let argDefs = ctx.FieldExecuteMap.GetArgs(ctx.ExecutionPlan.RootDef.Name, info.Definition.Name) - match getArgumentValues argDefs info.Ast.Arguments ctx.GetInputContext ctx.Variables with - | Error errs -> asyncVal { return Error (errs |> List.map GQLProblemDetails.OfError) } - | Ok args -> - let path = [ box info.Identifier ] - let fieldCtx = - { ExecutionInfo = info - Context = ctx - ReturnType = fDef.TypeDef - ParentType = objDef - Schema = ctx.Schema - Args = args - Variables = ctx.Variables - Path = normalizeErrorPath path } - let execute = ctx.FieldExecuteMap.GetExecute(ctx.ExecutionPlan.RootDef.Name, info.Definition.Name) - asyncVal { - let! result = - executeResolvers ctx.GetInputContext fieldCtx path rootValue (resolveField execute fieldCtx rootValue) - |> AsyncVal.rescue path ctx.Schema.ParseError - let result = - match result with - | Ok (Ok value) -> Ok value - | Ok (Error errs) - | Error errs -> Error errs + let path = [ box info.Identifier ] + let fieldCtx = + { ExecutionInfo = info + Context = ctx + ReturnType = fDef.TypeDef + ParentType = objDef + Schema = ctx.Schema + Args = args + Variables = ctx.Variables + Path = normalizeErrorPath path } + let execute = ctx.FieldExecuteMap.GetExecute(ctx.ExecutionPlan.RootDef.Name, info.Definition.Name) + asyncVal { + let! result = + executeResolvers ctx.GetInputContext fieldCtx path rootValue (resolveField execute fieldCtx rootValue) + |> AsyncVal.rescue path ctx.Schema.ParseError + let result = match result with - | Error errs when info.IsNullable -> return Ok (KeyValuePair(name, null), None, errs) - | Error errs -> return Error errs - | Ok r -> return Ok r - } + | Ok (Ok value) -> Ok value + | Ok (Error errs) + | Error errs -> Error errs + match result with + | Error errs when info.IsNullable -> return Ok (KeyValuePair(name, null), None, errs) + | Error errs -> return Error errs + | Ok r -> return Ok r + } asyncVal { let documentId = ctx.ExecutionPlan.DocumentId - match! resultSet |> Array.map executeRootOperation |> collectFields ctx.ExecutionPlan.Strategy with - | Ok (data, Some deferred, errs) -> return GQLExecutionResult.Deferred(documentId, NameValueLookup(data), errs, deferred, ctx.Metadata) - | Ok (data, None, errs) -> return GQLExecutionResult.Direct(documentId, NameValueLookup(data), errs, ctx.Metadata) - | Error errs -> return GQLExecutionResult.RequestError(documentId, errs, ctx.Metadata) + // Inline argument coercion is request validation, the same as variable coercion in Executor.eval's + // coerceVariables: it rejects the request before any root resolver runs, so its errors must never be + // reported as an execution result with null data + let coerced = SortedDictionary * IGQLError list)> () + resultSet + |> Array.iteri (fun i (_, info) -> + let argDefs = ctx.FieldExecuteMap.GetArgs(ctx.ExecutionPlan.RootDef.Name, info.Definition.Name) + match getArgumentValues argDefs info.Ast.Arguments ctx.GetInputContext ctx.Variables with + | Ok args -> coerced.Add(i, struct (args, [])) + | Error errs -> coerced.Add(i, struct (Map.empty, errs))) + let coercionErrors = coerced.Values |> Seq.collect (fun struct (_, errs) -> errs) |> List.ofSeq + if not coercionErrors.IsEmpty then + return GQLExecutionResult.Error(documentId, coercionErrors, ctx.Metadata) + else + let operations = + coerced + |> Seq.map (fun (KeyValue (i, struct (args, _))) -> executeRootOperation resultSet[i] args) + |> Array.ofSeq + match! operations |> collectFields ctx.ExecutionPlan.Strategy with + | Ok (data, Some deferred, errs) -> return GQLExecutionResult.Deferred(documentId, NameValueLookup(data), errs, deferred, ctx.Metadata) + | Ok (data, None, errs) -> return GQLExecutionResult.Direct(documentId, NameValueLookup(data), errs, ctx.Metadata) + // Only a non-null root field failing during execution reaches this branch: an execution result whose + // data is null, as the spec requires, unlike the request error returned above for a coercion failure + | Error errs -> return GQLExecutionResult.Direct(documentId, null, errs, ctx.Metadata) } let private executeSubscription (resultSet: (string * ExecutionInfo) []) (inputContext : InputExecutionContextProvider) (ctx: ExecutionContext) (objDef: SubscriptionObjectDef) value = result { diff --git a/src/FSharp.Data.GraphQL.Server/IO.fs b/src/FSharp.Data.GraphQL.Server/IO.fs index 463d8c3a..e4dc813f 100644 --- a/src/FSharp.Data.GraphQL.Server/IO.fs +++ b/src/FSharp.Data.GraphQL.Server/IO.fs @@ -71,7 +71,13 @@ type GQLExecutionResult = // TODO: Rename to PascalCase and GQLResponseContent = + /// The request was rejected before execution started: validation, planning, variable or inline argument + /// coercion, a middleware, or the executor itself failing. There is no data, unlike a Direct result whose + /// data happens to be null. | RequestError of Errors: GQLProblemDetails list + /// An execution result. Data is null when a non-null root field failed during execution and the error + /// propagated to the root, exactly as it would for a non-null nested field, rather than being rejected as a + /// RequestError. | Direct of Data : Output * Errors: GQLProblemDetails list | Deferred of Data : Output * Errors : GQLProblemDetails list * Defer : IObservable | Stream of Stream : IObservable diff --git a/src/FSharp.Data.GraphQL.Server/ObservableExtensions.fs b/src/FSharp.Data.GraphQL.Server/ObservableExtensions.fs index d44801a7..a3223971 100644 --- a/src/FSharp.Data.GraphQL.Server/ObservableExtensions.fs +++ b/src/FSharp.Data.GraphQL.Server/ObservableExtensions.fs @@ -1,7 +1,11 @@ namespace FSharp.Data.GraphQL open System +open System.Collections.Generic open System.Reactive.Linq +open System.Runtime.ExceptionServices +open System.Threading +open System.Threading.Tasks open FSharp.Control.Reactive.Observable /// Extension methods to observable, used in place of FSharp.Control.Observable @@ -36,3 +40,217 @@ module internal Observable = observer.OnCompleted() { new IDisposable with member _.Dispose() = () } } + + /// + /// Disposes the enumerator, if one was acquired, and returns the failure to report: the one captured while + /// enumerating, or the one raised by the disposal itself when there was none before. + /// + let internal disposeEnumerator (enumerator : IAsyncEnumerator<'T> voption) (failure : exn voption) : Task = task { + match enumerator with + | ValueNone -> return failure + | ValueSome enumerator -> + try + do! enumerator.DisposeAsync () + return failure + with ex -> + // An enumeration failure is the more useful one to report, the disposal failure is likely its consequence + return failure |> ValueOption.orElse (ValueSome ex) + } + + /// + /// Creates a cold observable, which enumerates the asynchronous sequence for every subscription. + /// + /// + /// Disposing the subscription cancels the enumeration and disposes the enumerator. + /// An exception raised by the sequence, when acquiring or disposing its enumerator as well as while enumerating, + /// is delivered through . + /// + let ofAsyncEnumerable (source : IAsyncEnumerable<'T>) : IObservable<'T> = + let enumerate (observer : IObserver<'T>) (cancellationToken : CancellationToken) : Task = task { + let mutable enumerator = ValueNone + let mutable failure = ValueNone + try + // Acquired inside the try, because a source may throw when asked for its enumerator + let acquired = source.GetAsyncEnumerator cancellationToken + enumerator <- ValueSome acquired + let mutable hasNext = true + // The token is checked explicitly, because a sequence is not obliged to observe the token it was given + while hasNext && not cancellationToken.IsCancellationRequested do + let! moved = acquired.MoveNextAsync () + if moved then observer.OnNext acquired.Current + else hasNext <- false + with ex -> + failure <- ValueSome ex + let! failure = disposeEnumerator enumerator failure + match failure with + // A failure caused by disposing the subscription has no observer left to be delivered to + | ValueSome ex when not cancellationToken.IsCancellationRequested -> observer.OnError ex + | _ -> () + } + Observable.Create<'T> (Func, CancellationToken, Task> (fun observer cancellationToken -> enumerate observer cancellationToken)) + + /// + /// Enumerates the sequence, resolving each item into a result with . At most + /// items are pulled from the source and resolved at the same time: once that + /// many resolutions are in flight, pulling the next item waits for one of them to be emitted. + /// + /// + /// A result produced synchronously is emitted immediately, keeping it in the order it was pulled. An exception + /// raised by the source, when acquiring or disposing its enumerator as well as while enumerating, is turned into + /// a result with and emitted only after every item pulled before it, so it can never + /// overtake a result that is still being resolved. A resolution whose computation throws — as opposed to + /// returning a result that merely carries errors, which is free to keep resolving + /// items after — stops the enumeration the same way and is delivered through once + /// every resolution already started has settled; no item pulled after such a failure is resolved. + /// Only the resolutions in flight are tracked, so a long-running source does not retain what it already delivered. + /// An observer whose throws while a result is delivered always has its + /// concurrency slot released, so the enumeration never deadlocks over it, but nothing further is delivered to it: + /// per the observable contract, the subscription is torn down by the caller as soon as OnNext throws, same + /// as for any other observer. + /// Disposing the subscription cancels the enumeration; resolutions already started are still awaited and, if + /// still relevant, emitted, but no further item is pulled. + /// + let ofAsyncEnumerableResolved + (maxConcurrency : int) + (resolve : int -> 'T -> AsyncVal<'Result>) + (onFailure : exn -> 'Result) + (source : IAsyncEnumerable<'T>) + : IObservable<'Result> = + let enumerate (observer : IObserver<'Result>) (cancellationToken : CancellationToken) : Task = task { + use slots = new SemaphoreSlim (maxConcurrency, maxConcurrency) + // Observer calls are not required to be thread-safe, but resolutions complete on arbitrary threads + let sync = obj () + let emit (result : 'Result) = + lock sync (fun () -> if not cancellationToken.IsCancellationRequested then observer.OnNext result) + // Only the number of resolutions still in flight is tracked, not the tasks themselves, so a long-running + // source does not retain one task per item; the last resolution to settle after the enumeration has ended + // completes drained. Ref cells, because the resolutions run on other threads. + let inFlight = ref 0 + let enumerationEnded = ref false + let drained = TaskCompletionSource () + let resolutionFailure = ref ValueNone + let failed () = lock sync (fun () -> resolutionFailure.Value.IsSome) + let stopped () = cancellationToken.IsCancellationRequested || failed () + let settle () = + lock sync (fun () -> + inFlight.Value <- inFlight.Value - 1 + if enumerationEnded.Value && inFlight.Value = 0 then drained.TrySetResult () |> ignore) + let resolveInBackground (pendingResult : AsyncVal<'Result>) = + lock sync (fun () -> inFlight.Value <- inFlight.Value + 1) + task { + try + try + let! result = pendingResult |> AsyncVal.toTask + emit result + with ex -> + // The first failure stops the enumeration; it is delivered once every started resolution has settled + lock sync (fun () -> if resolutionFailure.Value.IsNone then resolutionFailure.Value <- ValueSome ex) + finally + // Released whatever happened, otherwise the enumeration would wait for this slot forever. + // Released before settling, because settling lets the enumeration finish and dispose the semaphore. + slots.Release () |> ignore + settle () + } + |> ignore + let mutable enumerator = ValueNone + let mutable failure = ValueNone + try + // Acquired inside the try, because a source may throw when asked for its enumerator + let acquired = source.GetAsyncEnumerator cancellationToken + enumerator <- ValueSome acquired + let mutable index = 0 + let mutable hasNext = true + // The token is checked explicitly, because a sequence is not obliged to observe the token it was given + while hasNext && not (stopped ()) do + do! slots.WaitAsync cancellationToken + // A resolution may have failed, or the subscription been disposed, while this waited for a slot + // or while the source was producing the next item; rechecked after each await so nothing pulled + // after that is resolved, let alone emitted (an item the source already produced is dropped: + // the failure ends the stream anyway) + if stopped () then + slots.Release () |> ignore + hasNext <- false + else + let! moved = acquired.MoveNextAsync () + if not moved || stopped () then + slots.Release () |> ignore + hasNext <- false + else + let itemIndex = index + let item = acquired.Current + index <- index + 1 + match resolve itemIndex item with + // Items resolved synchronously are emitted immediately, which keeps them in the source order + | Immediate result -> + try + emit result + finally + slots.Release () |> ignore + | pendingResult -> resolveInBackground pendingResult + with ex -> + failure <- ValueSome ex + // Captured items no longer need the enumerator, so it is disposed before waiting for their resolutions + let! failure = disposeEnumerator enumerator failure + // Resolutions still in flight neither need the enumerator nor the loop, only their slots + lock sync (fun () -> + enumerationEnded.Value <- true + if inFlight.Value = 0 then drained.TrySetResult () |> ignore) + do! drained.Task + match failure |> ValueOption.orElse resolutionFailure.Value with + // A failure caused by disposing the subscription has no observer left to be delivered to + | ValueSome ex when not cancellationToken.IsCancellationRequested -> emit (onFailure ex) + | _ -> () + if not cancellationToken.IsCancellationRequested then + lock sync (fun () -> observer.OnCompleted ()) + } + Observable.Create<'Result> (Func, CancellationToken, Task> (fun observer cancellationToken -> enumerate observer cancellationToken)) + + /// + /// Wraps every element into and emits when the source completes. + /// + /// + /// A consumer can handle the completion like a regular element, for example to send a final message + /// before the completion itself is processed. + /// + let withCompletionMarker (source : IObservable<'T>) : IObservable<'T voption> = + Observable.Concat (Observable.Select (source, fun item -> ValueSome item), Observable.Return ValueNone) + +/// +/// Functions for consuming from computations. +/// +module internal AsyncEnumerableExtensions = + + /// + /// Enumerates the whole asynchronous sequence into an array using the cancellation token of the current computation. + /// + let toArrayAsync (source : IAsyncEnumerable<'T>) : Async<'T[]> = async { + let! cancellationToken = Async.CancellationToken + let enumerate () : Task> = task { + let items = ResizeArray<'T> () + let mutable enumerator = ValueNone + let mutable failure = ValueNone + try + // Acquired inside the try, because a source may throw when asked for its enumerator + let acquired = source.GetAsyncEnumerator cancellationToken + enumerator <- ValueSome acquired + let mutable hasNext = true + while hasNext do + cancellationToken.ThrowIfCancellationRequested () + let! moved = acquired.MoveNextAsync () + if moved then items.Add acquired.Current + else hasNext <- false + with ex -> + failure <- ValueSome ex + let! failure = Observable.disposeEnumerator enumerator failure + match failure with + | ValueSome ex -> return Error ex + | ValueNone -> return Ok (items.ToArray ()) + } + // The failure is returned as a value and rethrown here, because awaiting a faulted task + // would wrap the original exception into an AggregateException + match! enumerate () |> Async.AwaitTask with + | Ok items -> return items + | Error ex -> + ExceptionDispatchInfo.Capture(ex).Throw () + return Array.empty + } diff --git a/src/FSharp.Data.GraphQL.Server/Planning.fs b/src/FSharp.Data.GraphQL.Server/Planning.fs index 2d0ef475..9a5a5d53 100644 --- a/src/FSharp.Data.GraphQL.Server/Planning.fs +++ b/src/FSharp.Data.GraphQL.Server/Planning.fs @@ -179,16 +179,16 @@ let private getStreamBufferMode (field : Field) = ) let directive = field.Directives - |> List.tryFind (fun d -> d.Name = "stream") + |> List.vtryFind (fun d -> d.Name = "stream") let getArg argName (d : Directive) = d.Arguments - |> List.tryFind (fun x -> x.Name = argName) - |> Option.map (fun x -> x.Value |> cast argName) + |> List.vtryFind (fun x -> x.Name = argName) + |> ValueOption.map (fun x -> x.Value |> cast argName) let interval = getArg "interval" let preferredBatchSize = getArg "preferredBatchSize" match directive with - | Some d -> { Interval = interval d; PreferredBatchSize = preferredBatchSize d } - | None -> + | ValueSome d -> { Interval = interval d; PreferredBatchSize = preferredBatchSize d } + | ValueNone -> // Buffer options are read only for fields that have the @stream directive, so this indicates a planner bug Debug.Fail "Must be prevented by validation" raise (InvalidOperationException $"Field '%s{field.AliasOrName}' is planned as streamed, but it has no @stream directive") diff --git a/src/FSharp.Data.GraphQL.Server/Schema.fs b/src/FSharp.Data.GraphQL.Server/Schema.fs index b7752c82..1aca8292 100644 --- a/src/FSharp.Data.GraphQL.Server/Schema.fs +++ b/src/FSharp.Data.GraphQL.Server/Schema.fs @@ -153,14 +153,14 @@ type SchemaConfig = Define.Input( "interval", Nullable IntType, - defaultValue = streamOptions.Interval, + defaultValue = ValueOption.toOption streamOptions.Interval, description = "An optional argument used to buffer stream results. " + "When it's value is greater than zero, stream results will be buffered for milliseconds equal to the value, then sent to the client. " + "After that, starts buffering again until all results are streamed.") Define.Input( "preferredBatchSize", Nullable IntType, - defaultValue = streamOptions.PreferredBatchSize, + defaultValue = ValueOption.toOption streamOptions.PreferredBatchSize, description = "An optional argument used to buffer stream results. " + "When it's value is greater than zero, stream results will be buffered until item count reaches this value, then sent to the client. " + "After that, starts buffering again until all results are streamed.") |] diff --git a/src/FSharp.Data.GraphQL.Shared/FSharp.Data.GraphQL.Shared.fsproj b/src/FSharp.Data.GraphQL.Shared/FSharp.Data.GraphQL.Shared.fsproj index f23b49b1..270d1ea2 100644 --- a/src/FSharp.Data.GraphQL.Shared/FSharp.Data.GraphQL.Shared.fsproj +++ b/src/FSharp.Data.GraphQL.Shared/FSharp.Data.GraphQL.Shared.fsproj @@ -28,6 +28,7 @@ + diff --git a/src/FSharp.Data.GraphQL.Shared/SchemaDefinitions.fs b/src/FSharp.Data.GraphQL.Shared/SchemaDefinitions.fs index bc33ab3e..dba804d9 100644 --- a/src/FSharp.Data.GraphQL.Shared/SchemaDefinitions.fs +++ b/src/FSharp.Data.GraphQL.Shared/SchemaDefinitions.fs @@ -1059,6 +1059,286 @@ module SchemaDefinitions = DeprecationReason = deprecationReason Metadata = Metadata.Empty } + /// + /// Creates a list field defined inside object type, which items are produced by an asynchronous sequence. + /// + /// Without directives the sequence is enumerated completely and returned as a list. + /// With the @stream directive each item is delivered as soon as the sequence produces it, + /// grouped into batches by unless the directive specifies preferredBatchSize. + /// + /// + /// The resolver is captured as a quotation, so a taskSeq block that uses let! or yield! + /// must be defined in a separate function called from the resolver. + /// + /// + /// Field name. Must be unique in scope of the defining object. + /// GraphQL type definition of the current field's type. + /// Expression used to resolve the asynchronous sequence from defining object. + /// Optional grouping of streamed items into batches. + /// + /// Maximum number of items resolved, and pulled from the sequence, at the same time when the field is + /// streamed. Defaults to . Not applied outside @stream. + /// + /// Deprecation reason. + static member TaskSeqField(name : string, typedef : #OutputDef<'Item seq>, + [] resolve : Expr 'Val -> IAsyncEnumerable<'Item>>, + [] ?batching : StreamBatching<'Item>, + [] ?maxConcurrency : int, + ?deprecationReason : string) : FieldDef<'Val, 'Item seq> = + upcast { FieldDefinition.Name = name + Description = None + TypeDef = typedef + Resolve = TaskSeq(typeof<'Val>, typeof<'Item>, resolve, StreamBatching<'Item>.ToStreamingOptions (batching, maxConcurrency)) + Args = [||] + DeprecationReason = deprecationReason + Metadata = Metadata.Empty } + + /// + /// Creates a list field defined inside object type, which items are produced by an asynchronous sequence. + /// + /// Without directives the sequence is enumerated completely and returned as a list. + /// With the @stream directive each item is delivered as soon as the sequence produces it, + /// grouped into batches by unless the directive specifies preferredBatchSize. + /// + /// + /// The resolver is captured as a quotation, so a taskSeq block that uses let! or yield! + /// must be defined in a separate function called from the resolver. + /// + /// + /// Field name. Must be unique in scope of the defining object. + /// GraphQL type definition of the current field's type. + /// Field description. Useful for generating documentation. + /// Expression used to resolve the asynchronous sequence from defining object. + /// Optional grouping of streamed items into batches. + /// + /// Maximum number of items resolved, and pulled from the sequence, at the same time when the field is + /// streamed. Defaults to . Not applied outside @stream. + /// + /// Deprecation reason. + static member TaskSeqField(name : string, typedef : #OutputDef<'Item seq>, description : string, + [] resolve : Expr 'Val -> IAsyncEnumerable<'Item>>, + [] ?batching : StreamBatching<'Item>, + [] ?maxConcurrency : int, + ?deprecationReason : string) : FieldDef<'Val, 'Item seq> = + upcast { FieldDefinition.Name = name + Description = Some description + TypeDef = typedef + Resolve = TaskSeq(typeof<'Val>, typeof<'Item>, resolve, StreamBatching<'Item>.ToStreamingOptions (batching, maxConcurrency)) + Args = [||] + DeprecationReason = deprecationReason + Metadata = Metadata.Empty } + + /// + /// Creates a list field defined inside object type, which items are produced by an asynchronous sequence. + /// + /// Without directives the sequence is enumerated completely and returned as a list. + /// With the @stream directive each item is delivered as soon as the sequence produces it, + /// grouped into batches by unless the directive specifies preferredBatchSize. + /// + /// + /// The resolver is captured as a quotation, so a taskSeq block that uses let! or yield! + /// must be defined in a separate function called from the resolver. + /// + /// + /// Field name. Must be unique in scope of the defining object. + /// GraphQL type definition of the current field's type. + /// List of field arguments used to parametrize resolve expression output. + /// Expression used to resolve the asynchronous sequence from defining object. + /// Optional grouping of streamed items into batches. + /// + /// Maximum number of items resolved, and pulled from the sequence, at the same time when the field is + /// streamed. Defaults to . Not applied outside @stream. + /// + /// Deprecation reason. + static member TaskSeqField(name : string, typedef : #OutputDef<'Item seq>, args : InputFieldDef list, + [] resolve : Expr 'Val -> IAsyncEnumerable<'Item>>, + [] ?batching : StreamBatching<'Item>, + [] ?maxConcurrency : int, + ?deprecationReason : string) : FieldDef<'Val, 'Item seq> = + upcast { FieldDefinition.Name = name + Description = None + TypeDef = typedef + Resolve = TaskSeq(typeof<'Val>, typeof<'Item>, resolve, StreamBatching<'Item>.ToStreamingOptions (batching, maxConcurrency)) + Args = args |> List.toArray + DeprecationReason = deprecationReason + Metadata = Metadata.Empty } + + /// + /// Creates a list field defined inside object type, which items are produced by an asynchronous sequence. + /// + /// Without directives the sequence is enumerated completely and returned as a list. + /// With the @stream directive each item is delivered as soon as the sequence produces it, + /// grouped into batches by unless the directive specifies preferredBatchSize. + /// + /// + /// The resolver is captured as a quotation, so a taskSeq block that uses let! or yield! + /// must be defined in a separate function called from the resolver. + /// + /// + /// Field name. Must be unique in scope of the defining object. + /// GraphQL type definition of the current field's type. + /// Field description. Useful for generating documentation. + /// List of field arguments used to parametrize resolve expression output. + /// Expression used to resolve the asynchronous sequence from defining object. + /// Optional grouping of streamed items into batches. + /// + /// Maximum number of items resolved, and pulled from the sequence, at the same time when the field is + /// streamed. Defaults to . Not applied outside @stream. + /// + /// Deprecation reason. + static member TaskSeqField(name : string, typedef : #OutputDef<'Item seq>, description : string, args : InputFieldDef list, + [] resolve : Expr 'Val -> IAsyncEnumerable<'Item>>, + [] ?batching : StreamBatching<'Item>, + [] ?maxConcurrency : int, + ?deprecationReason : string) : FieldDef<'Val, 'Item seq> = + upcast { FieldDefinition.Name = name + Description = Some description + TypeDef = typedef + Resolve = TaskSeq(typeof<'Val>, typeof<'Item>, resolve, StreamBatching<'Item>.ToStreamingOptions (batching, maxConcurrency)) + Args = args |> List.toArray + DeprecationReason = deprecationReason + Metadata = Metadata.Empty } + + /// + /// Creates a nullable list field defined inside object type, which items are produced by an asynchronous sequence. + /// + /// Without directives the sequence is enumerated completely and returned as a list. + /// With the @stream directive each item is delivered as soon as the sequence produces it, + /// grouped into batches by unless the directive specifies preferredBatchSize. + /// + /// + /// The resolver is captured as a quotation, so a taskSeq block that uses let! or yield! + /// must be defined in a separate function called from the resolver. + /// + /// + /// Field name. Must be unique in scope of the defining object. + /// GraphQL type definition of the current field's type. + /// Expression used to resolve the asynchronous sequence from defining object. + /// Optional grouping of streamed items into batches. + /// + /// Maximum number of items resolved, and pulled from the sequence, at the same time when the field is + /// streamed. Defaults to . Not applied outside @stream. + /// + /// Deprecation reason. + static member TaskSeqField(name : string, typedef : #OutputDef<'Item seq option>, + [] resolve : Expr 'Val -> IAsyncEnumerable<'Item> option>, + [] ?batching : StreamBatching<'Item>, + [] ?maxConcurrency : int, + ?deprecationReason : string) : FieldDef<'Val, 'Item seq option> = + upcast { FieldDefinition.Name = name + Description = None + TypeDef = typedef + Resolve = TaskSeq(typeof<'Val>, typeof<'Item>, resolve, StreamBatching<'Item>.ToStreamingOptions (batching, maxConcurrency)) + Args = [||] + DeprecationReason = deprecationReason + Metadata = Metadata.Empty } + + /// + /// Creates a nullable list field defined inside object type, which items are produced by an asynchronous sequence. + /// + /// Without directives the sequence is enumerated completely and returned as a list. + /// With the @stream directive each item is delivered as soon as the sequence produces it, + /// grouped into batches by unless the directive specifies preferredBatchSize. + /// + /// + /// The resolver is captured as a quotation, so a taskSeq block that uses let! or yield! + /// must be defined in a separate function called from the resolver. + /// + /// + /// Field name. Must be unique in scope of the defining object. + /// GraphQL type definition of the current field's type. + /// Field description. Useful for generating documentation. + /// Expression used to resolve the asynchronous sequence from defining object. + /// Optional grouping of streamed items into batches. + /// + /// Maximum number of items resolved, and pulled from the sequence, at the same time when the field is + /// streamed. Defaults to . Not applied outside @stream. + /// + /// Deprecation reason. + static member TaskSeqField(name : string, typedef : #OutputDef<'Item seq option>, description : string, + [] resolve : Expr 'Val -> IAsyncEnumerable<'Item> option>, + [] ?batching : StreamBatching<'Item>, + [] ?maxConcurrency : int, + ?deprecationReason : string) : FieldDef<'Val, 'Item seq option> = + upcast { FieldDefinition.Name = name + Description = Some description + TypeDef = typedef + Resolve = TaskSeq(typeof<'Val>, typeof<'Item>, resolve, StreamBatching<'Item>.ToStreamingOptions (batching, maxConcurrency)) + Args = [||] + DeprecationReason = deprecationReason + Metadata = Metadata.Empty } + + /// + /// Creates a nullable list field defined inside object type, which items are produced by an asynchronous sequence. + /// + /// Without directives the sequence is enumerated completely and returned as a list. + /// With the @stream directive each item is delivered as soon as the sequence produces it, + /// grouped into batches by unless the directive specifies preferredBatchSize. + /// + /// + /// The resolver is captured as a quotation, so a taskSeq block that uses let! or yield! + /// must be defined in a separate function called from the resolver. + /// + /// + /// Field name. Must be unique in scope of the defining object. + /// GraphQL type definition of the current field's type. + /// List of field arguments used to parametrize resolve expression output. + /// Expression used to resolve the asynchronous sequence from defining object. + /// Optional grouping of streamed items into batches. + /// + /// Maximum number of items resolved, and pulled from the sequence, at the same time when the field is + /// streamed. Defaults to . Not applied outside @stream. + /// + /// Deprecation reason. + static member TaskSeqField(name : string, typedef : #OutputDef<'Item seq option>, args : InputFieldDef list, + [] resolve : Expr 'Val -> IAsyncEnumerable<'Item> option>, + [] ?batching : StreamBatching<'Item>, + [] ?maxConcurrency : int, + ?deprecationReason : string) : FieldDef<'Val, 'Item seq option> = + upcast { FieldDefinition.Name = name + Description = None + TypeDef = typedef + Resolve = TaskSeq(typeof<'Val>, typeof<'Item>, resolve, StreamBatching<'Item>.ToStreamingOptions (batching, maxConcurrency)) + Args = args |> List.toArray + DeprecationReason = deprecationReason + Metadata = Metadata.Empty } + + /// + /// Creates a nullable list field defined inside object type, which items are produced by an asynchronous sequence. + /// + /// Without directives the sequence is enumerated completely and returned as a list. + /// With the @stream directive each item is delivered as soon as the sequence produces it, + /// grouped into batches by unless the directive specifies preferredBatchSize. + /// + /// + /// The resolver is captured as a quotation, so a taskSeq block that uses let! or yield! + /// must be defined in a separate function called from the resolver. + /// + /// + /// Field name. Must be unique in scope of the defining object. + /// GraphQL type definition of the current field's type. + /// Field description. Useful for generating documentation. + /// List of field arguments used to parametrize resolve expression output. + /// Expression used to resolve the asynchronous sequence from defining object. + /// Optional grouping of streamed items into batches. + /// + /// Maximum number of items resolved, and pulled from the sequence, at the same time when the field is + /// streamed. Defaults to . Not applied outside @stream. + /// + /// Deprecation reason. + static member TaskSeqField(name : string, typedef : #OutputDef<'Item seq option>, description : string, args : InputFieldDef list, + [] resolve : Expr 'Val -> IAsyncEnumerable<'Item> option>, + [] ?batching : StreamBatching<'Item>, + [] ?maxConcurrency : int, + ?deprecationReason : string) : FieldDef<'Val, 'Item seq option> = + upcast { FieldDefinition.Name = name + Description = Some description + TypeDef = typedef + Resolve = TaskSeq(typeof<'Val>, typeof<'Item>, resolve, StreamBatching<'Item>.ToStreamingOptions (batching, maxConcurrency)) + Args = args |> List.toArray + DeprecationReason = deprecationReason + Metadata = Metadata.Empty } + /// /// Creates a custom defined field using a custom field execution function. /// diff --git a/src/FSharp.Data.GraphQL.Shared/SchemaDefinitionsExtensions.fs b/src/FSharp.Data.GraphQL.Shared/SchemaDefinitionsExtensions.fs index 1da3341c..9cfa31c6 100644 --- a/src/FSharp.Data.GraphQL.Shared/SchemaDefinitionsExtensions.fs +++ b/src/FSharp.Data.GraphQL.Shared/SchemaDefinitionsExtensions.fs @@ -59,6 +59,7 @@ type internal CustomResolveFieldDefinition<'Val, 'Res> (source : FieldDef<'Val>, | Sync (input, output, expr) -> Sync (input, output, changeResolver expr) | Async (input, output, expr) -> Async (input, output, changeResolver expr) | Undefined -> failwith "Field has no resolve function." + | TaskSeq _ -> raise (NotSupportedException "Resolve middleware is not supported for fields defined with Define.TaskSeqField.") | x -> failwith <| sprintf "Resolver '%A' is not supported." x interface IEquatable with member _.Equals (other) = source.Equals (other) diff --git a/src/FSharp.Data.GraphQL.Shared/Serialization/JsonConverters.fs b/src/FSharp.Data.GraphQL.Shared/Serialization/JsonConverters.fs index 88e83d62..9d16e351 100644 --- a/src/FSharp.Data.GraphQL.Shared/Serialization/JsonConverters.fs +++ b/src/FSharp.Data.GraphQL.Shared/Serialization/JsonConverters.fs @@ -130,7 +130,11 @@ type RawServerMessageConverter () = | ExecutionResult output -> writer.WritePropertyName ("payload") JsonSerializer.Serialize (writer, output, options) - | ErrorMessages msgs -> JsonSerializer.Serialize (writer, msgs, options) - | CustomResponse jsonDocument -> jsonDocument.WriteTo (writer) + | ErrorMessages msgs -> + writer.WritePropertyName ("payload") + JsonSerializer.Serialize (writer, msgs, options) + | CustomResponse jsonDocument -> + writer.WritePropertyName ("payload") + jsonDocument.WriteTo (writer) writer.WriteEndObject () diff --git a/src/FSharp.Data.GraphQL.Shared/TypeSystem.fs b/src/FSharp.Data.GraphQL.Shared/TypeSystem.fs index be8e79b5..5a4cc368 100644 --- a/src/FSharp.Data.GraphQL.Shared/TypeSystem.fs +++ b/src/FSharp.Data.GraphQL.Shared/TypeSystem.fs @@ -809,9 +809,33 @@ and ExecutionInfoKind = /// Buffered stream options. Used to specify how the buffer will behavior in a stream. and BufferedStreamOptions = { /// The maximum time in milliseconds that the buffer will be filled before being sent to the subscriber. - Interval : int option + Interval : int voption /// The maximum number of items that will be buffered before being sent to the subscriber. - PreferredBatchSize : int option + PreferredBatchSize : int voption +} + +/// +/// Untyped batching policy of a field, applied to its items +/// when the field is requested with the @stream directive. +/// +and [] StreamBatchingPolicy = + /// Items are delivered as soon as they are produced, unless the query requests buffering. + | NoBatching + /// Items are delivered in batches of the fixed size. + | FixedBatch of size : int + /// The batch size is computed from the resolved sequence. + /// A delegate is used instead of an F# function to keep equality on the containing types. + | BatchFromSource of getBatchSize : Func + +/// +/// Untyped options of a field, applied when the field is requested with the +/// @stream directive. +/// +and TaskSeqStreamingOptions = { + /// Batching policy applied to streamed items. + Batching : StreamBatchingPolicy + /// Maximum number of items resolved, and pulled from the sequence, at the same time. + MaxConcurrency : int } /// Wrapper for a resolve method defined by the user or generated by a runtime. @@ -845,6 +869,14 @@ and Resolve = | ResolveExpr of expr : Expr + /// Resolve field value as an asynchronous sequence of items. + /// input defines .NET type of the provided object + /// output defines .NET type of the sequence items + /// expr is untyped version of Expr'Input->IAsyncEnumerable<'Output>> + /// or Expr'Input->IAsyncEnumerable<'Output> option> + /// streaming defines how items are grouped and how many are resolved concurrently when the field is streamed + | TaskSeq of input : Type * output : Type * expr : Expr * streaming : TaskSeqStreamingOptions + /// Returns an expression defining resolver function. member x.Expr = @@ -852,6 +884,7 @@ and Resolve = | Sync (_, _, e) -> e | Async (_, _, e) -> e | ResolveExpr (e) -> e + | TaskSeq (_, _, e, _) -> e | Undefined -> failwith "Resolve function was not defined" | x -> failwith <| sprintf "Unexpected resolve function %A" x @@ -2278,6 +2311,91 @@ module SubscriptionExtensions = this.AsyncPublish typeName fieldName subType |> Async.RunSynchronously +/// +/// Defines how items of a field created with are grouped into batches +/// when the field is requested with the @stream directive. +/// +/// +/// The preferredBatchSize argument of the @stream directive takes precedence over this definition. +/// +[] +type StreamBatching<'Item> = + /// Items are delivered in batches of the fixed size. + | Fixed of size : int + /// + /// The batch size is computed from the resolved sequence, for example from the page size of a paged SDK sequence. + /// Returning delivers items as soon as they are produced. + /// + | FromSource of getBatchSize : (IAsyncEnumerable<'Item> -> int voption) + + /// Converts the typed batching definition into the untyped policy stored in . + static member internal ToPolicy (batching : StreamBatching<'Item> voption) : StreamBatchingPolicy = + match batching with + | ValueNone -> StreamBatchingPolicy.NoBatching + | ValueSome (StreamBatching.Fixed size) when size < 1 -> invalidArg (nameof batching) $"Batch size must be greater than zero, but was %i{size}." + | ValueSome (StreamBatching.Fixed size) -> StreamBatchingPolicy.FixedBatch size + | ValueSome (StreamBatching.FromSource getBatchSize) -> + StreamBatchingPolicy.BatchFromSource (Func (fun source -> getBatchSize (source :?> IAsyncEnumerable<'Item>))) + + /// + /// Converts the typed batching definition and the field's maxConcurrency into the untyped options stored + /// in . A missing defaults to + /// . + /// + static member internal ToStreamingOptions (batching : StreamBatching<'Item> voption, maxConcurrency : int voption) : TaskSeqStreamingOptions = + let maxConcurrency = + match maxConcurrency with + | ValueNone -> Environment.ProcessorCount + | ValueSome c when c < 1 -> invalidArg (nameof maxConcurrency) $"Max concurrency must be greater than zero, but was %i{c}." + | ValueSome c -> c + { Batching = StreamBatching<'Item>.ToPolicy batching; MaxConcurrency = maxConcurrency } + +/// Gives the executor access to a resolved asynchronous sequence field value without knowing its item type. +type internal IAsyncEnumerableFieldValue = + /// Items of the sequence returned by the field resolver. + abstract Items : IAsyncEnumerable + /// Maximum number of items resolved, and pulled from the sequence, at the same time when the field is streamed. + abstract MaxConcurrency : int + /// + /// Computes the batch size from the batching policy declared on the field, applied to this resolved sequence. + /// + /// + /// Evaluated lazily: only when the field is requested with the @stream directive and the query does not + /// specify its own preferredBatchSize, so a batching callback never runs for an ordinary or deferred query. + /// + abstract GetPreferredBatchSize : unit -> int voption + +/// +/// Wraps a typed asynchronous sequence returned by a field resolver. +/// Items are boxed one by one, because an asynchronous sequence of a value type is not an asynchronous sequence of . +/// +type internal AsyncEnumerableFieldValue<'Item> (source : IAsyncEnumerable<'Item>, streaming : TaskSeqStreamingOptions) = + + let items = + { new IAsyncEnumerable with + member _.GetAsyncEnumerator (cancellationToken) = + let enumerator = source.GetAsyncEnumerator cancellationToken + { new IAsyncEnumerator with + member _.Current = box enumerator.Current + member _.MoveNextAsync () = enumerator.MoveNextAsync () + interface IAsyncDisposable with + member _.DisposeAsync () = enumerator.DisposeAsync () + } + } + + interface IAsyncEnumerableFieldValue with + member _.Items = items + member _.MaxConcurrency = streaming.MaxConcurrency + member _.GetPreferredBatchSize () = + match streaming.Batching with + | StreamBatchingPolicy.NoBatching -> ValueNone + | StreamBatchingPolicy.FixedBatch size -> ValueSome size + | StreamBatchingPolicy.BatchFromSource getBatchSize -> + // A non-positive size cannot be used for buffering, so such items are delivered as they are produced + match getBatchSize.Invoke (box source) with + | ValueSome size when size > 0 -> ValueSome size + | _ -> ValueNone + [] module Resolve = type private Marker = @@ -2309,6 +2427,15 @@ module Resolve = else None + let private (|AsyncEnumerable|_|) (typ : Type) = + if + typ.GetTypeInfo().IsGenericType + && typ.GetGenericTypeDefinition () = typedefof> + then + Some (typ.GenericTypeArguments |> Array.head) + else + None + let private boxify<'T, 'U> (f : ResolveFieldContext -> 'T -> 'U) : ResolveFieldContext -> obj -> obj = <@@ fun ctx (x : obj) -> f ctx (x :?> 'T) |> box @@> |> LeafExpressionConverter.EvaluateQuotation @@ -2333,6 +2460,28 @@ module Resolve = |> LeafExpressionConverter.EvaluateQuotation |> unbox + let private wrapAsyncEnumerable<'U> (streaming : TaskSeqStreamingOptions) (source : IAsyncEnumerable<'U>) : obj = + match box source with + // A null sequence is reported by the executor the same way as any other null value + | null -> null + | _ -> AsyncEnumerableFieldValue<'U> (source, streaming) |> box + + // The resolve function is returned from a let binding instead of a lambda body, so the compiled method + // keeps exactly two parameters, which the reflection-based invocation in boxifyExprTaskSeq relies on. + let private boxifyTaskSeq<'T, 'U> (streaming : TaskSeqStreamingOptions) (f : ResolveFieldContext -> 'T -> IAsyncEnumerable<'U>) : ResolveFieldContext -> obj -> obj = + let resolve (ctx : ResolveFieldContext) (x : obj) = f ctx (x :?> 'T) |> wrapAsyncEnumerable streaming + resolve + + let private boxifyTaskSeqOption<'T, 'U> (streaming : TaskSeqStreamingOptions) (f : ResolveFieldContext -> 'T -> IAsyncEnumerable<'U> option) : ResolveFieldContext -> obj -> obj = + let resolve (ctx : ResolveFieldContext) (x : obj) = + match f ctx (x :?> 'T) with + | Some source -> + match wrapAsyncEnumerable streaming source with + | null -> null + | wrapped -> box (Some wrapped) + | None -> null + resolve + let private getRuntimeMethod name = let methods = typeof.DeclaringType.GetRuntimeMethods () methods |> Seq.find (fun m -> m.Name.Equals name) @@ -2345,6 +2494,10 @@ module Resolve = let private runtimeBoxifyAsyncFilter = getRuntimeMethod "boxifyAsyncFilter" + let private runtimeBoxifyTaskSeq = getRuntimeMethod "boxifyTaskSeq" + + let private runtimeBoxifyTaskSeqOption = getRuntimeMethod "boxifyTaskSeqOption" + let private unwrapExpr = function | WithValue (resolver, _, _) -> (resolver, resolver.GetType ()) @@ -2387,6 +2540,18 @@ module Resolve = resolveUntypedFilter resolver r i o runtimeBoxifyAsyncFilter | resolver, _ -> failwithf "Unsupported signature for Async Subscription Filter Resolve %A" (resolver.GetType ()) + let private boxifyExprTaskSeq (streaming : TaskSeqStreamingOptions) expr : ResolveFieldContext -> obj -> obj = + let invoke (methodInfo : MethodInfo) (input : Type) (item : Type) (resolver : obj) = + methodInfo + .GetGenericMethodDefinition() + .MakeGenericMethod(input, item) + .Invoke (null, [| box streaming; resolver |]) + |> unbox + match unwrapExpr expr with + | resolver, FSharpFunc (_, FSharpFunc (d, AsyncEnumerable (c))) -> invoke runtimeBoxifyTaskSeq d c resolver + | resolver, FSharpFunc (_, FSharpFunc (d, FSharpOption (AsyncEnumerable (c)))) -> invoke runtimeBoxifyTaskSeqOption d c resolver + | resolver, _ -> failwithf "Unsupported signature for TaskSeq Resolve %A" (resolver.GetType ()) + let (|BoxedSync|_|) = function | Sync (d, c, expr) -> ValueSome (d, c, boxifyExpr expr) @@ -2412,6 +2577,12 @@ module Resolve = | AsyncFilter (r, i, o, expr) -> ValueSome (r, i, o, boxifyAsyncFilterExpr expr) | _ -> ValueNone + /// Matches a resolver of an asynchronous sequence field and compiles it into an untyped resolve function. + let (|BoxedTaskSeq|_|) = + function + | TaskSeq (d, c, expr, streaming) -> ValueSome (d, c, boxifyExprTaskSeq streaming expr) + | _ -> ValueNone + let private genMethodResolve<'Val, 'Res> (typeInfo : TypeInfo) (methodInfo : MethodInfo) = let argInfo = typeof.GetTypeInfo().GetDeclaredMethod ("Arg") let valueVar = Var ("value", typeof<'Val>) diff --git a/src/FSharp.Data.GraphQL.Shared/WebSockets.fs b/src/FSharp.Data.GraphQL.Shared/WebSockets.fs index 473b0a8c..923edb7b 100644 --- a/src/FSharp.Data.GraphQL.Shared/WebSockets.fs +++ b/src/FSharp.Data.GraphQL.Shared/WebSockets.fs @@ -3,6 +3,7 @@ namespace FSharp.Data.GraphQL.Shared.WebSockets open System open System.Collections.Generic open System.Text.Json +open System.Text.Json.Serialization open FSharp.Data.GraphQL open FSharp.Data.GraphQL.Shared @@ -16,11 +17,67 @@ type SubscriptionsDict = IDictionary +/// Payload of a next message of the graphql-transport-ws protocol. +/// +/// +/// and are present +/// only in payloads of incremental delivery, which is produced by the @defer and @stream directives. +/// +type SubscriptionExecutionResult = { + /// Result data: an object for complete and initial payloads, or a deferred or streamed value for incremental payloads. + /// It is omitted from the final payload of an incremental delivery. + Data : obj Skippable + /// Errors raised while producing the payload. + Errors : GQLProblemDetails list + /// Path of a deferred or streamed value inside the initial result. + Path : FieldPath Skippable + /// Tells whether more incremental payloads follow. + HasNext : bool Skippable +} with + + /// Creates a payload of a complete execution result. + static member Create (data : Output, errors : GQLProblemDetails list) = { + Data = Include (box data) + Errors = errors + Path = Skip + HasNext = Skip + } + + /// Creates a payload that carries only errors. + static member CreateErrors (errors : GQLProblemDetails list) = { Data = Include null; Errors = errors; Path = Skip; HasNext = Skip } + + /// Creates the initial payload of an incremental delivery, which is always followed by incremental payloads. + static member CreateInitial (data : Output, errors : GQLProblemDetails list) = { + Data = Include (box data) + Errors = errors + Path = Skip + HasNext = Include true + } + + /// + /// Creates an incremental payload with a deferred or streamed value located at the path. + /// More payloads may follow, so is . + /// + static member CreateIncremental (data : objnull, errors : GQLProblemDetails list, path : FieldPath) = { + Data = Include data + Errors = errors + Path = Include path + HasNext = Include true + } + + /// + /// Creates the final payload of an incremental delivery, which only reports that no more payloads follow. + /// + /// + /// Payloads are sent as soon as they are produced, and whether a payload is the last one becomes known + /// only when the deferred results complete, so the end of the delivery is reported separately. + /// + static member CreateCompleted () = { Data = Skip; Errors = []; Path = Skip; HasNext = Include false } type ServerRawPayload = | ExecutionResult of SubscriptionExecutionResult - | ErrorMessages of NameValueLookup list + | ErrorMessages of GQLProblemDetails list | CustomResponse of JsonDocument type RawServerMessage = { Id : string voption; Type : string; Payload : ServerRawPayload voption } @@ -39,7 +96,7 @@ type ServerMessage = | ServerPing | ServerPong of JsonDocument voption | Next of id : string * payload : SubscriptionExecutionResult - | Error of id : string * err : NameValueLookup list + | Error of id : string * err : GQLProblemDetails list | Complete of id : string module CustomWebSocketStatus = diff --git a/tests/FSharp.Data.GraphQL.Tests/AspNetCore/IncrementalPayloadSplittingTests.fs b/tests/FSharp.Data.GraphQL.Tests/AspNetCore/IncrementalPayloadSplittingTests.fs new file mode 100644 index 00000000..ed4e5dc1 --- /dev/null +++ b/tests/FSharp.Data.GraphQL.Tests/AspNetCore/IncrementalPayloadSplittingTests.fs @@ -0,0 +1,62 @@ +module FSharp.Data.GraphQL.Tests.AspNetCore.IncrementalPayloadSplittingTests + +open Xunit +open FSharp.Data.GraphQL +open FSharp.Data.GraphQL.Server.AspNetCore.IncrementalPayloadSplitting + +[] +let ``BatchPath does not match a path already addressed by a single index`` () = + match [ box "items"; box 0 ] with + | BatchPath _ -> Assert.Fail "a single-index path was matched as a batch" + | _ -> () + +[] +let ``BatchPath matches a path ending in a list of indices`` () = + match [ box "items"; box [ box 2; box 1 ] ] with + | BatchPath (fieldPath, indices) -> + fieldPath |> equals [ box "items" ] + indices |> equals [ box 2; box 1 ] + | _ -> fail "expected a BatchPath match" + +[] +let ``splitBatch addresses every item at its own index, out-of-order and preserving order`` () = + // Regression test: a batch's path ends in the list of its items' indices (as produced by + // Execution.collectItems), such as ["items"; [2; 1]], which no graphql-transport-ws client can merge - + // no single index identifies where the payload belongs. Splitting must produce one independently + // addressed payload per item, in the same relative order as the batch's own data array. + let data = box [| box "Buffered 3"; box "Buffered 2" |] + let split = splitBatch [ box "items" ] [ box 2; box 1 ] data [] + split + |> List.map (fun (itemData, errors, path) -> (itemData :?> obj[]), errors, path) + |> seqEquals [ + [| box "Buffered 3" |], [], [ box "items"; box 2 ] + [| box "Buffered 2" |], [], [ box "items"; box 1 ] + ] + +[] +let ``splitBatch attributes each error only to the item whose path it belongs to`` () = + let itemError = GQLProblemDetails.CreateWithKind ("Boom", Execution, [ box "items"; box 0; box "value" ]) + let data = box [| box "zero"; box "one" |] + let split = splitBatch [ box "items" ] [ box 0; box 1 ] data [ itemError ] + let errorsOf index = + split + |> List.find (fun (_, _, path) -> path = [ box "items"; box index ]) + |> fun (_, errors, _) -> errors + errorsOf 0 |> seqEquals [ itemError ] + errorsOf 1 |> empty + +[] +let ``splitBatch handles a batch containing a failed item's null data slot`` () = + // Regression test for the tenth Copilot review thread PRRT_kwDOA0s7t86i5Vu-: Execution.collectItems leaves a + // null slot in `data` for a failed item, but still keeps its index in `indices` at the same position (see + // Execution.fs's `merge`, whose Error arm only skips `Array.set data i`, not the index) - so `indices` and + // `data` are always the same length and List.map2 does not throw. This pins the null slot's shape. + let itemError = GQLProblemDetails.CreateWithKind ("Boom", Execution, [ box "items"; box 0; box "value" ]) + let data = box [| null; box "one" |] + let split = splitBatch [ box "items" ] [ box 0; box 1 ] data [ itemError ] + split + |> List.map (fun (itemData, errors, path) -> (itemData :?> obj[]), errors, path) + |> seqEquals [ + [| null |], [ itemError ], [ box "items"; box 0 ] + [| box "one" |], [], [ box "items"; box 1 ] + ] diff --git a/tests/FSharp.Data.GraphQL.Tests/AspNetCore/SerializationTests.fs b/tests/FSharp.Data.GraphQL.Tests/AspNetCore/SerializationTests.fs index 3ba85567..2c1316c9 100644 --- a/tests/FSharp.Data.GraphQL.Tests/AspNetCore/SerializationTests.fs +++ b/tests/FSharp.Data.GraphQL.Tests/AspNetCore/SerializationTests.fs @@ -106,3 +106,78 @@ let ``Deserializes client subscription correctly`` () = Assert.Equal (Skip, payload.OperationName) Assert.Equal (Skip, payload.Variables) | other -> Assert.Fail ($"unexpected actual value: '%A{other}'") + +open FSharp.Data.GraphQL + +let private serializePayload (payload : SubscriptionExecutionResult) = + let message : RawServerMessage = { Id = ValueSome "1"; Type = "next"; Payload = ValueSome (ExecutionResult payload) } + JsonSerializer.Serialize (message, serializerOptions) + +let private hasProperty (name : string) (element : JsonElement) = + let mutable ignored = Unchecked.defaultof + element.TryGetProperty (name, &ignored) + +[] +let ``Serializes incremental payload with path and hasNext`` () = + let json = serializePayload (SubscriptionExecutionResult.CreateIncremental (box [| box 1 |], [], [ box "numbers"; box 0 ])) + use document = JsonDocument.Parse json + let payload = document.RootElement.GetProperty "payload" + let data = payload.GetProperty "data" + Assert.Equal (JsonValueKind.Array, data.ValueKind) + Assert.Equal (1, data[0].GetInt32 ()) + let path = payload.GetProperty "path" + Assert.Equal ("numbers", path[0].GetString ()) + Assert.Equal (0, path[1].GetInt32 ()) + Assert.True (payload.GetProperty("hasNext").GetBoolean (), $"Expected hasNext to be true in {json}") + +[] +let ``Serializes final incremental payload with hasNext only`` () = + let json = serializePayload (SubscriptionExecutionResult.CreateCompleted ()) + use document = JsonDocument.Parse json + let payload = document.RootElement.GetProperty "payload" + Assert.False (payload.GetProperty("hasNext").GetBoolean (), $"Expected hasNext to be false in {json}") + Assert.False (hasProperty "data" payload, $"Expected no data in {json}") + Assert.False (hasProperty "path" payload, $"Expected no path in {json}") + +[] +let ``Serializes complete payload without path and hasNext`` () = + let json = serializePayload (SubscriptionExecutionResult.Create (NameValueLookup.ofList [ "name", upcast "R2-D2" ], [])) + use document = JsonDocument.Parse json + let payload = document.RootElement.GetProperty "payload" + Assert.Equal ("R2-D2", payload.GetProperty("data").GetProperty("name").GetString ()) + Assert.False (hasProperty "path" payload, $"Expected no path in {json}") + Assert.False (hasProperty "hasNext" payload, $"Expected no hasNext in {json}") + +[] +let ``Serializes errors payload with null data as before`` () = + let json = serializePayload (SubscriptionExecutionResult.CreateErrors [ GQLProblemDetails.CreateWithKind ("Boom", Execution, [ box "numbers" ]) ]) + use document = JsonDocument.Parse json + let payload = document.RootElement.GetProperty "payload" + Assert.Equal (JsonValueKind.Null, payload.GetProperty("data").ValueKind) + Assert.Equal ("Boom", (payload.GetProperty "errors").Item(0).GetProperty("message").GetString ()) + +[] +let ``Serializes an error message with its problem details as the payload`` () = + // Regression test: RawServerMessageConverter used to write the ErrorMessages payload without a preceding + // WritePropertyName ("payload"), which Utf8JsonWriter rejects, so every "error" message failed to serialize + let message : RawServerMessage = + { Id = ValueSome "1"; Type = "error"; Payload = ValueSome (ErrorMessages [ GQLProblemDetails.Create "Boom" ]) } + let json = JsonSerializer.Serialize (message, serializerOptions) + use document = JsonDocument.Parse json + let root = document.RootElement + Assert.Equal ("error", root.GetProperty("type").GetString ()) + Assert.Equal ("1", root.GetProperty("id").GetString ()) + let payload = root.GetProperty "payload" + Assert.Equal (JsonValueKind.Array, payload.ValueKind) + Assert.Equal ("Boom", payload[0].GetProperty("message").GetString ()) + +[] +let ``Serializes a pong message with its payload`` () = + // Regression test: the same missing WritePropertyName ("payload") affected a pong carrying a custom response + use responseDocument = JsonDocument.Parse "\"pong!\"" + let message : RawServerMessage = { Id = ValueNone; Type = "pong"; Payload = ValueSome (CustomResponse responseDocument) } + let json = JsonSerializer.Serialize (message, serializerOptions) + use document = JsonDocument.Parse json + let root = document.RootElement + Assert.Equal ("pong", root.GetProperty("type").GetString ()) + Assert.Equal ("pong!", root.GetProperty("payload").GetString ()) diff --git a/tests/FSharp.Data.GraphQL.Tests/DeferredTests.fs b/tests/FSharp.Data.GraphQL.Tests/DeferredTests.fs index a687e5e6..22e27901 100644 --- a/tests/FSharp.Data.GraphQL.Tests/DeferredTests.fs +++ b/tests/FSharp.Data.GraphQL.Tests/DeferredTests.fs @@ -240,7 +240,7 @@ let Query = ]) let schemaConfig = - { SchemaConfig.DefaultWithBufferedStream(streamOptions = { Interval = None; PreferredBatchSize = None }) with Types = [ CType; DType ] } + { SchemaConfig.DefaultWithBufferedStream(streamOptions = { Interval = ValueNone; PreferredBatchSize = ValueNone }) with Types = [ CType; DType ] } let sub = diff --git a/tests/FSharp.Data.GraphQL.Tests/ExecutionTests.fs b/tests/FSharp.Data.GraphQL.Tests/ExecutionTests.fs index 07bc5827..a4f43aca 100644 --- a/tests/FSharp.Data.GraphQL.Tests/ExecutionTests.fs +++ b/tests/FSharp.Data.GraphQL.Tests/ExecutionTests.fs @@ -17,6 +17,9 @@ open FSharp.Data.GraphQL.Shared open FSharp.Data.GraphQL.Types open FSharp.Data.GraphQL.Parser open FSharp.Data.GraphQL.Execution +open FSharp.Data.GraphQL.Validation +open FSharp.Data.GraphQL.Validation.ValidationResult +open ErrorHelpers type TestSubject = { a: string @@ -449,7 +452,35 @@ let ``Execution handles errors: exceptions`` () = ])) let expectedError = GQLProblemDetails.CreateWithKind ("Resolver Error!", Execution, [ box "a" ]) let result = sync <| Executor(schema).AsyncExecute("query Test { a }", getMockInputContext, ()) - ensureRequestError result <| fun [ error ] -> error |> equals expectedError + ensureDirect result <| fun data [ error ] -> + Assert.Null data + error |> equals expectedError + +type CoercionGuardInput = { Country : string } + +let CoercionGuardInputType = + Define.InputObject ( + "CoercionGuardInput", + [ Define.Input ("country", StringType) ], + fun input -> + match input.Country with + | "US" -> Success + | _ -> ValidationError [ { new IGQLError with member _.Message = "Unsupported country" } ]) + +[] +let ``Execution rejects inline argument coercion failures on one root field before running another root field's resolver`` () = + let boomCalls = ref 0 + let schema = + Schema(Define.Object( + "Query", [ + Define.Field("boom", StringType, (fun _ _ -> boomCalls.Value <- boomCalls.Value + 1; failwith "Resolver Error!")) + Define.Field("bad", Nullable StringType, [ Define.Input("input", CoercionGuardInputType) ], fun _ _ -> None) + ])) + let query = """query Test { boom bad(input: { country: "FR" }) }""" + let result = sync <| Executor(schema).AsyncExecute(query, getMockInputContext, ()) + ensureRequestError result <| fun [ error ] -> + error |> ensureInputObjectValidationError (Argument "input") "Unsupported country" [] "CoercionGuardInput!" + Assert.Equal(0, boomCalls.Value) [] let ``Execution handles errors: nullable list fields`` () = @@ -572,7 +603,8 @@ let ``Execution handles errors: additional error added when exception is rised i let result = let variables = { Inner = { Kaboom = "Yes, Rico, Kaboom" }; InnerPartialSuccess = { Kaboom = "Yes, Rico, Kaboom" } } sync <| Executor(schema).AsyncExecute("query Example { inner { kaboom } }", getMockInputContext, variables) - ensureRequestError result <| fun errors -> + ensureDirect result <| fun data errors -> + Assert.Null data result.DocumentId |> notEquals Unchecked.defaultof errors |> equals expectedErrors @@ -600,6 +632,7 @@ let ``Execution handles errors: additional error added and when null returned fr let result = let variables = { Inner = { Kaboom = "Yes, Rico, Kaboom" }; InnerPartialSuccess = { Kaboom = "Yes, Rico, Kaboom" } } sync <| Executor(schema).AsyncExecute("query Example { inner { kaboom } }", getMockInputContext, variables) - ensureRequestError result <| fun errors -> + ensureDirect result <| fun data errors -> + Assert.Null data result.DocumentId |> notEquals Unchecked.defaultof errors |> equals expectedErrors diff --git a/tests/FSharp.Data.GraphQL.Tests/FSharp.Data.GraphQL.Tests.fsproj b/tests/FSharp.Data.GraphQL.Tests/FSharp.Data.GraphQL.Tests.fsproj index 10104f96..61563c85 100644 --- a/tests/FSharp.Data.GraphQL.Tests/FSharp.Data.GraphQL.Tests.fsproj +++ b/tests/FSharp.Data.GraphQL.Tests/FSharp.Data.GraphQL.Tests.fsproj @@ -15,9 +15,12 @@ + + + @@ -95,7 +98,9 @@ + + diff --git a/tests/FSharp.Data.GraphQL.Tests/Helpers and Extensions/ObservableExtensionsTests.fs b/tests/FSharp.Data.GraphQL.Tests/Helpers and Extensions/ObservableExtensionsTests.fs index a51c342d..e2bb88e6 100644 --- a/tests/FSharp.Data.GraphQL.Tests/Helpers and Extensions/ObservableExtensionsTests.fs +++ b/tests/FSharp.Data.GraphQL.Tests/Helpers and Extensions/ObservableExtensionsTests.fs @@ -10,11 +10,6 @@ open Helpers open System open FSharp.Control.Reactive - -let delay time x = async { - do! Async.Sleep(ms time) - return x } - [] let ``ofSeq should call OnComplete and return items in expected order`` () = let source = seq { for x in 1 .. 5 do yield x } @@ -252,3 +247,248 @@ let ``singleton should call OnComplete and return item`` () = use sub = Observer.create obs sub.WaitCompleted(timeout = ms 10) sub.Received |> seqEquals (Seq.singleton 1) + +open System.Threading +open System.Threading.Tasks +open FSharp.Control + +[] +let ``ofAsyncEnumerable should call OnComplete and return items in expected order`` () = + use sub = Observable.ofAsyncEnumerable (asyncItems [ 1 .. 5 ]) |> Observer.create + sub.WaitCompleted(timeout = ms 10) + sub.Received |> seqEquals [ 1; 2; 3; 4; 5 ] + +[] +let ``ofAsyncEnumerable should deliver items produced before an enumeration error`` () = + let source = taskSeq { + yield 1 + failwith "Boom" + } + use sub = Observable.ofAsyncEnumerable source |> Observable.materialize |> Observer.create + sub.WaitCompleted(timeout = ms 10) + Assert.Collection ( + sub.Received, + (fun (notification : System.Reactive.Notification) -> + Assert.Equal (System.Reactive.NotificationKind.OnNext, notification.Kind) + Assert.Equal (1, notification.Value)), + (fun (notification : System.Reactive.Notification) -> + Assert.Equal (System.Reactive.NotificationKind.OnError, notification.Kind) + Assert.Equal ("Boom", notification.Exception.Message)) + ) + +[] +let ``ofAsyncEnumerable should stop the enumeration when the subscription is disposed`` () : Task = task { + let pulled = ref 0 + let disposed = TaskCompletionSource () + let received = TaskCompletionSource () + let source = endlessNumbers pulled disposed + let subscription = Observable.ofAsyncEnumerable source |> Observable.subscribe (fun _ -> received.TrySetResult () |> ignore) + do! waitForTask (TimeSpan.FromSeconds (float (ms 5))) "Expected an item before the subscription is disposed" received.Task + subscription.Dispose () + do! waitForTask (TimeSpan.FromSeconds (float (ms 5))) "Expected the enumerator to be disposed with the subscription" disposed.Task + let pulledAfterDisposal = pulled.Value + // A still running enumeration would pull more items during this delay + do! Task.Delay 200 + Assert.Equal (pulledAfterDisposal, pulled.Value) +} + +[] +let ``ofAsyncEnumerableResolved should emit synchronously resolved results in order`` () = + use sub = + Observable.ofAsyncEnumerableResolved 3 (fun _ (n : int) -> AsyncVal.wrap (n * 10)) (fun _ -> -1) (asyncItems [ 1 .. 5 ]) + |> Observer.create + sub.WaitCompleted (timeout = ms 10) + sub.Received |> seqEquals [ 10; 20; 30; 40; 50 ] + +[] +let ``ofAsyncEnumerableResolved should never resolve more than maxConcurrency items at the same time`` () = + let inFlight = ref 0 + let maxObserved = ref 0 + let resolve _ (n : int) = + async { + let current = Interlocked.Increment inFlight + let mutable observed = maxObserved.Value + while current > observed && Interlocked.CompareExchange (maxObserved, current, observed) <> observed do + observed <- maxObserved.Value + do! Async.Sleep (ms 50) + Interlocked.Decrement inFlight |> ignore + return n + } + |> AsyncVal.ofAsync + use sub = Observable.ofAsyncEnumerableResolved 2 resolve (fun _ -> -1) (asyncItems [ 1 .. 6 ]) |> Observer.create + sub.WaitCompleted (timeout = ms 10) + sub.Received |> Seq.toList |> List.sort |> seqEquals [ 1; 2; 3; 4; 5; 6 ] + Assert.True (maxObserved.Value <= 2, $"Expected at most 2 concurrent resolutions, but observed {maxObserved.Value}") + +[] +let ``ofAsyncEnumerableResolved should emit the failure after a slower earlier item`` () = + let source = itemThenFailure 1 + let resolve index (n : int) = + if index = 0 then + async { + do! Async.Sleep (ms 200) + return n + } + |> AsyncVal.ofAsync + else + AsyncVal.wrap n + use sub = Observable.ofAsyncEnumerableResolved 4 resolve (fun _ -> -1) source |> Observer.create + sub.WaitCompleted (timeout = ms 10) + sub.Received |> seqEquals [ 1; -1 ] + +[] +let ``ofAsyncEnumerableResolved should stop resolving further items when the subscription is disposed`` () : Task = task { + let pulled = ref 0 + let disposed = TaskCompletionSource () + let received = TaskCompletionSource () + let source = endlessNumbers pulled disposed + let subscription = + Observable.ofAsyncEnumerableResolved 1 (fun _ (n : int) -> AsyncVal.wrap n) (fun _ -> -1) source + |> Observable.subscribe (fun _ -> received.TrySetResult () |> ignore) + do! waitForTask (TimeSpan.FromSeconds (float (ms 5))) "Expected an item before the subscription is disposed" received.Task + subscription.Dispose () + do! waitForTask (TimeSpan.FromSeconds (float (ms 5))) "Expected the enumerator to be disposed with the subscription" disposed.Task + let pulledAfterDisposal = pulled.Value + // A still running enumeration would pull more items during this delay + do! Task.Delay 200 + Assert.Equal (pulledAfterDisposal, pulled.Value) +} + +[] +let ``ofAsyncEnumerable should deliver OnError when GetAsyncEnumerator throws`` () = + // Regression test: acquiring the enumerator happens before the try, so a throwing source must not bypass + // the failure handling and fault the returned Task in a way that skips OnError + let source = ThrowingAsyncEnumerable "Boom acquiring the enumerator" + use sub = Observable.ofAsyncEnumerable source |> Observable.materialize |> Observer.create + sub.WaitCompleted (timeout = ms 10) + Assert.Collection ( + sub.Received, + fun (notification : System.Reactive.Notification) -> + Assert.Equal (System.Reactive.NotificationKind.OnError, notification.Kind) + Assert.Equal ("Boom acquiring the enumerator", notification.Exception.Message) + ) + +[] +let ``ofAsyncEnumerable should deliver OnError when DisposeAsync throws`` () = + let source = itemThenDisposalFailure 1 + use sub = Observable.ofAsyncEnumerable source |> Observable.materialize |> Observer.create + sub.WaitCompleted (timeout = ms 10) + Assert.Collection ( + sub.Received, + (fun (notification : System.Reactive.Notification) -> + Assert.Equal (System.Reactive.NotificationKind.OnNext, notification.Kind) + Assert.Equal (1, notification.Value)), + (fun (notification : System.Reactive.Notification) -> + Assert.Equal (System.Reactive.NotificationKind.OnError, notification.Kind) + Assert.Equal ("Boom disposing", notification.Exception.Message)) + ) + +[] +let ``ofAsyncEnumerableResolved should emit the failure through onFailure when GetAsyncEnumerator throws`` () = + // Regression test: this used to fault the returned Task instead of going through onFailure, which terminates + // the merged deferred stream of a query instead of producing this field's DeferredErrors + let source = ThrowingAsyncEnumerable "Boom acquiring the enumerator" + use sub = + Observable.ofAsyncEnumerableResolved 2 (fun _ (n : int) -> AsyncVal.wrap n) (fun _ -> -1) source + |> Observer.create + sub.WaitCompleted (timeout = ms 10) + sub.Received |> seqEquals [ -1 ] + +[] +let ``ofAsyncEnumerableResolved should emit the failure through onFailure after the item when DisposeAsync throws`` () = + let source = itemThenDisposalFailure 1 + use sub = + Observable.ofAsyncEnumerableResolved 2 (fun _ (n : int) -> AsyncVal.wrap n) (fun _ -> -1) source + |> Observer.create + sub.WaitCompleted (timeout = ms 10) + sub.Received |> seqEquals [ 1; -1 ] + +[] +let ``ofAsyncEnumerableResolved should stop and deliver the failure when a resolution fails`` () = + // Regression test: a failed resolution used to leave its concurrency slot held forever, so with + // maxConcurrency = 1 the enumeration would deadlock instead of ever reaching onFailure or OnCompleted + let resolve _ (_ : int) = AsyncVal.Failure (exn "Boom resolving") + use sub = Observable.ofAsyncEnumerableResolved 1 resolve (fun _ -> -1) (asyncItems [ 1 ]) |> Observer.create + sub.WaitCompleted (timeout = ms 10) + sub.Received |> seqEquals [ -1 ] + +[] +let ``ofAsyncEnumerableResolved should not pull another item after a resolution fails while waiting for a slot`` () = + // Regression test: with maxConcurrency = 1 the loop is parked in WaitAsync while the one in-flight resolution + // runs; once that resolution fails and releases the slot, the loop used to go straight to MoveNextAsync without + // rechecking the failure, so a synchronously resolved item 2 was pulled and emitted before the failure + let resolve _ (n : int) = + if n = 1 then + async { + do! Async.Sleep (ms 50) + return failwith "Boom resolving" + } + |> AsyncVal.ofAsync + else + AsyncVal.wrap n + use sub = Observable.ofAsyncEnumerableResolved 1 resolve (fun _ -> -1) (asyncItems [ 1; 2; 3 ]) |> Observer.create + sub.WaitCompleted (timeout = ms 10) + sub.Received |> seqEquals [ -1 ] + +[] +let ``ofAsyncEnumerableResolved should not resolve an item pulled after a resolution failed while the source produced it`` () = + // Regression test: a background resolution can fail while MoveNextAsync for the next item is still suspended; + // when that move completed the code used to go straight to resolving it without rechecking the failure, so a + // synchronously resolved item 2 was pulled and emitted before the failure that already happened + let source = + SuspendingAsyncEnumerable (fun _ index -> + task { + match index with + | 0 -> return ValueSome 1 + | 1 -> + do! Task.Delay (ms 150) + return ValueSome 2 + | _ -> return ValueNone + }) + let resolve _ (n : int) = + if n = 1 then + async { + do! Async.Sleep (ms 50) + return failwith "Boom resolving" + } + |> AsyncVal.ofAsync + else + AsyncVal.wrap n + use sub = Observable.ofAsyncEnumerableResolved 2 resolve (fun _ -> -1) source |> Observer.create + sub.WaitCompleted (timeout = ms 10) + sub.Received |> seqEquals [ -1 ] + +[] +let ``ofAsyncEnumerableResolved should release the slot and not hang when the observer throws`` () : Task = task { + // Regression test: an observer throwing while a background resolution is delivered used to skip the slot + // release entirely, deadlocking the enumeration the same way a failed resolution did. System.Reactive tears + // the subscription down itself (disposing it, which cancels the enumeration) as soon as OnNext throws, so + // onFailure/OnCompleted are never expected here: this only checks that DisposeAsync is still reached instead + // of the enumeration hanging forever on the concurrency slot the throwing resolution never released. + let disposed = TaskCompletionSource () + let source = + SuspendingAsyncEnumerable ( + (fun _ index -> task { return if index = 0 then ValueSome 1 else ValueNone }), + fun () -> disposed.TrySetResult () |> ignore + ) + let resolve _ (n : int) = async { return n } |> AsyncVal.ofAsync + let onReceived (_ : TestObserver) (value : int) = + if value = 1 then failwith "Boom in observer" + use sub = + Observable.ofAsyncEnumerableResolved 1 resolve (fun _ -> -1) source + |> Observer.createWithCallback onReceived + do! waitForTask (TimeSpan.FromSeconds (float (ms 5))) "Expected the enumerator to be disposed despite the observer throwing" disposed.Task + sub.Received |> seqEquals [ 1 ] +} + +[] +let ``withCompletionMarker should emit the items and then the marker when the source completes`` () = + use sub = Observable.ofSeq [ 1; 2 ] |> Observable.withCompletionMarker |> Observer.create + sub.WaitCompleted(timeout = ms 10) + sub.Received |> seqEquals [ ValueSome 1; ValueSome 2; ValueNone ] + +[] +let ``withCompletionMarker should emit only the marker for an empty source`` () = + use sub = Observable.ofSeq Seq.empty |> Observable.withCompletionMarker |> Observer.create + sub.WaitCompleted(timeout = ms 10) + sub.Received |> seqEquals [ ValueNone ] diff --git a/tests/FSharp.Data.GraphQL.Tests/Helpers.fs b/tests/FSharp.Data.GraphQL.Tests/Helpers.fs index b44fd611..52abfafd 100644 --- a/tests/FSharp.Data.GraphQL.Tests/Helpers.fs +++ b/tests/FSharp.Data.GraphQL.Tests/Helpers.fs @@ -128,7 +128,7 @@ let ms x = | _ -> 20 x * factor -type TestObserver<'T>(obs : IObservable<'T>, ?onReceived : TestObserver<'T> -> 'T -> unit) as this = +type TestObserver<'T>(obs : IObservable<'T>, [] ?onReceived : TestObserver<'T> -> 'T -> unit) as this = let received = List<'T>() let mutable isCompleted = false let mre = new ManualResetEvent(false) @@ -157,7 +157,7 @@ type TestObserver<'T>(obs : IObservable<'T>, ?onReceived : TestObserver<'T> -> ' member _.OnError (error) = error.Reraise() member _.OnNext (value) = received.Add (value) - onReceived |> Option.iter (fun evt -> evt this value) + onReceived |> ValueOption.iter (fun evt -> evt this value) interface IDisposable with member _.Dispose () = subscription.Dispose () @@ -220,3 +220,91 @@ module MockInputContext = let mockInputContextInstance = MockInputExecutionContext() let getMockInputContext = fun () -> MockInputContext.mockInputContextInstance :> IInputExecutionContext + +open System.Threading.Tasks +open IcedTasks + +/// +/// An asynchronous sequence that produces each item through a task created on demand. +/// +/// +/// Tests use it instead of a taskSeq block for sequences that really suspend, because taskSeq code compiled +/// without optimizations, as in Debug builds of this project, does not resume correctly after an await. +/// +type SuspendingAsyncEnumerable<'T> (produceItem : CancellationToken -> int -> Task<'T voption>, [] ?onDisposed : unit -> unit) = + interface IAsyncEnumerable<'T> with + member _.GetAsyncEnumerator cancellationToken = + let index = ref 0 + let current = ref Unchecked.defaultof<'T> + { new IAsyncEnumerator<'T> with + member _.Current = current.Value + member _.MoveNextAsync () = + valueTask { + match! produceItem cancellationToken index.Value with + | ValueSome item -> + current.Value <- item + index.Value <- index.Value + 1 + return true + | ValueNone -> return false + } + interface IAsyncDisposable with + member _.DisposeAsync () = + onDisposed |> ValueOption.iter (fun onDisposed -> onDisposed ()) + ValueTask.CompletedTask + } + +/// Awaits the task without blocking the test thread and fails the test with the message when the task does not complete in time +let waitForTask (timeout : TimeSpan) (message : string) (awaited : Task) : Task = task { + let! completed = Task.WhenAny (awaited, Task.Delay timeout) + if not (obj.ReferenceEquals (completed, awaited)) then + fail message +} + +open FSharp.Control + +/// Returns the value after the scaled delay +let delay time x = async { + do! Async.Sleep (ms time) + return x +} + +/// An asynchronous sequence of the items, safe to use as a taskSeq in Debug builds because it never awaits +let asyncItems (items : 'T list) = taskSeq { + for item in items do + yield item +} + +/// A source whose GetAsyncEnumerator throws instead of returning an enumerator +type ThrowingAsyncEnumerable<'T> (message : string) = + interface IAsyncEnumerable<'T> with + member _.GetAsyncEnumerator _ = failwith message + +/// Produces the item, then fails while pulling the next one +let itemThenFailure (item : 'T) = + SuspendingAsyncEnumerable<'T> (fun _ index -> + task { + match index with + | 0 -> return ValueSome item + | _ -> return failwith "Boom during enumeration" + }) + :> IAsyncEnumerable<'T> + +/// Produces the item, then completes, and throws from DisposeAsync +let itemThenDisposalFailure (item : 'T) = + SuspendingAsyncEnumerable<'T> ( + (fun _ index -> task { return if index = 0 then ValueSome item else ValueNone }), + fun () -> failwith "Boom disposing" + ) + :> IAsyncEnumerable<'T> + +/// Produces numbers forever with a small delay, recording how many were pulled and signalling disposal +let endlessNumbers (pulled : int ref) (disposed : TaskCompletionSource) = + SuspendingAsyncEnumerable ( + (fun _ index -> task { + pulled.Value <- index + 1 + do! Task.Delay 20 + return ValueSome (index + 1) + }), + fun () -> disposed.TrySetResult () |> ignore + ) + :> IAsyncEnumerable diff --git a/tests/FSharp.Data.GraphQL.Tests/LazyEnumerationExceptionTests.fs b/tests/FSharp.Data.GraphQL.Tests/LazyEnumerationExceptionTests.fs index 17532a3e..e0b77836 100644 --- a/tests/FSharp.Data.GraphQL.Tests/LazyEnumerationExceptionTests.fs +++ b/tests/FSharp.Data.GraphQL.Tests/LazyEnumerationExceptionTests.fs @@ -117,7 +117,9 @@ let ``Execution must propagate error when non-nullable list field throws during ])) let expectedError = GQLProblemDetails.CreateWithKind ("Boom during enumeration", Execution, [ box "tags" ]) let result = sync <| Executor(schema).AsyncExecute(parse "{ tags }", getMockInputContext, ()) - ensureRequestError result <| fun [ error ] -> error |> equals expectedError + ensureDirect result <| fun data [ error ] -> + Assert.Null data + error |> equals expectedError [] let ``Execution must return null with field error when nullable list of objects throws KeyNotFoundException during lazy enumeration`` () = diff --git a/tests/FSharp.Data.GraphQL.Tests/TaskSeqFieldTests.fs b/tests/FSharp.Data.GraphQL.Tests/TaskSeqFieldTests.fs new file mode 100644 index 00000000..c87b9d9b --- /dev/null +++ b/tests/FSharp.Data.GraphQL.Tests/TaskSeqFieldTests.fs @@ -0,0 +1,584 @@ +// The MIT License (MIT) + +module FSharp.Data.GraphQL.Tests.TaskSeqFieldTests + +open System +open System.Collections.Generic +open System.Threading +open System.Threading.Tasks +open FSharp.Control +open Azure +open Xunit + +open FSharp.Data.GraphQL +open FSharp.Data.GraphQL.Parser +open FSharp.Data.GraphQL.Types + +type StreamItem = { Id : int; Value : Async } + +// Resolvers are captured as quotations, which cannot contain every taskSeq builder member, +// so the sequences are produced by functions called from the resolvers. +// Sequences that complete synchronously use taskSeq blocks (Helpers.asyncItems), while sequences that really +// suspend use SuspendingAsyncEnumerable, because taskSeq blocks do not resume correctly in Debug builds. +let gatedNumbers (gate : Task) = + SuspendingAsyncEnumerable(fun _ index -> task { + match index with + | 0 -> return ValueSome 1 + | 1 -> + do! gate + return ValueSome 2 + | _ -> return ValueNone + }) + :> IAsyncEnumerable + +let failingNumbers () = taskSeq { + yield 1 + yield 2 + failwith "Boom during enumeration" +} + +/// Simulates a paged sequence that exposes the size of its pages +type PagedAsyncEnumerable<'T> (pageSize : int, items : 'T list) = + member _.PageSize = pageSize + interface IAsyncEnumerable<'T> with + member _.GetAsyncEnumerator cancellationToken = (asyncItems items).GetAsyncEnumerator cancellationToken + +let pageSizeOf (source : IAsyncEnumerable) = + match source with + | :? PagedAsyncEnumerable as paged -> ValueSome paged.PageSize + | _ -> ValueNone + +/// Splits the items into Azure SDK pages of the given size +let azurePages (pageSize : int) (items : int list) = + let chunks = items |> List.chunkBySize pageSize + chunks + |> List.mapi (fun index chunk -> + let continuationToken = + if index < chunks.Length - 1 then + string (index + 1) + else + null + // The pages are not produced by a service call, so there is no raw response to attach + Page.FromValues(List.toArray chunk, continuationToken, Unchecked.defaultof)) + +/// +/// An Azure SDK paged sequence that remembers the page size hint it was requested with. +/// +/// +/// does not expose a page size, because the size is only a hint passed to +/// , so an application has to keep the hint to batch streamed items by pages. +/// +type HintedAsyncPageable<'T> (pageSizeHint : int, pages : Page<'T> list) = + inherit AsyncPageable<'T> () + member _.PageSizeHint = pageSizeHint + override _.AsPages (continuationToken, pageSizeHint) = AsyncPageable<'T>.FromPages(pages).AsPages(continuationToken, pageSizeHint) + +let azurePageSizeOf (source : IAsyncEnumerable) = + match source with + | :? HintedAsyncPageable as pageable -> ValueSome pageable.PageSizeHint + | _ -> ValueNone + +let StreamItemType = + Define.Object( + "StreamItem", + [ + Define.Field ("id", IntType, fun _ (item : StreamItem) -> item.Id) + Define.AsyncField ("value", StringType, fun _ (item : StreamItem) -> item.Value) + ] + ) + +let immediateItems = [ { Id = 1; Value = async { return "one" } }; { Id = 2; Value = async { return "two" } } ] + +let slowAndFastItems = [ { Id = 1; Value = delay 3000 "slow" }; { Id = 2; Value = async { return "fast" } } ] + +let schemaConfig = + SchemaConfig.DefaultWithBufferedStream (streamOptions = { Interval = ValueNone; PreferredBatchSize = ValueNone }) + +let executorFor (fields : FieldDef list) = Executor (Schema (Define.Object("Query", fields), config = schemaConfig)) + +let executeQuery (executor : Executor) (query : string) = + executor.AsyncExecute (parse query, getMockInputContext, ()) + |> sync + +let fieldError (message : string) (fieldName : string) = GQLProblemDetails.CreateWithKind (message, Execution, [ box fieldName ]) + +/// Builds the deferred payload of streamed items given as (index, value) pairs +let streamedBatch (fieldName : string) (items : (int * int) list) = + match items with + | [ index, value ] -> DeferredResult ([| box value |], [ box fieldName; box index ]) + | _ -> DeferredResult (items |> List.map (snd >> box) |> List.toArray, [ box fieldName; box (items |> List.map (fst >> box)) ]) + +let waitForCompletion (deferred : IObservable) = + use subscription = Observer.create deferred + subscription.WaitCompleted (timeout = ms 10) + subscription.Received |> Seq.toList + +[] +let ``TaskSeq field without directives returns the whole sequence as a list`` () = + let executor = + executorFor [ + Define.TaskSeqField ("numbers", ListOf IntType, fun _ _ -> asyncItems [ 1; 2; 3 ]) + Define.TaskSeqField ("items", ListOf StreamItemType, fun _ _ -> asyncItems immediateItems) + ] + let expectedData = + NameValueLookup.ofList [ + "numbers", upcast [| box 1; box 2; box 3 |] + "items", + upcast + [| + box (NameValueLookup.ofList [ "id", upcast 1; "value", upcast "one" ]) + box (NameValueLookup.ofList [ "id", upcast 2; "value", upcast "two" ]) + |] + ] + let result = executeQuery executor "{ numbers items { id value } }" + ensureDirect result + <| fun data errors -> + empty errors + data |> equals (upcast expectedData) + +[] +let ``TaskSeq field without directives waits for a sequence that suspends`` () = + let executor = + executorFor [ Define.TaskSeqField ("numbers", ListOf IntType, fun _ _ -> gatedNumbers Task.CompletedTask) ] + let expectedData = NameValueLookup.ofList [ "numbers", upcast [| box 1; box 2 |] ] + let result = executeQuery executor "{ numbers }" + ensureDirect result + <| fun data errors -> + empty errors + data |> equals (upcast expectedData) + +[] +let ``TaskSeq field with defer directive delivers the whole list in one deferred payload`` () = + let executor = + executorFor [ + Define.TaskSeqField ("numbers", Nullable (ListOf IntType), fun _ _ -> Some (asyncItems [ 1; 2; 3 ])) + ] + let expectedData = NameValueLookup.ofList [ "numbers", null ] + let result = executeQuery executor "{ numbers @defer }" + ensureDeferred result + <| fun data errors deferred -> + empty errors + data |> equals (upcast expectedData) + waitForCompletion deferred + |> single + |> equals (DeferredResult ([| box 1; box 2; box 3 |], [ box "numbers" ])) + +[] +let ``TaskSeq field with stream directive delivers items before the sequence completes`` () = + let gate = TaskCompletionSource () + let executor = + executorFor [ Define.TaskSeqField ("numbers", ListOf IntType, fun _ _ -> gatedNumbers gate.Task) ] + let expectedData = NameValueLookup.ofList [ "numbers", upcast [] ] + use firstReceived = new ManualResetEventSlim false + let result = executeQuery executor "{ numbers @stream }" + ensureDeferred result + <| fun data errors deferred -> + empty errors + data |> equals (upcast expectedData) + use subscription = + deferred + |> Observer.createWithCallback (fun _ _ -> firstReceived.Set ()) + if not (firstReceived.Wait (TimeSpan.FromSeconds (float (ms 5)))) then + fail "Timeout while waiting for the first streamed item" + // The sequence is blocked on the gate, so only its first item can have been delivered + Assert.False (subscription.IsCompleted, "The stream must not complete before the sequence produces its last item") + subscription.Received + |> single + |> equals (streamedBatch "numbers" [ 0, 1 ]) + gate.SetResult () + subscription.WaitCompleted (timeout = ms 10) + subscription.Received + |> seqEquals [ streamedBatch "numbers" [ 0, 1 ]; streamedBatch "numbers" [ 1, 2 ] ] + +[] +let ``TaskSeq field with stream directive emits each item as soon as its fields are resolved`` () = + // maxConcurrency is explicit (rather than the Environment.ProcessorCount default) so the fast item is + // guaranteed a concurrent slot alongside the slow one and the assertion below does not depend on the runner's + // CPU count + let executor = + executorFor [ + Define.TaskSeqField ("items", ListOf StreamItemType, (fun _ _ -> asyncItems slowAndFastItems), maxConcurrency = 2) + ] + let result = executeQuery executor "{ items @stream { id value } }" + ensureDeferred result + <| fun _ errors deferred -> + empty errors + waitForCompletion deferred + |> seqEquals [ + DeferredResult ([| box (NameValueLookup.ofList [ "id", upcast 2; "value", upcast "fast" ]) |], [ box "items"; box 1 ]) + DeferredResult ([| box (NameValueLookup.ofList [ "id", upcast 1; "value", upcast "slow" ]) |], [ box "items"; box 0 ]) + ] + +[] +let ``TaskSeq field with stream directive groups items by the preferred batch size of the query`` () = + let executor = + executorFor [ Define.TaskSeqField ("numbers", ListOf IntType, fun _ _ -> asyncItems [ 1; 2; 3; 4; 5 ]) ] + let result = executeQuery executor "{ numbers @stream(preferredBatchSize: 2) }" + ensureDeferred result + <| fun _ errors deferred -> + empty errors + waitForCompletion deferred + |> seqEquals [ + streamedBatch "numbers" [ 0, 1; 1, 2 ] + streamedBatch "numbers" [ 2, 3; 3, 4 ] + streamedBatch "numbers" [ 4, 5 ] + ] + +[] +let ``TaskSeq field with fixed batching groups streamed items without query arguments`` () = + let executor = + executorFor [ + Define.TaskSeqField ("numbers", ListOf IntType, (fun _ _ -> asyncItems [ 1; 2; 3; 4; 5 ]), batching = StreamBatching.Fixed 2) + ] + let result = executeQuery executor "{ numbers @stream }" + ensureDeferred result + <| fun _ errors deferred -> + empty errors + waitForCompletion deferred + |> seqEquals [ + streamedBatch "numbers" [ 0, 1; 1, 2 ] + streamedBatch "numbers" [ 2, 3; 3, 4 ] + streamedBatch "numbers" [ 4, 5 ] + ] + +[] +let ``TaskSeq field with batching from source groups streamed items by the page size of the source`` () = + let executor = + executorFor [ + Define.TaskSeqField ( + "numbers", + ListOf IntType, + (fun _ _ -> PagedAsyncEnumerable (3, [ 1; 2; 3; 4; 5; 6 ]) :> IAsyncEnumerable), + batching = StreamBatching.FromSource pageSizeOf + ) + ] + let result = executeQuery executor "{ numbers @stream }" + ensureDeferred result + <| fun _ errors deferred -> + empty errors + waitForCompletion deferred + |> seqEquals [ streamedBatch "numbers" [ 0, 1; 1, 2; 2, 3 ]; streamedBatch "numbers" [ 3, 4; 4, 5; 5, 6 ] ] + +[] +let ``TaskSeq field with batching from source delivers items one by one when the source has no page size`` () = + let executor = + executorFor [ + Define.TaskSeqField ("numbers", ListOf IntType, (fun _ _ -> asyncItems [ 1; 2 ]), batching = StreamBatching.FromSource pageSizeOf) + ] + let result = executeQuery executor "{ numbers @stream }" + ensureDeferred result + <| fun _ errors deferred -> + empty errors + waitForCompletion deferred + |> seqEquals [ streamedBatch "numbers" [ 0, 1 ]; streamedBatch "numbers" [ 1, 2 ] ] + +[] +let ``TaskSeq field backed by Azure AsyncPageable returns items of all pages without directives`` () = + let executor = + executorFor [ + Define.TaskSeqField ("numbers", ListOf IntType, fun _ _ -> AsyncPageable.FromPages(azurePages 2 [ 1..5 ]) :> IAsyncEnumerable) + ] + let expectedData = NameValueLookup.ofList [ "numbers", upcast [| box 1; box 2; box 3; box 4; box 5 |] ] + let result = executeQuery executor "{ numbers }" + ensureDirect result + <| fun data errors -> + empty errors + data |> equals (upcast expectedData) + +[] +let ``TaskSeq field backed by Azure AsyncPageable streams items in batches of the kept page size hint`` () = + let executor = + executorFor [ + Define.TaskSeqField ( + "numbers", + ListOf IntType, + (fun _ _ -> HintedAsyncPageable (2, azurePages 2 [ 1..5 ]) :> IAsyncEnumerable), + batching = StreamBatching.FromSource azurePageSizeOf + ) + ] + let result = executeQuery executor "{ numbers @stream }" + ensureDeferred result + <| fun _ errors deferred -> + empty errors + waitForCompletion deferred + |> seqEquals [ + streamedBatch "numbers" [ 0, 1; 1, 2 ] + streamedBatch "numbers" [ 2, 3; 3, 4 ] + streamedBatch "numbers" [ 4, 5 ] + ] + +[] +let ``TaskSeq field backed by plain Azure AsyncPageable streams items one by one because it has no page size`` () = + let executor = + executorFor [ + Define.TaskSeqField ( + "numbers", + ListOf IntType, + (fun _ _ -> AsyncPageable.FromPages(azurePages 2 [ 1..3 ]) :> IAsyncEnumerable), + batching = StreamBatching.FromSource azurePageSizeOf + ) + ] + let result = executeQuery executor "{ numbers @stream }" + ensureDeferred result + <| fun _ errors deferred -> + empty errors + waitForCompletion deferred + |> seqEquals [ + streamedBatch "numbers" [ 0, 1 ] + streamedBatch "numbers" [ 1, 2 ] + streamedBatch "numbers" [ 2, 3 ] + ] + +[] +let ``Preferred batch size of the stream directive overrides the batching of the TaskSeq field`` () = + let executor = + executorFor [ + Define.TaskSeqField ("numbers", ListOf IntType, (fun _ _ -> asyncItems [ 1; 2; 3 ]), batching = StreamBatching.Fixed 3) + ] + let result = executeQuery executor "{ numbers @stream(preferredBatchSize: 1) }" + ensureDeferred result + <| fun _ errors deferred -> + empty errors + waitForCompletion deferred + |> seqEquals [ + streamedBatch "numbers" [ 0, 1 ] + streamedBatch "numbers" [ 1, 2 ] + streamedBatch "numbers" [ 2, 3 ] + ] + +[] +let ``Batching from source runs only for a stream query that does not override the batch size, and only once`` () = + let mutable callCount = 0 + let batching = + StreamBatching.FromSource (fun _ -> + callCount <- callCount + 1 + ValueNone) + let executor = + executorFor [ + Define.TaskSeqField ("numbers", ListOf IntType, (fun _ _ -> asyncItems [ 1; 2; 3 ]), batching = batching) + Define.TaskSeqField ("deferrable", Nullable (ListOf IntType), (fun _ _ -> Some (asyncItems [ 1; 2 ])), batching = batching) + ] + executeQuery executor "{ numbers }" |> ignore + callCount |> equals 0 + executeQuery executor "{ deferrable @defer }" |> ignore + callCount |> equals 0 + executeQuery executor "{ numbers @stream(preferredBatchSize: 1) }" |> ignore + callCount |> equals 0 + executeQuery executor "{ numbers @stream }" |> ignore + callCount |> equals 1 + +[] +let ``Throwing batching callback does not affect a query that does not stream the field`` () = + let throwingBatching = StreamBatching.FromSource (fun _ -> failwith "Batching must not run for this query") + let executor = + executorFor [ + Define.TaskSeqField ("numbers", ListOf IntType, (fun _ _ -> asyncItems [ 1; 2; 3 ]), batching = throwingBatching) + ] + let expectedData = NameValueLookup.ofList [ "numbers", upcast [| box 1; box 2; box 3 |] ] + let result = executeQuery executor "{ numbers }" + ensureDirect result + <| fun data errors -> + empty errors + data |> equals (upcast expectedData) + +[] +let ``Nullable TaskSeq field that fails during enumeration returns null with a field error`` () = + let executor = + executorFor [ Define.TaskSeqField ("numbers", Nullable (ListOf IntType), fun _ _ -> Some (failingNumbers ())) ] + let expectedData = NameValueLookup.ofList [ "numbers", null ] + let result = executeQuery executor "{ numbers }" + ensureDirect result + <| fun data errors -> + data |> equals (upcast expectedData) + errors + |> equals [ fieldError "Boom during enumeration" "numbers" ] + +[] +let ``Non-nullable TaskSeq field that fails during enumeration propagates the error`` () = + let executor = + executorFor [ Define.TaskSeqField ("numbers", ListOf IntType, fun _ _ -> failingNumbers ()) ] + let result = executeQuery executor "{ numbers }" + ensureDirect result + <| fun data errors -> + Assert.Null data + errors + |> single + |> equals (fieldError "Boom during enumeration" "numbers") + +[] +let ``Streamed TaskSeq field that fails during enumeration delivers produced items and then the error`` () = + let executor = + executorFor [ + Define.TaskSeqField ("failing", ListOf IntType, fun _ _ -> failingNumbers ()) + Define.TaskSeqField ("numbers", ListOf IntType, fun _ _ -> asyncItems [ 10; 20 ]) + ] + let expectedData = + NameValueLookup.ofList [ "failing", upcast []; "numbers", upcast [| box 10; box 20 |] ] + let result = executeQuery executor "{ failing @stream numbers }" + ensureDeferred result + <| fun data errors deferred -> + empty errors + data |> equals (upcast expectedData) + waitForCompletion deferred + |> seqEquals [ + streamedBatch "failing" [ 0, 1 ] + streamedBatch "failing" [ 1, 2 ] + DeferredErrors (null, [ fieldError "Boom during enumeration" "failing" ], [ box "failing" ]) + ] + +[] +let ``Streamed TaskSeq field that fails acquiring the enumerator still delivers its DeferredErrors`` () = + // Regression test: this used to fault the merged deferred observable of the whole query instead of producing + // this field's DeferredErrors, which would drop sibling deferred results and the final completion payload + let executor = + executorFor [ + Define.TaskSeqField ("failing", ListOf IntType, fun _ _ -> ThrowingAsyncEnumerable "Boom acquiring the enumerator" :> IAsyncEnumerable) + Define.TaskSeqField ("numbers", ListOf IntType, fun _ _ -> asyncItems [ 10; 20 ]) + ] + let expectedData = + NameValueLookup.ofList [ "failing", upcast []; "numbers", upcast [| box 10; box 20 |] ] + let result = executeQuery executor "{ failing @stream numbers }" + ensureDeferred result + <| fun data errors deferred -> + empty errors + data |> equals (upcast expectedData) + waitForCompletion deferred + |> seqEquals [ DeferredErrors (null, [ fieldError "Boom acquiring the enumerator" "failing" ], [ box "failing" ]) ] + +[] +let ``Streamed TaskSeq field emits a slower earlier item before the enumeration failure that follows it`` () = + // Regression test: an item resolved asynchronously must not be overtaken by a failure of the source that + // is pulled right after it, even though the failure itself completes immediately + let executor = + executorFor [ + Define.TaskSeqField ("items", ListOf StreamItemType, fun _ _ -> itemThenFailure { Id = 1; Value = delay 500 "slow" }) + ] + let result = executeQuery executor "{ items @stream { id value } }" + ensureDeferred result + <| fun _ errors deferred -> + empty errors + waitForCompletion deferred + |> seqEquals [ + DeferredResult ([| box (NameValueLookup.ofList [ "id", upcast 1; "value", upcast "slow" ]) |], [ box "items"; box 0 ]) + DeferredErrors (null, [ fieldError "Boom during enumeration" "items" ], [ box "items" ]) + ] + +[] +let ``Streamed TaskSeq field delivers an item's own resolver error and keeps streaming the items after it`` () = + // Regression test: an item whose own field resolution fails is a normal (non-throwing) result as far as the + // streaming operator is concerned, so it must not be mistaken for a failure of the source or of the enumeration + // itself: the item's error is delivered on its own path and later items keep streaming, exactly like @stream on + // an ordinary list (see DeferredTests."Resolver list error") + let items = [ { Id = 1; Value = async { return failwith "Boom resolving the item" } }; { Id = 2; Value = async { return "two" } } ] + let executor = + executorFor [ + Define.TaskSeqField ("items", ListOf StreamItemType, (fun _ _ -> asyncItems items), maxConcurrency = 1) + ] + let result = executeQuery executor "{ items @stream { id value } }" + ensureDeferred result + <| fun _ errors deferred -> + empty errors + waitForCompletion deferred + |> seqEquals [ + DeferredErrors ( + null, + [ GQLProblemDetails.CreateWithKind ("Boom resolving the item", Execution, [ box "items"; box 0; box "value" ]) ], + [ box "items"; box 0 ] + ) + DeferredResult ([| box (NameValueLookup.ofList [ "id", upcast 2; "value", upcast "two" ]) |], [ box "items"; box 1 ]) + ] + +[] +let ``A batch containing a failed item alongside a succeeding one is delivered as one DeferredErrors event`` () = + // Regression test for the tenth Copilot review thread PRRT_kwDOA0s7t86i5Vu-, which claimed that + // Execution.collectItems' chunk branch omits a failed item's index from `indicies` while still reserving its + // slot in `data`, so GraphQLWebsocketMiddleware.splitBatch's List.map2 would throw on a mixed success/error + // batch. It does not: both arms of `merge` prepend the item's index, so `indicies` and `data` always end up the + // same length as the chunk, with the failed item's slot left null. This pins that shape end to end. + let items = [ { Id = 1; Value = async { return failwith "Boom resolving item 0" } }; { Id = 2; Value = async { return "two" } } ] + let executor = + executorFor [ + Define.TaskSeqField ("items", ListOf StreamItemType, (fun _ _ -> asyncItems items), batching = StreamBatching.Fixed 2, maxConcurrency = 1) + ] + let result = executeQuery executor "{ items @stream { id value } }" + ensureDeferred result + <| fun _ errors deferred -> + empty errors + waitForCompletion deferred + |> seqEquals [ + DeferredErrors ( + [| null; box (NameValueLookup.ofList [ "id", upcast 2; "value", upcast "two" ]) |], + [ GQLProblemDetails.CreateWithKind ("Boom resolving item 0", Execution, [ box "items"; box 0; box "value" ]) ], + [ box "items"; box [ box 0; box 1 ] ] + ) + ] + +[] +let ``Disposing the stream subscription stops the enumeration of the TaskSeq field`` () : Task = task { + let pulled = ref 0 + let disposed = TaskCompletionSource () + let received = TaskCompletionSource () + let executor = + executorFor [ Define.TaskSeqField ("numbers", ListOf IntType, fun _ _ -> endlessNumbers pulled disposed) ] + let! result = executor.AsyncExecute (parse "{ numbers @stream }", getMockInputContext, ()) + match result.Content with + | Deferred (_, errors, deferred) -> + empty errors + let subscription = deferred |> Observable.subscribe (fun _ -> received.TrySetResult () |> ignore) + do! waitForTask (TimeSpan.FromSeconds (float (ms 5))) "Timeout while waiting for the first streamed item" received.Task + subscription.Dispose () + do! + waitForTask + (TimeSpan.FromSeconds (float (ms 5))) + "The sequence enumerator was not disposed after the subscription had been disposed" + disposed.Task + let pulledAfterDisposal = pulled.Value + // A still running enumeration would pull more items during this delay + do! Task.Delay 200 + pulled.Value |> equals pulledAfterDisposal + | response -> fail $"Expected a 'Deferred' GQLResponse but got\n{response}" +} + +[] +let ``TaskSeq field with stream directive never resolves more than maxConcurrency items at the same time`` () = + let inFlight = ref 0 + let maxObserved = ref 0 + let trackConcurrency (work : Async<'T>) : Async<'T> = async { + let current = Interlocked.Increment inFlight + let mutable observed = maxObserved.Value + while current > observed && Interlocked.CompareExchange (maxObserved, current, observed) <> observed do + observed <- maxObserved.Value + try + return! work + finally + Interlocked.Decrement inFlight |> ignore + } + let items = [ for id in 1 .. 6 -> { Id = id; Value = trackConcurrency (delay 100 (string id)) } ] + let executor = + executorFor [ + Define.TaskSeqField ("items", ListOf StreamItemType, (fun _ _ -> asyncItems items), maxConcurrency = 2) + ] + let result = executeQuery executor "{ items @stream { id value } }" + ensureDeferred result + <| fun _ errors deferred -> + empty errors + let received = waitForCompletion deferred + received |> List.length |> equals 6 + Assert.True (maxObserved.Value <= 2, $"Expected at most 2 concurrent item resolutions, but observed {maxObserved.Value}") + +[] +let ``TaskSeqField with a non-positive maxConcurrency fails at definition time`` () = + throws (fun () -> + Define.TaskSeqField ("numbers", ListOf IntType, (fun _ _ -> asyncItems [ 1 ]), maxConcurrency = 0) |> ignore) + +[] +let ``TaskSeq field resolved as null reports a non-null field error`` () = + let executor = + executorFor [ + Define.TaskSeqField ("numbers", ListOf IntType, fun _ _ -> Unchecked.defaultof>) + ] + let result = executeQuery executor "{ numbers }" + ensureDirect result + <| fun data errors -> + Assert.Null data + hasError "Non-Null field numbers resolved as a null!" errors