feat: Update prover job ordering (#3769)

## What ❔

This is a rework of previous PR #3561
The difference is, that instead of ordering jobs by `created_at` field,
we are ordering them by new `batch_created_at` column, which is assigned
on saving witness inputs.

## Why ❔

<!-- Why are these changes done? What goal do they contribute to? What
are the principles behind them? -->
<!-- The `Why` has to be clear to non-Matter Labs entities running their
own ZK Chain -->
<!-- Example: PR templates ensure PR reviewers, observers, and future
iterators are in context about the evolution of repos. -->

## Is this a breaking change?
- [x] Yes
- [ ] No

## Operational changes
There is a need to run migrations on prover side

## Checklist

<!-- Check your PR fulfills the following items. -->
<!-- For draft PRs check the boxes as you complete them. -->

- [ ] PR title corresponds to the body of PR (we generate changelog
entries from PRs).
- [ ] Tests for the changes have been added / updated.
- [ ] Documentation comments have been added / updated.
- [ ] Code has been formatted via `zkstack dev fmt` and `zkstack dev
lint`.

Author: Artemka374

core/Cargo.lock

@@ -13,6 +13,7 @@ use crate::{
 pub struct FriProverJobMetadata {
     pub id: u32,
     pub block_number: L1BatchNumber,
+    pub batch_sealed_at: DateTime<Utc>,
     pub circuit_id: u8,
     pub aggregation_round: AggregationRound,
     pub sequence_number: usize,

core/lib/basic_types/src/prover_dal.rs

@@ -23,6 +23,7 @@ bellman.workspace = true
 serde.workspace = true
 serde_with = { workspace = true, features = ["base64", "hex"] }
 ciborium.workspace = true
+chrono.workspace = true
 
 [dev-dependencies]
 tokio = { workspace = true, features = ["full"] }

core/lib/prover_interface/Cargo.toml

@@ -19,6 +19,8 @@ use crate::{
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ProofGenerationData {
     pub l1_batch_number: L1BatchNumber,
+    #[serde(default)]
+    pub batch_sealed_at: chrono::DateTime<chrono::Utc>,
     pub witness_input_data: WitnessInputData,
     pub protocol_version: ProtocolSemanticVersion,
     pub l1_verifier_config: L1VerifierConfig,

core/lib/prover_interface/src/api.rs

@@ -184,8 +184,15 @@ impl Processor {
             },
         };
 
+        let batch_sealed_at = conn
+            .blocks_dal()
+            .get_batch_sealed_at(l1_batch_number)
+            .await?
+            .ok_or(ProcessorError::Internal)?;
+
         Ok(ProofGenerationData {
             l1_batch_number,
+            batch_sealed_at,
             witness_input_data: blob,
             protocol_version: protocol_version.version,
             l1_verifier_config: protocol_version.l1_verifier_config,

core/node/external_proof_integration_api/src/processor.rs

@@ -197,8 +197,17 @@ impl RequestProcessor {
 
         METRICS.observe_blob_sizes(&blob);
 
+        let batch_sealed_at = conn
+            .blocks_dal()
+            .get_batch_sealed_at(l1_batch_number)
+            .await?
+            .ok_or(RequestProcessorError::GeneralError(format!(
+                "Batch {l1_batch_number} not found in blocks_dal"
+            )))?;
+
         Ok(ProofGenerationData {
             l1_batch_number,
+            batch_sealed_at,
             witness_input_data: blob,
             protocol_version: protocol_version.version,
             l1_verifier_config: protocol_version.l1_verifier_config,

core/node/proof_data_handler/src/request_processor.rs

@@ -18,6 +18,7 @@ bincode.workspace = true
 clap = { workspace = true, features = ["derive", "string"] }
 once_cell.workspace = true
 regex.workspace = true
+chrono.workspace = true
 
 shivini = { workspace = true, features = ["circuit_definitions", "zksync"] }
 zksync_circuit_prover_service.workspace = true

prover/Cargo.lock

@@ -5,6 +5,7 @@ use std::{
 };
 
 use anyhow::Context;
+use chrono::{DateTime, Utc};
 use clap::Parser;
 use once_cell::sync::Lazy;
 use regex::Regex;
@@ -159,6 +160,7 @@ fn get_metadata(path: &Path) -> anyhow::Result<FriProverJobMetadata> {
         depth: caps["depth"].parse()?,
         is_node_final_proof: false,
         pick_time: Instant::now(),
+        batch_sealed_at: DateTime::<Utc>::default(),
     })
 }
 
@@ -172,6 +174,7 @@ fn witness_vector_filename(metadata: FriProverJobMetadata) -> String {
         depth,
         is_node_final_proof: _,
         pick_time: _,
+        batch_sealed_at: _,
     } = metadata;
     format!("{block_number}_{sequence_number}_{circuit_id}_{aggregation_round:?}_{depth}.witness_vector")
 }

prover/crates/bin/gpu_checker/Cargo.toml

@@ -1,4 +1,5 @@
 use anyhow::Context as _;
+use chrono::{DateTime, Utc};
 use clap::Args as ClapArgs;
 use zksync_basic_types::{
     protocol_version::{ProtocolSemanticVersion, ProtocolVersionId, VersionPatch},
@@ -36,6 +37,7 @@ pub async fn run(args: Args, config: ProverCLIConfig) -> anyhow::Result<()> {
             args.number,
             &format!("witness_inputs_{}", args.number.0),
             ProtocolSemanticVersion::new(protocol_version, protocol_version_patch),
+            DateTime::<Utc>::default(),
         )
         .await;