cuprate_blockchain/service/
write.rs

1//! Database writer thread definitions and logic.
2
3use std::{
4    borrow::Cow,
5    sync::Arc,
6    task::{Context, Poll},
7};
8
9use crossbeam::channel::Receiver;
10use fjall::PersistMode;
11use futures::channel::oneshot;
12use tapes::TapesRead;
13use tower::Service;
14use tracing::instrument;
15
16use cuprate_helper::cast::u64_to_usize;
17use cuprate_types::{
18    blockchain::{BlockchainResponse, BlockchainWriteRequest},
19    AltBlockInformation, ChainId, VerifiedBlockInformation,
20};
21
22use crate::{
23    config::Persistence,
24    error::{BlockchainError, DbResult},
25    ops::block::add_blocks_to_tapes,
26    service::ResponseResult,
27    BlockchainDatabase,
28};
29
30//---------------------------------------------------------------------------------------------------- init_write_service
31/// Initialise the blockchain write service from a [`BlockchainDatabase`].
32pub fn init_write_service(env: Arc<BlockchainDatabase>) -> BlockchainWriteHandle {
33    let (sender, receiver) = crossbeam::channel::unbounded();
34
35    std::thread::Builder::new()
36        .name("cuprate_blockchain_writer".into())
37        .spawn(move || writer_thread(&env, &receiver))
38        .unwrap();
39
40    BlockchainWriteHandle { sender }
41}
42
43/// The [`tower::Service`] handle to write to the database.
44pub struct BlockchainWriteHandle {
45    /// Sender channel to the database write thread-pool.
46    ///
47    /// We provide the response channel for the thread-pool.
48    sender: crossbeam::channel::Sender<(
49        BlockchainWriteRequest,
50        oneshot::Sender<DbResult<BlockchainResponse>>,
51    )>,
52}
53
54impl Service<BlockchainWriteRequest> for BlockchainWriteHandle {
55    type Response = BlockchainResponse;
56    type Error = BlockchainError;
57    type Future = cuprate_helper::asynch::InfallibleOneshotReceiver<DbResult<BlockchainResponse>>;
58
59    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
60        Poll::Ready(Ok(()))
61    }
62
63    fn call(&mut self, req: BlockchainWriteRequest) -> Self::Future {
64        let (response_sender, receiver) = oneshot::channel();
65
66        self.sender.try_send((req, response_sender)).unwrap();
67
68        cuprate_helper::asynch::InfallibleOneshotReceiver::from(receiver)
69    }
70}
71
72#[instrument(
73    name = "blockchain_writer_thread",
74    skip(env, receiver),
75    level = "error"
76)]
77fn writer_thread(
78    env: &Arc<BlockchainDatabase>,
79    receiver: &Receiver<(
80        BlockchainWriteRequest,
81        oneshot::Sender<DbResult<BlockchainResponse>>,
82    )>,
83) {
84    while let Ok((req, response_sender)) = receiver.recv() {
85        let span = tracing::debug_span!("write_request");
86        span.in_scope(|| {
87            let response = handle_blockchain_request(env, &req);
88
89            match &response {
90                Ok(_) => tracing::debug!("Sending successful write response."),
91                Err(e) => {
92                    tracing::error!("Failed to handle write request: {e:?}");
93                }
94            }
95
96            let _ = response_sender.send(response).inspect_err(|_| {
97                tracing::warn!("Failed to send write response, rx wasn't waiting.");
98            });
99        });
100    }
101}
102
103//---------------------------------------------------------------------------------------------------- handle_bc_request
104/// Handle an incoming [`BlockchainWriteRequest`], returning a [`BlockchainResponse`].
105fn handle_blockchain_request(
106    env: &Arc<BlockchainDatabase>,
107    req: &BlockchainWriteRequest,
108) -> Result<BlockchainResponse, BlockchainError> {
109    match req {
110        BlockchainWriteRequest::WriteBlock(block) => write_block(env, block),
111        BlockchainWriteRequest::BatchWriteBlocks(blocks) => write_blocks(env, blocks),
112        BlockchainWriteRequest::WriteAltBlock(alt_block) => write_alt_block(env, alt_block),
113        BlockchainWriteRequest::PopBlocks(numb_blocks) => pop_blocks(env, *numb_blocks),
114        BlockchainWriteRequest::FlushAltBlocks => flush_alt_blocks(env),
115    }
116}
117
118//---------------------------------------------------------------------------------------------------- Handler functions
119// These are the actual functions that do stuff according to the incoming [`Request`].
120//
121// Each function name is a 1-1 mapping (from CamelCase -> snake_case) to
122// the enum variant name, e.g: `BlockExtendedHeader` -> `block_extended_header`.
123//
124// Each function will return the [`Response`] that we
125// should send back to the caller in [`map_request()`].
126
127/// [`BlockchainWriteRequest::WriteBlock`].
128#[inline]
129#[instrument(skip(db, block), level = "debug")]
130fn write_block(db: &BlockchainDatabase, block: &VerifiedBlockInformation) -> ResponseResult {
131    write_blocks(db, std::slice::from_ref(block))
132}
133
134/// [`BlockchainWriteRequest::BatchWriteBlocks`].
135#[inline]
136#[instrument(skip(db, blocks), level = "debug")]
137fn write_blocks(db: &BlockchainDatabase, blocks: &[VerifiedBlockInformation]) -> ResponseResult {
138    let (tapes_persist_mode, fjall_persist_mode) = match db.config.persistence {
139        Persistence::Buffer => (tapes::Persistence::Buffer, PersistMode::Buffer),
140        // We use the amount of blocks to write as a heuristic for if we are synced. When Cuprate starts downloading
141        // blocks it will do 1 at a time so for that those will be fully synced but that does not last long.
142        Persistence::BufferThenSync if blocks.len() > 1 => {
143            (tapes::Persistence::Buffer, PersistMode::Buffer)
144        }
145        Persistence::Sync | Persistence::BufferThenSync => {
146            (tapes::Persistence::SyncAll, PersistMode::SyncAll)
147        }
148    };
149
150    tracing::debug!("Writing {} block(s) to database.", blocks.len());
151
152    let mut tapes = db.linear_tapes.append();
153
154    let mut numb_transactions = tapes
155        .fixed_sized_tape_len(&db.tx_infos)
156        .expect("required tape not open");
157
158    add_blocks_to_tapes(blocks, db, &mut tapes)?;
159
160    tapes.commit(tapes_persist_mode)?;
161
162    let mut pre_rct_numb_outputs_cache = db.pre_rct_numb_outputs_cache.lock().unwrap();
163
164    let mut tx_rw = db.fjall.batch().durability(Some(fjall_persist_mode));
165
166    for block in blocks {
167        crate::ops::block::add_block_to_dynamic_tables(
168            db,
169            &block.block,
170            &block.block_hash,
171            block.txs.iter().map(|tx| Cow::Borrowed(&tx.tx)),
172            &mut numb_transactions,
173            &mut tx_rw,
174            &mut pre_rct_numb_outputs_cache,
175        )?;
176    }
177
178    tx_rw.commit()?;
179
180    Ok(BlockchainResponse::Ok)
181}
182
183/// [`BlockchainWriteRequest::WriteAltBlock`].
184#[inline]
185fn write_alt_block(db: &BlockchainDatabase, block: &AltBlockInformation) -> ResponseResult {
186    let mut tx_rw = db.fjall.batch().durability(Some(PersistMode::SyncAll));
187
188    crate::ops::alt_block::add_alt_block(db, block, &mut tx_rw)?;
189
190    tx_rw.commit()?;
191
192    Ok(BlockchainResponse::Ok)
193}
194
195/// [`BlockchainWriteRequest::PopBlocks`].
196fn pop_blocks(db: &BlockchainDatabase, numb_blocks: usize) -> ResponseResult {
197    let mut tapes = db.linear_tapes.truncate();
198    let mut tx_rw = db.fjall.batch().durability(Some(PersistMode::SyncAll));
199
200    // flush all the current alt blocks as they may reference blocks to be popped.
201    crate::ops::alt_block::flush_alt_blocks(db)?;
202
203    // generate a `ChainId` for the popped blocks.
204    let old_main_chain_id = ChainId(rand::random());
205
206    assert!(tapes
207        .fixed_sized_tape_len(&db.block_infos)
208        .is_some_and(|height| u64_to_usize(height) > numb_blocks));
209
210    // pop the blocks
211    for _ in 0..numb_blocks {
212        crate::ops::block::pop_block(db, Some(old_main_chain_id), &mut tx_rw, &mut tapes)?;
213    }
214
215    tapes.commit(tapes::Persistence::SyncAll)?;
216    tx_rw.commit()?;
217    Ok(BlockchainResponse::PopBlocks(old_main_chain_id))
218}
219
220/// [`BlockchainWriteRequest::FlushAltBlocks`].
221#[inline]
222fn flush_alt_blocks(db: &BlockchainDatabase) -> ResponseResult {
223    crate::ops::alt_block::flush_alt_blocks(db)?;
224
225    Ok(BlockchainResponse::Ok)
226}
cuprate_blockchain/service/write.rs

cuprate_blockchain/service/
write.rs
