channel config: add slog buffer capacity, doc capacities in config (#56)

Author: drozdziak1

config/config.toml

@@ -21,7 +21,45 @@ key_store.root_path = "/path/to/keystore"
 
 ### Optional fields ###
 
-# Path to publisher identity keypair. When not found, the network will expect a remote-loaded keypair. see remote_keypair_loader options for details.
+# Channel capacities. These refer to async messaging channels
+# internally used by the agent's subroutines
+
+# Capacity of the channel used to broadcast shutdown events to all
+# components
+# channel_capacities.shutdown = 10000
+
+# Capacity of the channel used to send updates from the primary Oracle
+# to the Global Store
+# channel_capacities.primary_oracle_updates = 10000
+
+# Capacity of the channel used to send updates from the secondary
+# Oracle to the Global Store
+# channel_capacities.secondary_oracle_updates = 10000
+
+# Capacity of the channel the Pythd API Adapter uses to send lookup
+# requests to the Global Store
+# channel_capacities.global_store_lookup = 10000
+
+# Capacity of the channel the Pythd API Adapter uses to communicate
+# with the Local Store
+# channel_capacities.local_store_lookup = 10000
+
+# Capacity of the channel on which the Local Store receives messages
+# channel_capacities.local_store = 10000
+
+# Capacity of the channel on which the Pythd API Adapter receives
+# messages
+# channel_capacities.pythd_adapter = 10000
+
+# Capacity of the slog logging channel. Adjust this value if you see
+# complaints about channel capacity from slog
+# channel_capacities.logger_buffer = 10000
+
+
+# Path to publisher identity keypair. When the specified path is not
+# found on startup, the relevant primary/secondary network will expect
+# a remote-loaded keypair. See remote_keypair_loader options for
+# details.
 # key_store.publish_keypair_path = "publish_key_pair.json" # I exist, remote loading disabled
 # key_store.publish_keypair_path = "none" # I do not exist, remote loading activated for the network
 

src/agent.rs

@@ -3,7 +3,7 @@
 +--------------------------------+ +--------------------------------+
 |     RPC Node, e.g. Pythnet     | | RPC Node, e.g. Solana Mainnet  |
 +--------------------------------+ +--------------------------------+
-    |                     ^            ^                     |
+|                     ^            ^                     |
 +---|---------------------|------+ +---|---------------------|------+
 |   |    Primary Network  |      | |      Secondary Network  |      |
 |   v                     |      | |   |                     v      |
@@ -39,14 +39,14 @@ Publisher data write path:
 - The Adapter then transforms this into the Pyth SDK data structures and sends it to the Local Store.
 - The Local Store holds the latest price data the user has submitted for each price feed.
 - The Exporters periodically query the Local Store for the latest user-submitted data,
-  and send it to the RPC node.
+and send it to the RPC node.
 
 Publisher data read path:
 - The Oracles continually fetch data from the RPC node, and pass this to the Global Store.
 - The Global Store holds a unified view of the latest observed data from both networks, in the Pyth SDK data structures.
 - When a user queries for this data using the Pythd JRPC Websocket API, the Adapter fetches
-  the latest data from the Global Store. It transforms this from the Pyth SDK data structures into the
-  Pythd JRPC Websocket API data structures.
+the latest data from the Global Store. It transforms this from the Pyth SDK data structures into the
+Pythd JRPC Websocket API data structures.
 - The Pythd JRPC Websocket API then sends this data to the user.
 
 Remote Keypair Loading:
@@ -270,6 +270,8 @@ pub mod config {
         pub local_store:              usize,
         /// Capacity of the channel on which the Pythd API Adapter receives messages
         pub pythd_adapter:            usize,
+        /// Capacity of the slog logging channel. Adjust this value if you see complaints about channel capacity from slog
+        pub logger_buffer:            usize,
     }
 
     impl Default for ChannelCapacities {
@@ -282,6 +284,7 @@ pub mod config {
                 local_store_lookup:       10000,
                 local_store:              10000,
                 pythd_adapter:            10000,
+                logger_buffer:            10000,
             }
         }
     }

src/bin/agent.rs

@@ -1,5 +1,8 @@
 use {
-    anyhow::Result,
+    anyhow::{
+        Context,
+        Result,
+    },
     clap::Parser,
     pyth_agent::agent::{
         config::Config,
@@ -11,6 +14,7 @@ use {
         Drain,
         Logger,
     },
+    slog_async::Async,
     slog_envlogger::LogBuilder,
     std::{
         env,
@@ -28,29 +32,36 @@ struct Arguments {
 }
 
 #[tokio::main]
-async fn main() {
-    let logger = slog::Logger::root(
-        slog_async::Async::default(
-            LogBuilder::new(
-                slog_term::FullFormat::new(slog_term::TermDecorator::new().stdout().build())
-                    .build()
-                    .fuse(),
-            )
-            .parse(&env::var("RUST_LOG").unwrap_or("info".to_string()))
-            .build(),
-        )
-        .fuse(),
-        o!(),
-    );
-
-    if let Err(err) = start(logger.clone()).await {
+async fn main() -> Result<()> {
+    // Parse config early for logging channel capacity
+    let config = Config::new(Arguments::parse().config).context("Could not parse config")?;
+
+    // A plain slog drain that sits inside an async drain instance
+    let inner_drain = LogBuilder::new(
+        slog_term::FullFormat::new(slog_term::TermDecorator::new().stdout().build())
+            .build()
+            .fuse(), // Yell loud on logger internal errors
+    )
+    .parse(&env::var("RUST_LOG").unwrap_or("info".to_string()))
+    .build();
+
+    // The top level async drain
+    let async_drain = Async::new(inner_drain)
+        .chan_size(config.channel_capacities.logger_buffer)
+        .build()
+        .fuse();
+
+    let logger = slog::Logger::root(async_drain, o!());
+
+    if let Err(err) = start(config, logger.clone()).await {
         error!(logger, "{:#}", err; "error" => format!("{:?}", err));
+        return Err(err);
     }
+
+    Ok(())
 }
 
-async fn start(logger: Logger) -> Result<()> {
-    Agent::new(Config::new(Arguments::parse().config)?)
-        .start(logger)
-        .await;
+async fn start(config: Config, logger: Logger) -> Result<()> {
+    Agent::new(config).start(logger).await;
     Ok(())
 }