graphql: adds transactions to graphql mutations

graphql/admin/schema.go

@@ -112,6 +112,7 @@ func (asr *updateSchemaResolver) Execute(
 		return &dgoapi.Response{Json: b}, err
 	}
 
+	req.CommitNow = true
 	resp, err := asr.baseMutationExecutor.Execute(ctx, req)
 	if err != nil {
 		return nil, err
@@ -126,6 +127,10 @@ func (asr *updateSchemaResolver) Execute(
 	return resp, nil
 }
 
+func (asr *updateSchemaResolver) CommitOrAbort(ctx context.Context, tc *dgoapi.TxnContext) error {
+	return asr.baseMutationExecutor.CommitOrAbort(ctx, tc)
+}
+
 func (gsr *getSchemaResolver) Rewrite(ctx context.Context,
 	gqlQuery schema.Query) (*gql.GraphQuery, error) {
 	gsr.gqlQuery = gqlQuery
@@ -140,6 +145,10 @@ func (gsr *getSchemaResolver) Execute(
 	return &dgoapi.Response{Json: b}, err
 }
 
+func (gsr *getSchemaResolver) CommitOrAbort(ctx context.Context, tc *dgoapi.TxnContext) error {
+	return nil
+}
+
 func doQuery(gql *gqlSchema, field schema.Field) ([]byte, error) {
 
 	var buf bytes.Buffer

graphql/dgraph/execute.go

@@ -57,3 +57,9 @@ func (dg *DgraphEx) Execute(ctx context.Context, req *dgoapi.Request) (*dgoapi.R
 
 	return resp, schema.GQLWrapf(err, "Dgraph execution failed")
 }
+
+// CommitOrAbort is the underlying dgraph implementation for commiting a Dgraph transaction
+func (dg *DgraphEx) CommitOrAbort(ctx context.Context, tc *dgoapi.TxnContext) error {
+	_, err := (&edgraph.Server{}).CommitOrAbort(ctx, tc)
+	return err
+}

graphql/e2e/common/error.go

@@ -268,6 +268,10 @@ func (dg *panicClient) Execute(ctx context.Context, req *dgoapi.Request) (*dgoap
 	return nil, nil
 }
 
+func (dg *panicClient) CommitOrAbort(ctx context.Context, tc *dgoapi.TxnContext) error {
+	return nil
+}
+
 // clientInfoLogin check whether the client info(IP address) is propagated in the request.
 // It mocks Dgraph like panicCatcher.
 func clientInfoLogin(t *testing.T) {

graphql/resolve/mutation.go

@@ -25,6 +25,7 @@ import (
 	"github.com/dgraph-io/dgraph/graphql/dgraph"
 	"github.com/dgraph-io/dgraph/graphql/schema"
 	"github.com/dgraph-io/dgraph/x"
+	"github.com/golang/glog"
 	otrace "go.opencensus.io/trace"
 )
 
@@ -103,6 +104,7 @@ type DgraphExecutor interface {
 	// occurs, that indicates that the execution failed in some way significant enough
 	// way as to not continue processing this mutation or others in the same request.
 	Execute(ctx context.Context, req *dgoapi.Request) (*dgoapi.Response, error)
+	CommitOrAbort(ctx context.Context, tc *dgoapi.TxnContext) error
 }
 
 // An UpsertMutation is the query and mutations needed for a Dgraph upsert.
@@ -185,6 +187,16 @@ func getNumUids(m schema.Mutation, a map[string]string, r map[string]interface{}
 func (mr *dgraphResolver) rewriteAndExecute(
 	ctx context.Context,
 	mutation schema.Mutation) (*Resolved, bool) {
+	var mutResp *dgoapi.Response
+	commit := false
+
+	defer func() {
+		if !commit && mutResp != nil && mutResp.Txn != nil {
+			mutResp.Txn.Aborted = true
+			err := mr.executor.CommitOrAbort(ctx, mutResp.Txn)
+			glog.Errorf("Error occured while aborting transaction: %f", err)
+		}
+	}()
 
 	emptyResult := func(err error) *Resolved {
 		return &Resolved{
@@ -206,7 +218,7 @@ func (mr *dgraphResolver) rewriteAndExecute(
 		Mutations: upsert.Mutations,
 	}
 
-	mutResp, err := mr.executor.Execute(ctx, req)
+	mutResp, err = mr.executor.Execute(ctx, req)
 	if err != nil {
 		gqlErr := schema.GQLWrapLocationf(
 			err, mutation.Location(), "mutation %s failed", mutation.Name())
@@ -231,6 +243,14 @@ func (mr *dgraphResolver) rewriteAndExecute(
 		return emptyResult(errs), resolverFailed
 	}
 
+	err = mr.executor.CommitOrAbort(ctx, mutResp.Txn)
+	if err != nil {
+		return emptyResult(
+				schema.GQLWrapf(err, "mutation failed, couldn't commit transaction")),
+			resolverFailed
+	}
+	commit = true
+
 	qryResp, err := mr.executor.Execute(ctx,
 		&dgoapi.Request{Query: dgraph.AsString(dgQuery), ReadOnly: true})
 	errs = schema.AppendGQLErrs(errs, schema.GQLWrapf(err,

graphql/resolve/resolver.go

@@ -159,11 +159,19 @@ func (aex *adminExecutor) Execute(ctx context.Context, req *dgoapi.Request) (
 	return aex.dg.Execute(ctx, req)
 }
 
+func (aex *adminExecutor) CommitOrAbort(ctx context.Context, tc *dgoapi.TxnContext) error {
+	return aex.dg.CommitOrAbort(ctx, tc)
+}
+
 func (de *dgraphExecutor) Execute(ctx context.Context, req *dgoapi.Request) (
 	*dgoapi.Response, error) {
 	return de.dg.Execute(ctx, req)
 }
 
+func (de *dgraphExecutor) CommitOrAbort(ctx context.Context, tc *dgoapi.TxnContext) error {
+	return de.dg.CommitOrAbort(ctx, tc)
+}
+
 func (rf *resolverFactory) WithQueryResolver(
 	name string, resolver func(schema.Query) QueryResolver) ResolverFactory {
 	rf.queryResolvers[name] = resolver

graphql/resolve/resolver_error_test.go

@@ -116,6 +116,10 @@ func (ex *executor) Execute(ctx context.Context, req *dgoapi.Request) (*dgoapi.R
 
 }
 
+func (ex *executor) CommitOrAbort(ctx context.Context, tc *dgoapi.TxnContext) error {
+	return nil
+}
+
 // Tests in resolver_test.yaml are about what gets into a completed result (addition
 // of "null", errors and error propagation).  Exact JSON result (e.g. order) doesn't
 // matter here - that makes for easier to format and read tests for these many cases.