concurrent action dispatch

aquila.yml

@@ -47,6 +47,9 @@ grpc:
   # Expose the standard gRPC health service.
   health_service: false
 
+  # Maximum number of post-logon messages handled concurrently per action stream.
+  action_transfer_concurrency_limit: 256
+
 runtime_status:
   # Frequency of Aquila's own heartbeat to the Sagittarius gateway, in minutes.
   heartbeat_interval_minutes: 5

docs/installation.mdx

@@ -101,6 +101,7 @@ provide a more specific runtime filter, for example `RUST_LOG=aquila::server=tra
 | `grpc.host` | Aquila gRPC bind host. |
 | `grpc.port` | Aquila gRPC bind port. |
 | `grpc.health_service` | Enables the gRPC health service. |
+| `grpc.action_transfer_concurrency_limit` | Maximum concurrent post-logon handlers per action stream (default: `256`). |
 | `runtime_status.not_responding_after_secs` | Heartbeat timeout before `not_responding`. |
 | `runtime_status.stopped_after_not_responding_secs` | Additional timeout before `stopped`. |
 | `runtime_status.monitor_interval_secs` | Heartbeat monitor interval. |

src/configuration/config/display.rs

@@ -50,6 +50,11 @@ impl fmt::Display for Config {
             "    Health service: {}",
             self.grpc.health_service
         )?;
+        writeln!(
+            formatter,
+            "    Action transfer concurrency: {}",
+            self.grpc.action_transfer_concurrency_limit
+        )?;
         writeln!(formatter, "  Static mode")?;
         writeln!(formatter, "    Flow path: {}", self.static_config.flow_path)?;
         writeln!(formatter, "  Dynamic mode")?;
@@ -93,6 +98,7 @@ mod tests {
         assert!(output.starts_with("Aquila configuration\n"));
         assert!(output.contains("  Environment: development"));
         assert!(output.contains("    Address:   127.0.0.1:8081"));
+        assert!(output.contains("    Action transfer concurrency: 256"));
         assert!(output.contains("    Request timeout: 5s"));
         assert!(output.contains("    Backend token:   [FILTERED]"));
         assert!(!output.contains("super-secret"));

src/configuration/config/mod.rs

@@ -74,6 +74,9 @@ pub struct Grpc {
     pub host: String,
     pub port: u16,
     pub health_service: bool,
+    /// Maximum number of post-logon messages processed concurrently for one
+    /// action transfer stream.
+    pub action_transfer_concurrency_limit: usize,
 }
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
@@ -138,6 +141,7 @@ impl Default for Grpc {
             host: "127.0.0.1".into(),
             port: 8081,
             health_service: false,
+            action_transfer_concurrency_limit: 256,
         }
     }
 }
@@ -261,4 +265,25 @@ mod tests {
     fn opentelemetry_default_service_name_is_aquila() {
         assert_eq!(Config::default().opentelemetry.service_name, "aquila");
     }
+
+    #[test]
+    fn action_transfer_concurrency_limit_is_configurable() {
+        let config: Config = ConfigLoader::builder()
+            .add_source(
+                ConfigLoader::try_from(&Config::default())
+                    .expect("default configuration should serialize"),
+            )
+            .set_override("grpc.action_transfer_concurrency_limit", 17)
+            .expect("concurrency override should apply")
+            .build()
+            .expect("configuration should build")
+            .try_deserialize()
+            .expect("configuration should deserialize");
+
+        assert_eq!(
+            Config::default().grpc.action_transfer_concurrency_limit,
+            256
+        );
+        assert_eq!(config.grpc.action_transfer_concurrency_limit, 17);
+    }
 }