From 9485ced68a81f8880f3b173e62b62b55f0dd80aa Mon Sep 17 00:00:00 2001
From: cai <cai@nbcai.cc>
Date: Wed, 12 Aug 2026 10:29:38 +0800
Subject: [PATCH] fix(helper): observe fixture attributes through probe ttl
---
src/main.rs | 124 ++++++++++++++++++++++++++++++++++------
1 files changed, 104 insertions(+), 20 deletions(-)
diff --git a/src/main.rs b/src/main.rs
index 056ed15..cd45c4d 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -63,7 +63,6 @@
const CONTROLLED_FIXTURE_ACK_TOPIC: &str = "controlled_fixture_attribute_ack";
const CONTROLLED_FIXTURE_PROTOCOL_VERSION: u64 = 1;
const CONTROLLED_FIXTURE_GENERATION: u64 = 1;
-const CONTROLLED_FIXTURE_PROBE_RECHECKS: usize = 3;
const CONTROLLED_FIXTURE_PROBE_RECHECK_DELAY: Duration = Duration::from_millis(25);
const CONTROLLED_FIXTURE_PROBE_TTL: Duration = Duration::from_millis(250);
@@ -171,6 +170,40 @@
return Err("wrong_sequence");
}
Ok(())
+}
+
+async fn observe_controlled_fixture_attributes<F>(
+ expires_at: Instant,
+ actual_participant: &str,
+ expected_participant: Option<&str>,
+ requested_sequence: &str,
+ mut read_attributes: F,
+) -> Result<(), &'static str>
+where
+ F: FnMut() -> std::collections::HashMap<String, String>,
+{
+ loop {
+ if Instant::now() > expires_at {
+ return Err("timeout");
+ }
+ let decision = classify_controlled_fixture_attributes(
+ actual_participant,
+ expected_participant,
+ &read_attributes(),
+ requested_sequence,
+ );
+ if decision.is_ok() || !matches!(decision, Err("missing_attributes")) {
+ return decision;
+ }
+ let Some(next_check) = Instant::now().checked_add(CONTROLLED_FIXTURE_PROBE_RECHECK_DELAY)
+ else {
+ return Err("timeout");
+ };
+ if next_check > expires_at {
+ return Err("timeout");
+ }
+ sleep(CONTROLLED_FIXTURE_PROBE_RECHECK_DELAY).await;
+ }
}
fn controlled_fixture_observer_gate(pending_probe: bool, probe_result: Option<bool>) -> bool {
@@ -1070,25 +1103,14 @@
if participant.identity() != probe.sender {
return Some(false);
}
- let mut decision = Err("timeout");
- for attempt in 0..CONTROLLED_FIXTURE_PROBE_RECHECKS {
- if Instant::now() > probe.expires_at {
- break;
- }
- let attributes = participant.attributes();
- decision = classify_controlled_fixture_attributes(
- &participant.identity().to_string(),
- expected_participant,
- &attributes,
- &probe.sequence,
- );
- if decision.is_ok() || !matches!(decision, Err("missing_attributes")) {
- break;
- }
- if attempt + 1 < CONTROLLED_FIXTURE_PROBE_RECHECKS {
- sleep(CONTROLLED_FIXTURE_PROBE_RECHECK_DELAY).await;
- }
- }
+ let decision = observe_controlled_fixture_attributes(
+ probe.expires_at,
+ &participant.identity().to_string(),
+ expected_participant,
+ &probe.sequence,
+ || participant.attributes(),
+ )
+ .await;
let (result, input_source_category, reject_reason) = match decision {
Ok(()) => ("observed", Some("controlled_fixture"), None),
Err(reason) => ("rejected", None, Some(reason)),
@@ -5041,6 +5063,68 @@
);
}
+ #[tokio::test]
+ async fn production_attribute_observation_accepts_server_visibility_within_probe_ttl() {
+ let expected = HashMap::from([
+ (
+ "inputSourceCategory".to_string(),
+ "controlled_fixture".to_string(),
+ ),
+ (
+ "clientFixtureSequence".to_string(),
+ "fixture-01".to_string(),
+ ),
+ ]);
+ let mut reads = 0;
+ let result = observe_controlled_fixture_attributes(
+ Instant::now() + CONTROLLED_FIXTURE_PROBE_TTL,
+ "user-1",
+ Some("user-1"),
+ "fixture-01",
+ || {
+ reads += 1;
+ if reads <= 3 {
+ HashMap::new()
+ } else {
+ expected.clone()
+ }
+ },
+ )
+ .await;
+ assert_eq!(result, Ok(()));
+ assert_eq!(reads, 4);
+ }
+
+ #[tokio::test]
+ async fn production_attribute_observation_rejects_wrong_sequence_without_audio_effect() {
+ let attributes = HashMap::from([
+ (
+ "inputSourceCategory".to_string(),
+ "controlled_fixture".to_string(),
+ ),
+ (
+ "clientFixtureSequence".to_string(),
+ "fixture-02".to_string(),
+ ),
+ ]);
+ let result = observe_controlled_fixture_attributes(
+ Instant::now() + CONTROLLED_FIXTURE_PROBE_TTL,
+ "user-1",
+ Some("user-1"),
+ "fixture-01",
+ || attributes.clone(),
+ )
+ .await;
+ let mut observer_starts = 0;
+ assert_eq!(result, Err("wrong_sequence"));
+ assert!(!start_observer_after_controlled_fixture_probe(
+ true,
+ Some(result.is_ok()),
+ || observer_starts += 1,
+ ));
+ assert_eq!(observer_starts, 0);
+ }
+
#[test]
fn controlled_fixture_ack_payload_is_reliable_and_redacted() {
let ack = ControlledFixtureAttributeAck {
--
Gitblit v1.9.3