rebase main

Author: glihm

.github/workflows/release-dispatch.yml

@@ -26,12 +26,12 @@ jobs:
       - run: |
           VERSION=${{ inputs.version }}
           VERSION=${VERSION#v}
-          cargo install cargo-release
+          cargo install cargo-release --locked
           cargo release version $VERSION --execute --no-confirm && cargo release replace --execute --no-confirm
 
       - id: version_info
         run: |
-          cargo install cargo-get
+          cargo install cargo-get --locked
           echo "version=$(cargo get workspace.package.version)" >> $GITHUB_OUTPUT
 
       - uses: peter-evans/create-pull-request@v7

Cargo.lock

@@ -21,7 +21,7 @@ edition = "2021"
 license = "Apache-2.0"
 license-file = "LICENSE"
 repository = "https://github.com/dojoengine/torii"
-version = "1.8.7"
+version = "1.8.8"
 
 [profile.performance]
 codegen-units = 1

Cargo.toml

@@ -62,8 +62,10 @@ where
     /// Publish an update message that all subscription streams can receive.
     pub fn publish(msg: Update<T>) {
         with_senders::<Update<T>, _, _>(|senders| {
+            // Use unbounded_send instead of start_send for better performance
+            // Unbounded channels never block or fail
             for (_, sender) in senders.0.iter_mut() {
-                sender.start_send(msg.clone()).ok();
+                let _ = sender.unbounded_send(msg.clone());
             }
         });
     }

crates/broker/src/memory.rs

@@ -26,7 +26,7 @@ pub const DEFAULT_RELAY_WEBRTC_PORT: u16 = 9091;
 pub const DEFAULT_RELAY_WEBSOCKET_PORT: u16 = 9092;
 pub const DEFAULT_GRPC_ADDR: IpAddr = IpAddr::V4(Ipv4Addr::LOCALHOST);
 pub const DEFAULT_GRPC_PORT: u16 = 50051;
-pub const DEFAULT_GRPC_SUBSCRIPTION_BUFFER_SIZE: usize = 256;
+pub const DEFAULT_GRPC_SUBSCRIPTION_BUFFER_SIZE: usize = 16384;
 pub const DEFAULT_GRPC_TCP_KEEPALIVE_SECS: u64 = 60;
 pub const DEFAULT_GRPC_HTTP2_KEEPALIVE_INTERVAL_SECS: u64 = 30;
 pub const DEFAULT_GRPC_HTTP2_KEEPALIVE_TIMEOUT_SECS: u64 = 10;

crates/cli/src/options.rs

@@ -119,48 +119,37 @@ impl<P: Provider + Sync> DojoWorld<P> {
         let achievement_progression_manager =
             Arc::new(AchievementProgressionManager::new(config.clone()));
 
-        // Spawn subscription services on the main runtime
-        // They use try_send and non-blocking operations to avoid starving other tasks
+        // Spawn subscription services - each polls its broker stream and dispatches to subscribers
         tokio::spawn(subscriptions::entity::Service::new(Arc::clone(
             &entity_manager,
         )));
-
         tokio::spawn(subscriptions::event_message::Service::new(Arc::clone(
             &event_message_manager,
         )));
-
         tokio::spawn(subscriptions::event::Service::new(Arc::clone(
             &event_manager,
         )));
-
         tokio::spawn(subscriptions::contract::Service::new(Arc::clone(
             &contract_manager,
         )));
-
-        tokio::spawn(subscriptions::token_balance::Service::new(Arc::clone(
-            &token_balance_manager,
-        )));
-
         tokio::spawn(subscriptions::token::Service::new(Arc::clone(
             &token_manager,
         )));
-
+        tokio::spawn(subscriptions::token_balance::Service::new(Arc::clone(
+            &token_balance_manager,
+        )));
         tokio::spawn(subscriptions::token_transfer::Service::new(Arc::clone(
             &token_transfer_manager,
         )));
-
         tokio::spawn(subscriptions::transaction::Service::new(Arc::clone(
             &transaction_manager,
         )));
-
         tokio::spawn(subscriptions::aggregation::Service::new(Arc::clone(
             &aggregation_manager,
         )));
-
         tokio::spawn(subscriptions::activity::Service::new(Arc::clone(
             &activity_manager,
         )));
-
         tokio::spawn(subscriptions::achievement::Service::new(Arc::clone(
             &achievement_progression_manager,
         )));
@@ -1298,8 +1287,9 @@ pub async fn new<P: Provider + Sync + Send + 'static>(
         .tcp_keepalive(Some(tcp_keepalive))
         .http2_keepalive_interval(Some(http2_keepalive_interval))
         .http2_keepalive_timeout(Some(http2_keepalive_timeout))
-        .initial_stream_window_size(Some(1024 * 1024))
-        .initial_connection_window_size(Some(1024 * 1024 * 10))
+        // Enable adaptive flow control for optimal streaming performance
+        // This automatically adjusts window sizes based on throughput and prevents flow control bottlenecks
+        .http2_adaptive_window(Some(true))
         // Should be enabled by default.
         .tcp_nodelay(true)
         .layer(