From c9341c43195cb020f19a4e8041291f5718d2b20f Mon Sep 17 00:00:00 2001 From: xubo9367 <147725110+xubo9367@users.noreply.github.com> Date: Fri, 23 Oct 2026 11:23:30 +0000 Subject: [PATCH 1/2] Bound interview token startup and cleanup Bound server and browser waits around token startup. Release consent when a dispatched room never admits a candidate, and preserve successful startup behavior across dispatch boundaries. --- src/accounts/interviews.rs | 16 +- src/dispatch.rs | 40 +++- src/livekit.rs | 131 +++++++---- src/web/mod.rs | 54 ++++- src/web/token.rs | 65 ++++-- tests/browser/dom-contract.test.js | 2 +- tests/browser/recording-consent.test.js | 23 ++ tests/unit/dispatch.rs | 126 +++++++++++ tests/unit/livekit.rs | 9 + tests/unit/web/token.rs | 11 + tests/web.rs | 72 +++++- tests/web/recording.rs | 280 ++++++++++++++++++++++++ web/interview.js | 12 +- 13 files changed, 765 insertions(+), 76 deletions(-) diff --git a/src/accounts/interviews.rs b/src/accounts/interviews.rs index 9377a0414..96b83d7a0 100644 --- a/src/accounts/interviews.rs +++ b/src/accounts/interviews.rs @@ -198,9 +198,8 @@ pub fn interview_for_account( /// /// The claim is spent before the interviewer is dispatched, so that two /// requests racing on one interview cannot both start a room. A dispatcher that -/// refuses then leaves a consent bound to a room nobody will join, and the -/// candidate is told to retry a request that would be refused. This is that -/// retry working. +/// refuses or a dispatched room whose candidate never arrives would otherwise +/// leave consent bound to a room nobody will join. This gives that claim back. /// /// Guarded on the room it is giving back, so a release arriving late cannot /// unbind a room some other request has since claimed. @@ -219,7 +218,16 @@ pub fn release_interview_room( connection.execute( " UPDATE interviews SET room_name = NULL - WHERE id = ?1 AND account_id = ?2 AND room_name = ?3 + WHERE id = ?1 + AND account_id = ?2 + AND room_name = ?3 + AND NOT EXISTS ( + SELECT 1 FROM interviews AS pending + WHERE pending.account_id = interviews.account_id + AND pending.consent_version = interviews.consent_version + AND pending.room_name IS NULL + AND pending.consent_withdrawn_at IS NULL + ) ", (id, account_id, room_name), )?; diff --git a/src/dispatch.rs b/src/dispatch.rs index 4f37e6b04..32e629093 100644 --- a/src/dispatch.rs +++ b/src/dispatch.rs @@ -9,7 +9,7 @@ use std::collections::{HashMap, HashSet}; use std::sync::{Arc, Mutex}; use crate::config::{AgentConfig, Provider}; -use crate::web::{DispatchRefusal, RoomDispatcher}; +use crate::web::{DispatchRefusal, NoShowConsent, RoomDispatcher}; const MAX_CONCURRENT_GUEST_INTERVIEWS: usize = 2; @@ -46,7 +46,16 @@ impl RoomDispatcher for LocalDispatcher { } fn ensure_agent(&self, room_name: &str, provider: &Provider) -> Result<(), DispatchRefusal> { - self.ensure_agent_with_kind(room_name, provider, false) + self.ensure_agent_with_kind(room_name, provider, false, None) + } + + fn ensure_agent_with_no_show_release( + &self, + room_name: &str, + provider: &Provider, + no_show: Option, + ) -> Result<(), DispatchRefusal> { + self.ensure_agent_with_kind(room_name, provider, false, no_show) } fn ensure_guest_agent( @@ -54,7 +63,7 @@ impl RoomDispatcher for LocalDispatcher { room_name: &str, provider: &Provider, ) -> Result<(), DispatchRefusal> { - self.ensure_agent_with_kind(room_name, provider, true) + self.ensure_agent_with_kind(room_name, provider, true, None) } } @@ -64,6 +73,7 @@ impl LocalDispatcher { room_name: &str, provider: &Provider, guest: bool, + no_show: Option, ) -> Result<(), DispatchRefusal> { let slot = match self.reserve_with_kind(room_name, guest) { Reservation::Existing => return Ok(()), @@ -93,7 +103,7 @@ impl LocalDispatcher { // outlives the response by the length of the interview. self.runtime.spawn(async move { eprintln!("codetrial dispatch room={}", slot.room_name); - if let Err(error) = crate::livekit::run_room_with_slot( + match crate::livekit::run_room_with_slot( &config, &slot.room_name, crate::web::current_epoch_seconds(), @@ -101,13 +111,33 @@ impl LocalDispatcher { ) .await { - eprintln!("codetrial agent_failed room={}: {error}", slot.room_name); + Ok(crate::livekit::RoomRunOutcome::NoCandidate) => { + release_no_show(no_show, &slot.room_name).await; + } + Ok(crate::livekit::RoomRunOutcome::FailedBeforeCandidate(error)) => { + eprintln!("codetrial agent_failed room={}: {error}", slot.room_name); + release_no_show(no_show, &slot.room_name).await; + } + Ok(crate::livekit::RoomRunOutcome::Completed) => {} + Err(error) => { + eprintln!("codetrial agent_failed room={}: {error}", slot.room_name); + } } }); Ok(()) } } +async fn release_no_show(no_show: Option, room_name: &str) { + if let Some(no_show) = no_show + && let Err(error) = no_show.release().await + { + eprintln!( + "could not release interview consent after a no-show in room={room_name}: {error}" + ); + } +} + enum Reservation { Existing, New(Slot), diff --git a/src/livekit.rs b/src/livekit.rs index a7cca112c..29e1812ce 100644 --- a/src/livekit.rs +++ b/src/livekit.rs @@ -1531,6 +1531,12 @@ struct OpenSession<'a> { restarts: usize, } +enum OpenSessionOutcome<'a> { + Started(Box>), + NoCandidate, + FailedBeforeCandidate(Box), +} + /// Joins, waits for a candidate, brings up audio and Gemini, and greets. /// /// Split from [`run_room`] because none of it repeats: it runs once, in order, @@ -1538,29 +1544,40 @@ struct OpenSession<'a> { /// before it starts. What is left in `run_room` is the part that runs many /// times, which is the part worth reading as a loop. /// -/// `Ok(None)` is nobody joined. That is not an error: the room was named, the -/// candidate never arrived, and there is nothing to run or report. +/// `NoCandidate` is nobody joined. That is not an error: the room was named, +/// the candidate never arrived, and there is nothing to run or report. async fn open_session<'a>( config: &'a AgentConfig, keys: &Arc, room_name: &'a str, now_seconds: u64, session_id: u64, -) -> Result>, Box> { +) -> Result, Box> { let agent_identity = agent_identity(room_name); - let (room, mut events) = join_room(config, room_name, &agent_identity, now_seconds).await?; + let (room, mut events) = match join_room(config, room_name, &agent_identity, now_seconds).await + { + Ok(joined) => joined, + Err(error) => return Ok(OpenSessionOutcome::FailedBeforeCandidate(error)), + }; let room = RoomLink::live(room); let mut agent_state = session::AgentState::default(); - session::publish_opening_attributes(&room, &mut agent_state, config.gemini_silence_ms).await?; + if let Err(error) = + session::publish_opening_attributes(&room, &mut agent_state, config.gemini_silence_ms).await + { + return Ok(OpenSessionOutcome::FailedBeforeCandidate(error)); + } // The candidate's metadata picks the problem, so nothing else can start // until someone joins. let mut deferred_events = Vec::new(); - let Some((candidate_identity, candidate_metadata)) = - candidate_metadata(room.connected(), &mut events, &mut deferred_events).await? - else { + let candidate = + match candidate_metadata(room.connected(), &mut events, &mut deferred_events).await { + Ok(candidate) => candidate, + Err(error) => return Ok(OpenSessionOutcome::FailedBeforeCandidate(error)), + }; + let Some((candidate_identity, candidate_metadata)) = candidate else { eprintln!("no candidate joined room={room_name}; leaving it"); - return Ok(None); + return Ok(OpenSessionOutcome::NoCandidate); }; let events = tokio::sync::Mutex::new(events); let boot = candidate_bootstrap(config, room_name, Some(&candidate_metadata)); @@ -1699,7 +1716,7 @@ async fn open_session<'a>( turn.activity .mark_prompted(Instant::now(), Some(&greeting), false); - Ok(Some(OpenSession { + Ok(OpenSessionOutcome::Started(Box::new(OpenSession { room, events, agent_identity, @@ -1711,7 +1728,7 @@ async fn open_session<'a>( turn, started_at, restarts, - })) + }))) } /// The room loop's own bookkeeping, distinct from the session it runs on. @@ -2585,7 +2602,37 @@ pub async fn run_room( room_name: &str, now_seconds: u64, ) -> Result<(), Box> { - run_room_with_slot(config, room_name, now_seconds, None).await + match run_room_with_slot(config, room_name, now_seconds, None).await? { + RoomRunOutcome::Completed | RoomRunOutcome::NoCandidate => Ok(()), + RoomRunOutcome::FailedBeforeCandidate(error) => Err(error), + } +} + +#[derive(Debug)] +pub(crate) enum RoomRunOutcome { + Completed, + NoCandidate, + FailedBeforeCandidate(Box), +} + +fn startup_failure_line( + model: &str, + room_name: &str, + session_id: u64, + error: &(dyn std::error::Error + Send + Sync + 'static), +) -> String { + // A socket may have opened before the failure, so no count is claimed; no + // generation was asked for, so there is no usage. + let outcome = LiveOutcome::from_error(error, LiveOutcome::Error); + live_usage_line( + room_name, + session_id, + model, + 0, + outcome, + None, + &crate::gemini::TokenUsage::default(), + ) } pub(crate) async fn run_room_with_slot( @@ -2593,7 +2640,7 @@ pub(crate) async fn run_room_with_slot( room_name: &str, now_seconds: u64, slot: Option<&crate::dispatch::Slot>, -) -> Result<(), Box> { +) -> Result> { // Names this run on every usage line. Milliseconds rather than the // dispatch's seconds, so a room retried within the same second is not // summed with the run before it. @@ -2603,7 +2650,36 @@ pub(crate) async fn run_room_with_slot( u64::try_from(since.as_millis()).unwrap_or(u64::MAX) }); let keys = Arc::new(GeminiKeys::from_config(config)); - let Some(OpenSession { + let session = open_session(config, &keys, room_name, now_seconds, session_id) + .await + .inspect_err(|error| { + eprintln!( + "{}", + startup_failure_line( + &config.gemini_live_model, + room_name, + session_id, + error.as_ref(), + ) + ); + })?; + let session = match session { + OpenSessionOutcome::Started(session) => session, + OpenSessionOutcome::NoCandidate => return Ok(RoomRunOutcome::NoCandidate), + OpenSessionOutcome::FailedBeforeCandidate(error) => { + eprintln!( + "{}", + startup_failure_line( + &config.gemini_live_model, + room_name, + session_id, + error.as_ref(), + ) + ); + return Ok(RoomRunOutcome::FailedBeforeCandidate(error)); + } + }; + let OpenSession { room, events, agent_identity, @@ -2615,30 +2691,7 @@ pub(crate) async fn run_room_with_slot( mut turn, started_at, restarts, - }) = open_session(config, &keys, room_name, now_seconds, session_id) - .await - .inspect_err(|error| { - // The interview never started, and the summary is where an operator - // counts those, a first open a depleted project refused above all. - // A socket may have opened before the failure, so no count is - // claimed; no generation was asked for, so there is no usage. - let outcome = LiveOutcome::from_error(error.as_ref(), LiveOutcome::Error); - eprintln!( - "{}", - live_usage_line( - room_name, - session_id, - &config.gemini_live_model, - 0, - outcome, - None, - &crate::gemini::TokenUsage::default(), - ) - ); - })? - else { - return Ok(()); - }; + } = *session; let interview = InterviewContext { config, @@ -2910,7 +2963,7 @@ pub(crate) async fn run_room_with_slot( &turn.activity.live_usage, ) ); - result + result.map(|()| RoomRunOutcome::Completed) } /// A session's usage summary, in the one spelling the log analyzer reads. diff --git a/src/web/mod.rs b/src/web/mod.rs index 492c05802..e5fc66f5d 100644 --- a/src/web/mod.rs +++ b/src/web/mod.rs @@ -12,7 +12,7 @@ use axum::response::{IntoResponse, Response}; use axum::routing::{delete, get, post}; use serde_json::{Value, json}; -use crate::accounts::{Accounts, GitHubLoginConfig}; +use crate::accounts::{Accounts, GitHubLoginConfig, blocking, release_interview_room}; pub use crate::accounts::{ACCOUNT_SCHEMA_VERSION, initialize_account_database}; pub use crate::current_epoch_seconds; @@ -529,6 +529,45 @@ impl AppState { } } +/// The recorded consent to release if a candidate never reaches its room. +#[derive(Clone)] +pub struct NoShowConsent { + accounts: Arc, + interview_id: String, + account_id: i64, + room_name: String, +} + +impl NoShowConsent { + pub(crate) fn new( + accounts: Arc, + interview_id: String, + account_id: i64, + room_name: String, + ) -> Self { + Self { + accounts, + interview_id, + account_id, + room_name, + } + } + + /// Gives a recorded interview's consent back only if it is still bound to + /// the room whose candidate never arrived. + pub async fn release(self) -> rusqlite::Result<()> { + blocking(move || { + release_interview_room( + &self.accounts, + &self.interview_id, + self.account_id, + &self.room_name, + ) + }) + .await + } +} + /// Rate limiting needs the peer address, which only exists when the router is /// served with connect info attached. /// Puts an interviewer in a room this server just named. @@ -559,6 +598,19 @@ pub trait RoomDispatcher: Send + Sync + 'static { provider: &crate::config::Provider, ) -> Result<(), DispatchRefusal>; + /// Starts an interviewer and releases recorded consent if no candidate + /// reaches the dispatched room. Dispatchers that cannot observe a no-show + /// keep their existing behavior. + fn ensure_agent_with_no_show_release( + &self, + room_name: &str, + provider: &crate::config::Provider, + no_show: Option, + ) -> Result<(), DispatchRefusal> { + let _ = no_show; + self.ensure_agent(room_name, provider) + } + /// Whether this dispatcher enforces guest-specific admission limits. /// /// Custom dispatchers must opt in only when they enforce those limits. diff --git a/src/web/token.rs b/src/web/token.rs index 64da31b47..213b1a990 100644 --- a/src/web/token.rs +++ b/src/web/token.rs @@ -7,14 +7,13 @@ use std::net::{IpAddr, SocketAddr}; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; -use axum::body::{Body, to_bytes}; +use axum::body::{Body, Bytes, to_bytes}; use axum::extract::{ConnectInfo, State}; use axum::http::{Request, StatusCode}; use axum::response::{IntoResponse, Response}; use ring::rand::SecureRandom; use serde_json::{Value, json}; -use crate::accounts::{blocking, release_interview_room}; use crate::config::{DEFAULT_DURATION_MIN, MAX_DURATION_MIN, MIN_DURATION_MIN}; use crate::current_epoch_seconds; use crate::token::{LivekitTokenInput, TOKEN_TTL_SECONDS, livekit_observer_token, livekit_token}; @@ -23,8 +22,8 @@ use super::auth::{Owner, current_user}; use super::consent::{checked_consent, claim_consent}; use super::pool::{ProviderChoice, room_and_available_provider}; use super::{ - AppState, MAX_BODY_BYTES, client_ip, json_response, number_json, rate_limited_response, - unauthorized_response, + AppState, MAX_BODY_BYTES, NoShowConsent, client_ip, json_response, number_json, + rate_limited_response, unauthorized_response, }; /// No `Debug`, for the reason [`crate::config::AgentConfig`] has none: one @@ -91,6 +90,24 @@ pub const TOKEN_RATE_LIMIT: u32 = 30; pub(crate) const TOKEN_RATE_WINDOW: Duration = Duration::from_secs(60); +/// A partial upload must not hold a token request open forever. This deadline +/// ends before any room is reserved, consent is claimed, or agent is started. +const TOKEN_BODY_READ_LIMIT: Duration = Duration::from_secs(10); + +#[derive(Debug, PartialEq, Eq)] +enum TokenBodyReadError { + TimedOut, + Rejected, +} + +async fn read_token_body(body: Body, limit: Duration) -> Result { + match tokio::time::timeout(limit, to_bytes(body, MAX_BODY_BYTES)).await { + Err(_) => Err(TokenBodyReadError::TimedOut), + Ok(Err(_)) => Err(TokenBodyReadError::Rejected), + Ok(Ok(body)) => Ok(body), + } +} + /// The key is what the route can name a caller by. `/api/token` and /// `/api/login` are reachable without a credential, so they key on the address; /// replay ingestion runs behind `Owner` and keys on the account. The window is @@ -533,28 +550,29 @@ async fn dispatch_or_release_consent( return None; }; + let no_show = owner.zip(interview).map(|(owner, interview)| { + NoShowConsent::new( + owner.accounts.clone(), + interview.clone(), + owner.user.id, + room_name.to_string(), + ) + }); + let dispatch_result = if owner.is_none() { dispatcher.ensure_guest_agent(room_name, provider) } else { - dispatcher.ensure_agent(room_name, provider) + dispatcher.ensure_agent_with_no_show_release(room_name, provider, no_show.clone()) }; let Err(refusal) = dispatch_result else { return None; }; - if let (Some(owner), Some(interview)) = (owner, interview) { - let accounts = owner.accounts.clone(); - let interview = interview.clone(); - let room_name = room_name.to_string(); - let owner_id = owner.user.id; - - if let Err(error) = - blocking(move || release_interview_room(&accounts, &interview, owner_id, &room_name)) - .await - { - eprintln!("could not release interview consent after a refused dispatch: {error}"); - } + if let Some(no_show) = no_show + && let Err(error) = no_show.release().await + { + eprintln!("could not release interview consent after a refused dispatch: {error}"); } let error = match refusal { @@ -614,8 +632,17 @@ pub(crate) async fn token_handler( Err(response) => return response, }; - let Ok(body) = to_bytes(request.into_body(), MAX_BODY_BYTES).await else { - return StatusCode::PAYLOAD_TOO_LARGE.into_response(); + let body = match read_token_body(request.into_body(), TOKEN_BODY_READ_LIMIT).await { + Ok(body) => body, + Err(TokenBodyReadError::TimedOut) => { + return json_response( + StatusCode::REQUEST_TIMEOUT, + json!({ "error": "Session request body timed out." }), + ); + } + Err(TokenBodyReadError::Rejected) => { + return StatusCode::PAYLOAD_TOO_LARGE.into_response(); + } }; // Before the token is minted, and therefore before any Egress call can be diff --git a/tests/browser/dom-contract.test.js b/tests/browser/dom-contract.test.js index f40e22a69..d0d0a24fc 100644 --- a/tests/browser/dom-contract.test.js +++ b/tests/browser/dom-contract.test.js @@ -861,7 +861,7 @@ test("a refused interview says why instead of silently degrading", () => { const script = interviewSource(); const connect = functionBody(script, "connect"); - assert.match(connect, /providerUiState\("degraded", error\?\.message\)/); + assert.match(connect, /providerUiState\("degraded", failure\?\.message\)/); // The reason has to reach the candidate, not only the console. const bannerAt = connect.indexOf('setBanner("connection", degraded.message)'); const offlineAt = connect.indexOf("Offline mode is ready"); diff --git a/tests/browser/recording-consent.test.js b/tests/browser/recording-consent.test.js index 0d2a697fd..f4c4a8075 100644 --- a/tests/browser/recording-consent.test.js +++ b/tests/browser/recording-consent.test.js @@ -196,6 +196,29 @@ test("recording-consent is recorded before a token is asked for", () => { ); }); +test("recording-consent bounds a stalled token request before offline practice", () => { + const body = withoutComments(functionBody(script, "connect")); + assert.ok( + code.includes("const TOKEN_REQUEST_TIMEOUT_MS = 10_000;"), + "the token request needs a named ten-second deadline", + ); + assertIncludesCompact( + body, + "signal: AbortSignal.timeout(TOKEN_REQUEST_TIMEOUT_MS)", + "the deadline signal must cover the token request and its response body", + ); + assert.ok( + body.includes('error?.name === "TimeoutError"'), + "a timeout must not expose browser-specific exception text", + ); + assert.ok( + body.includes( + "The interview connection timed out. Offline practice is ready.", + ), + "the candidate needs a stable explanation before the existing offline fallback", + ); +}); + test("recording-consent sends the version it displayed", () => { const body = withoutComments(functionBody(script, "recordConsent")); assert.ok( diff --git a/tests/unit/dispatch.rs b/tests/unit/dispatch.rs index 058efb8d2..f9d1009a4 100644 --- a/tests/unit/dispatch.rs +++ b/tests/unit/dispatch.rs @@ -78,6 +78,132 @@ async fn the_configured_cap_is_the_one_enforced() { ); } +/// The no-show-aware entry point must retain the local dispatcher's admission +/// checks instead of accepting a room without reserving or staffing it. +#[tokio::test] +async fn no_show_dispatch_enforces_the_configured_cap() { + let state = Arc::new(Mutex::new(DispatchState::default())); + state + .lock() + .unwrap() + .live + .insert("interview-first".to_string(), true); + let dispatcher = LocalDispatcher { + config: crate::config::load_from_pairs([ + ("LIVEKIT_URL", "wss://primary.example"), + ("LIVEKIT_API_KEY", "primary-key"), + ("LIVEKIT_API_SECRET", "primary-secret"), + ("GOOGLE_API_KEY", "primary-google"), + ]) + .unwrap(), + runtime: tokio::runtime::Handle::current(), + state, + max_concurrent: 1, + }; + let provider = Provider { + id: crate::config::PRIMARY_PROVIDER_ID.to_string(), + url: "wss://primary.example".to_string(), + api_key: "primary-key".to_string(), + api_secret: "primary-secret".to_string(), + google_api_keys: Vec::new(), + }; + + assert_eq!( + dispatcher.ensure_agent_with_no_show_release("interview-second", &provider, None), + Err(DispatchRefusal::AtCapacity) + ); +} + +/// Removes a temporary account database even when its test unwinds early. +struct Scratch(std::path::PathBuf); + +impl Scratch { + fn new(label: &str) -> Self { + Self(std::env::temp_dir().join(format!( + "codetrial-{label}-{}-{}.db", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + ))) + } +} + +impl Drop for Scratch { + fn drop(&mut self) { + for suffix in ["", "-wal", "-shm"] { + let _ = std::fs::remove_file(format!("{}{suffix}", self.0.display())); + } + } +} + +/// The task's shared cleanup helper is the production edge between a no-show +/// outcome and the database release. Exercising it directly makes removing the +/// call observable without waiting through a LiveKit room deadline. +#[tokio::test] +async fn release_no_show_gives_the_claimed_room_back() { + let path = Scratch::new("dispatch-release"); + crate::accounts::initialize_account_database(&path.0).unwrap(); + let accounts = Arc::new( + crate::accounts::Accounts::open(crate::accounts::GitHubLoginConfig { + oauth: None, + session_secret: "session".into(), + db_path: path.0.clone(), + oauth_base_url: String::new(), + api_base_url: String::new(), + }) + .unwrap(), + ); + let session = crate::accounts::create_session( + &accounts, + &crate::accounts::GitHubProfile { + github_id: -1, + login: "release-no-show".into(), + avatar_url: None, + verified_email: None, + }, + ) + .unwrap(); + let account_id = crate::accounts::session_user(&accounts, &session) + .unwrap() + .unwrap() + .id; + let interview_id = "interview-release-no-show"; + let room_name = "interview-release-room"; + crate::accounts::create_interview(&accounts, interview_id, account_id, "v1", 1) + .unwrap() + .unwrap(); + assert!( + crate::accounts::claim_interview_room(&accounts, interview_id, account_id, room_name) + .unwrap() + ); + + release_no_show( + Some(NoShowConsent::new( + Arc::clone(&accounts), + interview_id.into(), + account_id, + room_name.into(), + )), + room_name, + ) + .await; + + let room: Option = accounts + .with(|connection| { + connection.query_row( + "SELECT room_name FROM interviews WHERE id = ?1", + [interview_id], + |row| row.get(0), + ) + }) + .unwrap(); + assert_eq!(room, None); + + drop(accounts); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn a_concurrent_burst_never_overbooks_and_released_capacity_returns() { let state = Arc::new(Mutex::new(DispatchState::default())); diff --git a/tests/unit/livekit.rs b/tests/unit/livekit.rs index 95091c3e2..d95abfcbd 100644 --- a/tests/unit/livekit.rs +++ b/tests/unit/livekit.rs @@ -3859,6 +3859,15 @@ fn both_usage_summaries_share_one_spelling() { ), "{startup}" ); + + let error = std::io::Error::other("startup failed"); + let failed = startup_failure_line("live", "room-a", 7, &error); + assert!( + failed.starts_with( + "codetrial live_usage room=room-a session=7 model=live elapsed_s=0 outcome=error phase=startup prompt_tokens=0" + ), + "{failed}" + ); } /// The window reaches the interview's state, which is the one place the diff --git a/tests/unit/web/token.rs b/tests/unit/web/token.rs index a69fd934c..2f237f360 100644 --- a/tests/unit/web/token.rs +++ b/tests/unit/web/token.rs @@ -21,6 +21,17 @@ fn peer() -> SocketAddr { SocketAddr::from(([10, 0, 0, 9], 4000)) } +#[tokio::test] +async fn token_body_read_has_a_deadline() { + let stalled = futures_util::stream::pending::>(); + let body = Body::from_stream(stalled); + + assert_eq!( + read_token_body(body, Duration::from_millis(10)).await, + Err(TokenBodyReadError::TimedOut), + ); +} + #[test] fn generated_candidate_identity_has_the_livekit_prefix() { let generated = generated_candidate_identity(); diff --git a/tests/web.rs b/tests/web.rs index 82f05d92d..eaebe5f4f 100644 --- a/tests/web.rs +++ b/tests/web.rs @@ -24,9 +24,9 @@ use codetrial::token::{ livekit_token, }; use codetrial::web::{ - DispatchRefusal, MAX_BODY_BYTES, MAX_REPORT_BYTES, READ_RATE_LIMIT, REPLAY_RATE_LIMIT, - RoomDispatcher, TOKEN_RATE_LIMIT, TokenConfig, WebServerConfig, initialize_account_database, - login_config, static_file_meta, token_response, + DispatchRefusal, MAX_BODY_BYTES, MAX_REPORT_BYTES, NoShowConsent, READ_RATE_LIMIT, + REPLAY_RATE_LIMIT, RoomDispatcher, TOKEN_RATE_LIMIT, TokenConfig, WebServerConfig, + initialize_account_database, login_config, static_file_meta, token_response, }; #[path = "common/http.rs"] @@ -348,6 +348,7 @@ async fn spawn_web_server(config: WebServerConfig) -> (String, TestServer) { #[derive(Default)] struct RecordingDispatcher { staffed: std::sync::Mutex>, + no_show_releases: std::sync::Mutex>, /// Set to refuse, standing in for a process already at capacity. at_capacity: std::sync::atomic::AtomicBool, /// Set to refuse, standing in for a fixed room still finishing a report. @@ -376,6 +377,22 @@ impl RoomDispatcher for RecordingDispatcher { .push((room_name.to_string(), provider.url.clone())); Ok(()) } + + fn ensure_agent_with_no_show_release( + &self, + room_name: &str, + provider: &codetrial::config::Provider, + no_show: Option, + ) -> Result<(), DispatchRefusal> { + self.ensure_agent(room_name, provider)?; + if let Some(no_show) = no_show { + self.no_show_releases + .lock() + .unwrap() + .push((room_name.to_string(), no_show)); + } + Ok(()) + } fn ensure_guest_agent( &self, room_name: &str, @@ -389,6 +406,28 @@ impl RoomDispatcher for RecordingDispatcher { } impl RecordingDispatcher { + fn no_show_release(&self, room_name: &str) -> NoShowConsent { + self.no_show_releases + .lock() + .unwrap() + .iter() + .find(|(room, _)| room == room_name) + .map(|(_, no_show)| no_show.clone()) + .expect("the dispatched recorded room should carry its no-show release") + } + + async fn release_no_show(&self, room_name: &str) { + let no_show = { + let mut releases = self.no_show_releases.lock().unwrap(); + let index = releases + .iter() + .position(|(room, _)| room == room_name) + .expect("the dispatched recorded room should carry its no-show release"); + releases.swap_remove(index).1 + }; + no_show.release().await.unwrap(); + } + fn rooms(&self) -> Vec { self.staffed .lock() @@ -403,12 +442,35 @@ impl RecordingDispatcher { } } -async fn spawn_web_server_with_dispatcher( +#[derive(Default)] +struct DefaultMethodDispatcher { + staffed: std::sync::Mutex>, +} + +impl RoomDispatcher for DefaultMethodDispatcher { + fn ensure_agent( + &self, + room_name: &str, + _provider: &codetrial::config::Provider, + ) -> Result<(), DispatchRefusal> { + self.staffed.lock().unwrap().push(room_name.to_string()); + Ok(()) + } +} + +impl DefaultMethodDispatcher { + fn rooms(&self) -> Vec { + self.staffed.lock().unwrap().clone() + } +} + +async fn spawn_web_server_with_dispatcher( config: WebServerConfig, - dispatcher: std::sync::Arc, + dispatcher: std::sync::Arc, ) -> (String, TestServer) { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); + let dispatcher: std::sync::Arc = dispatcher; let server = spawn_test_server( listener, codetrial::web::web_service_with_dispatcher(config, Some(dispatcher)), diff --git a/tests/web/recording.rs b/tests/web/recording.rs index 8133dd87c..ac1cd687b 100644 --- a/tests/web/recording.rs +++ b/tests/web/recording.rs @@ -546,6 +546,286 @@ async fn a_refused_interview_leaves_its_consent_unspent() { remove_database(path).await; } +/// A dispatcher that only implements `ensure_agent` still staffs recorded +/// interviews through the trait's no-show-aware default method. +#[tokio::test] +async fn recorded_tokens_use_the_default_dispatcher_method() { + let (mut config, cookie, path) = signed_in_web_config("default-recording-dispatcher"); + config.recording = Some(recording_config()); + rusqlite::Connection::open(&path) + .unwrap() + .execute( + "UPDATE users SET email = 'one@example.test', email_verified = 1 WHERE id = 1", + [], + ) + .unwrap(); + let dispatcher = std::sync::Arc::new(DefaultMethodDispatcher::default()); + let (base, server) = + spawn_web_server_with_dispatcher(config, std::sync::Arc::clone(&dispatcher)).await; + let client = http_client(); + + let interview = start_interview(&client, &base, &cookie).await; + let response = client + .post(format!("{base}/api/token")) + .header("cookie", cookie) + .json(&json!({ "interviewId": interview })) + .send() + .await + .unwrap(); + + assert_eq!(response.status(), 200); + assert_eq!(dispatcher.rooms().len(), 1); + + server.shutdown().await; + remove_database(path).await; +} + +/// Evidence for issue #380: a deadline at the browser cannot tell whether the +/// request is still before dispatch or whether only the response path stalled. +/// The two cases have different ownership and therefore cannot share the same +/// cleanup rule. +#[tokio::test] +async fn token_stalls_before_and_after_dispatch_have_different_side_effects() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let (mut config, cookie, path) = signed_in_web_config("token-stall-phases"); + config.recording = Some(recording_config()); + { + let connection = rusqlite::Connection::open(&path).unwrap(); + connection + .execute( + "UPDATE users SET email = 'one@example.test', email_verified = 1 WHERE id = 1", + [], + ) + .unwrap(); + } + let dispatcher = std::sync::Arc::new(RecordingDispatcher::default()); + let (base, server) = + spawn_web_server_with_dispatcher(config, std::sync::Arc::clone(&dispatcher)).await; + let address = base.strip_prefix("http://").unwrap(); + let client = http_client(); + + // Case 1: advertise a JSON body larger than the bytes actually sent. The + // server admits the caller, then waits while reading the body. Nothing has + // reserved a room, claimed consent, or dispatched an agent at this point. + let pending_interview = start_interview(&client, &base, &cookie).await; + let partial_body = format!(r#"{{"interviewId":"{pending_interview}""#); + let mut socket = tokio::net::TcpStream::connect(address).await.unwrap(); + socket + .write_all( + format!( + "POST /api/token HTTP/1.1\r\nHost: {address}\r\nCookie: {cookie}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{partial_body}", + partial_body.len() + 100 + ) + .as_bytes(), + ) + .await + .unwrap(); + let before_started = std::time::Instant::now(); + let mut byte = [0u8; 1]; + assert!( + tokio::time::timeout(Duration::from_secs(1), socket.read(&mut byte)) + .await + .is_err(), + "an incomplete request body should leave the handler awaiting body bytes" + ); + let mut response = Vec::new(); + tokio::time::timeout(Duration::from_secs(11), socket.read_to_end(&mut response)) + .await + .expect("the server-side body deadline should finish the request") + .unwrap(); + let response = String::from_utf8(response).unwrap(); + assert!( + response.starts_with("HTTP/1.1 408 Request Timeout\r\n"), + "the body deadline should answer 408, got {response:?}" + ); + assert!( + before_started.elapsed() >= Duration::from_secs(10), + "the production deadline should not fire early" + ); + let pending_room: Option = rusqlite::Connection::open(&path) + .unwrap() + .query_row( + "SELECT room_name FROM interviews WHERE id = ?1", + [&pending_interview], + |row| row.get(0), + ) + .unwrap(); + eprintln!( + "issue380 before_dispatch pending_ms={} dispatched_rooms={} consent_room={pending_room:?}", + before_started.elapsed().as_millis(), + dispatcher.rooms().len() + ); + assert!(dispatcher.rooms().is_empty()); + assert_eq!(pending_room, None); + + // Case 2: a reverse proxy forwards the complete request and receives the + // upstream response, but withholds that response from the browser. The + // browser still sees a pending fetch although dispatch and consent claim + // have already completed on the server. + let dispatched_interview = start_interview(&client, &base, &cookie).await; + let (upstream_done_tx, upstream_done_rx) = tokio::sync::oneshot::channel(); + let upstream_done_tx = std::sync::Arc::new(std::sync::Mutex::new(Some(upstream_done_tx))); + let proxy_base = base.clone(); + let proxy_signal = std::sync::Arc::clone(&upstream_done_tx); + let proxy = axum::Router::new().route( + "/api/token", + axum::routing::post(move |request: axum::extract::Request| { + let proxy_base = proxy_base.clone(); + let proxy_signal = std::sync::Arc::clone(&proxy_signal); + async move { + let cookie = request + .headers() + .get(axum::http::header::COOKIE) + .unwrap() + .to_str() + .unwrap() + .to_string(); + let body = axum::body::to_bytes(request.into_body(), MAX_BODY_BYTES) + .await + .unwrap(); + let upstream = http_client() + .post(format!("{proxy_base}/api/token")) + .header("cookie", cookie) + .header("content-type", "application/json") + .body(body) + .send() + .await + .unwrap(); + if let Some(done) = proxy_signal.lock().unwrap().take() { + let _ = done.send(upstream.status().as_u16()); + } + std::future::pending::().await + } + }), + ); + let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let proxy_address = proxy_listener.local_addr().unwrap(); + let proxy_task = tokio::spawn(async move { + axum::serve( + proxy_listener, + proxy.into_make_service_with_connect_info::(), + ) + .await + }); + let downstream_started = std::time::Instant::now(); + let downstream_interview = dispatched_interview.clone(); + let downstream = tokio::spawn({ + let cookie = cookie.clone(); + async move { + http_client() + .post(format!("http://{proxy_address}/api/token")) + .header("cookie", cookie) + .json(&json!({ "interviewId": downstream_interview })) + .send() + .await + } + }); + let upstream_status = tokio::time::timeout(Duration::from_secs(2), upstream_done_rx) + .await + .expect("the real token handler should finish") + .unwrap(); + assert_eq!(upstream_status, 200); + tokio::time::sleep(Duration::from_secs(1)).await; + let dispatched_room: Option = rusqlite::Connection::open(&path) + .unwrap() + .query_row( + "SELECT room_name FROM interviews WHERE id = ?1", + [&dispatched_interview], + |row| row.get(0), + ) + .unwrap(); + eprintln!( + "issue380 after_dispatch downstream_pending_ms={} dispatched_rooms={} consent_room={dispatched_room:?}", + downstream_started.elapsed().as_millis(), + dispatcher.rooms().len() + ); + assert!( + !downstream.is_finished(), + "the proxy withheld response headers" + ); + assert_eq!(dispatcher.rooms().len(), 1); + assert!(dispatched_room.is_some()); + + let dispatched_room = dispatched_room.unwrap(); + let guarded_release = dispatcher.no_show_release(&dispatched_room); + let replacement_room = "interview-replacement-room"; + rusqlite::Connection::open(&path) + .unwrap() + .execute( + "UPDATE interviews SET room_name = ?1 WHERE id = ?2", + rusqlite::params![replacement_room, &dispatched_interview], + ) + .unwrap(); + guarded_release.release().await.unwrap(); + let room_after_stale_release: Option = rusqlite::Connection::open(&path) + .unwrap() + .query_row( + "SELECT room_name FROM interviews WHERE id = ?1", + [&dispatched_interview], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(room_after_stale_release.as_deref(), Some(replacement_room)); + + rusqlite::Connection::open(&path) + .unwrap() + .execute( + "UPDATE interviews SET room_name = ?1 WHERE id = ?2", + rusqlite::params![&dispatched_room, &dispatched_interview], + ) + .unwrap(); + let pending_after_reload = start_interview(&client, &base, &cookie).await; + assert_ne!(pending_after_reload, dispatched_interview); + + // Returning the claimed row to pending would violate the one-pending-row + // index while the reload's row exists, so the no-show release must wait. + dispatcher + .no_show_release(&dispatched_room) + .release() + .await + .unwrap(); + let (room_while_pending, pending_room): (Option, Option) = + rusqlite::Connection::open(&path) + .unwrap() + .query_row( + "SELECT claimed.room_name, pending.room_name + FROM interviews AS claimed, interviews AS pending + WHERE claimed.id = ?1 AND pending.id = ?2", + rusqlite::params![&dispatched_interview, &pending_after_reload], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .unwrap(); + assert_eq!( + room_while_pending.as_deref(), + Some(dispatched_room.as_str()) + ); + assert_eq!(pending_room, None); + rusqlite::Connection::open(&path) + .unwrap() + .execute( + "DELETE FROM interviews WHERE id = ?1", + [&pending_after_reload], + ) + .unwrap(); + dispatcher.release_no_show(&dispatched_room).await; + let released_room: Option = rusqlite::Connection::open(&path) + .unwrap() + .query_row( + "SELECT room_name FROM interviews WHERE id = ?1", + [&dispatched_interview], + |row| row.get(0), + ) + .unwrap(); + eprintln!("issue380 after_no_show consent_room={released_room:?}"); + assert_eq!(released_room, None); + + downstream.abort(); + proxy_task.abort(); + server.shutdown().await; + remove_database(path).await; +} + /// The per-batch caps bound one request and the stored quota bounds what an /// interview keeps. Neither bounds how often one account may ask, and an ask /// that is not refused before the body is a parse and a stored batch. This is diff --git a/web/interview.js b/web/interview.js index 7080a53b0..ddb3187ee 100644 --- a/web/interview.js +++ b/web/interview.js @@ -221,6 +221,7 @@ let editorPaintFrame = null; // The agent reads the editor once per 2s watch tick, so publishing every // keystroke sends ~10x more full-buffer packets than anyone consumes. const CODE_PUBLISH_DEBOUNCE_MS = 300; +const TOKEN_REQUEST_TIMEOUT_MS = 10_000; /// Long enough to read twice, short enough that it is gone before the answer /// it is about. Measured against the hint text, not chosen round. const FRAMEWORK_HINT_MS = 12000; @@ -1671,6 +1672,7 @@ async function connect(preflight, presenting = false) { const response = await fetch("/api/token", { method: "POST", headers: { "Content-Type": "application/json" }, + signal: AbortSignal.timeout(TOKEN_REQUEST_TIMEOUT_MS), body: JSON.stringify({ problemId: problem.page, durationMin, @@ -1738,11 +1740,17 @@ async function connect(preflight, presenting = false) { }); } } catch (error) { + const failure = + error?.name === "TimeoutError" + ? new Error( + "The interview connection timed out. Offline practice is ready.", + ) + : error; // Swallowed for the candidate, logged for everyone else. Offline practice // mode is a reasonable fallback but a terrible diagnosis: it looks the same // whether the server has no LiveKit credentials, the token was refused, or // the microphone failed to publish after a successful connect. - console.warn("codetrial connect_failed", error); + console.warn("codetrial connect_failed", failure); // Whatever threw may have thrown after `connectLiveKit` already joined and // published, and dropping the reference does not leave the room. That is // how a page ran "offline practice" with a live LiveKit room underneath it, @@ -1753,7 +1761,7 @@ async function connect(preflight, presenting = false) { state.room = null; if (joined) void joined.disconnect?.()?.catch?.(() => {}); stopPreflight(preflight); - const degraded = providerUiState("degraded", error?.message); + const degraded = providerUiState("degraded", failure?.message); setAgentStateLabel(degraded.label); // A degraded start without a reason reads as the product working. The server // says why it refused, in words written for a candidate, and a busy server From b706fc4f496ade2126815a85ba9b0e3e0a5e4a74 Mon Sep 17 00:00:00 2001 From: xubo9367 <147725110+xubo9367@users.noreply.github.com> Date: Fri, 23 Oct 2026 11:22:01 +0000 Subject: [PATCH 2/2] Close token startup release races Keep claim and dispatch together after a client disconnects. Guard release against pending or recording state, and document when browser fallback can precede completed dispatch. --- src/accounts/interviews.rs | 14 +++- src/dispatch.rs | 48 +++++++++---- src/web/mod.rs | 4 +- src/web/token.rs | 96 +++++++++++++++++-------- tests/browser/recording-consent.test.js | 4 +- tests/unit/dispatch.rs | 94 +++++++++++++++++++++--- tests/web/recording.rs | 14 ++-- web/interview.js | 4 +- 8 files changed, 210 insertions(+), 68 deletions(-) diff --git a/src/accounts/interviews.rs b/src/accounts/interviews.rs index 96b83d7a0..028209d85 100644 --- a/src/accounts/interviews.rs +++ b/src/accounts/interviews.rs @@ -204,6 +204,10 @@ pub fn interview_for_account( /// Guarded on the room it is giving back, so a release arriving late cannot /// unbind a room some other request has since claimed. /// +/// Returns whether the claim was released. `false` means the room no longer +/// matches, another pending interview holds this consent version's pending +/// slot, or recording has started for this interview. +/// /// The ceiling: a process that dies between the claim and the release leaves /// the interview bound to a room that never ran, and the candidate has to /// reload. That is one page reload against a lock nobody can take back, which @@ -213,14 +217,18 @@ pub fn release_interview_room( id: &str, account_id: i64, room_name: &str, -) -> rusqlite::Result<()> { +) -> rusqlite::Result { accounts.with(|connection| { - connection.execute( + let changed = connection.execute( " UPDATE interviews SET room_name = NULL WHERE id = ?1 AND account_id = ?2 AND room_name = ?3 + AND NOT EXISTS ( + SELECT 1 FROM recordings + WHERE recordings.interview_id = interviews.id + ) AND NOT EXISTS ( SELECT 1 FROM interviews AS pending WHERE pending.account_id = interviews.account_id @@ -231,7 +239,7 @@ pub fn release_interview_room( ", (id, account_id, room_name), )?; - Ok(()) + Ok(changed > 0) }) } diff --git a/src/dispatch.rs b/src/dispatch.rs index 32e629093..75f903372 100644 --- a/src/dispatch.rs +++ b/src/dispatch.rs @@ -103,20 +103,20 @@ impl LocalDispatcher { // outlives the response by the length of the interview. self.runtime.spawn(async move { eprintln!("codetrial dispatch room={}", slot.room_name); - match crate::livekit::run_room_with_slot( + let outcome = crate::livekit::run_room_with_slot( &config, &slot.room_name, crate::web::current_epoch_seconds(), Some(&slot), ) - .await - { - Ok(crate::livekit::RoomRunOutcome::NoCandidate) => { - release_no_show(no_show, &slot.room_name).await; - } + .await; + if outcome.as_ref().is_ok_and(no_show_release_required) { + release_no_show(no_show, &slot.room_name).await; + } + match outcome { + Ok(crate::livekit::RoomRunOutcome::NoCandidate) => {} Ok(crate::livekit::RoomRunOutcome::FailedBeforeCandidate(error)) => { eprintln!("codetrial agent_failed room={}: {error}", slot.room_name); - release_no_show(no_show, &slot.room_name).await; } Ok(crate::livekit::RoomRunOutcome::Completed) => {} Err(error) => { @@ -128,13 +128,33 @@ impl LocalDispatcher { } } -async fn release_no_show(no_show: Option, room_name: &str) { - if let Some(no_show) = no_show - && let Err(error) = no_show.release().await - { - eprintln!( - "could not release interview consent after a no-show in room={room_name}: {error}" - ); +fn no_show_release_required(outcome: &crate::livekit::RoomRunOutcome) -> bool { + matches!( + outcome, + crate::livekit::RoomRunOutcome::NoCandidate + | crate::livekit::RoomRunOutcome::FailedBeforeCandidate(_) + ) +} + +async fn release_no_show(no_show: Option, room_name: &str) -> bool { + let Some(no_show) = no_show else { + return false; + }; + match no_show.release().await { + Ok(true) => true, + Ok(false) => { + eprintln!( + "codetrial consent_release_skipped room={room_name} \ + reason=claim_changed_pending_or_recording_started" + ); + false + } + Err(error) => { + eprintln!( + "could not release interview consent after a no-show in room={room_name}: {error}" + ); + false + } } } diff --git a/src/web/mod.rs b/src/web/mod.rs index e5fc66f5d..d519cfee9 100644 --- a/src/web/mod.rs +++ b/src/web/mod.rs @@ -554,8 +554,8 @@ impl NoShowConsent { } /// Gives a recorded interview's consent back only if it is still bound to - /// the room whose candidate never arrived. - pub async fn release(self) -> rusqlite::Result<()> { + /// the room whose candidate never arrived, and reports whether it did. + pub async fn release(self) -> rusqlite::Result { blocking(move || { release_interview_room( &self.accounts, diff --git a/src/web/token.rs b/src/web/token.rs index 213b1a990..6989871ec 100644 --- a/src/web/token.rs +++ b/src/web/token.rs @@ -569,10 +569,17 @@ async fn dispatch_or_release_consent( return None; }; - if let Some(no_show) = no_show - && let Err(error) = no_show.release().await - { - eprintln!("could not release interview consent after a refused dispatch: {error}"); + if let Some(no_show) = no_show { + match no_show.release().await { + Ok(true) => {} + Ok(false) => eprintln!( + "codetrial consent_release_skipped room={room_name} \ + reason=claim_changed_or_pending_interview" + ), + Err(error) => { + eprintln!("could not release interview consent after a refused dispatch: {error}") + } + } } let error = match refusal { @@ -590,6 +597,37 @@ async fn dispatch_or_release_consent( )) } +/// Keeps the consent claim and its dispatch in one task if the client drops +/// the request while the blocking database claim is still finishing. +async fn claim_and_dispatch( + state: AppState, + owner: Option, + interview: Option, + room_name: String, + provider: crate::config::Provider, +) -> Option { + if let (Some(owner), Some(interview)) = (owner.as_ref(), interview.as_ref()) + && let Err(response) = claim_consent( + &owner.accounts, + owner.user.id, + interview.clone(), + &room_name, + ) + .await + { + return Some(response); + } + + dispatch_or_release_consent( + &state, + owner.as_ref(), + interview.as_ref(), + &room_name, + &provider, + ) + .await +} + fn minted_token( state: &AppState, provider: &crate::config::Provider, @@ -693,32 +731,30 @@ pub(crate) async fn token_handler( // Before the dispatcher, so two requests racing on one interview cannot // both start an interviewer, and after the token, so a signing failure - // costs nothing. The loser of that race is refused here without dispatching - // anything. - if let (Some(owner), Some(interview)) = (owner.as_ref(), interview.as_ref()) - && let Err(response) = claim_consent( - &owner.accounts, - owner.user.id, - interview.clone(), - &room_name, - ) - .await - { - return response; - } - - // After the token and the claim, so a request that was going to be refused - // anyway never leaves an interviewer running behind it. - if let Some(refusal) = dispatch_or_release_consent( - &state, - owner.as_ref(), - interview.as_ref(), - &room_name, - provider, - ) - .await - { - return refusal; + // costs nothing. Spawned as one operation because dropping a request must + // not cancel the handler after the blocking claim commits but before the + // matching dispatcher sees it. Dropping a Tokio join handle detaches its + // task, so this pair finishes even when the browser has gone away. + let startup = tokio::spawn(claim_and_dispatch( + state.clone(), + owner.as_ref().map(|owner| Owner { + accounts: Arc::clone(&owner.accounts), + user: owner.user.clone(), + }), + interview.clone(), + room_name.clone(), + provider.clone(), + )); + match startup.await { + Ok(Some(refusal)) => return refusal, + Ok(None) => {} + Err(error) => { + eprintln!("could not finish the consent claim and room dispatch: {error}"); + return json_response( + StatusCode::INTERNAL_SERVER_ERROR, + json!({ "error": "Could not start the interview." }), + ); + } } if let Some(owner) = owner.as_ref() { diff --git a/tests/browser/recording-consent.test.js b/tests/browser/recording-consent.test.js index f4c4a8075..f1cf78c28 100644 --- a/tests/browser/recording-consent.test.js +++ b/tests/browser/recording-consent.test.js @@ -199,8 +199,8 @@ test("recording-consent is recorded before a token is asked for", () => { test("recording-consent bounds a stalled token request before offline practice", () => { const body = withoutComments(functionBody(script, "connect")); assert.ok( - code.includes("const TOKEN_REQUEST_TIMEOUT_MS = 10_000;"), - "the token request needs a named ten-second deadline", + code.includes("const TOKEN_REQUEST_TIMEOUT_MS = 20_000;"), + "the browser deadline must leave the server's body deadline time to answer", ); assertIncludesCompact( body, diff --git a/tests/unit/dispatch.rs b/tests/unit/dispatch.rs index f9d1009a4..e2cc4e9e1 100644 --- a/tests/unit/dispatch.rs +++ b/tests/unit/dispatch.rs @@ -179,16 +179,30 @@ async fn release_no_show_gives_the_claimed_room_back() { .unwrap() ); - release_no_show( - Some(NoShowConsent::new( - Arc::clone(&accounts), - interview_id.into(), - account_id, - room_name.into(), - )), - room_name, - ) - .await; + assert!( + release_no_show( + Some(NoShowConsent::new( + Arc::clone(&accounts), + interview_id.into(), + account_id, + room_name.into(), + )), + room_name, + ) + .await + ); + assert!( + !release_no_show( + Some(NoShowConsent::new( + Arc::clone(&accounts), + interview_id.into(), + account_id, + room_name.into(), + )), + room_name, + ) + .await + ); let room: Option = accounts .with(|connection| { @@ -201,9 +215,69 @@ async fn release_no_show_gives_the_claimed_room_back() { .unwrap(); assert_eq!(room, None); + assert!( + crate::accounts::claim_interview_room(&accounts, interview_id, account_id, room_name) + .unwrap() + ); + accounts + .with(|connection| { + connection.execute( + "INSERT INTO recordings ( + id, account_id, interview_id, room_name, idempotency_key, + recipient_email, state, created_at, updated_at + ) VALUES (?1, ?2, ?3, ?4, ?3, ?5, 'starting', 1, 1)", + ( + "recording-release-race", + account_id, + interview_id, + room_name, + "candidate@example.com", + ), + )?; + Ok(()) + }) + .unwrap(); + assert!( + !release_no_show( + Some(NoShowConsent::new( + Arc::clone(&accounts), + interview_id.into(), + account_id, + room_name.into(), + )), + room_name, + ) + .await + ); + let recorded_room: Option = accounts + .with(|connection| { + connection.query_row( + "SELECT room_name FROM interviews WHERE id = ?1", + [interview_id], + |row| row.get(0), + ) + }) + .unwrap(); + assert_eq!(recorded_room.as_deref(), Some(room_name)); + drop(accounts); } +#[test] +fn pre_candidate_outcomes_release_consent() { + assert!(no_show_release_required( + &crate::livekit::RoomRunOutcome::NoCandidate + )); + assert!(no_show_release_required( + &crate::livekit::RoomRunOutcome::FailedBeforeCandidate(Box::new(std::io::Error::other( + "startup failed" + ))) + )); + assert!(!no_show_release_required( + &crate::livekit::RoomRunOutcome::Completed + )); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn a_concurrent_burst_never_overbooks_and_released_capacity_returns() { let state = Arc::new(Mutex::new(DispatchState::default())); diff --git a/tests/web/recording.rs b/tests/web/recording.rs index ac1cd687b..73238c0ad 100644 --- a/tests/web/recording.rs +++ b/tests/web/recording.rs @@ -757,7 +757,7 @@ async fn token_stalls_before_and_after_dispatch_have_different_side_effects() { rusqlite::params![replacement_room, &dispatched_interview], ) .unwrap(); - guarded_release.release().await.unwrap(); + assert!(!guarded_release.release().await.unwrap()); let room_after_stale_release: Option = rusqlite::Connection::open(&path) .unwrap() .query_row( @@ -780,11 +780,13 @@ async fn token_stalls_before_and_after_dispatch_have_different_side_effects() { // Returning the claimed row to pending would violate the one-pending-row // index while the reload's row exists, so the no-show release must wait. - dispatcher - .no_show_release(&dispatched_room) - .release() - .await - .unwrap(); + assert!( + !dispatcher + .no_show_release(&dispatched_room) + .release() + .await + .unwrap() + ); let (room_while_pending, pending_room): (Option, Option) = rusqlite::Connection::open(&path) .unwrap() diff --git a/web/interview.js b/web/interview.js index ddb3187ee..e2cbbf88b 100644 --- a/web/interview.js +++ b/web/interview.js @@ -221,7 +221,9 @@ let editorPaintFrame = null; // The agent reads the editor once per 2s watch tick, so publishing every // keystroke sends ~10x more full-buffer packets than anyone consumes. const CODE_PUBLISH_DEBOUNCE_MS = 300; -const TOKEN_REQUEST_TIMEOUT_MS = 10_000; +// Bounds startup through the response headers. The server may finish claim and +// dispatch after the browser has fallen back to offline practice. +const TOKEN_REQUEST_TIMEOUT_MS = 20_000; /// Long enough to read twice, short enough that it is gone before the answer /// it is about. Measured against the hint text, not chosen round. const FRAMEWORK_HINT_MS = 12000;