Merge #2333

2333: Fix triggers r=Deniallugo a=Deniallugo



Co-authored-by: Danil <[email protected]>

core/bin/tx_count_migration/src/main.rs

@@ -1,3 +1,4 @@
+use std::time::Duration;
 use zksync_storage::StorageProcessor;
 
 #[tokio::main]
@@ -13,16 +14,38 @@ async fn main() -> anyhow::Result<()> {
             .await
         {
             Some((start_account, final_account)) => {
-                transaction
-                    .chain()
-                    .operations_ext_schema()
-                    .update_txs_count(start_account, final_account)
-                    .await;
-                println!(
-                    "Data for accounts from {:?} to {:?} has been updated",
-                    &start_account, &final_account,
-                );
-                first_account = Some(final_account);
+                for i in 0..=10 {
+                    let mut tr = transaction.start_transaction().await?;
+                    match tr
+                        .chain()
+                        .operations_ext_schema()
+                        .update_txs_count(start_account, final_account)
+                        .await
+                    {
+                        Ok(_) => {
+                            println!(
+                                "Data for accounts from {:?} to {:?} has been updated",
+                                &start_account, &final_account,
+                            );
+                            first_account = Some(final_account);
+                            tr.commit().await?;
+                            break;
+                        }
+                        Err(err) => {
+                            let text = format!(
+                                "Error for accounts from {:?} to {:?} detected {:?}",
+                                &start_account, &final_account, err
+                            );
+                            if i != 10 {
+                                println!("{}", text);
+                            } else {
+                                panic!("{}", text);
+                            }
+                        }
+                    }
+
+                    tokio::time::sleep(Duration::from_secs(5)).await;
+                }
             }
             None => {
                 // We can forget about tx because we will close
@@ -31,6 +54,7 @@ async fn main() -> anyhow::Result<()> {
                 break;
             }
         }
+        tokio::time::sleep(Duration::from_secs(1)).await;
         transaction.commit().await?;
     }
 

core/lib/storage/migrations/2023-01-18-164812_txs_count/down.sql

@@ -1,7 +1,7 @@
-DROP TRIGGER increase_txs_count_for_address_tr ON tx_filters;
-DROP TRIGGER decrease_txs_count_for_address_tr ON tx_filters;
+DROP TRIGGER IF EXISTS increase_txs_count_for_address_tr ON tx_filters;
+DROP TRIGGER IF EXISTS decrease_txs_count_for_address_tr ON tx_filters;
 
-DROP FUNCTION increase_txs_count_for_address;
-DROP FUNCTION decrease_txs_count_for_address;
+DROP FUNCTION IF EXISTS increase_txs_count_for_address;
+DROP FUNCTION IF EXISTS decrease_txs_count_for_address;
 
 DROP TABLE txs_count;

core/lib/storage/migrations/2023-02-21-204507_fix-triggers/down.sql

@@ -0,0 +1,5 @@
+DROP TRIGGER IF EXISTS increase_txs_count_for_address_tr ON tx_filters;
+DROP TRIGGER IF EXISTS decrease_txs_count_for_address_tr ON tx_filters;
+
+DROP FUNCTION IF EXISTS increase_txs_count_for_address;
+DROP FUNCTION IF EXISTS decrease_txs_count_for_address;

core/lib/storage/migrations/2023-02-21-204507_fix-triggers/up.sql

@@ -0,0 +1,45 @@
+DROP TRIGGER IF EXISTS increase_txs_count_for_address_tr ON tx_filters;
+DROP TRIGGER IF EXISTS decrease_txs_count_for_address_tr ON tx_filters;
+
+CREATE OR REPLACE FUNCTION decrease_txs_count_for_address() RETURNS TRIGGER AS $$
+BEGIN
+    IF (TG_OP = 'DELETE') THEN
+        -- Postgresql doesn't support unique indexes for nullable fields, so we have to use
+        -- artificial token which means no token
+        UPDATE txs_count SET count = txs_count.count -
+                                     CASE WHEN (SELECT count(*) = 1  FROM tx_filters WHERE address = OLD.address AND tx_hash = OLD.tx_hash) THEN 1 ELSE 0 END
+        WHERE address=OLD.address AND token = -1;
+
+        UPDATE txs_count SET count = txs_count.count -
+                                     CASE WHEN (SELECT count(*) = 1 FROM tx_filters WHERE address = OLD.address AND tx_hash = OLD.tx_hash AND token = OLD.token) THEN 1 ELSE 0 END
+        WHERE address=OLD.address AND token=OLD.token;
+
+    END IF;
+    RETURN OLD;
+END;
+$$ LANGUAGE plpgsql;
+
+CREATE OR REPLACE FUNCTION increase_txs_count_for_address() RETURNS TRIGGER AS $$
+BEGIN
+    IF (TG_OP = 'INSERT') THEN
+        INSERT INTO txs_count (address, token, count) VALUES (NEW.address, -1       , 0) ON CONFLICT (address, token) DO NOTHING;
+        INSERT INTO txs_count (address, token, count) VALUES (NEW.address, NEW.token, 0) ON CONFLICT (address, token) DO NOTHING;
+
+        UPDATE txs_count SET count = txs_count.count +
+                                     CASE WHEN EXISTS(SELECT 1 FROM tx_filters WHERE address = NEW.address AND tx_hash = NEW.tx_hash FOR UPDATE) THEN 0 ELSE 1 END
+        WHERE address = NEW.address AND token = -1;
+
+        UPDATE txs_count SET count = txs_count.count +
+                                     CASE WHEN EXISTS(SELECT 1 FROM tx_filters WHERE address = NEW.address AND tx_hash = NEW.tx_hash AND token = NEW.token) THEN 0 ELSE 1 END
+        WHERE address = NEW.address AND token = NEW.token;
+
+    END IF;
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+CREATE TRIGGER increase_txs_count_for_address_tr BEFORE INSERT ON tx_filters
+    FOR EACH ROW  EXECUTE FUNCTION increase_txs_count_for_address();
+
+CREATE TRIGGER decrease_txs_count_for_address_tr BEFORE DELETE ON tx_filters
+    FOR EACH ROW  EXECUTE FUNCTION decrease_txs_count_for_address();

core/lib/storage/src/chain/operations_ext/mod.rs

@@ -1445,7 +1445,11 @@ impl<'a, 'c> OperationsExtSchema<'a, 'c> {
     }
 
     // TODO Remove it after migration is complete
-    pub async fn update_txs_count(&mut self, start_account: Address, finish_account: Address) {
+    pub async fn update_txs_count(
+        &mut self,
+        start_account: Address,
+        finish_account: Address,
+    ) -> QueryResult<()> {
         sqlx::query!(
             r#"
             INSERT INTO txs_count (address, token, count)
@@ -1459,8 +1463,7 @@ impl<'a, 'c> OperationsExtSchema<'a, 'c> {
             finish_account.as_bytes(),
         )
         .execute(self.0.conn())
-        .await
-        .unwrap();
+        .await?;
 
         sqlx::query!(
             r#"
@@ -1475,8 +1478,8 @@ impl<'a, 'c> OperationsExtSchema<'a, 'c> {
             finish_account.as_bytes(),
         )
         .execute(self.0.conn())
-        .await
-        .unwrap();
+        .await?;
+        Ok(())
     }
 
     pub async fn get_account_transactions_count(