Repository: aptos-labs/aptos-core
File: consensus/src/pipeline/decryption_pipeline_builder.rs (Rust)

Code before:
        // TODO(ibalajiarun): Fix this wrapping
        let encryption_round = block.round() % 200;
        let (digest, proofs_promise) = FPTXWeighted::digest(
            secret_share_config.digest_key(),
            &txn_ciphertexts,
            encryption_round,
        )?;

        let metadata = SecretShareMetadata::new(
            block.epoch(),
            block.round(),
            block.timestamp_usecs(),
            block.id(),
            digest.clone(),
        );

        let derived_key_share =
            FPTXWeighted::derive_decryption_key_share(secret_share_config.msk_share(), &digest)?;
        if derived_self_key_share_tx
            .send(Some(SecretShare::new(
                author,
                metadata.clone(),
                derived_key_share,
            )))
            .is_err()
        {
            return Err(anyhow!(
                "derived_self_key_share_tx receiver dropped, pipeline likely aborted"
            )
            .into());
        }

        // TODO(ibalajiarun): improve perf
        let proofs = FPTXWeighted::eval_proofs_compute_all(
            &proofs_promise,
            secret_share_config.digest_key(),
        );

        let maybe_decryption_key = secret_shared_key_rx
            .await
            .map_err(|_| anyhow!("secret_shared_key_rx dropped"))?;
        // TODO(ibalajiarun): account for the case where decryption key is not available
        let decryption_key = maybe_decryption_key.expect("decryption key should be available");

        info!(
            "Successfully received decryption key for block {}: metadata={:?}",
            block.round(),
            decryption_key.metadata
        );
        let decrypted_txns: Vec<_> = encrypted_txns
            .into_par_iter()

Review comment (suggestion):
We don't want to restrict the number of threads that will do this work?

Asking because in @danielxiangzl 's initial prototype, I believe he used a threadpool to restrict to 16 threads

Code after:
        let metadata = SecretShareMetadata::new(
            block.epoch(),
            block.round(),
            block.timestamp_usecs(),
            block.id(),
            digest.clone(),
        );

        let derived_key_share =
            FPTXWeighted::derive_decryption_key_share(secret_share_config.msk_share(), &digest)?;
        if derived_self_key_share_tx
            .send(Some(SecretShare::new(
                author,
                metadata.clone(),
                derived_key_share,
            )))
            .is_err()
        {
            return Err(anyhow!(
                "derived_self_key_share_tx receiver dropped, pipeline likely aborted"
            )
            .into());
        }

        // TODO(ibalajiarun): improve perf
        let proofs = FPTXWeighted::eval_proofs_compute_all(
            &proofs_promise,
            secret_share_config.digest_key(),
        );

        let maybe_decryption_key = secret_shared_key_rx
            .await
            .map_err(|_| anyhow!("secret_shared_key_rx dropped"))?;
        // TODO(ibalajiarun): account for the case where decryption key is not available
        let decryption_key = maybe_decryption_key.expect("decryption key should be available");

        info!(
            "Successfully received decryption key for block {}: metadata={:?}",
            block.round(),
            decryption_key.metadata
        );
        let decrypted_txns: Vec<_> = encrypted_txns
            .into_par_iter()
            .zip(txn_ciphertexts)
            .map(|(mut txn, ciphertext)| {
                let eval_proof = proofs.get(&ciphertext.id()).expect("must exist");
                match FPTXWeighted::decrypt_individual::<DecryptedPayload>(
                    &decryption_key.key,
                    &ciphertext,
                    &digest,
                    &eval_proof,