Repository: MystenLabs/sui
File: crates/sui-core/src/transaction_driver/effects_certifier.rs (Rust)

Code before:
        // Track validators we've already sent requests to
        let mut requested_validators = HashSet::new();
        requested_validators.insert(initial_target);

        let mut fallback_futures: FuturesUnordered<
            BoxFuture<'_, (AuthorityName, FullEffectsResult)>,
        > = FuturesUnordered::new();

        loop {
            tokio::select! {
                result = &mut initial_request => {
                    return (result, initial_target);
                }

                Some((validator, result)) = fallback_futures.next(), if !fallback_futures.is_empty() => {
                    if result.is_ok() {
                        return (result, validator);
                    }
                    tracing::debug!(?validator, "Fallback get_full_effects failed, continuing to wait");
                }

                // After delay, try to start a fallback request to an acked validator
                _ = fallback_delay.tick() => {
                    // Drain all available acked validators and pick one we haven't tried
                    while let Ok(acked_validator) = acked_validators_rx.try_recv() {
                        if !requested_validators.contains(&acked_validator) {
                            requested_validators.insert(acked_validator);

                            if let Some(client) = authority_aggregator.authority_clients.get(&acked_validator) {
                                tracing::debug!(
                                    ?acked_validator,
                                    "Starting fallback get_full_effects request"
                                );

                                let client = client.clone();
                                let fut = self.get_full_effects(
                                    client,
                                    tx_digest,
                                    tx_type,
                                    consensus_position,
                                    options,
                                );

                                fallback_futures.push(Box::pin(async move {
                                    (acked_validator, fut.await)
                                }));

                                // Only start one fallback per interval
                                break;
                            }
                        }

Review comment (security):
nit:
```
if requested_validators.contains(&acked_validator) {
    break;
}
...
let Some(client) = authority_aggregator.authority_clients.get(&acked_validator) else {
    break
};
...
```

Code after:
        pending_requests.push(Box::pin(
            async move { (initial_target, initial_request.await) },
        ));

        let mut fallback_delay = tokio::time::interval(GET_FULL_EFFECTS_FALLBACK_DELAY);
        fallback_delay.reset();

        loop {
            tokio::select! {
                Some((validator, result)) = pending_requests.next() => {
                    // Return as soon as any request (including fallback)completes - the caller handles retries for errors
                    return (result, validator);
                }

                // After delay, try to start a fallback request to an acked validator
                _ = fallback_delay.tick() => {
                    // Drain all available acked validators and pick one we haven't tried
                    while let Ok(acked_validator) = acked_validators_rx.try_recv() {
                        // We send ack requests to all validators, so skip if the acked validator was the initial target
                        if acked_validator == initial_target {
                            continue;
                        }

                        let Some(client) = authority_aggregator.authority_clients.get(&acked_validator) else {
                            continue;
                        };

                        tracing::debug!(
                            ?acked_validator,
                            "Starting fallback get_full_effects request"
                        );

                        let fut = self.get_full_effects(
                            client.clone(),
                            tx_digest,
                            tx_type,
                            consensus_position,
                            options,
                        );

                        pending_requests.push(Box::pin(async move {
                            (acked_validator, fut.await)
                        }));

                        // Only start one fallback per interval
                        break;
                    }
                }
            }
        }
    }