bug: validators removed on unbond

chain/src/main.rs

@@ -185,17 +185,18 @@ async fn crawling_fn(
     let bonds_updates = bonds
         .iter()
         .cloned()
-        .filter_map(|(_, bond)| bond)
+        .filter_map(|(_, _, bond)| bond)
         .collect::<Vec<_>>();
 
     let removed_bonds_addresses = bonds
         .iter()
         .cloned()
-        .filter_map(|(addr, bond)| match bond {
+        .filter_map(|(source, validator, bond)| match bond {
             Some(_) => None,
-            None => Some(addr),
+            None => Some((source, validator)),
         })
-        .collect::<Vec<Id>>();
+        .collect::<Vec<(Id, Id)>>();
+
     let addresses = block.unbond_addresses();
     let unbonds = namada_service::query_unbonds(&client, addresses)
         .await

chain/src/repository/pos.rs

@@ -18,15 +18,21 @@ use shared::validator::ValidatorMetadataChange;
 
 pub fn clear_bonds(
     transaction_conn: &mut PgConnection,
-    addresses: Vec<Id>,
+    addresses: Vec<(Id, Id)>,
 ) -> anyhow::Result<()> {
-    let addresses = addresses
+    let (sources, validators): (Vec<String>, Vec<String>) = addresses
         .into_iter()
-        .map(|a| a.to_string())
-        .collect::<Vec<_>>();
+        .map(|(source, validator)| (source.to_string(), validator.to_string()))
+        .unzip();
 
     diesel::delete(bonds::table)
-        .filter(bonds::address.eq_any(addresses))
+        .filter(bonds::address.eq_any(sources).and(
+            bonds::validator_id.eq_any(
+                validators::table.select(validators::columns::id).filter(
+                    validators::columns::namada_address.eq_any(validators),
+                ),
+            ),
+        ))
         .execute(transaction_conn)
         .context("Failed to remove bonds from db")?;
 

chain/src/services/namada.rs

@@ -329,14 +329,14 @@ pub async fn query_next_governance_id(
 pub async fn query_bonds(
     client: &HttpClient,
     addresses: HashSet<BondAddresses>,
-) -> anyhow::Result<Vec<(Id, Option<Bond>)>> {
+) -> anyhow::Result<Vec<(Id, Id, Option<Bond>)>> {
     let nested_bonds = futures::stream::iter(addresses)
         .filter_map(|BondAddresses { source, target }| async move {
             // TODO: if this is too slow do not use query_all_bonds_and_unbonds
             let (bonds_res, _) = query_all_bonds_and_unbonds(
                 client,
                 Some(source.clone()),
-                Some(target),
+                Some(target.clone()),
             )
             .await
             .context("Failed to query all bonds and unbonds")
@@ -345,10 +345,10 @@ pub async fn query_bonds(
             let bonds = if !bonds_res.is_empty() {
                 bonds_res
                     .into_iter()
-                    .map(|bond| (source.clone(), Some(bond)))
+                    .map(|bond| (source.clone(), target.clone(), Some(bond)))
                     .collect::<Vec<_>>()
             } else {
-                vec![(source, None)]
+                vec![(source, target, None)]
             };
 
             Some(bonds)