diff --git a/crates/punktfunk-host/src/audio/windows/wasapi_cap.rs b/crates/punktfunk-host/src/audio/windows/wasapi_cap.rs index 17469cbc..443a37df 100644 --- a/crates/punktfunk-host/src/audio/windows/wasapi_cap.rs +++ b/crates/punktfunk-host/src/audio/windows/wasapi_cap.rs @@ -48,7 +48,8 @@ impl WasapiLoopbackCapturer { let (tx, rx) = sync_channel::>(64); let stop = Arc::new(AtomicBool::new(false)); // Bring-up handshake: report open success/failure before returning, so a missing render - // endpoint surfaces as Err (caller continues without audio) rather than a silent dead thread. + // endpoint surfaces as Err (the native plane then keeps retrying the open with backoff) + // rather than a silent dead thread. let (ready_tx, ready_rx) = sync_channel::>(1); let stop_t = stop.clone(); let join = thread::Builder::new() @@ -134,6 +135,14 @@ enum Next { const REOPEN_BACKOFF: Duration = Duration::from_secs(2); /// Watchdog cadence for "did the default render device change under us?" checks. const DEFAULT_CHECK_EVERY: Duration = Duration::from_secs(1); +/// Total attempts for the FIRST open before its failure surfaces through the `ready` handshake. +/// Session start is peak endpoint churn — the virtual-display attach and this module's own +/// IPolicyConfig default flips race the activate, which then fails transiently (0x80070002, +/// endpoint mid-re-registration) — so a couple of quick retries absorb it within the +/// handshake budget. +const FIRST_OPEN_ATTEMPTS: u32 = 3; +/// Pause between first-open attempts (endpoint churn settles in well under a second). +const FIRST_OPEN_RETRY_PAUSE: Duration = Duration::from_secs(1); fn capture_thread( tx: SyncSender>, @@ -151,11 +160,14 @@ fn capture_thread( } // Self-heal for the capturer's whole life: each `capture_once` is one endpoint open + inner // capture loop; it returns to reopen (default-device change) or errors (device invalidated, - // engine restart), and only the FIRST open's failure is fatal — it surfaces as `open()`'s Err - // so the session honestly runs without audio (the native plane retries with its own backoff). + // engine restart). The FIRST open gets [`FIRST_OPEN_ATTEMPTS`] tries (session-start endpoint + // churn — see the constant) before its failure surfaces as `open()`'s Err; the caller keeps + // retrying the whole open with its own backoff after that, so a bad start delays audio + // rather than ending it. let mut ready = Some(ready); let mut mode = TargetMode::Assert; let mut failures: u64 = 0; + let mut first_attempts: u32 = 0; while !stop.load(Ordering::Relaxed) { match capture_once(&tx, &stop, &mut ready, channels, mode) { Ok(Next::Stopped) => break, @@ -163,11 +175,21 @@ fn capture_thread( mode = m; failures = 0; } - Err(e) => { - if let Some(r) = ready.take() { - let _ = r.send(Err(anyhow!("{e:#}"))); + Err(e) if ready.is_some() => { + first_attempts += 1; + if first_attempts >= FIRST_OPEN_ATTEMPTS || stop.load(Ordering::Relaxed) { + let _ = ready.take().unwrap().send(Err(anyhow!("{e:#}"))); break; } + tracing::info!(error = %format!("{e:#}"), attempt = first_attempts, + "audio loopback first open failed — retrying"); + // Stop-responsive pause (same discipline as the reopen backoff below). + let until = Instant::now() + FIRST_OPEN_RETRY_PAUSE; + while Instant::now() < until && !stop.load(Ordering::Relaxed) { + thread::sleep(Duration::from_millis(100)); + } + } + Err(e) => { failures += 1; if failures.is_power_of_two() { tracing::warn!(error = %format!("{e:#}"), count = failures, @@ -197,7 +219,8 @@ fn default_render(en: &DeviceEnumerator) -> Option<(Device, String)> { } /// One endpoint open + capture loop. Returns how to continue ([`Next`]) or an error (first open: -/// fatal via the `ready` handshake; later: reopen with backoff). +/// retried [`FIRST_OPEN_ATTEMPTS`] times, then fatal via the `ready` handshake; later: reopen +/// with backoff). fn capture_once( tx: &SyncSender>, stop: &AtomicBool, diff --git a/crates/punktfunk-host/src/native/audio.rs b/crates/punktfunk-host/src/native/audio.rs index bbb1909e..493015a4 100644 --- a/crates/punktfunk-host/src/native/audio.rs +++ b/crates/punktfunk-host/src/native/audio.rs @@ -71,18 +71,23 @@ pub(super) fn audio_thread( // Reuse the cached capturer ONLY when its channel count matches this session's; a stereo // capturer left by a prior session must not feed a 5.1/7.1 session (the encoder + the client's // decoder are sized for `want`, so a mismatched capturer would garble/desync the audio). - let mut capturer = match audio_cap.lock().unwrap().take() { + // A FAILED first open does not end the session's audio: session start is peak endpoint churn + // on Windows (the virtual-display attach and the wiring plan's own default-device flips race + // the WASAPI activate — 0x80070002 mid-re-registration), so it enters the same + // reopen-with-backoff loop a mid-session capture death does; audio then starts a few seconds + // late instead of never. + let capturer = match audio_cap.lock().unwrap().take() { Some(mut c) if c.channels() == want as u32 => { c.drain(); // discard audio captured between sessions (also re-claims routing) - c + Some(c) } prev => { drop(prev); // wrong channel count (or none): clean teardown, open fresh at `want` match crate::audio::open_audio_capture(want as u32) { - Ok(c) => c, + Ok(c) => Some(c), Err(e) => { - tracing::warn!(error = %format!("{e:#}"), "punktfunk/1 audio unavailable — session continues without it"); - return; + tracing::warn!(error = %format!("{e:#}"), "punktfunk/1 audio failed to open — retrying in the background until it comes up"); + None } } } @@ -91,8 +96,10 @@ pub(super) fn audio_thread( Ok(e) => e, Err(e) => { tracing::warn!(error = %e, "opus encoder init failed — session continues without audio"); - capturer.idle(); // parked, not streaming — release the routing claim - crate::audio::park_audio_capture(&audio_cap, capturer); + if let Some(mut c) = capturer { + c.idle(); // parked, not streaming — release the routing claim + crate::audio::park_audio_capture(&audio_cap, c); + } return; } }; @@ -103,20 +110,22 @@ pub(super) fn audio_thread( let mut opus_buf = vec![0u8; 4096]; let mut seq: u32 = 0; // Reopen-with-backoff: hold the capturer in an Option so a mid-session capture-thread death - // (device unplug, daemon restart) reopens instead of muting the rest of a multi-hour session. - // A quiet sink is NOT a death — `next_chunk` returns an empty chunk on its idle timeout — so only - // a genuine thread-ended Err drops the capturer. Reopens are throttled by INJECTOR_REOPEN_BACKOFF. - // The Opus encoder and the monotonic `seq` are kept across reopens (the client sees a gap, not a - // restart). The first open already happened above; failing THAT still ends the session quietly. - let mut capturer = Some(capturer); - let mut last_failed: Option = None; + // (device unplug, daemon restart) — or a first open lost to session-start churn above — + // reopens instead of muting the rest of a multi-hour session. A quiet sink is NOT a death — + // `next_chunk` returns an empty chunk on its idle timeout — so only a genuine thread-ended + // Err drops the capturer. Reopens are throttled by INJECTOR_REOPEN_BACKOFF. The Opus encoder + // and the monotonic `seq` are kept across reopens (the client sees a gap, not a restart). + let mut last_failed = capturer.is_none().then(std::time::Instant::now); + let mut capturer = capturer; // A stuck Opus encoder would fail on every 5 ms frame (~200/s); power-of-two throttle the // warn so it can't flood stderr + the log ring while still surfacing that it's failing. let mut opus_encode_errs: u64 = 0; - tracing::info!( - channels = want, - "punktfunk/1 audio streaming (Opus 48 kHz, 5 ms datagrams)" - ); + if capturer.is_some() { + tracing::info!( + channels = want, + "punktfunk/1 audio streaming (Opus 48 kHz, 5 ms datagrams)" + ); + } 'session: while !stop.load(Ordering::SeqCst) { if capturer.is_none() { if last_failed.is_some_and(|t| t.elapsed() < INJECTOR_REOPEN_BACKOFF) {