fix handling of async query errors

policy/policy_handler/async.go

@@ -4,7 +4,9 @@ import (
 	"context"
 	b64 "encoding/base64"
 	"fmt"
+
 	"github.com/atomist-skills/go-skill/policy/goals"
+	"github.com/atomist-skills/go-skill/policy/types"
 
 	"github.com/atomist-skills/go-skill"
 	"github.com/atomist-skills/go-skill/policy/data"
@@ -72,6 +74,9 @@ func buildAsyncDataSources(multipleQuerySupport bool) dataSourceProvider {
 		}
 
 		metaEdn, err := b64.StdEncoding.DecodeString(req.Event.Context.AsyncQueryResult.Metadata)
+		if err != nil {
+			return nil, fmt.Errorf("failed to decode metadata: %w", err)
+		}
 
 		var metadata data.AsyncResultMetadata
 		err = edn.Unmarshal(metaEdn, &metadata)
@@ -84,6 +89,13 @@ func buildAsyncDataSources(multipleQuerySupport bool) dataSourceProvider {
 		if err != nil {
 			return nil, fmt.Errorf("failed to unmarshal async query result: %w", err)
 		}
+		if len(queryResponse.Errors) > 0 {
+			errorMessage := queryResponse.Errors[0].Message
+			if errorMessage == "An unexpected error has occurred" {
+				return nil, fmt.Errorf("async query contained error: %s", errorMessage).(types.RetryableExecutionError)
+			}
+			return nil, fmt.Errorf("async query contained error: %s", errorMessage)
+		}
 		metadata.AsyncQueryResults[metadata.InFlightQueryName] = queryResponse
 
 		return []data.DataSource{

policy/policy_handler/handler.go

@@ -11,6 +11,7 @@ import (
 	"github.com/atomist-skills/go-skill/policy/data"
 	"github.com/atomist-skills/go-skill/policy/goals"
 	"github.com/atomist-skills/go-skill/policy/storage"
+	"github.com/atomist-skills/go-skill/policy/types"
 	"github.com/atomist-skills/go-skill/util"
 	"olympos.io/encoding/edn"
 )
@@ -94,8 +95,8 @@ func (h EventHandler) handle(ctx context.Context, req skill.RequestContext) skil
 	for _, provider := range h.dataSourceProviders {
 		ds, err := provider(ctx, req, *evaluationMetadata)
 		if err != nil {
-			if err.Error() == "An unexpected error has occurred" {
-				return skill.NewRetryableStatus(fmt.Sprintf("Failed to create data source [%s]", err.Error()))
+			if retryableError, ok := err.(types.RetryableExecutionError); ok {
+				return skill.NewRetryableStatus(fmt.Sprintf("Failed to create data source [%s]", retryableError.Error()))
 			}
 			return skill.NewFailedStatus(fmt.Sprintf("failed to create data source [%s]", err.Error()))
 		}

policy/types/errors.go

@@ -0,0 +1,9 @@
+package types
+
+import "fmt"
+
+type RetryableExecutionError string
+
+func (e RetryableExecutionError) Error() string {
+	return fmt.Sprint(string(e))
+}