Consume results of write-only transactions (guacsec#176)

While it's not strictly necessary, since the Neo4j server will execute the transaction queries just before commit at latest, consuming results makes sure the queries are run right when the client starts pulling results. Another advantage is that some query errors will surface when you consume results and not when you commit.
Qiwen-Zhou · Oct 21, 2022 · 03e2c33 · 03e2c33
1 parent 4b98664
commit 03e2c33
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 8 deletions.
diff --git a/pkg/assembler/graphdb.go b/pkg/assembler/graphdb.go
@@ -76,7 +76,12 @@ func StoreGraph(g Graph, client graphdb.Client) error {
 	_, err := session.WriteTransaction(
 		func(tx graphdb.Transaction) (interface{}, error) {
 			for i, query := range queries {
-				if _, err := tx.Run(query, params[i]); err != nil {
+				result, err := tx.Run(query, params[i])
+				if err != nil {
+					return nil, err
+				}
+				_, err = result.Consume()
+				if err != nil {
 					return nil, err
 				}
 			}

diff --git a/pkg/assembler/graphdb/graphdb_test_utils.go b/pkg/assembler/graphdb/graphdb_test_utils.go
@@ -35,7 +35,12 @@ func WriteQueryForTesting(client Client, query string, args map[string]interface
 
 	_, err := session.WriteTransaction(
 		func(tx Transaction) (interface{}, error) {
-			return tx.Run(query, args)
+			result, err := tx.Run(query, args)
+			if err != nil {
+				return nil, err
+			}
+			_, err = result.Consume()
+			return nil, err
 		})
 	return err
 }
@@ -48,15 +53,14 @@ func ReadQueryForTesting(client Client, query string, args map[string]interface{
 	session := client.NewSession(neo4j.SessionConfig{})
 	defer session.Close()
 
-	// For this testing function, we just collect all values.
-	values := make([][]interface{}, 0)
-
-	_, err := session.ReadTransaction(
+	result, err := session.ReadTransaction(
+		// For this testing function, we just collect all values.
 		func(tx Transaction) (interface{}, error) {
 			records, err := tx.Run(query, args)
 			if err != nil {
 				return nil, err
 			}
+			values := make([][]interface{}, 0)
 			// Since `records` is valid only while `tx` is in
 			// scope, we have to process all data here.
 			for records.Next() {
@@ -72,7 +76,7 @@ func ReadQueryForTesting(client Client, query string, args map[string]interface{
 	if err != nil {
 		return nil, err
 	}
-	return values, err
+	return result.([][]interface{}), nil
 }
 
 // ClearDBForTesting clears the entire database.
@@ -84,7 +88,12 @@ func ClearDBForTesting(client Client) error {
 
 	_, err := session.WriteTransaction(
 		func(tx Transaction) (interface{}, error) {
-			return tx.Run("MATCH (n) DETACH DELETE n", nil)
+			results, err := tx.Run("MATCH (n) DETACH DELETE n", nil)
+			if err != nil {
+				return nil, err
+			}
+			_, err = results.Consume()
+			return nil, err
 		})
 	return err
 }