feat: Adding EN snapshots applier (#882)

Parts of this component that I decided to do in separate PRs:
-resilience

Cargo.toml

@@ -38,6 +38,7 @@ members = [
     "core/lib/multivm",
     "core/lib/vm_utils",
     "core/lib/web3_decl",
+    "core/lib/snapshots_applier",
 
     # Test infrastructure
     "core/tests/test_account",

core/bin/snapshots_creator/src/tests.rs

@@ -205,7 +205,8 @@ async fn prepare_postgres(
         let factory_deps = gen_factory_deps(rng, 10);
         conn.storage_dal()
             .insert_factory_deps(MiniblockNumber(block_number), &factory_deps)
-            .await;
+            .await
+            .unwrap();
 
         // Since we generate `logs` randomly, all of them are written the first time.
         create_l1_batch(conn, L1BatchNumber(block_number), &logs).await;

core/lib/dal/migrations/20240112194527_snapshots_applier_processed_chunks_format_change.down.sql

@@ -0,0 +1,11 @@
+ALTER TABLE snapshot_recovery ADD COLUMN last_finished_chunk_id NOT NULL;
+ALTER TABLE snapshot_recovery ADD COLUMN total_chunk_count NOT NULL;
+
+ALTER TABLE snapshot_recovery DROP COLUMN storage_logs_chunks_processed;
+
+ALTER TABLE factory_deps ADD CONSTRAINT factory_deps_miniblock_number_fkey
+    FOREIGN KEY (miniblock_number) REFERENCES miniblocks (number);
+ALTER TABLE initial_writes ADD CONSTRAINT initial_writes_l1_batch_number_fkey
+    FOREIGN KEY (l1_batch_number) REFERENCES l1_batches (number);
+ALTER TABLE storage_logs ADD CONSTRAINT storage_logs_miniblock_number_fkey
+    FOREIGN KEY (miniblock_number) REFERENCES miniblocks (number);

core/lib/dal/migrations/20240112194527_snapshots_applier_processed_chunks_format_change.up.sql

@@ -0,0 +1,8 @@
+ALTER TABLE snapshot_recovery DROP COLUMN last_finished_chunk_id;
+ALTER TABLE snapshot_recovery DROP COLUMN total_chunk_count;
+
+ALTER TABLE snapshot_recovery ADD COLUMN storage_logs_chunks_processed BOOL[] NOT NULL;
+
+ALTER TABLE factory_deps DROP CONSTRAINT factory_deps_miniblock_number_fkey;
+ALTER TABLE initial_writes DROP CONSTRAINT initial_writes_l1_batch_number_fkey;
+ALTER TABLE storage_logs DROP CONSTRAINT storage_logs_miniblock_number_fkey;

core/lib/dal/src/blocks_dal.rs

@@ -1812,6 +1812,31 @@ impl BlocksDal<'_, '_> {
         .collect())
     }
 
+    pub async fn delete_initial_writes(
+        &mut self,
+        last_batch_to_keep: L1BatchNumber,
+    ) -> sqlx::Result<()> {
+        self.delete_initial_writes_inner(Some(last_batch_to_keep))
+            .await
+    }
+
+    pub async fn delete_initial_writes_inner(
+        &mut self,
+        last_batch_to_keep: Option<L1BatchNumber>,
+    ) -> sqlx::Result<()> {
+        let block_number = last_batch_to_keep.map_or(-1, |number| number.0 as i64);
+        sqlx::query!(
+            r#"
+            DELETE FROM initial_writes
+            WHERE
+                l1_batch_number > $1
+            "#,
+            block_number
+        )
+        .execute(self.storage.conn())
+        .await?;
+        Ok(())
+    }
     /// Deletes all L1 batches from the storage so that the specified batch number is the last one left.
     pub async fn delete_l1_batches(
         &mut self,
@@ -2181,6 +2206,9 @@ impl BlocksDal<'_, '_> {
         self.delete_l1_batches_inner(None)
             .await
             .context("delete_l1_batches_inner()")?;
+        self.delete_initial_writes_inner(None)
+            .await
+            .context("delete_initial_writes_inner()")?;
         Ok(())
     }
 }
@@ -2204,6 +2232,10 @@ mod tests {
             .delete_l1_batches(L1BatchNumber(0))
             .await
             .unwrap();
+        conn.blocks_dal()
+            .delete_initial_writes(L1BatchNumber(0))
+            .await
+            .unwrap();
         conn.protocol_versions_dal()
             .save_protocol_version_with_tx(ProtocolVersion::default())
             .await;
@@ -2265,6 +2297,10 @@ mod tests {
             .delete_l1_batches(L1BatchNumber(0))
             .await
             .unwrap();
+        conn.blocks_dal()
+            .delete_initial_writes(L1BatchNumber(0))
+            .await
+            .unwrap();
         conn.protocol_versions_dal()
             .save_protocol_version_with_tx(ProtocolVersion::default())
             .await;

core/lib/dal/src/blocks_web3_dal.rs

@@ -767,11 +767,10 @@ mod tests {
             l1_batch_root_hash: H256::zero(),
             miniblock_number: MiniblockNumber(42),
             miniblock_root_hash: H256::zero(),
-            last_finished_chunk_id: None,
-            total_chunk_count: 100,
+            storage_logs_chunks_processed: vec![true; 100],
         };
         conn.snapshot_recovery_dal()
-            .set_applied_snapshot_status(&snapshot_recovery)
+            .insert_initial_recovery_status(&snapshot_recovery)
             .await
             .unwrap();