diff --git a/crates/test-support/tests/streaming_and_cache.rs b/crates/test-support/tests/streaming_and_cache.rs index dd286e5a..3a5e7e64 100644 --- a/crates/test-support/tests/streaming_and_cache.rs +++ b/crates/test-support/tests/streaming_and_cache.rs @@ -227,11 +227,12 @@ async fn streaming_client_disconnect_emits_cancelled_gateway_log() { // // Recipe: an upstream that holds the response open (long initial // delay) so the gateway's SSE body stream is parked waiting on - // the first chunk when the client times out. + // the first chunk when the client goes. let app = TestApp::spawn_with_clickhouse().await; let user = fixtures::create_random_user(&app.db).await.unwrap(); - // Slow upstream — 5s delay before any chunk; we'll drop after ~150ms. + // An upstream that does not answer within the test: the stream is + // still waiting on it when the client leaves. let server = MockServer::start().await; Mock::given(method("POST")) .and(path("/v1/chat/completions")) @@ -241,7 +242,7 @@ async fn streaming_client_disconnect_emits_cancelled_gateway_log() { b"data: {\"id\":\"x\",\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\"}}]}\n\n", "text/event-stream", ) - .set_delay(std::time::Duration::from_secs(5)), + .set_delay(std::time::Duration::from_secs(60)), ) .mount(&server) .await; @@ -270,15 +271,17 @@ async fn streaming_client_disconnect_emits_cancelled_gateway_log() { .await .unwrap(); - // Raw reqwest with an aggressive total-request timeout — when it - // fires, the in-flight request future is dropped, which closes - // the gateway-side TCP connection. The gateway's SSE body future - // is dropped, which drops `done_tx`, which the spawned on_done - // task picks up as ClientCancelled. - let client = reqwest::Client::builder() - .timeout(std::time::Duration::from_millis(150)) - .build() - .unwrap(); + // The client leaves once the stream has started: the response + // headers are back (the gateway sends them before calling the + // upstream), and dropping the response closes the connection. The + // gateway's SSE body future is dropped, which drops `done_tx`, which + // the spawned tail picks up as ClientCancelled. + // + // This used to be a 150 ms client timeout, which raced the handler: + // when auth, limits and routing took longer than that under load, the + // client left before the stream existed, hyper dropped the handler, + // and no row was ever written. + let client = reqwest::Client::new(); let url = format!("{}/v1/chat/completions", app.gateway_url); let body = serde_json::json!({ "model": "cancel-stream", @@ -286,15 +289,15 @@ async fn streaming_client_disconnect_emits_cancelled_gateway_log() { "stream": true, "temperature": 0.7 }); - let result = client + let resp = client .post(&url) .bearer_auth(&key.plaintext) .json(&body) .send() - .await; - // Either we time out (expected) or we got a partial response that - // we now drop. Both end with the gateway seeing a disconnect. - drop(result); + .await + .expect("the stream starts"); + assert_eq!(resp.status(), 200); + drop(resp); // Wait for the cancelled row to land in ClickHouse. The audit // pipeline batches with a small flush interval; give it a few