Added possibility to propagate subscription error properly

[maoueh]

After a subscription has started and sent some results to the user, from the resolver perspective, it's not right now possible to propagate error correctly to the user using the top-level object.

Since a chan <type> type is required for the top-level subscription resolver, it cannot convey any error information. By adding a SubscriptionError interface that top-level resolver's object can implement, the library upon reception of the result from the channel can check if it implements the SubscriptionError interface, if it's the case and the interface's method returned a non-nil error, the subscription is terminated, the channel is closed and the error is propagated to the user.

Extracted from #317

Design

Right now, our use case as always been "terminal" error, i.e. that once we set SubscriptionError to be non-nil, we always terminate the stream.

While extracting the code to submit the PR, I realized that in some situations, it could be desired that this error is treated as a transient error in which case, it should behaves like if a sub-field resolver would have a second return value of type error. Moreover, someone could also want to print the "top-level" error as well as the rest of the resolvable fields of this top-level object.

So I think it would be beneficial to handle terminal & non-terminal errors here. One suggestion I could make would be to change the interface SubscriptionError to become:

type SubscriptionError interface {
  SubscriptionError() (err error, terminal bool) // Or inverted with `transient bool` instead
}

And in the subscribe implementation where we check for the SubscriptionError interface, if the terminal return value is true, the code uses the path implemented in this PR.

Otherwise if it's false, then we keep the error, resolve the fields and add the errors to the other ones if any.

Thoughts?

[maoueh]

I'll wait for #417 to be merge before resolving conflicts

[pavelnikolov]

I am not convinced that this should be part of the library. You can have your errors treated as data and have them in your schema resolver as fields. For example you might return an optional error field. Which in turn might have a subfield isTerminal or something like this. I'm not 100% convinced that I want this to be part of the library. What exactly in the current implementation is limiting you? Why can't you terminate the subscription with the current implementation?

[maoueh]

Then my goal will be to convince you :)

Don't stick on terminal error and extra fields, I should not have talked about this, it's not necessary for the core element this PR does.

The main goal here is to report "streaming" errors for each subscription's message through the standard GraphQL format so via the errors field.

Your suggestion has major drawbacks in my opinion:

• It brings another error model that GraphQL might not be directly aware of expecting error(s) to be in the errors field. They need to be discovered and queried otherwise they are not "received"
• It requires to clutter the schema definition with various "error" field at a multitude of locations
• It cannot work properly with union value returned at top-level, requiring useless schema "workaround" on certain circumstances (see my StarWars subscription that I present below)

With the proposed PR, errors can be appended correctly to the errors field for subscription message with little effort and be in the format expected by developers used to GraphQL.

StarWars Subscription Example

I re-used the StarWars example but added a subscription on the search so each search result are streamed.

schema {
  ...
  subscription: Subscription
}

type Subscription {
  search(text: String!): SearchResult!
}

With this PR

With this PR, no schema change is required, only the searchResultResolver needs to be modified to implement the SubscribeError method.

type searchResultResolver struct {
	result interface{}
	err error
}

...

func (t *searchResultResolver) SubscriptionError() error {
	return t.err
}

And here the message received when there is an error while subscribing:

{"data":{"search":{"kind":"Human","id":"1000","name":"Luke Skywalker"}}}
{"errors":[{"message":"backend had an error"}]}

Within Schema

There is a big problem here with SearchResult which is an union. It's not possible to add an error field here because what kind should I returned when the search itself encountered an error? I started adding errors to each union's elements, but it does not make sense since it the "search result" itself that has the error.

So I needed in fact to wrap all this in another message that has the union and an error field. Here what messages looked like in this experiment:

{"data":{"search":{"errors":null,"node":{"kind":"Human","id":"1000","name":"Luke Skywalker"}}}}
{"data":{"search":{"errors":[{"message":"backend had an error"}],"node":{"kind":""}}}}

Personally, I find this output really bad, it has the feeling that it worked mostly but it's no the case. I could have tweaked the node field to be optional, but that wouldn't improve it much.

Wrap-up

This is a rather large reply, I wanted to play with your proposed approach to better understand other possibilities. I hope it convinced you that this is a good change for the library. I'm happy to revisit the PR and find alternative, my goal is really to have the output of subscription message like I pasted above

{"data":{"search":{"kind":"Human","id":"1000","name":"Luke Skywalker"}}}
{"errors":[{"message":"backend had an error"}]}

How it looks like in the library, I'm happy to explore other way to achieve the same thing.

Diff (between code with this PR and code within schema)

Here the diff I need to implement the error in schema.

diff --git a/resolver.go b/resolver.go
index 8303217..e2c11f0 100644
--- a/resolver.go
+++ b/resolver.go
@@ -38,17 +38,17 @@ func (r *Resolver) QuerySearch(args struct{ Text string }) []*searchResultResolv
	var l []*searchResultResolver
	for _, h := range humans {
		if strings.Contains(h.Name, args.Text) {
-			l = append(l, &searchResultResolver{&humanResolver{h}, nil})
+			l = append(l, &searchResultResolver{&humanResolver{h}})
		}
	}
	for _, d := range droids {
		if strings.Contains(d.Name, args.Text) {
-			l = append(l, &searchResultResolver{&droidResolver{d}, nil})
+			l = append(l, &searchResultResolver{&droidResolver{d}})
		}
	}
	for _, s := range starships {
		if strings.Contains(s.Name, args.Text) {
-			l = append(l, &searchResultResolver{&starshipResolver{s}, nil})
+			l = append(l, &searchResultResolver{&starshipResolver{s}})
		}
	}
	return l
@@ -85,8 +85,8 @@ func (r *Resolver) QueryStarship(args struct{ ID graphql.ID }) *starshipResolver
	return nil
}

-func (r *Resolver) SubscriptionSearch(ctx context.Context, args struct{ Text string }) (<-chan *searchResultResolver, error) {
-	c := make(chan *searchResultResolver)
+func (r *Resolver) SubscriptionSearch(ctx context.Context, args struct{ Text string }) (<-chan *searchResultMessageResolver, error) {
+	c := make(chan *searchResultMessageResolver)

	go func() {
		defer func() {
@@ -96,7 +96,7 @@ func (r *Resolver) SubscriptionSearch(ctx context.Context, args struct{ Text str
		shouldQuitDueToError := func() bool {
			if ctx.Err() != nil {
				fmt.Println("Context timeout, closing!")
-				c <- &searchResultResolver{nil, ctx.Err()}
+				c <- &searchResultMessageResolver{&searchResultResolver{nil}, []error{ctx.Err()}}
				return true
			}

@@ -109,13 +109,13 @@ func (r *Resolver) SubscriptionSearch(ctx context.Context, args struct{ Text str
			}

			if strings.Contains(h.Name, args.Text) {
-				c <- &searchResultResolver{&humanResolver{h}, nil}
+				c <- &searchResultMessageResolver{&searchResultResolver{&humanResolver{h}}, nil}
			}
		}

		time.Sleep(750 * time.Millisecond)
		if rand.Float64() > 0.80 {
-			c <- &searchResultResolver{nil, errors.New("backend had an error")}
+			c <- &searchResultMessageResolver{&searchResultResolver{nil}, []error{errors.New("backend had an error")}}
			return
		}

@@ -125,13 +125,13 @@ func (r *Resolver) SubscriptionSearch(ctx context.Context, args struct{ Text str
			}

			if strings.Contains(d.Name, args.Text) {
-				c <- &searchResultResolver{&droidResolver{d}, nil}
+				c <- &searchResultMessageResolver{&searchResultResolver{&droidResolver{d}}, nil}
			}
		}

		time.Sleep(750 * time.Millisecond)
		if rand.Float64() > 0.80 {
-			c <- &searchResultResolver{nil, errors.New("backend had an error")}
+			c <- &searchResultMessageResolver{&searchResultResolver{nil}, []error{errors.New("backend had an error")}}
			return
		}

@@ -141,7 +141,7 @@ func (r *Resolver) SubscriptionSearch(ctx context.Context, args struct{ Text str
			}

			if strings.Contains(s.Name, args.Text) {
-				c <- &searchResultResolver{&starshipResolver{s}, nil}
+				c <- &searchResultMessageResolver{&searchResultResolver{&starshipResolver{s}}, nil}
			}
		}
	}()
@@ -425,9 +425,37 @@ func (r *starshipResolver) Length(args struct{ Unit string }) float64 {
	return convertLength(r.s.Length, args.Unit)
}

+type errorDef struct {
+	message string
+}
+
+func (e *errorDef) Message() string {
+	return e.message
+}
+
+type searchResultMessageResolver struct {
+	node   *searchResultResolver
+	errors []error
+}
+
+func (r *searchResultMessageResolver) Node() *searchResultResolver {
+	return r.node
+}
+
+func (r *searchResultMessageResolver) Errors() *[]*errorDef {
+	if len(r.errors) == 0 {
+		return nil
+	}
+
+	out := make([]*errorDef, len(r.errors))
+	for i, err := range r.errors {
+		out[i] = &errorDef{err.Error()}
+	}
+	return &out
+}
+
type searchResultResolver struct {
	result interface{}
-	err    error
}

func (r *searchResultResolver) ToHuman() (*humanResolver, bool) {
@@ -445,10 +473,6 @@ func (r *searchResultResolver) ToStarship() (*starshipResolver, bool) {
	return res, ok
}

-func (t *searchResultResolver) SubscriptionError() error {
-	return t.err
-}
-
func convertLength(meters float64, unit string) float64 {
	switch unit {
	case "METER":
diff --git a/schema.graphql b/schema.graphql
index b692677..401b5f7 100644
--- a/schema.graphql
+++ b/schema.graphql
@@ -16,7 +16,7 @@ type Query {
}

type Subscription {
-  search(text: String!): SearchResult!
+  search(text: String!): SearchResultMessage!
}

# The mutation type, represents all updates we can make to our data
@@ -56,6 +56,10 @@ enum LengthUnit {
  FOOT
}

+type Error {
+  message: String!
+}
+
# A humanoid creature from the Star Wars universe
type Human implements Character {
  # The ID of the human
@@ -144,4 +148,9 @@ type Starship {
  length(unit: LengthUnit = METER): Float!
}

+type SearchResultMessage {
+  node: SearchResult!
+  errors: [Error]
+}
+
union SearchResult = Human | Droid | Starship
diff --git a/subscribe_ok.graphql b/subscribe_ok.graphql
index f6e3b5e..07bc389 100644
--- a/subscribe_ok.graphql
+++ b/subscribe_ok.graphql
@@ -1,20 +1,22 @@
subscription {
  search(text: "Luke") {
-    kind: __typename
-
-    ... on Human {
-      id
-      name
+    errors {
+      message
    }
-
-    ... on Droid {
-      id
-      name
-    }
-
-    ... on Starship {
-      id
-      name
+    node {
+      kind: __typename
+      ... on Human {
+        id
+        name
+      }
+      ... on Droid {
+        id
+        name
+      }
+      ... on Starship {
+        id
+        name
+      }
    }
  }
}

[pavelnikolov]

Thank you for the detailed explanation and for your patience on this but I still don't understand how this PR improves the library. Currently the library satisfies the GraphQL spec and allows messages to contain both data and errors. Could you give me a unit test or an executable example that that I can run which demonstrates the issue? I believe that this unit test demonstrates the ability to propagate subscription errors to the user.