Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions apps/desktop/src/backend/DesktopBackendManager.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1585,6 +1585,7 @@ describe("DesktopBackendManager", () => {
list: Effect.succeed([instance1, instance2]),
get: () => Effect.succeedNone,
primary: Effect.die(new Error("primary not implemented")),
changes: Stream.empty,
register: () => Effect.die(new Error("register not implemented")),
unregister: () => Effect.die(new Error("unregister not implemented")),
});
Expand Down
59 changes: 34 additions & 25 deletions apps/desktop/src/backend/DesktopBackendManager.ts
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ import * as Fiber from "effect/Fiber";
import * as FileSystem from "effect/FileSystem";
import * as Option from "effect/Option";
import * as PlatformError from "effect/PlatformError";
import * as Ref from "effect/Ref";
import * as SubscriptionRef from "effect/SubscriptionRef";
import * as Schedule from "effect/Schedule";
import * as Schema from "effect/Schema";
import * as Semaphore from "effect/Semaphore";
Expand Down Expand Up @@ -276,6 +276,10 @@ export interface DesktopBackendInstance {
readonly stop: (options?: { readonly timeout?: Duration.Duration }) => Effect.Effect<void>;
readonly currentConfig: Effect.Effect<Option.Option<DesktopBackendStartConfig>>;
readonly snapshot: Effect.Effect<DesktopBackendSnapshot>;
// Emits once on subscribe and again after every lifecycle state write
// (start, ready, exit, restart scheduling). Writes that leave the observable
// config and snapshot unchanged still emit, so consumers dedupe.
readonly changes: Stream.Stream<void>;
// Polls desiredRunning + the instance's own ready flag until the
// backend reports ready, or the timeout elapses. Returns true on
// ready, false on timeout. Used by the WSL backend swap to drive its
Expand Down Expand Up @@ -696,16 +700,16 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
const wslEnvironment = yield* DesktopWslEnvironment.DesktopWslEnvironment;
const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
const httpClient = yield* HttpClient.HttpClient;
const state = yield* Ref.make(initialState);
const state = yield* SubscriptionRef.make(initialState);
const mutex = yield* Semaphore.make(1);

const { logWarning: logInstanceWarning, logError: logInstanceError } =
DesktopObservability.makeComponentLogger(`desktop-backend-instance:${spec.id}`);

const updateActiveRun = (runId: number, f: (run: ActiveBackendRun) => ActiveBackendRun) =>
Ref.update(state, withActiveRun(runId, f));
SubscriptionRef.update(state, withActiveRun(runId, f));

const snapshot = Ref.get(state).pipe(
const snapshot = SubscriptionRef.get(state).pipe(
Effect.map((current): DesktopBackendSnapshot => ({
desiredRunning: current.desiredRunning,
ready: current.ready,
Expand All @@ -714,10 +718,11 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
restartScheduled: Option.isSome(current.restartFiber),
})),
);
const currentConfig = Ref.get(state).pipe(Effect.map((current) => current.config));
const currentConfig = SubscriptionRef.get(state).pipe(Effect.map((current) => current.config));
const changes = SubscriptionRef.changes(state).pipe(Stream.map(() => undefined));

const cancelRestart = Effect.gen(function* () {
const restartFiber = yield* Ref.modify(state, (current) => [
const restartFiber = yield* SubscriptionRef.modify(state, (current) => [
current.restartFiber,
{
...current,
Expand All @@ -734,10 +739,10 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
const start: Effect.Effect<void> = Effect.suspend(() =>
mutex.withPermits(1)(
Effect.gen(function* () {
const current = yield* Ref.get(state);
const current = yield* SubscriptionRef.get(state);
if (Option.isSome(current.active)) {
if (!current.desiredRunning) {
yield* Ref.update(state, (latest) => ({
yield* SubscriptionRef.update(state, (latest) => ({
...latest,
desiredRunning: true,
}));
Expand All @@ -747,7 +752,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (

if (current.ready) {
yield* spec.onShutdown?.() ?? Effect.void;
yield* Ref.update(state, (latest) =>
yield* SubscriptionRef.update(state, (latest) =>
latest.ready ? { ...latest, ready: false } : latest,
);
}
Expand All @@ -772,7 +777,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
const resetFatalPreflightCounter =
!current.desiredRunning && current.preflightFailureAttempt > 0;
yield* cancelRestart;
yield* Ref.update(state, (latest) => ({
yield* SubscriptionRef.update(state, (latest) => ({
...latest,
desiredRunning: true,
ready: false,
Expand All @@ -787,7 +792,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
// Transient (WSL cold-starting, wslpath while the VM boots). Keep
// retrying so the backend self-heals once WSL is ready. Reset a
// prior bounded/fatal streak because this is a different failure.
yield* Ref.update(state, (latest) =>
yield* SubscriptionRef.update(state, (latest) =>
latest.preflightFailureAttempt === 0
? latest
: { ...latest, preflightFailureAttempt: 0 },
Expand All @@ -796,7 +801,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
return;
}
const attemptLimit = retryLimit ?? MAX_PREFLIGHT_FAILURE_ATTEMPTS;
const attempt = yield* Ref.modify(state, (latest) => {
const attempt = yield* SubscriptionRef.modify(state, (latest) => {
const next = latest.preflightFailureAttempt + 1;
return [next, { ...latest, preflightFailureAttempt: next }] as const;
});
Expand All @@ -808,7 +813,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
reason,
attempt,
});
yield* Ref.update(state, (latest) => ({
yield* SubscriptionRef.update(state, (latest) => ({
...latest,
desiredRunning: false,
ready: false,
Expand All @@ -830,7 +835,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
if (shouldRestart) {
yield* scheduleRestart(reason);
} else {
yield* Ref.update(state, (latest) => ({
yield* SubscriptionRef.update(state, (latest) => ({
...latest,
desiredRunning: false,
ready: false,
Expand All @@ -843,7 +848,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
}
// Clean preflight — reset the fatal counter so a later failure gets a
// fresh allowance.
yield* Ref.update(state, (latest) =>
yield* SubscriptionRef.update(state, (latest) =>
latest.preflightFailureAttempt === 0 ? latest : { ...latest, preflightFailureAttempt: 0 },
);

Expand All @@ -853,7 +858,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
}

const runScope = yield* Scope.make("sequential");
const runId = yield* Ref.modify(state, (latest) => [
const runId = yield* SubscriptionRef.modify(state, (latest) => [
latest.nextRunId,
{
...latest,
Expand All @@ -877,7 +882,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
yield* mutex.withPermits(1)(
Effect.gen(function* () {
const { isCurrentRun, nextState, pid, exitObserved, stopRequested, wasReady } =
yield* Ref.modify(
yield* SubscriptionRef.modify(
state,
(
latest,
Expand Down Expand Up @@ -944,7 +949,10 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (

if (isCurrentRun && nextState.desiredRunning) {
if (stateDirOwned && !stopRequested) {
yield* Ref.update(state, (latest) => ({ ...latest, desiredRunning: false }));
yield* SubscriptionRef.update(state, (latest) => ({
...latest,
desiredRunning: false,
}));
yield* (spec.onStateDirOwned?.() ?? Effect.void).pipe(Effect.ignore);
return;
}
Expand Down Expand Up @@ -977,7 +985,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
exitObserved: true,
})),
onReady: Effect.fn("desktop.backendInstance.onReady")(function* () {
const isCurrentRun = yield* Ref.modify(state, (latest) => {
const isCurrentRun = yield* SubscriptionRef.modify(state, (latest) => {
const activeRun = Option.getOrUndefined(latest.active);
if (activeRun?.id !== runId) {
return [false, latest] as const;
Expand Down Expand Up @@ -1041,7 +1049,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
const scheduleRestart = Effect.fn("desktop.backendInstance.scheduleRestart")(function* (
reason: string,
) {
const scheduled = yield* Ref.modify(state, (latest) => {
const scheduled = yield* SubscriptionRef.modify(state, (latest) => {
if (!latest.desiredRunning || Option.isSome(latest.restartFiber)) {
return [Option.none<Duration.Duration>(), latest] as const;
}
Expand All @@ -1066,7 +1074,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
const restartFiber = yield* Effect.forkIn(
Effect.sleep(delay).pipe(
Effect.andThen(
Ref.modify(state, (latest) => {
SubscriptionRef.modify(state, (latest) => {
const shouldRestart = latest.desiredRunning;
return [
shouldRestart,
Expand All @@ -1086,7 +1094,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
),
parentScope,
);
yield* Ref.update(state, (latest) =>
yield* SubscriptionRef.update(state, (latest) =>
Option.isNone(latest.restartFiber)
? {
...latest,
Expand All @@ -1103,7 +1111,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
}) {
const { active, restartFiber, notifyShutdown } = yield* mutex.withPermits(1)(
Effect.gen(function* () {
const result = yield* Ref.modify(state, (latest) => {
const result = yield* SubscriptionRef.modify(state, (latest) => {
const active = Option.map(latest.active, (run) =>
run.exitObserved ? run : { ...run, stopRequested: true },
);
Expand Down Expand Up @@ -1142,7 +1150,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
return;
}
const cleanup = yield* mutex.withPermits(1)(
Ref.modify(
SubscriptionRef.modify(
state,
(
latest,
Expand Down Expand Up @@ -1192,7 +1200,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (

const waitForReady = (timeout: Duration.Duration): Effect.Effect<boolean> =>
Effect.gen(function* () {
const current = yield* Ref.get(state);
const current = yield* SubscriptionRef.get(state);
// Return false early if an external `stop()` flipped desiredRunning off
// — no point polling for a backend that is being torn down.
if (!current.desiredRunning) return { done: true, ready: false };
Expand All @@ -1216,6 +1224,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
stop,
currentConfig,
snapshot,
changes,
waitForReady,
} satisfies DesktopBackendInstance;
});
43 changes: 43 additions & 0 deletions apps/desktop/src/backend/DesktopBackendPool.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import * as Effect from "effect/Effect";
import * as FileSystem from "effect/FileSystem";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Queue from "effect/Queue";
import * as Ref from "effect/Ref";
import * as Stream from "effect/Stream";
import { HttpClient } from "effect/http";
Expand Down Expand Up @@ -39,6 +40,7 @@ function makeStubInstance(
stop: () => Effect.void,
currentConfig: Effect.succeed(Option.none<DesktopBackendStartConfig>()),
snapshot: Effect.succeed(snapshot),
changes: Stream.empty,
waitForReady: (_timeout: Duration.Duration) => Effect.succeed(false),
};
}
Expand Down Expand Up @@ -158,4 +160,45 @@ describe("DesktopBackendPool", () => {
}),
),
);

it.effect("reports registry changes so the renderer topology can be re-read", () =>
Effect.scoped(
Effect.gen(function* () {
const pool = yield* DesktopBackendPool.DesktopBackendPool.pipe(
Effect.provide(layerPool(yield* Ref.make("Windows"))),
);
const wslId = DesktopBackendPool.BackendInstanceId("wsl:ubuntu");
// Each emission records the registry as observed right after it, so
// only an emission caused by the change can satisfy the next wait.
const observed = yield* Queue.unbounded<ReadonlyArray<string>>();
yield* pool.changes.pipe(
Stream.runForEach(() =>
pool.list.pipe(
Effect.flatMap((instances) =>
Queue.offer(
observed,
instances.map((instance) => instance.id),
),
),
),
),
Effect.forkScoped,
);
const awaitRegistry = (matches: (ids: ReadonlyArray<string>) => boolean) =>
Effect.gen(function* () {
while (!matches(yield* Queue.take(observed))) {}
});

yield* awaitRegistry((ids) => ids.length === 1);
yield* pool.register({
id: wslId,
label: Effect.succeed("WSL (Ubuntu)"),
configResolve: Effect.die("unexpected WSL config resolve"),
});
yield* awaitRegistry((ids) => ids.includes(wslId));
yield* pool.unregister(wslId);
yield* awaitRegistry((ids) => !ids.includes(wslId));
}),
),
);
});
Loading
Loading