cuprated/txpool/dandelion/
tx_store.rs

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
use std::task::{Context, Poll};

use bytes::Bytes;
use futures::{future::BoxFuture, FutureExt};
use tower::{Service, ServiceExt};

use cuprate_dandelion_tower::{
    traits::{TxStoreRequest, TxStoreResponse},
    State,
};
use cuprate_database::RuntimeError;
use cuprate_txpool::service::{
    interface::{TxpoolReadRequest, TxpoolReadResponse, TxpoolWriteRequest},
    TxpoolReadHandle, TxpoolWriteHandle,
};

use super::{DandelionTx, TxId};

/// The dandelion tx-store service.
///
/// This is just mapping the interface [`cuprate_dandelion_tower`] wants to what [`cuprate_txpool`] provides.
pub struct TxStoreService {
    pub txpool_read_handle: TxpoolReadHandle,
    pub txpool_write_handle: TxpoolWriteHandle,
}

impl Service<TxStoreRequest<TxId>> for TxStoreService {
    type Response = TxStoreResponse<DandelionTx>;
    type Error = tower::BoxError;
    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;

    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        Poll::Ready(Ok(()))
    }

    fn call(&mut self, req: TxStoreRequest<TxId>) -> Self::Future {
        match req {
            TxStoreRequest::Get(tx_id) => self
                .txpool_read_handle
                .clone()
                .oneshot(TxpoolReadRequest::TxBlob(tx_id))
                .map(|res| match res {
                    Ok(TxpoolReadResponse::TxBlob {
                        tx_blob,
                        state_stem,
                    }) => {
                        let state = if state_stem {
                            State::Stem
                        } else {
                            State::Fluff
                        };

                        Ok(TxStoreResponse::Transaction(Some((
                            DandelionTx(Bytes::from(tx_blob)),
                            state,
                        ))))
                    }
                    Err(RuntimeError::KeyNotFound) => Ok(TxStoreResponse::Transaction(None)),
                    Err(e) => Err(e.into()),
                    Ok(_) => unreachable!(),
                })
                .boxed(),
            TxStoreRequest::Promote(tx_id) => self
                .txpool_write_handle
                .clone()
                .oneshot(TxpoolWriteRequest::Promote(tx_id))
                .map(|res| match res {
                    Ok(_) | Err(RuntimeError::KeyNotFound) => Ok(TxStoreResponse::Ok),
                    Err(e) => Err(e.into()),
                })
                .boxed(),
        }
    }
}