perf(desktop,host-service): enqueue workspace creation off pooled sockets and batch the host-service client

apps/desktop/src/renderer/lib/host-service-client.ts

@@ -1,5 +1,5 @@
 import type { AppRouter } from "@superset/host-service";
-import { createTRPCClient, httpLink } from "@trpc/client";
+import { createTRPCClient, httpBatchStreamLink } from "@trpc/client";
 import superjson from "superjson";
 import { getHostServiceHeaders } from "./host-service-auth";
 
@@ -20,7 +20,13 @@ export function getHostServiceClientByUrl(hostUrl: string): HostServiceClient {
 
 	const client = createTRPCClient<AppRouter>({
 		links: [
-			httpLink({
+			// Streaming batch link: same-tick calls share one HTTP request and
+			// one CORS preflight, but each result streams as soon as it's ready
+			// — no slowest-in-batch latency (the reason #3879 unbatched the old
+			// buffering httpBatchLink). All renderer clients share Chromium's
+			// 6-connections-per-origin pool with every other host-service
+			// request, so sockets are the scarce resource here.
+			httpBatchStreamLink({
 				url: `${hostUrl}/trpc`,
 				transformer: superjson,
 				headers: () => getHostServiceHeaders(hostUrl),

apps/desktop/src/renderer/stores/workspace-creates/useWorkspaceCreates.ts

@@ -1,9 +1,18 @@
+import {
+	getEventBus,
+	type WorkspaceCreateSettledPayload,
+} from "@superset/workspace-client";
+import { TRPCClientError } from "@trpc/client";
 import { useCallback } from "react";
 import { resolveHostUrl } from "renderer/hooks/host-service/useHostTargetUrl";
 import { useRelayUrl } from "renderer/hooks/useRelayUrl";
 import { authClient } from "renderer/lib/auth-client";
 import { electronTrpc } from "renderer/lib/electron-trpc";
-import { getHostServiceClientByUrl } from "renderer/lib/host-service-client";
+import { getHostServiceWsToken } from "renderer/lib/host-service-auth";
+import {
+	getHostServiceClientByUrl,
+	type HostServiceClient,
+} from "renderer/lib/host-service-client";
 import { getHostServiceUnavailableMessage } from "renderer/lib/host-service-unavailable";
 import { electronTrpcClient } from "renderer/lib/trpc-client";
 import { useCollections } from "renderer/routes/_authenticated/providers/CollectionsProvider";
@@ -38,6 +47,179 @@ export interface UseWorkspaceCreatesApi {
 	submit: (args: SubmitArgs) => SubmitHandle;
 }
 
+/** Sessions create a folder + git init — quick. A wedged transport must not
+ * hold the pending-create UI forever. */
+const SESSION_CREATE_TIMEOUT_MS = 30_000;
+/** The enqueue call validates and returns — it does none of the git work. */
+const ENQUEUE_TIMEOUT_MS = 15_000;
+/** Backstop for a lost `workspace:create-settled` event (host restart or WS
+ * drop mid-create). Sized for worktree adds on monorepo-scale repos. */
+const CREATE_SETTLE_TIMEOUT_MS = 10 * 60_000;
+
+/** What the completed-create pipeline needs, from either transport. */
+interface CreateOutcome {
+	workspace: { id: string; projectId: string | null };
+	terminals: Array<{ terminalId: string; label?: string }>;
+	agents: Array<
+		| { ok: true; kind: "terminal" | "chat"; sessionId: string; label: string }
+		| { ok: false; error: string }
+	>;
+}
+
+/** Older hosts don't have `workspaces.createEnqueued` yet. */
+function isMissingProcedureError(error: unknown): boolean {
+	return (
+		error instanceof TRPCClientError &&
+		/no procedure found on path/i.test(error.message)
+	);
+}
+
+/** An `AbortSignal.timeout` firing — host reachable but slow, as opposed to
+ * unreachable (network errors reject differently). */
+function isTimeoutAbort(error: unknown): boolean {
+	const cause = error instanceof TRPCClientError ? error.cause : error;
+	return (
+		cause instanceof DOMException &&
+		(cause.name === "TimeoutError" || cause.name === "AbortError")
+	);
+}
+
+/**
+ * Bounded row-existence probe: did the host actually create this workspace?
+ * "unknown" means the probe itself failed (transport/auth) — callers must not
+ * treat that as absence.
+ */
+async function probeWorkspaceRow(
+	client: HostServiceClient,
+	workspaceId: string,
+): Promise<"exists" | "absent" | "unknown"> {
+	try {
+		const existing = await client.workspace.get.query(
+			{ id: workspaceId },
+			{ signal: AbortSignal.timeout(15_000) },
+		);
+		return existing != null ? "exists" : "absent";
+	} catch (error) {
+		if (error instanceof TRPCClientError && error.data?.code === "NOT_FOUND") {
+			return "absent";
+		}
+		return "unknown";
+	}
+}
+
+/** A few spaced probes: a client-side timeout can race a create that hasn't
+ * inserted its row yet. Resolves "exists" as soon as any probe sees the row. */
+async function probeWorkspaceRowWithRetries(
+	client: HostServiceClient,
+	workspaceId: string,
+	attempts = 3,
+	delayMs = 4_000,
+): Promise<"exists" | "absent" | "unknown"> {
+	let last: "absent" | "unknown" = "absent";
+	for (let attempt = 0; attempt < attempts; attempt++) {
+		if (attempt > 0) {
+			await new Promise((resolve) => setTimeout(resolve, delayMs));
+		}
+		const result = await probeWorkspaceRow(client, workspaceId);
+		if (result === "exists") return result;
+		last = result;
+	}
+	return last;
+}
+
+/**
+ * Enqueue the create and resolve from the `workspace:create-settled` event.
+ * The synchronous `workspaces.create` mutation holds one of Chromium's six
+ * pooled sockets for the whole create (minutes on big repos) and exceeds the
+ * relay's 30s exchange cap outright; the enqueue variant returns immediately
+ * and the result arrives over the eventBus WebSocket instead.
+ */
+async function createViaEnqueue(
+	client: HostServiceClient,
+	hostUrl: string,
+	workspaceId: string,
+	payload: WorkspacesCreateInput,
+): Promise<CreateOutcome> {
+	const bus = getEventBus(hostUrl, () => getHostServiceWsToken(hostUrl));
+	const releaseBus = bus.retain();
+	let unsubscribe: () => void = () => {};
+	try {
+		// Subscribe before enqueueing so the settled event can't slip past.
+		const settled = new Promise<WorkspaceCreateSettledPayload>((resolve) => {
+			unsubscribe = bus.on(
+				"workspace:create-settled",
+				workspaceId,
+				(_workspaceId, eventPayload) => resolve(eventPayload),
+			);
+		});
+
+		try {
+			await client.workspaces.createEnqueued.mutate(payload, {
+				signal: AbortSignal.timeout(ENQUEUE_TIMEOUT_MS),
+			});
+		} catch (error) {
+			if (isMissingProcedureError(error)) {
+				// Legacy host: fall back to the long-held synchronous create.
+				const result = await client.workspaces.create.mutate(payload);
+				return result;
+			}
+			// A timed-out enqueue may still have reached the host (reachable but
+			// slow — e.g. a saturated socket pool). Hard-failing here could tear
+			// down a workspace the host is actually creating; fall through and
+			// let the settled event / backstop decide. Anything else (network
+			// error, rejection) is a definitive no-enqueue: fail now.
+			if (!isTimeoutAbort(error)) throw error;
+		}
+
+		let backstopTimer: ReturnType<typeof setTimeout> | undefined;
+		const outcome = await Promise.race([
+			settled,
+			new Promise<"timeout">((resolve) => {
+				backstopTimer = setTimeout(
+					() => resolve("timeout"),
+					CREATE_SETTLE_TIMEOUT_MS,
+				);
+			}),
+		]).finally(() => clearTimeout(backstopTimer));
+
+		if (outcome === "timeout") {
+			// The event was lost, not necessarily the create. If the row exists
+			// the workspace is real — recover with an EMPTY pane seed rather
+			// than tearing down a workspace the user can already see. Launched
+			// terminals/agents aren't recoverable without the event; the
+			// background-session auto-adopt path surfaces them on open.
+			const row = await probeWorkspaceRow(client, workspaceId);
+			if (row === "exists") {
+				return {
+					workspace: { id: workspaceId, projectId: payload.projectId },
+					terminals: [],
+					agents: [],
+				};
+			}
+			throw new Error(
+				row === "unknown"
+					? "Timed out waiting for the host to create the workspace, and the host could not be reached to verify it"
+					: "Timed out waiting for the host to create the workspace",
+			);
+		}
+
+		if (!outcome.ok) {
+			throw new Error(outcome.error ?? "Workspace creation failed");
+		}
+		return {
+			workspace: {
+				id: outcome.canonicalWorkspaceId ?? workspaceId,
+				projectId: outcome.projectId,
+			},
+			terminals: outcome.terminals,
+			agents: outcome.agents,
+		};
+	} finally {
+		unsubscribe();
+		releaseBus();
+	}
+}
+
 export function useWorkspaceCreates(): UseWorkspaceCreatesApi {
 	const hostService = useLocalHostService();
 	const { machineId, activeHostUrl } = hostService;
@@ -138,14 +320,37 @@ export function useWorkspaceCreates(): UseWorkspaceCreatesApi {
 			// agent behind the setup commands. On a cold cache the hook value is
 			// still undefined — resolve it directly so an early create can't
 			// silently skip the gate (failures fall back to default-off).
-			const createPromise = (async () => {
+			const createPromise: Promise<CreateOutcome> = (async () => {
 				const client = getHostServiceClientByUrl(hostUrl);
 				if (snapshot.projectId === null) {
-					// Sessions have no setup scripts, so no wait-for-setup gate.
+					// Sessions have no setup scripts, so no wait-for-setup gate —
+					// and no git worktree work, so the synchronous mutation stays.
 					const { projectId: _null, ...sessionInput } = snapshot;
-					const result =
-						await client.workspaces.createSession.mutate(sessionInput);
-					return { ...result, alreadyExists: false };
+					try {
+						const result = await client.workspaces.createSession.mutate(
+							sessionInput,
+							{ signal: AbortSignal.timeout(SESSION_CREATE_TIMEOUT_MS) },
+						);
+						return result;
+					} catch (error) {
+						// The abort cancels the client, not the host — a slow but
+						// ultimately successful session create must not be recorded
+						// as failed (the row would pop back via workspace:changed).
+						// Spaced probes cover a create that hasn't inserted its row
+						// yet; the empty pane seed is healed by session auto-adopt.
+						if (
+							isTimeoutAbort(error) &&
+							(await probeWorkspaceRowWithRetries(client, workspaceId)) ===
+								"exists"
+						) {
+							return {
+								workspace: { id: workspaceId, projectId: null },
+								terminals: [],
+								agents: [],
+							};
+						}
+						throw error;
+					}
 				}
 				let waitForSetup = waitForSetupBeforeAgent;
 				if (waitForSetup === undefined && snapshot.agents?.length) {
@@ -158,7 +363,7 @@ export function useWorkspaceCreates(): UseWorkspaceCreatesApi {
 					snapshot.agents?.length && waitForSetup
 						? { ...snapshot, waitForSetupBeforeAgents: true }
 						: snapshot;
-				return client.workspaces.create.mutate(payload);
+				return createViaEnqueue(client, hostUrl, workspaceId, payload);
 			})();
 
 			writeWorkspacePaneLayout(

packages/host-service/src/app.ts

@@ -78,6 +78,7 @@ export interface CreateAppResult {
 	injectWebSocket: ReturnType<typeof createNodeWebSocket>["injectWebSocket"];
 	api: ApiClient;
 	db: HostDb;
+	eventBus: EventBus;
 	dispose: () => Promise<void>;
 }
 
@@ -365,5 +366,5 @@ export function createApp(options: CreateAppOptions): CreateAppResult {
 		}
 	};
 
-	return { app, injectWebSocket, api, db, dispose };
+	return { app, injectWebSocket, api, db, eventBus, dispose };
 }

packages/host-service/src/events/event-bus.ts

@@ -224,6 +224,20 @@ export class EventBus {
 		return () => this.workspaceChangedListeners.delete(listener);
 	}
 
+	/**
+	 * Terminal event for an enqueued workspaces.createEnqueued call — carries
+	 * what the synchronous create response used to (canonical id + launched
+	 * terminals/agents), keyed by the client-minted enqueue id.
+	 */
+	broadcastWorkspaceCreateSettled(
+		message: Omit<
+			Extract<ServerMessage, { type: "workspace:create-settled" }>,
+			"type"
+		>,
+	): void {
+		this.broadcast({ type: "workspace:create-settled", ...message });
+	}
+
 	/**
 	 * Fan out project lifecycle changes (create/rename/delete) from the
 	 * host-owned projects table. Broadcast to all clients — list consumers

packages/host-service/src/events/types.ts

@@ -111,6 +111,36 @@ export interface ProjectChangedMessage {
 	occurredAt: number;
 }
 
+export interface WorkspaceCreateTerminalLaunch {
+	terminalId: string;
+	label?: string;
+}
+
+export type WorkspaceCreateAgentLaunch =
+	| { ok: true; kind: "terminal" | "chat"; sessionId: string; label: string }
+	| { ok: false; error: string };
+
+/**
+ * Terminal event for an enqueued `workspaces.createEnqueued` call. The HTTP
+ * response returns immediately; this carries what the synchronous
+ * `workspaces.create` response used to: the canonical row id (which can
+ * differ from the enqueue id when the create resolved to an existing
+ * workspace) and the launched terminals/agents for the pane-layout seed.
+ */
+export interface WorkspaceCreateSettledMessage {
+	type: "workspace:create-settled";
+	/** The client-minted id from the enqueue call — the correlation key. */
+	workspaceId: string;
+	ok: boolean;
+	canonicalWorkspaceId: string | null;
+	projectId: string | null;
+	terminals: WorkspaceCreateTerminalLaunch[];
+	agents: WorkspaceCreateAgentLaunch[];
+	alreadyExists: boolean;
+	error?: string;
+	occurredAt: number;
+}
+
 export interface EventBusErrorMessage {
 	type: "error";
 	message: string;
@@ -123,6 +153,7 @@ export type ServerMessage =
 	| TerminalLifecycleMessage
 	| PortChangedMessage
 	| WorkspaceChangedMessage
+	| WorkspaceCreateSettledMessage
 	| ProjectChangedMessage
 	| EventBusErrorMessage;
 

packages/host-service/src/terminal/terminal.create-on-attach.node-test.ts

@@ -20,6 +20,7 @@ import { fileURLToPath } from "node:url";
 import { serve } from "@hono/node-server";
 import { createNodeWebSocket } from "@hono/node-ws";
 import { Server } from "@superset/pty-daemon";
+import { eq } from "drizzle-orm";
 import { Hono } from "hono";
 import { createDb, type HostDb } from "../db/index.ts";
 import { projects, terminalSessions, workspaces } from "../db/schema.ts";
@@ -162,6 +163,15 @@ test("attach with create=1 creates the session for a brand-new id", async () =>
 	assert.deepEqual(result, { kind: "attached" });
 	assert.ok(isLiveTerminalSession(terminalId));
 
+	// The session must be persisted, not just in-memory — the row is what
+	// future attaches (and the session-gone contract) key off after restarts.
+	const row = db.query.terminalSessions
+		.findFirst({ where: eq(terminalSessions.id, terminalId) })
+		.sync();
+	assert.ok(row);
+	assert.equal(row.originWorkspaceId, workspaceId);
+	assert.equal(row.status, "active");
+
 	// The session now exists like any other: a plain re-attach works.
 	const reattach = await dial(terminalId, `?workspaceId=${workspaceId}`);
 	assert.deepEqual(reattach, { kind: "attached" });

packages/host-service/src/trpc/index.ts

@@ -98,6 +98,7 @@ const sentryMiddleware = t.middleware(async ({ next, path, type }) => {
 const baseProcedure = t.procedure.use(sentryMiddleware);
 
 export const router = t.router;
+export const createCallerFactory = t.createCallerFactory;
 export const publicProcedure = baseProcedure;
 
 export const protectedProcedure = baseProcedure.use(async ({ ctx, next }) => {