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)); +}