Can't return error after returning channel in subscription

[figboy9]

What happened?

Can't return an error after returning a channel. For example, errors in redis-pubsub , gRPC client streaming, etc.

What did you expect?

I want to return an error after returning a channel.

Minimal graphql.schema and models to reproduce

type Subscription {
  messageAdded: Message!
}

type Message {
  id: ID!
  body: String!
}

func (r *subscriptionResolver) MessageAdded(ctx context.Context) (<-chan *model.Message, error) {
    ch := make(chan *model.Message)
    go func() {
        // some kind of block processing (e.g.: gRPC client streaming)
        stream, err := gRPCClientStreamRequest(ctx)
        if err != nil {
            log.Println(err) // i want to return this error
        }
        for {
            m, err := stream.Recv()
            if err == io.EOF {
                return
            }
            if err != nil {
                log.Println(err) // i want to return this error
            }
            ch <- m
        }
    }()

    return ch, nil
}

versions

• gqlgen version: 0.11.3
• go version: 1.13
• dep or go modules: go modules

[maaft]

Did you find out how to return the error? I'm currently looking for the same thing.

[carldunham]

Why is this closed? Is there a solution for this?

[xiehuc]

i think there a only one way

add error field to Message

type Message { id: ID! body: String! error: String }

[zhixinwen]

Some client can retry on error automatically, so custom error does not work for those clients.

Is it possible to use graphql.AddError or some other similar methods to add an error to context, and process the error https://github.com/99designs/gqlgen/blob/master/graphql/handler/transport/websocket.go#L365.

it would break some code encapsulation, but I don’t see a good way to get the error without changing resolver interface.

[zhixinwen]

@StevenACoffman could you please reopen the issue?

[noltrifork]

Bump.

[StevenACoffman]

Please verify whether this is still an issue, or if #2506 resolved it If it is still an issue, then PRs to address it are welcome.

The project is maintained by the community's contributions. Commenting "Bump" to an issue is not actually a helpful contribution.

[wiegell]

I experience the same problem and have examined it a bit further. I think @zhixinwen hit the main problem - grahpl.AddError() does not propagate through the system when set after the resolver has returned the channel.

I've tried reading through the nested contexts by making a test middleware with both gqlHandler.AroundOperations() and gqlHandler.AroundResponses(). Only the ladder actually has a response context (or result_context as the string enum evaluates to). But that response context does not hold the error set with the grahpl.AddError() as would be expected! Instead len(errs) = 0. Subscription errors do stick on the other hand (transport.AddSubscriptionError()), but they are not suited for the task since they accumulate over time and are not reset between each emit to the resolver return channel.

This lib has quite a few levels of contexts going on and it's hard to pinpoint where the error is lost / the response context is wrongly refreshed - @StevenACoffman do you have a clue?

[wiegell]

So i've tried checking out 11c3a4d, where #2506 should be implemented. That does not return type error on the websocket in the browser when adding transport.AddSubscriptionError() as would be expected. I've added the error pretty much like the code snippet shown here

I've tried to modify websocket.go by replacing the for loop at line 384 (0.17.36: sha 2d8673a691deffe6ddd1f4d5f013a52dc91aef91) to try and get the functionality we want:

  ForLoop:
    for {
      response := responses(ctx)
      errs := getSubscriptionError(ctx)
      switch {
      case len(errs) != 0:
        c.sendError(msg.id, errs...)
      case response == nil:
        break ForLoop
      default:
        c.sendResponse(msg.id, response)
      }

    }

I then add an error on each emit to the resolver return channel with transport.AddSubscriptionError(). This does return type error in the browser, but they accumulate in the response and you cannot emit without the error anymore.

So... either we need to flush the subscription errors at an appropriate place or make the correct response context available when adding errors with grahpl.AddError() after the connection is established.

[vlad-tokarev]

According to the GraphQL specification, subscription streams an "Execution result" which comprises two fields: data and errors. Not all of these errors are fatal, meaning the server can continue streaming.

However, in the current implementation, the subscription resolver is designed only with a channel for data, excluding errors. The transport processes errors only after the subscription completes, which is when the resolver closes the data channel. If errors are present, they are sent as a payload of the graphql-transport-ws message of the Error type. This type of message signifies a "fatal" subscription error, implying that the subscription ended due to an error and no further messages are anticipated.

It appears there are two distinct mechanisms:

1. One that propagates errors in the same way as a regular query resolver (data+errors).
2. Another that notifies the client when a subscription has been terminated due to an error. Currently, gqlgen only supports the second mechanism.

While this isn't a proposal, a potential solution could be to provide an API in the generated subscription resolver that allows for both data and errors to be pushed into channels. This could either be in the same channel or separate ones.

[StevenACoffman]

@telemenar @UnAfraid @zhixinwen Any thoughts on how we might resolve this? I don't personally use websockets at work, so I would really appreciate a PR from someone with hands-on experience.

[zhixinwen]

I don’t have access to a real websocket project now. A quick fix would be whenever we send errors, we remove them from the context value.

[UnAfraid]

I do not use ws at work yet, but i do have few personal projects in which i do.

The current subscription api only gives you channel with the type to send back, we don't have a way to put error in there. We can probably add new configuration option to switch generated subscription resolvers to (<-chan transport.SubscriptionResult[Type], error) which is either the actual type or an error. And for fatal errors we can just close the channel

[comerc]

// AddSubscriptionError is used to let websocket return an error message after subscription resolver returns a channel.
// for example:
//
//  func (r *subscriptionResolver) Method(ctx context.Context) (<-chan *model.Message, error) {
//      ch := make(chan *model.Message)
//      go func() {
//       defer func() {
//              close(ch)
//       }
//          // some kind of block processing (e.g.: gRPC client streaming)
//          stream, err := gRPCClientStreamRequest(ctx)
//          if err != nil {
//                 transport.AddSubscriptionError(ctx, err)
//              return // must return and close channel so websocket can send error back
//       }
//          for {
//              m, err := stream.Recv()
//              if err == io.EOF {
//                  return
//              }
//              if err != nil {
//                 transport.AddSubscriptionError(ctx, err)
//              return // must return and close channel so websocket can send error back
//              }
//              ch <- m
//          }
//      }()
//
//      return ch, nil
//  }
//
// see https://github.com/99designs/gqlgen/pull/2506 for more details

[comerc]

Pseudo code in directive:

func Auth(ctx context.Context, obj interface{}, next graphql.Resolver) (interface{}, error) {
    payload := jwt.GetPayload(ctx)
    if payload.IsExpired() {
        return nil, errors.New("JWT was expired - case 1")
    }
    if payload.MemberID == 0 {
        return nil, errors.New("Unauthorised")
    }
    return next(ctx)
}

Result:

{
  "errors": [
    {
      "message": "JWT was expired - case 1",
      "path": [
        "currentTime"
      ]
    }
  ],
  "data": null
}

Pseudo code in resolver:

if jwt.GetPayload(ctx).IsExpired() {
    transport.AddSubscriptionError(ctx, gqlerror.Errorf("JWT was expired - case 2"))
    return
}

Result:

{
  "errors": [
    {
      "message": "JWT was expired - case 2"
    }
  ]
}

How to add path to errors (like case 1)?