exocore-chain 0.1.24

Storage of Exocore (Distributed applications framework)
Documentation 
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
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119

use exocore_core::{
    cell::{Cell, CellNodes, NodeId},
    framing::CapnpFrameBuilder,
};
use exocore_protos::generated::data_transport_capnp::{
    chain_sync_request, chain_sync_response, pending_sync_request,
};
use exocore_transport::{OutMessage, ServiceType};

use super::{EngineError, Event};
use crate::block::{BlockHeight, BlockOffset};

/// Synchronization context used by `chain_sync`, `pending_sync` and
/// `commit_manager` to dispatch messages to other nodes, and dispatch events to
/// be sent to engine handles.
pub struct SyncContext {
    pub events: Vec<Event>,
    pub messages: Vec<SyncContextMessage>,
    pub sync_state: SyncState,
}

impl SyncContext {
    pub fn new(sync_state: SyncState) -> SyncContext {
        SyncContext {
            events: Vec::new(),
            messages: Vec::new(),
            sync_state,
        }
    }

    pub fn push_pending_sync_request(
        &mut self,
        node_id: NodeId,
        request_builder: CapnpFrameBuilder<pending_sync_request::Owned>,
    ) {
        self.messages.push(SyncContextMessage::PendingSyncRequest(
            node_id,
            request_builder,
        ));
    }

    pub fn push_chain_sync_request(
        &mut self,
        node_id: NodeId,
        request_builder: CapnpFrameBuilder<chain_sync_request::Owned>,
    ) {
        self.messages.push(SyncContextMessage::ChainSyncRequest(
            node_id,
            request_builder,
        ));
    }

    pub fn push_chain_sync_response(
        &mut self,
        node_id: NodeId,
        response_builder: CapnpFrameBuilder<chain_sync_response::Owned>,
    ) {
        self.messages.push(SyncContextMessage::ChainSyncResponse(
            node_id,
            response_builder,
        ));
    }

    pub fn push_event(&mut self, event: Event) {
        self.events.push(event);
    }
}

pub enum SyncContextMessage {
    PendingSyncRequest(NodeId, CapnpFrameBuilder<pending_sync_request::Owned>),
    ChainSyncRequest(NodeId, CapnpFrameBuilder<chain_sync_request::Owned>),
    ChainSyncResponse(NodeId, CapnpFrameBuilder<chain_sync_response::Owned>),
}

impl SyncContextMessage {
    pub fn into_out_message(self, cell: &Cell) -> Result<OutMessage, EngineError> {
        let cell_nodes = cell.nodes();
        let dest_node = cell_nodes
            .get(self.dest_node())
            .map(|n| n.node().clone())
            .ok_or_else(|| EngineError::NodeNotFound(self.dest_node().clone()))?;

        let message = match self {
            SyncContextMessage::PendingSyncRequest(_, request_builder) => {
                OutMessage::from_framed_message(cell, ServiceType::Chain, request_builder)?
                    .with_destination(dest_node)
            }
            SyncContextMessage::ChainSyncRequest(_, request_builder) => {
                OutMessage::from_framed_message(cell, ServiceType::Chain, request_builder)?
                    .with_destination(dest_node)
            }
            SyncContextMessage::ChainSyncResponse(_, response_builder) => {
                OutMessage::from_framed_message(cell, ServiceType::Chain, response_builder)?
                    .with_destination(dest_node)
            }
        };

        Ok(message)
    }

    fn dest_node(&self) -> &NodeId {
        match self {
            SyncContextMessage::PendingSyncRequest(to_node, _) => to_node,
            SyncContextMessage::ChainSyncRequest(to_node, _) => to_node,
            SyncContextMessage::ChainSyncResponse(to_node, _) => to_node,
        }
    }
}

/// State of the synchronization, used to communicate information between the
/// `ChainSynchronizer`, `CommitManager` and `PendingSynchronizer`.
#[derive(Default, Clone, Copy)]
pub struct SyncState {
    /// Indicates what is the last block that got cleaned up from pending store,
    /// and that is now only available from the chain. This is used by the
    /// `PendingSynchronizer` to know which operations it should not include
    /// anymore in its requests.
    pub pending_last_cleanup_block: Option<(BlockOffset, BlockHeight)>,
}