libtracker-data: Fix transaction handling

Make sure to rollback all changes when an error happens during a
SPARQL update.

7 files changed, 102 insertions, 89 deletions

diff --git a/src/libtracker-data/tracker-data-update.c b/src/libtracker-data/tracker-data-update.c
index 60e2b1419..aa90dbf77 100644
--- a/src/libtracker-data/tracker-data-update.c
+++ b/src/libtracker-data/tracker-data-update.c
@@ -86,8 +86,7 @@ struct _TrackerDataBlankBuffer {
 	GArray *objects;
 };
 
-
-static gint transaction_level = 0;
+static gboolean in_transaction = FALSE;
 static TrackerDataUpdateBuffer update_buffer;
 static TrackerDataBlankBuffer blank_buffer;
 
@@ -448,6 +447,30 @@ tracker_data_update_buffer_flush (void)
 }
 
 static void
+tracker_data_update_buffer_clear (void)
+{
+	if (update_buffer.subject == NULL) {
+		/* nothing to clear */
+		return;
+	}
+
+	g_free (update_buffer.new_subject);
+	update_buffer.new_subject = NULL;
+
+	g_hash_table_remove_all (update_buffer.predicates);
+	g_hash_table_remove_all (update_buffer.tables);
+	g_free (update_buffer.subject);
+	update_buffer.subject = NULL;
+
+	g_ptr_array_foreach (update_buffer.types, (GFunc) g_free, NULL);
+	g_ptr_array_free (update_buffer.types, TRUE);
+	update_buffer.types = NULL;
+
+	tracker_fts_update_rollback ();
+	update_buffer.fts_ever_updated = FALSE;
+}
+
+static void
 tracker_data_blank_buffer_flush (void)
 {
 	/* end of blank node */
@@ -893,16 +916,12 @@ tracker_data_delete_statement (const gchar            *subject,
 	g_return_if_fail (subject != NULL);
 	g_return_if_fail (predicate != NULL);
 	g_return_if_fail (object != NULL);
-
-	tracker_data_begin_transaction ();
+	g_return_if_fail (in_transaction);
 
 	subject_id = query_resource_id (subject);
 	
 	if (subject_id == 0) {
 		/* subject not in database */
-
-		tracker_data_commit_transaction ();
-
 		return;
 	}
 
@@ -1098,8 +1117,6 @@ tracker_data_delete_statement (const gchar            *subject,
 		g_ptr_array_foreach (types, (GFunc) g_free, NULL);
 		g_ptr_array_free (types, TRUE);
 	}
-
-	tracker_data_commit_transaction ();
 }
 
 static gboolean
@@ -1177,6 +1194,7 @@ tracker_data_insert_statement (const gchar            *subject,
 	g_return_if_fail (subject != NULL);
 	g_return_if_fail (predicate != NULL);
 	g_return_if_fail (object != NULL);
+	g_return_if_fail (in_transaction);
 
 	property = tracker_ontology_get_property_by_uri (predicate);
 	if (property != NULL) {
@@ -1206,6 +1224,7 @@ tracker_data_insert_statement_with_uri (const gchar            *subject,
 	g_return_if_fail (subject != NULL);
 	g_return_if_fail (predicate != NULL);
 	g_return_if_fail (object != NULL);
+	g_return_if_fail (in_transaction);
 
 	property = tracker_ontology_get_property_by_uri (predicate);
 	if (property == NULL) {
@@ -1222,8 +1241,6 @@ tracker_data_insert_statement_with_uri (const gchar            *subject,
 		return;
 	}
 
-	tracker_data_begin_transaction ();
-
 	/* subjects and objects starting with `:' are anonymous blank nodes */
 	if (g_str_has_prefix (object, ":")) {
 		/* anonymous blank node used as object in a statement */
@@ -1244,8 +1261,6 @@ tracker_data_insert_statement_with_uri (const gchar            *subject,
 
 			g_hash_table_remove (blank_buffer.table, object);
 
-			tracker_data_commit_transaction ();
-
 			return;
 		} else {
 			g_critical ("Blank node '%s' not found", object);
@@ -1253,7 +1268,6 @@ tracker_data_insert_statement_with_uri (const gchar            *subject,
 	}
 
 	if (!tracker_data_insert_statement_common (subject, predicate, object)) {
-		tracker_data_commit_transaction ();
 		return;
 	}
 
@@ -1274,8 +1288,7 @@ tracker_data_insert_statement_with_uri (const gchar            *subject,
 		/* add value to metadata database */
 		cache_set_metadata_decomposed (property, object, &actual_error);
 		if (actual_error) {
-			/* FIXME rollback instead of commit */
-			tracker_data_commit_transaction ();
+			tracker_data_update_buffer_clear ();
 			g_propagate_error (error, actual_error);
 			return;
 		}
@@ -1284,8 +1297,6 @@ tracker_data_insert_statement_with_uri (const gchar            *subject,
 			insert_callback (subject, predicate, object, update_buffer.types, insert_data);
 		}
 	}
-
-	tracker_data_commit_transaction ();
 }
 
 void
@@ -1300,6 +1311,7 @@ tracker_data_insert_statement_with_string (const gchar            *subject,
 	g_return_if_fail (subject != NULL);
 	g_return_if_fail (predicate != NULL);
 	g_return_if_fail (object != NULL);
+	g_return_if_fail (in_transaction);
 
 	property = tracker_ontology_get_property_by_uri (predicate);
 	if (property == NULL) {
@@ -1312,18 +1324,14 @@ tracker_data_insert_statement_with_string (const gchar            *subject,
 		return;
 	}
 
-	tracker_data_begin_transaction ();
-
 	if (!tracker_data_insert_statement_common (subject, predicate, object)) {
-		tracker_data_commit_transaction ();
 		return;
 	}
 
 	/* add value to metadata database */
 	cache_set_metadata_decomposed (property, object, &actual_error);
 	if (actual_error) {
-		/* FIXME rollback instead of commit */
-		tracker_data_commit_transaction ();
+		tracker_data_update_buffer_clear ();
 		g_propagate_error (error, actual_error);
 		return;
 	}
@@ -1331,8 +1339,6 @@ tracker_data_insert_statement_with_string (const gchar            *subject,
 	if (insert_callback) {
 		insert_callback (subject, predicate, object, update_buffer.types, insert_data);
 	}
-
-	tracker_data_commit_transaction ();
 }
 
 static void
@@ -1549,21 +1555,21 @@ tracker_data_begin_transaction (void)
 {
 	TrackerDBInterface *iface;
 
-	if (transaction_level == 0) {
-		update_buffer.resource_cache = g_hash_table_new_full (g_str_hash, g_str_equal, g_free, NULL);
-		update_buffer.predicates = g_hash_table_new_full (g_direct_hash, g_direct_equal, g_object_unref, (GDestroyNotify) g_value_array_free);
-		update_buffer.tables = g_hash_table_new_full (g_str_hash, g_str_equal,
-							      g_free, (GDestroyNotify) cache_table_free);
-		if (blank_buffer.table == NULL) {
-			blank_buffer.table = g_hash_table_new_full (g_str_hash, g_str_equal, g_free, g_free);
-		}
-
-		iface = tracker_db_manager_get_db_interface ();
+	g_return_if_fail (!in_transaction);
 
-		tracker_db_interface_start_transaction (iface);
+	update_buffer.resource_cache = g_hash_table_new_full (g_str_hash, g_str_equal, g_free, NULL);
+	update_buffer.predicates = g_hash_table_new_full (g_direct_hash, g_direct_equal, g_object_unref, (GDestroyNotify) g_value_array_free);
+	update_buffer.tables = g_hash_table_new_full (g_str_hash, g_str_equal,
+						      g_free, (GDestroyNotify) cache_table_free);
+	if (blank_buffer.table == NULL) {
+		blank_buffer.table = g_hash_table_new_full (g_str_hash, g_str_equal, g_free, g_free);
 	}
 
-	transaction_level++;
+	iface = tracker_db_manager_get_db_interface ();
+
+	tracker_db_interface_start_transaction (iface);
+
+	in_transaction = TRUE;
 }
 
 void
@@ -1571,27 +1577,27 @@ tracker_data_commit_transaction (void)
 {
 	TrackerDBInterface *iface;
 
-	transaction_level--;
+	g_return_if_fail (in_transaction);
 
-	if (transaction_level == 0) {
-		tracker_data_update_buffer_flush ();
+	in_transaction = FALSE;
 
-		if (update_buffer.fts_ever_updated) {
-			tracker_fts_update_commit ();
-			update_buffer.fts_ever_updated = FALSE;
-		}
+	tracker_data_update_buffer_flush ();
 
-		iface = tracker_db_manager_get_db_interface ();
+	if (update_buffer.fts_ever_updated) {
+		tracker_fts_update_commit ();
+		update_buffer.fts_ever_updated = FALSE;
+	}
+
+	iface = tracker_db_manager_get_db_interface ();
 
-		tracker_db_interface_end_transaction (iface);
+	tracker_db_interface_end_transaction (iface);
 
-		g_hash_table_unref (update_buffer.resource_cache);
-		g_hash_table_unref (update_buffer.predicates);
-		g_hash_table_unref (update_buffer.tables);
+	g_hash_table_unref (update_buffer.resource_cache);
+	g_hash_table_unref (update_buffer.predicates);
+	g_hash_table_unref (update_buffer.tables);
 
-		if (commit_callback) {
-			commit_callback (commit_data);
-		}
+	if (commit_callback) {
+		commit_callback (commit_data);
 	}
 }
 
@@ -1770,14 +1776,26 @@ void
 tracker_data_update_sparql (const gchar  *update,
 			    GError      **error)
 {
+	TrackerDBInterface *iface;
 	TrackerSparqlQuery *sparql_query;
 
 	g_return_if_fail (update != NULL);
 
+	iface = tracker_db_manager_get_db_interface ();
+
 	sparql_query = tracker_sparql_query_new_update (update);
 
+	tracker_db_interface_execute_query (iface, NULL, "SAVEPOINT sparql");
+
 	tracker_sparql_query_execute (sparql_query, error);
 
+	if (*error) {
+		tracker_data_update_buffer_clear ();
+		tracker_db_interface_execute_query (iface, NULL, "ROLLBACK TO sparql");
+	}
+
+	tracker_db_interface_execute_query (iface, NULL, "RELEASE sparql");
+
 	g_object_unref (sparql_query);
 }
 
diff --git a/src/libtracker-data/tracker-sparql-query.vala b/src/libtracker-data/tracker-sparql-query.vala
index 105ce3563..657065079 100644
--- a/src/libtracker-data/tracker-sparql-query.vala
+++ b/src/libtracker-data/tracker-sparql-query.vala
@@ -443,32 +443,25 @@ public class Tracker.SparqlQuery : Object {
 		} else {
 			// SPARQL update supports multiple operations in a single query
 
-			// all updates should be committed in one transaction
-			Data.begin_transaction ();
-
-			try {
-				while (current () != SparqlTokenType.EOF) {
-					switch (current ()) {
-					case SparqlTokenType.INSERT:
-						execute_insert ();
-						break;
-					case SparqlTokenType.DELETE:
-						execute_delete ();
-						break;
-					case SparqlTokenType.DROP:
-						execute_drop_graph ();
-						break;
-					case SparqlTokenType.SELECT:
-					case SparqlTokenType.CONSTRUCT:
-					case SparqlTokenType.DESCRIBE:
-					case SparqlTokenType.ASK:
-						throw get_error ("SELECT, CONSTRUCT, DESCRIBE, and ASK are not supported in update mode");
-					default:
-						throw get_error ("expected INSERT or DELETE");
-					}
+			while (current () != SparqlTokenType.EOF) {
+				switch (current ()) {
+				case SparqlTokenType.INSERT:
+					execute_insert ();
+					break;
+				case SparqlTokenType.DELETE:
+					execute_delete ();
+					break;
+				case SparqlTokenType.DROP:
+					execute_drop_graph ();
+					break;
+				case SparqlTokenType.SELECT:
+				case SparqlTokenType.CONSTRUCT:
+				case SparqlTokenType.DESCRIBE:
+				case SparqlTokenType.ASK:
+					throw get_error ("SELECT, CONSTRUCT, DESCRIBE, and ASK are not supported in update mode");
+				default:
+					throw get_error ("expected INSERT or DELETE");
 				}
-			} finally {
-				Data.commit_transaction ();
 			}
 
 			return null;
diff --git a/src/libtracker-data/tracker-turtle-reader.vala b/src/libtracker-data/tracker-turtle-reader.vala
index 138c65849..f04cca5d0 100644
--- a/src/libtracker-data/tracker-turtle-reader.vala
+++ b/src/libtracker-data/tracker-turtle-reader.vala
@@ -363,19 +363,13 @@ public class Tracker.TurtleReader : Object {
 	}
 
 	public static void load (string path) throws FileError, SparqlError, DataError {
-		try {
-			Data.begin_transaction ();
-
-			var reader = new TurtleReader (path);
-			while (reader.next ()) {
-				if (reader.object_is_uri) {
-					Data.insert_statement_with_uri (reader.subject, reader.predicate, reader.object);
-				} else {
-					Data.insert_statement_with_string (reader.subject, reader.predicate, reader.object);
-				}
+		var reader = new TurtleReader (path);
+		while (reader.next ()) {
+			if (reader.object_is_uri) {
+				Data.insert_statement_with_uri (reader.subject, reader.predicate, reader.object);
+			} else {
+				Data.insert_statement_with_string (reader.subject, reader.predicate, reader.object);
 			}
-		} finally {
-			Data.commit_transaction ();
 		}
 	}
 
diff --git a/src/tracker-store/tracker-store.c b/src/tracker-store/tracker-store.c
index a31520f36..c9bb9c79b 100644
--- a/src/tracker-store/tracker-store.c
+++ b/src/tracker-store/tracker-store.c
@@ -524,7 +524,9 @@ tracker_store_sparql_update (const gchar *sparql,
 		private->batch_count = 0;
 	}
 
+	tracker_data_begin_transaction ();
 	tracker_data_update_sparql (sparql, error);
+	tracker_data_commit_transaction ();
 
 	if (private->start_log) {
 		log_to_journal (private, sparql);
diff --git a/tests/libtracker-data/tracker-ontology-test.c b/tests/libtracker-data/tracker-ontology-test.c
index fbd5f584a..b108b4892 100644
--- a/tests/libtracker-data/tracker-ontology-test.c
+++ b/tests/libtracker-data/tracker-ontology-test.c
@@ -100,7 +100,9 @@ test_query (gconstpointer test_data)
 
 	/* load data set */
 	data_filename = g_strconcat (data_prefix, ".ttl", NULL);
+	tracker_data_begin_transaction ();
 	tracker_turtle_reader_load (data_filename, &error);
+	tracker_data_commit_transaction ();
 	g_assert (error == NULL);
 
 	query_filename = g_strconcat (test_prefix, ".rq", NULL);
diff --git a/tests/libtracker-data/tracker-sparql-test.c b/tests/libtracker-data/tracker-sparql-test.c
index c51899d82..254f6ca87 100644
--- a/tests/libtracker-data/tracker-sparql-test.c
+++ b/tests/libtracker-data/tracker-sparql-test.c
@@ -115,7 +115,9 @@ test_sparql_query (gconstpointer test_data)
 
 	/* load data set */
 	data_filename = g_strconcat (data_prefix, ".ttl", NULL);
+	tracker_data_begin_transaction ();
 	tracker_turtle_reader_load (data_filename, &error);
+	tracker_data_commit_transaction ();
 	g_assert (error == NULL);
 
 	query_filename = g_strconcat (test_prefix, ".rq", NULL);
diff --git a/tests/libtracker-fts/tracker-fts-test.c b/tests/libtracker-fts/tracker-fts-test.c
index 8acd702a5..6e07a076c 100644
--- a/tests/libtracker-fts/tracker-fts-test.c
+++ b/tests/libtracker-fts/tracker-fts-test.c
@@ -79,7 +79,9 @@ test_sparql_query (gconstpointer test_data)
 	g_file_get_contents (update_filename, &update, NULL, &error);
 	g_assert (error == NULL);
 
+	tracker_data_begin_transaction ();
 	tracker_data_update_sparql (update, &error);
+	tracker_data_commit_transaction ();
 	g_assert (error == NULL);
 
 	g_free (update_filename);