diff --git a/runner/Cargo.toml b/runner/Cargo.toml index 5b4444a8..dabcff2f 100644 --- a/runner/Cargo.toml +++ b/runner/Cargo.toml @@ -68,3 +68,9 @@ features = ["signal", "process"] lto = true codegen-units = 1 strip = true + +[profile.dev] +debug = "line-tables-only" + +[profile.dev.package."*"] +debug = false diff --git a/runner/README.md b/runner/README.md index 6bdbb43b..6da74783 100644 --- a/runner/README.md +++ b/runner/README.md @@ -121,6 +121,16 @@ cargo test cargo run --bin zone-server ``` +### Debug Info + +Dev and test builds carry line tables for workspace crates and no debuginfo +for dependencies, so backtraces keep file and line while the build cache stays +small. Full debuginfo for workspace crates, when a debugger needs variables: + +```bash +CARGO_PROFILE_DEV_DEBUG=full cargo build -p zone_server +``` + ### sqlx Offline Query Cache `runner/.cargo/config.toml` sets `SQLX_OFFLINE=true`, so every `sqlx::query!` diff --git a/runner/zone_server/src/services/login/status.rs b/runner/zone_server/src/services/login/status.rs index b1f4fbb3..a33c40b8 100644 --- a/runner/zone_server/src/services/login/status.rs +++ b/runner/zone_server/src/services/login/status.rs @@ -80,14 +80,8 @@ impl AgentStatus { viewer: Viewer, attempt: Option, ) -> Result { - let logins = agent_logins::list_for(state.db(), organization, agent.as_str()).await?; - let snapshots = refreshed(state, &logins).await; - let now = Utc::now(); - let readings: Vec = logins - .iter() - .zip(snapshots) - .map(|(login, snapshot)| read(state, login, snapshot, logins.len(), now)) - .collect(); + // The sign-in in flight is read before the logins: a codex sign-in records its login + // before it drops its attempt, so a read that finds no attempt finds the login. let mut status = Self::signed_out(agent); let pending = match agent { AgentKind::Claude => { @@ -97,10 +91,19 @@ impl AgentStatus { None } AgentKind::Codex => { + let pending = devices::pending(organization); status.error = devices::failure(organization); - devices::pending(organization) + pending } }; + let logins = agent_logins::list_for(state.db(), organization, agent.as_str()).await?; + let snapshots = refreshed(state, &logins).await; + let now = Utc::now(); + let readings: Vec = logins + .iter() + .zip(snapshots) + .map(|(login, snapshot)| read(state, login, snapshot, logins.len(), now)) + .collect(); let best = readings .iter() diff --git a/runner/zone_server/tests/integration/agent_login.rs b/runner/zone_server/tests/integration/agent_login.rs index 8e340e74..e2a6a816 100644 --- a/runner/zone_server/tests/integration/agent_login.rs +++ b/runner/zone_server/tests/integration/agent_login.rs @@ -457,13 +457,23 @@ impl Stage { /// Codex runs from `executable`, with its homes under a private state root. async fn codex(executable: &Path) -> Self { + Self::codex_reading_usage( + executable, + MockServer::start().await, + AgentConfig::default().usage_ttl, + ) + .await + } + + /// A codex server that reads its logins' usage from `api` once a reading is `ttl` old. + async fn codex_reading_usage(executable: &Path, api: MockServer, ttl: Duration) -> Self { let state = TempDir::new().expect("an agent state root"); - let api = MockServer::start().await; let config = Config { model_backend: codex_at(executable.to_path_buf()), agents: AgentConfig { state: state.path().join("agents"), host_login: false, + usage_ttl: ttl, claude_api_url: api.uri(), codex_api_url: api.uri(), ..AgentConfig::default() @@ -3511,6 +3521,37 @@ async fn a_member_reads_every_login_and_never_a_token() { assert_eq!(stage.status(organization, "claude", &member).await, *claude); } +#[tokio::test] +async fn a_status_read_while_a_codex_sign_in_ends_never_misses_the_login_it_saved() { + const SLOW: Duration = Duration::from_secs(3); + let api = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/backend-api/wham/usage")) + .respond_with(ResponseTemplate::new(503).set_delay(SLOW)) + .mount(&api) + .await; + let codex = Codex::new(); + let stage = Stage::codex_reading_usage(&codex.executable, api, Duration::ZERO).await; + let owner = person(&stage.client).await; + let organization = organization(&stage.client, &owner).await; + stage.sign_in_codex(organization, &codex, &owner).await; + fs::remove_file(codex.marker(Codex::APPROVE)).expect("the first approval withdrawn"); + codex.saves(OTHER_ACCOUNT); + stage.start(organization, "codex", json!({}), &owner).await; + + let (status, ()) = tokio::join!(stage.status(organization, "codex", &owner), async { + tokio::time::sleep(SLOW / 6).await; + codex.approve(); + }); + + let logins = status["logins"].as_array().map_or(0, Vec::len); + assert!( + status["state"] == "pending" || logins == 2, + "a status that is no longer pending left out the login that ended it: {status}" + ); + stage.sign_out(organization, "codex", &owner).await; +} + #[tokio::test] async fn cancelling_a_codex_device_sign_in_keeps_the_logins_already_held() { let codex = Codex::new(); diff --git a/runner/zone_server/tests/integration/chat_ws.rs b/runner/zone_server/tests/integration/chat_ws.rs index 234d6b35..533d5165 100644 --- a/runner/zone_server/tests/integration/chat_ws.rs +++ b/runner/zone_server/tests/integration/chat_ws.rs @@ -11,8 +11,8 @@ use crate::common; use crate::common::{ - MEDIA, TestClient, create_test_pool, create_test_router, create_test_state, init_tracing, - test_config, test_email, test_password, + MEDIA, REFUSED_URL, TestClient, create_test_pool, create_test_router, create_test_state, + init_tracing, test_config, test_email, test_password, }; use base64::Engine; use futures_util::{SinkExt, StreamExt}; @@ -1979,16 +1979,32 @@ async fn test_upscale_output_too_large_says_so_rather_than_claiming_nothing_came let _ = tokio::fs::remove_dir_all(artifact_root).await; } +#[tokio::test] +async fn the_refused_url_refuses_connections_on_a_port_never_handed_out() { + let address: std::net::SocketAddr = REFUSED_URL.trim_start_matches("http://").parse().unwrap(); + let connection = tokio::time::timeout( + Duration::from_secs(5), + tokio::net::TcpStream::connect(address), + ) + .await + .expect("a refused connection fails at once rather than timing out"); + assert_eq!( + connection.map(|_| ()).unwrap_err().kind(), + std::io::ErrorKind::ConnectionRefused + ); + assert!( + address.port() < 1024, + "binding port 0 hands out only unprivileged ports, so no parallel test can take this one" + ); +} + #[tokio::test] async fn test_upscale_reports_an_unreachable_comfyui_without_an_empty_message() { - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let refused = listener.local_addr().unwrap(); - drop(listener); let artifact_root = std::env::temp_dir().join(format!("zone-ws-upscale-refused-{}", uuid::Uuid::new_v4())); let mut config = test_config(); config.comfyui.enabled = true; - config.comfyui.base_url = format!("http://{refused}"); + config.comfyui.base_url = REFUSED_URL.to_string(); config.comfyui.poll_interval_ms = 50; config.comfyui.artifact_root = artifact_root.clone(); let (_client, _token, _chat_id, mut socket) = image_socket(config).await; @@ -2635,7 +2651,7 @@ async fn test_image_status_precedes_stalled_prompt_and_timeout_is_visible() { config.comfyui.enabled = true; config.comfyui.base_url = comfy.uri(); config.comfyui.request_timeout_secs = 1; - config.litellm_host = "http://127.0.0.1:9".to_string(); + config.litellm_host = REFUSED_URL.to_string(); let (_, _, _, mut socket) = image_socket(config).await; socket.send(WsMessage::Text(json!({"type":"send", "content":"Generate an image of the same rooster facing the other way"}).to_string().into())).await.unwrap(); let mut status = false; @@ -2666,12 +2682,9 @@ async fn test_image_status_precedes_stalled_prompt_and_timeout_is_visible() { #[tokio::test] async fn test_image_refused_connection_reports_error_without_empty_message() { - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let address = listener.local_addr().unwrap(); - drop(listener); let mut config = test_config(); config.comfyui.enabled = true; - config.comfyui.base_url = format!("http://{address}"); + config.comfyui.base_url = REFUSED_URL.to_string(); let (client, token, chat_id, mut socket) = image_socket(config).await; socket .send(WsMessage::Text( diff --git a/runner/zone_server/tests/integration/common/mod.rs b/runner/zone_server/tests/integration/common/mod.rs index 8f2e0617..e21ab29a 100644 --- a/runner/zone_server/tests/integration/common/mod.rs +++ b/runner/zone_server/tests/integration/common/mod.rs @@ -28,6 +28,10 @@ use zone_server::config::Config; use zone_server::routes::create_router; use zone_server::state::AppState; +/// Refuses every connection. A privileged port: binding port 0 never hands it +/// out, so unlike a freed ephemeral port no parallel test's server can take it. +pub const REFUSED_URL: &str = "http://127.0.0.1:1"; + /// The server runs one media generation per process. Under plain `cargo test` every module /// shares that process, so a test that needs the lane to itself holds this first. pub static MEDIA: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); diff --git a/runner/zone_server/tests/integration/lora_flows.rs b/runner/zone_server/tests/integration/lora_flows.rs index c166779b..aeac6ad5 100644 --- a/runner/zone_server/tests/integration/lora_flows.rs +++ b/runner/zone_server/tests/integration/lora_flows.rs @@ -234,13 +234,6 @@ async fn mock_ollama() -> String { format!("http://{addr}") } -async fn dead_ollama() -> String { - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = listener.local_addr().unwrap(); - drop(listener); - format!("http://{addr}") -} - fn plant_image_bases(models_dir: &std::path::Path) { fs::write( models_dir.join("checkpoints/flux1-schnell-fp8.safetensors"), @@ -524,7 +517,7 @@ async fn comfy_pull_writes_lora_from_hub_origin() { axum::serve(listener, hub).await.unwrap(); }); - let mut config = common::test_config_with_ollama_host("http://127.0.0.1:9"); + let mut config = common::test_config_with_ollama_host(common::REFUSED_URL); config.huggingface_models_url = format!("http://{hub_addr}/api/models"); config.comfyui.models_dir = models_dir.clone(); let pool = common::create_test_pool().await; @@ -1476,9 +1469,9 @@ async fn train_bases_lists_chat_models_alongside_image_rows() { async fn train_bases_keep_image_rows_when_ollama_is_down() { let models_dir = temp_models(); plant_image_bases(&models_dir); - let ollama = dead_ollama().await; let catalog = start_catalog(split_catalog).await; - let (router, token) = router_with(&ollama, &catalog, models_dir.clone(), None).await; + let (router, token) = + router_with(common::REFUSED_URL, &catalog, models_dir.clone(), None).await; let (status, body) = get_json(router, &token, "/api/models/train/bases").await; assert_eq!(status, StatusCode::OK, "{body}"); let bases = body.as_array().expect("bases list");