Merge pull request #18 from ShimmerGlass/rollback-tx

Fix: mariadb: rollback transactions on error

go.mod

@@ -5,16 +5,13 @@ go 1.13
 require (
 	github.com/abbot/go-http-auth v0.4.0
 	github.com/antlr/antlr4 v0.0.0-20200103163232-691acdc23f1f
-	github.com/cespare/reflex v0.2.0 // indirect
 	github.com/cheggaaa/pb v2.0.7+incompatible
 	github.com/deckarep/golang-set v1.7.1
 	github.com/go-sql-driver/mysql v1.5.0
 	github.com/golang-collections/go-datastructures v0.0.0-20150211160725-59788d5eb259
 	github.com/gorilla/mux v1.7.3
-	github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 // indirect
 	github.com/mattn/go-colorable v0.0.9 // indirect
 	github.com/mattn/go-isatty v0.0.3 // indirect
-	github.com/ogier/pflag v0.0.1 // indirect
 	github.com/spf13/cobra v0.0.5
 	github.com/spf13/viper v1.6.1
 	github.com/stretchr/testify v1.3.0

go.sum

@@ -11,8 +11,6 @@ github.com/antlr/antlr4 v0.0.0-20200103163232-691acdc23f1f/go.mod h1:T7PbCXFs94r
 github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
-github.com/cespare/reflex v0.2.0 h1:6d9WpWJseKjJvZEevKP7Pk42nPx2+BUTqmhNk8wZPwM=
-github.com/cespare/reflex v0.2.0/go.mod h1:ooqOLJ4algvHP/oYvKWfWJ9tFUzCLDk5qkIJduMYrgI=
 github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
 github.com/cheggaaa/pb v2.0.7+incompatible h1:gLKifR1UkZ/kLkda5gC0K6c8g+jU2sINPtBeOiNlMhU=
 github.com/cheggaaa/pb v2.0.7+incompatible/go.mod h1:pQciLPpbU0oxA0h+VJYYLxO+XeDQb5pZijXscXHm81s=
@@ -67,8 +65,6 @@ github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22
 github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo=
 github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
 github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
-github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 h1:Z9n2FFNUXsshfwJMBgNA0RU6/i7WVaAegv3PtuIHPMs=
-github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51/go.mod h1:CzGEWj7cYgsdH8dAjBGEr58BoE7ScuLd+fwFZ44+/x8=
 github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
@@ -92,8 +88,6 @@ github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrk
 github.com/mitchellh/mapstructure v1.1.2 h1:fmNYVwqnSfB9mZU6OS2O6GsXM+wcskZDuKQzvN1EDeE=
 github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
 github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
-github.com/ogier/pflag v0.0.1 h1:RW6JSWSu/RkSatfcLtogGfFgpim5p7ARQ10ECk5O750=
-github.com/ogier/pflag v0.0.1/go.mod h1:zkFki7tvTa0tafRvTBIZTvzYyAu6kQhPZFnshFFPE+g=
 github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U=
 github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181zc=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=

internal/database/mariadb.go

@@ -163,19 +163,22 @@ func (m *MariaDB) resolveAssets(assets []knowledge.AssetKey, registry *AssetRegi
 
 	stmt, err := tx.PrepareContext(context.Background(), "SELECT id FROM assets WHERE type = ? AND value = ?")
 	if err != nil {
+		tx.Rollback()
 		return fmt.Errorf("Unable to prepare statement: %v", err)
 	}
 
 	for _, a := range assets {
 		q, err := stmt.QueryContext(context.Background(), a.Type, a.Key)
 		if err != nil {
+			tx.Rollback()
 			return fmt.Errorf("Unable to query asset %v: %v", a, err)
 		}
 		defer q.Close()
 
 		for q.Next() {
 			var idx int64
 			if err := q.Scan(&idx); err != nil {
+				tx.Rollback()
 				return fmt.Errorf("Unable to retrieve id for %v: %v", a, err)
 			}
 			registry.Set(knowledge.AssetKey(a), idx)
@@ -223,10 +226,12 @@ INSERT INTO assets (type, value) VALUES (?, ?)`)
 
 			res, err := insertQuery.ExecContext(context.Background(), a.Type, a.Key)
 			if err != nil {
+				tx.Rollback()
 				return 0, fmt.Errorf("Unable to insert asset %v: %v", a, err)
 			}
 			idx, err := res.LastInsertId()
 			if err != nil {
+				tx.Rollback()
 				return 0, fmt.Errorf("Unable to retrieve last inserted ID: %v", err)
 			}
 			registry.Set(knowledge.AssetKey(a), idx)
@@ -262,6 +267,7 @@ func (m *MariaDB) upsertRelations(source string, relations []knowledge.Relation,
 		q, err := tx.PrepareContext(context.Background(),
 			"INSERT INTO relations (from_id, to_id, type, source) VALUES (?, ?, ?, ?)")
 		if err != nil {
+			tx.Rollback()
 			return 0, fmt.Errorf("Unable to prepare relation insertion query: %v", err)
 		}
 		defer q.Close()
@@ -286,6 +292,7 @@ func (m *MariaDB) upsertRelations(source string, relations []knowledge.Relation,
 					bar.Increment()
 					continue
 				}
+				tx.Rollback()
 				return 0, fmt.Errorf("Unable to insert relation %v (%d -> %d): %v", r, idxFrom, idxTo, err)
 			}
 			bar.Increment()
@@ -318,6 +325,7 @@ INNER JOIN assets a ON r.from_id = a.id
 INNER JOIN assets b ON r.to_id = b.id
 WHERE a.type = ? AND a.value = ? AND b.type = ? AND b.value = ? AND r.type = ? AND r.source = ?`)
 	if err != nil {
+		tx.Rollback()
 		return 0, 0, err
 	}
 	defer stmt.Close()
@@ -326,11 +334,13 @@ WHERE a.type = ? AND a.value = ? AND b.type = ? AND b.value = ? AND r.type = ? A
 		res, err := stmt.ExecContext(context.Background(),
 			r.From.Type, r.From.Key, r.To.Type, r.To.Key, r.Type, source)
 		if err != nil {
+			tx.Rollback()
 			return 0, 0, fmt.Errorf("Unable to detete relation %v: %v", r, err)
 		}
 		bar.Increment()
 		rCount, err := res.RowsAffected()
 		if err != nil {
+			tx.Rollback()
 			return 0, 0, fmt.Errorf("Unable to count rows affected by relations deletion: %v", err)
 		}
 		removedCount += rCount
@@ -342,11 +352,13 @@ DELETE a FROM assets a
 WHERE id NOT IN (select from_id from relations)
 AND id NOT IN (select to_id from relations)`)
 	if err != nil {
+		tx.Rollback()
 		return 0, 0, fmt.Errorf("Unable to delete assets: %v", err)
 	}
 
 	removedAssetsCount, err := res.RowsAffected()
 	if err != nil {
+		tx.Rollback()
 		return 0, 0, fmt.Errorf("Unable to count rows affected by assets deletion: %v", err)
 	}