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

cuprate_blockchain/service/
write.rs
