Use database transactions where appropriate

Signed-off-by: Alexis Sellier <alexis@radicle.xyz>

radicle-node/src/address/store.rs

@@ -11,6 +11,7 @@ use crate::address::{KnownAddress, Source};
 use crate::clock::Timestamp;
 use crate::prelude::Address;
 use crate::service::NodeId;
+use crate::sql::transaction;
 use crate::wire::message::AddressType;
 
 #[derive(Error, Debug)]
@@ -120,52 +121,56 @@ impl Store for Book {
         timestamp: Timestamp,
         addrs: impl IntoIterator<Item = KnownAddress>,
     ) -> Result<bool, Error> {
-        let mut stmt = self.db.prepare(
+        transaction(&self.db, move |db| {
-            "INSERT INTO nodes (id, features, alias, timestamp)
+            let mut stmt = db.prepare(
-             VALUES (?1, ?2, ?3, ?4)
+                "INSERT INTO nodes (id, features, alias, timestamp)
-             ON CONFLICT DO UPDATE
+                 VALUES (?1, ?2, ?3, ?4)
-             SET features = ?2, alias = ?3, timestamp = ?4
-             WHERE timestamp < ?4",
-        )?;
-
-        stmt.bind(1, node)?;
-        stmt.bind(2, features)?;
-        stmt.bind(3, alias)?;
-        stmt.bind(4, timestamp as i64)?;
-        stmt.next()?;
-
-        for addr in addrs {
-            let mut stmt = self.db.prepare(
-                "INSERT INTO addresses (node, type, value, source, timestamp)
-                 VALUES (?1, ?2, ?3, ?4, ?5)
                  ON CONFLICT DO UPDATE
-                 SET timestamp = ?5
+                 SET features = ?2, alias = ?3, timestamp = ?4
-                 WHERE timestamp < ?5",
+                 WHERE timestamp < ?4",
             )?;
+
             stmt.bind(1, node)?;
-            stmt.bind(2, AddressType::from(&addr.addr))?;
+            stmt.bind(2, features)?;
-            stmt.bind(3, addr.addr)?;
+            stmt.bind(3, alias)?;
-            stmt.bind(4, addr.source)?;
+            stmt.bind(4, timestamp as i64)?;
-            stmt.bind(5, timestamp as i64)?;
             stmt.next()?;
-        }
+
-        Ok(self.db.change_count() > 0)
+            for addr in addrs {
+                let mut stmt = db.prepare(
+                    "INSERT INTO addresses (node, type, value, source, timestamp)
+                     VALUES (?1, ?2, ?3, ?4, ?5)
+                     ON CONFLICT DO UPDATE
+                     SET timestamp = ?5
+                     WHERE timestamp < ?5",
+                )?;
+                stmt.bind(1, node)?;
+                stmt.bind(2, AddressType::from(&addr.addr))?;
+                stmt.bind(3, addr.addr)?;
+                stmt.bind(4, addr.source)?;
+                stmt.bind(5, timestamp as i64)?;
+                stmt.next()?;
+            }
+            Ok(db.change_count() > 0)
+        })
+        .map_err(Error::from)
     }
 
     fn remove(&mut self, node: &NodeId) -> Result<bool, Error> {
-        self.db
+        transaction(&self.db, move |db| {
-            .prepare("DELETE FROM nodes WHERE id = ?")?
+            db.prepare("DELETE FROM nodes WHERE id = ?")?
-            .into_cursor()
+                .into_cursor()
-            .bind(&[(*node).into()])?
+                .bind(&[(*node).into()])?
-            .next();
+                .next();
 
-        self.db
+            db.prepare("DELETE FROM addresses WHERE node = ?")?
-            .prepare("DELETE FROM addresses WHERE node = ?")?
+                .into_cursor()
-            .into_cursor()
+                .bind(&[(*node).into()])?
-            .bind(&[(*node).into()])?
+                .next();
-            .next();
 
-        Ok(self.db.change_count() > 0)
+            Ok(db.change_count() > 0)
+        })
+        .map_err(Error::from)
     }
 
     fn entries(&self) -> Result<Box<dyn Iterator<Item = (NodeId, KnownAddress)>>, Error> {

radicle-node/src/lib.rs

@@ -5,6 +5,7 @@ pub mod control;
 pub mod decoder;
 pub mod logger;
 pub mod service;
+pub mod sql;
 #[cfg(test)]
 pub mod test;
 pub mod transport;

radicle-node/src/sql.rs

@@ -0,0 +1,21 @@
+use sqlite as sql;
+
+/// Run an SQL query inside a transaction.
+/// Commits the transaction on success, and rolls back on error.
+pub fn transaction<T>(
+    db: &sql::Connection,
+    query: impl FnOnce(&sql::Connection) -> Result<T, sql::Error>,
+) -> Result<T, sql::Error> {
+    db.execute("BEGIN")?;
+
+    match query(db) {
+        Ok(result) => {
+            db.execute("COMMIT")?;
+            Ok(result)
+        }
+        Err(err) => {
+            db.execute("ROLLBACK")?;
+            Err(err)
+        }
+    }
+}