From 2bbdc6e1270460fb93817eaa7b647fd6c2b7b48b Mon Sep 17 00:00:00 2001 From: Vlad Panfilov Date: Tue, 20 Oct 2026 14:48:41 -0700 Subject: [PATCH 1/3] service: rebind the local listener instead of killing the whole run loop Service::run()'s doc comment says it "never completes (unless it errors)", but its local_server.accept() arm propagated any accept error with `?`, which does exactly that: kills the entire service - all local AND remote connections, repository expiration, everything - not just local ones. The local listener's socket can be invalidated by the OS independently of anything the service did (e.g. iOS tearing down a backgrounded app's sockets), and nothing ever recreated it, so once that happened the backend was permanently dead until the app was restarted: the Dart client's own reconnect-with-backoff kept retrying forever against a service that had already stopped listening, which is what presented as file previews and network status hanging indefinitely after a background/foreground cycle. Rebind on an accept error instead, with capped exponential backoff. local_server becomes Option (None only transiently during a rebind) so the old (dead) listener is actually dropped before attempting to bind the new one - binding to the same port while the old listener value is still alive fails with AddrInUse even when its socket is already unusable at the OS level, which the included regression test caught. --- service/src/lib.rs | 91 +++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 85 insertions(+), 6 deletions(-) diff --git a/service/src/lib.rs b/service/src/lib.rs index 7152ec394..dd3bc1019 100644 --- a/service/src/lib.rs +++ b/service/src/lib.rs @@ -60,7 +60,9 @@ const LOCAL_ENDPOINT_KEY: ConfigKey = ConfigKey::new("local_endpo pub struct Service { state: Arc, - local_server: LocalServer, + // `None` only transiently, while rebinding after the old listener died - see + // `rebind_local_server`. + local_server: Option, connections: FuturesUnordered + Send>>>, closed: watch::Sender, } @@ -101,7 +103,7 @@ impl Service { Ok(Self { state, - local_server, + local_server: Some(local_server), connections: FuturesUnordered::new(), closed: watch::Sender::new(false), }) @@ -119,9 +121,23 @@ impl Service { loop { select! { - result = self.local_server.accept() => { - let conn = result.map_err(Error::Accept)?; - self.start_connection(AcceptedConnection::Local(conn)); + result = self.local_server.as_ref().expect( + "local_server is only ever None transiently inside rebind_local_server, \ + which this branch's body awaits to completion before looping back here" + ).accept() => { + match result { + Ok(conn) => self.start_connection(AcceptedConnection::Local(conn)), + // The local listener's socket can be invalidated by the OS independently + // of anything we did (e.g. iOS tearing down a backgrounded app's sockets), + // and nothing else will ever recreate it - propagating this with `?` would + // permanently kill the entire service (all local AND remote connections, + // repository expiration, everything), not just local connections, since + // this whole function "never completes (unless it errors)". Rebind instead. + Err(error) => { + tracing::warn!(?error, "local server accept failed, rebinding"); + self.rebind_local_server().await; + } + } } result = self.state.remote_accept() => { let conn = result.map_err(Error::Accept)?; @@ -135,6 +151,38 @@ impl Service { } } + /// Rebinds the local server listener to the same endpoint it was previously bound to. + /// Retries with capped exponential backoff instead of giving up, since giving up here would + /// mean no local client (i.e. the app's own UI) could ever talk to this service again until + /// restarted. + async fn rebind_local_server(&mut self) { + let endpoint = *self.local_server.as_ref().unwrap().endpoint(); + + // Drop the old listener *before* trying to bind the new one: even though its socket may + // already be invalid at the OS level (that's exactly why we're here), our process still + // holds a reference to it until it's dropped, which is enough to make the OS refuse to + // bind a new socket to the same port (`AddrInUse`). + self.local_server = None; + + let mut delay = Duration::from_millis(50); + const MAX_DELAY: Duration = Duration::from_secs(5); + + loop { + match LocalServer::bind(endpoint).await { + Ok(server) => { + tracing::info!(port = server.endpoint().port, "local server rebound"); + self.local_server = Some(server); + return; + } + Err(error) => { + tracing::warn!(?error, ?delay, "local server rebind failed, retrying"); + time::sleep(delay).await; + delay = (delay * 2).min(MAX_DELAY); + } + } + } + } + pub async fn close(&mut self) { // Signal connections to close themselves self.closed.send(true).ok(); @@ -148,7 +196,7 @@ impl Service { } pub fn local_endpoint(&self) -> &LocalEndpoint { - self.local_server.endpoint() + self.local_server.as_ref().unwrap().endpoint() } pub fn store_dirs(&self) -> Vec { @@ -254,6 +302,37 @@ mod tests { service0.close().await; } + // Regression test for the local listener's socket being invalidated independently of + // anything the service did (e.g. iOS tearing down a backgrounded app's sockets): the service + // must rebind and keep accepting new local connections on the same port, instead of its + // whole run loop dying because one `accept()` call failed. + #[tokio::test] + async fn rebind_local_server() { + let temp_dir = TempDir::new().unwrap(); + let mut service = Service::init(temp_dir.path().join("config")).await.unwrap(); + let port = service.local_endpoint().port; + + // Exercise the same rebind path `run()` takes on an accept error, without needing to + // simulate the OS invalidating the fd (which is OS/platform specific) - this still + // covers the tricky part: binding to the *same* port while the old (still live, not yet + // dropped) listener technically still holds it. + time::timeout(Duration::from_secs(5), service.rebind_local_server()) + .await + .expect("rebind_local_server timed out"); + + assert_eq!(service.local_endpoint().port, port); + + let runner = ServiceRunner::start(service); + + // A fresh connection to the same port must still be accepted after rebinding. + tokio::net::TcpStream::connect((Ipv4Addr::LOCALHOST, port)) + .await + .expect("failed to connect after rebind"); + + let mut service = runner.stop().await; + service.close().await; + } + #[tokio::test] async fn tls_cert_renewal() { async fn gen_cert(expires_in: Duration) -> (Certificate, KeyPair) { From ba856cafb74d76ff00b24c68a52e01e431d57ecf Mon Sep 17 00:00:00 2001 From: Vlad Panfilov Date: Thu, 29 Oct 2026 16:47:16 -0700 Subject: [PATCH 2/3] service: fix run() panic when restarted after a cancelled rebind run() is documented as safe to cancel and restart, but rebind_local_server left `local_server` as None across its backoff awaits. Since the Service outlives the run() future, a cancel during a rebind left the Service with no listener, and the next run() panicked on the `expect`/`unwrap`. Store the resolved local endpoint in its own field and rebind at the top of run()'s loop whenever `local_server` is None (the accept-error arm now just drops the dead listener). A restart that finds no listener rebinds instead of panicking, and local_endpoint() no longer unwraps an Option. Adds a regression test that starts run() with a missing listener. --- service/src/lib.rs | 68 ++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 56 insertions(+), 12 deletions(-) diff --git a/service/src/lib.rs b/service/src/lib.rs index dd3bc1019..88cc6bfe5 100644 --- a/service/src/lib.rs +++ b/service/src/lib.rs @@ -60,9 +60,12 @@ const LOCAL_ENDPOINT_KEY: ConfigKey = ConfigKey::new("local_endpo pub struct Service { state: Arc, - // `None` only transiently, while rebinding after the old listener died - see - // `rebind_local_server`. + // `None` only transiently, between the local listener dying and `run()` rebinding it at the + // top of its loop - see `run`/`rebind_local_server`. local_server: Option, + // The (resolved) endpoint the local listener binds to. Kept separately so we can rebind to it + // even while `local_server` is `None`, and so `local_endpoint()` never has to unwrap. + local_endpoint: LocalEndpoint, connections: FuturesUnordered + Send>>>, closed: watch::Sender, } @@ -93,6 +96,8 @@ impl Service { local_endpoint_entry.set(local_server.endpoint()).await?; + let resolved_endpoint = *local_server.endpoint(); + tracing::debug!( port = local_server.endpoint().port, "local server listening" @@ -104,6 +109,7 @@ impl Service { Ok(Self { state, local_server: Some(local_server), + local_endpoint: resolved_endpoint, connections: FuturesUnordered::new(), closed: watch::Sender::new(false), }) @@ -120,11 +126,17 @@ impl Service { repo_expiration_interval.set_missed_tick_behavior(MissedTickBehavior::Delay); loop { + // If the local listener is gone - the accept arm below dropped it after a fatal error, + // or a previous `run()` was cancelled mid-rebind - rebind it before polling it. Doing + // this here rather than inline in the accept arm is what keeps `run()` safe to cancel + // and restart: a restart that finds `local_server` still `None` recovers here instead + // of panicking on an `unwrap`. + if self.local_server.is_none() { + self.rebind_local_server().await; + } + select! { - result = self.local_server.as_ref().expect( - "local_server is only ever None transiently inside rebind_local_server, \ - which this branch's body awaits to completion before looping back here" - ).accept() => { + result = self.local_server.as_ref().unwrap().accept() => { match result { Ok(conn) => self.start_connection(AcceptedConnection::Local(conn)), // The local listener's socket can be invalidated by the OS independently @@ -132,10 +144,11 @@ impl Service { // and nothing else will ever recreate it - propagating this with `?` would // permanently kill the entire service (all local AND remote connections, // repository expiration, everything), not just local connections, since - // this whole function "never completes (unless it errors)". Rebind instead. + // this whole function "never completes (unless it errors)". Drop the dead + // listener so the top of the loop rebinds it. Err(error) => { tracing::warn!(?error, "local server accept failed, rebinding"); - self.rebind_local_server().await; + self.local_server = None; } } } @@ -156,8 +169,6 @@ impl Service { /// mean no local client (i.e. the app's own UI) could ever talk to this service again until /// restarted. async fn rebind_local_server(&mut self) { - let endpoint = *self.local_server.as_ref().unwrap().endpoint(); - // Drop the old listener *before* trying to bind the new one: even though its socket may // already be invalid at the OS level (that's exactly why we're here), our process still // holds a reference to it until it's dropped, which is enough to make the OS refuse to @@ -168,7 +179,7 @@ impl Service { const MAX_DELAY: Duration = Duration::from_secs(5); loop { - match LocalServer::bind(endpoint).await { + match LocalServer::bind(self.local_endpoint).await { Ok(server) => { tracing::info!(port = server.endpoint().port, "local server rebound"); self.local_server = Some(server); @@ -196,7 +207,7 @@ impl Service { } pub fn local_endpoint(&self) -> &LocalEndpoint { - self.local_server.as_ref().unwrap().endpoint() + &self.local_endpoint } pub fn store_dirs(&self) -> Vec { @@ -333,6 +344,39 @@ mod tests { service.close().await; } + // Regression test for `run()` being cancelled while `rebind_local_server` had already dropped + // the old listener (so the `Service` is left with no local listener), then restarted: `run()` + // must rebind on entry rather than panic on a missing listener. + #[tokio::test] + async fn run_rebinds_missing_local_server_on_start() { + let temp_dir = TempDir::new().unwrap(); + let mut service = Service::init(temp_dir.path().join("config")).await.unwrap(); + let port = service.local_endpoint().port; + + // Simulate the state left behind by a cancel during rebind: the listener is gone. + service.local_server = None; + + let runner = ServiceRunner::start(service); + + // The rebind happens inside `run()` right after start, so retry briefly until the listener + // comes back up on the same port. + let connect = async { + loop { + match tokio::net::TcpStream::connect((Ipv4Addr::LOCALHOST, port)).await { + Ok(stream) => break stream, + Err(_) => time::sleep(Duration::from_millis(10)).await, + } + } + }; + time::timeout(Duration::from_secs(5), connect) + .await + .expect("listener never came back up after restart with a missing listener"); + + let mut service = runner.stop().await; + assert_eq!(service.local_endpoint().port, port); + service.close().await; + } + #[tokio::test] async fn tls_cert_renewal() { async fn gen_cert(expires_in: Duration) -> (Certificate, KeyPair) { From c2ecaa23cd407644ec7d2b582e249c96d643566b Mon Sep 17 00:00:00 2001 From: Vlad Panfilov Date: Thu, 29 Oct 2026 17:13:50 -0700 Subject: [PATCH 3/3] service: document that inline local-server rebind pauses run()'s loop A reviewer noted rebind_local_server is awaited inline, so while it retries the run() select! is paused and remote accepts, connection cleanup and repo expiration don't progress. Rebinds succeed fast in practice and a persistent failure means the local listener is unusable anyway, so keep it inline for now; record the trade-off and how to make it concurrent if needed. --- service/src/lib.rs | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/service/src/lib.rs b/service/src/lib.rs index 88cc6bfe5..6862c4fe1 100644 --- a/service/src/lib.rs +++ b/service/src/lib.rs @@ -131,6 +131,13 @@ impl Service { // this here rather than inline in the accept arm is what keeps `run()` safe to cancel // and restart: a restart that finds `local_server` still `None` recovers here instead // of panicking on an `unwrap`. + // + // Known limitation: this rebind is awaited inline, so while it retries (capped backoff) + // the `select!` below is paused - remote accepts, connection cleanup and repository + // expiration don't make progress until it succeeds. That's acceptable because a rebind + // normally succeeds on the first try within milliseconds; a rebind that keeps failing + // means the local listener is unusable anyway. If that ever needs to run concurrently, + // drive the rebind as its own `select!` arm instead. if self.local_server.is_none() { self.rebind_local_server().await; }