Client subscriber (#130)

Subscriber client

narwhal/consensus/Cargo.toml

@@ -11,6 +11,7 @@ rand = { version = "0.7.3", optional = true }
 crypto = { path = "../crypto" }
 config = { path = "../config" }
 primary = { path = "../primary" }
+blake2 = "0.9"
 tracing = "0.1.31"
 tokio-util = { version = "0.7.0", features = ["codec"] }
 serde = { version = "1.0.136", features = ["derive"] }

narwhal/consensus/src/subscriber.rs

@@ -272,7 +272,7 @@ impl<PublicKey: VerifyingKey> SubscriberConnection<PublicKey> {
             .expect("Failed to send new subscriber to core");
 
         // Interact with the subscriber.
-        // TODO [issue #120]: Better error handling (we have a log of prints and breaks here).
+        // TODO [issue #120]: Better error handling (we have a lot of prints and breaks here).
         loop {
             tokio::select! {
                 // Update the subscriber every time a certificate is sequenced.

narwhal/worker/src/helper.rs

@@ -1,6 +1,7 @@
 // Copyright (c) 2021, Facebook, Inc. and its affiliates
 // Copyright (c) 2022, Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
+use crate::worker::SerializedBatchMessage;
 use bytes::Bytes;
 use config::{Committee, WorkerId};
 use crypto::traits::VerifyingKey;
@@ -10,8 +11,6 @@ use store::Store;
 use tokio::sync::mpsc::{Receiver, Sender};
 use tracing::{error, warn};
 
-use crate::processor::SerializedBatchMessage;
-
 #[cfg(test)]
 #[path = "tests/helper_tests.rs"]
 pub mod helper_tests;

narwhal/worker/src/lib.rs

@@ -20,4 +20,4 @@ mod worker;
 #[path = "tests/common.rs"]
 mod common;
 
-pub use crate::worker::Worker;
+pub use crate::worker::{SerializedBatchMessage, Worker, WorkerMessage};

narwhal/worker/src/processor.rs

@@ -1,10 +1,9 @@
 // Copyright (c) 2021, Facebook, Inc. and its affiliates
 // Copyright (c) 2022, Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
-use crate::worker::SerializedWorkerPrimaryMessage;
+use crate::worker::{SerializedBatchMessage, SerializedWorkerPrimaryMessage};
 use blake2::digest::Update;
 use config::WorkerId;
-
 use primary::{BatchDigest, WorkerPrimaryMessage};
 use store::Store;
 use tokio::sync::mpsc::{Receiver, Sender};
@@ -13,9 +12,6 @@ use tokio::sync::mpsc::{Receiver, Sender};
 #[path = "tests/processor_tests.rs"]
 pub mod processor_tests;
 
-/// Indicates a serialized `WorkerMessage::Batch` message.
-pub type SerializedBatchMessage = Vec<u8>;
-
 /// Hashes and stores batches, it then outputs the batch's digest.
 pub struct Processor;
 

narwhal/worker/src/quorum_waiter.rs

@@ -1,7 +1,7 @@
 // Copyright (c) 2021, Facebook, Inc. and its affiliates
 // Copyright (c) 2022, Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
-use crate::processor::SerializedBatchMessage;
+use crate::worker::SerializedBatchMessage;
 use config::{Committee, Stake};
 use crypto::traits::VerifyingKey;
 use futures::stream::{futures_unordered::FuturesUnordered, StreamExt as _};

narwhal/worker/src/synchronizer.rs

@@ -1,10 +1,7 @@
 // Copyright (c) 2021, Facebook, Inc. and its affiliates
 // Copyright (c) 2022, Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
-use crate::{
-    processor::SerializedBatchMessage,
-    worker::{Round, SerializedWorkerPrimaryMessage, WorkerMessage},
-};
+use crate::worker::{Round, SerializedBatchMessage, SerializedWorkerPrimaryMessage, WorkerMessage};
 use bytes::Bytes;
 use config::{Committee, WorkerId};
 use crypto::traits::VerifyingKey;

narwhal/worker/src/tests/common.rs

@@ -1,7 +1,7 @@
 // Copyright (c) 2021, Facebook, Inc. and its affiliates
 // Copyright (c) 2022, Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
-use crate::{processor::SerializedBatchMessage, worker::WorkerMessage};
+use crate::{worker::WorkerMessage, SerializedBatchMessage};
 use blake2::digest::Update;
 use bytes::Bytes;
 use config::{Authority, Committee, PrimaryAddresses, WorkerAddresses};

narwhal/worker/src/worker.rs

@@ -2,12 +2,8 @@
 // Copyright (c) 2022, Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 use crate::{
-    batch_maker::BatchMaker,
-    helper::Helper,
-    primary_connector::PrimaryConnector,
-    processor::{Processor, SerializedBatchMessage},
-    quorum_waiter::QuorumWaiter,
-    synchronizer::Synchronizer,
+    batch_maker::BatchMaker, helper::Helper, primary_connector::PrimaryConnector,
+    processor::Processor, quorum_waiter::QuorumWaiter, synchronizer::Synchronizer,
 };
 use async_trait::async_trait;
 use bytes::Bytes;
@@ -39,6 +35,9 @@ pub type Round = u64;
 /// Indicates a serialized `WorkerPrimaryMessage` message.
 pub type SerializedWorkerPrimaryMessage = Vec<u8>;
 
+/// Indicates a serialized `WorkerMessage::Batch` message.
+pub type SerializedBatchMessage = Vec<u8>;
+
 /// The message exchanged between workers.
 #[derive(Debug, Serialize, Deserialize)]
 #[serde(bound(deserialize = "PublicKey: VerifyingKey"))]
@@ -290,6 +289,7 @@ struct WorkerReceiverHandler<PublicKey: VerifyingKey> {
 impl<PublicKey: VerifyingKey> MessageHandler for WorkerReceiverHandler<PublicKey> {
     async fn dispatch(&self, writer: &mut Writer, serialized: Bytes) -> Result<(), Box<dyn Error>> {
         // Deserialize and parse the message.
+        // TODO [issue #7]: Do some accounting to prevent bad actors from use all our resources.
         match bincode::deserialize(&serialized) {
             Ok(WorkerMessage::Batch(..)) => {
                 self.tx_processor