feat(starfish): improving transaction synchronizer (#8157)

# Description of change

In this PR, we improve the transaction synchronizer in Starfish. Main
changes:
- we join fetching and processing transactions into one method
`fetch_and_process_transactions_from_authorities` to reuse this for live
and periodic syncing
- we make at most one fetch request to a given authority in one call of
`fetch_and_process_transactions_from_authorities`
- we track the number of concurrent requests to each authority and limit
that value using a new struct in synchronizer called `active_requests`
- we process transactions after fetching for each authority
individually. While this creates much more calls to the core thread,
this improves the syncing process significantly since it was slowed down
by the slowest response in the previous implementation. Later on we can
add one/a few active thread(s) that will take all fetched bytes and
process them, removing duplicates and sending processed transactions to
the core thread on a regular basis.
- we changed some parameters for fetching transactions, but they might
require further changes once the protocol is tested on a larger network
with longer discrepancies.

## Links to any relevant issues

Fixes #8147 

## How the change has been tested

- [x] Basic tests (linting, compilation, formatting, unit/integration
tests)
- [x] Patch-specific tests (correctness, functionality coverage)
- [ ] I have added tests that prove my fix is effective or that my
feature works
- [ ] I have checked that new and existing unit tests pass locally with
my changes

### Release Notes

- [x] Protocol: Improvements in Starfish tx synchronizer
- [ ] Nodes (Validators and Full nodes):
- [ ] Indexer:
- [ ] JSON-RPC:
- [ ] GraphQL:
- [ ] CLI:
- [ ] Rust SDK:
- [ ] REST API:

Author: polinikita

crates/starfish/config/src/parameters.rs

@@ -151,7 +151,7 @@ impl Parameters {
 
     // Maximum number of block headers to fetch per periodic or live sync request.
     pub(crate) fn default_max_transactions_per_fetch() -> usize {
-        if cfg!(msim) { 10 } else { 1000 }
+        if cfg!(msim) { 10 } else { 100 }
     }
 
     pub(crate) fn default_sync_last_known_own_block_timeout() -> Duration {

crates/starfish/config/tests/snapshots/parameters_test__parameters.snap

@@ -13,7 +13,7 @@ max_forward_time_drift:
   nanos: 500000000
 max_headers_per_commit_sync_fetch: 1000
 max_headers_per_regular_sync_fetch: 100
-max_transactions_per_fetch: 1000
+max_transactions_per_fetch: 100
 sync_last_known_own_block_timeout:
   secs: 5
   nanos: 0

crates/starfish/core/src/metrics.rs

@@ -165,7 +165,7 @@ pub(crate) struct NodeMetrics {
     pub(crate) transactions_synchronizer_fetched_transactions_by_authority: IntCounterVec,
     pub(crate) transactions_synchronizer_missing_transactions_by_authority: IntCounterVec,
     pub(crate) transactions_synchronizer_current_missing_transactions_by_authority: IntGaugeVec,
-    pub(crate) transactions_synchronizer_scheduler_inflight: IntGauge,
+    pub(crate) transactions_synchronizer_periodic_inflight: IntGauge,
     pub(crate) transactions_synchronizer_fetch_latency: HistogramVec,
     pub(crate) transactions_synchronizer_success_by_peer: IntCounterVec,
     pub(crate) transactions_synchronizer_failure_by_peer: IntCounterVec,
@@ -598,7 +598,7 @@ impl NodeMetrics {
                 &["authority"],
                 registry,
             ).unwrap(),
-            transactions_synchronizer_scheduler_inflight: register_int_gauge_with_registry!(
+            transactions_synchronizer_periodic_inflight: register_int_gauge_with_registry!(
                 "fetch_transactions_scheduler_inflight",
                 "Designates whether the transaction synchronizer scheduler task to fetch transactions is currently running",
                 registry,

crates/starfish/core/src/synchronizer.rs

@@ -1375,6 +1375,8 @@ impl<C: NetworkClient, V: BlockVerifier, D: CoreThreadDispatcher> Synchronizer<C
                 },
                 _ = &mut fetcher_timeout => {
                     debug!("Timed out while fetching missing block headers");
+                    // Drop all pending requests immediately — frees all block guards
+                    drop(request_futures);
                     break;
                 }
             }