1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

use solana_client::{
    client_error::Result as ClientResult, rpc_client::RpcClient, rpc_request::RpcError,
};
use solana_sdk::transaction::Transaction;
use solana_sdk::{
    clock::Slot, commitment_config::CommitmentConfig, signature::Signature,
    transaction::uses_durable_nonce,
};

use std::{thread, time};

// #[allow(dead_code)]
// pub fn retry<T>(request: impl Fn() -> Result<T, anchor_client::ClientError>) -> anyhow::Result<T> {
//     for _i in 0..5 {
//         match request() {
//             Ok(res) => return Ok(res),
//             Err(err) => {
//                 // TODO: only retry for recoverable errors
//                 log::error!("{:#?}", err);
//                 continue;
//             }
//         }
//     }
//     Err(anyhow!("Retry failed"))
// }

/// Some Result<> types don't convert to anyhow::Result nicely. Force them through stringification.
pub trait AnyhowWrap {
    type Value;
    fn map_err_anyhow(self) -> anyhow::Result<Self::Value>;
}

impl<T, E: std::fmt::Debug> AnyhowWrap for Result<T, E> {
    type Value = T;
    fn map_err_anyhow(self) -> anyhow::Result<Self::Value> {
        self.map_err(|err| anyhow::anyhow!("{:?}", err))
    }
}

/// Push to an async_channel::Sender and ignore if the channel is full
pub trait AsyncChannelSendUnlessFull<T> {
    /// Send a message if the channel isn't full
    fn send_unless_full(&self, msg: T) -> Result<(), async_channel::SendError<T>>;
}

impl<T> AsyncChannelSendUnlessFull<T> for async_channel::Sender<T> {
    fn send_unless_full(&self, msg: T) -> Result<(), async_channel::SendError<T>> {
        use async_channel::*;
        match self.try_send(msg) {
            Ok(()) => Ok(()),
            Err(TrySendError::Closed(msg)) => Err(async_channel::SendError(msg)),
            Err(TrySendError::Full(_)) => Ok(()),
        }
    }
}

/// A copy of RpcClient::send_and_confirm_transaction that returns the slot the
/// transaction confirmed in.
pub fn send_and_confirm_transaction(
    rpc_client: &RpcClient,
    transaction: &Transaction,
) -> ClientResult<(Signature, Slot)> {
    const SEND_RETRIES: usize = 1;
    const GET_STATUS_RETRIES: usize = usize::MAX;

    'sending: for _ in 0..SEND_RETRIES {
        let signature = rpc_client.send_transaction(transaction)?;

        let recent_blockhash = if uses_durable_nonce(transaction).is_some() {
            let (recent_blockhash, ..) =
                rpc_client.get_latest_blockhash_with_commitment(CommitmentConfig::processed())?;
            recent_blockhash
        } else {
            transaction.message.recent_blockhash
        };

        for status_retry in 0..GET_STATUS_RETRIES {
            let response = rpc_client.get_signature_statuses(&[signature])?.value;
            match response[0]
                .clone()
                .filter(|result| result.satisfies_commitment(rpc_client.commitment()))
            {
                Some(tx_status) => {
                    return if let Some(e) = tx_status.err {
                        Err(e.into())
                    } else {
                        Ok((signature, tx_status.slot))
                    };
                }
                None => {
                    if !rpc_client
                        .is_blockhash_valid(&recent_blockhash, CommitmentConfig::processed())?
                    {
                        // Block hash is not found by some reason
                        break 'sending;
                    } else if cfg!(not(test))
                        // Ignore sleep at last step.
                        && status_retry < GET_STATUS_RETRIES
                    {
                        // Retry twice a second
                        thread::sleep(time::Duration::from_millis(500));
                        continue;
                    }
                }
            }
        }
    }

    Err(RpcError::ForUser(
        "unable to confirm transaction. \
            This can happen in situations such as transaction expiration \
            and insufficient fee-payer funds"
            .to_string(),
    )
    .into())
}