media: drain mic input during playout waits

AGENTS.md

@@ -227,3 +227,4 @@ Context: 0.16.x proved that queue tweaks and static calibration cannot guarantee
 - 2026-05-01: 0.17.1 mirrored probe failed with video about `1.18-1.31s` behind audio and 761 planner video freezes. Root cause candidate: the client rebaser forced independent camera/mic pipelines onto one first-packet capture base, so a later-starting camera path was timestamped too early and looked permanently behind audio. Patch 0.17.2 anchors each stream to the shared monotonic clock at its own first packet time.
 - 2026-05-02: 0.17.2 mirrored probe and Google Meet test showed major improvement but persistent sub-second late video. Root cause follow-up: the temporary `+350ms` factory MJPEG video playout offset matched the observed browser skew and also made the server skew guard freeze video against its own offset. Patch 0.17.3 restores factory video offset to `0ms`, migrates untouched `+350ms` install/calibration defaults back to `0ms`, and makes the skew guard offset-aware for intentional site calibration.
 - 2026-05-02: 0.17.3 Google Meet manual test improved to roughly sub-second/near-quarter-second lip sync, but the mirrored analyzer could not pair pulses and the user still heard choppy background audio. Client logs showed Pulse microphone packets arriving unevenly with ages around `90-240ms`; patch 0.17.4 lowers Pulse mic `buffer-time`/`latency-time`, bounds the mic queue/appsink, and keeps mirrored-probe after-run planner diagnostics even when analysis fails.
+- 2026-05-02: 0.17.4 mirrored run was salvageable after an SCP banner timeout, but analysis still failed with no close pulse pairs. The client log still showed `180-240ms` microphone delivery ages, pointing at server playout sleeps backpressuring the gRPC microphone stream. Patch 0.17.5 drains inbound microphone packets while waiting for scheduled UAC playout and retries browser-capture SCP fetches.

Cargo.lock

@@ -1652,7 +1652,7 @@ checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2"
 
 [[package]]
 name = "lesavka_client"
-version = "0.17.4"
+version = "0.17.5"
 dependencies = [
  "anyhow",
  "async-stream",
@@ -1686,7 +1686,7 @@ dependencies = [
 
 [[package]]
 name = "lesavka_common"
-version = "0.17.4"
+version = "0.17.5"
 dependencies = [
  "anyhow",
  "base64",
@@ -1698,7 +1698,7 @@ dependencies = [
 
 [[package]]
 name = "lesavka_server"
-version = "0.17.4"
+version = "0.17.5"
 dependencies = [
  "anyhow",
  "base64",

client/Cargo.toml

@@ -4,7 +4,7 @@ path                    = "src/main.rs"
 
 [package]
 name                    = "lesavka_client"
-version                 = "0.17.4"
+version                 = "0.17.5"
 edition                 = "2024"
 
 [dependencies]

common/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name                    = "lesavka_common"
-version                 = "0.17.4"
+version                 = "0.17.5"
 edition                 = "2024"
 build                   = "build.rs"
 

scripts/manual/run_upstream_browser_av_sync.sh

@@ -171,7 +171,19 @@ while true; do
 done
 
 echo "==> fetching capture back to ${LOCAL_CAPTURE}"
-scp ${SSH_OPTS} "${TETHYS_HOST}:${REMOTE_CAPTURE}" "${LOCAL_CAPTURE}"
+fetch_status=1
+for attempt in 1 2 3 4 5; do
+  if scp ${SSH_OPTS} "${TETHYS_HOST}:${REMOTE_CAPTURE}" "${LOCAL_CAPTURE}"; then
+    fetch_status=0
+    break
+  fi
+  echo "capture fetch attempt ${attempt} failed; retrying" >&2
+  sleep 3
+done
+if ((fetch_status != 0)); then
+  echo "failed to fetch browser capture from ${TETHYS_HOST}:${REMOTE_CAPTURE}" >&2
+  exit "${fetch_status}"
+fi
 
 echo "==> analyzing browser capture"
 analyze_args=(--report-dir "${LOCAL_REPORT_DIR}")

server/Cargo.toml

@@ -10,7 +10,7 @@ bench                   = false
 
 [package]
 name                    = "lesavka_server"
-version                 = "0.17.4"
+version                 = "0.17.5"
 edition                 = "2024"
 autobins                = false
 

server/src/main/relay_service.rs

@@ -151,7 +151,7 @@ impl Relay for Handler {
             let stale_drop_budget = upstream_stale_drop_budget();
             static CNT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
 
-            loop {
+            'microphone_loop: loop {
                 if !upstream_media_rt.is_microphone_active(lease.generation) {
                     info!(rpc_id, session_id = lease.session_id, "🎤 stream_microphone session superseded");
                     cleanup.mark_superseded();
@@ -245,7 +245,39 @@ impl Relay for Handler {
                     );
                     continue;
                 }
-                tokio::time::sleep_until(plan.due_at).await;
+                while tokio::time::Instant::now() < plan.due_at {
+                    let sleep_until_due = tokio::time::sleep_until(plan.due_at);
+                    tokio::pin!(sleep_until_due);
+                    tokio::select! {
+                        _ = &mut sleep_until_due => break,
+                        next_packet = inbound.next(), if !inbound_closed => {
+                            match next_packet.transpose() {
+                                Ok(Some(next_pkt)) => {
+                                    pending.push_back(next_pkt);
+                                    let coalesced = retain_freshest_audio_packet(&mut pending);
+                                    if coalesced > 0 {
+                                        tracing::debug!(
+                                            rpc_id,
+                                            session_id = lease.session_id,
+                                            dropped = coalesced,
+                                            "🎤 coalesced stale upstream audio while waiting for scheduled playout"
+                                        );
+                                    }
+                                }
+                                Ok(None) => inbound_closed = true,
+                                Err(err) => {
+                                    cleanup.mark_aborted();
+                                    warn!(
+                                        rpc_id,
+                                        session_id = lease.session_id,
+                                        "🎤 stream_microphone inbound error before clean EOF: {err}"
+                                    );
+                                    break 'microphone_loop;
+                                }
+                            }
+                        }
+                    }
+                }
                 let actual_late_by = tokio::time::Instant::now()
                     .checked_duration_since(plan.due_at)
                     .unwrap_or_default();

testing/tests/client_manual_sync_script_contract.rs

@@ -78,6 +78,9 @@ fn browser_sync_script_can_delegate_to_a_real_path_driver() {
         "bash -lc \"${BROWSER_SYNC_DRIVER_COMMAND}\"",
         "--event-width-codes",
         "--report-dir \"${LOCAL_REPORT_DIR}\"",
+        "for attempt in 1 2 3 4 5",
+        "capture fetch attempt ${attempt} failed; retrying",
+        "failed to fetch browser capture from ${TETHYS_HOST}:${REMOTE_CAPTURE}",
     ] {
         assert!(
             BROWSER_SYNC_SCRIPT.contains(expected),