From da9563428e990244ff2145eb6a9cb453f9ef9d22 Mon Sep 17 00:00:00 2001
From: cai <cai@nbcai.cc>
Date: Wed, 12 Aug 2026 12:02:47 +0800
Subject: [PATCH] Fail closed when fixture ACK publish fails
---
src/main.rs | 123 +++++++++++++++++++++++++++++++++++------
1 files changed, 105 insertions(+), 18 deletions(-)
diff --git a/src/main.rs b/src/main.rs
index 6a0ca0e..d40ca9e 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -6,6 +6,7 @@
borrow::Cow,
collections::HashSet,
env, fs,
+ future::Future,
path::{Path, PathBuf},
sync::{
Arc,
@@ -209,6 +210,45 @@
classification.1,
);
classification
+}
+
+async fn complete_controlled_fixture_ack_publish<F, E>(
+ publish: F,
+ observed: bool,
+ ack_result: &'static str,
+ reject_reason: Option<&'static str>,
+ call_id: &str,
+ trace_id: &str,
+ sequence: &str,
+ acknowledged_probe_sequences: &mut HashSet<String>,
+) -> bool
+where
+ F: Future<Output = Result<(), E>>,
+{
+ if publish.await.is_ok() {
+ record_controlled_fixture_probe_event(
+ "ack_publish_completed",
+ call_id,
+ trace_id,
+ sequence,
+ observed,
+ Some(ack_result),
+ reject_reason,
+ );
+ acknowledged_probe_sequences.insert(sequence.to_string());
+ observed
+ } else {
+ record_controlled_fixture_probe_event(
+ "ack_publish_completed",
+ call_id,
+ trace_id,
+ sequence,
+ false,
+ Some("publish_failed"),
+ Some("ack_publish_failed"),
+ );
+ false
+ }
}
fn controlled_fixture_probe(
@@ -1325,29 +1365,19 @@
reliable: true,
destination_identities: vec![probe.sender],
});
- if publish.await.is_ok() {
- record_controlled_fixture_probe_event(
- "ack_publish_completed",
- call_id,
- trace_id,
- &probe.sequence,
+ Some(
+ complete_controlled_fixture_ack_publish(
+ publish,
observed,
- Some(ack_result),
+ ack_result,
reject_reason,
- );
- acknowledged_probe_sequences.insert(probe.sequence);
- } else {
- record_controlled_fixture_probe_event(
- "ack_publish_completed",
call_id,
trace_id,
&probe.sequence,
- false,
- Some("publish_failed"),
- Some("ack_publish_failed"),
- );
- }
- Some(result == "observed")
+ acknowledged_probe_sequences,
+ )
+ .await,
+ )
}
async fn handle_finished_turn(
@@ -5367,6 +5397,63 @@
}
#[tokio::test]
+ async fn production_ack_publish_failure_keeps_observer_session_and_audio_closed() {
+ let mut acknowledged = HashSet::new();
+ let probe_result = complete_controlled_fixture_ack_publish(
+ async { Err::<(), ()>(()) },
+ true,
+ "observed",
+ None,
+ "call-publish-failure",
+ "trace-publish-failure",
+ "fixture-01",
+ &mut acknowledged,
+ )
+ .await;
+ let mut observer_starts = 0;
+ let mut session_starts = 0;
+ let mut audio_starts = 0;
+ let mut speaking_starts = 0;
+ assert!(!start_observer_after_controlled_fixture_probe(
+ true,
+ Some(probe_result),
+ || {
+ observer_starts += 1;
+ session_starts += 1;
+ audio_starts += 1;
+ speaking_starts += 1;
+ },
+ ));
+ assert!(!probe_result);
+ assert!(acknowledged.is_empty());
+ assert_eq!(observer_starts, 0);
+ assert_eq!(session_starts, 0);
+ assert_eq!(audio_starts, 0);
+ assert_eq!(speaking_starts, 0);
+
+ let observed_result = complete_controlled_fixture_ack_publish(
+ async { Ok::<(), ()>(()) },
+ true,
+ "observed",
+ None,
+ "call-publish-success",
+ "trace-publish-success",
+ "fixture-02",
+ &mut acknowledged,
+ )
+ .await;
+ let mut successful_observer_starts = 0;
+ assert!(start_observer_after_controlled_fixture_probe(
+ true,
+ Some(observed_result),
+ || successful_observer_starts += 1,
+ ));
+ assert!(observed_result);
+ assert!(acknowledged.contains("fixture-02"));
+ assert_eq!(successful_observer_starts, 1);
+ }
+
+ #[tokio::test]
async fn production_attribute_observation_accepts_server_visibility_within_probe_ttl() {
let expected = HashMap::from([
(
--
Gitblit v1.9.3