fix(desktop): serialize background service connects through one state machine (#61748)

This commit is contained in:
Luke Parker
2026-11-04 14:38:43 -06:00
committed by GitHub
Unverified
parent bbf71d6205
commit 2047c0d4fd
13 changed files with 921 additions and 189 deletions
+3 -3
View File
@@ -106,9 +106,9 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
},
})
// The plugin reports a new endpoint: requests use it at once, and a connected stream moves to it now. An endpoint
// the stream's own reconnect already moved to needs neither.
if (server.type === "plugin") {
// The plugin, or the desktop after another window's reconnect, reports a new endpoint: requests use it at once, and
// a connected stream moves to it now. An endpoint the stream's own reconnect already moved to needs neither.
if (server.type === "plugin" || server.type === "sidecar") {
createEffect(
on(
() => `${server.http.url}\0${server.http.password ?? ""}`,
@@ -0,0 +1,408 @@
import { describe, expect, test } from "bun:test"
import { Deferred, Effect, Fiber } from "effect"
import { BackgroundServiceMachine } from "./background-service-machine"
const A = { url: "http://127.0.0.1:4100", password: "a" }
const B = { url: "http://127.0.0.1:4200", password: "b" }
const C = { url: "http://127.0.0.1:4300", password: "c" }
const reply = (): BackgroundServiceMachine.Reply => ({ resolve: () => undefined, reject: () => undefined })
// Folds events through the machine and keeps the commands of each step.
function replay(
events: readonly BackgroundServiceMachine.Event[],
from: BackgroundServiceMachine.State = BackgroundServiceMachine.initialTransition().state,
) {
return events.reduce<{
state: BackgroundServiceMachine.State
commands: (readonly BackgroundServiceMachine.Command[])[]
}>(
(result, event) => {
const next = BackgroundServiceMachine.transition(result.state, event)
return { state: next.state, commands: [...result.commands, next.commands] }
},
{ state: from, commands: [] },
)
}
const readyA = replay([{ type: "connect.succeeded", attempt: 1, data: A }]).state
describe("background service connection", () => {
test("starts the initial connect at once", () => {
expect(BackgroundServiceMachine.initialTransition()).toEqual({
state: { status: "connecting", attempt: 1, waiters: [] },
commands: [{ type: "connect", attempt: 1, mode: "initial" }],
})
})
test("windows that reconnect together share one attempt and all move to its service", () => {
const [r1, r2, r3] = [reply(), reply(), reply()]
const result = replay(
[
{ type: "reconnect.requested", reply: r1 },
{ type: "reconnect.requested", reply: r2 },
{ type: "connection.requested", reply: r3 },
{ type: "connect.succeeded", attempt: 2, data: B },
],
readyA,
)
expect(result.commands).toEqual([
[{ type: "connect", attempt: 2, mode: "reconnect" }],
[],
[],
[
{ type: "publish", data: B },
{ type: "resolve", replies: [r1, r2, r3], data: B },
],
])
expect(result.state).toEqual({ status: "ready", attempt: 2, data: B })
})
test("a reconnect that finds the same service publishes nothing", () => {
const r1 = reply()
const result = replay(
[
{ type: "reconnect.requested", reply: r1 },
{ type: "connect.succeeded", attempt: 2, data: { ...A } },
],
readyA,
)
expect(result.commands.at(-1)).toEqual([{ type: "resolve", replies: [r1], data: A }])
})
test("a probed service is adopted at once and replaced by the one its verification settles on", () => {
const r1 = reply()
const probed = replay([
{ type: "connection.requested", reply: r1 },
{ type: "probe.succeeded", attempt: 1, data: A },
])
expect(probed.commands.at(-1)).toEqual([
{ type: "publish", data: A },
{ type: "resolve", replies: [r1], data: A },
{ type: "verify", attempt: 1 },
])
const replaced = replay([{ type: "verify.succeeded", attempt: 1, data: B }], probed.state)
expect(replaced.state).toEqual({ status: "ready", attempt: 1, data: B })
expect(replaced.commands).toEqual([[{ type: "publish", data: B }]])
const confirmed = replay([{ type: "verify.succeeded", attempt: 1, data: { ...A } }], probed.state)
expect(confirmed.state).toBe(probed.state)
expect(confirmed.commands).toEqual([[]])
})
test("a failed verification keeps the probed service", () => {
const probed = replay([{ type: "probe.succeeded", attempt: 1, data: A }]).state
const result = replay([{ type: "verify.failed", attempt: 1, error: new Error("handoff failed") }], probed)
expect(result.state).toBe(probed)
expect(result.commands).toEqual([[]])
})
test("a failed initial connect is retried by the next window instead of replayed", () => {
const [r1, r2] = [reply(), reply()]
const error = new Error("config invalid")
const result = replay([
{ type: "connection.requested", reply: r1 },
{ type: "connect.failed", attempt: 1, error },
{ type: "connection.requested", reply: r2 },
{ type: "connect.succeeded", attempt: 2, data: A },
])
expect(result.commands).toEqual([
[],
[{ type: "reject", replies: [r1], error }],
[{ type: "connect", attempt: 2, mode: "initial" }],
[
{ type: "publish", data: A },
{ type: "resolve", replies: [r2], data: A },
],
])
})
test("a failed reconnect keeps the last service and retries as a reconnect", () => {
const [r1, r2] = [reply(), reply()]
const error = new Error("port in use")
const failed = replay(
[
{ type: "reconnect.requested", reply: r1 },
{ type: "connect.failed", attempt: 2, error },
],
readyA,
)
expect(failed.state).toEqual({ status: "failed", attempt: 2, error, last: A })
expect(failed.commands.at(-1)).toEqual([{ type: "reject", replies: [r1], error }])
const retried = replay([{ type: "connection.requested", reply: r2 }], failed.state)
expect(retried.state).toEqual({ status: "reconnecting", attempt: 3, data: A, waiters: [], joiners: [r2] })
expect(retried.commands).toEqual([[{ type: "connect", attempt: 3, mode: "reconnect" }]])
})
test("a window that opens during a reconnect that fails still gets the last service", () => {
const [reconnect, opened, retry] = [reply(), reply(), reply()]
const error = new Error("port in use")
const result = replay(
[
{ type: "reconnect.requested", reply: reconnect },
{ type: "connection.requested", reply: opened },
{ type: "connect.failed", attempt: 2, error },
// The next window to open retries, and keeps the last service if that fails too.
{ type: "connection.requested", reply: retry },
{ type: "connect.failed", attempt: 3, error },
],
readyA,
)
expect(result.commands).toEqual([
[{ type: "connect", attempt: 2, mode: "reconnect" }],
[],
[
{ type: "resolve", replies: [opened], data: A },
{ type: "reject", replies: [reconnect], error },
],
[{ type: "connect", attempt: 3, mode: "reconnect" }],
[{ type: "resolve", replies: [retry], data: A }],
])
expect(result.state).toEqual({ status: "failed", attempt: 3, error, last: A })
})
test("without a previous service, every caller of a failed attempt gets the error", () => {
const [opened, reconnect] = [reply(), reply()]
const error = new Error("config invalid")
const result = replay([
{ type: "connection.requested", reply: opened },
{ type: "reconnect.requested", reply: reconnect },
{ type: "connect.failed", attempt: 1, error },
])
expect(result.commands.at(-1)).toEqual([{ type: "reject", replies: [opened, reconnect], error }])
})
test("ignores results of an attempt a reconnect superseded", () => {
const r1 = reply()
const probed = replay([{ type: "probe.succeeded", attempt: 1, data: A }]).state
const reconnecting = replay([{ type: "reconnect.requested", reply: r1 }], probed).state
const result = replay(
[
{ type: "verify.succeeded", attempt: 1, data: C },
{ type: "verify.failed", attempt: 1, error: new Error("late") },
{ type: "connect.succeeded", attempt: 1, data: C },
{ type: "connect.failed", attempt: 1, error: new Error("late") },
{ type: "probe.succeeded", attempt: 2, data: C },
],
reconnecting,
)
expect(result.state).toBe(reconnecting)
expect(result.commands.flat()).toEqual([])
})
test("a window that closes while it waits for a reconnect is not answered", () => {
const [reconnect, opened] = [reply(), reply()]
const error = new Error("port in use")
const result = replay(
[
{ type: "reconnect.requested", reply: reconnect },
{ type: "connection.requested", reply: opened },
{ type: "waiter.cancelled", reply: opened },
{ type: "connect.failed", attempt: 2, error },
],
readyA,
)
expect(result.commands.at(-1)).toEqual([{ type: "reject", replies: [reconnect], error }])
})
test("a window that stops waiting leaves the attempt to the others", () => {
const [r1, r2] = [reply(), reply()]
const waiting = replay([
{ type: "connection.requested", reply: r1 },
{ type: "connection.requested", reply: r2 },
]).state
const cancelled = replay([{ type: "waiter.cancelled", reply: r1 }], waiting)
const unknown = replay([{ type: "waiter.cancelled", reply: reply() }], cancelled.state)
const done = replay([{ type: "connect.succeeded", attempt: 1, data: A }], cancelled.state)
expect(cancelled.state).toEqual({ status: "connecting", attempt: 1, waiters: [r2] })
expect(unknown.state).toBe(cancelled.state)
expect(done.commands.at(-1)).toContainEqual({ type: "resolve", replies: [r2], data: A })
})
test("disposal rejects every waiter and every later caller", () => {
const [r1, r2, r3] = [reply(), reply(), reply()]
const reconnecting = replay([{ type: "reconnect.requested", reply: r1 }], readyA).state
const result = replay(
[
{ type: "disposed" },
{ type: "connect.succeeded", attempt: 2, data: B },
{ type: "connection.requested", reply: r2 },
{ type: "reconnect.requested", reply: r3 },
],
reconnecting,
)
expect(result.state).toEqual({ status: "disposed" })
expect(result.commands.map((commands) => commands.map((command) => command.type))).toEqual([
["reject"],
[],
["reject"],
["reject"],
])
expect(result.commands.flat().flatMap((command) => (command.type === "reject" ? command.replies : []))).toEqual([
r1,
r2,
r3,
])
})
})
describe("background service driver", () => {
test("callers wait for the initial connect", async () => {
const result = await Effect.runPromise(
Effect.scoped(
Effect.gen(function* () {
const gate = yield* Deferred.make<void>()
const machine = yield* BackgroundServiceMachine.make({
probe: Effect.succeed(undefined),
ensure: () => Deferred.await(gate).pipe(Effect.as(A)),
publish: () => undefined,
})
const consumer = yield* Effect.forkChild(machine.connection, { startImmediately: true })
yield* Effect.yieldNow
expect(consumer.pollUnsafe()).toBeUndefined()
yield* Deferred.succeed(gate, undefined)
return yield* Fiber.join(consumer)
}),
),
)
expect(result).toEqual(A)
})
test("a connect that dies fails its callers, and the next caller connects again", async () => {
const result = await Effect.runPromise(
Effect.scoped(
Effect.gen(function* () {
const gate = yield* Deferred.make<void>()
const attempts: string[] = []
const machine = yield* BackgroundServiceMachine.make({
probe: Effect.succeed(undefined),
ensure: (mode) =>
Effect.suspend(() => {
attempts.push(mode)
return attempts.length === 1
? Deferred.await(gate).pipe(Effect.andThen(Effect.die(new Error("CLI missing"))))
: Effect.succeed(A)
}),
publish: () => undefined,
})
const waiting = yield* Effect.forkChild(machine.connection.pipe(Effect.exit), { startImmediately: true })
yield* Deferred.succeed(gate, undefined)
const first = yield* Fiber.join(waiting)
return { first: first._tag, second: yield* machine.connection, attempts }
}),
),
)
expect(result).toEqual({ first: "Failure", second: A, attempts: ["initial", "initial"] })
})
test("concurrent reconnects run one ensure and publish each new service once", async () => {
const result = await Effect.runPromise(
Effect.scoped(
Effect.gen(function* () {
const gate = yield* Deferred.make<void>()
const attempts: string[] = []
const published: string[] = []
const machine = yield* BackgroundServiceMachine.make({
probe: Effect.succeed(undefined),
ensure: (mode) =>
Effect.suspend(() => {
attempts.push(mode)
return mode === "initial" ? Effect.succeed(A) : Deferred.await(gate).pipe(Effect.as(B))
}),
publish: (data) => published.push(data.url),
})
yield* machine.connection
const fibers = yield* Effect.forEach([1, 2, 3], () =>
Effect.forkChild(machine.reconnect, { startImmediately: true }),
)
yield* Effect.yieldNow
yield* Deferred.succeed(gate, undefined)
return { replies: yield* Fiber.joinAll(fibers), attempts, published }
}),
),
)
expect(result).toEqual({
replies: [B, B, B],
attempts: ["initial", "reconnect"],
published: [A.url, B.url],
})
})
test("adopts a probed service and moves to the one its verification starts", async () => {
const result = await Effect.runPromise(
Effect.scoped(
Effect.gen(function* () {
const gate = yield* Deferred.make<void>()
const verified = yield* Deferred.make<void>()
const published: string[] = []
const machine = yield* BackgroundServiceMachine.make({
probe: Effect.succeed(A),
ensure: () => Deferred.await(gate).pipe(Effect.as(B)),
publish: (data) => {
published.push(data.url)
if (data.url === B.url) Deferred.doneUnsafe(verified, Effect.void)
},
})
const first = yield* machine.connection
yield* Deferred.succeed(gate, undefined)
yield* Deferred.await(verified)
return { first, next: yield* machine.connection, published }
}),
),
)
expect(result).toEqual({ first: A, next: B, published: [A.url, B.url] })
})
})
@@ -0,0 +1,351 @@
export * as BackgroundServiceMachine from "./background-service-machine"
import { Cause, Effect, Exit, FiberSet } from "effect"
import type { SidecarCredentials } from "./sidecar-credentials"
type Data = SidecarCredentials.Data
/** `initial` may adopt a service the entry module found; `reconnect` always runs `Service.ensure`. */
export type ConnectMode = "initial" | "reconnect"
/** Settles one caller of `connection` or `reconnect`. */
export type Reply = {
readonly resolve: (data: Data) => void
readonly reject: (error: Error) => void
}
/**
* The connection to the background service every window shares. Each attempt to reach it has an id, and a result
* of any other attempt is stale. Callers that arrive while an attempt runs wait for that attempt.
*/
export type State =
/** No service has been reached yet. */
| { readonly status: "connecting"; readonly attempt: number; readonly waiters: readonly Reply[] }
/** `data` is the active service. A probed service is still verified by `attempt`'s `Service.ensure`. */
| { readonly status: "ready"; readonly attempt: number; readonly data: Data }
/**
* Requests keep using `data` until the reconnect replaces it. `waiters` asked to reconnect and share its outcome.
* `joiners` opened or reloaded a window meanwhile: a failed reconnect answers them with `data`, the service they
* would have received before it started.
*/
| {
readonly status: "reconnecting"
readonly attempt: number
readonly data: Data
readonly waiters: readonly Reply[]
readonly joiners: readonly Reply[]
}
/**
* The last attempt failed; the next caller starts a new one instead of receiving `error` again. `last` is the
* service reached before it, if any.
*/
| { readonly status: "failed"; readonly attempt: number; readonly error: Error; readonly last: Data | undefined }
| { readonly status: "disposed" }
export type Event =
| { readonly type: "connection.requested"; readonly reply: Reply }
| { readonly type: "reconnect.requested"; readonly reply: Reply }
| { readonly type: "waiter.cancelled"; readonly reply: Reply }
| { readonly type: "probe.succeeded"; readonly attempt: number; readonly data: Data }
| { readonly type: "connect.succeeded"; readonly attempt: number; readonly data: Data }
| { readonly type: "connect.failed"; readonly attempt: number; readonly error: Error }
| { readonly type: "verify.succeeded"; readonly attempt: number; readonly data: Data }
| { readonly type: "verify.failed"; readonly attempt: number; readonly error: Error }
| { readonly type: "disposed" }
export type Command =
| { readonly type: "connect"; readonly attempt: number; readonly mode: ConnectMode }
| { readonly type: "verify"; readonly attempt: number }
| { readonly type: "publish"; readonly data: Data }
| { readonly type: "resolve"; readonly replies: readonly Reply[]; readonly data: Data }
| { readonly type: "reject"; readonly replies: readonly Reply[]; readonly error: Error }
export type Transition = {
readonly state: State
readonly commands: readonly Command[]
}
const DISPOSED_ERROR = new Error("Background service is disposed")
export function initialTransition(): Transition {
return {
state: { status: "connecting", attempt: 1, waiters: [] },
commands: [{ type: "connect", attempt: 1, mode: "initial" }],
}
}
export function transition(state: State, event: Event): Transition {
if (state.status === "disposed") {
if (event.type === "connection.requested" || event.type === "reconnect.requested")
return { state, commands: [{ type: "reject", replies: [event.reply], error: DISPOSED_ERROR }] }
return unchanged(state)
}
if (event.type === "disposed")
return {
state: { status: "disposed" },
commands: reject(pending(state), DISPOSED_ERROR),
}
if (state.status === "connecting") return connecting(state, event)
if (state.status === "reconnecting") return reconnecting(state, event)
if (state.status === "ready") return ready(state, event)
if (event.type !== "connection.requested" && event.type !== "reconnect.requested") return unchanged(state)
const attempt = state.attempt + 1
// A service that was reached before is replaced the way a reconnect replaces it; otherwise this is a new first
// connect, so a failure that has since been fixed (a config error, a port race) is not replayed.
if (state.last)
return {
state: {
status: "reconnecting",
attempt,
data: state.last,
waiters: event.type === "reconnect.requested" ? [event.reply] : [],
joiners: event.type === "connection.requested" ? [event.reply] : [],
},
commands: [{ type: "connect", attempt, mode: "reconnect" }],
}
return {
state: { status: "connecting", attempt, waiters: [event.reply] },
commands: [{ type: "connect", attempt, mode: "initial" }],
}
}
// No service was reached yet, so every caller shares the attempt's outcome, failure included.
function connecting(state: Extract<State, { status: "connecting" }>, event: Event): Transition {
// Every caller joins the attempt in flight, so windows that start together run one `Service.ensure`.
if (event.type === "connection.requested" || event.type === "reconnect.requested")
return { state: { ...state, waiters: [...state.waiters, event.reply] }, commands: [] }
if (event.type === "waiter.cancelled") {
if (!state.waiters.includes(event.reply)) return unchanged(state)
return { state: { ...state, waiters: without(state.waiters, event.reply) }, commands: [] }
}
if (!("attempt" in event) || event.attempt !== state.attempt) return unchanged(state)
if (event.type === "probe.succeeded")
return {
state: { status: "ready", attempt: state.attempt, data: event.data },
commands: [
{ type: "publish", data: event.data },
...resolve(state.waiters, event.data),
{ type: "verify", attempt: state.attempt },
],
}
if (event.type === "connect.succeeded")
return {
state: { status: "ready", attempt: state.attempt, data: event.data },
commands: [{ type: "publish", data: event.data }, ...resolve(state.waiters, event.data)],
}
if (event.type === "connect.failed")
return {
state: { status: "failed", attempt: state.attempt, error: event.error, last: undefined },
commands: reject(state.waiters, event.error),
}
return unchanged(state)
}
function reconnecting(state: Extract<State, { status: "reconnecting" }>, event: Event): Transition {
// Windows that reconnect together run one `Service.ensure`; a window that opens meanwhile waits for it too.
if (event.type === "reconnect.requested")
return { state: { ...state, waiters: [...state.waiters, event.reply] }, commands: [] }
if (event.type === "connection.requested")
return { state: { ...state, joiners: [...state.joiners, event.reply] }, commands: [] }
if (event.type === "waiter.cancelled") {
if (!state.waiters.includes(event.reply) && !state.joiners.includes(event.reply)) return unchanged(state)
return {
state: { ...state, waiters: without(state.waiters, event.reply), joiners: without(state.joiners, event.reply) },
commands: [],
}
}
if (event.type !== "connect.succeeded" && event.type !== "connect.failed") return unchanged(state)
if (event.attempt !== state.attempt) return unchanged(state)
if (event.type === "connect.succeeded")
return {
state: { status: "ready", attempt: state.attempt, data: event.data },
commands: [
...(same(state.data, event.data) ? [] : [{ type: "publish" as const, data: event.data }]),
...resolve([...state.waiters, ...state.joiners], event.data),
],
}
// A window that only asked for the connection still gets the last service, as it would have before the reconnect
// started; only the windows that asked to reconnect see the failure.
return {
state: { status: "failed", attempt: state.attempt, error: event.error, last: state.data },
commands: [...resolve(state.joiners, state.data), ...reject(state.waiters, event.error)],
}
}
function ready(state: Extract<State, { status: "ready" }>, event: Event): Transition {
if (event.type === "connection.requested")
return { state, commands: [{ type: "resolve", replies: [event.reply], data: state.data }] }
if (event.type === "reconnect.requested") {
const attempt = state.attempt + 1
// The new attempt also supersedes a verification still running for the current one.
return {
state: { status: "reconnecting", attempt, data: state.data, waiters: [event.reply], joiners: [] },
commands: [{ type: "connect", attempt, mode: "reconnect" }],
}
}
// `Service.ensure` replaced the probed service (it had exited or was incompatible): move every window to it.
if (event.type === "verify.succeeded" && event.attempt === state.attempt && !same(state.data, event.data))
return {
state: { status: "ready", attempt: state.attempt, data: event.data },
commands: [{ type: "publish", data: event.data }],
}
// A failed verification keeps the probed service, which answered the probe; a window that loses it reconnects.
return unchanged(state)
}
function resolve(replies: readonly Reply[], data: Data): Command[] {
return replies.length > 0 ? [{ type: "resolve", replies, data }] : []
}
function reject(replies: readonly Reply[], error: Error): Command[] {
return replies.length > 0 ? [{ type: "reject", replies, error }] : []
}
/** Every caller still waiting on an attempt. */
function pending(state: State) {
if (state.status === "connecting") return state.waiters
if (state.status === "reconnecting") return [...state.waiters, ...state.joiners]
return []
}
function without(replies: readonly Reply[], reply: Reply) {
return replies.filter((item) => item !== reply)
}
function same(a: Data, b: Data) {
return a.url === b.url && a.password === b.password
}
function unchanged(state: State): Transition {
return { state, commands: [] }
}
/**
* Runs the machine for the scope's lifetime. The initial connect starts at once, in the background: windows are
* created as soon as the main process is up, and their renderers load while the CLI service is resolved.
*/
export const make = Effect.fn("BackgroundServiceMachine.make")(function* (input: {
/** The compatible service the entry module found, if any. */
readonly probe: Effect.Effect<Data | undefined, unknown>
readonly ensure: (mode: ConnectMode) => Effect.Effect<Data, unknown>
/** Called when the active service changes. */
readonly publish: (data: Data) => void
}) {
const fibers = yield* FiberSet.make()
const run = yield* FiberSet.runtime(fibers)()
const initial = initialTransition()
const current = { state: initial.state }
const dispatch = (event: Event) => {
const next = transition(current.state, event)
current.state = next.state
next.commands.forEach(perform)
}
// A failure of `ensure` can be a defect (`DesktopCli.resolve` dies), so every exit reports back as an event. An
// interrupt only happens on disposal, which the machine already left.
const report = (effect: Effect.Effect<Event, unknown>, failed: (error: Error) => Event) =>
run(
effect.pipe(
Effect.exit,
Effect.flatMap((exit) => {
if (Exit.isSuccess(exit)) return Effect.sync(() => dispatch(exit.value))
if (Cause.hasInterruptsOnly(exit.cause)) return Effect.void
const error = Cause.squash(exit.cause)
const event = failed(error instanceof Error ? error : new Error(String(error)))
return Effect.logError("v2 CLI background service failed", { event }).pipe(
Effect.andThen(Effect.sync(() => dispatch(event))),
)
}),
),
)
const perform = (command: Command) => {
if (command.type === "connect") {
const attempt = command.attempt
const connect = (mode: ConnectMode) =>
input.ensure(mode).pipe(Effect.map((data): Event => ({ type: "connect.succeeded", attempt, data })))
return void report(
command.mode === "reconnect"
? connect("reconnect")
: input.probe.pipe(
Effect.orElseSucceed(() => undefined),
Effect.flatMap((data) =>
data ? Effect.succeed<Event>({ type: "probe.succeeded", attempt, data }) : connect("initial"),
),
),
(error) => ({ type: "connect.failed", attempt, error }),
)
}
if (command.type === "verify") {
const attempt = command.attempt
return void report(
input.ensure("initial").pipe(Effect.map((data): Event => ({ type: "verify.succeeded", attempt, data }))),
(error) => ({ type: "verify.failed", attempt, error }),
)
}
if (command.type === "publish") return input.publish(command.data)
if (command.type === "resolve") return command.replies.forEach((reply) => reply.resolve(command.data))
command.replies.forEach((reply) => reply.reject(command.error))
}
const request = (type: "connection.requested" | "reconnect.requested") =>
Effect.callback<Data>((resume) => {
const reply: Reply = {
resolve: (data) => resume(Effect.succeed(data)),
reject: (error) => resume(Effect.die(error)),
}
dispatch({ type, reply })
// A window that closes or aborts its reconnect leaves the attempt, which keeps running for the others.
return Effect.sync(() => dispatch({ type: "waiter.cancelled", reply }))
})
yield* Effect.addFinalizer(() => Effect.sync(() => dispatch({ type: "disposed" })))
initial.commands.forEach(perform)
return {
connection: request("connection.requested"),
reconnect: request("reconnect.requested"),
}
})
@@ -1,65 +0,0 @@
import { expect, test } from "bun:test"
import { Deferred, Effect, Fiber } from "effect"
import { BackgroundServiceState } from "./background-service-state"
test("new consumers receive the latest reconnected service", async () => {
const initial = { url: "http://127.0.0.1:4100", password: "first" }
const replacement = { url: "http://127.0.0.1:4200", password: "second" }
const service = await Effect.runPromise(
Effect.scoped(
Effect.gen(function* () {
const state = yield* BackgroundServiceState.make({
initial: Effect.succeed(initial),
reconnect: Effect.succeed(replacement),
})
expect(yield* state.connection).toEqual(initial)
expect(yield* state.reconnect).toEqual(replacement)
return yield* state.connection
}),
),
)
expect(service).toEqual(replacement)
})
test("the state is ready before the initial connect finishes and consumers wait for it", async () => {
const initial = { url: "http://127.0.0.1:4100", password: "first" }
await Effect.runPromise(
Effect.scoped(
Effect.gen(function* () {
const gate = yield* Deferred.make<void>()
const state = yield* BackgroundServiceState.make({
initial: Deferred.await(gate).pipe(Effect.as(initial)),
reconnect: Effect.succeed(initial),
})
const consumer = yield* Effect.forkChild(state.connection)
yield* Effect.sleep("10 millis")
expect(consumer.pollUnsafe()).toBeUndefined()
yield* Deferred.succeed(gate, undefined)
expect(yield* Fiber.join(consumer)).toEqual(initial)
}),
),
)
})
test("a failed initial connect fails consumers instead of hanging them", async () => {
const result = await Effect.runPromise(
Effect.scoped(
Effect.gen(function* () {
const state = yield* BackgroundServiceState.make({
initial: Effect.fail(new Error("service unavailable")),
reconnect: Effect.succeed({ url: "http://127.0.0.1:4100", password: "later" }),
})
return yield* state.connection.pipe(Effect.exit)
}),
),
)
expect(result._tag).toBe("Failure")
})
@@ -1,31 +0,0 @@
export * as BackgroundServiceState from "./background-service-state"
import { Deferred, Effect, Exit, Ref } from "effect"
import type { SidecarCredentials } from "./sidecar-credentials"
// The initial connect runs in the background: windows are created as soon as the main process
// is up, and the renderer loads its bundle while the CLI service is being resolved. Consumers
// await the first outcome through `connection`, so nothing observes the service before it exists.
export const make = Effect.fn("BackgroundServiceState.make")(function* (options: {
readonly initial: Effect.Effect<SidecarCredentials.Data, unknown>
readonly reconnect: Effect.Effect<SidecarCredentials.Data>
}) {
// Every Exit is an Effect, so the latest resolution replays directly for each consumer.
const current = yield* Ref.make<Exit.Exit<SidecarCredentials.Data, unknown> | undefined>(undefined)
const first = yield* Deferred.make<void>()
yield* options.initial.pipe(
Effect.exit,
Effect.flatMap((exit) => Ref.set(current, exit)),
Effect.ensuring(Deferred.succeed(first, undefined)),
Effect.forkScoped,
)
return {
connection: Deferred.await(first).pipe(
Effect.flatMap(() => Ref.get(current)),
Effect.flatMap((exit) => exit ?? Exit.die(new Error("background service connect did not resolve"))),
Effect.orDie,
),
reconnect: options.reconnect.pipe(Effect.tap((next) => Ref.set(current, Exit.succeed(next)))),
}
})
@@ -1,12 +1,16 @@
import { app } from "electron"
import type { Endpoint } from "@opencode/client/service"
import { Context, Effect, FileSystem, Layer, Option, Path, Schema } from "effect"
import { SidecarChanged } from "../../shared/ipc-rpc/events"
import { CHANNEL } from "../constants"
import { BackgroundServiceState } from "./background-service-state"
import { emitIpcEvent } from "../ipc-events"
import { getMainWindows } from "../windows"
import { BackgroundServiceMachine } from "./background-service-machine"
import { cleanStages } from "./cli-stages"
import { keptStages, serviceVersion, shouldRetryEnsure } from "./cli-version"
import { DesktopCli } from "./desktop-cli"
import { SidecarCredentials } from "./sidecar-credentials"
import { sidecarProbe } from "./sidecar-probe"
import { takeSidecarProbe } from "./sidecar-probe"
export * as BackgroundService from "./background-service"
@@ -25,20 +29,78 @@ export const layer = Layer.effect(
const context = yield* Effect.context<FileSystem.FileSystem | Path.Path | DesktopCli.Service>()
return Service.of(
yield* BackgroundServiceState.make({
initial: connect("initial").pipe(Effect.provide(context)),
reconnect: connect("reconnect").pipe(Effect.provide(context), Effect.orDie),
yield* BackgroundServiceMachine.make({
probe: probe().pipe(Effect.provide(context)),
ensure: (mode) => ensure(mode).pipe(Effect.provide(context)),
publish: (data) => {
SidecarCredentials.set(data)
// Every window moves its requests and event stream to the new origin, not only the one that reconnected.
const event = new SidecarChanged({ url: data.url })
getMainWindows().forEach((win) => emitIpcEvent(win.webContents, event))
},
}),
)
}),
)
const connect = Effect.fn("BackgroundService.connect")(function* (mode: "initial" | "reconnect") {
// A compatible service the entry module already found is adopted at once; the machine still verifies it with
// `ensure("initial")` (which also completes a terminal handoff), off the renderer's path.
const probe = Effect.fn("BackgroundService.probe")(function* () {
const service = yield* Effect.promise(takeSidecarProbe)
if (!service) return undefined
const setup = yield* prepare("initial")
return (yield* adopt(setup, service, true)).credentials
})
const ensure = Effect.fn("BackgroundService.ensure")(function* (mode: BackgroundServiceMachine.ConnectMode) {
const runFork = Effect.runForkWith(yield* Effect.context<FileSystem.FileSystem | Path.Path>())
const setup = yield* prepare(mode)
const run = (version = setup.version) =>
setup.client.Service.ensure({
file: setup.file,
version,
// A fixed port makes a second contender fail to bind and back off; port 0 never collides, so two
// services could boot against the same database.
command: [
...setup.cli.command,
"serve",
"--service",
...(setup.isolated ? ["--hostname", "0.0.0.0", "--port", String(0x0c0c)] : []),
],
onStart: (reason, previousVersion) =>
runFork(Effect.logInfo("v2 CLI background service starting", { reason, previousVersion })),
})
const service = yield* Effect.tryPromise(() =>
run().catch((error) => {
if (!shouldRetryEnsure(setup.version, error instanceof Error ? error.message : String(error))) throw error
return run(setup.cli.version)
}),
)
const ready = yield* adopt(setup, service, false)
// Only packaged and isolated builds run a staged copy; any other binary sits in the source tree.
// Keep both the bundled stage and the running server's stage so its executable stays on disk. This runs only
// once `Service.ensure` settled which server runs, never for a probed one it may still replace.
if (mode === "initial" && (app.isPackaged || setup.isolated) && setup.cli.binary && ready.runningVersion)
runFork(
cleanStages(setup.cli.binary, { keep: keptStages(setup.cli.version, ready.runningVersion) }).pipe(
Effect.catch((error) => Effect.logError("failed to clean staged v2 CLIs", { error })),
),
)
return ready.credentials
})
const prepare = Effect.fn("BackgroundService.prepare")(function* (mode: BackgroundServiceMachine.ConnectMode) {
yield* Effect.logInfo("starting v2 background service")
const fs = yield* FileSystem.FileSystem
const path = yield* Path.Path
const desktopCli = yield* DesktopCli.Service
const runFork = Effect.runForkWith(yield* Effect.context<FileSystem.FileSystem | Path.Path>())
const isolated = !app.isPackaged && process.env.OPENCODE_DESKTOP_ISOLATED_SERVER === "1"
const cli = yield* desktopCli.resolve
const version = serviceVersion({ channel: CHANNEL, isolated, mode, version: cli.version })
@@ -55,42 +117,22 @@ const connect = Effect.fn("BackgroundService.connect")(function* (mode: "initial
)
: undefined
const ensure = (required = version) =>
client.Service.ensure({
file,
version: required,
// A fixed port makes a second contender fail to bind and back off; port 0 never collides, so two
// services could boot against the same database.
command: [
...cli.command,
"serve",
"--service",
...(isolated ? ["--hostname", "0.0.0.0", "--port", String(0x0c0c)] : []),
],
onStart: (reason, previousVersion) =>
runFork(Effect.logInfo("v2 CLI background service starting", { reason, previousVersion })),
})
return { cli, client, file, isolated, version }
})
const runEnsure = () =>
ensure().catch((error) => {
if (!shouldRetryEnsure(version, error instanceof Error ? error.message : String(error))) throw error
return ensure(cli.version)
})
// A compatible service the entry module already found is adopted at once; ensure() still runs
// afterwards for its side effects (terminal handoff completion), off the renderer's path.
const early = mode === "initial" && !isolated ? yield* Effect.promise(sidecarProbe) : undefined
if (early) yield* Effect.sync(() => void runEnsure().catch(() => undefined))
const service = early ?? (yield* Effect.tryPromise(runEnsure))
const adopt = Effect.fn("BackgroundService.adopt")(function* (
setup: Effect.Success<ReturnType<typeof prepare>>,
service: Endpoint,
probed: boolean,
) {
const fs = yield* FileSystem.FileSystem
if (service.auth?.type !== "basic") throw new Error("V2 CLI background service did not provide authentication")
const url = new URL(service.url)
if (url.hostname === "0.0.0.0") url.hostname = "127.0.0.1"
const runningVersion = yield* fs.readFileString(file ?? client.fallback()).pipe(
const runningVersion = yield* fs.readFileString(setup.file ?? setup.client.fallback()).pipe(
Effect.map(
(text) =>
Option.getOrUndefined(Schema.decodeUnknownOption(Schema.fromJsonString(RegisteredService))(text))?.version,
@@ -99,25 +141,16 @@ const connect = Effect.fn("BackgroundService.connect")(function* (mode: "initial
)
yield* Effect.logInfo("v2 CLI background service ready", {
version,
version: setup.version,
runningVersion,
probed: !!early,
probed,
...endpoint(url.origin),
})
// Only packaged and isolated builds run a staged copy; any other binary sits in the source tree.
// Keep both the bundled stage and any adopted server's stage so its executable stays on disk.
if (mode === "initial" && (app.isPackaged || isolated) && cli.binary && runningVersion)
runFork(
cleanStages(cli.binary, { keep: keptStages(cli.version, runningVersion) }).pipe(
Effect.catch((error) => Effect.logError("failed to clean staged v2 CLIs", { error })),
),
)
const ready = { url: url.origin, password: service.auth.password } satisfies SidecarCredentials.Data
SidecarCredentials.set(ready)
return ready
return {
credentials: { url: url.origin, password: service.auth.password } satisfies SidecarCredentials.Data,
runningVersion,
}
})
function endpoint(url: string | undefined) {
@@ -16,13 +16,15 @@ export function startSidecarProbe() {
const version = CHANNEL === "beta" ? undefined : bundledVersion()
if (version === "") return
probe = import("@opencode/client/service")
.then(({ Service }) => Service.discover({ version }))
.catch(() => undefined)
probe = import("@opencode/client/service").then(({ Service }) => Service.discover({ version })).catch(() => undefined)
}
export function sidecarProbe() {
return probe ?? Promise.resolve(undefined)
/** The probe's result, once: a later connect must not adopt a service that may have exited since boot. */
export function takeSidecarProbe() {
const result = probe ?? Promise.resolve(undefined)
probe = undefined
return result
}
function bundledVersion() {
+2 -1
View File
@@ -14,7 +14,8 @@ import type {
export type ElectronAPI = {
awaitInitialization(): Promise<ServerReadyData>
reconnectService(): Promise<ServerReadyData>
reconnectService(signal?: AbortSignal): Promise<ServerReadyData>
onSidecarChanged(cb: (data: ServerReadyData) => void): () => void
readyDeepLinks(): void
acknowledgeDeepLinks(attempt: number): void
isFirstLaunchOnboardingPending(): Promise<boolean>
+3 -2
View File
@@ -1,5 +1,5 @@
import type { ElectronAPI } from "./api-types"
import { invoke, listen, send } from "./ipc-client"
import { cancellable, invoke, listen, send } from "./ipc-client"
type Mutable<Value> =
Value extends ReadonlyArray<unknown>
@@ -22,7 +22,8 @@ const seeded = window.electron.storageSnapshot.then((snapshot) => new Map(Object
export const api: ElectronAPI = {
awaitInitialization: () => invoke("AppAwaitInitialization"),
reconnectService: () => invoke("AppReconnectService"),
reconnectService: (signal) => cancellable("AppReconnectService", undefined, signal),
onSidecarChanged: (cb) => listen("SidecarChanged", (event) => cb({ url: event.url })),
readyDeepLinks: () => send("AppReadyDeepLinks"),
acknowledgeDeepLinks: (attempt) => send("AppAcknowledgeDeepLinks", { attempt }),
isFirstLaunchOnboardingPending: () => invoke("AppIsFirstLaunchOnboardingPending"),
+39 -17
View File
@@ -16,8 +16,9 @@ import {
} from "@opencode/app/desktop"
import { useTheme } from "@opencode/ui/theme/context"
import type { BaseRouterProps } from "@solidjs/router"
import { createEffect, createMemo, createResource, lazy, Show, Suspense } from "solid-js"
import { createEffect, createMemo, createResource, createSignal, lazy, onCleanup, Show, Suspense } from "solid-js"
import { createStore } from "solid-js/store"
import type { ServerReadyData } from "../shared/ipc-contract"
import type { ElectronAPI } from "./api-types"
import { DesktopFirstLaunchOnboarding } from "./onboarding"
import { createDesktopPlatform } from "./platform"
@@ -28,6 +29,8 @@ import { LoadingSplash } from "./startup/splash"
import { getLastActiveUrl } from "./window/route-storage"
import { DesktopMemoryRouter } from "./window/router"
const offline = { url: "http://127.0.0.1:0" }
const MigrationStatus = lazy(() => import("./migration-status").then((module) => ({ default: module.MigrationStatus })))
export function DesktopApp(props: { api: ElectronAPI; version: string }) {
@@ -60,7 +63,15 @@ export function DesktopApp(props: { api: ElectronAPI; version: string }) {
)
const platform = createDesktopPlatform(props.api, windowState)
const [sidecar, { mutate: setSidecar }] = createResource(() => props.api.awaitInitialization())
const [sidecar] = createResource(() => props.api.awaitInitialization())
// The origin main moved every window to after initialization: this window's reconnect, another's, or a probed
// service that `Service.ensure` replaced.
const [movedSidecar, setMovedSidecar] = createSignal<ServerReadyData | undefined>(undefined, {
equals: (a, b) => a?.url === b?.url,
})
onCleanup(props.api.onSidecarChanged((data) => setMovedSidecar(data)))
const [locale] = createResource(() => preloadStoredLocale(platform))
@@ -85,23 +96,34 @@ export function DesktopApp(props: { api: ElectronAPI; version: string }) {
const pluginServers = usePluginServers()
const ready = createMemo(() => !firstLaunch.loading && !sidecar.loading && !locale.loading && pluginServers.ready())
const servers = createMemo(() => {
const data = initializationData(sidecar)
const list: ServerConnection.Any[] = []
const activeSidecar = createMemo(
() => {
const data = movedSidecar() ?? initializationData(sidecar)
if (data) {
list.push({
type: "sidecar",
variant: "base",
http: sidecarHttp(data),
reconnect: createSidecarResolver({ api: props.api, current: sidecar, update: setSidecar }),
})
}
return data ? sidecarHttp(data) : undefined
},
undefined,
{ equals: (a, b) => a?.url === b?.url },
)
list.push(...pluginServers.list())
// One connection for the window's lifetime: the server SDK follows its `http` to a new origin in place.
const sidecarServer: ServerConnection.Sidecar = {
type: "sidecar",
variant: "base",
get http() {
return activeSidecar() ?? offline
},
reconnect: createSidecarResolver({
api: props.api,
current: () => movedSidecar() ?? (sidecar.state === "ready" ? sidecar.latest : undefined),
update: (data) => setMovedSidecar(data),
}),
}
return list
})
const servers = createMemo((): ServerConnection.Any[] => [
...(activeSidecar() ? [sidecarServer] : []),
...pluginServers.list(),
])
return (
<Show when={ready()}>
@@ -119,7 +141,7 @@ export function DesktopApp(props: { api: ElectronAPI; version: string }) {
/>
<DesktopEffects api={props.api} />
<Suspense fallback={null}>
<Show when={initializationData(sidecar)} keyed>
<Show when={activeSidecar()} keyed>
{(server) => <MigrationStatus server={server} />}
</Show>
</Suspense>
@@ -1,7 +1,7 @@
import { describe, expect, test } from "bun:test"
import { createSidecarResolver, initializationData } from "./initialization"
function failure(error: unknown) {
function failure(error: Error | string) {
try {
initializationData(Object.assign(() => undefined, { error }))
} catch (caught) {
@@ -53,13 +53,20 @@ describe("desktop renderer initialization", () => {
expect(updates).toEqual([])
})
test("does not publish a sidecar resolved after cancellation", async () => {
test("does not publish a sidecar resolved after cancellation, and cancels the wait in main", async () => {
const sidecar = { url: "http://127.0.0.1:4321" }
const pending = Promise.withResolvers<typeof sidecar>()
const updates: (typeof sidecar)[] = []
const signals: (AbortSignal | undefined)[] = []
const resolve = createSidecarResolver({
api: { reconnectService: () => pending.promise },
api: {
reconnectService: (signal) => {
signals.push(signal)
return pending.promise
},
},
current: () => undefined,
update: (next) => updates.push(next),
})
@@ -71,5 +78,6 @@ describe("desktop renderer initialization", () => {
await expect(result).rejects.toBe(abort.signal.reason)
expect(updates).toEqual([])
expect(signals).toEqual([abort.signal])
})
})
@@ -3,9 +3,11 @@ import type { ElectronAPI } from "../api-types"
type SidecarData = Awaited<ReturnType<ElectronAPI["awaitInitialization"]>>
export function initializationData<A>(state: (() => A | undefined) & { error: unknown }) {
if (state.error !== undefined) throw markLocalServerStartup(state.error)
if (state.error === undefined) return state()
const failure = state.error instanceof Error ? state.error : new Error(String(state.error))
Object.defineProperty(failure, "localServerStartup", { value: true })
return state()
throw failure
}
// The main process adds Authorization to sidecar requests (`wireRendererHeaders`); the renderer never
@@ -21,7 +23,8 @@ export function createSidecarResolver(input: {
}) {
return async (signal: AbortSignal) => {
if (signal.aborted) throw signal.reason
const next = await input.api.reconnectService()
// Aborting interrupts this window's wait in main; the reconnect keeps running for other windows.
const next = await input.api.reconnectService(signal)
if (signal.aborted) throw signal.reason
@@ -34,10 +37,3 @@ export function createSidecarResolver(input: {
function sameSidecar(current: SidecarData | undefined, next: SidecarData) {
return current?.url === next.url
}
function markLocalServerStartup(error: unknown) {
const failure = error instanceof Error ? error : new Error(String(error))
Object.defineProperty(failure, "localServerStartup", { value: true })
return failure
}
@@ -71,8 +71,14 @@ export class PluginMenubarItemsChanged extends Schema.TaggedClass<PluginMenubarI
{ items: Schema.Array(PluginMenubarItem) },
) {}
// Main adopted another background service origin; every window moves its sidecar requests to it.
export class SidecarChanged extends Schema.TaggedClass<SidecarChanged>()("SidecarChanged", {
url: Schema.String,
}) {}
export const DesktopEvent = Schema.Union([
DeepLinksOpened,
SidecarChanged,
MenuCommandTriggered,
WindowFullscreenChanged,
WindowFocusChanged,