Skip to content
Closed
Show file tree
Hide file tree
Changes from 2 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