Prepolish and test fixes

core/bin/zksync_server/src/node_builder.rs

@@ -421,7 +421,7 @@ impl MainNodeBuilder {
     fn add_house_keeper_layer(mut self) -> anyhow::Result<Self> {
         let house_keeper_config = try_load_config!(self.configs.house_keeper_config);
         let fri_prover_config = try_load_config!(self.configs.prover_config);
-        let fri_witness_generator_config = try_load_config!(self.configs.witness_generator);
+        let fri_witness_generator_config = try_load_config!(self.configs.witness_generator_config);
         let fri_prover_group_config = try_load_config!(self.configs.prover_group_config);
         let fri_proof_compressor_config = try_load_config!(self.configs.proof_compressor_config);
 

core/lib/basic_types/src/basic_fri_types.rs

@@ -2,11 +2,19 @@
 
 // TODO (PLA-773): Should be moved to the prover workspace.
 
-use std::{convert::TryFrom, str::FromStr};
+use std::{
+    collections::{hash_map::IntoIter, HashMap},
+    convert::TryFrom,
+    iter::once,
+    str::FromStr,
+};
 
 use serde::{Deserialize, Serialize};
 
-use crate::protocol_version::{ProtocolSemanticVersion, ProtocolVersionId, VersionPatch};
+use crate::{
+    protocol_version::{ProtocolSemanticVersion, ProtocolVersionId, VersionPatch},
+    prover_dal::JobCountStatistics,
+};
 
 const BLOB_CHUNK_SIZE: usize = 31;
 const ELEMENTS_PER_4844_BLOCK: usize = 4096;
@@ -136,6 +144,16 @@ impl AggregationRound {
             AggregationRound::Scheduler => None,
         }
     }
+
+    pub fn all_rounds() -> Vec<AggregationRound> {
+        vec![
+            AggregationRound::BasicCircuits,
+            AggregationRound::LeafAggregation,
+            AggregationRound::NodeAggregation,
+            AggregationRound::RecursionTip,
+            AggregationRound::Scheduler,
+        ]
+    }
 }
 
 impl std::fmt::Display for AggregationRound {
@@ -187,6 +205,118 @@ impl TryFrom<i32> for AggregationRound {
     }
 }
 
+/// TODO
+pub struct ProtocolVersionedCircuitProverStats {
+    protocol_versioned_circuit_stats: HashMap<ProtocolSemanticVersion, CircuitProverStats>,
+}
+
+impl FromIterator<CircuitProverStatsEntry> for ProtocolVersionedCircuitProverStats {
+    fn from_iter<I: IntoIterator<Item = CircuitProverStatsEntry>>(iter: I) -> Self {
+        let mut mapping = HashMap::new();
+        for entry in iter {
+            let protocol_semantic_version = entry.protocol_semantic_version;
+            let circuit_prover_stats: &mut CircuitProverStats =
+                mapping.entry(protocol_semantic_version).or_default();
+            circuit_prover_stats.add(entry.circuit_id_round_tuple, entry.job_count_statistics);
+        }
+        Self {
+            protocol_versioned_circuit_stats: mapping,
+        }
+    }
+}
+
+impl IntoIterator for ProtocolVersionedCircuitProverStats {
+    type Item = (ProtocolSemanticVersion, CircuitProverStats);
+    type IntoIter = IntoIter<ProtocolSemanticVersion, CircuitProverStats>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        self.protocol_versioned_circuit_stats.into_iter()
+    }
+}
+
+/// TODO
+pub struct CircuitProverStats {
+    circuits_prover_stats: HashMap<CircuitIdRoundTuple, JobCountStatistics>,
+}
+
+impl IntoIterator for CircuitProverStats {
+    type Item = (CircuitIdRoundTuple, JobCountStatistics);
+    type IntoIter = IntoIter<CircuitIdRoundTuple, JobCountStatistics>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        self.circuits_prover_stats.into_iter()
+    }
+}
+
+impl CircuitProverStats {
+    fn add(
+        &mut self,
+        circuit_id_round_tuple: CircuitIdRoundTuple,
+        job_count_statistics: JobCountStatistics,
+    ) {
+        let stats = self
+            .circuits_prover_stats
+            .entry(circuit_id_round_tuple)
+            .or_default();
+        stats.queued += job_count_statistics.queued;
+        stats.in_progress += job_count_statistics.in_progress;
+    }
+}
+
+impl Default for CircuitProverStats {
+    fn default() -> Self {
+        let mut stats = HashMap::new();
+        for circuit in (1..=15).chain(once(255)) {
+            stats.insert(
+                CircuitIdRoundTuple::new(circuit, 0),
+                JobCountStatistics::default(),
+            );
+        }
+        for circuit in 3..=18 {
+            stats.insert(
+                CircuitIdRoundTuple::new(circuit, 1),
+                JobCountStatistics::default(),
+            );
+        }
+        stats.insert(
+            CircuitIdRoundTuple::new(2, 2),
+            JobCountStatistics::default(),
+        );
+        stats.insert(
+            CircuitIdRoundTuple::new(255, 3),
+            JobCountStatistics::default(),
+        );
+        stats.insert(
+            CircuitIdRoundTuple::new(1, 4),
+            JobCountStatistics::default(),
+        );
+        Self {
+            circuits_prover_stats: stats,
+        }
+    }
+}
+
+/// TODO
+pub struct CircuitProverStatsEntry {
+    circuit_id_round_tuple: CircuitIdRoundTuple,
+    protocol_semantic_version: ProtocolSemanticVersion,
+    job_count_statistics: JobCountStatistics,
+}
+
+impl CircuitProverStatsEntry {
+    pub fn new(
+        circuit_id_round_tuple: CircuitIdRoundTuple,
+        protocol_semantic_version: ProtocolSemanticVersion,
+        job_count_statistics: JobCountStatistics,
+    ) -> Self {
+        CircuitProverStatsEntry {
+            circuit_id_round_tuple,
+            protocol_semantic_version,
+            job_count_statistics,
+        }
+    }
+}
+
 #[derive(Debug, Deserialize, Serialize, Clone, PartialEq, Eq, Hash)]
 pub struct JobIdentifiers {
     pub circuit_id: u8,

core/lib/basic_types/src/prover_dal.rs

@@ -52,6 +52,8 @@ pub struct StuckJobs {
     pub status: String,
     pub attempts: u64,
     pub circuit_id: Option<u32>,
+    pub picked_by: Option<String>,
+    pub error: Option<String>,
 }
 
 // TODO (PLA-774): Redundant structure, should be replaced with `std::net::SocketAddr`.

core/lib/config/src/configs/general.rs

@@ -34,7 +34,7 @@ pub struct GeneralConfig {
     pub prover_gateway: Option<FriProverGatewayConfig>,
     pub witness_vector_generator: Option<FriWitnessVectorGeneratorConfig>,
     pub prover_group_config: Option<FriProverGroupConfig>,
-    pub witness_generator: Option<FriWitnessGeneratorConfig>,
+    pub witness_generator_config: Option<FriWitnessGeneratorConfig>,
     pub prometheus_config: Option<PrometheusConfig>,
     pub proof_data_handler_config: Option<ProofDataHandlerConfig>,
     pub db_config: Option<DBConfig>,

core/lib/config/src/configs/prover_job_monitor.rs

@@ -5,61 +5,15 @@ pub struct ProverJobMonitorConfig {
     pub prometheus_port: u16,
     pub max_db_connections: u32,
     pub graceful_shutdown_timeout_ms: u32,
+    pub gpu_prover_archiver_run_interval_ms: u64,
+    pub gpu_prover_archiver_archive_prover_after_secs: u64,
+    pub prover_jobs_archiver_run_interval_ms: u64,
+    pub prover_jobs_archiver_archive_jobs_after_secs: u64,
+    pub proof_compressor_job_requeuer_run_interval_ms: u64,
+    pub prover_job_requeuer_run_interval_ms: u64,
+    pub witness_generator_job_requeuer_run_interval_ms: u64,
+    pub proof_compressor_queue_reporter_run_interval_ms: u64,
+    pub prover_queue_reporter_run_interval_ms: u64,
+    pub witness_generator_queue_reporter_run_interval_ms: u64,
+    pub witness_job_queuer_run_interval_ms: u64,
 }
-
-// use serde::{Deserialize, Serialize};
-// use zksync_basic_types::{
-//     commitment::L1BatchCommitmentMode,
-//     protocol_version::{ProtocolSemanticVersion, ProtocolVersionId},
-//     Address, L1ChainId, L2ChainId, SLChainId, H256,
-// };
-//
-// /// This config represents the genesis state of the chain.
-// /// Each chain has this config immutable and we update it only during the protocol upgrade
-// #[derive(Debug, Serialize, Deserialize, Clone, PartialEq)]
-// pub struct GenesisConfig {
-//     // TODO make fields non optional, once we fully moved to file based configs.
-//     // Now for backward compatibility we keep it optional
-//     pub protocol_version: Option<ProtocolSemanticVersion>,
-//     pub genesis_root_hash: Option<H256>,
-//     pub rollup_last_leaf_index: Option<u64>,
-//     pub genesis_commitment: Option<H256>,
-//     pub bootloader_hash: Option<H256>,
-//     pub default_aa_hash: Option<H256>,
-//     pub l1_chain_id: L1ChainId,
-//     pub sl_chain_id: Option<SLChainId>,
-//     pub l2_chain_id: L2ChainId,
-//     pub recursion_scheduler_level_vk_hash: H256,
-//     pub fee_account: Address,
-//     pub dummy_verifier: bool,
-//     pub l1_batch_commit_data_generator_mode: L1BatchCommitmentMode,
-// }
-//
-// impl GenesisConfig {
-//     pub fn settlement_layer_id(&self) -> SLChainId {
-//         self.sl_chain_id.unwrap_or(self.l1_chain_id.into())
-//     }
-// }
-//
-// impl GenesisConfig {
-//     pub fn for_tests() -> Self {
-//         GenesisConfig {
-//             genesis_root_hash: Some(H256::repeat_byte(0x01)),
-//             rollup_last_leaf_index: Some(26),
-//             recursion_scheduler_level_vk_hash: H256::repeat_byte(0x02),
-//             fee_account: Default::default(),
-//             genesis_commitment: Some(H256::repeat_byte(0x17)),
-//             bootloader_hash: Default::default(),
-//             default_aa_hash: Default::default(),
-//             l1_chain_id: L1ChainId(9),
-//             sl_chain_id: None,
-//             protocol_version: Some(ProtocolSemanticVersion {
-//                 minor: ProtocolVersionId::latest(),
-//                 patch: 0.into(),
-//             }),
-//             l2_chain_id: L2ChainId::default(),
-//             dummy_verifier: false,
-//             l1_batch_commit_data_generator_mode: L1BatchCommitmentMode::Rollup,
-//         }
-//     }
-// }

core/lib/config/src/testonly.rs

@@ -1020,6 +1020,17 @@ impl Distribution<configs::prover_job_monitor::ProverJobMonitorConfig> for Encod
             prometheus_port: self.sample(rng),
             max_db_connections: self.sample(rng),
             graceful_shutdown_timeout_ms: self.sample(rng),
+            gpu_prover_archiver_run_interval_ms: self.sample(rng),
+            gpu_prover_archiver_archive_prover_after_secs: self.sample(rng),
+            prover_jobs_archiver_run_interval_ms: self.sample(rng),
+            prover_jobs_archiver_archive_jobs_after_secs: self.sample(rng),
+            proof_compressor_job_requeuer_run_interval_ms: self.sample(rng),
+            prover_job_requeuer_run_interval_ms: self.sample(rng),
+            witness_generator_job_requeuer_run_interval_ms: self.sample(rng),
+            proof_compressor_queue_reporter_run_interval_ms: self.sample(rng),
+            prover_queue_reporter_run_interval_ms: self.sample(rng),
+            witness_generator_queue_reporter_run_interval_ms: self.sample(rng),
+            witness_job_queuer_run_interval_ms: self.sample(rng),
         }
     }
 }
@@ -1040,7 +1051,7 @@ impl Distribution<configs::GeneralConfig> for EncodeDist {
             prover_gateway: self.sample(rng),
             witness_vector_generator: self.sample(rng),
             prover_group_config: self.sample(rng),
-            witness_generator: self.sample(rng),
+            witness_generator_config: self.sample(rng),
             prometheus_config: self.sample(rng),
             proof_data_handler_config: self.sample(rng),
             db_config: self.sample(rng),

core/lib/protobuf_config/src/general.rs

@@ -22,7 +22,7 @@ impl ProtoRepr for proto::GeneralConfig {
             prover_group_config: read_optional_repr(&self.prover_group),
             prometheus_config: read_optional_repr(&self.prometheus),
             proof_data_handler_config: read_optional_repr(&self.data_handler),
-            witness_generator: read_optional_repr(&self.witness_generator),
+            witness_generator_config: read_optional_repr(&self.witness_generator),
             api_config: read_optional_repr(&self.api),
             db_config: read_optional_repr(&self.db),
             eth: read_optional_repr(&self.eth),
@@ -59,7 +59,7 @@ impl ProtoRepr for proto::GeneralConfig {
             proof_compressor: this.proof_compressor_config.as_ref().map(ProtoRepr::build),
             prover: this.prover_config.as_ref().map(ProtoRepr::build),
             prover_group: this.prover_group_config.as_ref().map(ProtoRepr::build),
-            witness_generator: this.witness_generator.as_ref().map(ProtoRepr::build),
+            witness_generator: this.witness_generator_config.as_ref().map(ProtoRepr::build),
             prover_gateway: this.prover_gateway.as_ref().map(ProtoRepr::build),
             witness_vector_generator: this.witness_vector_generator.as_ref().map(ProtoRepr::build),
             prometheus: this.prometheus_config.as_ref().map(ProtoRepr::build),

core/lib/protobuf_config/src/proto/config/prover_job_monitor.proto

@@ -6,4 +6,15 @@ message ProverJobMonitor {
   optional uint32 prometheus_port = 1; // required; u32
   optional uint32 max_db_connections = 2; // required; u32
   optional uint32 graceful_shutdown_timeout_ms = 3; // required; u32
+  optional uint64 gpu_prover_archiver_run_interval_ms = 4; // required; u64
+  optional uint64 gpu_prover_archiver_archive_prover_after_secs = 5; // required; u64
+  optional uint64 prover_jobs_archiver_run_interval_ms = 6; // required; u64
+  optional uint64 prover_jobs_archiver_archive_jobs_after_secs = 7; // required; u64
+  optional uint64 proof_compressor_job_requeuer_run_interval_ms = 8; // required; u64
+  optional uint64 prover_job_requeuer_run_interval_ms = 9; // required; u64
+  optional uint64 witness_generator_job_requeuer_run_interval_ms = 10; // required; u64
+  optional uint64 proof_compressor_queue_reporter_run_interval_ms = 11; // required; u64
+  optional uint64 prover_queue_reporter_run_interval_ms = 12; // required; u64
+  optional uint64 witness_generator_queue_reporter_run_interval_ms = 13; // required; u64
+  optional uint64 witness_job_queuer_run_interval_ms = 14; // required; u64
 }