Test: Missing subscription field should not be an "internal error"

[leebyron]

Added a test to illustrate a broken issue and potentially underspecified in spec:

If a source event stream emits an error instead of an event, then that error is passing up through the whole stack and throwing at the consumer of the response event stream. That's very likely not what we want. I have a proposal in this test case for what should happen in that case, similar to what would happen if an error occurred during during the second step of executing an event from the source stream.

cc @Urigo @robzhu

[leebyron]

Per the spec, @robzhu my thought is that in MapSourceToResponseEvent we may need a step which determines if each event represents an error or data, and if an error then describes what to do.

[robzhu]

Let me make sure I'm understanding the issue you're highlighting. In the first test case, you expect a response consistent with the an error encountered during query execution. In the second case, we don't actually get to the execution step because the source stream has thrown an error, and we don't actually have a root value for the execution step?

[robzhu]

If a source event stream emits an error instead of an event, then that error is passing up through the whole stack and throwing at the consumer of the response event stream. That's very likely not what we want.

Why not? I imagine the consumer of the response stream is in a better position to determine what to do with the error.

[Urigo]

@leebyron @robzhu

Looking at what you wrote, I completely agree and this is the behavior I would expect from the GraphQL execution engine, because this event source errors should be treated just like any other GraphQL error, and handle it the same way, just like the GraphQL engine does for queries.
If an error occurred during the subscriptions's payload execution - GraphQL should reflect it as a regular error (using errors: [ ... ]).

With the current implementation, if an error is emitted from the AsyncIterable it will also cause the listener to stop (if using forAwaitOf - it will call the catch or the return Promise, causing it to stop the iteration, right?).

Currently, we are using catch of forAwaitOf to catch those errors, but it might also stop the iteration over the AsyncIterator.

I think that in order to catch this error correctly, we need to extend mapAsyncIterator to accept an error mapping function, something like that:

  return mapAsyncIterator(
    subscription.subscription,
    payload => execute(
      schema,
      document,
      payload,
      contextValue,
      variableValues,
      operationName,
      fieldResolver
    ),
    (error: Error) => {
      return {
        value: {
          errors: [
            new GraphQLError(error.message, fieldNodes), // What is the correct way to build this error?
          ],
        },
        done: false,
      };
    }
  );

And to use this error handler inside mapAsyncIterator, with catch:

    next() {
      return iterator.next()
        .then(mapResult)
        .catch(mapError);
    },

This way we make sure that the error is caught and handled, and emitted as another value of the AsyncIterable, instead of crashing the whole iteration.

The only thing is that we need somehow to expose the fieldNodes (and maybe also the path?) from createSourceEventStream, because the error should be formatted correctly with the location and path and I'm not sure how to do it correctly.

[leebyron]

Why not? I imagine the consumer of the response stream is in a better position to determine what to do with the error.

Possibly! I think behavior here is definitely up for discussion. What would a consumer do with this error? Say concretely for a case where a Redis stream creates an error and a websockets consumer encounters it?

My thought was that most of the time a consumer like websockets would prefer to not know about errors and just treat it like the last payload in the sequence (side note: does the websocket subscriptions impl have a clear way of communicating to a client that a sequence has completed?)

@Urigo - I don't think we need to derive that information from createSourceEventStream since we have the original document, but I agree with you that our map operation will need this second mapping funtion

[Urigo]

@leebyron
In subscriptions-transport-ws protocol we have a message for GQL_COMPLETE that indicates that the stream is done.
And I agree that createSourceEventStream should remain as, but then we need to resolve the document again and extract the location and path when error occurred?

[robzhu]

@Urigo

The only thing is that we need somehow to expose the fieldNodes (and maybe also the path?) from createSourceEventStream, because the error should be formatted correctly with the location and path and I'm not sure how to do it correctly.

We won't have location and path handy unless we attempt to resolve the selection, but since the source stream yielded an error instead of an event, we don't have a root value, so resolving might not yield anything useful.

Say concretely for a case where a Redis stream creates an error and a websockets consumer encounters it?

In the Redis/Websocket scenario, I imagine there would be a piece of code that lives between the responseStream and dispatching the message over the client connection. For example:

const subscription = subscribe(...);
try {
  for await (const responseEvent of subscription) {
    const payload = responseEvent.value;
    // don't send the client any errors
    if (!payload.errors) {
      webSocket.send(payload);  
    }
  }
} catch (e) {
  webSocket.send(PROTOCOL_ERROR);
  webSocket.close();
}

If such a piece of code exists, we can side-step the question of how to deal with these sorts of errors in the spec/reference, while providing more flexibility for error handling behavior.

If an asyncIterator throws, does its contract enforce whether it can or cannot throw again?

[leebyron]

If an asyncIterator throws, does its contract enforce whether it can or cannot throw again?

Good question. I don't think so. I think it can catch errors thrown internally:

async function* valueAfterThrow() {
  try {
    throw new Error('shucks');
  } finally {
    yield 'value';
  }
}

But the finally block is executed before the error is thrown, ensuring the error is the last iterated step.

[leebyron]

@robzhu I think your example here is pretty compelling. To make sure I understand, this suggests that beyond GraphQL's yielded errors, the transport is responsible for additional messages "complete" and "error" which are not in the form of graphql responses.

[leebyron]

Perhaps there's just some parity to figure out with execution. Calling the graphql() function results in a Promise which always resolves. Even when there is an error, that error is returned as a resolved promise with a { errors: [...] } payload.

Maybe the difference here is if the error is an error within the confines of GraphQL execution (if the query is syntactically incorrect or invalid, or if execution fails) then it returns in a GraphQL payload, but if the error occurs outside of these then it's a transport issue?

[leebyron]

This reminds me of another missing piece of the public API. Right now we export both execute() and graphql() which is responsible for validating before executing. We should include a single use function which describes how to validate before subscribing. Should we build that into graphql() and have it return Promise | AsyncIterator or perhaps add a new function which can always return AsyncIterator?

[robzhu]

To make sure I understand, this suggests that beyond GraphQL's yielded errors, the transport is responsible for additional messages "complete" and "error" which are not in the form of graphql responses.

Yep, and I would take it further. Before a payload is dispatched to the client, the server must handle four cases for the subscription responseStream:

1. GraphQL executed, valid response.
2. GraphQL executed, error response.
3. responseStream ended.
4. responseStream threw an error.

It should be possible to vary the logic to handle all four of these cases on a per-application basis. For example, in the case of a responseStream error, I may want to have some sort of retry logic, or in the case of a valid response I might want deduping/encoding. Since sourceStream is user-provided, I think the most intuitive behavior is to have errors pass through.

In other words, I would rewrite the second test case to validate that responseStream throws any errors it encounters from sourceStream.

[robzhu]

Good question. I don't think so. I think it can catch errors thrown internally:

My previous response is predicated on this. Basically, I'm wondering if we can treat an AsyncIterator as a state machine, where upon encountering the first error, the iterator is then in the "faulted" state and will neither yield additional events nor throw more errors.

[jamesgorman2]

Have you guys looked at the error handing in RxJava 2? David Karnok put a lot of thinking into handling next/error/complete and what to do about errors that can't be captured. The core description is in the wiki, but there are a bunch of (closed) issues with people talking about what happened when they upgraded and things started breaking (especially in android). Might help spark some more ideas for you.

[leebyron]

My previous response is predicated on this. Basically, I'm wondering if we can treat an AsyncIterator as a state machine, where upon encountering the first error, the iterator is then in the "faulted" state and will neither yield additional events nor throw more errors.

This is a safe assumption for Async Generators. The spec proposal literally describes async generators it via a state machine where return and throw (or any abrupt completion) result in the AG's state changing to "completed" - which makes sense, the goal is to mirror semantics of typical functions, where return/throw end the function's execution.

However Async Iterators are a set which includes Async Generators, and there's no such rule that a rejected Promise must be the last in the sequence. In fact, the spec mentions a few things that should happen (but aren't enforced) and this didn't even make the list (https://tc39.github.io/proposal-async-iteration/#sec-asynciterator-interface)

Here's an example:

async function* clockTick() {...} // yields a value every second

const wtfStream = mapAsyncIterable(clockTick(), time => {
  if (time % 2 === 1) throw new Error('Odd number of seconds, what gives!')
  return 'Even number of seconds, how peaceful'
})

Here, wtfStream will alternate between yielding resolved values and yielding rejected ones. (Perhaps it shouldn't, our implementation of map could close the parent ai after this!)

More evidence for rejecting expecting to be the last in a sequence is for await(let val of ai) loops, which when encountering a rejected item from ai will behave as if it "threw" an error and scope upwards until finding a try {} block. (https://tc39.github.io/proposal-async-iteration/#sec-runtime-semantics-forin-div-ofbodyevaluation-lhs-stmt-iterator-lhskind-labelset step "L")

[leebyron]

Updated with new test cases. @robzhu let me know what you think of this behavior.

These tests still fail, but the fix should be much easier.

[robzhu]

This looks great. Any more thoughts on whether we should accompany this with an update to the spec?

[leebyron]

Coming back to this after GraphQL EU travel :)

Some new thoughts on this - I think both of my previous approaches were correct, and both incorrect. Depends on context of the error.

I think we can split errors into two categories: internal errors and GraphQL user errors. For example, if a user provides a GraphQL document which causes an issue then the error should be returned as described by the spec (matching that of execute()) in a response body with {"errors":}. However internal errors with the source stream should cause the response stream to fail.

For updates to the spec, we should define these kinds of errors and explain what to do with them (like "return an event stream yielding a single event containing response" when appropriate), and perhaps abstractly describe both standard and error completion of event streams with a note about supporting all of these in a transport

[leebyron]

Ok, I just updated this with more tests, utility functions, and some real semantic changes. I'd appreciate a thorough review.

[robzhu]

Thanks for highlighting these error-handling cases. I think is definitely something we need to be more thorough about. I see a few concerns all bunched together under "error handling":

• The source stream returning/throwing errors
• The resolver returning/throwing errors
• Parsing/creation of subscription throwing errors

Perhaps it makes sense to tackle these individually if you think the concerns are separable. Otherwise, we should try and put together a table of expected behavior.

[robzhu]

This doesn't feel right to me. I would expect this to behave the same as when we try to execute an invalid query. Maybe I'm missing something here? How do we get the error in this case?

[leebyron]

Validation will produce an error, however once validated a document should be operable according to the algorithm in the spec.

[leebyron]

Alternatively, we could change the spec to make this an assertion instead

[robzhu]

So we won't actually get to this step because subscription { unknownField } would fail validation and return an informative error?

[leebyron]

In typical validate-then-operate usage, most definitely. Attempting to use an undefined field is a validation error.

However we need to handle these corner cases nonetheless. In the case that validation was thwarted in some way. Perhaps a doc was validated before being persisted, and then a breaking change was made to a schema which caused the doc to become invalid - but future runs of it have skipped validation due to being persisted.

This is the sort of thing that's rare, but needs definition.

[robzhu]

Nit: this is a weird place for this function. Perhaps move it to the top of the file or before the block of tests that use it?

[leebyron]

Ya

[robzhu]

In the doc block above:

If an error is encountered during creation, an Error is thrown.

Why shouldn't we treat a missing fieldDef as an error?

[leebyron]

I think this should mirror the behavior of execution of an undefined field:

await execute(schema, parse('{ unknownField }'))
{ "data": {} }

That is, an event stream from a non existent source is the empty stream.

[leebyron]

All the tests from this PR have since been added or refined in separate commits