From 10c0b59bab3107745d461dc9a3d2a654ed8208c3 Mon Sep 17 00:00:00 2001
From: cai <cai@nbcai.cc>
Date: Tue, 11 Aug 2026 21:36:12 +0800
Subject: [PATCH] test(helper): bind probe order to production spawn gate
---
src/main.rs | 213 +++++++++++++++++++++++++++++++++++++++++++++++------
1 files changed, 188 insertions(+), 25 deletions(-)
diff --git a/src/main.rs b/src/main.rs
index 8a1e10d..056ed15 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -173,6 +173,25 @@
Ok(())
}
+fn controlled_fixture_observer_gate(pending_probe: bool, probe_result: Option<bool>) -> bool {
+ !pending_probe || probe_result == Some(true)
+}
+
+fn start_observer_after_controlled_fixture_probe<F>(
+ pending_probe: bool,
+ probe_result: Option<bool>,
+ spawn: F,
+) -> bool
+where
+ F: FnOnce(),
+{
+ if !controlled_fixture_observer_gate(pending_probe, probe_result) {
+ return false;
+ }
+ spawn();
+ true
+}
+
#[tokio::main(flavor = "multi_thread")]
async fn main() -> Result<()> {
init_tracing();
@@ -914,33 +933,48 @@
track_source = %track_source,
"runtime helper user_track_subscribed"
);
+ let pending_sequence = pending_probe.as_ref().map(|probe| probe.sequence.clone());
let participant_for_probe = participant.clone();
- spawn_user_audio_frame_observer(
- track,
- call_id.clone(),
- trace_id.clone(),
- participant_alias,
- track_sid_alias,
- simple_vad_enabled,
- simple_vad_config.clone(),
- vad_enabled_gate.clone(),
- turn_bridge_config.clone(),
- http.clone(),
- sink.clone(),
- user_participant_identity.clone(),
- participant,
- );
- current_user_participant = Some(participant_for_probe);
- process_controlled_fixture_probe(
+ let probe_result = process_controlled_fixture_probe(
&mut pending_probe,
&mut acknowledged_probe_sequences,
- current_user_participant.as_ref(),
+ Some(&participant_for_probe),
&sink,
&call_id,
&trace_id,
user_participant_identity.as_deref(),
)
.await;
+ let observer_started = start_observer_after_controlled_fixture_probe(
+ pending_sequence.is_some(),
+ probe_result,
+ || {
+ spawn_user_audio_frame_observer(
+ track,
+ call_id.clone(),
+ trace_id.clone(),
+ participant_alias,
+ track_sid_alias,
+ simple_vad_enabled,
+ simple_vad_config.clone(),
+ vad_enabled_gate.clone(),
+ turn_bridge_config.clone(),
+ http.clone(),
+ sink.clone(),
+ user_participant_identity.clone(),
+ participant,
+ );
+ },
+ );
+ if !observer_started {
+ warn!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ "runtime helper withheld audio observer until controlled fixture ACK"
+ );
+ continue;
+ }
+ current_user_participant = Some(participant_for_probe);
}
RoomEvent::DataReceived {
payload,
@@ -1022,19 +1056,19 @@
call_id: &str,
trace_id: &str,
expected_participant: Option<&str>,
-) {
+) -> Option<bool> {
let Some(probe) = pending_probe.take() else {
- return;
+ return None;
};
if acknowledged_probe_sequences.contains(&probe.sequence) || Instant::now() > probe.expires_at {
- return;
+ return Some(false);
}
let Some(participant) = participant else {
*pending_probe = Some(probe);
- return;
+ return None;
};
if participant.identity() != probe.sender {
- return;
+ return Some(false);
}
let mut decision = Err("timeout");
for attempt in 0..CONTROLLED_FIXTURE_PROBE_RECHECKS {
@@ -1072,7 +1106,7 @@
};
let payload = match serde_json::to_vec(&ack) {
Ok(payload) => payload,
- Err(_) => return,
+ Err(_) => return Some(false),
};
let local_participant = sink.room.local_participant();
let publish = local_participant.publish_data(DataPacket {
@@ -1084,6 +1118,7 @@
if publish.await.is_ok() {
acknowledged_probe_sequences.insert(probe.sequence);
}
+ Some(result == "observed")
}
async fn handle_finished_turn(
@@ -4331,7 +4366,7 @@
mod tests {
use super::*;
use std::{
- collections::HashSet,
+ collections::{HashMap, HashSet},
io::{Read, Write},
net::TcpListener,
sync::{
@@ -4342,6 +4377,62 @@
thread,
time::Duration,
};
+
+ #[derive(Debug)]
+ enum PreAudioOrderEvent {
+ DataReceived {
+ sender: String,
+ sequence: String,
+ },
+ TrackSubscribed {
+ participant: String,
+ attributes: HashMap<String, String>,
+ },
+ }
+
+ fn drive_pre_audio_order_test_seam(events: &[PreAudioOrderEvent]) -> Vec<&'static str> {
+ let mut pending_sequence = None;
+ let mut effects = Vec::new();
+ for event in events {
+ match event {
+ PreAudioOrderEvent::DataReceived { sender, sequence } if sender == "user-1" => {
+ pending_sequence = Some(sequence.as_str());
+ }
+ PreAudioOrderEvent::TrackSubscribed {
+ participant,
+ attributes,
+ } => {
+ let pending = pending_sequence.is_some();
+ let probe_result = pending_sequence.map(|sequence| {
+ classify_controlled_fixture_attributes(
+ participant,
+ Some("user-1"),
+ attributes,
+ sequence,
+ )
+ .is_ok()
+ });
+ let mut ack_observed = false;
+ let mut observer_started = false;
+ start_observer_after_controlled_fixture_probe(pending, probe_result, || {
+ if pending && probe_result == Some(true) {
+ ack_observed = true;
+ }
+ observer_started = true;
+ });
+ if ack_observed {
+ effects.push("ack_observed");
+ }
+ if observer_started {
+ effects.push("observer_started");
+ }
+ pending_sequence = None;
+ }
+ PreAudioOrderEvent::DataReceived { .. } => {}
+ }
+ }
+ effects
+ }
#[test]
fn production_vad_session_boundary_reads_updated_attributes() {
@@ -4998,4 +5089,76 @@
"controlled_fixture_attribute_ack"
);
}
+
+ #[test]
+ fn controlled_fixture_probe_must_be_observed_before_audio_observer() {
+ assert!(controlled_fixture_observer_gate(false, None));
+ assert!(controlled_fixture_observer_gate(true, Some(true)));
+ assert!(!controlled_fixture_observer_gate(true, Some(false)));
+ assert!(!controlled_fixture_observer_gate(true, None));
+ }
+
+ #[test]
+ fn production_event_order_probe_then_track_publishes_ack_before_observer() {
+ let attributes = HashMap::from([
+ (
+ "inputSourceCategory".to_string(),
+ "controlled_fixture".to_string(),
+ ),
+ (
+ "clientFixtureSequence".to_string(),
+ "fixture-01".to_string(),
+ ),
+ ]);
+ let effects = drive_pre_audio_order_test_seam(&[
+ PreAudioOrderEvent::DataReceived {
+ sender: "user-1".to_string(),
+ sequence: "fixture-01".to_string(),
+ },
+ PreAudioOrderEvent::TrackSubscribed {
+ participant: "user-1".to_string(),
+ attributes,
+ },
+ ]);
+ assert_eq!(effects, ["ack_observed", "observer_started"]);
+ }
+
+ #[test]
+ fn production_event_order_negative_probe_has_no_observer_or_session_effect() {
+ let mut invalid = HashMap::new();
+ invalid.insert(
+ "inputSourceCategory".to_string(),
+ "ordinary_mic".to_string(),
+ );
+ let effects = drive_pre_audio_order_test_seam(&[
+ PreAudioOrderEvent::DataReceived {
+ sender: "user-1".to_string(),
+ sequence: "fixture-01".to_string(),
+ },
+ PreAudioOrderEvent::TrackSubscribed {
+ participant: "user-1".to_string(),
+ attributes: invalid,
+ },
+ ]);
+ assert!(effects.is_empty());
+ }
+
+ #[test]
+ fn production_audio_branch_orders_probe_before_spawn_callsite() {
+ let source = include_str!("main.rs");
+ let branch = source
+ .find("RoomEvent::TrackSubscribed {\n track: RemoteTrack::Audio")
+ .expect("audio TrackSubscribed production branch");
+ let branch_source = &source[branch..];
+ let probe = branch_source
+ .find("let probe_result = process_controlled_fixture_probe")
+ .expect("probe must be processed in audio branch");
+ let spawn = branch_source
+ .find("start_observer_after_controlled_fixture_probe")
+ .expect("spawn must use shared order entry");
+ assert!(
+ probe < spawn,
+ "probe must precede shared observer spawn entry"
+ );
+ }
}
--
Gitblit v1.9.3