feat: L1 batch QC database (BFT-476) (matter-labs#2340)

## What ❔

- [x] Add an `l1_batches_consensus` table to hold [L1 batch Quorum
Certificates](https://github.com/matter-labs/era-consensus/blob/177881457f392fca990dbb3df1695737d90fd0c7/node/libs/roles/src/attester/messages/batch.rs#L67)
from Attesters
- [x] Add attesters to the config
- [x] Implement methods in `PersistentBatchStore` 
  - [x] `persisted`
  - [x] `last_batch`
  - [x] `last_batch_qc`
  - [x] `get_batch`
  - [x] `get_batch_qc`
  - [x] `store_qc`
  - [ ] `queue_next_batch` - _not going to implement for now_
  - [ ] assign `SyncBatch::proof` - _not going to implement for now_
- [x] Add tests for all new methods in `ConsensusDal` and the
`PersistentBatchStore`

### Caveat

Implemented the updating of `persisted` with a loop that polls the
database for newly available `SyncBatch` records, even if they have no
proof. This inevitably triggers the gossiping of batch statuses and the
pulling of `SyncBatch` between peers. For this reason `queue_next_batch`
just drop the data, since we can't do anything with it without the proof
yet. Returning an error or panicking would stop the consensus tasks.

I ended up disabling the `persisted` by leaving its dummy implementation
in place because when enabled the full node tests keep going on forever,
printing the following logs in a loop:

```console
❯ RUST_LOG=info zk test rust test_full_nodes --no-capture
...
2024-07-03T14:22:57.882784Z  INFO in{addr=[::1]:53082}: zksync_consensus_network: 191: new connection
2024-07-03T14:22:57.883457Z  INFO in{addr=[::1]:53082}:gossip: zksync_consensus_network::gossip::runner: 383: peer = node:public:ed25519:068ffa0b3fedbbe5c2a6da3defd26e0d084248f12bfe98db85f7785b0b08b63e
2024-07-03T14:22:57.883764Z  INFO out{addr="[::1]:52998"}:gossip: zksync_consensus_network::gossip::runner: 416: peer = node:public:ed25519:7710ed90aad9f5859dfba06e13fb4e6fb0fe4d686f81f9d819464ad1fdc371bd
2024-07-03T14:22:57.886204Z  INFO in{addr=[::1]:53082}:gossip: zksync_consensus_network::rpc: 222: message too large: max = 10240B, got 13773B
2024-07-03T14:22:57.886280Z  INFO out{addr="[::1]:52998"}:gossip: zksync_consensus_network::rpc: 222: message too large: max = 10240B, got 13773B
2024-07-03T14:22:57.886633Z  INFO in{addr=[::1]:53082}:gossip: zksync_consensus_network::rpc: 222: canceled
...
2024-07-03T14:22:57.888143Z  INFO out{addr="[::1]:52998"}:gossip: zksync_consensus_network::rpc: 222: disconnected
...
2024-07-03T14:22:57.888390Z  INFO zksync_consensus_network: 216: [::1]:53082: gossip.run_inbound_stream(): push_batch_store_state.: end of stream
2024-07-03T14:22:57.888446Z  INFO zksync_consensus_network: 158: gossip.run_outbound_stream("[::1]:52998"): push_batch_store_state.: end of stream
```

So in the tests the message size exceeds the maximum. I think it's
[hardcoded
here](https://github.com/matter-labs/era-consensus/blob/decb988eb9e1a45fd5171d2cc540a360d9ca5f1f/node/actors/network/src/gossip/runner.rs#L109).
Since this functionality isn't expected to work, I think we can disable
it for now.

## Why ❔

The workflow of signing and submitting L1 batch certificates will be
like this:
1. Data is inserted into the `l1_batches` table. 
2. If the node is one of the Attesters it picks up the batch, signs and
sends it to the gossip layer via
matter-labs/era-consensus#137
3. The consensus collects votes about the L1 batch, and when the
threshold is reached it saves the quorum certificate into Postgres
4. The node monitors Main Node (later L1) for new batch QCs and upserts
them into the database (the QC can be different than what a particular
node inserted based on gossip). This way a node which has been down for
a period of time can backfill any QCs it missed. It is assumed that the
Main Node API only serves QCs that have no gaps following them, ie. they
are final - if it was L1 it wouldn't allow submissions with gaps, and
this simulates that semantic.
5. The last height that doesn't have any gaps following it is used as a
floor for what needs to be (re)signed and gossiped

This PR supports the above workflow up to step 3.

## Checklist

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

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

---------

Co-authored-by: Bruno França <[email protected]>

Cargo.toml

@@ -283,4 +283,4 @@ zksync_node_consensus = { path = "core/node/consensus" }
 zksync_contract_verification_server = { path = "core/node/contract_verification_server" }
 zksync_node_api_server = { path = "core/node/api_server" }
 zksync_tee_verifier_input_producer = { path = "core/node/tee_verifier_input_producer" }
-zksync_base_token_adjuster = {path = "core/node/base_token_adjuster"}
+zksync_base_token_adjuster = { path = "core/node/base_token_adjuster" }

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

@@ -12,6 +12,14 @@ pub struct ValidatorPublicKey(pub String);
 #[derive(Debug, Clone)]
 pub struct ValidatorSecretKey(pub Secret<String>);
 
+/// `zksync_consensus_crypto::TextFmt` representation of `zksync_consensus_roles::attester::PublicKey`.
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct AttesterPublicKey(pub String);
+
+/// `zksync_consensus_crypto::TextFmt` representation of `zksync_consensus_roles::attester::SecretKey`.
+#[derive(Debug, Clone)]
+pub struct AttesterSecretKey(pub Secret<String>);
+
 /// `zksync_consensus_crypto::TextFmt` representation of `zksync_consensus_roles::node::PublicKey`.
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub struct NodePublicKey(pub String);
@@ -26,6 +34,12 @@ impl PartialEq for ValidatorSecretKey {
     }
 }
 
+impl PartialEq for AttesterSecretKey {
+    fn eq(&self, other: &Self) -> bool {
+        self.0.expose_secret().eq(other.0.expose_secret())
+    }
+}
+
 impl PartialEq for NodeSecretKey {
     fn eq(&self, other: &Self) -> bool {
         self.0.expose_secret().eq(other.0.expose_secret())
@@ -41,6 +55,15 @@ pub struct WeightedValidator {
     pub weight: u64,
 }
 
+/// Copy-paste of `zksync_consensus_roles::attester::WeightedAttester`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct WeightedAttester {
+    /// Attester key
+    pub key: AttesterPublicKey,
+    /// Attester weight inside the Committee.
+    pub weight: u64,
+}
+
 /// Copy-paste of `zksync_concurrency::net::Host`.
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub struct Host(pub String);
@@ -61,6 +84,8 @@ pub struct GenesisSpec {
     pub protocol_version: ProtocolVersion,
     /// The validator committee. Represents `zksync_consensus_roles::validator::Committee`.
     pub validators: Vec<WeightedValidator>,
+    /// The attester committee. Represents `zksync_consensus_roles::attester::Committee`.
+    pub attesters: Vec<WeightedAttester>,
     /// Leader of the committee. Represents
     /// `zksync_consensus_roles::validator::LeaderSelectionMode::Sticky`.
     pub leader: ValidatorPublicKey,
@@ -119,9 +144,10 @@ impl ConsensusConfig {
     }
 }
 
-/// Secrets need for consensus.
+/// Secrets needed for consensus.
 #[derive(Debug, Clone, PartialEq)]
 pub struct ConsensusSecrets {
     pub validator_key: Option<ValidatorSecretKey>,
+    pub attester_key: Option<AttesterSecretKey>,
     pub node_key: Option<NodeSecretKey>,
 }

core/lib/config/src/testonly.rs

@@ -725,13 +725,24 @@ impl Distribution<configs::consensus::WeightedValidator> for EncodeDist {
     }
 }
 
+impl Distribution<configs::consensus::WeightedAttester> for EncodeDist {
+    fn sample<R: Rng + ?Sized>(&self, rng: &mut R) -> configs::consensus::WeightedAttester {
+        use configs::consensus::{AttesterPublicKey, WeightedAttester};
+        WeightedAttester {
+            key: AttesterPublicKey(self.sample(rng)),
+            weight: self.sample(rng),
+        }
+    }
+}
+
 impl Distribution<configs::consensus::GenesisSpec> for EncodeDist {
     fn sample<R: Rng + ?Sized>(&self, rng: &mut R) -> configs::consensus::GenesisSpec {
         use configs::consensus::{GenesisSpec, ProtocolVersion, ValidatorPublicKey};
         GenesisSpec {
             chain_id: L2ChainId::default(),
             protocol_version: ProtocolVersion(self.sample(rng)),
             validators: self.sample_collect(rng),
+            attesters: self.sample_collect(rng),
             leader: ValidatorPublicKey(self.sample(rng)),
         }
     }
@@ -769,9 +780,12 @@ impl Distribution<configs::consensus::RpcConfig> for EncodeDist {
 
 impl Distribution<configs::consensus::ConsensusSecrets> for EncodeDist {
     fn sample<R: Rng + ?Sized>(&self, rng: &mut R) -> configs::consensus::ConsensusSecrets {
-        use configs::consensus::{ConsensusSecrets, NodeSecretKey, ValidatorSecretKey};
+        use configs::consensus::{
+            AttesterSecretKey, ConsensusSecrets, NodeSecretKey, ValidatorSecretKey,
+        };
         ConsensusSecrets {
             validator_key: self.sample_opt(|| ValidatorSecretKey(String::into(self.sample(rng)))),
+            attester_key: self.sample_opt(|| AttesterSecretKey(String::into(self.sample(rng)))),
             node_key: self.sample_opt(|| NodeSecretKey(String::into(self.sample(rng)))),
         }
     }

core/lib/dal/migrations/20240627142548_l1_batches_consensus.down.sql

@@ -0,0 +1 @@
+DROP TABLE l1_batches_consensus;

core/lib/dal/migrations/20240627142548_l1_batches_consensus.up.sql

@@ -0,0 +1,9 @@
+CREATE TABLE l1_batches_consensus (
+  l1_batch_number BIGINT PRIMARY KEY REFERENCES l1_batches (number) ON DELETE CASCADE,
+  certificate JSONB NOT NULL,
+
+  created_at TIMESTAMP NOT NULL,
+  updated_at TIMESTAMP NOT NULL,
+
+  CHECK((certificate->'message'->'number')::jsonb::numeric = l1_batch_number)
+);