diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 8508dc49f..64e3b9627 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -288,7 +288,7 @@ * **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** `SubscriptionExecutionResult.Data` is now `obj voption Skippable`, and the record has new `Path` and `HasNext` fields for incremental delivery * **Breaking Change** `BufferedStreamOptions.Interval` and `BufferedStreamOptions.PreferredBatchSize` are now `int voption` * **Breaking Change** `ServerMessage.Error` and `ServerRawPayload.ErrorMessages` now carry `GQLProblemDetails list` instead of `NameValueLookup list`, so an `error` message's `payload` is a standard GraphQL error array as the `graphql-transport-ws` protocol requires * **Breaking Change** A query or mutation whose non-null root field fails during execution now produces a `Direct` (execution) result with `null` data instead of a `RequestError`, which is now only ever produced for a request rejected before execution (validation, planning, variable or inline argument coercion, a middleware, or the executor itself failing); HTTP and `graphql-transport-ws` responses for such a failure now carry `data: null` as the spec requires, instead of omitting `data` entirely. This also changes the public `GQLResponse.Data`, `GQLResponseContent.Direct.Data`, `DeferredErrors.Data`, and `SubscriptionErrors.Data` signatures to use `voption` diff --git a/src/FSharp.Data.GraphQL.Server.AspNetCore/GraphQLSubscriptionsManagement.fs b/src/FSharp.Data.GraphQL.Server.AspNetCore/GraphQLSubscriptionsManagement.fs index 7cd4ba431..8c80e63b4 100644 --- a/src/FSharp.Data.GraphQL.Server.AspNetCore/GraphQLSubscriptionsManagement.fs +++ b/src/FSharp.Data.GraphQL.Server.AspNetCore/GraphQLSubscriptionsManagement.fs @@ -1,14 +1,16 @@ module internal FSharp.Data.GraphQL.Server.AspNetCore.GraphQLSubscriptionsManagement +open System + open FSharp.Data.GraphQL.Shared.WebSockets let addSubscription (id : SubscriptionId, unsubscriber : SubscriptionUnsubscriber, onUnsubscribe : OnUnsubscribeAction) (subscriptions : SubscriptionsDict) = - subscriptions.Add (id, (unsubscriber, onUnsubscribe)) + lock subscriptions (fun () -> subscriptions.Add (id, (unsubscriber, onUnsubscribe))) -let isIdTaken (id : SubscriptionId) (subscriptions : SubscriptionsDict) = subscriptions.ContainsKey (id) +let isIdTaken (id : SubscriptionId) (subscriptions : SubscriptionsDict) = lock subscriptions (fun () -> subscriptions.ContainsKey (id)) let executeOnUnsubscribeAndDispose (id : SubscriptionId) (subscription : SubscriptionUnsubscriber * OnUnsubscribeAction) = match subscription with @@ -19,15 +21,37 @@ let executeOnUnsubscribeAndDispose (id : SubscriptionId) (subscription : Subscri unsubscriber.Dispose () let removeSubscription (id : SubscriptionId) (subscriptions : SubscriptionsDict) = - match subscriptions.TryGetValue id with - | true, sub -> - sub |> executeOnUnsubscribeAndDispose id - subscriptions.Remove (id) |> ignore - | false, _ -> () + let subscription = + lock subscriptions (fun () -> + match subscriptions.TryGetValue id with + | true, sub -> + subscriptions.Remove (id) |> ignore + ValueSome sub + | false, _ -> ValueNone) + + match subscription with + | ValueSome sub -> sub |> executeOnUnsubscribeAndDispose id + | ValueNone -> () let removeAllSubscriptions (subscriptions : SubscriptionsDict) = - subscriptions - |> Seq.iter (fun subscription -> - subscription.Value - |> executeOnUnsubscribeAndDispose subscription.Key) - subscriptions.Clear () + let subscriptionsToDispose = + lock subscriptions (fun () -> + let snapshot = + subscriptions + |> Seq.map (fun subscription -> struct (subscription.Key, subscription.Value)) + |> Seq.toArray + + subscriptions.Clear () + snapshot) + + let exceptions = ResizeArray () + + subscriptionsToDispose + |> Array.iter (fun struct (id, subscription) -> + try + subscription |> executeOnUnsubscribeAndDispose id + with ex -> + exceptions.Add ex) + + if exceptions.Count > 0 then + raise (AggregateException ("One or more subscriptions failed to unsubscribe.", exceptions)) diff --git a/src/FSharp.Data.GraphQL.Server.AspNetCore/GraphQLWebsocketMiddleware.fs b/src/FSharp.Data.GraphQL.Server.AspNetCore/GraphQLWebsocketMiddleware.fs index d399a879b..b4fcd6cc3 100644 --- a/src/FSharp.Data.GraphQL.Server.AspNetCore/GraphQLWebsocketMiddleware.fs +++ b/src/FSharp.Data.GraphQL.Server.AspNetCore/GraphQLWebsocketMiddleware.fs @@ -40,18 +40,22 @@ 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 trySkipPathPrefix (prefix : obj list) (path : obj list) = - let rec loop prefix path = - match prefix, path with - | [], remainingPath -> ValueSome remainingPath - | _ :: _, [] -> ValueNone - | prefixHead :: prefixTail, pathHead :: pathTail when prefixHead = pathHead -> loop prefixTail pathTail - | _ -> ValueNone - loop prefix path + let pathStartsWith (prefix : obj list) (path : obj list) = + let prefixLength = List.length prefix + List.length path >= prefixLength + && List.truncate prefixLength path = prefix + + let tryGetPathItemIndex (fieldPath : obj list) (path : obj list) = + let fieldPathLength = List.length fieldPath + + if pathStartsWith fieldPath path then + path |> List.vtryItem fieldPathLength + else + ValueNone /// 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 -> ValueSome (List.rev fieldPathRev, indices) @@ -62,23 +66,67 @@ module internal IncrementalPayloadSplitting = /// 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[] - let errorsByIndex = + let errorsByItemIndex = errors |> Seq.vchoose (fun error -> error.Path |> Skippable.toValueOption - |> ValueOption.bind (trySkipPathPrefix fieldPath) - |> ValueOption.bind (function - | itemIndex :: _ -> ValueSome struct (itemIndex, error) - | [] -> ValueNone)) - |> _.ToLookup((fun struct (itemIndex, _) -> itemIndex), (fun struct (_, error) -> error)) + |> ValueOption.bind (tryGetPathItemIndex fieldPath) + |> ValueOption.map (fun index -> struct (index, error))) + |> _.ToLookup((fun struct (index, _) -> index), (fun struct (_, error) -> error)) + (indices, List.ofArray items) ||> List.map2 (fun index item -> let itemPath = [ yield! fieldPath; yield index ] - let itemErrors = errorsByIndex[index] |> List.ofSeq + let itemErrors = errorsByItemIndex[index] |> Seq.toList box [| item |], itemErrors, itemPath) +module internal ObservableErrorHandling = + + [] + let UnexpectedObservableErrorMessage = "Unexpected error during subscription" + + let private deduplicationKey (problem : GQLProblemDetails) = + let extensions = + problem.Extensions + |> Skippable.toValueOption + |> ValueOption.map ( + Seq.sortBy _.Key + >> Seq.map (fun kvp -> kvp.Key, kvp.Value) + >> Seq.toList + ) + + problem.Message, problem.Path, problem.Locations, extensions + + let rec problemDetailsOfObservableError (ex : exn) = + match ex with + | :? AggregateException as aggregate -> + let problemDetails = + aggregate.Flatten().InnerExceptions + |> Seq.collect problemDetailsOfObservableError + |> Seq.distinctBy deduplicationKey + |> Seq.toList + + match problemDetails with + | [] -> [ GQLProblemDetails.Create UnexpectedObservableErrorMessage ] + | _ -> problemDetails + | _ -> + match box ex with + | :? IGQLError as error -> [ GQLProblemDetails.OfError error ] + | _ -> [ GQLProblemDetails.Create UnexpectedObservableErrorMessage ] + + let sanitizeRequestError (problemDetails : GQLProblemDetails) = + match + problemDetails.Exception + |> ValueOption.map box + |> ValueOption.toObj + with + | :? IGQLError -> problemDetails + | :? exn -> GQLProblemDetails.Create UnexpectedObservableErrorMessage + | _ -> problemDetails + open IncrementalPayloadSplitting +open ObservableErrorHandling type GraphQLWebSocketMiddleware<'Root> ( @@ -93,6 +141,7 @@ type GraphQLWebSocketMiddleware<'Root> let serializerOptions = options.SerializerOptions let pingHandler = options.WebsocketOptions.CustomPingHandler let connectionInitTimeout = options.WebsocketOptions.ConnectionInitTimeout + let gracefulCloseTimeout : TimeSpan = TimeSpan.FromSeconds 5.0 let serializeServerMessage (jsonSerializerOptions : JsonSerializerOptions) (serverMessage : ServerMessage) = task { let raw = @@ -100,30 +149,41 @@ type GraphQLWebSocketMiddleware<'Root> | ConnectionAck -> { Id = ValueNone; Type = "connection_ack"; Payload = ValueNone } | ServerPing -> { Id = ValueNone; Type = "ping"; Payload = ValueNone } | ServerPong p -> { Id = ValueNone; Type = "pong"; Payload = p |> ValueOption.map CustomResponse } - | Next (id, payload) -> { Id = ValueSome id; Type = "next"; Payload = ValueSome <| ExecutionResult payload } + | Next (id, payload) -> { + Id = ValueSome id + Type = "next" + Payload = ValueSome <| ExecutionResult payload + } | Complete id -> { Id = ValueSome id; Type = "complete"; Payload = ValueNone } - | Error (id, errMessages) -> { Id = ValueSome id; Type = "error"; Payload = ValueSome <| ErrorMessages errMessages } + | Error (id, errMessages) -> { + Id = ValueSome id + Type = "error" + Payload = ValueSome <| ErrorMessages errMessages + } return JsonSerializer.Serialize (raw, jsonSerializerOptions) } static let invalidJsonInClientMessageError = - Result.Error <| InvalidMessage (4400, "Invalid json in client message") + Result.Error + <| InvalidMessage (4400, "Invalid json in client message") let deserializeClientMessage (serializerOptions : JsonSerializerOptions) (msg : IReadOnlyPooledList) = taskResult { try - return JsonSerializer.Deserialize (msg.Span, serializerOptions) + return JsonSerializer.Deserialize(msg.Span, serializerOptions) with | :? InvalidWebsocketMessageException as ex -> - logger.LogError(ex, "Invalid websocket message:\n{payload}", msg) - return! Result.Error <| InvalidMessage (4400, ex.Message.ToString ()) - | :? JsonException as ex when logger.IsEnabled(LogLevel.Trace) -> - logger.LogError(ex, "Cannot deserialize WebSocket message:\n{payload}", msg) + logger.LogError (ex, "Invalid websocket message:\n{payload}", msg) + return! + Result.Error + <| InvalidMessage (4400, ex.Message.ToString ()) + | :? JsonException as ex when logger.IsEnabled (LogLevel.Trace) -> + logger.LogError (ex, "Cannot deserialize WebSocket message:\n{payload}", msg) return! invalidJsonInClientMessageError | :? JsonException as ex -> - logger.LogError(ex, "Cannot deserialize WebSocket message") + logger.LogError (ex, "Cannot deserialize WebSocket message") return! invalidJsonInClientMessageError | ex -> - logger.LogError(ex, $"Unexpected exception '{ex.GetType().Name}' in GraphQLWebsocketMiddleware") + logger.LogError (ex, $"Unexpected exception '{ex.GetType().Name}' in GraphQLWebsocketMiddleware") return! invalidJsonInClientMessageError } @@ -156,10 +216,10 @@ type GraphQLWebSocketMiddleware<'Root> let message = completeMessage |> Seq.filter (fun x -> x > 0uy) - |> Array.ofSeq + |> Seq.toArray |> System.Text.Encoding.UTF8.GetString logger.LogInformation ("-> Request: {request}", message) - if completeMessage.All(fun b -> b = 0uy) then + if completeMessage.All (fun b -> b = 0uy) then return ValueNone else let! result = deserializeClientMessage serializerOptions completeMessage @@ -168,38 +228,65 @@ type GraphQLWebSocketMiddleware<'Root> ArrayPool.Shared.Return buffer } - let sendMessageViaSocket (jsonSerializerOptions) (socket : WebSocket) (message : ServerMessage) : Task = task { - if not (socket.State = WebSocketState.Open) then - logger.LogTrace ($"Ignoring message to be sent via socket, since its state is not '{nameof WebSocketState.Open}', but '{{state}}'", socket.State) - else - // TODO: Allocate string only if a debugger is attached - let! serializedMessage = message |> serializeServerMessage jsonSerializerOptions - let segment = ArraySegment(System.Text.Encoding.UTF8.GetBytes (serializedMessage)) - if not (socket.State = WebSocketState.Open) then - logger.LogTrace ($"Ignoring message to be sent via socket, since its state is not '{nameof WebSocketState.Open}', but '{{state}}'", socket.State) - else - do! socket.SendAsync (segment, WebSocketMessageType.Text, endOfMessage = true, cancellationToken = CancellationToken.None) + let sendMessageViaSocket (sendGate : SemaphoreSlim) (jsonSerializerOptions) (socket : WebSocket) (message : ServerMessage) : Task = task { + do! sendGate.WaitAsync () + try logger.LogTrace ("<- Response: {response}", message) + + if not (socket.State = WebSocketState.Open) then + logger.LogTrace ( + $"Ignoring message to be sent via socket, since its state is not '{nameof WebSocketState.Open}', but '{{state}}'", + socket.State + ) + else + // TODO: Allocate string only if a debugger is attached + let! serializedMessage = message |> serializeServerMessage jsonSerializerOptions + let segment = ArraySegment(System.Text.Encoding.UTF8.GetBytes serializedMessage) + + if not (socket.State = WebSocketState.Open) then + logger.LogTrace ( + $"Ignoring message to be sent via socket, since its state is not '{nameof WebSocketState.Open}', but '{{state}}'", + socket.State + ) + else + do! socket.SendAsync (segment, WebSocketMessageType.Text, endOfMessage = true, cancellationToken = CancellationToken.None) + finally + sendGate.Release () |> ignore } let addClientSubscription (id : SubscriptionId) (howToSendDataOnNext : SubscriptionId -> 'ResponseContent -> Task) - (subscriptions : SubscriptionsDict, - socket : WebSocket, - streamSource : IObservable<'ResponseContent>, - jsonSerializerOptions : JsonSerializerOptions) + (subscriptions : SubscriptionsDict, streamSource : IObservable<'ResponseContent>, sendMsg : ServerMessage -> Task) = + let sendTerminalError (ex : exn) = sendMsg (Error (id, problemDetailsOfObservableError ex)) + let observer = new Reactive.AnonymousObserver<'ResponseContent> ( - onNext = (fun theOutput -> (howToSendDataOnNext id theOutput).Wait ()), - onError = (fun ex -> logger.LogError (ex, "Error on subscription with Id = '{id}'", id)), + onNext = + (fun theOutput -> + try + (howToSendDataOnNext id theOutput).Wait() + with _ -> + subscriptions + |> GraphQLSubscriptionsManagement.removeSubscription id + reraise ()), + onError = + (fun ex -> + logger.LogError (ex, "Error on subscription with Id = '{id}'", id) + try + (sendTerminalError ex).Wait() + finally + subscriptions + |> GraphQLSubscriptionsManagement.removeSubscription (id)), onCompleted = (fun () -> - (sendMessageViaSocket jsonSerializerOptions socket (Complete id)).Wait () - subscriptions - |> GraphQLSubscriptionsManagement.removeSubscription (id)) + try + (sendMsg (Complete id)).Wait() + finally + subscriptions + |> GraphQLSubscriptionsManagement.removeSubscription id) ) // Registered before subscribing, so a stream that completes synchronously (from inside Subscribe) still @@ -215,40 +302,67 @@ type GraphQLWebSocketMiddleware<'Root> 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 + subscriptions + |> GraphQLSubscriptionsManagement.removeSubscription id reraise () - let tryToGracefullyCloseSocket (code, message) theSocket = - if theSocket |> canCloseSocket then - theSocket.CloseAsync (code, message, CancellationToken.None) - else - Task.CompletedTask + let tryToGracefullyCloseSocket (sendGate : SemaphoreSlim) (cancellationToken : CancellationToken) (code, message) (theSocket : WebSocket) : Task = + task { + do! sendGate.WaitAsync () + + try + if theSocket |> canCloseSocket then + use closeCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource cancellationToken + closeCancellationTokenSource.CancelAfter gracefulCloseTimeout - let tryToGracefullyCloseSocketWithDefaultBehavior = - tryToGracefullyCloseSocket (WebSocketCloseStatus.NormalClosure, "Normal Closure") + try + do! theSocket.CloseAsync (code, message, closeCancellationTokenSource.Token) + with :? OperationCanceledException -> + logger.LogWarning ( + "Aborting WebSocket after graceful close did not complete before cancellation. State = '{state}'", + theSocket.State + ) + theSocket.Abort () + else + logger.LogTrace ( + $"Ignoring socket close request, since its state is neither writable nor closeable, but '{{state}}'", + theSocket.State + ) + finally + sendGate.Release () |> ignore + } - let handleMessages (cancellationToken : CancellationToken) (httpContext : HttpContext) (socket : WebSocket) : Task = + let tryToGracefullyCloseSocketWithDefaultBehavior sendGate cancellationToken = + tryToGracefullyCloseSocket sendGate cancellationToken (WebSocketCloseStatus.NormalClosure, "Normal Closure") + + let handleMessages (sendGate : SemaphoreSlim) (cancellationToken : CancellationToken) (httpContext : HttpContext) (socket : WebSocket) : Task = let subscriptions = Dictionary() // ----------> // Helpers --> // ----------> let rcvMsgViaSocket = receiveMessageViaSocket (CancellationToken.None) - let sendMsg = sendMessageViaSocket serializerOptions socket + let sendMsg = sendMessageViaSocket sendGate serializerOptions socket let rcv () = socket |> rcvMsgViaSocket serializerOptions - let sendOutput id (output : SubscriptionExecutionResult) = - sendMsg (Next (id, output)) + let sendOutput id (output : SubscriptionExecutionResult) = sendMsg (Next (id, output)) let sendSubscriptionResponseOutput id subscriptionResult = match subscriptionResult with - | SubscriptionResult output -> SubscriptionExecutionResult.Create (output, []) |> sendOutput id + | SubscriptionResult output -> + SubscriptionExecutionResult.Create (output, []) + |> sendOutput id | SubscriptionErrors (output, errors) -> + // TODO: Use StringBuilder logger.LogWarning ("Subscription errors: {subscriptionErrors}", (String.Join ('\n', errors |> Seq.map (fun x -> $"- %s{x.Message}")))) // The executor may still have resolved partial data alongside the field errors; forward it as-is match output with - | ValueNone -> SubscriptionExecutionResult.CreateErrors errors |> sendOutput id - | ValueSome output -> SubscriptionExecutionResult.Create (output, errors) |> sendOutput id + | ValueNone -> + SubscriptionExecutionResult.CreateErrors errors + |> sendOutput id + | ValueSome 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. @@ -258,33 +372,48 @@ type GraphQLWebSocketMiddleware<'Root> match deferredResult with | ValueSome (DeferredResult (data, BatchPath (fieldPath, indices))) -> for itemData, _, itemPath in splitBatch fieldPath indices data [] do - do! SubscriptionExecutionResult.CreateIncremental (itemData, [], itemPath) |> sendOutput id + do! + SubscriptionExecutionResult.CreateIncremental (itemData, [], itemPath) + |> sendOutput id | ValueSome (DeferredResult (data, path)) -> - do! SubscriptionExecutionResult.CreateIncremental (data, [], path) |> sendOutput id + do! + SubscriptionExecutionResult.CreateIncremental (data, [], path) + |> sendOutput id | ValueSome (DeferredErrors (ValueSome data, errors, BatchPath (fieldPath, indices))) -> logger.LogWarning ( "Deferred response errors: {deferredErrors}", + // TODO: Use StringBuilder (String.Join ('\n', errors |> Seq.map (fun x -> $"- %s{x.Message}"))) ) for itemData, itemErrors, itemPath in splitBatch fieldPath indices data errors do - do! SubscriptionExecutionResult.CreateIncremental (itemData, itemErrors, itemPath) |> sendOutput id + do! + SubscriptionExecutionResult.CreateIncremental (itemData, itemErrors, itemPath) + |> sendOutput id | ValueSome (DeferredErrors (data, errors, path)) -> logger.LogWarning ( "Deferred response errors: {deferredErrors}", + // TODO: Use StringBuilder (String.Join ('\n', errors |> Seq.map (fun x -> $"- %s{x.Message}"))) ) - do! SubscriptionExecutionResult.CreateIncremental (data |> ValueOption.toObj, errors, path) |> sendOutput id - | ValueNone -> do! SubscriptionExecutionResult.CreateCompleted () |> sendOutput id + do! + SubscriptionExecutionResult.CreateIncremental (data |> ValueOption.toObj, errors, path) + |> sendOutput id + | ValueNone -> + do! + SubscriptionExecutionResult.CreateCompleted () + |> sendOutput id } let applyPlanExecutionResult (id : SubscriptionId) (socket) (executionResult : GQLExecutionResult) : Task = task { match executionResult with | Stream observableOutput -> - (subscriptions, socket, observableOutput, serializerOptions) + (subscriptions, observableOutput, sendMsg) |> addClientSubscription id sendSubscriptionResponseOutput | Deferred (data, errors, observableOutput) -> - do! SubscriptionExecutionResult.CreateInitial (data, errors) |> sendOutput id - (subscriptions, socket, observableOutput |> Observable.withCompletionMarker, serializerOptions) + do! + SubscriptionExecutionResult.CreateInitial (data, errors) + |> sendOutput id + (subscriptions, observableOutput |> Observable.withCompletionMarker, sendMsg) |> addClientSubscription id sendDeferredResponseOutput | Direct (data, errors) -> // An execution result, whose data is null when a non-null root field failed during execution; @@ -292,15 +421,18 @@ type GraphQLWebSocketMiddleware<'Root> // message below if not errors.IsEmpty then logger.LogWarning ("Execution errors:\n{errors}", errors) - do! SubscriptionExecutionResult.Create (data |> ValueOption.toObj, errors) |> sendOutput id + do! + SubscriptionExecutionResult.Create (data |> ValueOption.toObj, 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) + let sanitizedProblemDetails = problemDetails |> List.map sanitizeRequestError + logger.LogWarning ("Request errors:\n{errors}", problemDetails) // 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)) + do! sendMsg (Error (id, sanitizedProblemDetails)) } let logMsgReceivedWithOptionalPayload optionalPayload (msgAsStr : string) = @@ -319,85 +451,103 @@ type GraphQLWebSocketMiddleware<'Root> // -------> task { try - while not cancellationToken.IsCancellationRequested - && socket |> isSocketOpen do - let! receivedMessage = rcv () - match receivedMessage with - | Result.Error failureMessages -> - nameof InvalidMessage - |> logMsgReceivedWithOptionalPayload ValueNone - match failureMessages with - | InvalidMessage (code, explanation) -> do! socket.CloseAsync (enum code, explanation, CancellationToken.None) - | Ok ValueNone -> logger.LogTrace ("WebSocket received empty message! State = '{socketState}'", socket.State) - | Ok (ValueSome msg) -> - match msg with - | ConnectionInit p -> - nameof ConnectionInit |> logMsgReceivedWithOptionalPayload p - do! - socket.CloseAsync ( - enum CustomWebSocketStatus.TooManyInitializationRequests, - "Too many initialization requests", - CancellationToken.None - ) - | ClientPing p -> - nameof ClientPing |> logMsgReceivedWithOptionalPayload p - match pingHandler with - | ValueSome func -> - let! customP = p |> func serviceProvider - do! ServerPong customP |> sendMsg - | ValueNone -> do! ServerPong p |> sendMsg - | ClientPong p -> nameof ClientPong |> logMsgReceivedWithOptionalPayload p - | Subscribe (id, query) -> - try - nameof Subscribe |> logMsgWithIdReceived id - if subscriptions |> GraphQLSubscriptionsManagement.isIdTaken id then - do! - let warningMsg : FormattableString = $"Subscriber for Id = '{id}' already exists" - logger.LogWarning (String.Format (warningMsg.Format, "id"), id) - socket.CloseAsync ( - enum CustomWebSocketStatus.SubscriberAlreadyExists, - warningMsg.ToString (), - CancellationToken.None - ) - else - let variables = query.Variables |> Skippable.toValueOption - let getInputContext() = httpContext.RequestServices.GetRequiredService() - let! planExecutionResult = - let root = options.RootFactory httpContext - options.SchemaExecutor.AsyncExecute (query.Query, getInputContext, root, ?variables = variables) - do! planExecutionResult |> applyPlanExecutionResult id socket - with ex -> - logger.LogError (ex, "Unexpected error during subscription with id '{id}'", id) - do! sendMsg (Error (id, [ GQLProblemDetails.Create "Unexpected error during subscription" ])) - | ClientComplete id -> - "ClientComplete" |> logMsgWithIdReceived id - subscriptions - |> GraphQLSubscriptionsManagement.removeSubscription (id) - logger.LogTrace "Leaving the 'graphql-ws' connection loop..." - do! socket |> tryToGracefullyCloseSocketWithDefaultBehavior - with ex -> - logger.LogError (ex, "Cannot handle a message; dropping a websocket connection") - // At this point, only something really weird must have happened. - // In order to avoid faulty state scenarios and unimagined damages, - // just close the socket without further ado. - do! socket |> tryToGracefullyCloseSocketWithDefaultBehavior + try + while not cancellationToken.IsCancellationRequested + && socket |> isSocketOpen do + let! receivedMessage = rcv () + match receivedMessage with + | Result.Error failureMessages -> + nameof InvalidMessage + |> logMsgReceivedWithOptionalPayload ValueNone + match failureMessages with + | InvalidMessage (code, explanation) -> + do! + socket + |> tryToGracefullyCloseSocket sendGate cancellationToken (enum code, explanation) + | Ok ValueNone -> logger.LogTrace ("WebSocket received empty message! State = '{socketState}'", socket.State) + | Ok (ValueSome msg) -> + match msg with + | ConnectionInit p -> + nameof ConnectionInit |> logMsgReceivedWithOptionalPayload p + do! + socket + |> tryToGracefullyCloseSocket + sendGate + cancellationToken + (enum CustomWebSocketStatus.TooManyInitializationRequests, "Too many initialization requests") + | ClientPing p -> + nameof ClientPing |> logMsgReceivedWithOptionalPayload p + match pingHandler with + | ValueSome func -> + let! customP = p |> func serviceProvider + do! ServerPong customP |> sendMsg + | ValueNone -> do! ServerPong p |> sendMsg + | ClientPong p -> nameof ClientPong |> logMsgReceivedWithOptionalPayload p + | Subscribe (id, query) -> + try + nameof Subscribe |> logMsgWithIdReceived id + if subscriptions |> GraphQLSubscriptionsManagement.isIdTaken id then + do! + let warningMsg : FormattableString = $"Subscriber for Id = '{id}' already exists" + logger.LogWarning (String.Format (warningMsg.Format, "id"), id) + socket + |> tryToGracefullyCloseSocket + sendGate + cancellationToken + (enum CustomWebSocketStatus.SubscriberAlreadyExists, warningMsg.ToString ()) + else + let variables = query.Variables |> Skippable.toValueOption + let getInputContext () = httpContext.RequestServices.GetRequiredService() + let! planExecutionResult = + let root = options.RootFactory httpContext + options.SchemaExecutor.AsyncExecute (query.Query, getInputContext, root, ?variables = variables) + do! planExecutionResult |> applyPlanExecutionResult id socket + with ex -> + logger.LogError (ex, "Unexpected error during subscription with id '{id}'", id) + do! sendMsg (Error (id, [ GQLProblemDetails.Create UnexpectedObservableErrorMessage ])) + | ClientComplete id -> + "ClientComplete" |> logMsgWithIdReceived id + subscriptions + |> GraphQLSubscriptionsManagement.removeSubscription (id) + logger.LogTrace "Leaving the 'graphql-ws' connection loop..." + do! + socket + |> tryToGracefullyCloseSocketWithDefaultBehavior sendGate cancellationToken + with ex -> + logger.LogError (ex, "Cannot handle a message; dropping a websocket connection") + // At this point, only something really weird must have happened. + // In order to avoid faulty state scenarios and unimagined damages, + // just close the socket without further ado. + do! + socket + |> tryToGracefullyCloseSocketWithDefaultBehavior sendGate cancellationToken + finally + subscriptions + |> GraphQLSubscriptionsManagement.removeAllSubscriptions } // <-------- // <-- Main // <-------- - let waitForConnectionInitAndRespondToClient (socket : WebSocket) : TaskResult = task { + let waitForConnectionInitAndRespondToClient + (sendGate : SemaphoreSlim) + (cancellationToken : CancellationToken) + (socket : WebSocket) + : TaskResult = task { let timerTokenSource = new CancellationTokenSource () timerTokenSource.CancelAfter connectionInitTimeout let detonationRegistration = timerTokenSource.Token.Register (fun _ -> (socket - |> tryToGracefullyCloseSocket (enum CustomWebSocketStatus.ConnectionTimeout, "Connection initialization timeout")) - .Wait ()) + |> tryToGracefullyCloseSocket + sendGate + cancellationToken + (enum CustomWebSocketStatus.ConnectionTimeout, "Connection initialization timeout")) + .Wait()) let! connectionInitSucceeded = - TaskResult.Run ( + TaskResult.Run( (fun _ -> task { logger.LogDebug ($"Waiting for {nameof ConnectionInit}...") let! receivedMessage = receiveMessageViaSocket CancellationToken.None serializerOptions socket @@ -407,20 +557,22 @@ type GraphQLWebSocketMiddleware<'Root> detonationRegistration.Unregister () |> ignore do! ConnectionAck - |> sendMessageViaSocket serializerOptions socket + |> sendMessageViaSocket sendGate serializerOptions socket return true | Ok (ValueSome (Subscribe _)) -> do! socket - |> tryToGracefullyCloseSocket (enum CustomWebSocketStatus.Unauthorized, "Unauthorized") + |> tryToGracefullyCloseSocket sendGate cancellationToken (enum CustomWebSocketStatus.Unauthorized, "Unauthorized") return false | Result.Error (InvalidMessage (code, explanation)) -> do! socket - |> tryToGracefullyCloseSocket (enum code, explanation) + |> tryToGracefullyCloseSocket sendGate cancellationToken (enum code, explanation) return false | _ -> - do! socket |> tryToGracefullyCloseSocketWithDefaultBehavior + do! + socket + |> tryToGracefullyCloseSocketWithDefaultBehavior sendGate cancellationToken return false }), timerTokenSource.Token @@ -438,18 +590,25 @@ type GraphQLWebSocketMiddleware<'Root> if ctx.WebSockets.IsWebSocketRequest then task { use! socket = ctx.WebSockets.AcceptWebSocketAsync ("graphql-transport-ws") - let! connectionInitResult = socket |> waitForConnectionInitAndRespondToClient + let sendGate = new SemaphoreSlim (1, 1) + use connectionLifetimeCancellationTokenSource = + CancellationTokenSource.CreateLinkedTokenSource (ctx.RequestAborted, applicationLifetime.ApplicationStopping) + let connectionLifetimeCancellationToken = connectionLifetimeCancellationTokenSource.Token + let! connectionInitResult = + socket + |> waitForConnectionInitAndRespondToClient sendGate connectionLifetimeCancellationToken match connectionInitResult with | Result.Error errMsg -> logger.LogWarning errMsg | Ok _ -> - let longRunningCancellationToken = - (CancellationTokenSource - .CreateLinkedTokenSource(ctx.RequestAborted, applicationLifetime.ApplicationStopping) - .Token) - longRunningCancellationToken.Register (fun _ -> (socket |> tryToGracefullyCloseSocketWithDefaultBehavior).Wait ()) + connectionLifetimeCancellationToken.Register (fun _ -> + (socket + |> tryToGracefullyCloseSocketWithDefaultBehavior sendGate connectionLifetimeCancellationToken) + .Wait()) |> ignore try - do! socket |> handleMessages longRunningCancellationToken ctx + do! + socket + |> handleMessages sendGate connectionLifetimeCancellationToken ctx with ex -> logger.LogError (ex, "Cannot handle WebSocket message.") } @@ -458,5 +617,6 @@ type GraphQLWebSocketMiddleware<'Root> title = "WebSocket connection expected.", detail = $"'{options.WebsocketOptions.EndpointUrl}' endpoint only accepts WebSocket connections.", statusCode = StatusCodes.Status400BadRequest - ) :> IResult + ) + :> IResult |> _.ExecuteAsync(ctx) 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 270d1ea22..3bd336b2c 100644 --- a/src/FSharp.Data.GraphQL.Shared/FSharp.Data.GraphQL.Shared.fsproj +++ b/src/FSharp.Data.GraphQL.Shared/FSharp.Data.GraphQL.Shared.fsproj @@ -15,6 +15,7 @@ + diff --git a/src/FSharp.Data.GraphQL.Shared/Helpers/ObjAndStructConversions.fs b/src/FSharp.Data.GraphQL.Shared/Helpers/ObjAndStructConversions.fs index b68d5a9d4..b2eeaec9e 100644 --- a/src/FSharp.Data.GraphQL.Shared/Helpers/ObjAndStructConversions.fs +++ b/src/FSharp.Data.GraphQL.Shared/Helpers/ObjAndStructConversions.fs @@ -1,6 +1,5 @@ namespace rec FSharp.Data.GraphQL -open System.Linq open System.Collections.Generic open FsToolkit.ErrorHandling @@ -27,26 +26,12 @@ module internal ValueTuple = [] module Seq = - let vchoose mapping seq = - seq - |> Seq.map mapping - |> Seq.where ValueOption.isSome - |> Seq.map ValueOption.get - - let vtryFind predicate seq = - seq - |> Seq.where predicate - |> Seq.map ValueSome - |> _.FirstOrDefault() - let vtryHead (source : 'T seq) = use enumerator = source.GetEnumerator () if not (enumerator.MoveNext ()) then ValueNone else - match enumerator.Current with - | null -> ValueNone - | head -> ValueSome head + ValueSome enumerator.Current let vtryLast (source : 'T seq) = use enumerator = source.GetEnumerator () @@ -56,19 +41,59 @@ module Seq = let mutable last = enumerator.Current while enumerator.MoveNext () do last <- enumerator.Current - match last with - | null -> ValueNone - | last -> ValueSome last + ValueSome last + + let vchoose mapping seq = + seq + |> Seq.map mapping + |> Seq.where ValueOption.isSome + |> Seq.map ValueOption.get + + let vtryFind predicate (source : 'T seq) = source |> Seq.where predicate |> Seq.vtryHead + + let vtryItem index (source : 'T seq) = + if index < 0 then + ValueNone + else + use enumerator = source.GetEnumerator () + let mutable currentIndex = 0 + let mutable result = ValueNone + let mutable found = false + + while not found && enumerator.MoveNext () do + if currentIndex = index then + result <- ValueSome enumerator.Current + found <- true + else + currentIndex <- currentIndex + 1 + + result module internal List = - let vchoose mapping list = list |> Seq.ofList |> Seq.vchoose mapping |> Seq.toList + let vchoose mapping list = list |> Seq.vchoose mapping |> Seq.toList + + let vtryFind predicate list = list |> Seq.where predicate |> Seq.vtryHead + + let vtryItem index list = + let rec loop currentIndex list = + match currentIndex, list with + | _, [] -> ValueNone + | 0, head :: _ -> ValueSome head + | currentIndex, _ :: tail when currentIndex > 0 -> loop (currentIndex - 1) tail + | _ -> ValueNone - let vtryFind predicate list = list |> Seq.ofList |> Seq.vtryFind predicate + loop index list module internal Array = - let vchoose mapping array = array |> Seq.vchoose mapping |> Array.ofSeq + let vchoose mapping array = array |> Seq.vchoose mapping |> Seq.toArray + + let vtryItem index (array : 'T array) = + if index < 0 || index >= array.Length then + ValueNone + else + ValueSome array[index] module internal Map = diff --git a/src/FSharp.Data.GraphQL.Shared/WebSockets.fs b/src/FSharp.Data.GraphQL.Shared/WebSockets.fs index 923edb7b7..bd27bfdd7 100644 --- a/src/FSharp.Data.GraphQL.Shared/WebSockets.fs +++ b/src/FSharp.Data.GraphQL.Shared/WebSockets.fs @@ -27,7 +27,7 @@ type RawMessage = { Id : string voption; Type : string; Payload : JsonDocument v 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 + Data : obj voption Skippable /// Errors raised while producing the payload. Errors : GQLProblemDetails list /// Path of a deferred or streamed value inside the initial result. @@ -37,19 +37,29 @@ type SubscriptionExecutionResult = { } with /// Creates a payload of a complete execution result. - static member Create (data : Output, errors : GQLProblemDetails list) = { - Data = Include (box data) + static member Create (data : Output | null, errors : GQLProblemDetails list) = { + Data = + Include ( + Option.ofObj data + |> ValueOption.ofOption + |> ValueOption.map box + ) 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 } + static member CreateErrors (errors : GQLProblemDetails list) = { Data = Include ValueNone; 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) + static member CreateInitial (data : Output | null, errors : GQLProblemDetails list) = { + Data = + Include ( + Option.ofObj data + |> ValueOption.ofOption + |> ValueOption.map box + ) Errors = errors Path = Skip HasNext = Include true @@ -60,7 +70,7 @@ type SubscriptionExecutionResult = { /// More payloads may follow, so is . /// static member CreateIncremental (data : objnull, errors : GQLProblemDetails list, path : FieldPath) = { - Data = Include data + Data = Include (data |> ValueOption.ofObj) Errors = errors Path = Include path HasNext = Include true diff --git a/tests/FSharp.Data.GraphQL.Tests/AspNetCore/SerializationTests.fs b/tests/FSharp.Data.GraphQL.Tests/AspNetCore/SerializationTests.fs index 2c1316c99..490e481a6 100644 --- a/tests/FSharp.Data.GraphQL.Tests/AspNetCore/SerializationTests.fs +++ b/tests/FSharp.Data.GraphQL.Tests/AspNetCore/SerializationTests.fs @@ -1,18 +1,25 @@ module FSharp.Data.GraphQL.Tests.AspNetCore.SerializationTests -open Xunit +open System +open System.Collections.Concurrent +open System.Collections.Generic open System.Text.Json +open System.Text.Json.Serialization + +open Xunit + open FSharp.Data.GraphQL.Ast open FSharp.Data.GraphQL.Shared +open FSharp.Data.GraphQL.Server.AspNetCore.GraphQLSubscriptionsManagement +open FSharp.Data.GraphQL.Server.AspNetCore.ObservableErrorHandling open FSharp.Data.GraphQL.Shared.WebSockets -open System.Text.Json.Serialization [] let ``Deserializes ConnectionInit correctly`` () = let input = "{\"type\":\"connection_init\"}" - let result = JsonSerializer.Deserialize (input, serializerOptions) + let result = JsonSerializer.Deserialize(input, serializerOptions) match result with | ConnectionInit ValueNone -> () // <-- expected @@ -23,7 +30,7 @@ let ``Deserializes ConnectionInit with payload correctly`` () = let input = "{\"type\":\"connection_init\", \"payload\":\"hello\"}" - let result = JsonSerializer.Deserialize (input, serializerOptions) + let result = JsonSerializer.Deserialize(input, serializerOptions) match result with | ConnectionInit _ -> () // <-- expected @@ -34,7 +41,7 @@ let ``Deserializes ClientPing correctly`` () = let input = "{\"type\":\"ping\"}" - let result = JsonSerializer.Deserialize (input, serializerOptions) + let result = JsonSerializer.Deserialize(input, serializerOptions) match result with | ClientPing ValueNone -> () // <-- expected @@ -45,7 +52,7 @@ let ``Deserializes ClientPing with payload correctly`` () = let input = "{\"type\":\"ping\", \"payload\":\"ping!\"}" - let result = JsonSerializer.Deserialize (input, serializerOptions) + let result = JsonSerializer.Deserialize(input, serializerOptions) match result with | ClientPing _ -> () // <-- expected @@ -56,7 +63,7 @@ let ``Deserializes ClientPong correctly`` () = let input = "{\"type\":\"pong\"}" - let result = JsonSerializer.Deserialize (input, serializerOptions) + let result = JsonSerializer.Deserialize(input, serializerOptions) match result with | ClientPong ValueNone -> () // <-- expected @@ -67,7 +74,7 @@ let ``Deserializes ClientPong with payload correctly`` () = let input = "{\"type\":\"pong\", \"payload\": \"pong!\"}" - let result = JsonSerializer.Deserialize (input, serializerOptions) + let result = JsonSerializer.Deserialize(input, serializerOptions) match result with | ClientPong _ -> () // <-- expected @@ -78,7 +85,7 @@ let ``Deserializes ClientComplete correctly`` () = let input = "{\"id\": \"65fca2b5-f149-4a70-a055-5123dea4628f\", \"type\":\"complete\"}" - let result = JsonSerializer.Deserialize (input, serializerOptions) + let result = JsonSerializer.Deserialize(input, serializerOptions) match result with | ClientComplete id -> Assert.Equal ("65fca2b5-f149-4a70-a055-5123dea4628f", id) @@ -97,7 +104,7 @@ let ``Deserializes client subscription correctly`` () = } """ - let result = JsonSerializer.Deserialize (input, serializerOptions) + let result = JsonSerializer.Deserialize(input, serializerOptions) match result with | Subscribe (id, payload) -> @@ -110,7 +117,11 @@ let ``Deserializes client subscription correctly`` () = open FSharp.Data.GraphQL let private serializePayload (payload : SubscriptionExecutionResult) = - let message : RawServerMessage = { Id = ValueSome "1"; Type = "next"; Payload = ValueSome (ExecutionResult payload) } + let message : RawServerMessage = { + Id = ValueSome "1" + Type = "next" + Payload = ValueSome (ExecutionResult payload) + } JsonSerializer.Serialize (message, serializerOptions) let private hasProperty (name : string) (element : JsonElement) = @@ -119,65 +130,162 @@ let private hasProperty (name : string) (element : JsonElement) = [] let ``Serializes incremental payload with path and hasNext`` () = - let json = serializePayload (SubscriptionExecutionResult.CreateIncremental (box [| box 1 |], [], [ box "numbers"; box 0 ])) + 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 ()) + 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}") + 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 (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" ], [])) + 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.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" ]) ]) + 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 ()) + 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 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 ()) + 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 ()) + 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 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 ()) + Assert.Equal ("pong", root.GetProperty("type").GetString()) + Assert.Equal ("pong!", root.GetProperty("payload").GetString()) + +[] +let ``Observable error details sanitize non-GraphQL exception messages`` () = + let actual = problemDetailsOfObservableError (Exception "sensitive backend failure") + let error = Assert.Single actual + Assert.Equal (UnexpectedObservableErrorMessage, error.Message) + +[] +let ``Observable error details preserve GraphQL-facing messages inside aggregates`` () = + let actual = + AggregateException [| Exception "sensitive backend failure"; GQLMessageException "Visible to client" |] + |> problemDetailsOfObservableError + |> List.map _.Message + + Assert.Contains (UnexpectedObservableErrorMessage, actual) + Assert.Contains ("Visible to client", actual) + Assert.DoesNotContain ("sensitive backend failure", actual) + +[] +let ``Observable error details fall back to the generic message for empty aggregates`` () = + let actual = problemDetailsOfObservableError (AggregateException ()) + let error = Assert.Single actual + Assert.Equal (UnexpectedObservableErrorMessage, error.Message) + +[] +let ``Observable error details do not duplicate repeated aggregate errors`` () = + let actual = + AggregateException [| + GQLMessageException ("Visible to client", Dictionary(dict [ "a", box 1; "b", box 2 ])) :> exn + GQLMessageException ("Visible to client", Dictionary(dict [ "b", box 2; "a", box 1 ])) :> exn + |] + |> problemDetailsOfObservableError + + let error = Assert.Single actual + Assert.Equal ("Visible to client", error.Message) + +[] +let ``Request error sanitization replaces backend exception messages`` () = + let actual = + sanitizeRequestError (GQLProblemDetails.Create ("sensitive backend failure", Exception "sensitive backend failure")) + Assert.Equal (UnexpectedObservableErrorMessage, actual.Message) + +[] +let ``Request error sanitization preserves GraphQL-facing errors`` () = + let expected = GQLProblemDetails.OfError (GQLMessageException "Visible to client") + let actual = sanitizeRequestError expected + Assert.Equal (expected, actual) + +type private TrackingSubscription (onDispose : unit -> unit) = + interface IDisposable with + member _.Dispose () = onDispose () + +[] +let ``Removing all subscriptions attempts every disposal before raising aggregate failure`` () = + let disposedIds = ConcurrentQueue () + let unsubscribedIds = ConcurrentQueue () + let subscriptions = + Dictionary() :> SubscriptionsDict + + let createSubscription id shouldThrow = + let subscription = + new TrackingSubscription (fun () -> + disposedIds.Enqueue id + + if shouldThrow then + raise (InvalidOperationException $"Dispose failed for {id}")) + + let onUnsubscribe removedId = + unsubscribedIds.Enqueue removedId + + if shouldThrow then + raise (InvalidOperationException $"Unsubscribe failed for {removedId}") + + id, (subscription :> SubscriptionUnsubscriber), onUnsubscribe + + subscriptions + |> addSubscription (createSubscription "first" true) + subscriptions + |> addSubscription (createSubscription "second" false) + + let error = Assert.Throws(fun () -> subscriptions |> removeAllSubscriptions) + + Assert.False (subscriptions.ContainsKey "first") + Assert.False (subscriptions.ContainsKey "second") + Assert.Equal(set [ "first"; "second" ], set disposedIds) + Assert.Equal(set [ "first"; "second" ], set unsubscribedIds) + Assert.Single error.InnerExceptions