cli: allow installation as a service from the UI (#187869)

- When turning on remote tunnel access, a quickpick is now shown asking
  users whether it should be installed as a service or just run in
	the session.
- Picking the service install will install the tunnel as a service on
  the machine, and start it.
- Turning off remote tunnel access will uninstall the service only if
  we were the ones to install it.
- This involved some refactoring to add extra state to the RemoteTunnelService.
  There's now a "mode" that includes the previous "session" and reflects
	the desired end state.
- I also did a cleanup with a `StreamSplitter` to ensure output of the
  CLI gets read line-by-line. This was depended upon by the remote tunnel
	service code, but it's not actually guaranteed.
- Changes in the CLI: allow setting the tunnel name while installing the
  service, and make both service un/installation and renames idempotent.

Closes https://github.com/microsoft/vscode/issues/184663

cli/src/commands/args.rs

@@ -649,6 +649,10 @@ pub struct TunnelServiceInstallArgs {
 	/// If set, the user accepts the server license terms and the server will be started without a user prompt.
 	#[clap(long)]
 	pub accept_server_license_terms: bool,
+
+	/// Sets the machine name for port forwarding service
+	#[clap(long)]
+	pub name: Option<String>,
 }
 
 #[derive(Args, Debug, Clone)]

cli/src/commands/tunnels.rs

@@ -135,10 +135,17 @@ pub async fn service(
 	let manager = create_service_manager(ctx.log.clone(), &ctx.paths);
 	match service_args {
 		TunnelServiceSubCommands::Install(args) => {
-			// ensure logged in, otherwise subsequent serving will fail
-			Auth::new(&ctx.paths, ctx.log.clone())
-				.get_credential()
-				.await?;
+			let auth = Auth::new(&ctx.paths, ctx.log.clone());
+
+			if let Some(name) = &args.name {
+				// ensure the name matches, and tunnel exists
+				dev_tunnels::DevTunnels::new(&ctx.log, auth, &ctx.paths)
+					.rename_tunnel(name)
+					.await?;
+			} else {
+				// still ensure they're logged in, otherwise subsequent serving will fail
+				auth.get_credential().await?;
+			}
 
 			// likewise for license consent
 			legal::require_consent(&ctx.paths, args.accept_server_license_terms)?;
@@ -203,20 +210,20 @@ pub async fn user(ctx: CommandContext, user_args: TunnelUserSubCommands) -> Resu
 	Ok(0)
 }
 
-/// Remove the tunnel used by this gateway, if any.
+/// Remove the tunnel used by this tunnel, if any.
 pub async fn rename(ctx: CommandContext, rename_args: TunnelRenameArgs) -> Result<i32, AnyError> {
 	let auth = Auth::new(&ctx.paths, ctx.log.clone());
 	let mut dt = dev_tunnels::DevTunnels::new(&ctx.log, auth, &ctx.paths);
 	dt.rename_tunnel(&rename_args.name).await?;
 	ctx.log.result(format!(
-		"Successfully renamed this gateway to {}",
+		"Successfully renamed this tunnel to {}",
 		&rename_args.name
 	));
 
 	Ok(0)
 }
 
-/// Remove the tunnel used by this gateway, if any.
+/// Remove the tunnel used by this tunnel, if any.
 pub async fn unregister(ctx: CommandContext) -> Result<i32, AnyError> {
 	let auth = Auth::new(&ctx.paths, ctx.log.clone());
 	let mut dt = dev_tunnels::DevTunnels::new(&ctx.log, auth, &ctx.paths);

cli/src/rpc.rs

@@ -117,7 +117,7 @@ impl<S: Serialization, C: Send + Sync + 'static> RpcMethodBuilder<S, C> {
 					Ok(p) => p,
 					Err(err) => {
 						return id.map(|id| {
-							serial.serialize(&ErrorResponse {
+							serial.serialize(ErrorResponse {
 								id,
 								error: ResponseError {
 									code: 0,
@@ -131,7 +131,7 @@ impl<S: Serialization, C: Send + Sync + 'static> RpcMethodBuilder<S, C> {
 				match callback(param.params, &context) {
 					Ok(result) => id.map(|id| serial.serialize(&SuccessResponse { id, result })),
 					Err(err) => id.map(|id| {
-						serial.serialize(&ErrorResponse {
+						serial.serialize(ErrorResponse {
 							id,
 							error: ResponseError {
 								code: -1,
@@ -161,7 +161,7 @@ impl<S: Serialization, C: Send + Sync + 'static> RpcMethodBuilder<S, C> {
 					Ok(p) => p,
 					Err(err) => {
 						return future::ready(id.map(|id| {
-							serial.serialize(&ErrorResponse {
+							serial.serialize(ErrorResponse {
 								id,
 								error: ResponseError {
 									code: 0,
@@ -182,7 +182,7 @@ impl<S: Serialization, C: Send + Sync + 'static> RpcMethodBuilder<S, C> {
 							id.map(|id| serial.serialize(&SuccessResponse { id, result }))
 						}
 						Err(err) => id.map(|id| {
-							serial.serialize(&ErrorResponse {
+							serial.serialize(ErrorResponse {
 								id,
 								error: ResponseError {
 									code: -1,
@@ -222,7 +222,7 @@ impl<S: Serialization, C: Send + Sync + 'static> RpcMethodBuilder<S, C> {
 						return (
 							None,
 							future::ready(id.map(|id| {
-								serial.serialize(&ErrorResponse {
+								serial.serialize(ErrorResponse {
 									id,
 									error: ResponseError {
 										code: 0,
@@ -255,7 +255,7 @@ impl<S: Serialization, C: Send + Sync + 'static> RpcMethodBuilder<S, C> {
 					match callback(servers, param.params, context).await {
 						Ok(r) => id.map(|id| serial.serialize(&SuccessResponse { id, result: r })),
 						Err(err) => id.map(|id| {
-							serial.serialize(&ErrorResponse {
+							serial.serialize(ErrorResponse {
 								id,
 								error: ResponseError {
 									code: -1,
@@ -427,7 +427,7 @@ impl<S: Serialization, C: Send + Sync> RpcDispatcher<S, C> {
 				Some(Method::Async(callback)) => MaybeSync::Future(callback(id, body)),
 				Some(Method::Duplex(callback)) => MaybeSync::Stream(callback(id, body)),
 				None => MaybeSync::Sync(id.map(|id| {
-					self.serializer.serialize(&ErrorResponse {
+					self.serializer.serialize(ErrorResponse {
 						id,
 						error: ResponseError {
 							code: -1,

cli/src/tunnels/dev_tunnels.rs

@@ -275,7 +275,9 @@ impl DevTunnels {
 
 	/// Renames the current tunnel to the new name.
 	pub async fn rename_tunnel(&mut self, name: &str) -> Result<(), AnyError> {
-		self.update_tunnel_name(None, name).await.map(|_| ())
+		self.update_tunnel_name(self.launcher_tunnel.load(), name)
+			.await
+			.map(|_| ())
 	}
 
 	/// Updates the name of the existing persisted tunnel to the new name.
@@ -286,28 +288,34 @@ impl DevTunnels {
 		name: &str,
 	) -> Result<(Tunnel, PersistedTunnel), AnyError> {
 		let name = name.to_ascii_lowercase();
-		self.check_is_name_free(&name).await?;
-
-		debug!(self.log, "Tunnel name changed, applying updates...");
 
 		let (mut full_tunnel, mut persisted, is_new) = match persisted {
 			Some(persisted) => {
+				debug!(
+					self.log,
+					"Found a persisted tunnel, seeing if the name matches..."
+				);
 				self.get_or_create_tunnel(persisted, Some(&name), NO_REQUEST_OPTIONS)
 					.await
 			}
-			None => self
-				.create_tunnel(&name, NO_REQUEST_OPTIONS)
-				.await
-				.map(|(pt, t)| (t, pt, true)),
+			None => {
+				debug!(self.log, "Creating a new tunnel with the requested name");
+				self.create_tunnel(&name, NO_REQUEST_OPTIONS)
+					.await
+					.map(|(pt, t)| (t, pt, true))
+			}
 		}?;
 
-		if is_new {
+		let desired_tags = self.get_tags(&name);
+		if is_new || vec_eq_as_set(&full_tunnel.tags, &desired_tags) {
 			return Ok((full_tunnel, persisted));
 		}
 
-		full_tunnel.tags = self.get_tags(&name);
+		debug!(self.log, "Tunnel name changed, applying updates...");
 
-		let new_tunnel = spanf!(
+		full_tunnel.tags = desired_tags;
+
+		let updated_tunnel = spanf!(
 			self.log,
 			self.log.span("dev-tunnel.tag.update"),
 			self.client.update_tunnel(&full_tunnel, NO_REQUEST_OPTIONS)
@@ -317,7 +325,7 @@ impl DevTunnels {
 		persisted.name = name;
 		self.launcher_tunnel.save(Some(persisted.clone()))?;
 
-		Ok((new_tunnel, persisted))
+		Ok((updated_tunnel, persisted))
 	}
 
 	/// Gets the persisted tunnel from the service, or creates a new one.
@@ -443,6 +451,8 @@ impl DevTunnels {
 	) -> Result<(PersistedTunnel, Tunnel), AnyError> {
 		info!(self.log, "Creating tunnel with the name: {}", name);
 
+		self.check_is_name_free(name).await?;
+
 		let mut tried_recycle = false;
 
 		let new_tunnel = Tunnel {
@@ -527,7 +537,7 @@ impl DevTunnels {
 		options: &TunnelRequestOptions,
 	) -> Result<Tunnel, AnyError> {
 		let new_tags = self.get_tags(name);
-		if vec_eq_unsorted(&tunnel.tags, &new_tags) {
+		if vec_eq_as_set(&tunnel.tags, &new_tags) {
 			return Ok(tunnel);
 		}
 
@@ -610,7 +620,7 @@ impl DevTunnels {
 	}
 
 	async fn check_is_name_free(&mut self, name: &str) -> Result<(), AnyError> {
-		let existing = spanf!(
+		let existing: Vec<Tunnel> = spanf!(
 			self.log,
 			self.log.span("dev-tunnel.rename.search"),
 			self.client.list_all_tunnels(&TunnelRequestOptions {
@@ -998,7 +1008,7 @@ fn clean_hostname_for_tunnel(hostname: &str) -> String {
 	}
 }
 
-fn vec_eq_unsorted(a: &[String], b: &[String]) -> bool {
+fn vec_eq_as_set(a: &[String], b: &[String]) -> bool {
 	if a.len() != b.len() {
 		return false;
 	}

cli/src/tunnels/service_windows.rs

@@ -78,6 +78,7 @@ impl CliServiceManager for WindowsService {
 		cmd.stderr(Stdio::null());
 		cmd.stdout(Stdio::null());
 		cmd.stdin(Stdio::null());
+		cmd.creation_flags(CREATE_NEW_PROCESS_GROUP | DETACHED_PROCESS);
 		cmd.spawn()
 			.map_err(|e| wrapdbg(e, "error starting service"))?;
 
@@ -121,8 +122,12 @@ impl CliServiceManager for WindowsService {
 
 	async fn unregister(&self) -> Result<(), AnyError> {
 		let key = WindowsService::open_key()?;
-		key.delete_value(TUNNEL_ACTIVITY_NAME)
-			.map_err(|e| AnyError::from(wrap(e, "error deleting registry key")))?;
+		match key.delete_value(TUNNEL_ACTIVITY_NAME) {
+			Ok(_) => {}
+			Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
+			Err(e) => return Err(wrap(e, "error deleting registry key").into()),
+		}
+
 		info!(self.log, "Tunnel service uninstalled");
 
 		let r = do_single_rpc_call::<_, ()>(