From 949be8e5f1a75286b8b517dacda76ed27646e77b Mon Sep 17 00:00:00 2001
From: cai <cai@nbcai.cc>
Date: Tue, 11 Aug 2026 21:23:54 +0800
Subject: [PATCH] fix(helper): gate audio observer on fixture ack
---
src/main.rs | 54 ++++++++++++++++++++++++++++++++++++++----------------
1 files changed, 38 insertions(+), 16 deletions(-)
diff --git a/src/main.rs b/src/main.rs
index 8a1e10d..94b6589 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -173,6 +173,10 @@
Ok(())
}
+fn controlled_fixture_observer_gate(pending_probe: bool, probe_result: Option<bool>) -> bool {
+ !pending_probe || probe_result == Some(true)
+}
+
#[tokio::main(flavor = "multi_thread")]
async fn main() -> Result<()> {
init_tracing();
@@ -914,7 +918,26 @@
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();
+ let probe_result = process_controlled_fixture_probe(
+ &mut pending_probe,
+ &mut acknowledged_probe_sequences,
+ Some(&participant_for_probe),
+ &sink,
+ &call_id,
+ &trace_id,
+ user_participant_identity.as_deref(),
+ )
+ .await;
+ if !controlled_fixture_observer_gate(pending_sequence.is_some(), probe_result) {
+ warn!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ "runtime helper withheld audio observer until controlled fixture ACK"
+ );
+ continue;
+ }
spawn_user_audio_frame_observer(
track,
call_id.clone(),
@@ -931,16 +954,6 @@
participant,
);
current_user_participant = Some(participant_for_probe);
- process_controlled_fixture_probe(
- &mut pending_probe,
- &mut acknowledged_probe_sequences,
- current_user_participant.as_ref(),
- &sink,
- &call_id,
- &trace_id,
- user_participant_identity.as_deref(),
- )
- .await;
}
RoomEvent::DataReceived {
payload,
@@ -1022,19 +1035,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 +1085,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 +1097,7 @@
if publish.await.is_ok() {
acknowledged_probe_sequences.insert(probe.sequence);
}
+ Some(result == "observed")
}
async fn handle_finished_turn(
@@ -4998,4 +5012,12 @@
"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));
+ }
}
--
Gitblit v1.9.3