From 8fb27b39efbbb009fdb6cd811ebfa53066e52c7f Mon Sep 17 00:00:00 2001 From: Simon Klee Date: Fri, 6 Nov 2026 15:27:01 +0100 Subject: [PATCH] fix(cli-rs): find a crashed service between event stream attempts The event stream found a restarted service inside its 2 s connect timeout, which cancelled ensure before the 5 s spawn delay let it start a replacement, so a crashed managed service was never replaced. The transport now exposes reconnect, Follow implements it, and the store's event source runs it between attempts as TS createClientConnection does. The stream's own request no longer resolves the service inside the timeout. --- packages/cli-rs/Cargo.lock | 3 + packages/cli-rs/crates/api/src/transport.rs | 7 + packages/cli-rs/crates/service/src/follow.rs | 28 +++- packages/cli-rs/crates/store/Cargo.toml | 3 + .../cli-rs/crates/store/src/connection.rs | 5 + .../cli-rs/crates/store/tests/reconnect.rs | 141 ++++++++++++++++++ 6 files changed, 179 insertions(+), 8 deletions(-) create mode 100644 packages/cli-rs/crates/store/tests/reconnect.rs diff --git a/packages/cli-rs/Cargo.lock b/packages/cli-rs/Cargo.lock index a5bfc0477e6..eec05c5949d 100644 --- a/packages/cli-rs/Cargo.lock +++ b/packages/cli-rs/Cargo.lock @@ -1167,8 +1167,11 @@ dependencies = [ "indexmap", "log", "opencode-api", + "opencode-local", + "opencode-service", "serde", "serde_json", + "tempfile", "tokio", "tokio-util", ] diff --git a/packages/cli-rs/crates/api/src/transport.rs b/packages/cli-rs/crates/api/src/transport.rs index 0a6c532e141..f5d9755c053 100644 --- a/packages/cli-rs/crates/api/src/transport.rs +++ b/packages/cli-rs/crates/api/src/transport.rs @@ -77,6 +77,13 @@ pub trait Transport: Send + Sync { fn endpoint(&self, url: url::Url, auth: Option<&Auth>) -> (url::Url, Option) { (self.route(url), auth.cloned()) } + + /// Finds the server again after the event stream disconnected (the TS `service.reconnect`). Returns true when + /// a server was resolved. The connection loop runs this between stream attempts, outside the connect timeout, + /// so a managed service has time to start a replacement. + fn reconnect(&self) -> BoxFuture<'_, bool> { + Box::pin(async { false }) + } } /// The checked response of a server-sent events endpoint: status and content type are verified, and diff --git a/packages/cli-rs/crates/service/src/follow.rs b/packages/cli-rs/crates/service/src/follow.rs index 974b945c243..58daab2184f 100644 --- a/packages/cli-rs/crates/service/src/follow.rs +++ b/packages/cli-rs/crates/service/src/follow.rs @@ -7,7 +7,8 @@ //! origin the client was built for go to the current service with its current password. A request //! that fails to connect or is refused with 401 finds the service again, through //! [`connect::reconnect`], before it returns its error. The request is not retried, so a request -//! that may have reached the old service is never sent twice; the event stream retries itself. +//! that may have reached the old service is never sent twice. The event stream retries itself and +//! finds the service through [`Transport::reconnect`], between attempts. use std::sync::{Arc, RwLock}; @@ -67,16 +68,26 @@ impl Transport for Follow { if request.url.origin() != self.origin { return self.inner.send(request).await; } + // The event stream opens under the connection loop's connect timeout, which would cancel a + // service start; the loop finds the service through `Transport::reconnect` instead. + let stream = request.url.path() == "/api/event"; let generation = self.reroute(&mut request); let result = self.inner.send(request).await; // A transport error or 401. - if !result.as_ref().is_ok_and(|response| response.status != 401) { - self.reconnect(generation).await; + if !stream && !result.as_ref().is_ok_and(|response| response.status != 401) { + self.resolve(generation).await; } result }) } + fn reconnect(&self) -> BoxFuture<'_, bool> { + Box::pin(async move { + let generation = self.route.read().expect("route lock").generation; + self.resolve(generation).await + }) + } + fn route(&self, url: url::Url) -> url::Url { self.endpoint(url, None).0 } @@ -122,26 +133,27 @@ impl Follow { route.generation } - // Finds the service again, unless another request did since `generation`. - async fn reconnect(&self, generation: u64) { + // Finds the service again, unless another request did since `generation`. True when it is found. + async fn resolve(&self, generation: u64) -> bool { let _resolving = self.resolving.lock().await; if self.route.read().expect("route lock").generation != generation { - return; + return true; } let found = connect::reconnect(self.inner.clone(), &self.paths, &self.env).await; let endpoint = match found { Ok(found) => found.service.endpoint, Err(error) => { log::warn!("service.reconnect failed: {error}"); - return; + return false; } }; let Ok(url) = url::Url::parse(&endpoint.url) else { log::warn!("service.reconnect found an invalid url: {}", endpoint.url); - return; + return false; }; log::info!("service.reconnect url={url}"); *self.route.write().expect("route lock") = Route { url, auth: endpoint.auth, generation: generation + 1 }; + true } } diff --git a/packages/cli-rs/crates/store/Cargo.toml b/packages/cli-rs/crates/store/Cargo.toml index 4d690f28535..675d7ddc0f7 100644 --- a/packages/cli-rs/crates/store/Cargo.toml +++ b/packages/cli-rs/crates/store/Cargo.toml @@ -20,5 +20,8 @@ tokio-util.workspace = true [dev-dependencies] bytes.workspace = true +opencode-local.workspace = true +opencode-service.workspace = true serde.workspace = true +tempfile.workspace = true tokio = { workspace = true, features = ["test-util"] } diff --git a/packages/cli-rs/crates/store/src/connection.rs b/packages/cli-rs/crates/store/src/connection.rs index 5fdca4ea09f..30b92d447ad 100644 --- a/packages/cli-rs/crates/store/src/connection.rs +++ b/packages/cli-rs/crates/store/src/connection.rs @@ -8,6 +8,7 @@ use futures_util::stream::BoxStream; use opencode_api::client::Client; use opencode_api::events::{self, Activity, Connection, Source, Status}; use opencode_api::generated::*; +use opencode_api::transport::BoxFuture; use tokio::sync::{mpsc, watch}; use tokio::task::JoinHandle; @@ -381,6 +382,10 @@ impl Source for Events { }) .boxed() } + + fn reconnect(&mut self) -> BoxFuture<'_, bool> { + self.client.transport().reconnect() + } } /// Connects the store to its client's event stream on the store's runtime, so the UI thread can diff --git a/packages/cli-rs/crates/store/tests/reconnect.rs b/packages/cli-rs/crates/store/tests/reconnect.rs new file mode 100644 index 00000000000..4e1d0e25702 --- /dev/null +++ b/packages/cli-rs/crates/store/tests/reconnect.rs @@ -0,0 +1,141 @@ +//! The event stream of a managed client recovers when its service crashes (review carbon-1). The crashed +//! service leaves its registration behind, so `ensure` waits the spawn delay (5 s) before it starts a +//! replacement: longer than the connect timeout (2 s) a stream attempt may take. The connection loop +//! must find the service between attempts, as TS `createClientConnection` runs `reconnect`. + +use std::ffi::OsString; +use std::path::PathBuf; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use bytes::Bytes; +use futures_util::StreamExt; +use opencode_api::events::{self, Status}; +use opencode_api::transport::{BoxFuture, Request, Response, Transport}; +use opencode_local::paths::Paths; +use opencode_local::registration; +use opencode_service::connect::{self, Env}; +use opencode_service::follow; +use opencode_service::ts_command::write_script; +use opencode_store::connection::Events; +use tokio::sync::watch; + +const OLD: u16 = 4096; +const NEW: u16 = 5000; + +/// The old service on port 4096 until it dies, and its replacement on port 5000 once the contender +/// registered it. +struct Services { + registration: PathBuf, + alive: watch::Sender, + streams: Mutex>, +} + +impl Transport for Services { + fn send(&self, request: Request) -> BoxFuture<'_, Result> { + let port = request.url.port().unwrap_or_default(); + let path = request.url.path().to_owned(); + Box::pin(async move { + let pid = match port { + OLD if *self.alive.borrow() => 9, + NEW => match registration::read(&self.registration) { + Some(info) if info.url.ends_with(":5000") => info.pid, + _ => return Err("connection refused".to_owned()), + }, + _ => return Err("connection refused".to_owned()), + }; + match path.as_str() { + "/api/info" => Ok(reply( + 200, + "application/json", + format!(r#"{{"version":"2.0.0","pid":{pid},"urls":[],"paths":{{"tmp":"/tmp"}}}}"#), + futures_util::stream::empty().boxed(), + )), + "/api/event" => { + self.streams.lock().unwrap().push(port); + let mut alive = self.alive.subscribe(); + let tail = match port { + OLD => futures_util::stream::once(async move { + let _ = alive.wait_for(|alive| !alive).await; + Err("connection reset".to_owned()) + }) + .boxed(), + _ => futures_util::stream::pending().boxed(), + }; + Ok(reply( + 200, + "text/event-stream", + "data: {\"id\":\"evt_1\",\"type\":\"server.connected\",\"data\":{}}\n\n".to_owned(), + tail, + )) + } + _ => Ok(reply(404, "application/json", "{}".to_owned(), futures_util::stream::empty().boxed())), + } + }) + } +} + +fn reply( + status: u16, + content_type: &str, + body: String, + tail: futures_util::stream::BoxStream<'static, Result>, +) -> Response { + Response { + status, + headers: vec![("content-type".to_owned(), content_type.to_owned())], + body: Box::pin(futures_util::stream::iter([Ok(Bytes::from(body))]).chain(tail)), + } +} + +#[tokio::test(flavor = "multi_thread")] +async fn the_event_stream_follows_a_service_that_crashed_and_was_replaced() { + let dir = tempfile::tempdir().unwrap(); + let paths = Paths::from_vars(|name| match name { + "HOME" => Some(OsString::from(dir.path())), + "XDG_STATE_HOME" => Some(dir.path().join("state").into()), + "XDG_CONFIG_HOME" => Some(dir.path().join("config").into()), + _ => None, + }) + .unwrap(); + std::fs::create_dir_all(&paths.state).unwrap(); + let file = paths.state.join("service-oc-x.json"); + std::fs::write(&file, format!(r#"{{"url":"http://127.0.0.1:{OLD}","pid":9,"password":"old"}}"#)).unwrap(); + // The contender registers the replacement under its own pid. + let program = dir.path().join("opencode-ts"); + write_script( + &program, + &format!( + "#!/bin/sh\nprintf '{{\"url\":\"http://127.0.0.1:{NEW}\",\"pid\":%s,\"password\":\"new\"}}' $$ > {0}.tmp\nmv {0}.tmp {0}\nexec sleep 30\n", + file.display() + ), + ); + let services = Arc::new(Services { + registration: file.clone(), + alive: watch::channel(true).0, + streams: Mutex::new(Vec::new()), + }); + let env = Env { + ts_bin: Some(program.to_string_lossy().into_owned()), + channel: Some("oc-x".to_owned()), + ..Env::default() + }; + let connection = connect::resolve(services.clone(), &paths, &env, &connect::Options::default()).await.unwrap(); + let client = follow::client(services.clone(), &connection, &paths, &env).unwrap(); + let (connection, _batches) = events::connect(Events { client }, events::Options::default()); + let mut state = connection.watch(); + state.wait_for(|state| state.status == Status::Connected).await.unwrap(); + + services.alive.send_replace(false); + state.wait_for(|state| state.status != Status::Connected).await.unwrap(); + let started = Instant::now(); + let reconnected = + tokio::time::timeout(Duration::from_secs(15), state.wait_for(|state| state.status == Status::Connected)).await; + let replacement = registration::read(&file).filter(|info| info.pid != 9); + if let Some(info) = &replacement { + let _ = std::process::Command::new("kill").arg(info.pid.to_string()).status(); + } + assert!(reconnected.is_ok(), "no reconnect after {:?}: {:?}", started.elapsed(), connection.history()); + assert!(replacement.is_some(), "no replacement service was started"); + assert_eq!(services.streams.lock().unwrap().last(), Some(&NEW)); +}