cli: allow installation as a service from the UI

- 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 #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/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...");
+
+		full_tunnel.tags = desired_tags;
 
-		let new_tunnel = spanf!(
+		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::<_, ()>(

src/vs/base/common/buffer.ts

@@ -172,51 +172,57 @@ export class VSBuffer {
 		writeUInt8(this.buffer, value, offset);
 	}
 
-	indexOf(subarray: VSBuffer | Uint8Array) {
-		const needle = subarray instanceof VSBuffer ? subarray.buffer : subarray;
-		const needleLen = needle.byteLength;
-		const haystack = this.buffer;
-		const haystackLen = haystack.byteLength;
-
-		if (needleLen === 0) {
-			return 0;
-		}
+	indexOf(subarray: VSBuffer | Uint8Array, offset = 0) {
+		return binaryIndexOf(this.buffer, subarray instanceof VSBuffer ? subarray.buffer : subarray, offset);
+	}
+}
 
-		if (needleLen === 1) {
-			return haystack.indexOf(needle[0]);
-		}
+/**
+ * Like String.indexOf, but works on Uint8Arrays.
+ * Uses the boyer-moore-horspool algorithm to be reasonably speedy.
+ */
+export function binaryIndexOf(haystack: Uint8Array, needle: Uint8Array, offset = 0): number {
+	const needleLen = needle.byteLength;
+	const haystackLen = haystack.byteLength;
 
-		if (needleLen > haystackLen) {
-			return -1;
-		}
+	if (needleLen === 0) {
+		return 0;
+	}
 
-		// find index of the subarray using boyer-moore-horspool algorithm
-		const table = indexOfTable.value;
-		table.fill(needle.length);
-		for (let i = 0; i < needle.length; i++) {
-			table[needle[i]] = needle.length - i - 1;
-		}
+	if (needleLen === 1) {
+		return haystack.indexOf(needle[0]);
+	}
+
+	if (needleLen > haystackLen - offset) {
+		return -1;
+	}
+
+	// find index of the subarray using boyer-moore-horspool algorithm
+	const table = indexOfTable.value;
+	table.fill(needle.length);
+	for (let i = 0; i < needle.length; i++) {
+		table[needle[i]] = needle.length - i - 1;
+	}
 
-		let i = needle.length - 1;
-		let j = i;
-		let result = -1;
-		while (i < haystackLen) {
-			if (haystack[i] === needle[j]) {
-				if (j === 0) {
-					result = i;
-					break;
-				}
-
-				i--;
-				j--;
-			} else {
-				i += Math.max(needle.length - j, table[haystack[i]]);
-				j = needle.length - 1;
+	let i = offset + needle.length - 1;
+	let j = i;
+	let result = -1;
+	while (i < haystackLen) {
+		if (haystack[i] === needle[j]) {
+			if (j === 0) {
+				result = i;
+				break;
 			}
-		}
 
-		return result;
+			i--;
+			j--;
+		} else {
+			i += Math.max(needle.length - j, table[haystack[i]]);
+			j = needle.length - 1;
+		}
 	}
+
+	return result;
 }
 
 export function readUInt16LE(source: Uint8Array, offset: number): number {

src/vs/base/node/nodeStreams.ts

@@ -0,0 +1,62 @@
+/*---------------------------------------------------------------------------------------------
+ *  Copyright (c) Microsoft Corporation. All rights reserved.
+ *  Licensed under the MIT License. See License.txt in the project root for license information.
+ *--------------------------------------------------------------------------------------------*/
+import { Transform } from 'stream';
+import { binaryIndexOf } from 'vs/base/common/buffer';
+
+/**
+ * A Transform stream that splits the input on the "splitter" substring.
+ * The resulting chunks will contain (and trail with) the splitter match.
+ * The last chunk when the stream ends will be emitted even if a splitter
+ * is not encountered.
+ */
+export class StreamSplitter extends Transform {
+	private buffer: Buffer | undefined;
+	private readonly splitter: Buffer | number;
+	private readonly spitterLen: number;
+
+	constructor(splitter: string | number | Buffer) {
+		super();
+		if (typeof splitter === 'number') {
+			this.splitter = splitter;
+			this.spitterLen = 1;
+		} else {
+			const buf = Buffer.isBuffer(splitter) ? splitter : Buffer.from(splitter);
+			this.splitter = buf.length === 1 ? buf[0] : buf;
+			this.spitterLen = buf.length;
+		}
+	}
+
+	override _transform(chunk: Buffer, _encoding: string, callback: (error?: Error | null, data?: any) => void): void {
+		if (!this.buffer) {
+			this.buffer = chunk;
+		} else {
+			this.buffer = Buffer.concat([this.buffer, chunk]);
+		}
+
+		let offset = 0;
+		while (offset < this.buffer.length) {
+			const index = typeof this.splitter === 'number'
+				? this.buffer.indexOf(this.splitter, offset)
+				: binaryIndexOf(this.buffer, this.splitter, offset);
+			if (index === -1) {
+				break;
+			}
+
+			this.push(this.buffer.slice(offset, index + this.spitterLen));
+			offset = index + this.spitterLen;
+		}
+
+		this.buffer = offset === this.buffer.length ? undefined : this.buffer.slice(offset);
+		callback();
+	}
+
+	override _flush(callback: (error?: Error | null, data?: any) => void): void {
+		if (this.buffer) {
+			this.push(this.buffer);
+		}
+
+		callback();
+	}
+}

src/vs/base/test/node/nodeStreams.test.ts

@@ -0,0 +1,51 @@
+/*---------------------------------------------------------------------------------------------
+ *  Copyright (c) Microsoft Corporation. All rights reserved.
+ *  Licensed under the MIT License. See License.txt in the project root for license information.
+ *--------------------------------------------------------------------------------------------*/
+
+
+import { Writable } from 'stream';
+import * as assert from 'assert';
+import { StreamSplitter } from 'vs/base/node/nodeStreams';
+
+suite('StreamSplitter', () => {
+	test('should split a stream on a single character splitter', (done) => {
+		const chunks: string[] = [];
+		const splitter = new StreamSplitter('\n');
+		const writable = new Writable({
+			write(chunk, _encoding, callback) {
+				chunks.push(chunk.toString());
+				callback();
+			},
+		});
+
+		splitter.pipe(writable);
+		splitter.write('hello\nwor');
+		splitter.write('ld\n');
+		splitter.write('foo\nbar\nz');
+		splitter.end(() => {
+			assert.deepStrictEqual(chunks, ['hello\n', 'world\n', 'foo\n', 'bar\n', 'z']);
+			done();
+		});
+	});
+
+	test('should split a stream on a multi-character splitter', (done) => {
+		const chunks: string[] = [];
+		const splitter = new StreamSplitter('---');
+		const writable = new Writable({
+			write(chunk, _encoding, callback) {
+				chunks.push(chunk.toString());
+				callback();
+			},
+		});
+
+		splitter.pipe(writable);
+		splitter.write('hello---wor');
+		splitter.write('ld---');
+		splitter.write('foo---bar---z');
+		splitter.end(() => {
+			assert.deepStrictEqual(chunks, ['hello---', 'world---', 'foo---', 'bar---', 'z']);
+			done();
+		});
+	});
+});