feat: concurrent da_dispatcher (#288)

* initial commit

* use Notify for a more deterministic approach

* replace atomic for mutex

* move const to config

core/lib/config/src/configs/da_dispatcher.rs

@@ -6,6 +6,7 @@ pub const DEFAULT_POLLING_INTERVAL_MS: u32 = 5000;
 pub const DEFAULT_MAX_ROWS_TO_DISPATCH: u32 = 100;
 pub const DEFAULT_MAX_RETRIES: u16 = 5;
 pub const DEFAULT_USE_DUMMY_INCLUSION_DATA: bool = false;
+pub const DEFAULT_MAX_CONCURRENT_REQUESTS: u32 = 100;
 
 #[derive(Debug, Clone, PartialEq, Deserialize)]
 pub struct DADispatcherConfig {
@@ -19,6 +20,8 @@ pub struct DADispatcherConfig {
     // TODO: run a verification task to check if the L1 contract expects the inclusion proofs to
     // avoid the scenario where contracts expect real proofs, and server is using dummy proofs.
     pub use_dummy_inclusion_data: Option<bool>,
+    /// The maximun number of concurrent request to send to the DA server.
+    pub max_concurrent_requests: Option<u32>,
 }
 
 impl DADispatcherConfig {
@@ -28,6 +31,7 @@ impl DADispatcherConfig {
             max_rows_to_dispatch: Some(DEFAULT_MAX_ROWS_TO_DISPATCH),
             max_retries: Some(DEFAULT_MAX_RETRIES),
             use_dummy_inclusion_data: Some(DEFAULT_USE_DUMMY_INCLUSION_DATA),
+            max_concurrent_requests: Some(DEFAULT_MAX_CONCURRENT_REQUESTS),
         }
     }
 

core/lib/config/src/testonly.rs

@@ -948,6 +948,7 @@ impl Distribution<configs::da_dispatcher::DADispatcherConfig> for EncodeDist {
             max_rows_to_dispatch: self.sample(rng),
             max_retries: self.sample(rng),
             use_dummy_inclusion_data: self.sample(rng),
+            max_concurrent_requests: self.sample(rng),
         }
     }
 }

core/lib/env_config/src/da_dispatcher.rs

@@ -21,12 +21,14 @@ mod tests {
         interval: u32,
         rows_limit: u32,
         max_retries: u16,
+        max_concurrent_requests: u32,
     ) -> DADispatcherConfig {
         DADispatcherConfig {
             polling_interval_ms: Some(interval),
             max_rows_to_dispatch: Some(rows_limit),
             max_retries: Some(max_retries),
             use_dummy_inclusion_data: Some(true),
+            max_concurrent_requests: Some(max_concurrent_requests),
         }
     }
 
@@ -38,9 +40,10 @@ mod tests {
             DA_DISPATCHER_MAX_ROWS_TO_DISPATCH=60
             DA_DISPATCHER_MAX_RETRIES=7
             DA_DISPATCHER_USE_DUMMY_INCLUSION_DATA="true"
+            DA_DISPATCHER_MAX_CONCURRENT_REQUESTS=10
         "#;
         lock.set_env(config);
         let actual = DADispatcherConfig::from_env().unwrap();
-        assert_eq!(actual, expected_da_layer_config(5000, 60, 7));
+        assert_eq!(actual, expected_da_layer_config(5000, 60, 7, 10));
     }
 }

core/lib/protobuf_config/src/da_dispatcher.rs

@@ -12,6 +12,7 @@ impl ProtoRepr for proto::DataAvailabilityDispatcher {
             max_rows_to_dispatch: self.max_rows_to_dispatch,
             max_retries: self.max_retries.map(|x| x as u16),
             use_dummy_inclusion_data: self.use_dummy_inclusion_data,
+            max_concurrent_requests: self.max_concurrent_requests,
         })
     }
 
@@ -21,6 +22,7 @@ impl ProtoRepr for proto::DataAvailabilityDispatcher {
             max_rows_to_dispatch: this.max_rows_to_dispatch,
             max_retries: this.max_retries.map(Into::into),
             use_dummy_inclusion_data: this.use_dummy_inclusion_data,
+            max_concurrent_requests: this.max_concurrent_requests,
         }
     }
 }

core/lib/protobuf_config/src/proto/config/da_dispatcher.proto

@@ -7,4 +7,5 @@ message DataAvailabilityDispatcher {
   optional uint32 max_rows_to_dispatch = 2;
   optional uint32 max_retries = 3;
   optional bool use_dummy_inclusion_data = 4;
+  optional uint32 max_concurrent_requests = 5;
 }