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

Code before:
        initial_client: Arc<SafeClient<A>>,
        initial_target: AuthorityName,
        tx_digest: Option<TransactionDigest>,
        tx_type: TxType,
        consensus_position: Option<ConsensusPosition>,
        options: &SubmitTransactionOptions,
        mut acked_validators_rx: Receiver<AuthorityName>,
    ) -> (FullEffectsResult, AuthorityName)
    where
        A: AuthorityAPI + Send + Sync + 'static + Clone,
    {
        let initial_request = self.get_full_effects(
            initial_client,
            tx_digest,
            tx_type,
            consensus_position,
            options,
        );
        tokio::pin!(initial_request);

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

        // 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);


Review comment (suggestion):
since we assume that each validator is sending at most one ack , then do we need the `HashSet` to dedupe?  Overall I think it's fine to have some additional protection in place in case the upstream logic changes.

Code after:
        initial_client: Arc<SafeClient<A>>,
        initial_target: AuthorityName,
        tx_digest: Option<TransactionDigest>,
        tx_type: TxType,
        consensus_position: Option<ConsensusPosition>,
        options: &SubmitTransactionOptions,
        mut acked_validators_rx: Receiver<AuthorityName>,
    ) -> (FullEffectsResult, AuthorityName)
    where
        A: AuthorityAPI + Send + Sync + 'static + Clone,
    {
        let mut pending_requests: FuturesUnordered<
            BoxFuture<'_, (AuthorityName, FullEffectsResult)>,
        > = FuturesUnordered::new();

        // Add initial request to the pending set alongside fallbacks for uniform handling
        let initial_request = self.get_full_effects(
            initial_client,
            tx_digest,
            tx_type,
            consensus_position,
            options,
        );
        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!(