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

Code before:
    ///
    /// This prevents slow validators from blocking the entire operation when faster
    /// validators are available, while still preferring the initial validator if it responds quickly.
    #[instrument(level = "debug", skip_all, fields(tx_digest = ?tx_digest, initial_validator = ?initial_target))]
    async fn get_full_effects_with_fallback<A>(
        &self,
        authority_aggregator: &Arc<AuthorityAggregator<A>>,
        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");
                }


Review comment (suggestion):
Maybe you could insert the future of this to the `fallback_futures` directly so you don't really differentiate , and let the execution be driven by the loop.

Code after:
    ///
    /// This prevents slow validators from blocking the entire operation when faster
    /// validators are available, while still preferring the initial validator if it responds quickly.
    #[instrument(level = "debug", skip_all, fields(tx_digest = ?tx_digest, initial_validator = ?initial_target))]
    async fn get_full_effects_with_fallback<A>(
        &self,
        authority_aggregator: &Arc<AuthorityAggregator<A>>,
        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;