Merge mysten_infra into sui repo

Cargo.toml

@@ -1,6 +1,13 @@
 [workspace]
 resolver = "2"
 members = [
+    "crates/component", 
+    "crates/mysten-network", 
+    "crates/mysten-util-mem",
+    "crates/mysten-util-mem-derive", 
+    "crates/name-variant", 
+    "crates/prometheus-closure-metric", 
+    "crates/rccheck",
     "crates/sui",
     "crates/sui-adapter",
     "crates/sui-adapter-transactional-tests",
@@ -37,10 +44,25 @@ members = [
     "crates/sui-types",
     "crates/sui-verifier",
     "crates/sui-verifier-transactional-tests",
+    "crates/telemetry-subscribers",
     "crates/test-utils",
+    "crates/typed-store", 
+    "crates/typed-store-derive",
     "crates/workspace-hack",
     "crates/x",
-    "narwhal/config", "narwhal/consensus", "narwhal/crypto", "narwhal/dag", "narwhal/examples", "narwhal/executor", "narwhal/network", "narwhal/node", "narwhal/primary", "narwhal/storage", "narwhal/test-utils", "narwhal/types", "narwhal/worker",
+    "narwhal/config", 
+    "narwhal/consensus", 
+    "narwhal/crypto", 
+    "narwhal/dag", 
+    "narwhal/examples", 
+    "narwhal/executor", 
+    "narwhal/network", 
+    "narwhal/node", 
+    "narwhal/primary", 
+    "narwhal/storage", 
+    "narwhal/test-utils", 
+    "narwhal/types", 
+    "narwhal/worker",
 ]
 
 [workspace.package]
@@ -70,14 +92,6 @@ opt-level = 1
 # Dependencies that should be kept in sync through the whole workspace
 [workspace.dependencies]
 
-# github.com/MystenLabs/mysten-infra dependencies
-typed-store = "0.4.0"
-typed-store-derive = "0.3.0"
-telemetry-subscribers = { version = "0.2.0", features = ["jaeger", "tokio-console"] }
-mysten-network = { git = "https://github.com/MystenLabs/mysten-infra.git", rev = "fee7617899ac605f4aaf77312a394751ae577546" }
-name-variant = "0.1.0"
-store = { version = "0.4.0", package = "typed-store" }
-
 # Move dependencies
 move-binary-format = { git = "https://github.com/move-language/move", rev = "3ac0cd7b78cc2a6d470b7fefbfd82b0c044ef576" }
 move-bytecode-utils = { git = "https://github.com/move-language/move", rev = "3ac0cd7b78cc2a6d470b7fefbfd82b0c044ef576" }

crates/component/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "component"
+version = "0.1.0"
+license = "Apache-2.0"
+authors = ["Laura Makdah <laura@mystenlabs.com>"]
+description = "library for resilient component supervision"
+repository = "https://github.com/mystenlabs/mysten-infra"
+edition = "2021"
+publish = false
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+async-trait = "0.1.56"
+eyre = "0.6.8"
+tokio = { version = "1.21.2", features = ["sync", "macros", "rt", "rt-multi-thread"] }
+workspace-hack.workspace = true

crates/component/README.md

@@ -0,0 +1,47 @@
+# Resilient Components
+
+This library provides supervision of components that should be consistently running without downtime.
+We automatically restart a component when it stops running, and provide a contract for the component to
+communicate with the supervisor to signal when a situation arises that requires a restart.
+
+
+## Use
+
+For each component that needs to be registered with supervision, we will instantiate a `Supervisor` object to do supervision for that component.
+The supervisor should be initialized by passing in an object that implements the `Manageable` trait. To start the component,
+call the spawn method and await.
+```asm
+ let supervisor = Supervisor::new(stream_component);
+
+ supervisor.spawn().await?
+```
+
+A user-provided async function will be called by the supervisor to start a tokio task. We wrap the tokio task with communication channels, and one end of the channels will be passed into the user-provided function which will run the component, and the other ends of the channels will be held by the supervisor to receive irrecoverable error messages and to send cancellation signals to the tokio task.
+
+In order to have a component that gets managed, a user needs to implement the following async functions on their struct to be implementing the Manageable trait:
+
+1. start - This function is responsible for launching the tokio task that needs parental supervision and panic-tolerance. It will have as input a channel sender for passing irrecoverable errors to the supervisor, as well as a channel receiver that will be listened to after sending a message to the supervisor to receive an ack that the message was received. The function should then return on reception of the cancellation signal.
+2. handle_irrecoverable - This function will be called upon receiving an irrecoverable error. This allows the user to decide what to log or alert, and what actions to take, if any, for this particular component in terms of resource cleanup. After this function is called, the component is restarted again via the start function above.
+
+After these functions are implemented, a user can create a Supervisor object with the constructor and then call spawn on the supervisor. This will run the supervision on the component and will consistently ensure that it restarted after any irrecoverable errors.
+
+Note that since we never expect these tasks to complete, the user does not need to call join on these handles created in the start function. The constantly running supervision of the task will also handle the event that the async task execution completes by ending supervision, although this is not an expected use case.
+
+
+## Example
+
+The executable example is used as an integration test for this library. To run
+
+You can run it with
+```
+cargo run --example stream_example
+```
+and then ctrl + c to stop the component.
+
+This should generate the following output.
+```
+starting component task
+Received irrecoverable error: missing something required
+starting component task
+terminating component task
+```

crates/component/examples/stream_example.rs

@@ -0,0 +1,152 @@
+// Copyright (c) Mysten Labs, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+extern crate component;
+
+use async_trait::async_trait;
+use component::{IrrecoverableError, Manageable, Supervisor};
+use eyre::eyre;
+use std::cmp::min;
+use std::sync::Once;
+use tokio::sync::mpsc::Sender;
+use tokio::sync::oneshot::Receiver as oneshotReceiver;
+use tokio::task::JoinHandle;
+
+static mut SHOULD_FAIL: bool = true;
+static FIXER: Once = Once::new();
+
+fn fix() {
+    FIXER.call_once(|| unsafe {
+        SHOULD_FAIL = false;
+    })
+}
+
+/// We create two structs, an empty struct as a component that will contain functions
+/// that create and fully encapsulate an instance of the actual type.
+/// An instance of the actual type needs to be instantiated inside the supervised task
+/// in order to have the correct lifetime.
+pub struct MockTcpStreamComponent {}
+
+pub struct MockTcpStream {
+    read_data: Vec<u8>,
+}
+
+impl MockTcpStream {
+    pub fn new() -> Self {
+        let read_data = Vec::new();
+        MockTcpStream { read_data }
+    }
+
+    /// This function will fail on the first call and then succeed on preceding calls to create
+    /// a situation where we have an irrecoverable error.
+    fn mock_read(&self, buf: &mut [u8]) -> Result<usize, eyre::Report> {
+        // failure should happen once
+        unsafe {
+            if SHOULD_FAIL {
+                fix();
+                return Result::Err(eyre!("Could not read from stream."));
+            }
+        }
+
+        let size: usize = min(self.read_data.len(), buf.len());
+        buf[..size].copy_from_slice(&self.read_data[..size]);
+        Ok(size)
+    }
+}
+
+impl Default for MockTcpStream {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl MockTcpStreamComponent {
+    /// This is a function that should run continuously doing some operation, here we are
+    /// continuously listening on a mocked TCP Stream. This is ultimately the function that we
+    /// are supervising.
+    /// Inside this function we first initialize any state that will be used in this component so
+    /// that the scope is also correctly reset on a restart.
+    ///
+    /// This would be an excellent place to also add a scopeguard with a defer_panic so that if the
+    /// component panics without a user caught irrecoverable error, a descriptive error message and/
+    /// or stacktrace can be forwarded to the supervisor.
+    pub async fn listen(
+        tx_irrecoverable: Sender<eyre::Report>,
+        rx_cancellation: oneshotReceiver<()>,
+    ) {
+        // Initialize the concrete type
+        let m_tcp = MockTcpStream::new();
+
+        loop {
+            let mut buf = [0; 10];
+            match m_tcp.mock_read(&mut buf) {
+                Ok(_) => {} // process
+                Err(_) => {
+                    let e = eyre!("missing something required");
+                    tx_irrecoverable
+                        .send(e)
+                        .await
+                        .expect("Could not send irrecoverable signal.");
+                    wait_for_cancellation(rx_cancellation).await;
+                    return;
+                }
+            };
+        }
+    }
+}
+
+/// Wait for the cancellation signal in order to ensure that the message we sent to the
+/// supervisor was received before we return which causes the join handle to complete. If we
+/// were to return immediately, there would be no guarantee the message we send will be received.
+async fn wait_for_cancellation(rx_cancellation: oneshotReceiver<()>) {
+    loop {
+        tokio::select! {
+            _ = rx_cancellation => {
+                println!("terminating component task");
+                break;
+            }
+        }
+    }
+}
+
+#[async_trait]
+impl Manageable for MockTcpStreamComponent {
+    #[allow(clippy::async_yields_async)]
+    /// The start function spawns a tokio task supplied with a function that
+    /// should be constantly running.
+    async fn start(
+        &self,
+        tx_irrecoverable: Sender<eyre::Report>,
+        rx_cancellation: oneshotReceiver<()>,
+    ) -> tokio::task::JoinHandle<()> {
+        println!("starting component task");
+        let handle: JoinHandle<()> = tokio::spawn(Self::listen(tx_irrecoverable, rx_cancellation));
+        handle
+    }
+
+    /// Implement this function to log the error messages or take any task-specific action such as
+    /// closing a file or terminating children tasks.
+    fn handle_irrecoverable(
+        &mut self,
+        irrecoverable: IrrecoverableError,
+    ) -> Result<(), eyre::Report> {
+        println!("Received irrecoverable error: {irrecoverable}");
+        Ok(())
+    }
+}
+
+#[tokio::main]
+pub async fn main() -> Result<(), eyre::Report> {
+    // Create a component
+    let stream_component = MockTcpStreamComponent {};
+
+    // Create a supervisor for the component
+    let supervisor = Supervisor::new(stream_component);
+
+    // Spawn the supervisor to start the component and supervision.
+    match supervisor.spawn().await {
+        Ok(_) => {}
+        Err(e) => println!("Got this error {:?}", e),
+    };
+    Ok(())
+}