thruput_node_rpc/
chunk.rs

1//! The protocol's chunked RPCs, as a Rust server receives them.
2//!
3//! The database publishes a committed block's balance deltas and auth-data
4//! changes as runs of [`NativeTokensTransferredChunk`] and [`KeyChangeChunk`] records,
5//! because a block can touch more accounts than one RPC record carries.
6//! Every record of a run carries the same `chunk_count`, each `chunk_index`
7//! in `[0, chunk_count)` is sent exactly once, and a run always has at least
8//! one record, so an empty message is a run of one empty record. The records
9//! of a run may be sent concurrently and over several connections, so they
10//! can arrive in any order. The `KeyChangeChunk` runs of several heights can
11//! interleave; the database sends one height's `NativeTokensTransferredChunk` run
12//! only once the previous one is acknowledged.
13//!
14//! The [`ChunkRecord`] impls here take the two records apart into height,
15//! [`ChunkPosition`] and items. A server implementing `SchedDb1`, `AuthDB1`
16//! or `MempoolAuth1` (over which the auth server forwards `KeyChangeChunk`
17//! runs to the mempool) hands every record it receives to a
18//! `thruput_rpc_driver::ChunkRunAssembler`, which returns the height's whole message
19//! once every chunk of its run is in, with the parts read as one sequence.
20//! This crate depends on `thruput_rpc_api` alone, so the assembler is not
21//! named here:
22//!
23//! ```
24//! use thruput_rpc_driver::{ChunkRunAssembler, ProgramError};
25//! use thruput_node_rpc::{Delta, NativeTokensTransferredChunk};
26//! use thruput_xdr::types::BlockHeight;
27//! use xdrs::VecM;
28//!
29//! struct Mempool {
30//!     transfers: ChunkRunAssembler<BlockHeight, VecM<Delta>>,
31//! }
32//!
33//! impl Mempool {
34//!     fn scheddb_native_tokens_transferred_chunk(
35//!         &self,
36//!         chunk: NativeTokensTransferredChunk,
37//!     ) -> Result<(), ProgramError> {
38//!         // A chunk that breaks its run's framing is a sender bug; the
39//!         // error converts into the ProgramError a handler returns.
40//!         if let Some(run) = self.transfers.push_record(chunk)? {
41//!             let block: &BlockHeight = &run.key;
42//!             for delta in &run.parts {
43//!                 let _: &Delta = delta;
44//!                 // apply the block's delta ...
45//!             }
46//!         }
47//!         Ok(())
48//!     }
49//! }
50//! ```
51
52use thruput_rpc_api::{ChunkPosition, ChunkRecord};
53use thruput_xdr::types::BlockHeight;
54use xdrs::VecM;
55
56use crate::{
57    Delta, KeyChangeBatch, KeyChangeChunk, NativeTokensTransferredChunk,
58};
59
60impl ChunkRecord for NativeTokensTransferredChunk {
61    type Key = BlockHeight;
62    type Part = VecM<Delta>;
63
64    fn into_chunk(self) -> (BlockHeight, ChunkPosition, VecM<Delta>) {
65        (
66            self.block,
67            ChunkPosition::new(self.chunk_index, self.chunk_count),
68            self.deltas,
69        )
70    }
71}
72
73impl ChunkRecord for KeyChangeChunk {
74    type Key = BlockHeight;
75    type Part = KeyChangeBatch;
76
77    fn into_chunk(self) -> (BlockHeight, ChunkPosition, KeyChangeBatch) {
78        (
79            self.height,
80            ChunkPosition::new(self.chunk_index, self.chunk_count),
81            self.changes,
82        )
83    }
84}
85
86#[cfg(test)]
87mod tests {
88    use super::*;
89    use crate::KeyChange;
90    use thruput_xdr::types::{ContractAddress, int128_xdr};
91
92    fn delta(lowbits: u64) -> Delta {
93        Delta {
94            address: ContractAddress::default(),
95            native_token_delta: int128_xdr {
96                highbits: 0,
97                lowbits,
98            },
99        }
100    }
101
102    #[test]
103    fn a_transfer_chunk_comes_apart_into_block_position_and_deltas() {
104        // The record a handler receives is what the dispatcher decoded from
105        // the wire, so go through the wire form.
106        let chunk = NativeTokensTransferredChunk {
107            block: BlockHeight::new(42),
108            chunk_index: 1,
109            chunk_count: 3,
110            deltas: VecM(vec![delta(1), delta(2)]),
111        };
112        let bytes = xdrs::txs::encode(&chunk).expect("encodes");
113        let decoded: NativeTokensTransferredChunk =
114            xdrs::txs::decode(&bytes).expect("decodes");
115        let (block, position, deltas) = decoded.into_chunk();
116        assert_eq!(block, BlockHeight::new(42));
117        assert_eq!(position, ChunkPosition::new(1, 3));
118        assert_eq!(deltas, VecM(vec![delta(1), delta(2)]));
119    }
120
121    #[test]
122    fn a_key_change_chunk_comes_apart_into_height_position_and_changes() {
123        let change = KeyChange {
124            address: ContractAddress::default(),
125            auth_data: None,
126        };
127        let chunk = KeyChangeChunk {
128            height: BlockHeight::new(5),
129            chunk_index: 0,
130            chunk_count: 1,
131            changes: VecM(vec![change.clone()]),
132        };
133        let (height, position, changes) = chunk.into_chunk();
134        assert_eq!(height, BlockHeight::new(5));
135        assert_eq!(position, ChunkPosition::new(0, 1));
136        assert_eq!(changes, VecM(vec![change]));
137    }
138}
thruput_node_rpc/chunk.rs

thruput_node_rpc/
chunk.rs
