1#[cfg(any(test, feature = "test-support"))]
2pub mod test;
3
4use anyhow::{anyhow, Context, Result};
5use async_recursion::async_recursion;
6use async_tungstenite::tungstenite::{
7 error::Error as WebsocketError,
8 http::{Request, StatusCode},
9};
10use gpui::{AsyncAppContext, Entity, ModelContext, Task};
11use lazy_static::lazy_static;
12use parking_lot::RwLock;
13use postage::{prelude::Stream, watch};
14use rand::prelude::*;
15use std::{
16 any::TypeId,
17 collections::HashMap,
18 convert::TryFrom,
19 fmt::Write as _,
20 future::Future,
21 sync::{Arc, Weak},
22 time::{Duration, Instant},
23};
24use surf::Url;
25use thiserror::Error;
26use util::ResultExt;
27pub use zrpc::{proto, ConnectionId, PeerId, TypedEnvelope};
28use zrpc::{
29 proto::{AnyTypedEnvelope, EntityMessage, EnvelopedMessage, RequestMessage},
30 Connection, Peer, Receipt,
31};
32
33lazy_static! {
34 static ref ZED_SERVER_URL: String =
35 std::env::var("ZED_SERVER_URL").unwrap_or("https://zed.dev:443".to_string());
36 static ref IMPERSONATE_LOGIN: Option<String> = std::env::var("ZED_IMPERSONATE")
37 .ok()
38 .and_then(|s| if s.is_empty() { None } else { Some(s) });
39}
40
41pub struct Client {
42 peer: Arc<Peer>,
43 state: RwLock<ClientState>,
44 authenticate:
45 Option<Box<dyn 'static + Send + Sync + Fn(&AsyncAppContext) -> Task<Result<Credentials>>>>,
46 establish_connection: Option<
47 Box<
48 dyn 'static
49 + Send
50 + Sync
51 + Fn(
52 &Credentials,
53 &AsyncAppContext,
54 ) -> Task<Result<Connection, EstablishConnectionError>>,
55 >,
56 >,
57}
58
59#[derive(Error, Debug)]
60pub enum EstablishConnectionError {
61 #[error("upgrade required")]
62 UpgradeRequired,
63 #[error("unauthorized")]
64 Unauthorized,
65 #[error("{0}")]
66 Other(#[from] anyhow::Error),
67 #[error("{0}")]
68 Io(#[from] std::io::Error),
69 #[error("{0}")]
70 Http(#[from] async_tungstenite::tungstenite::http::Error),
71}
72
73impl From<WebsocketError> for EstablishConnectionError {
74 fn from(error: WebsocketError) -> Self {
75 if let WebsocketError::Http(response) = &error {
76 match response.status() {
77 StatusCode::UNAUTHORIZED => return EstablishConnectionError::Unauthorized,
78 StatusCode::UPGRADE_REQUIRED => return EstablishConnectionError::UpgradeRequired,
79 _ => {}
80 }
81 }
82 EstablishConnectionError::Other(error.into())
83 }
84}
85
86impl EstablishConnectionError {
87 pub fn other(error: impl Into<anyhow::Error> + Send + Sync) -> Self {
88 Self::Other(error.into())
89 }
90}
91
92#[derive(Copy, Clone, Debug)]
93pub enum Status {
94 SignedOut,
95 UpgradeRequired,
96 Authenticating,
97 Connecting,
98 ConnectionError,
99 Connected { connection_id: ConnectionId },
100 ConnectionLost,
101 Reauthenticating,
102 Reconnecting,
103 ReconnectionError { next_reconnection: Instant },
104}
105
106struct ClientState {
107 credentials: Option<Credentials>,
108 status: (watch::Sender<Status>, watch::Receiver<Status>),
109 entity_id_extractors: HashMap<TypeId, Box<dyn Send + Sync + Fn(&dyn AnyTypedEnvelope) -> u64>>,
110 model_handlers: HashMap<
111 (TypeId, u64),
112 Box<dyn Send + Sync + FnMut(Box<dyn AnyTypedEnvelope>, &mut AsyncAppContext)>,
113 >,
114 _maintain_connection: Option<Task<()>>,
115 heartbeat_interval: Duration,
116}
117
118#[derive(Clone)]
119pub struct Credentials {
120 pub user_id: u64,
121 pub access_token: String,
122}
123
124impl Default for ClientState {
125 fn default() -> Self {
126 Self {
127 credentials: None,
128 status: watch::channel_with(Status::SignedOut),
129 entity_id_extractors: Default::default(),
130 model_handlers: Default::default(),
131 _maintain_connection: None,
132 heartbeat_interval: Duration::from_secs(5),
133 }
134 }
135}
136
137pub struct Subscription {
138 client: Weak<Client>,
139 id: (TypeId, u64),
140}
141
142impl Drop for Subscription {
143 fn drop(&mut self) {
144 if let Some(client) = self.client.upgrade() {
145 drop(
146 client
147 .state
148 .write()
149 .model_handlers
150 .remove(&self.id)
151 .unwrap(),
152 );
153 }
154 }
155}
156
157impl Client {
158 pub fn new() -> Arc<Self> {
159 Arc::new(Self {
160 peer: Peer::new(),
161 state: Default::default(),
162 authenticate: None,
163 establish_connection: None,
164 })
165 }
166
167 #[cfg(any(test, feature = "test-support"))]
168 pub fn override_authenticate<F>(&mut self, authenticate: F) -> &mut Self
169 where
170 F: 'static + Send + Sync + Fn(&AsyncAppContext) -> Task<Result<Credentials>>,
171 {
172 self.authenticate = Some(Box::new(authenticate));
173 self
174 }
175
176 #[cfg(any(test, feature = "test-support"))]
177 pub fn override_establish_connection<F>(&mut self, connect: F) -> &mut Self
178 where
179 F: 'static
180 + Send
181 + Sync
182 + Fn(&Credentials, &AsyncAppContext) -> Task<Result<Connection, EstablishConnectionError>>,
183 {
184 self.establish_connection = Some(Box::new(connect));
185 self
186 }
187
188 pub fn user_id(&self) -> Option<u64> {
189 self.state
190 .read()
191 .credentials
192 .as_ref()
193 .map(|credentials| credentials.user_id)
194 }
195
196 pub fn status(&self) -> watch::Receiver<Status> {
197 self.state.read().status.1.clone()
198 }
199
200 fn set_status(self: &Arc<Self>, status: Status, cx: &AsyncAppContext) {
201 let mut state = self.state.write();
202 *state.status.0.borrow_mut() = status;
203
204 match status {
205 Status::Connected { .. } => {
206 let heartbeat_interval = state.heartbeat_interval;
207 let this = self.clone();
208 let foreground = cx.foreground();
209 state._maintain_connection = Some(cx.foreground().spawn(async move {
210 loop {
211 foreground.timer(heartbeat_interval).await;
212 let _ = this.request(proto::Ping {}).await;
213 }
214 }));
215 }
216 Status::ConnectionLost => {
217 let this = self.clone();
218 let foreground = cx.foreground();
219 let heartbeat_interval = state.heartbeat_interval;
220 state._maintain_connection = Some(cx.spawn(|cx| async move {
221 let mut rng = StdRng::from_entropy();
222 let mut delay = Duration::from_millis(100);
223 while let Err(error) = this.authenticate_and_connect(&cx).await {
224 log::error!("failed to connect {}", error);
225 this.set_status(
226 Status::ReconnectionError {
227 next_reconnection: Instant::now() + delay,
228 },
229 &cx,
230 );
231 foreground.timer(delay).await;
232 delay = delay
233 .mul_f32(rng.gen_range(1.0..=2.0))
234 .min(heartbeat_interval);
235 }
236 }));
237 }
238 Status::SignedOut | Status::UpgradeRequired => {
239 state._maintain_connection.take();
240 }
241 _ => {}
242 }
243 }
244
245 pub fn subscribe<T, M, F>(
246 self: &Arc<Self>,
247 cx: &mut ModelContext<M>,
248 mut handler: F,
249 ) -> Subscription
250 where
251 T: EnvelopedMessage,
252 M: Entity,
253 F: 'static
254 + Send
255 + Sync
256 + FnMut(&mut M, TypedEnvelope<T>, Arc<Self>, &mut ModelContext<M>) -> Result<()>,
257 {
258 let subscription_id = (TypeId::of::<T>(), Default::default());
259 let client = self.clone();
260 let mut state = self.state.write();
261 let model = cx.handle().downgrade();
262 let prev_extractor = state
263 .entity_id_extractors
264 .insert(subscription_id.0, Box::new(|_| Default::default()));
265 if prev_extractor.is_some() {
266 panic!("registered a handler for the same entity twice")
267 }
268
269 state.model_handlers.insert(
270 subscription_id,
271 Box::new(move |envelope, cx| {
272 if let Some(model) = model.upgrade(cx) {
273 let envelope = envelope.into_any().downcast::<TypedEnvelope<T>>().unwrap();
274 model.update(cx, |model, cx| {
275 if let Err(error) = handler(model, *envelope, client.clone(), cx) {
276 log::error!("error handling message: {}", error)
277 }
278 });
279 }
280 }),
281 );
282
283 Subscription {
284 client: Arc::downgrade(self),
285 id: subscription_id,
286 }
287 }
288
289 pub fn subscribe_to_entity<T, M, F>(
290 self: &Arc<Self>,
291 remote_id: u64,
292 cx: &mut ModelContext<M>,
293 mut handler: F,
294 ) -> Subscription
295 where
296 T: EntityMessage,
297 M: Entity,
298 F: 'static
299 + Send
300 + Sync
301 + FnMut(&mut M, TypedEnvelope<T>, Arc<Self>, &mut ModelContext<M>) -> Result<()>,
302 {
303 let subscription_id = (TypeId::of::<T>(), remote_id);
304 let client = self.clone();
305 let mut state = self.state.write();
306 let model = cx.handle().downgrade();
307 state
308 .entity_id_extractors
309 .entry(subscription_id.0)
310 .or_insert_with(|| {
311 Box::new(|envelope| {
312 let envelope = envelope
313 .as_any()
314 .downcast_ref::<TypedEnvelope<T>>()
315 .unwrap();
316 envelope.payload.remote_entity_id()
317 })
318 });
319 let prev_handler = state.model_handlers.insert(
320 subscription_id,
321 Box::new(move |envelope, cx| {
322 if let Some(model) = model.upgrade(cx) {
323 let envelope = envelope.into_any().downcast::<TypedEnvelope<T>>().unwrap();
324 model.update(cx, |model, cx| {
325 if let Err(error) = handler(model, *envelope, client.clone(), cx) {
326 log::error!("error handling message: {}", error)
327 }
328 });
329 }
330 }),
331 );
332 if prev_handler.is_some() {
333 panic!("registered a handler for the same entity twice")
334 }
335
336 Subscription {
337 client: Arc::downgrade(self),
338 id: subscription_id,
339 }
340 }
341
342 #[async_recursion(?Send)]
343 pub async fn authenticate_and_connect(
344 self: &Arc<Self>,
345 cx: &AsyncAppContext,
346 ) -> anyhow::Result<()> {
347 let was_disconnected = match *self.status().borrow() {
348 Status::SignedOut => true,
349 Status::ConnectionError | Status::ConnectionLost | Status::ReconnectionError { .. } => {
350 false
351 }
352 Status::Connected { .. }
353 | Status::Connecting { .. }
354 | Status::Reconnecting { .. }
355 | Status::Authenticating
356 | Status::Reauthenticating => return Ok(()),
357 Status::UpgradeRequired => return Err(EstablishConnectionError::UpgradeRequired)?,
358 };
359
360 if was_disconnected {
361 self.set_status(Status::Authenticating, cx);
362 } else {
363 self.set_status(Status::Reauthenticating, cx)
364 }
365
366 let mut used_keychain = false;
367 let credentials = self.state.read().credentials.clone();
368 let credentials = if let Some(credentials) = credentials {
369 credentials
370 } else if let Some(credentials) = read_credentials_from_keychain(cx) {
371 used_keychain = true;
372 credentials
373 } else {
374 let credentials = match self.authenticate(&cx).await {
375 Ok(credentials) => credentials,
376 Err(err) => {
377 self.set_status(Status::ConnectionError, cx);
378 return Err(err);
379 }
380 };
381 credentials
382 };
383
384 if was_disconnected {
385 self.set_status(Status::Connecting, cx);
386 } else {
387 self.set_status(Status::Reconnecting, cx);
388 }
389
390 match self.establish_connection(&credentials, cx).await {
391 Ok(conn) => {
392 log::info!("connected to rpc address {}", *ZED_SERVER_URL);
393 self.state.write().credentials = Some(credentials.clone());
394 if !used_keychain && IMPERSONATE_LOGIN.is_none() {
395 write_credentials_to_keychain(&credentials, cx).log_err();
396 }
397 self.set_connection(conn, cx).await;
398 Ok(())
399 }
400 Err(EstablishConnectionError::Unauthorized) => {
401 self.state.write().credentials.take();
402 if used_keychain {
403 cx.platform().delete_credentials(&ZED_SERVER_URL).log_err();
404 self.set_status(Status::SignedOut, cx);
405 self.authenticate_and_connect(cx).await
406 } else {
407 self.set_status(Status::ConnectionError, cx);
408 Err(EstablishConnectionError::Unauthorized)?
409 }
410 }
411 Err(EstablishConnectionError::UpgradeRequired) => {
412 self.set_status(Status::UpgradeRequired, cx);
413 Err(EstablishConnectionError::UpgradeRequired)?
414 }
415 Err(error) => {
416 self.set_status(Status::ConnectionError, cx);
417 Err(error)?
418 }
419 }
420 }
421
422 async fn set_connection(self: &Arc<Self>, conn: Connection, cx: &AsyncAppContext) {
423 let (connection_id, handle_io, mut incoming) = self.peer.add_connection(conn).await;
424 cx.foreground()
425 .spawn({
426 let mut cx = cx.clone();
427 let this = self.clone();
428 async move {
429 while let Some(message) = incoming.recv().await {
430 let mut state = this.state.write();
431 if let Some(extract_entity_id) =
432 state.entity_id_extractors.get(&message.payload_type_id())
433 {
434 let payload_type_id = message.payload_type_id();
435 let entity_id = (extract_entity_id)(message.as_ref());
436 let handler_key = (payload_type_id, entity_id);
437 if let Some(mut handler) = state.model_handlers.remove(&handler_key) {
438 drop(state); // Avoid deadlocks if the handler interacts with rpc::Client
439 let start_time = Instant::now();
440 log::info!("RPC client message {}", message.payload_type_name());
441 (handler)(message, &mut cx);
442 log::info!(
443 "RPC message handled. duration:{:?}",
444 start_time.elapsed()
445 );
446 this.state
447 .write()
448 .model_handlers
449 .insert(handler_key, handler);
450 } else {
451 log::info!("unhandled message {}", message.payload_type_name());
452 }
453 } else {
454 log::info!("unhandled message {}", message.payload_type_name());
455 }
456 }
457 }
458 })
459 .detach();
460
461 self.set_status(Status::Connected { connection_id }, cx);
462
463 let handle_io = cx.background().spawn(handle_io);
464 let this = self.clone();
465 let cx = cx.clone();
466 cx.foreground()
467 .spawn(async move {
468 match handle_io.await {
469 Ok(()) => this.set_status(Status::SignedOut, &cx),
470 Err(err) => {
471 log::error!("connection error: {:?}", err);
472 this.set_status(Status::ConnectionLost, &cx);
473 }
474 }
475 })
476 .detach();
477 }
478
479 fn authenticate(self: &Arc<Self>, cx: &AsyncAppContext) -> Task<Result<Credentials>> {
480 if let Some(callback) = self.authenticate.as_ref() {
481 callback(cx)
482 } else {
483 self.authenticate_with_browser(cx)
484 }
485 }
486
487 fn establish_connection(
488 self: &Arc<Self>,
489 credentials: &Credentials,
490 cx: &AsyncAppContext,
491 ) -> Task<Result<Connection, EstablishConnectionError>> {
492 if let Some(callback) = self.establish_connection.as_ref() {
493 callback(credentials, cx)
494 } else {
495 self.establish_websocket_connection(credentials, cx)
496 }
497 }
498
499 fn establish_websocket_connection(
500 self: &Arc<Self>,
501 credentials: &Credentials,
502 cx: &AsyncAppContext,
503 ) -> Task<Result<Connection, EstablishConnectionError>> {
504 let request = Request::builder()
505 .header(
506 "Authorization",
507 format!("{} {}", credentials.user_id, credentials.access_token),
508 )
509 .header("X-Zed-Protocol-Version", zrpc::PROTOCOL_VERSION);
510 cx.background().spawn(async move {
511 if let Some(host) = ZED_SERVER_URL.strip_prefix("https://") {
512 let stream = smol::net::TcpStream::connect(host).await?;
513 let request = request.uri(format!("wss://{}/rpc", host)).body(())?;
514 let (stream, _) =
515 async_tungstenite::async_tls::client_async_tls(request, stream).await?;
516 Ok(Connection::new(stream))
517 } else if let Some(host) = ZED_SERVER_URL.strip_prefix("http://") {
518 let stream = smol::net::TcpStream::connect(host).await?;
519 let request = request.uri(format!("ws://{}/rpc", host)).body(())?;
520 let (stream, _) = async_tungstenite::client_async(request, stream).await?;
521 Ok(Connection::new(stream))
522 } else {
523 Err(anyhow!("invalid server url: {}", *ZED_SERVER_URL))?
524 }
525 })
526 }
527
528 pub fn authenticate_with_browser(
529 self: &Arc<Self>,
530 cx: &AsyncAppContext,
531 ) -> Task<Result<Credentials>> {
532 let platform = cx.platform();
533 let executor = cx.background();
534 executor.clone().spawn(async move {
535 // Generate a pair of asymmetric encryption keys. The public key will be used by the
536 // zed server to encrypt the user's access token, so that it can'be intercepted by
537 // any other app running on the user's device.
538 let (public_key, private_key) =
539 zrpc::auth::keypair().expect("failed to generate keypair for auth");
540 let public_key_string =
541 String::try_from(public_key).expect("failed to serialize public key for auth");
542
543 // Start an HTTP server to receive the redirect from Zed's sign-in page.
544 let server = tiny_http::Server::http("127.0.0.1:0").expect("failed to find open port");
545 let port = server.server_addr().port();
546
547 // Open the Zed sign-in page in the user's browser, with query parameters that indicate
548 // that the user is signing in from a Zed app running on the same device.
549 let mut url = format!(
550 "{}/sign_in?native_app_port={}&native_app_public_key={}",
551 *ZED_SERVER_URL, port, public_key_string
552 );
553
554 if let Some(impersonate_login) = IMPERSONATE_LOGIN.as_ref() {
555 log::info!("impersonating user @{}", impersonate_login);
556 write!(&mut url, "&impersonate={}", impersonate_login).unwrap();
557 }
558
559 platform.open_url(&url);
560
561 // Receive the HTTP request from the user's browser. Retrieve the user id and encrypted
562 // access token from the query params.
563 //
564 // TODO - Avoid ever starting more than one HTTP server. Maybe switch to using a
565 // custom URL scheme instead of this local HTTP server.
566 let (user_id, access_token) = executor
567 .spawn(async move {
568 if let Some(req) = server.recv_timeout(Duration::from_secs(10 * 60))? {
569 let path = req.url();
570 let mut user_id = None;
571 let mut access_token = None;
572 let url = Url::parse(&format!("http://example.com{}", path))
573 .context("failed to parse login notification url")?;
574 for (key, value) in url.query_pairs() {
575 if key == "access_token" {
576 access_token = Some(value.to_string());
577 } else if key == "user_id" {
578 user_id = Some(value.to_string());
579 }
580 }
581 req.respond(
582 tiny_http::Response::from_string(LOGIN_RESPONSE).with_header(
583 tiny_http::Header::from_bytes("Content-Type", "text/html").unwrap(),
584 ),
585 )
586 .context("failed to respond to login http request")?;
587 Ok((
588 user_id.ok_or_else(|| anyhow!("missing user_id parameter"))?,
589 access_token
590 .ok_or_else(|| anyhow!("missing access_token parameter"))?,
591 ))
592 } else {
593 Err(anyhow!("didn't receive login redirect"))
594 }
595 })
596 .await?;
597
598 let access_token = private_key
599 .decrypt_string(&access_token)
600 .context("failed to decrypt access token")?;
601 platform.activate(true);
602
603 Ok(Credentials {
604 user_id: user_id.parse()?,
605 access_token,
606 })
607 })
608 }
609
610 pub async fn disconnect(self: &Arc<Self>, cx: &AsyncAppContext) -> Result<()> {
611 let conn_id = self.connection_id()?;
612 self.peer.disconnect(conn_id).await;
613 self.set_status(Status::SignedOut, cx);
614 Ok(())
615 }
616
617 fn connection_id(&self) -> Result<ConnectionId> {
618 if let Status::Connected { connection_id, .. } = *self.status().borrow() {
619 Ok(connection_id)
620 } else {
621 Err(anyhow!("not connected"))
622 }
623 }
624
625 pub async fn send<T: EnvelopedMessage>(&self, message: T) -> Result<()> {
626 self.peer.send(self.connection_id()?, message).await
627 }
628
629 pub async fn request<T: RequestMessage>(&self, request: T) -> Result<T::Response> {
630 self.peer.request(self.connection_id()?, request).await
631 }
632
633 pub fn respond<T: RequestMessage>(
634 &self,
635 receipt: Receipt<T>,
636 response: T::Response,
637 ) -> impl Future<Output = Result<()>> {
638 self.peer.respond(receipt, response)
639 }
640}
641
642fn read_credentials_from_keychain(cx: &AsyncAppContext) -> Option<Credentials> {
643 if IMPERSONATE_LOGIN.is_some() {
644 return None;
645 }
646
647 let (user_id, access_token) = cx
648 .platform()
649 .read_credentials(&ZED_SERVER_URL)
650 .log_err()
651 .flatten()?;
652 Some(Credentials {
653 user_id: user_id.parse().ok()?,
654 access_token: String::from_utf8(access_token).ok()?,
655 })
656}
657
658fn write_credentials_to_keychain(credentials: &Credentials, cx: &AsyncAppContext) -> Result<()> {
659 cx.platform().write_credentials(
660 &ZED_SERVER_URL,
661 &credentials.user_id.to_string(),
662 credentials.access_token.as_bytes(),
663 )
664}
665
666const WORKTREE_URL_PREFIX: &'static str = "zed://worktrees/";
667
668pub fn encode_worktree_url(id: u64, access_token: &str) -> String {
669 format!("{}{}/{}", WORKTREE_URL_PREFIX, id, access_token)
670}
671
672pub fn decode_worktree_url(url: &str) -> Option<(u64, String)> {
673 let path = url.trim().strip_prefix(WORKTREE_URL_PREFIX)?;
674 let mut parts = path.split('/');
675 let id = parts.next()?.parse::<u64>().ok()?;
676 let access_token = parts.next()?;
677 if access_token.is_empty() {
678 return None;
679 }
680 Some((id, access_token.to_string()))
681}
682
683const LOGIN_RESPONSE: &'static str = "
684<!DOCTYPE html>
685<html>
686<script>window.close();</script>
687</html>
688";
689
690#[cfg(test)]
691mod tests {
692 use super::*;
693 use crate::test::FakeServer;
694 use gpui::TestAppContext;
695
696 #[gpui::test(iterations = 10)]
697 async fn test_heartbeat(cx: TestAppContext) {
698 cx.foreground().forbid_parking();
699
700 let user_id = 5;
701 let mut client = Client::new();
702 let server = FakeServer::for_client(user_id, &mut client, &cx).await;
703
704 cx.foreground().advance_clock(Duration::from_secs(10));
705 let ping = server.receive::<proto::Ping>().await.unwrap();
706 server.respond(ping.receipt(), proto::Ack {}).await;
707
708 cx.foreground().advance_clock(Duration::from_secs(10));
709 let ping = server.receive::<proto::Ping>().await.unwrap();
710 server.respond(ping.receipt(), proto::Ack {}).await;
711
712 client.disconnect(&cx.to_async()).await.unwrap();
713 assert!(server.receive::<proto::Ping>().await.is_err());
714 }
715
716 #[gpui::test(iterations = 10)]
717 async fn test_reconnection(cx: TestAppContext) {
718 cx.foreground().forbid_parking();
719
720 let user_id = 5;
721 let mut client = Client::new();
722 let server = FakeServer::for_client(user_id, &mut client, &cx).await;
723 let mut status = client.status();
724 assert!(matches!(
725 status.recv().await,
726 Some(Status::Connected { .. })
727 ));
728 assert_eq!(server.auth_count(), 1);
729
730 server.forbid_connections();
731 server.disconnect().await;
732 while !matches!(status.recv().await, Some(Status::ReconnectionError { .. })) {}
733
734 server.allow_connections();
735 cx.foreground().advance_clock(Duration::from_secs(10));
736 while !matches!(status.recv().await, Some(Status::Connected { .. })) {}
737 assert_eq!(server.auth_count(), 1); // Client reused the cached credentials when reconnecting
738
739 server.forbid_connections();
740 server.disconnect().await;
741 while !matches!(status.recv().await, Some(Status::ReconnectionError { .. })) {}
742
743 // Clear cached credentials after authentication fails
744 server.roll_access_token();
745 server.allow_connections();
746 cx.foreground().advance_clock(Duration::from_secs(10));
747 assert_eq!(server.auth_count(), 1);
748 cx.foreground().advance_clock(Duration::from_secs(10));
749 while !matches!(status.recv().await, Some(Status::Connected { .. })) {}
750 assert_eq!(server.auth_count(), 2); // Client re-authenticated due to an invalid token
751 }
752
753 #[test]
754 fn test_encode_and_decode_worktree_url() {
755 let url = encode_worktree_url(5, "deadbeef");
756 assert_eq!(decode_worktree_url(&url), Some((5, "deadbeef".to_string())));
757 assert_eq!(
758 decode_worktree_url(&format!("\n {}\t", url)),
759 Some((5, "deadbeef".to_string()))
760 );
761 assert_eq!(decode_worktree_url("not://the-right-format"), None);
762 }
763}