Fix race condition in maximum clients per call check

backend/src/call.rs

@@ -424,6 +424,7 @@ struct CallInfo {
     persist_approval_for_all_users_who_join: bool,
     initial_target_send_rate: DataRate,
     default_requested_max_send_rate: DataRate,
+    max_clients: usize,
 }
 
 pub struct CreateCallArgs {
@@ -441,6 +442,7 @@ pub struct CreateCallArgs {
     pub persist_approval_for_all_users_who_join: bool,
     pub endorsement_issuer: Option<Arc<Mutex<EndorsementIssuer>>>,
     pub drop_fragmentable_updates: bool,
+    pub max_clients: usize,
 }
 
 impl Call {
@@ -460,6 +462,7 @@ impl Call {
             persist_approval_for_all_users_who_join,
             endorsement_issuer,
             drop_fragmentable_updates,
+            max_clients,
         } = create_args;
 
         let loggable_call_id = LoggableCallId::from(&call_id);
@@ -489,6 +492,7 @@ impl Call {
             persist_approval_for_all_users_who_join,
             initial_target_send_rate,
             default_requested_max_send_rate,
+            max_clients,
         };
 
         Self { inner, call_info }
@@ -523,11 +527,6 @@ impl Call {
         self.inner.lock().size()
     }
 
-    #[inline(always)]
-    pub fn size_including_pending_clients(&self) -> usize {
-        self.inner.lock().size_including_pending_clients()
-    }
-
     #[inline(always)]
     pub fn is_approved_users_busy(&self) -> bool {
         self.inner.lock().approved_users.is_busy()
@@ -880,10 +879,6 @@ impl CallInner {
         self.clients.len()
     }
 
-    fn size_including_pending_clients(&self) -> usize {
-        self.clients.len() + self.pending_clients.len()
-    }
-
     pub fn peak_call_size(&self) -> usize {
         self.call_stats.peak_call_size
     }
@@ -950,6 +945,11 @@ impl CallInner {
         call_info: &CallInfo,
         now: Instant,
     ) -> ClientStatus {
+        let client_count = self.clients.len() + self.pending_clients.len();
+        if client_count >= call_info.max_clients {
+            return ClientStatus::Rejected;
+        }
+
         let member_ciphertext = (&user_id).try_into().ok();
         let pending_client = NonParticipantClient {
             demux_id,
@@ -4945,6 +4945,8 @@ mod call_tests {
         );
     }
 
+    const MAX_CLIENTS_IN_CALL: usize = usize::MAX;
+
     fn create_call(
         call_id: &[u8],
         now: Instant,
@@ -4969,6 +4971,7 @@ mod call_tests {
             persist_approval_for_all_users_who_join: false,
             endorsement_issuer: None,
             drop_fragmentable_updates: false,
+            max_clients: MAX_CLIENTS_IN_CALL,
         })
     }
 
@@ -4991,6 +4994,7 @@ mod call_tests {
             persist_approval_for_all_users_who_join: false,
             endorsement_issuer: None,
             drop_fragmentable_updates: false,
+            max_clients: MAX_CLIENTS_IN_CALL,
         })
     }
 

backend/src/connection.rs

@@ -1220,6 +1220,8 @@ mod connection_tests {
         transportcc as tcc,
     };
 
+    const MAX_CLIENTS_IN_CALL: usize = 8;
+
     fn new_call(
         call_id: &[u8],
         now: Instant,
@@ -1244,6 +1246,7 @@ mod connection_tests {
             persist_approval_for_all_users_who_join: false,
             endorsement_issuer: None,
             drop_fragmentable_updates: false,
+            max_clients: MAX_CLIENTS_IN_CALL,
         })
     }
 

backend/src/sfu.rs

@@ -704,6 +704,7 @@ impl Sfu {
                     .persist_approval_for_all_users_who_join,
                 endorsement_issuer: self.endorsement_issuer.clone(),
                 drop_fragmentable_updates: true,
+                max_clients: self.config.max_clients_per_call as usize,
             })
         });
 
@@ -711,10 +712,6 @@ impl Sfu {
             return Err(SfuError::DuplicateDemuxIdDetected);
         }
 
-        if call.size_including_pending_clients() == self.config.max_clients_per_call as usize {
-            return Err(SfuError::TooManyClients);
-        }
-
         info!(
             "call_id: {} adding demux_id: {}, join region {}",
             loggable_call_id,
@@ -743,6 +740,10 @@ impl Sfu {
             Instant::now(), // Now after taking the lock
         );
 
+        if client_status == ClientStatus::Rejected {
+            return Err(SfuError::TooManyClients);
+        }
+
         // ACKs can be sent from any SSRC that the client is configured to send with, which includes the
         // video base layer, so use that.
         let ack_ssrc = call::LayerId::Video0.to_ssrc(demux_id);

common/src/client_status.rs

@@ -13,6 +13,7 @@ pub enum ClientStatus {
     Active,
     Pending,
     Blocked,
+    Rejected,
 }
 
 impl fmt::Display for ClientStatus {
@@ -21,6 +22,7 @@ impl fmt::Display for ClientStatus {
             ClientStatus::Active => "ACTIVE",
             ClientStatus::Pending => "PENDING",
             ClientStatus::Blocked => "BLOCKED",
+            ClientStatus::Rejected => "REJECTED",
         })
     }
 }