Stabilize remote bridge privacy stress

Use the deterministic Python runner during concurrent privacy checks, allow realistic scheduling margins, assert thread termination, and drain child output asynchronously to avoid redirected-pipe stalls.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Signed-off-by: Colin Neilens <coneilen@microsoft.com>

Author: coneilen

Tools/windows/Tests/RemoteBridgePrivacyRace.Tests.ps1

@@ -19,21 +19,17 @@ function Start-CapturedProcess([string] $fileName, [string[]] $arguments) {
   $process = [Diagnostics.Process]::new()
   $process.StartInfo = $startInfo
   [void] $process.Start()
-  $process.BeginOutputReadLine()
-  $process.BeginErrorReadLine()
-  return $process
+  return [pscustomobject]@{
+    Process = $process
+    Stdout = $process.StandardOutput.ReadToEndAsync()
+    Stderr = $process.StandardError.ReadToEndAsync()
+  }
 }
 
 $testPath = Join-Path $repoRoot "investigation\spikes\remote-bridge"
 $remoteArguments = @(
   "-B",
-  "-m",
-  "unittest",
-  "discover",
-  "-s",
-  $testPath,
-  "-p",
-  "test_*.py"
+  (Join-Path $testPath "run_tests.py")
 )
 $privacyArguments = @(
   "-NoProfile",
@@ -55,28 +51,32 @@ $privacyProcesses = @(
 )
 
 try {
-  foreach ($process in $remoteProcesses + $privacyProcesses) {
-    $process.WaitForExit()
+  foreach ($entry in $remoteProcesses + $privacyProcesses) {
+    $entry.Process.WaitForExit()
   }
 
-  $remoteFailures = $remoteProcesses | Where-Object ExitCode -ne 0
+  $remoteFailures = $remoteProcesses | Where-Object { $_.Process.ExitCode -ne 0 }
   if ($remoteFailures) {
     throw "Remote bridge test process failed during privacy race: $(
-      @($remoteFailures | ForEach-Object { "pid=$($_.Id), exit=$($_.ExitCode)" }) -join "; "
+      @($remoteFailures | ForEach-Object {
+        "pid=$($_.Process.Id), exit=$($_.Process.ExitCode), stdout=$($_.Stdout.Result), stderr=$($_.Stderr.Result)"
+      }) -join "; "
     )"
   }
-  $privacyFailures = $privacyProcesses | Where-Object ExitCode -ne 0
+  $privacyFailures = $privacyProcesses | Where-Object { $_.Process.ExitCode -ne 0 }
   if ($privacyFailures) {
     throw "Privacy validation failed while remote tests ran concurrently: $(
-      @($privacyFailures | ForEach-Object { "pid=$($_.Id), exit=$($_.ExitCode)" }) -join "; "
+      @($privacyFailures | ForEach-Object {
+        "pid=$($_.Process.Id), exit=$($_.Process.ExitCode), stdout=$($_.Stdout.Result), stderr=$($_.Stderr.Result)"
+      }) -join "; "
     )"
   }
 } finally {
-  foreach ($process in $remoteProcesses + $privacyProcesses) {
-    if (-not $process.HasExited) {
-      $process.Kill()
+  foreach ($entry in $remoteProcesses + $privacyProcesses) {
+    if (-not $entry.Process.HasExited) {
+      $entry.Process.Kill()
     }
-    $process.Dispose()
+    $entry.Process.Dispose()
   }
 }
 

investigation/spikes/remote-bridge/test_remote_bridge.py

@@ -13,6 +13,7 @@
 
 
 SPIKE_ROOT = Path(__file__).resolve().parent
+THREAD_TIMEOUT = 5.0
 sys.path.insert(0, str(SPIKE_ROOT))
 
 from remote_bridge import (  # noqa: E402
@@ -369,14 +370,14 @@ def interleaving_read():
         store.read = interleaving_read
 
         def replace_state():
-            read_started.wait(1)
+            read_started.wait(THREAD_TIMEOUT)
             store.write(replacement)
             replacement_done.set()
 
         replacement_thread = threading.Thread(target=replace_state, daemon=True)
         replacement_thread.start()
         self.bridge.stop()
-        replacement_thread.join(1)
+        replacement_thread.join(THREAD_TIMEOUT)
 
         self.assertFalse(replacement_thread.is_alive())
         self.assertTrue(replacement_done.is_set())
@@ -500,7 +501,7 @@ def test_rotation_overlap_starts_after_state_lock_release(self):
         def hold_state_lock():
             with self.bridge.state_store.transaction():
                 lock_ready.set()
-                release_lock.wait(1)
+                release_lock.wait(THREAD_TIMEOUT)
 
         def rotate():
             try:
@@ -512,15 +513,15 @@ def rotate():
 
         holder = threading.Thread(target=hold_state_lock, daemon=True)
         holder.start()
-        lock_ready.wait(1)
+        lock_ready.wait(THREAD_TIMEOUT)
         rotation = threading.Thread(target=rotate, daemon=True)
         rotation.start()
         time.sleep(0.2)
         self.assertFalse(rotation_done.is_set())
         release_at = time.time()
         release_lock.set()
-        holder.join(1)
-        rotation.join(1)
+        holder.join(THREAD_TIMEOUT)
+        rotation.join(THREAD_TIMEOUT)
 
         self.assertFalse(holder.is_alive())
         self.assertFalse(rotation.is_alive())
@@ -576,7 +577,7 @@ def read_state():
                         if wait_for_readers:
                             missing_reads += 1
                     if wait_for_readers:
-                        read_barrier.wait(1)
+                        read_barrier.wait(THREAD_TIMEOUT)
                     raise
 
             return read_state
@@ -585,7 +586,7 @@ def read_state():
             bridge.state_store.read = synchronized_missing(read)
 
         def start_bridge(bridge):
-            start_barrier.wait(1)
+            start_barrier.wait(THREAD_TIMEOUT)
             try:
                 bridge.start()
                 outcomes.append((bridge, "started"))
@@ -598,9 +599,9 @@ def start_bridge(bridge):
         ]
         for thread in threads:
             thread.start()
-        start_barrier.wait(1)
+        start_barrier.wait(THREAD_TIMEOUT)
         for thread in threads:
-            thread.join(1)
+            thread.join(THREAD_TIMEOUT)
 
         self.assertTrue(all(not thread.is_alive() for thread in threads))
         started = [bridge for bridge, result in outcomes if result == "started"]
@@ -629,7 +630,7 @@ def test_stop_waits_for_start_publication(self):
 
         def delayed_publish(expected, state):
             publication_started.set()
-            release_publication.wait(1)
+            release_publication.wait(THREAD_TIMEOUT)
             return original_publish(expected, state)
 
         store.write_if_matches = delayed_publish
@@ -651,13 +652,13 @@ def stop_bridge():
         starter = threading.Thread(target=start_bridge, daemon=True)
         stopper = threading.Thread(target=stop_bridge, daemon=True)
         starter.start()
-        publication_started.wait(1)
+        publication_started.wait(THREAD_TIMEOUT)
         stopper.start()
         time.sleep(0.1)
         self.assertFalse(stop_done.is_set())
         release_publication.set()
-        starter.join(1)
-        stopper.join(1)
+        starter.join(THREAD_TIMEOUT)
+        stopper.join(THREAD_TIMEOUT)
 
         self.assertFalse(starter.is_alive())
         self.assertFalse(stopper.is_alive())
@@ -722,7 +723,7 @@ def read_states():
         finally:
             stop_readers.set()
             for reader in readers:
-                reader.join(1)
+                reader.join(THREAD_TIMEOUT)
 
         self.assertTrue(all(not reader.is_alive() for reader in readers))
         self.assertEqual(errors, [])