Repository: breez/breez-sdk-liquid (language: python)

Issue:
Real-time Sync: Add persistency layer
Closes #501.
Real-time Sync: Add pull and merge
Closes #502, closes #503.
Based on #555.
Real-time Sync: Add a method to push changes

Real-time Sync: Add a method to clean incoming and outgoing changes



Discussion / hints:

Fix patch:
diff --git a/lib/Cargo.lock b/lib/Cargo.lock
index f90d07f54..67f83b082 100644
--- a/lib/Cargo.lock
+++ b/lib/Cargo.lock
@@ -188,6 +188,12 @@ dependencies = [
  "backtrace",
 ]
 
+[[package]]
+name = "arrayref"
+version = "0.3.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "76a2e8124351fda1ef8aaaa3bbd7ebbcb486bbcd4225aca0aa0d84bb2db8fecb"
+
 [[package]]
 name = "arrayvec"
 version = "0.7.6"
diff --git a/lib/core/src/model.rs b/lib/core/src/model.rs
index 015350edd..b67e8dead 100644
--- a/lib/core/src/model.rs
+++ b/lib/core/src/model.rs
@@ -1615,3 +1615,19 @@ macro_rules! get_invoice_description {
         }
     };
 }
+
+#[macro_export]
+macro_rules! get_updated_fields {
+    ($($var:ident),* $(,)?) => {{
+        let mut options = Vec::new();
+        $(
+            if $var.is_some() {
+                options.push(stringify!($var).to_string());
+            }
+        )*
+        match options.len() > 0 {
+            true => Some(options),
+            false => None,
+        }
+    }};
+}
diff --git a/lib/core/src/persist/chain.rs b/lib/core/src/persist/chain.rs
index 2425b9c42..a9b5143ee 100644
--- a/lib/core/src/persist/chain.rs
+++ b/lib/core/src/persist/chain.rs
@@ -2,22 +2,25 @@ use std::collections::HashMap;
 
 use anyhow::Result;
 use boltz_client::swaps::boltz::{ChainSwapDetails, CreateChainResponse};
-use rusqlite::{named_params, params, Connection, Row};
+use rusqlite::{named_params, params, Connection, Row, TransactionBehavior};
 use sdk_common::bitcoin::hashes::{hex::ToHex, sha256, Hash};
 use serde::{Deserialize, Serialize};
 
-use crate::ensure_sdk;
 use crate::error::PaymentError;
 use crate::model::*;
 use crate::persist::{get_where_clause_state_in, Persister};
+use crate::sync::model::RecordType;
+use crate::{ensure_sdk, get_updated_fields};
 
 impl Persister {
     pub(crate) fn insert_chain_swap(&self, chain_swap: &ChainSwap) -> Result<()> {
-        let con = self.get_connection()?;
+        let mut con = self.get_connection()?;
+        let tx = con.transaction_with_behavior(TransactionBehavior::Immediate)?;
 
         // There is a limit of 16 param elements in a single tuple in rusqlite,
         // so we split up the insert into two statements.
-        let mut stmt = con.prepare(
+        let id_hash = sha256::Hash::hash(chain_swap.id.as_bytes()).to_hex();
+        tx.execute(
             "
             INSERT INTO chain_swaps (
                 id,
@@ -38,28 +41,27 @@ impl Persister {
                 state
             )
             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            (
+                &chain_swap.id,
+                &id_hash,
+                &chain_swap.direction,
+                &chain_swap.claim_address,
+                &chain_swap.lockup_address,
+                &chain_swap.timeout_block_height,
+                &chain_swap.preimage,
+                &chain_swap.payer_amount_sat,
+                &chain_swap.receiver_amount_sat,
+                &chain_swap.accept_zero_conf,
+                &chain_swap.create_response_json,
+                &chain_swap.claim_private_key,
+                &chain_swap.refund_private_key,
+                &chain_swap.claim_fees_sat,
+                &chain_swap.created_at,
+                &chain_swap.state,
+            ),
         )?;
-        let id_hash = sha256::Hash::hash(chain_swap.id.as_bytes()).to_hex();
-        _ = stmt.execute((
-            &chain_swap.id,
-            &id_hash,
-            &chain_swap.direction,
-            &chain_swap.claim_address,
-            &chain_swap.lockup_address,
-            &chain_swap.timeout_block_height,
-            &chain_swap.preimage,
-            &chain_swap.payer_amount_sat,
-            &chain_swap.receiver_amount_sat,
-            &chain_swap.accept_zero_conf,
-            &chain_swap.create_response_json,
-            &chain_swap.claim_private_key,
-            &chain_swap.refund_private_key,
-            &chain_swap.claim_fees_sat,
-            &chain_swap.created_at,
-            &chain_swap.state,
-        ))?;
 
-        con.execute(
+        tx.execute(
             "UPDATE chain_swaps
             SET
                 description = :description,
@@ -79,6 +81,10 @@ impl Persister {
             },
         )?;
 
+        Self::commit_outgoing(&tx, &chain_swap.id, RecordType::Chain, None)?;
+
+        tx.commit()?;
+
         Ok(())
     }
 
@@ -319,8 +325,10 @@ impl Persister {
         refund_tx_id: Option<&str>,
     ) -> Result<(), PaymentError> {
         // Do not overwrite server_lockup_tx_id, user_lockup_tx_id, claim_tx_id, refund_tx_id
-        let con: Connection = self.get_connection()?;
-        con.execute(
+        let mut con = self.get_connection()?;
+        let tx = con.transaction_with_behavior(TransactionBehavior::Immediate)?;
+
+        tx.execute(
             "UPDATE chain_swaps
             SET
                 server_lockup_tx_id =
@@ -360,6 +368,11 @@ impl Persister {
             },
         )?;
 
+        let updated_fields = get_updated_fields!(server_lockup_tx_id);
+        Self::commit_outgoing(&tx, swap_id, RecordType::Chain, updated_fields)?;
+
+        tx.commit()?;
+
         Ok(())
     }
 }
diff --git a/lib/core/src/persist/receive.rs b/lib/core/src/persist/receive.rs
index 13a31f004..8d04b0d27 100644
--- a/lib/core/src/persist/receive.rs
+++ b/lib/core/src/persist/receive.rs
@@ -2,20 +2,23 @@ use std::collections::HashMap;
 
 use anyhow::Result;
 use boltz_client::swaps::boltz::CreateReverseResponse;
-use rusqlite::{named_params, params, Connection, Row};
+use rusqlite::{named_params, params, Connection, Row, TransactionBehavior};
 use sdk_common::bitcoin::hashes::{hex::ToHex, sha256, Hash};
 use serde::{Deserialize, Serialize};
 
-use crate::ensure_sdk;
 use crate::error::PaymentError;
 use crate::model::*;
 use crate::persist::{get_where_clause_state_in, Persister};
+use crate::sync::model::RecordType;
+use crate::{ensure_sdk, get_updated_fields};
 
 impl Persister {
     pub(crate) fn insert_receive_swap(&self, receive_swap: &ReceiveSwap) -> Result<()> {
-        let con = self.get_connection()?;
+        let mut con = self.get_connection()?;
+        let tx = con.transaction_with_behavior(TransactionBehavior::Immediate)?;
 
-        let mut stmt = con.prepare(
+        let id_hash = sha256::Hash::hash(receive_swap.id.as_bytes()).to_hex();
+        tx.execute(
             "
             INSERT INTO receive_swaps (
                 id,
@@ -34,26 +37,25 @@ impl Persister {
                 state
             )
             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            (
+                &receive_swap.id,
+                id_hash,
+                &receive_swap.preimage,
+                &receive_swap.create_response_json,
+                &receive_swap.claim_private_key,
+                &receive_swap.invoice,
+                &receive_swap.payment_hash,
+                &receive_swap.payer_amount_sat,
+                &receive_swap.receiver_amount_sat,
+                &receive_swap.created_at,
+                &receive_swap.claim_fees_sat,
+                &receive_swap.mrh_address,
+                &receive_swap.mrh_script_pubkey,
+                &receive_swap.state,
+            ),
         )?;
-        let id_hash = sha256::Hash::hash(receive_swap.id.as_bytes()).to_hex();
-        _ = stmt.execute((
-            &receive_swap.id,
-            id_hash,
-            &receive_swap.preimage,
-            &receive_swap.create_response_json,
-            &receive_swap.claim_private_key,
-            &receive_swap.invoice,
-            &receive_swap.payment_hash,
-            &receive_swap.payer_amount_sat,
-            &receive_swap.receiver_amount_sat,
-            &receive_swap.created_at,
-            &receive_swap.claim_fees_sat,
-            &receive_swap.mrh_address,
-            &receive_swap.mrh_script_pubkey,
-            &receive_swap.state,
-        ))?;
 
-        con.execute(
+        tx.execute(
             "UPDATE receive_swaps
             SET
                 description = :description,
@@ -69,6 +71,10 @@ impl Persister {
             },
         )?;
 
+        Self::commit_outgoing(&tx, &receive_swap.id, RecordType::Receive, None)?;
+
+        tx.commit()?;
+
         Ok(())
     }
 
@@ -282,8 +288,10 @@ impl Persister {
         mrh_amount_sat: Option<u64>,
     ) -> Result<(), PaymentError> {
         // Do not overwrite claim_tx_id or lockup_tx_id
-        let con: Connection = self.get_connection()?;
-        con.execute(
+        let mut con = self.get_connection()?;
+        let tx = con.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
+
+        tx.execute(
             "UPDATE receive_swaps
             SET
                 claim_tx_id =
@@ -316,6 +324,12 @@ impl Persister {
             },
         )?;
 
+        // NOTE: Receive currently does not update any fields, bypassing the commit logic for now
+        // let updated_fields = None;
+        // Self::commit_outgoing(&tx, swap_id, RecordType::Receive, updated_fields)?;
+
+        tx.commit()?;
+
         Ok(())
     }
 }
diff --git a/lib/core/src/persist/send.rs b/lib/core/src/persist/send.rs
index 2cd3f123b..99ecb6dc1 100644
--- a/lib/core/src/persist/send.rs
+++ b/lib/core/src/persist/send.rs
@@ -6,16 +6,19 @@ use rusqlite::{named_params, params, Connection, Row};
 use sdk_common::bitcoin::hashes::{hex::ToHex, sha256, Hash};
 use serde::{Deserialize, Serialize};
 
-use crate::ensure_sdk;
 use crate::error::PaymentError;
 use crate::model::*;
 use crate::persist::{get_where_clause_state_in, Persister};
+use crate::sync::model::RecordType;
+use crate::{ensure_sdk, get_updated_fields};
 
 impl Persister {
     pub(crate) fn insert_send_swap(&self, send_swap: &SendSwap) -> Result<()> {
-        let con = self.get_connection()?;
+        let mut con = self.get_connection()?;
+        let tx = con.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
 
-        let mut stmt = con.prepare(
+        let id_hash = sha256::Hash::hash(send_swap.id.as_bytes()).to_hex();
+        tx.execute(
             "
             INSERT INTO send_swaps (
                 id,
@@ -34,24 +37,27 @@ impl Persister {
                 state
             )
             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            (
+                &send_swap.id,
+                &id_hash,
+                &send_swap.invoice,
+                &send_swap.bolt12_offer,
+                &send_swap.payment_hash,
+                &send_swap.description,
+                &send_swap.payer_amount_sat,
+                &send_swap.receiver_amount_sat,
+                &send_swap.create_response_json,
+                &send_swap.refund_private_key,
+                &send_swap.lockup_tx_id,
+                &send_swap.refund_tx_id,
+                &send_swap.created_at,
+                &send_swap.state,
+            ),
         )?;
-        let id_hash = sha256::Hash::hash(send_swap.id.as_bytes()).to_hex();
-        _ = stmt.execute((
-            &send_swap.id,
-            &id_hash,
-            &send_swap.invoice,
-            &send_swap.bolt12_offer,
-            &send_swap.payment_hash,
-            &send_swap.description,
-            &send_swap.payer_amount_sat,
-            &send_swap.receiver_amount_sat,
-            &send_swap.create_response_json,
-            &send_swap.refund_private_key,
-            &send_swap.lockup_tx_id,
-            &send_swap.refund_tx_id,
-            &send_swap.created_at,
-            &send_swap.state,
-        ))?;
+
+        Self::commit_outgoing(&tx, &send_swap.id, RecordType::Send, None)?;
+
+        tx.commit()?;
 
         Ok(())
     }
@@ -207,8 +213,10 @@ impl Persister {
         refund_tx_id: Option<&str>,
     ) -> Result<(), PaymentError> {
         // Do not overwrite preimage, lockup_tx_id, refund_tx_id
-        let con: Connection = self.get_connection()?;
-        con.execute(
+        let mut con = self.get_connection()?;
+        let tx = con.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
+
+        tx.execute(
             "UPDATE send_swaps
             SET
                 preimage =
@@ -241,6 +249,11 @@ impl Persister {
             },
         )?;
 
+        let updated_fields = get_updated_fields!(preimage);
+        Self::commit_outgoing(&tx, swap_id, RecordType::Send, updated_fields)?;
+
+        tx.commit()?;
+
         Ok(())
     }
 
diff --git a/lib/core/src/persist/sync.rs b/lib/core/src/persist/sync.rs
index 72a9d1724..f80e20dec 100644
--- a/lib/core/src/persist/sync.rs
+++ b/lib/core/src/persist/sync.rs
@@ -105,18 +105,19 @@ impl Persister {
         Ok(sync_settings)
     }
 
+    fn set_sync_setting_stmt(con: &Connection) -> rusqlite::Result<Statement> {
+        con.prepare("INSERT OR REPLACE INTO sync_settings(key, value) VALUES(:key, :value)")
+    }
+
     pub(crate) fn set_sync_settings(&self, map: HashMap<&'static str, String>) -> Result<()> {
         let mut con = self.get_connection()?;
         let tx = con.transaction_with_behavior(TransactionBehavior::Immediate)?;
 
         for (key, value) in map {
-            tx.execute(
-                "INSERT OR REPLACE INTO sync_settings(key, value) VALUES(:key, :value)",
-                named_params! {
-                    ":key": key,
-                    ":value": value,
-                },
-            )?;
+            Self::set_sync_setting_stmt(&tx)?.execute(named_params! {
+                ":key": key,
+                ":value": value,
+            })?;
         }
 
         tx.commit()?;
@@ -124,6 +125,52 @@ impl Persister {
         Ok(())
     }
 
+    pub(crate) fn set_new_remote(&self, remote_url: String) -> Result<()> {
+        let mut con = self.get_connection()?;
+        let tx = con.transaction_with_behavior(TransactionBehavior::Immediate)?;
+
+        tx.execute("DELETE FROM sync_state", [])?;
+        tx.execute("DELETE FROM sync_incoming", [])?;
+        tx.execute("DELETE FROM sync_outgoing", [])?;
+
+        let swap_tables = HashMap::from([
+            ("receive_swaps", RecordType::Receive),
+            ("send_swaps", RecordType::Send),
+            ("chain_swaps", RecordType::Chain),
+        ]);
+        for (table_name, record_type) in swap_tables {
+            let mut stmt = tx.prepare(&format!("SELECT id FROM {table_name}"))?;
+            let mut rows = stmt.query([])?;
+
+            while let Some(row) = rows.next()? {
+                let data_id: String = row.get(0)?;
+                let record_id = Record::get_id_from_record_type(record_type, &data_id);
+
+                tx.execute(
+                    "
+                    INSERT INTO sync_outgoing(record_id, data_id, record_type, commit_time)
+                    VALUES(:record_id, :data_id, :record_type, :commit_time)
+                ",
+                    named_params! {
+                        ":record_id": record_id,
+                        ":data_id": data_id,
+                        ":record_type": record_type,
+                        ":commit_time": utils::now(),
+                    },
+                )?;
+            }
+        }
+
+        Self::set_sync_setting_stmt(&tx)?.execute(named_params! {
+            ":key": "remote_url",
+            ":value": remote_url
+        })?;
+
+        tx.commit()?;
+
+        Ok(())
+    }
+
     pub(crate) fn get_incoming_records(&self) -> Result<Vec<Record>> {
         let con = self.get_connection()?;
 
diff --git a/lib/core/src/sync/mod.rs b/lib/core/src/sync/mod.rs
index 2eb80240d..a0710db87 100644
--- a/lib/core/src/sync/mod.rs
+++ b/lib/core/src/sync/mod.rs
@@ -1,18 +1,22 @@
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::time::Duration;
 
 use anyhow::{anyhow, Result};
+use futures_util::TryFutureExt;
+use tokio::sync::watch;
 
+use crate::sync::model::sync::{Record, SetRecordRequest, SetRecordStatus};
+use crate::utils;
 use crate::{persist::Persister, prelude::Signer};
 
 use self::client::SyncerClient;
-use self::model::sync::Record;
-use self::model::DecryptedRecord;
 use self::model::{
     data::{ChainSyncData, ReceiveSyncData, SendSyncData, SyncData},
     sync::ListChangesRequest,
     RecordType, SyncState,
 };
+use self::model::{DecryptedRecord, SyncOutgoingChanges};
 
 pub(crate) mod client;
 pub(crate) mod model;
@@ -39,6 +43,47 @@ impl SyncService {
         }
     }
 
+    fn check_remote_change(&self) -> Result<()> {
+        match self
+            .persister
+            .get_sync_settings()?
+            .remote_url
+            .is_some_and(|url| url == self.remote_url)
+        {
+            true => Ok(()),
+            false => self.persister.set_new_remote(self.remote_url.clone()),
+        }
+    }
+
+    pub(crate) async fn run(self: Arc<Self>, mut shutdown: watch::Receiver<()>) -> Result<()> {
+        self.client.connect(self.remote_url.clone()).await?;
+
+        self.check_remote_change()?;
+
+        tokio::spawn(async move {
+            let mut event_loop_interval = tokio::time::interval(Duration::from_secs(30));
+
+            loop {
+                tokio::select! {
+                    _ = event_loop_interval.tick() => {
+                        if let Err(err) = self.pull().and_then(|_| self.push()).await {
+                            log::debug!("Could not run sync event loop: {err:?}");
+                        }
+                    }
+                    _ = shutdown.changed() => {
+                        log::info!("Received shutdown signal, exiting realtime sync service loop");
+                        if let Err(err) = self.client.disconnect().await {
+                            log::debug!("Could not disconnect sync service client: {err:?}");
+                        };
+                        return;
+                    }
+                }
+            }
+        });
+
+        Ok(())
+    }
+
     fn commit_record(
         &self,
         decrypted_record: &DecryptedRecord,
@@ -210,18 +255,92 @@ impl SyncService {
 
         Ok(())
     }
+
+    async fn handle_push(
+        &self,
+        record_id: &str,
+        data_id: &str,
+        record_type: RecordType,
+    ) -> Result<()> {
+        log::info!("Handling push for record record_id {record_id} data_id {data_id}");
+
+        // Step 1: Get the sync state, if it exists, to compute the revision
+        let maybe_sync_state = self.persister.get_sync_state_by_record_id(record_id)?;
+        let record_revision = maybe_sync_state
+            .as_ref()
+            .map(|s| s.record_revision)
+            .unwrap_or(0);
+        let is_local = maybe_sync_state.map(|s| s.is_local).unwrap_or(true);
+
+        // Step 2: Fetch the sync data
+        let sync_data = self.load_sync_data(data_id, record_type)?;
+
+        // Step 3: Create the record to push outwards
+        let record = Record::new(sync_data, record_revision, self.signer.clone())?;
+
+        // Step 4: Push the record
+        let req = SetRecordRequest::new(record, utils::now(), self.signer.clone())?;
+        let reply = self.client.push(req).await?;
+
+        // Step 5: Check for conflict. If present, skip and retry on the next call
+        if reply.status() == SetRecordStatus::Conflict {
+            return Err(anyhow!(
+                "Got conflict status when attempting to push record"
+            ));
+        }
+
+        // Step 6: Set/update the state revision
+        self.persister.set_sync_state(SyncState {
+            data_id: data_id.to_string(),
+            record_id: record_id.to_string(),
+            record_revision: reply.new_revision,
+            is_local,
+        })?;
+
+        log::info!("Successfully pushed record record_id {record_id}");
+
+        Ok(())
+    }
+
+    async fn push(&self) -> Result<()> {
+        let outgoing_changes = self.persister.get_sync_outgoing_changes()?;
+
+        let mut succeded = vec![];
+        for SyncOutgoingChanges {
+            record_id,
+            data_id,
+            record_type,
+            ..
+        } in outgoing_changes
+        {
+            if let Err(err) = self.handle_push(&record_id, &data_id, record_type).await {
+                log::debug!("Could not handle push for record {record_id}: {err:?}");
+                continue;
+            }
+            succeded.push(record_id);
+        }
+
+        if !succeded.is_empty() {
+            self.persister.remove_sync_outgoing_changes(succeded)?;
+        }
+
+        Ok(())
+    }
 }
 
 #[cfg(test)]
 mod tests {
     use anyhow::{anyhow, Result};
-    use std::sync::Arc;
-    use tokio::sync::mpsc;
+    use std::{collections::HashMap, sync::Arc};
+    use tokio::sync::{mpsc, Mutex};
 
     use crate::{
-        prelude::Signer,
+        persist::Persister,
+        prelude::{Direction, PaymentState, Signer},
+        sync::model::SyncState,
         test_utils::{
-            persist::new_persister,
+            chain_swap::new_chain_swap,
+            persist::{new_persister, new_receive_swap, new_send_swap},
             sync::{
                 new_chain_sync_data, new_receive_sync_data, new_send_sync_data, MockSyncerClient,
             },
@@ -230,7 +349,7 @@ mod tests {
     };
 
     use super::{
-        model::{data::SyncData, sync::Record},
+        model::{data::SyncData, sync::Record, RecordType},
         SyncService,
     };
 
@@ -253,7 +372,8 @@ mod tests {
         ];
 
         let (incoming_tx, incoming_rx) = mpsc::channel::<Record>(10);
-        let client = Box::new(MockSyncerClient::new(incoming_rx));
+        let outgoing_records = Arc::new(Mutex::new(HashMap::new()));
+        let client = Box::new(MockSyncerClient::new(incoming_rx, outgoing_records.clone()));
         let sync_service =
             SyncService::new("".to_string(), persister.clone(), signer.clone(), client);
 
@@ -285,7 +405,6 @@ mod tests {
 
         let new_preimage = Some("preimage".to_string());
         let new_accept_zero_conf = false;
-        let new_server_lockup_tx_id = Some("server_lockup_tx_id".to_string());
         let sync_data = vec![
             SyncData::Send(new_send_sync_data(new_preimage.clone())),
             SyncData::Chain(new_chain_sync_data(Some(new_accept_zero_conf))),
@@ -308,11 +427,203 @@ mod tests {
         }
         if let Some(chain_swap) = persister.fetch_chain_swap_by_id(&sync_data[2].id())? {
             assert_eq!(chain_swap.accept_zero_conf, new_accept_zero_conf);
-            assert_eq!(chain_swap.server_lockup_tx_id, new_server_lockup_tx_id);
         } else {
             return Err(anyhow!("Chain swap not found"));
         }
 
         Ok(())
     }
+
+    fn get_outgoing_record<'a, 'b>(
+        persister: Arc<Persister>,
+        outgoing: &'a HashMap<String, Record>,
+        data_id: &'b str,
+        record_type: RecordType,
+    ) -> Result<&'a Record> {
+        let record_id = Record::get_id_from_record_type(record_type, data_id);
+        let sync_state = persister
+            .get_sync_state_by_record_id(&record_id)?
+            .ok_or(anyhow::anyhow!("Expected existing swap state"))?;
+        let Some(record) = outgoing.get(&sync_state.record_id) else {
+            return Err(anyhow::anyhow!(
+                "Expecting existing record in client's outgoing list"
+            ));
+        };
+        Ok(record)
+    }
+
+    #[tokio::test]
+    async fn test_outgoing_sync() -> Result<()> {
+        let (_temp_dir, persister) = new_persister()?;
+        let persister = Arc::new(persister);
+
+        let signer: Arc<Box<dyn Signer>> = Arc::new(Box::new(MockSigner::new()));
+
+        let (_incoming_tx, incoming_rx) = mpsc::channel::<Record>(10);
+        let outgoing_records = Arc::new(Mutex::new(HashMap::new()));
+        let client = Box::new(MockSyncerClient::new(incoming_rx, outgoing_records.clone()));
+        let sync_service =
+            SyncService::new("".to_string(), persister.clone(), signer.clone(), client);
+
+        // Test insert
+        persister.insert_receive_swap(&new_receive_swap(None))?;
+        persister.insert_send_swap(&new_send_swap(None))?;
+        persister.insert_chain_swap(&new_chain_swap(Direction::Incoming, None, true, None))?;
+
+        sync_service.push().await?;
+
+        let outgoing = outgoing_records.lock().await;
+        assert_eq!(outgoing.len(), 3);
+        drop(outgoing);
+
+        // Test conflict
+        let swap = new_receive_swap(None);
+        persister.insert_receive_swap(&swap)?;
+
+        sync_service.push().await?;
+
+        let outgoing = outgoing_records.lock().await;
+        assert_eq!(outgoing.len(), 4);
+        let record =
+            get_outgoing_record(persister.clone(), &outgoing, &swap.id, RecordType::Receive)?;
+        persister.set_sync_state(SyncState {
+            data_id: swap.id.clone(),
+            record_id: record.id.clone(),
+            record_revision: 90, // Set a wrong record revision
+            is_local: true,
+        })?;
+        drop(outgoing);
+
+        sync_service.push().await?;
+
+        let outgoing = outgoing_records.lock().await;
+        assert_eq!(outgoing.len(), 4); // No records were added
+        drop(outgoing);
+
+        // Test update before push
+        let swap = new_send_swap(None);
+        persister.insert_send_swap(&swap)?;
+        let new_preimage = Some("new-preimage");
+        persister.try_handle_send_swap_update(
+            &swap.id,
+            PaymentState::Pending,
+            new_preimage.clone(),
+            None,
+            None,
+        )?;
+
+        sync_service.push().await?;
+
+        let outgoing = outgoing_records.lock().await;
+
+        let record = get_outgoing_record(persister.clone(), &outgoing, &swap.id, RecordType::Send)?;
+        let decrypted_record = record.clone().decrypt(signer.clone())?;
+        assert_eq!(decrypted_record.data.id(), &swap.id);
+        match decrypted_record.data {
+            SyncData::Send(data) => {
+                assert_eq!(data.preimage, new_preimage.map(|p| p.to_string()));
+            }
+            _ => {
+                return Err(anyhow::anyhow!("Unexpected sync data type received."));
+            }
+        }
+        drop(outgoing);
+
+        // Test update after push
+        let swap = new_send_swap(None);
+        persister.insert_send_swap(&swap)?;
+
+        sync_service.push().await?;
+
+        let new_preimage = Some("new-preimage");
+        persister.try_handle_send_swap_update(
+            &swap.id,
+            PaymentState::Pending,
+            new_preimage.clone(),
+            None,
+            None,
+        )?;
+
+        sync_service.push().await?;
+
+        let outgoing = outgoing_records.lock().await;
+        let record = get_outgoing_record(persister.clone(), &outgoing, &swap.id, RecordType::Send)?;
+        let decrypted_record = record.clone().decrypt(signer.clone())?;
+        assert_eq!(decrypted_record.data.id(), &swap.id);
+        match decrypted_record.data {
+            SyncData::Send(data) => {
+                assert_eq!(data.preimage, new_preimage.map(|p| p.to_string()),);
+            }
+            _ => {
+                return Err(anyhow::anyhow!("Unexpected sync data type received."));
+            }
+        }
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_sync_clean() -> Result<()> {
+        let (_temp_dir, persister) = new_persister()?;
+        let persister = Arc::new(persister);
+
+        let signer: Arc<Box<dyn Signer>> = Arc::new(Box::new(MockSigner::new()));
+
+        let (incoming_tx, incoming_rx) = mpsc::channel::<Record>(10);
+        let outgoing_records = Arc::new(Mutex::new(HashMap::new()));
+        let client = Box::new(MockSyncerClient::new(incoming_rx, outgoing_records.clone()));
+        let sync_service =
+            SyncService::new("".to_string(), persister.clone(), signer.clone(), client);
+
+        // Clean incoming
+        let record = Record::new(
+            SyncData::Receive(new_receive_sync_data()),
+            1,
+            signer.clone(),
+        )?;
+        incoming_tx.send(record).await?;
+        sync_service.pull().await?;
+
+        let incoming_records = persister.get_incoming_records()?;
+        assert_eq!(incoming_records.len(), 0); // Records have been cleaned
+
+        let mut inapplicable_record = Record::new(
+            SyncData::Receive(new_receive_sync_data()),
+            2,
+            signer.clone(),
+        )?;
+        inapplicable_record.schema_version = "9.9.9".to_string();
+        incoming_tx.send(inapplicable_record).await?;
+        sync_service.pull().await?;
+
+        let incoming_records = persister.get_incoming_records()?;
+        assert_eq!(incoming_records.len(), 1); // Inapplicable records are stored for later
+
+        // Clean outgoing
+        let swap = new_send_swap(None);
+        persister.insert_send_swap(&swap)?;
+        let outgoing_changes = persister.get_sync_outgoing_changes()?;
+        assert_eq!(outgoing_changes.len(), 1); // Changes have been set
+
+        sync_service.push().await?;
+        let outgoing_changes = persister.get_sync_outgoing_changes()?;
+        assert_eq!(outgoing_changes.len(), 0); // Changes have been cleaned
+
+        let new_preimage = Some("new-preimage");
+        persister.try_handle_send_swap_update(
+            &swap.id,
+            PaymentState::Pending,
+            new_preimage.clone(),
+            None,
+            None,
+        )?;
+        let outgoing_changes = persister.get_sync_outgoing_changes()?;
+        assert_eq!(outgoing_changes.len(), 1); // Changes have been set
+
+        sync_service.push().await?;
+        let outgoing_changes = persister.get_sync_outgoing_changes()?;
+        assert_eq!(outgoing_changes.len(), 0); // Changes have been cleaned
+
+        Ok(())
+    }
 }
diff --git a/lib/core/src/sync/model/client.rs b/lib/core/src/sync/model/client.rs
index 84f053b90..897ff8109 100644
--- a/lib/core/src/sync/model/client.rs
+++ b/lib/core/src/sync/model/client.rs
@@ -1,4 +1,5 @@
 use anyhow::Result;
+use lwk_wollet::hashes::hex::DisplayHex as _;
 use openssl::sha::sha256;
 use std::sync::Arc;
 
@@ -7,7 +8,10 @@ use crate::{
     utils,
 };
 
-use super::{sync::ListChangesRequest, MESSAGE_PREFIX};
+use super::{
+    sync::{ListChangesRequest, Record, SetRecordRequest},
+    CURRENT_SCHEMA_VERSION, MESSAGE_PREFIX,
+};
 
 fn sign_message(msg: &[u8], signer: Arc<Box<dyn Signer>>) -> Result<String, SignerError> {
     let msg = [MESSAGE_PREFIX, msg].concat();
@@ -29,3 +33,25 @@ impl ListChangesRequest {
         })
     }
 }
+impl SetRecordRequest {
+    pub(crate) fn new(
+        record: Record,
+        request_time: u32,
+        signer: Arc<Box<dyn Signer>>,
+    ) -> Result<Self, SignerError> {
+        let msg = format!(
+            "{}-{}-{}-{}-{}",
+            record.id,
+            record.data.to_lower_hex_string(),
+            record.revision,
+            *CURRENT_SCHEMA_VERSION,
+            request_time,
+        );
+        let signature = sign_message(msg.as_bytes(), signer)?;
+        Ok(Self {
+            record: Some(record),
+            request_time,
+            signature,
+        })
+    }
+}
diff --git a/lib/core/src/sync/model/sync.rs b/lib/core/src/sync/model/sync.rs
index aee7f261c..2e707b828 100644
--- a/lib/core/src/sync/model/sync.rs
+++ b/lib/core/src/sync/model/sync.rs
@@ -80,10 +80,10 @@ pub mod syncer_client {
         dead_code,
         missing_docs,
         clippy::wildcard_imports,
-        clippy::let_unit_value,
+        clippy::let_unit_value
     )]
-    use tonic::codegen::*;
     use tonic::codegen::http::Uri;
+    use tonic::codegen::*;
     #[derive(Debug, Clone)]
     pub struct SyncerClient<T> {
         inner: tonic::client::Grpc<T>,
@@ -127,9 +127,8 @@ pub mod syncer_client {
                     <T as tonic::client::GrpcService<tonic::body::BoxBody>>::ResponseBody,
                 >,
             >,
-            <T as tonic::codegen::Service<
-                http::Request<tonic::body::BoxBody>,
-            >>::Error: Into<StdError> + std::marker::Send + std::marker::Sync,
+            <T as tonic::codegen::Service<http::Request<tonic::body::BoxBody>>>::Error:
+                Into<StdError> + std::marker::Send + std::marker::Sync,
         {
             SyncerClient::new(InterceptedService::new(inner, interceptor))
         }
@@ -168,39 +167,28 @@ pub mod syncer_client {
             &mut self,
             request: impl tonic::IntoRequest<super::SetRecordRequest>,
         ) -> std::result::Result<tonic::Response<super::SetRecordReply>, tonic::Status> {
-            self.inner
-                .ready()
-                .await
-                .map_err(|e| {
-                    tonic::Status::unknown(
-                        format!("Service was not ready: {}", e.into()),
-                    )
-                })?;
+            self.inner.ready().await.map_err(|e| {
+                tonic::Status::unknown(format!("Service was not ready: {}", e.into()))
+            })?;
             let codec = tonic::codec::ProstCodec::default();
             let path = http::uri::PathAndQuery::from_static("/sync.Syncer/SetRecord");
             let mut req = request.into_request();
-            req.extensions_mut().insert(GrpcMethod::new("sync.Syncer", "SetRecord"));
+            req.extensions_mut()
+                .insert(GrpcMethod::new("sync.Syncer", "SetRecord"));
             self.inner.unary(req, path, codec).await
         }
         pub async fn list_changes(
             &mut self,
             request: impl tonic::IntoRequest<super::ListChangesRequest>,
-        ) -> std::result::Result<
-            tonic::Response<super::ListChangesReply>,
-            tonic::Status,
-        > {
-            self.inner
-                .ready()
-                .await
-                .map_err(|e| {
-                    tonic::Status::unknown(
-                        format!("Service was not ready: {}", e.into()),
-                    )
-                })?;
+        ) -> std::result::Result<tonic::Response<super::ListChangesReply>, tonic::Status> {
+            self.inner.ready().await.map_err(|e| {
+                tonic::Status::unknown(format!("Service was not ready: {}", e.into()))
+            })?;
             let codec = tonic::codec::ProstCodec::default();
             let path = http::uri::PathAndQuery::from_static("/sync.Syncer/ListChanges");
             let mut req = request.into_request();
-            req.extensions_mut().insert(GrpcMethod::new("sync.Syncer", "ListChanges"));
+            req.extensions_mut()
+                .insert(GrpcMethod::new("sync.Syncer", "ListChanges"));
             self.inner.unary(req, path, codec).await
         }
         pub async fn track_changes(
@@ -210,18 +198,14 @@ pub mod syncer_client {
             tonic::Response<tonic::codec::Streaming<super::Record>>,
             tonic::Status,
         > {
-            self.inner
-                .ready()
-                .await
-                .map_err(|e| {
-                    tonic::Status::unknown(
-                        format!("Service was not ready: {}", e.into()),
-                    )
-                })?;
+            self.inner.ready().await.map_err(|e| {
+                tonic::Status::unknown(format!("Service was not ready: {}", e.into()))
+            })?;
             let codec = tonic::codec::ProstCodec::default();
             let path = http::uri::PathAndQuery::from_static("/sync.Syncer/TrackChanges");
             let mut req = request.into_request();
-            req.extensions_mut().insert(GrpcMethod::new("sync.Syncer", "TrackChanges"));
+            req.extensions_mut()
+                .insert(GrpcMethod::new("sync.Syncer", "TrackChanges"));
             self.inner.server_streaming(req, path, codec).await
         }
     }


Test patch:
diff --git a/lib/core/src/test_utils/sync.rs b/lib/core/src/test_utils/sync.rs
index aaa5d8027..a451ced46 100644
--- a/lib/core/src/test_utils/sync.rs
+++ b/lib/core/src/test_utils/sync.rs
@@ -1,5 +1,7 @@
 #![cfg(test)]
 
+use std::{collections::HashMap, sync::Arc};
+
 use crate::{
     prelude::Direction,
     sync::{
@@ -8,6 +10,7 @@ use crate::{
             data::{ChainSyncData, ReceiveSyncData, SendSyncData},
             sync::{
                 ListChangesReply, ListChangesRequest, Record, SetRecordReply, SetRecordRequest,
+                SetRecordStatus,
             },
         },
     },
@@ -18,12 +21,17 @@ use tokio::sync::{mpsc::Receiver, Mutex};
 
 pub(crate) struct MockSyncerClient {
     pub(crate) incoming_rx: Mutex<Receiver<Record>>,
+    pub(crate) outgoing_records: Arc<Mutex<HashMap<String, Record>>>,
 }
 
 impl MockSyncerClient {
-    pub(crate) fn new(incoming_rx: Receiver<Record>) -> Self {
+    pub(crate) fn new(
+        incoming_rx: Receiver<Record>,
+        outgoing_records: Arc<Mutex<HashMap<String, Record>>>,
+    ) -> Self {
         Self {
             incoming_rx: Mutex::new(incoming_rx),
+            outgoing_records,
         }
     }
 }
@@ -34,8 +42,30 @@ impl SyncerClient for MockSyncerClient {
         todo!()
     }
 
-    async fn push(&self, _req: SetRecordRequest) -> Result<SetRecordReply> {
-        todo!()
+    async fn push(&self, req: SetRecordRequest) -> Result<SetRecordReply> {
+        if let Some(mut record) = req.record {
+            let mut outgoing_records = self.outgoing_records.lock().await;
+
+            if let Some(existing_record) = outgoing_records.get(&record.id) {
+                if existing_record.revision != record.revision {
+                    return Ok(SetRecordReply {
+                        status: SetRecordStatus::Conflict as i32,
+                        new_revision: 0,
+                    });
+                }
+            }
+
+            record.revision = outgoing_records.len() as u64 + 1;
+            let record_revision = record.revision;
+
+            outgoing_records.insert(record.id.clone(), record);
+            return Ok(SetRecordReply {
+                status: SetRecordStatus::Success as i32,
+                new_revision: record_revision,
+            });
+        }
+
+        return Err(anyhow::anyhow!("No record was sent"));
     }
 
     async fn pull(&self, _req: ListChangesRequest) -> Result<ListChangesReply> {