Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
118 changes: 117 additions & 1 deletion apps/desktop/src-tauri/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@ use std::error::Error;
use std::fmt;
use std::net::{Ipv4Addr, SocketAddr, TcpListener as StdTcpListener};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

Expand Down Expand Up @@ -54,6 +54,7 @@ const CAPTURE_RUNTIME_ERROR_EVENT: &str = "capture-runtime-error";
const MAX_PENDING_CAPTURE_OUTCOMES: usize = 256;
const CAPTURES_PER_RETENTION_RUN: usize = 100;
const SIDECAR_STARTUP_TIMEOUT: Duration = Duration::from_secs(10);
const SIDECAR_MONITOR_INTERVAL: Duration = Duration::from_millis(250);
const PROXY_RECOVERY_JOURNAL_FILENAME: &str = "proxy-recovery.v0.1.json";

type SharedStore = Arc<Mutex<Option<EncryptedStore>>>;
Expand All @@ -64,6 +65,7 @@ struct DesktopState {
proxy: AsyncMutex<Option<ProxyRuntime>>,
mode: AsyncMutex<CaptureMode>,
capture_active: Arc<AtomicBool>,
next_proxy_generation: AtomicU64,
proxy_recovery_checked: AtomicBool,
sidecar_bundle: Option<Arc<SidecarBundle>>,
sidecar_error: Mutex<Option<String>>,
Expand All @@ -84,6 +86,7 @@ impl Drop for GatewayRuntime {
}

struct ProxyRuntime {
generation: u64,
system_proxy: Option<PlatformProxySession>,
process: Option<SidecarProcess>,
endpoint: String,
Expand Down Expand Up @@ -277,6 +280,7 @@ pub fn run() {
proxy: AsyncMutex::new(None),
mode: AsyncMutex::new(CaptureMode::Gateway),
capture_active: Arc::new(AtomicBool::new(true)),
next_proxy_generation: AtomicU64::new(1),
proxy_recovery_checked: AtomicBool::new(false),
sidecar_bundle,
sidecar_error: Mutex::new(sidecar_error),
Expand Down Expand Up @@ -664,7 +668,9 @@ async fn ensure_proxy(app: &AppHandle, state: &DesktopState) -> Result<(), Strin
.map_err(|error| format!("explicit proxy launch task failed: {error}"))?
.map_err(|error| error.to_string())?;
let endpoint = format!("http://{}", process.endpoint());
let generation = state.next_proxy_generation.fetch_add(1, Ordering::Relaxed);
let mut next_runtime = ProxyRuntime {
generation,
system_proxy: None,
process: Some(process),
endpoint: endpoint.clone(),
Expand All @@ -683,6 +689,8 @@ async fn ensure_proxy(app: &AppHandle, state: &DesktopState) -> Result<(), Strin
));
next_runtime.shutdown = Some(shutdown);
*runtime = Some(next_runtime);
drop(runtime);
tauri::async_runtime::spawn(monitor_proxy_process(app.clone(), generation));
Ok(())
}

Expand All @@ -696,6 +704,75 @@ async fn stop_proxy(state: &DesktopState) -> Result<(), String> {
Ok(())
}

async fn monitor_proxy_process(app: AppHandle, generation: u64) {
let state = app.state::<DesktopState>();
loop {
tokio::time::sleep(SIDECAR_MONITOR_INTERVAL).await;
let status = {
let mut proxy = state.proxy.lock().await;
let Some(runtime) = proxy.as_mut() else {
return;
};
if runtime.generation != generation {
return;
}
let Some(process) = runtime.process.as_mut() else {
return;
};
process.try_wait()
};
let (code, mut detail) = match status {
Ok(None) => continue,
Ok(Some(status)) => (
"sidecar_process_exited",
format!("The explicit proxy process exited unexpectedly ({status})."),
),
Err(error) => (
"sidecar_process_monitor_failed",
format!("The explicit proxy process could not be monitored: {error}"),
),
};
let Some(runtime) = take_failed_proxy_runtime(&state, generation).await else {
return;
};
let restore = tauri::async_runtime::spawn_blocking(move || runtime.shutdown()).await;
match restore {
Ok(Ok(())) => {}
Ok(Err(error)) => {
detail.push_str(&format!(
" Original system proxy restoration failed: {error}"
));
}
Err(error) => {
detail.push_str(&format!(" Proxy shutdown task failed: {error}"));
}
}
emit_runtime_error(&app, code, detail);
return;
}
}

async fn take_failed_proxy_runtime(state: &DesktopState, generation: u64) -> Option<ProxyRuntime> {
let runtime = {
let mut proxy = state.proxy.lock().await;
if proxy.as_ref().map(|runtime| runtime.generation) != Some(generation) {
return None;
}
proxy.take()
};
let mut mode = state.mode.lock().await;
if *mode == CaptureMode::Proxy {
*mode = CaptureMode::Gateway;
let gateway = state.gateway.lock().await;
if let Some(gateway) = gateway.as_ref() {
gateway
.capture
.set_enabled(state.capture_active.load(Ordering::Acquire));
}
}
runtime
}

#[cfg(windows)]
async fn activate_system_proxy(app: &AppHandle, runtime: &mut ProxyRuntime) -> Result<(), String> {
if runtime.system_proxy.is_some() {
Expand Down Expand Up @@ -1391,6 +1468,45 @@ mod tests {
})
}

#[tokio::test]
async fn failed_proxy_runtime_returns_to_gateway_and_ignores_stale_monitors() {
let (capture, _receiver, _) = GatewayCapture::defaults();
capture.set_enabled(false);
let state = DesktopState {
store: Arc::new(Mutex::new(None)),
gateway: AsyncMutex::new(Some(GatewayRuntime {
capture: capture.clone(),
endpoint: "http://127.0.0.1:8787".to_owned(),
shutdown: None,
})),
proxy: AsyncMutex::new(Some(ProxyRuntime {
generation: 7,
system_proxy: None,
process: None,
endpoint: "http://127.0.0.1:43125".to_owned(),
shutdown: None,
})),
mode: AsyncMutex::new(CaptureMode::Proxy),
capture_active: Arc::new(AtomicBool::new(true)),
next_proxy_generation: AtomicU64::new(8),
proxy_recovery_checked: AtomicBool::new(true),
sidecar_bundle: None,
sidecar_error: Mutex::new(None),
};

assert!(take_failed_proxy_runtime(&state, 6).await.is_none());
assert_eq!(*state.mode.lock().await, CaptureMode::Proxy);
assert!(state.proxy.lock().await.is_some());

let runtime = take_failed_proxy_runtime(&state, 7)
.await
.expect("current failed runtime must be removed");
assert_eq!(*state.mode.lock().await, CaptureMode::Gateway);
assert!(state.proxy.lock().await.is_none());
assert!(capture.is_enabled());
runtime.shutdown().expect("empty runtime must shut down");
}

#[test]
fn disk_pressure_pauses_capture_without_stopping_the_gateway() {
let (capture, receiver, _) = GatewayCapture::defaults();
Expand Down
57 changes: 57 additions & 0 deletions apps/desktop/src/App.test.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -387,6 +387,63 @@ describe("request workbench", () => {
);
});

it("reloads the safe gateway state after the proxy sidecar exits", async () => {
window.__TAURI_INTERNALS__ = {};
const proxy = structuredClone(fixture) as unknown as WorkspaceBootstrap;
proxy.capture = {
...proxy.capture,
active: true,
canControl: true,
proxyAvailable: true,
mode: "proxy",
profile: "System-managed explicit TLS proxy",
endpoint: "http://127.0.0.1:43125",
};
const recovered = structuredClone(proxy);
recovered.capture.mode = "gateway";
recovered.capture.profile = "OpenAI-compatible local gateway";
recovered.capture.endpoint = "http://127.0.0.1:8787";
let notifyRuntimeError = () => {};
vi.mocked(listen).mockImplementation(async (event, handler) => {
if (event === "capture-runtime-error") {
notifyRuntimeError = () => handler({
event: "capture-runtime-error",
id: 3,
payload: {
code: "sidecar_process_exited",
detail: "The explicit proxy process exited unexpectedly (exit code: 1).",
},
});
}
return () => {};
});
let bootstrapCount = 0;
vi.mocked(invoke).mockImplementation(async (command) => {
if (command === "bootstrap_workspace") {
bootstrapCount += 1;
return structuredClone(bootstrapCount === 1 ? proxy : recovered);
}
throw new Error(`Unexpected command: ${command}`);
});

render(<App />);
expect(await screen.findByRole("button", { name: "Proxy" })).toHaveAttribute(
"aria-pressed",
"true",
);
await act(async () => notifyRuntimeError());

expect(await screen.findByRole("button", { name: "Gateway" })).toHaveAttribute(
"aria-pressed",
"true",
);
expect(bootstrapCount).toBe(2);
expect(screen.getByText("Issue").parentElement).toHaveAttribute(
"title",
"The explicit proxy process exited unexpectedly (exit code: 1).",
);
});

it("virtualizes one thousand requests and keeps filtered selection coherent", async () => {
const user = userEvent.setup();
window.__TAURI_INTERNALS__ = {};
Expand Down
6 changes: 6 additions & 0 deletions apps/desktop/src/App.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -118,6 +118,12 @@ export function App() {
capture: { ...current.capture, active: false },
});
}
if (
event.code === "sidecar_process_exited" ||
event.code === "sidecar_process_monitor_failed"
) {
setReloadToken((value) => value + 1);
}
},
})
.then((value) => {
Expand Down
40 changes: 39 additions & 1 deletion crates/sidecar-runtime/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ use std::fs::{self, File};
use std::io::{self, Read};
use std::net::{IpAddr, SocketAddr, TcpStream};
use std::path::{Component, Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::process::{Child, Command, ExitStatus, Stdio};
use std::thread;
use std::time::{Duration, Instant};

Expand Down Expand Up @@ -1323,6 +1323,13 @@ impl SidecarProcess {
self.endpoint
}

pub fn try_wait(&mut self) -> Result<Option<ExitStatus>, SidecarError> {
let Some(child) = self.child.as_mut() else {
return Ok(None);
};
child.try_wait().map_err(SidecarError::Io)
}

pub fn stop(&mut self) -> Result<(), SidecarError> {
let Some(mut child) = self.child.take() else {
return Ok(());
Expand Down Expand Up @@ -2310,6 +2317,37 @@ MAoGCCqGSM49BAMCA0gAMEUCIQC1PB8+NumezrQf5unFGhVeufUcyw/sjH6p1aqs
}
}

#[test]
fn sidecar_exit_status_can_be_polled_without_blocking() {
let child = if cfg!(windows) {
Command::new("cmd")
.args(["/C", "exit", "7"])
.spawn()
.expect("Windows fixture process must start")
} else {
Command::new("sh")
.args(["-c", "exit 7"])
.spawn()
.expect("Unix fixture process must start")
};
let mut process = SidecarProcess {
child: Some(child),
endpoint: "127.0.0.1:41002".parse().unwrap(),
#[cfg(windows)]
job: None,
};
let deadline = Instant::now() + Duration::from_secs(5);
let status = loop {
if let Some(status) = process.try_wait().expect("process status must be readable") {
break status;
}
assert!(Instant::now() < deadline, "fixture process did not exit");
thread::sleep(Duration::from_millis(10));
};

assert_eq!(status.code(), Some(7));
}

#[test]
fn allow_host_patterns_are_anchored_and_escape_regex_metacharacters() {
assert_eq!(
Expand Down
Loading
Loading