bug: bonds do not clear when unbonding whole amount

chain/src/main.rs

@@ -27,6 +27,7 @@ use shared::checksums::Checksums;
 use shared::crawler::crawl;
 use shared::crawler_state::ChainCrawlerState;
 use shared::error::{AsDbError, AsRpcError, ContextDbInteractError, MainError};
+use shared::id::Id;
 use tendermint_rpc::HttpClient;
 use tokio::time::sleep;
 use tracing::Level;
@@ -181,6 +182,20 @@ async fn crawling_fn(
     let bonds = query_bonds(&client, addresses).await.into_rpc_error()?;
     tracing::info!("Updating bonds for {} addresses", bonds.len());
 
+    let bonds_updates = bonds
+        .iter()
+        .cloned()
+        .filter_map(|(_, bond)| bond)
+        .collect::<Vec<_>>();
+
+    let removed_bonds_addresses = bonds
+        .iter()
+        .cloned()
+        .filter_map(|(addr, bond)| match bond {
+            Some(_) => None,
+            None => Some(addr),
+        })
+        .collect::<Vec<Id>>();
     let addresses = block.unbond_addresses();
     let unbonds = namada_service::query_unbonds(&client, addresses)
         .await
@@ -226,7 +241,14 @@ async fn crawling_fn(
                     proposals_votes,
                 )?;
 
-                repository::pos::insert_bonds(transaction_conn, bonds)?;
+                // We first remove all the bonds that are not in the storage
+                // anymore
+                repository::pos::clear_bonds(
+                    transaction_conn,
+                    removed_bonds_addresses,
+                )?;
+                repository::pos::insert_bonds(transaction_conn, bonds_updates)?;
+
                 repository::pos::insert_unbonds(transaction_conn, unbonds)?;
                 repository::pos::remove_withdraws(
                     transaction_conn,

chain/src/repository/pos.rs

@@ -16,6 +16,23 @@ use shared::id::Id;
 use shared::unbond::{UnbondAddresses, Unbonds};
 use shared::validator::ValidatorMetadataChange;
 
+pub fn clear_bonds(
+    transaction_conn: &mut PgConnection,
+    addresses: Vec<Id>,
+) -> anyhow::Result<()> {
+    let addresses = addresses
+        .into_iter()
+        .map(|a| a.to_string())
+        .collect::<Vec<_>>();
+
+    diesel::delete(bonds::table)
+        .filter(bonds::address.eq_any(addresses))
+        .execute(transaction_conn)
+        .context("Failed to remove bonds from db")?;
+
+    anyhow::Ok(())
+}
+
 pub fn insert_bonds(
     transaction_conn: &mut PgConnection,
     bonds: Bonds,

chain/src/services/namada.rs

@@ -329,15 +329,27 @@ pub async fn query_next_governance_id(
 pub async fn query_bonds(
     client: &HttpClient,
     addresses: HashSet<BondAddresses>,
-) -> anyhow::Result<Bonds> {
+) -> anyhow::Result<Vec<(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, _) =
-                query_all_bonds_and_unbonds(client, Some(source), Some(target))
-                    .await
-                    .context("Failed to query all bonds and unbonds")
-                    .ok()?;
+            let (bonds_res, _) = query_all_bonds_and_unbonds(
+                client,
+                Some(source.clone()),
+                Some(target),
+            )
+            .await
+            .context("Failed to query all bonds and unbonds")
+            .ok()?;
+
+            let bonds = if !bonds_res.is_empty() {
+                bonds_res
+                    .into_iter()
+                    .map(|bond| (source.clone(), Some(bond)))
+                    .collect::<Vec<_>>()
+            } else {
+                vec![(source, None)]
+            };
 
             Some(bonds)
         })