[Consensus] Garbage Collection - 2 (#19385)

## Description 

This is the second part of Garbage Collection which implements the
following ticked next steps as outlined on the previous PR

- [x] BlockManager to respect the gc_round when accepting blocks and
trigger clean ups for new gc rounds
- [x] Skip blocks that are received which are <= gc_round
- [x] Not propose ancestors that are <= gc_round
- [x] Subscriber to ask for blocks from `gc_round` when
`last_fetched_round < gc_round` for a peer to prevent us from fetching
unnecessary blocks

Next steps:

- [ ] Re-propose GC'ed transactions (probably not all of them)
- [ ] Implement new timestamp approach so ancestor verification is not
needed
- [ ] Harden testing for GC & edge cases

## Test plan 

CI/PT

---

## Release notes

Check each box that your changes affect. If none of the boxes relate to
your changes, release notes aren't required.

For each box you select, include information after the relevant heading
that describes the impact of your changes that a user might notice and
any actions they must take to implement updates.

- [ ] Protocol: 
- [ ] Nodes (Validators and Full nodes): 
- [ ] Indexer: 
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:
- [ ] REST API:

consensus/core/src/authority_node.rs

@@ -487,12 +487,16 @@ mod tests {
     #[tokio::test(flavor = "current_thread")]
     async fn test_authority_committee(
         #[values(ConsensusNetwork::Anemo, ConsensusNetwork::Tonic)] network_type: ConsensusNetwork,
+        #[values(0, 5, 10)] gc_depth: u32,
     ) {
         let db_registry = Registry::new();
         DBMetrics::init(&db_registry);
 
         const NUM_OF_AUTHORITIES: usize = 4;
         let (committee, keypairs) = local_committee_and_keys(0, [1; NUM_OF_AUTHORITIES].to_vec());
+        let mut protocol_config = ProtocolConfig::get_for_max_version_UNSAFE();
+        protocol_config.set_consensus_gc_depth_for_testing(gc_depth);
+
         let temp_dirs = (0..NUM_OF_AUTHORITIES)
             .map(|_| TempDir::new().unwrap())
             .collect::<Vec<_>>();
@@ -509,6 +513,7 @@ mod tests {
                 keypairs.clone(),
                 network_type,
                 boot_counters[index],
+                protocol_config.clone(),
             )
             .await;
             boot_counters[index] += 1;
@@ -565,6 +570,7 @@ mod tests {
             keypairs.clone(),
             network_type,
             boot_counters[index],
+            protocol_config.clone(),
         )
         .await;
         boot_counters[index] += 1;
@@ -582,6 +588,7 @@ mod tests {
     #[tokio::test(flavor = "current_thread")]
     async fn test_amnesia_recovery_success(
         #[values(ConsensusNetwork::Anemo, ConsensusNetwork::Tonic)] network_type: ConsensusNetwork,
+        #[values(0, 5, 10)] gc_depth: u32,
     ) {
         telemetry_subscribers::init_for_testing();
         let db_registry = Registry::new();
@@ -594,6 +601,9 @@ mod tests {
         let mut temp_dirs = BTreeMap::new();
         let mut boot_counters = [0; NUM_OF_AUTHORITIES];
 
+        let mut protocol_config = ProtocolConfig::get_for_max_version_UNSAFE();
+        protocol_config.set_consensus_gc_depth_for_testing(gc_depth);
+
         for (index, _authority_info) in committee.authorities() {
             let dir = TempDir::new().unwrap();
             let (authority, receiver) = make_authority(
@@ -603,6 +613,7 @@ mod tests {
                 keypairs.clone(),
                 network_type,
                 boot_counters[index],
+                protocol_config.clone(),
             )
             .await;
             assert!(authority.sync_last_known_own_block_enabled(), "Expected syncing of last known own block to be enabled as all authorities are of empty db and boot for first time.");
@@ -651,6 +662,7 @@ mod tests {
             keypairs.clone(),
             network_type,
             boot_counters[index_1],
+            protocol_config.clone(),
         )
         .await;
         assert!(
@@ -671,6 +683,7 @@ mod tests {
             keypairs,
             network_type,
             boot_counters[index_2],
+            protocol_config.clone(),
         )
         .await;
         assert!(
@@ -704,6 +717,7 @@ mod tests {
         keypairs: Vec<(NetworkKeyPair, ProtocolKeyPair)>,
         network_type: ConsensusNetwork,
         boot_counter: u64,
+        protocol_config: ProtocolConfig,
     ) -> (ConsensusAuthority, UnboundedReceiver<CommittedSubDag>) {
         let registry = Registry::new();
 
@@ -729,7 +743,7 @@ mod tests {
             index,
             committee,
             parameters,
-            ProtocolConfig::get_for_max_version_UNSAFE(),
+            protocol_config,
             protocol_keypair,
             network_keypair,
             Arc::new(txn_verifier),