fix: increase channel size and pending cert interval

crates/topos-config/src/tce/synchronization.rs

@@ -23,7 +23,7 @@ impl Default for SynchronizationConfig {
 }
 
 impl SynchronizationConfig {
-    pub const INTERVAL_SECONDS: u64 = 10;
+    pub const INTERVAL_SECONDS: u64 = 30;
     pub const LIMIT_PER_SUBNET: usize = 100;
 
     const fn default_interval_seconds() -> u64 {

crates/topos-p2p/src/behaviour/gossip.rs

@@ -25,7 +25,7 @@ use crate::{constants, event::ComposedEvent, TOPOS_ECHO, TOPOS_GOSSIP, TOPOS_REA
 
 use super::HealthStatus;
 
-const MAX_BATCH_SIZE: usize = 10;
+const MAX_BATCH_SIZE: usize = 1024;
 
 pub struct Behaviour {
     batch_size: usize,
@@ -76,7 +76,7 @@ impl Behaviour {
             .unwrap_or(Ok(MAX_BATCH_SIZE))
             .unwrap();
         let gossipsub = gossipsub::ConfigBuilder::default()
-            .max_transmit_size(2 * 1024 * 1024)
+            .max_transmit_size(2 * 2048 * 2048)
             .validation_mode(gossipsub::ValidationMode::Strict)
             .message_id_fn(|msg_id| {
                 // Content based id

crates/topos-sequencer-subnet-runtime/src/proxy.rs

@@ -68,7 +68,7 @@ impl SubnetRuntimeProxy {
              address: {}, ",
             &config.http_endpoint, &config.ws_endpoint, &config.subnet_contract_address
         );
-        let (command_sender, mut command_rcv) = mpsc::channel::<SubnetRuntimeProxyCommand>(256);
+        let (command_sender, mut command_rcv) = mpsc::channel::<SubnetRuntimeProxyCommand>(1024);
         let ws_runtime_endpoint = config.ws_endpoint.clone();
         let http_runtime_endpoint = config.http_endpoint.clone();
         let subnet_contract_address = Arc::new(config.subnet_contract_address.clone());

crates/topos-tce-api/src/grpc/mod.rs

@@ -29,7 +29,7 @@ pub(crate) mod console;
 #[cfg(test)]
 mod tests;
 
-const DEFAULT_CHANNEL_STREAM_CAPACITY: usize = 100;
+const DEFAULT_CHANNEL_STREAM_CAPACITY: usize = 1024;
 
 pub(crate) mod builder;
 pub(crate) mod messaging;
@@ -272,7 +272,7 @@ impl ApiService for TceGrpcService {
             .map(move |message| Self::parse_stream(message, stream_id))
             .boxed();
 
-        let (command_sender, command_receiver) = mpsc::channel(2048);
+        let (command_sender, command_receiver) = mpsc::channel(6144);
 
         let (outbound_stream, rx) = mpsc::channel::<Result<(Option<Uuid>, OutboundMessage), Status>>(
             DEFAULT_CHANNEL_STREAM_CAPACITY,

crates/topos-tce-api/src/lib.rs

@@ -9,10 +9,10 @@ mod tests;
 
 pub(crate) mod constants {
     /// Constant size of every channel in the crate
-    pub(crate) const CHANNEL_SIZE: usize = 2048;
+    pub(crate) const CHANNEL_SIZE: usize = 6144;
 
     /// Constant size of every transient stream channel in the crate
-    pub(crate) const TRANSIENT_STREAM_CHANNEL_SIZE: usize = 1024;
+    pub(crate) const TRANSIENT_STREAM_CHANNEL_SIZE: usize = 2048;
 }
 pub use runtime::{
     error::RuntimeError, Runtime, RuntimeClient, RuntimeCommand, RuntimeContext, RuntimeEvent,

crates/topos-tce-broadcast/src/constant.rs

@@ -6,19 +6,19 @@ lazy_static! {
         std::env::var("TOPOS_DOUBLE_ECHO_COMMAND_CHANNEL_SIZE")
             .ok()
             .and_then(|s| s.parse().ok())
-            .unwrap_or(2048);
+            .unwrap_or(6144);
     /// Size of the channel between double echo and the task manager
     pub static ref BROADCAST_TASK_MANAGER_CHANNEL_SIZE: usize =
         std::env::var("TOPOS_BROADCAST_TASK_MANAGER_CHANNEL_SIZE")
             .ok()
             .and_then(|s| s.parse().ok())
-            .unwrap_or(20_480);
+            .unwrap_or(6144);
     /// Size of the channel to send protocol events from the double echo
     pub static ref PROTOCOL_CHANNEL_SIZE: usize =
         std::env::var("TOPOS_PROTOCOL_CHANNEL_SIZE")
             .ok()
             .and_then(|s| s.parse().ok())
-            .unwrap_or(2048);
+            .unwrap_or(6144);
     /// Capacity alert threshold for the double echo command channel
     pub static ref COMMAND_CHANNEL_CAPACITY: usize = COMMAND_CHANNEL_SIZE
         .checked_mul(10)

crates/topos-tce-broadcast/src/double_echo/mod.rs

@@ -55,7 +55,7 @@ pub struct DoubleEcho {
 }
 
 impl DoubleEcho {
-    pub const MAX_BUFFER_SIZE: usize = 2048;
+    pub const MAX_BUFFER_SIZE: usize = 6144;
 
     #[allow(clippy::too_many_arguments)]
     pub fn new(

crates/topos-tce-broadcast/src/task_manager/mod.rs

@@ -107,7 +107,7 @@ impl TaskManager {
     }
 
     pub async fn run(mut self, shutdown_receiver: CancellationToken) {
-        let mut interval = tokio::time::interval(Duration::from_secs(1));
+        let mut interval = tokio::time::interval(Duration::from_millis(15));
 
         loop {
             tokio::select! {

crates/topos-tce-proxy/src/client.rs

@@ -22,9 +22,9 @@ use topos_core::{
 use tracing::{debug, error, info, info_span, warn, Instrument, Span};
 use tracing_opentelemetry::OpenTelemetrySpanExt;
 
-const CERTIFICATE_OUTBOUND_CHANNEL_SIZE: usize = 100;
-const CERTIFICATE_INBOUND_CHANNEL_SIZE: usize = 100;
-const TCE_PROXY_COMMAND_CHANNEL_SIZE: usize = 100;
+const CERTIFICATE_OUTBOUND_CHANNEL_SIZE: usize = 2048;
+const CERTIFICATE_INBOUND_CHANNEL_SIZE: usize = 2048;
+const TCE_PROXY_COMMAND_CHANNEL_SIZE: usize = 2048;
 
 // Maximum backoff retry timeout in seconds (1 hour)
 const TCE_SUBMIT_CERTIFICATE_BACKOFF_TIMEOUT: Duration = Duration::from_secs(3600);

crates/topos-tce-proxy/src/worker.rs

@@ -19,11 +19,11 @@ pub struct TceProxyWorker {
 }
 
 impl TceProxyWorker {
-    /// Construct a new [`TceProxyWorker`] with a 128 items deep channel to send commands to and receive events from a TCE node on the given subnet.
+    /// Construct a new [`TceProxyWorker`] with a 2048 items deep channel to send commands to and receive events from a TCE node on the given subnet.
     /// The worker holds a [`crate::client::TceClient`]
     pub async fn new(config: TceProxyConfig) -> Result<(Self, Option<(Certificate, u64)>), Error> {
-        let (command_sender, mut command_rcv) = mpsc::channel::<TceProxyCommand>(128);
-        let (evt_sender, evt_rcv) = mpsc::channel::<TceProxyEvent>(128);
+        let (command_sender, mut command_rcv) = mpsc::channel::<TceProxyCommand>(2048);
+        let (evt_sender, evt_rcv) = mpsc::channel::<TceProxyEvent>(2048);
         let (tce_client_shutdown_channel, shutdown_receiver) =
             mpsc::channel::<oneshot::Sender<()>>(1);
 

crates/topos-tce-synchronizer/src/builder.rs

@@ -29,7 +29,7 @@ impl Default for SynchronizerBuilder {
             network_client: None,
             store: None,
             config: SynchronizationConfig::default(),
-            event_channel_size: 100,
+            event_channel_size: 1024,
             shutdown: None,
         }
     }

crates/topos-tce/src/app_context.rs

@@ -69,7 +69,7 @@ impl AppContext {
         validator_store: Arc<ValidatorStore>,
         api_context: RuntimeContext,
     ) -> (Self, mpsc::Receiver<Events>) {
-        let (events, receiver) = mpsc::channel(100);
+        let (events, receiver) = mpsc::channel(2048);
         (
             Self {
                 is_validator,