battleware-node 0.0.1

Validator that participates in a battleware network.
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
120
121
122
123
124
125
126
127

use battleware_types::{Activity, Seed};
use bytes::Bytes;
use commonware_consensus::{
    threshold_simplex::types::{Seedable, View},
    Reporter,
};
use commonware_resolver::{p2p::Producer, Consumer};
use commonware_utils::sequence::U64;
use futures::{
    channel::{mpsc, oneshot},
    SinkExt,
};

pub enum Message {
    Put(Seed),
    Get {
        view: View,
        response: oneshot::Sender<Seed>,
    },
    Deliver {
        view: View,
        signature: Bytes,
        response: oneshot::Sender<bool>,
    },
    Produce {
        view: View,
        response: oneshot::Sender<Bytes>,
    },
    Uploaded {
        view: View,
    },
}

#[derive(Clone)]
pub struct Mailbox {
    sender: mpsc::Sender<Message>,
}

impl Mailbox {
    pub(super) fn new(sender: mpsc::Sender<Message>) -> Self {
        Self { sender }
    }

    pub async fn put(&mut self, seed: Seed) {
        self.sender
            .send(Message::Put(seed))
            .await
            .expect("failed to send put");
    }

    pub async fn get(&mut self, view: View) -> Seed {
        let (sender, receiver) = oneshot::channel();
        self.sender
            .send(Message::Get {
                view,
                response: sender,
            })
            .await
            .expect("failed to send get");
        receiver.await.expect("failed to receive get")
    }

    pub async fn uploaded(&mut self, view: View) {
        self.sender
            .send(Message::Uploaded { view })
            .await
            .expect("failed to send uploaded");
    }
}

impl Consumer for Mailbox {
    type Key = U64;
    type Value = Bytes;
    type Failure = ();

    async fn deliver(&mut self, key: Self::Key, value: Self::Value) -> bool {
        let (sender, receiver) = oneshot::channel();
        self.sender
            .send(Message::Deliver {
                view: key.into(),
                signature: value,
                response: sender,
            })
            .await
            .expect("failed to send deliver");
        receiver.await.unwrap_or(true) // default to true to avoid blocking
    }

    async fn failed(&mut self, _: Self::Key, _: Self::Failure) {
        // We don't need to do anything on failure, the resolver will retry.
    }
}

impl Producer for Mailbox {
    type Key = U64;

    async fn produce(&mut self, key: Self::Key) -> oneshot::Receiver<Bytes> {
        let (sender, receiver) = oneshot::channel();
        self.sender
            .send(Message::Produce {
                view: key.into(),
                response: sender,
            })
            .await
            .expect("failed to send produce");
        receiver
    }
}

impl Reporter for Mailbox {
    type Activity = Activity;

    async fn report(&mut self, activity: Self::Activity) {
        match activity {
            Activity::Notarization(notarization) => {
                self.put(notarization.seed()).await;
            }
            Activity::Nullification(nullification) => {
                self.put(nullification.seed()).await;
            }
            Activity::Finalization(finalization) => {
                self.put(finalization.seed()).await;
            }
            _ => {}
        }
    }
}