Add seq no to tx filters (#2273)

* Use seq-no for filtering

Signed-off-by: deniallugo <[email protected]>

* Fix add seq-no

Signed-off-by: deniallugo <[email protected]>

* Fix is priority

Signed-off-by: deniallugo <[email protected]>

* Add indexes

Signed-off-by: deniallugo <[email protected]>

Cargo.toml

@@ -9,6 +9,7 @@ members = [
     "core/bin/block_revert",
     "core/bin/remove_proofs",
     "core/bin/tree_cache_updater",
+    "core/bin/add_seq_no",
 
     # Server micro-services
     "core/bin/zksync_api",

core/bin/add_seq_no/Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "add_seq_no"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+zksync_storage = { path = "../../lib/storage", version = "1.0" }
+zksync_config = { path = "../../lib/config", version = "1.0" }
+
+tokio = { version = "1", features = ["full"] }
+anyhow = "1.0"

core/bin/add_seq_no/src/main.rs

@@ -0,0 +1,63 @@
+use std::time::Duration;
+use tokio::time::sleep;
+use zksync_storage::StorageProcessor;
+
+#[tokio::main]
+async fn main() {
+    let mut storage = StorageProcessor::establish_connection().await.unwrap();
+    let mut last_seq_no_executed_txs = storage
+        .chain()
+        .operations_ext_schema()
+        .get_last_seq_no()
+        .await;
+    println!("Last seq_no {}", last_seq_no_executed_txs);
+    let mut last_seq_no_priority_ops = last_seq_no_executed_txs;
+    let (_, mut max_seq_no) = storage
+        .chain()
+        .stats_schema()
+        .count_total_transactions((last_seq_no_executed_txs as u64).into())
+        .await
+        .unwrap();
+    let updated_tx_hashes = storage
+        .chain()
+        .operations_ext_schema()
+        .update_non_unique_tx_filters_for_priority_ops()
+        .await;
+
+    println!("Finish updating non unique tx_filters");
+    loop {
+        let mut transaction = storage.start_transaction().await.unwrap();
+        if last_seq_no_executed_txs < max_seq_no.0 as i64 {
+            dbg!((
+                last_seq_no_executed_txs,
+                last_seq_no_priority_ops,
+                max_seq_no
+            ));
+            last_seq_no_executed_txs = transaction
+                .chain()
+                .operations_ext_schema()
+                .set_seq_no_for_executed_txs(last_seq_no_executed_txs)
+                .await;
+            println!("Last seq_no {}", last_seq_no_executed_txs);
+            last_seq_no_priority_ops = transaction
+                .chain()
+                .operations_ext_schema()
+                .set_unique_sequence_number_for_priority_operations(
+                    last_seq_no_priority_ops,
+                    &updated_tx_hashes,
+                )
+                .await;
+            println!("Last seq_no priority {}", last_seq_no_priority_ops);
+            transaction.commit().await.unwrap()
+        } else {
+            sleep(Duration::from_secs(10)).await;
+            let (_, new_max_seq_no) = transaction
+                .chain()
+                .stats_schema()
+                .count_total_transactions((last_seq_no_executed_txs as u64).into())
+                .await
+                .unwrap();
+            max_seq_no = new_max_seq_no;
+        }
+    }
+}

core/lib/storage/migrations/2022-07-13-134305_add_seq_no_to_tx_filters/down.sql

@@ -0,0 +1,2 @@
+ALTER TABLE tx_filters DROP COLUMN sequence_number;
+ALTER TABLE tx_filters DROP COLUMN is_priority;

core/lib/storage/migrations/2022-07-13-134305_add_seq_no_to_tx_filters/up.sql

@@ -0,0 +1,13 @@
+ALTER TABLE tx_filters ADD COLUMN sequence_number BIGINT;
+ALTER TABLE tx_filters ADD COLUMN is_priority bool;
+
+CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS uq_executed_transactions_sequence_number ON public.executed_transactions USING btree (sequence_number);
+DROP INDEX IF EXISTS executed_transactions_sequence_number;
+DROP INDEX IF EXISTS ix_executed_transactions_tx_hash_sequence_number;
+
+CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS uq_executed_priority_operations_sequence_number ON public.executed_priority_operations USING btree (sequence_number);
+DROP INDEX IF EXISTS executed_priority_operations_sequence_number;
+DROP INDEX IF EXISTS ix_executed_priority_operations_tx_hash_sequence_number;
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_tx_filters_address_sequence_number ON public.tx_filters USING btree (address, sequence_number) include(is_priority);
+DROP INDEX IF EXISTS ix_tx_filters_address_tx_hash;