sawtooth_sdk/consensus/
engine.rs

1/*
2 * Copyright 2018 Intel Corporation
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 *     http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 * ------------------------------------------------------------------------------
16 */
17
18use std::error;
19use std::fmt;
20use std::sync::mpsc::Receiver;
21
22use crate::consensus::service::Service;
23
24/// An update from the validator
25#[derive(Debug)]
26#[allow(clippy::large_enum_variant)]
27pub enum Update {
28    PeerConnected(PeerInfo),
29    PeerDisconnected(PeerId),
30    PeerMessage(PeerMessage, PeerId),
31    BlockNew(Block),
32    BlockValid(BlockId),
33    BlockInvalid(BlockId),
34    BlockCommit(BlockId),
35    Shutdown,
36}
37
38pub type BlockId = Vec<u8>;
39
40/// All information about a block that is relevant to consensus
41#[derive(Clone, Default, PartialEq, Hash)]
42pub struct Block {
43    pub block_id: BlockId,
44    pub previous_id: BlockId,
45    pub signer_id: PeerId,
46    pub block_num: u64,
47    pub payload: Vec<u8>,
48    pub summary: Vec<u8>,
49}
50impl Eq for Block {}
51impl fmt::Debug for Block {
52    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
53        write!(
54            f,
55            "Block(block_num: {:?}, block_id: {:?}, previous_id: {:?}, signer_id: {:?}, payload: {}, summary: {})",
56            self.block_num,
57            self.block_id,
58            self.previous_id,
59            self.signer_id,
60            hex::encode(&self.payload),
61            hex::encode(&self.summary),
62        )
63    }
64}
65
66pub type PeerId = Vec<u8>;
67
68/// Information about a peer that is relevant to consensus
69#[derive(Default, Debug, PartialEq, Hash)]
70pub struct PeerInfo {
71    pub peer_id: PeerId,
72}
73impl Eq for PeerInfo {}
74
75/// A consensus-related message sent between peers
76#[derive(Default, Debug, Clone)]
77pub struct PeerMessage {
78    pub header: PeerMessageHeader,
79    pub header_bytes: Vec<u8>,
80    pub header_signature: Vec<u8>,
81    pub content: Vec<u8>,
82}
83
84/// A header associated with a consensus-related message sent from a peer, can be used to verify
85/// the origin of the message
86#[derive(Default, Debug, Clone)]
87pub struct PeerMessageHeader {
88    /// The public key of the validator where this message originated
89    ///
90    /// NOTE: This may not be the validator that sent the message
91    pub signer_id: Vec<u8>,
92    pub content_sha512: Vec<u8>,
93    pub message_type: String,
94    pub name: String,
95    pub version: String,
96}
97
98/// Engine is the only trait that needs to be implemented when adding a new consensus engine.
99///
100/// The consensus engine should listen for notifications from the validator about the status of
101/// blocks and messages from peers. It must also determine internally when to build and publish
102/// blocks based on its view of the network and the consensus algorithm it implements. Often this
103/// will be some sort of timer-based interrupt.
104///
105/// Based on the updates the engine receives through the `Receiver<Update>` and the specifics of
106/// the algorithm being implemented, the engine utilizes the provided `Service` to create new
107/// blocks, communicate with its peers, request that certain blocks be committed, and fail or
108/// ignore blocks that should not be committed.
109///
110/// While the validator may take actions beyond what the engine instructs it to do for performance
111/// optimization reasons, it is the consensus engine's responsibility to drive the progress of the
112/// validator and ensure liveness.
113///
114/// It is not the engine's responsibility to manage blocks or memory, other than to ensure it
115/// responds to every new block with a commit, fail, or ignore within a "reasonable amount of
116/// time". The validator is responsible for guaranteeing the integrity of all blocks sent to the
117/// engine until the engine responds. After the engine responds, the validator does not guarantee
118/// that the block and its predecessors continue to be available unless the block was committed.
119///
120/// Finally, as an optimization, the consensus engine can send prioritized lists of blocks to the
121/// chain controller for checking instead of sending them one at a time, which allows the chain
122/// controller to intelligently work ahead while the consensus engine makes its decisions.
123pub trait Engine {
124    /// Called after the engine is initialized, when a connection to the validator has been
125    /// established. Notifications from the validator are sent along `updates`. `service` is used
126    /// to send requests to the validator.
127    fn start(
128        &mut self,
129        updates: Receiver<Update>,
130        service: Box<dyn Service>,
131        startup_state: StartupState,
132    ) -> Result<(), Error>;
133
134    /// Get the version of this engine
135    fn version(&self) -> String;
136
137    /// Get the name of the engine, typically the algorithm being implemented
138    fn name(&self) -> String;
139
140    /// Any additional name/version pairs this engine supports
141    fn additional_protocols(&self) -> Vec<(String, String)>;
142}
143
144/// State provided to an engine when it is started
145#[derive(Debug, Default)]
146pub struct StartupState {
147    pub chain_head: Block,
148    pub peers: Vec<PeerInfo>,
149    pub local_peer_info: PeerInfo,
150}
151
152#[derive(Debug)]
153pub enum Error {
154    EncodingError(String),
155    SendError(String),
156    ReceiveError(String),
157    InvalidState(String),
158    UnknownBlock(String),
159    UnknownPeer(String),
160    NoChainHead,
161    BlockNotReady,
162}
163
164impl error::Error for Error {}
165
166impl fmt::Display for Error {
167    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
168        use self::Error::*;
169        match *self {
170            EncodingError(ref s) => write!(f, "EncodingError: {}", s),
171            SendError(ref s) => write!(f, "SendError: {}", s),
172            ReceiveError(ref s) => write!(f, "ReceiveError: {}", s),
173            InvalidState(ref s) => write!(f, "InvalidState: {}", s),
174            UnknownBlock(ref s) => write!(f, "UnknownBlock: {}", s),
175            UnknownPeer(ref s) => write!(f, "UnknownPeer: {}", s),
176            NoChainHead => write!(f, "NoChainHead"),
177            BlockNotReady => write!(f, "BlockNotReady"),
178        }
179    }
180}
181
182#[cfg(test)]
183pub mod tests {
184    use super::*;
185
186    use std::default::Default;
187    use std::sync::mpsc::{channel, RecvTimeoutError};
188    use std::sync::{Arc, Mutex};
189    use std::thread;
190    use std::time::Duration;
191
192    use crate::consensus::service::tests::MockService;
193
194    pub struct MockEngine {
195        calls: Arc<Mutex<Vec<String>>>,
196    }
197
198    impl MockEngine {
199        pub fn new() -> Self {
200            MockEngine {
201                calls: Arc::new(Mutex::new(Vec::new())),
202            }
203        }
204
205        pub fn with(amv: Arc<Mutex<Vec<String>>>) -> Self {
206            MockEngine { calls: amv }
207        }
208
209        pub fn calls(&self) -> Vec<String> {
210            let calls = self.calls.lock().unwrap();
211            let mut v = Vec::with_capacity((*calls).len());
212            v.clone_from(&*calls);
213            v
214        }
215    }
216
217    impl Engine for MockEngine {
218        fn start(
219            &mut self,
220            updates: Receiver<Update>,
221            _service: Box<dyn Service>,
222            _startup_state: StartupState,
223        ) -> Result<(), Error> {
224            (*self.calls.lock().unwrap()).push("start".into());
225            loop {
226                match updates.recv_timeout(Duration::from_millis(100)) {
227                    Ok(update) => {
228                        // We don't check for exit() here because we want to drain all the updates
229                        // before we exit. In a real implementation, exit() should also be checked
230                        // here since there is no guarantee the queue will ever be empty.
231                        match update {
232                            Update::PeerConnected(_) => {
233                                (*self.calls.lock().unwrap()).push("PeerConnected".into())
234                            }
235                            Update::PeerDisconnected(_) => {
236                                (*self.calls.lock().unwrap()).push("PeerDisconnected".into())
237                            }
238                            Update::PeerMessage(_, _) => {
239                                (*self.calls.lock().unwrap()).push("PeerMessage".into())
240                            }
241                            Update::BlockNew(_) => {
242                                (*self.calls.lock().unwrap()).push("BlockNew".into())
243                            }
244                            Update::BlockValid(_) => {
245                                (*self.calls.lock().unwrap()).push("BlockValid".into())
246                            }
247                            Update::BlockInvalid(_) => {
248                                (*self.calls.lock().unwrap()).push("BlockInvalid".into())
249                            }
250                            Update::BlockCommit(_) => {
251                                (*self.calls.lock().unwrap()).push("BlockCommit".into())
252                            }
253                            Update::Shutdown => {
254                                println!("shutdown");
255                                break;
256                            }
257                        };
258                    }
259                    Err(RecvTimeoutError::Disconnected) => {
260                        println!("disconnected");
261                        break;
262                    }
263                    Err(RecvTimeoutError::Timeout) => {
264                        println!("timeout");
265                    }
266                }
267            }
268
269            Ok(())
270        }
271        fn version(&self) -> String {
272            "0".into()
273        }
274        fn name(&self) -> String {
275            "mock".into()
276        }
277        fn additional_protocols(&self) -> Vec<(String, String)> {
278            vec![("1".into(), "Mock".into())]
279        }
280    }
281
282    #[test]
283    fn test_engine() {
284        // Create the mock engine with this vec so we can refer to it later. Once we put the engine
285        // in a box, it is hard to get the vec back out.
286        let calls = Arc::new(Mutex::new(Vec::new()));
287
288        // We are going to run two threads to simulate the validator and the driver
289        let mut mock_engine = MockEngine::with(calls.clone());
290
291        let (sender, receiver) = channel();
292        sender
293            .send(Update::PeerConnected(Default::default()))
294            .unwrap();
295        sender
296            .send(Update::PeerDisconnected(Default::default()))
297            .unwrap();
298        sender
299            .send(Update::PeerMessage(Default::default(), Default::default()))
300            .unwrap();
301        sender.send(Update::BlockNew(Default::default())).unwrap();
302        sender.send(Update::BlockValid(Default::default())).unwrap();
303        sender
304            .send(Update::BlockInvalid(Default::default()))
305            .unwrap();
306        sender
307            .send(Update::BlockCommit(Default::default()))
308            .unwrap();
309        let handle = thread::spawn(move || {
310            let svc = Box::new(MockService {});
311            mock_engine
312                .start(receiver, svc, Default::default())
313                .unwrap();
314        });
315        sender.send(Update::Shutdown).unwrap();
316        handle.join().unwrap();
317        assert!(contains(&calls, "start"));
318        assert!(contains(&calls, "PeerConnected"));
319        assert!(contains(&calls, "PeerDisconnected"));
320        assert!(contains(&calls, "PeerMessage"));
321        assert!(contains(&calls, "BlockNew"));
322        assert!(contains(&calls, "BlockValid"));
323        assert!(contains(&calls, "BlockInvalid"));
324        assert!(contains(&calls, "BlockCommit"));
325    }
326
327    fn contains(calls: &Arc<Mutex<Vec<String>>>, expected: &str) -> bool {
328        for call in &*(calls.lock().unwrap()) {
329            if expected == call {
330                return true;
331            }
332        }
333        false
334    }
335}
sawtooth_sdk/consensus/engine.rs

sawtooth_sdk/consensus/
engine.rs
