diff --git a/apps/server/src/device/DeviceHost.ts b/apps/server/src/device/DeviceHost.ts index eafa3b6319de3..326f2eb5ac86f 100644 --- a/apps/server/src/device/DeviceHost.ts +++ b/apps/server/src/device/DeviceHost.ts @@ -198,12 +198,19 @@ export class DeviceHost extends Context.Service< ) => Effect.Effect; readonly ensureReady: ( onPhase: (phase: "installing" | "starting", detail?: string) => Effect.Effect, - ) => Effect.Effect; + ) => Effect.Effect< + DeviceHostReady, + DeviceHostError | DeviceHostUnavailableError | NodeRuntimeUnavailableError + >; readonly ensureAgentReady: ( onPhase: (phase: "installing" | "starting", detail?: string) => Effect.Effect, ) => Effect.Effect< DeviceHostAgentReady, - DeviceHostError | DeviceHostTimeoutError | AgentDeviceStartError | NodeRuntimeUnavailableError + | DeviceHostError + | DeviceHostUnavailableError + | DeviceHostTimeoutError + | AgentDeviceStartError + | NodeRuntimeUnavailableError >; readonly current: Effect.Effect; readonly stopAgent: Effect.Effect; diff --git a/apps/server/src/device/DeviceMultiHost.test.ts b/apps/server/src/device/DeviceMultiHost.test.ts index adf01a5cda5d6..7491155c203a7 100644 --- a/apps/server/src/device/DeviceMultiHost.test.ts +++ b/apps/server/src/device/DeviceMultiHost.test.ts @@ -1,5 +1,10 @@ import { expect, it } from "@effect/vitest"; -import { ThreadId } from "@t3tools/contracts"; +import { + DeviceHostUnavailableError, + LOCAL_DEVICE_HOST_ID, + ThreadId, + type DeviceHostSummary, +} from "@t3tools/contracts"; import * as Deferred from "effect/Deferred"; import * as Fiber from "effect/Fiber"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; @@ -145,6 +150,17 @@ it.effect("keeps hosts independent when serials collide and another host fails", expect((yield* service.state).sessions).toEqual([]); yield* service.agentReadinessIfSupported("a"); expect((yield* service.state).hostStatuses.a?.status).toBe("ready"); + const unavailable = new DeviceHostUnavailableError({ + hostId: "a", + reason: "Host configuration changed. Retry the operation.", + }); + hosts.set("a", { + ...host("a"), + ensureReady: () => Effect.fail(unavailable), + ensureAgentReady: () => Effect.fail(unavailable), + }); + expect(yield* service.readiness("a").pipe(Effect.flip)).toBe(unavailable); + expect(yield* service.agentReadinessIfSupported("a").pipe(Effect.flip)).toBe(unavailable); yield* service.configure({ enabled: false }); expect((yield* service.state).hostStatuses).toEqual({}); }).pipe( @@ -156,3 +172,331 @@ it.effect("keeps hosts independent when serials collide and another host fails", ), ), ); + +it.effect("starts other hosts while one host is still starting", () => + Effect.gen(function* () { + const slowHost = yield* pausedHost("ensureReady", "slow"); + const fastHost = yield* pausedHost("ensureAgentReady", "fast"); + const hosts = new Map([slowHost.host, fastHost.host].map((host) => [host.id, host])); + const service = yield* DeviceService.makeWithHosts(hosts).pipe( + Effect.provide(NodeCrypto.layer), + ); + const slow = yield* service.readiness("slow").pipe(Effect.flip, Effect.forkChild); + yield* Deferred.await(slowHost.started); + expect((yield* service.readiness("fast")).hostId).toBe("fast"); + expect((yield* service.state).hostStatuses.fast?.status).toBe("ready"); + yield* service.configure({ enabled: false }); + yield* Deferred.succeed(slowHost.finish, undefined); + const error = yield* Fiber.join(slow); + expect(error._tag === "DeviceHostUnavailableError" && error.reason).toBe( + "Device support was turned off while the host was starting.", + ); + expect(slowHost.calls).toEqual(["stop", "stop"]); + expect(fastHost.calls).toEqual(["stop"]); + expect((yield* service.state).hostStatuses).toEqual({}); + }).pipe(Effect.provide(ServerSettings.layerTest({ enableDeviceSupport: true }))), +); + +it.effect("reads host summaries outside the lifecycle lock when agent tools start", () => + Effect.gen(function* () { + const summaryEntered = yield* Deferred.make(); + const finishSummary = yield* Deferred.make(); + let pauseSummary = false; + const { host: agentHost } = yield* pausedHost("ensureReady", "agent"); + const { host: otherHost } = yield* pausedHost("ensureAgentReady", "other"); + const summary = Effect.suspend(() => + pauseSummary + ? Deferred.succeed(summaryEntered, undefined).pipe( + Effect.andThen(Deferred.await(finishSummary)), + Effect.andThen(otherHost.summary), + ) + : otherHost.summary, + ); + const hosts = new Map([agentHost, { ...otherHost, summary }].map((host) => [host.id, host])); + const service = yield* DeviceService.makeWithHosts(hosts).pipe( + Effect.provide(NodeCrypto.layer), + ); + pauseSummary = true; + const agent = yield* service.agentReadinessIfSupported("agent").pipe(Effect.forkChild); + yield* Deferred.await(summaryEntered); + expect((yield* service.readiness("other")).hostId).toBe("other"); + yield* service.withLifecycleLock( + Effect.sync(() => hosts.delete("other")).pipe(Effect.andThen(service.refreshHosts)), + ); + yield* Deferred.succeed(finishSummary, undefined); + expect((yield* Fiber.join(agent))?.hostId).toBe("agent"); + expect((yield* service.state).hosts.map((summary) => summary.id)).toEqual(["agent"]); + }).pipe( + Effect.provide( + ServerSettings.layerTest({ enableDeviceSupport: true, enableAgentDeviceAccess: true }), + ), + ), +); + +it.effect("stops a start that fails after device support was turned off before it began", () => + Effect.gen(function* () { + const summaryEntered = yield* Deferred.make(); + const finishSummary = yield* Deferred.make(); + let pauseSummary = false; + let running = false; + const stops: string[] = []; + const summary: DeviceHostSummary = { + id: LOCAL_DEVICE_HOST_ID, + label: "Local", + kind: "local", + hubInstalled: true, + agentDeviceInstalled: false, + platforms: [{ platform: "android", available: true }], + }; + const run = () => Effect.succeed({ stdout: "", stderr: "", code: 0 }); + const ready: DeviceHost.DeviceHostReady = { + nodePath: process.execPath, + hub: { origin: "http://local" }, + devices: makeDeviceHubOperations({ + hostId: LOCAL_DEVICE_HOST_ID, + origin: "http://local", + http: HttpClient.make(() => Effect.die("unused")), + run, + platformAvailable: () => Effect.succeed(true), + }), + run, + helpers: { serveSimAxSettings: null, serveSimCli: null }, + }; + const host: DeviceHost.DeviceHost["Service"] = { + id: LOCAL_DEVICE_HOST_ID, + label: "Local", + summary: Effect.suspend(() => + pauseSummary + ? Deferred.succeed(summaryEntered, undefined).pipe( + Effect.andThen(Deferred.await(finishSummary)), + Effect.as(summary), + ) + : Effect.succeed(summary), + ), + platformAvailability: (platform) => Effect.succeed({ platform, available: true }), + ensureReady: () => Effect.succeed(ready), + // Like LocalDeviceHost: the hub is up before the agent tools install fails. + ensureAgentReady: (onPhase) => + Effect.sync(() => { + running = true; + }).pipe( + Effect.andThen(onPhase("installing")), + Effect.tap(() => + service.state.pipe( + Effect.map((state) => { + expect(state.hostStatus).toBe("disabled"); + expect(state.hostStatuses).toEqual({}); + }), + ), + ), + Effect.andThen( + Effect.fail( + new DeviceHost.DeviceHostError({ + hostId: LOCAL_DEVICE_HOST_ID, + step: "installing agent tools", + cause: new Error("offline"), + }), + ), + ), + ), + current: Effect.sync(() => (running ? ready : null)), + stopAgent: Effect.void, + stop: Effect.sync(() => { + stops.push("stop"); + running = false; + }), + }; + const service = yield* DeviceService.makeWithHosts(new Map([[host.id, host]])).pipe( + Effect.provide(NodeCrypto.layer), + ); + pauseSummary = true; + const starting = yield* service.agentReadinessIfSupported().pipe(Effect.flip, Effect.forkChild); + yield* Deferred.await(summaryEntered); + yield* service.configure({ enabled: false }); + yield* Deferred.succeed(finishSummary, undefined); + const error = yield* Fiber.join(starting); + expect(error._tag === "DeviceHostUnavailableError" && error.reason).toBe( + "Device support failed during installing agent tools.", + ); + expect(stops).toEqual(["stop", "stop"]); + expect(yield* service.currentReadiness()).toBeNull(); + const state = yield* service.state; + expect(state.hostStatus).toBe("disabled"); + expect(state.hostStatuses).toEqual({}); + }).pipe( + Effect.provide( + ServerSettings.layerTest({ enableDeviceSupport: true, enableAgentDeviceAccess: true }), + ), + ), +); + +const pausedHost = ( + pausing: "ensureReady" | "ensureAgentReady", + id: string = LOCAL_DEVICE_HOST_ID, +) => + Effect.gen(function* () { + const started = yield* Deferred.make(); + const finish = yield* Deferred.make(); + const calls: string[] = []; + const run = () => Effect.succeed({ stdout: "", stderr: "", code: 0 }); + const ready: DeviceHost.DeviceHostAgentReady = { + nodePath: process.execPath, + hub: { origin: `http://${id}` }, + devices: makeDeviceHubOperations({ + hostId: id, + origin: `http://${id}`, + http: HttpClient.make((request) => + Effect.succeed( + HttpClientResponse.fromWeb(request, Response.json({ simulators: [], emulators: [] })), + ), + ), + run, + platformAvailable: () => Effect.succeed(false), + }), + agentDevice: null, + run, + helpers: { serveSimAxSettings: null, serveSimCli: null }, + }; + const start = (step: typeof pausing) => + step === pausing + ? Deferred.succeed(started, undefined).pipe( + Effect.andThen(Deferred.await(finish)), + Effect.as(ready), + ) + : Effect.succeed(ready); + const host: DeviceHost.DeviceHost["Service"] = { + id, + label: "Local", + summary: Effect.succeed({ + id, + label: "Local", + kind: id === LOCAL_DEVICE_HOST_ID ? "local" : "ssh", + hubInstalled: true, + agentDeviceInstalled: true, + platforms: [{ platform: "android", available: true }], + }), + platformAvailability: (platform) => Effect.succeed({ platform, available: true }), + ensureReady: () => start("ensureReady"), + ensureAgentReady: () => start("ensureAgentReady"), + current: Effect.succeed(null), + stopAgent: Effect.sync(() => { + calls.push("stopAgent"); + }), + stop: Effect.sync(() => { + calls.push("stop"); + }), + }; + return { host, started, finish, calls }; + }); + +it.effect("stays disabled when device support is turned off while listing starts a host", () => + Effect.gen(function* () { + const { host, started, finish, calls } = yield* pausedHost("ensureReady"); + const service = yield* DeviceService.makeWithHosts(new Map([[host.id, host]])).pipe( + Effect.provide(NodeCrypto.layer), + ); + const listing = yield* service.list.pipe(Effect.forkChild); + yield* Deferred.await(started); + yield* service.configure({ enabled: false }); + yield* Deferred.succeed(finish, undefined); + yield* Fiber.join(listing); + expect(calls).toEqual(["stop", "stop"]); + const state = yield* service.state; + expect(state.hostStatus).toBe("disabled"); + expect(state.hostStatuses).toEqual({}); + }).pipe(Effect.provide(ServerSettings.layerTest({ enableDeviceSupport: true }))), +); + +it.effect("retries without agent tools when agent access is turned off while they start", () => + Effect.gen(function* () { + const { host, started, finish, calls } = yield* pausedHost("ensureAgentReady"); + const service = yield* DeviceService.makeWithHosts(new Map([[host.id, host]])).pipe( + Effect.provide(NodeCrypto.layer), + ); + const retrying = yield* service.retryHost(LOCAL_DEVICE_HOST_ID).pipe(Effect.forkChild); + yield* Deferred.await(started); + yield* service.configure({ agentAccessEnabled: false }); + yield* Deferred.succeed(finish, undefined); + const state = yield* Fiber.join(retrying); + expect(calls).toEqual(["stopAgent", "stopAgent"]); + expect(state.agentAccessEnabled).toBe(false); + expect(state.hostStatus).toBe("ready"); + }).pipe( + Effect.provide( + ServerSettings.layerTest({ enableDeviceSupport: true, enableAgentDeviceAccess: true }), + ), + ), +); + +it.effect("keeps the replacement ready when an old host finishes starting", () => + Effect.gen(function* () { + for (const fail of [false, true]) { + const { host, started, finish, calls } = yield* pausedHost("ensureReady"); + const pendingHost: DeviceHost.DeviceHost["Service"] = { + ...host, + ensureReady: (onPhase) => + host.ensureReady(onPhase).pipe( + Effect.tap(() => onPhase("starting")), + Effect.filterOrFail( + () => !fail, + () => + new DeviceHost.DeviceHostError({ + hostId: host.id, + step: "start", + cause: "offline", + }), + ), + ), + }; + const hosts = new Map([[host.id, pendingHost]]); + const service = yield* DeviceService.makeWithHosts(hosts).pipe( + Effect.provide(NodeCrypto.layer), + ); + const listing = yield* service.list.pipe(Effect.forkChild); + yield* Deferred.await(started); + hosts.set(host.id, { ...host }); + yield* service.setHostStatus(host.id, { status: "ready" }); + yield* Deferred.succeed(finish, undefined); + const state = yield* Fiber.join(listing); + expect(calls).toEqual(["stop"]); + expect(state.hostStatuses[host.id]?.status).toBe("ready"); + } + }).pipe(Effect.provide(ServerSettings.layerTest({ enableDeviceSupport: true }))), +); + +it.effect("retries the hub when revoked agent startup fails", () => + Effect.gen(function* () { + const { host, started, finish } = yield* pausedHost("ensureAgentReady"); + const failingHost = { + ...host, + ensureAgentReady: () => + host + .ensureAgentReady(() => Effect.void) + .pipe( + Effect.andThen( + Effect.fail( + new DeviceHost.DeviceHostError({ + hostId: host.id, + step: "agent", + cause: "offline", + }), + ), + ), + ), + }; + const service = yield* DeviceService.makeWithHosts(new Map([[host.id, failingHost]])).pipe( + Effect.provide(NodeCrypto.layer), + ); + const retrying = yield* service.retryHost(host.id).pipe(Effect.forkChild); + yield* Deferred.await(started); + yield* service.configure({ agentAccessEnabled: false }); + yield* Deferred.succeed(finish, undefined); + const state = yield* Fiber.join(retrying); + expect(state.agentAccessEnabled).toBe(false); + expect(state.hostStatuses[host.id]?.status).toBe("ready"); + }).pipe( + Effect.provide( + ServerSettings.layerTest({ enableDeviceSupport: true, enableAgentDeviceAccess: true }), + ), + ), +); diff --git a/apps/server/src/device/DeviceService.test.ts b/apps/server/src/device/DeviceService.test.ts index e196b4cb7c3b4..b60ed21663385 100644 --- a/apps/server/src/device/DeviceService.test.ts +++ b/apps/server/src/device/DeviceService.test.ts @@ -2017,7 +2017,7 @@ describe("agent session release on close", () => { sessions.includes(session) ? "old" : sessions.includes(reopened) ? "reopened" : "none", ) .filter((label, index, labels) => label !== labels[index - 1]); - expect(published).toEqual(["none", "reopened"]); + expect(published[0] === "old" ? published.slice(1) : published).toEqual(["none", "reopened"]); }), ); }); diff --git a/apps/server/src/device/DeviceService.ts b/apps/server/src/device/DeviceService.ts index b365a6b9fe5c1..1aab127dd892d 100644 --- a/apps/server/src/device/DeviceService.ts +++ b/apps/server/src/device/DeviceService.ts @@ -379,14 +379,25 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function* ); // Uninterruptible so a committed retirement always prunes, even if its caller is cancelled. - const publish = (update: (state: DeviceServiceState) => DeviceServiceState) => - SynchronizedRef.modifyEffect(stateRef, ({ state }) => { - const next = { ...update(state), revision: state.revision + 1 }; - const retired = state.sessions.some((session) => !next.sessions.includes(session)); - return PubSub.publish(statePubSub, next).pipe( - Effect.as([{ next, retired }, { state: next }] as const), - ); - }).pipe( + const publish = ( + update: (state: DeviceServiceState) => DeviceServiceState, + onlyWhileEnabled = false, + ) => + SynchronizedRef.modifyEffect(stateRef, ({ state }) => + Effect.gen(function* () { + const enabled = + !onlyWhileEnabled || + (yield* readDeviceSettings.pipe( + Effect.map((current) => current.enabled), + Effect.orElseSucceed(() => false), + )); + const next = { ...(enabled ? update(state) : state), revision: state.revision + 1 }; + const retired = state.sessions.some((session) => !next.sessions.includes(session)); + return yield* PubSub.publish(statePubSub, next).pipe( + Effect.as([{ next, retired }, { state: next }] as const), + ); + }), + ).pipe( Effect.tap(({ retired }) => (retired ? pruneLaunchers : Effect.void)), Effect.map(({ next }) => next), Effect.uninterruptible, @@ -409,11 +420,13 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function* const setHostStatus = ( hostId: DeviceHostId, status: DeviceServiceState["hostStatuses"][string], + expectedHost?: DeviceHost.DeviceHost["Service"], ) => Effect.suspend(() => !hosts.has(hostId) ? SynchronizedRef.get(stateRef).pipe(Effect.map(({ state }) => state)) : publish((state) => { + if (expectedHost && hosts.get(hostId) !== expectedHost) return state; const previous = state.hostStatuses[hostId]; const detail = status.detail ?? @@ -440,9 +453,58 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function* }, }, }; - }), + }, true), ); + // Host startup can take minutes; only revocation cleanup holds the shared lock. + const stopIfRevoked = + (host: DeviceHost.DeviceHost["Service"], agent: boolean) => + (start: Effect.Effect) => + start.pipe( + Effect.ensuring( + lifecycleLock + .withPermit( + Effect.gen(function* () { + const current = yield* readDeviceSettings; + if (hosts.get(host.id) !== host) { + yield* host.stop; + return; + } + if (current.enabled && (!agent || current.agentAccessEnabled)) return; + yield* current.enabled ? host.stopAgent : host.stop; + yield* publish((state) => ({ + ...state, + ...(host.id === LOCAL_DEVICE_HOST_ID + ? { + hostStatus: current.enabled ? "idle" : "disabled", + hostStatusDetail: undefined, + } + : {}), + hostStatuses: Object.fromEntries( + Object.entries(state.hostStatuses).filter(([id]) => id !== host.id), + ), + })); + }), + ) + .pipe(Effect.ignore), + ), + ); + + const confirmStarted = Effect.fn("DeviceService.confirmStarted")(function* ( + host: DeviceHost.DeviceHost["Service"], + agent: boolean, + ) { + const current = yield* readDeviceSettings; + if (!current.enabled || (agent && !current.agentAccessEnabled)) return false; + if (hosts.get(host.id) !== host) + return yield* new DeviceHostUnavailableError({ + hostId: host.id, + hostLabel: host.label, + reason: "Host configuration changed. Retry the operation.", + }); + return true; + }); + const readiness: DeviceService["Service"]["readiness"] = Effect.fn("DeviceService.readiness")( function* (hostId) { const host = yield* resolveHost(hostId); @@ -456,43 +518,53 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function* } const ready = yield* host .ensureReady((status, detail) => - setHostStatus(host.id, { status, detail }).pipe(Effect.asVoid), + setHostStatus(host.id, { status, detail }, host).pipe(Effect.asVoid), ) .pipe( Effect.tapError((error) => - setHostStatus(host.id, { - status: "failed", - detail: - (error._tag === "DeviceHostError" && actionableHostFailure(error)) || error.message, - }), + setHostStatus( + host.id, + { + status: "failed", + detail: + (error._tag === "DeviceHostError" && actionableHostFailure(error)) || + error.message, + }, + host, + ), ), - Effect.mapError( - (error) => - new DeviceHostUnavailableError({ - hostId: host.id, - hostLabel: host.label, - reason: - error._tag === "NodeRuntimeUnavailableError" - ? nodeRuntimeUnavailableMessage("Local device support") - : (actionableHostFailure(error) ?? - `Device support failed during ${error.step}.`), - cause: error, - }), + Effect.mapError((error) => + error._tag === "DeviceHostUnavailableError" + ? error + : new DeviceHostUnavailableError({ + hostId: host.id, + hostLabel: host.label, + reason: + error._tag === "NodeRuntimeUnavailableError" + ? nodeRuntimeUnavailableMessage("Local device support") + : (actionableHostFailure(error) ?? + `Device support failed during ${error.step}.`), + cause: error, + }), ), + stopIfRevoked(host, false), ); - if (hosts.get(host.id) !== host) - return yield* new DeviceHostUnavailableError({ - hostId: host.id, - hostLabel: host.label, - reason: "Host configuration changed. Retry the operation.", - }); - const { state } = yield* SynchronizedRef.get(stateRef); - if (state.hostStatuses[host.id]?.status !== "ready") { - yield* setHostStatus(host.id, { status: "ready" }); - } - return { hostId: host.id, ...ready }; + return yield* lifecycleLock.withPermit( + Effect.gen(function* () { + if (!(yield* confirmStarted(host, false))) + return yield* new DeviceHostUnavailableError({ + hostId: host.id, + hostLabel: host.label, + reason: "Device support was turned off while the host was starting.", + }); + const { state } = yield* SynchronizedRef.get(stateRef); + if (state.hostStatuses[host.id]?.status !== "ready") { + yield* setHostStatus(host.id, { status: "ready" }); + } + return { hostId: host.id, ...ready }; + }), + ); }, - lifecycleLock.withPermit, ); const readinessIfSupported: DeviceService["Service"]["readinessIfSupported"] = Effect.fn( @@ -516,39 +588,57 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function* return null; const ready = yield* host .ensureAgentReady((phase, detail) => - setHostStatus(host.id, { status: phase, detail }).pipe(Effect.asVoid), + setHostStatus(host.id, { status: phase, detail }, host).pipe(Effect.asVoid), ) .pipe( Effect.tapError((error) => - setHostStatus(host.id, { - status: "failed", - detail: - (error._tag === "DeviceHostError" && actionableHostFailure(error)) || error.message, - }), + setHostStatus( + host.id, + { + status: "failed", + detail: + (error._tag === "DeviceHostError" && actionableHostFailure(error)) || + error.message, + }, + host, + ), ), - Effect.mapError( - (error) => - new DeviceHostUnavailableError({ - hostId: host.id, - hostLabel: host.label, - reason: - error._tag === "NodeRuntimeUnavailableError" - ? nodeRuntimeUnavailableMessage("Local device support") - : error._tag === "DeviceHostTimeoutError" - ? `Agent tools did not start within ${error.timeoutMs} ms.` - : error._tag === "AgentDeviceStartError" - ? error.message - : (actionableHostFailure(error) ?? - `Device support failed during ${error.step}.`), - cause: error, - }), + Effect.mapError((error) => + error._tag === "DeviceHostUnavailableError" + ? error + : new DeviceHostUnavailableError({ + hostId: host.id, + hostLabel: host.label, + reason: + error._tag === "NodeRuntimeUnavailableError" + ? nodeRuntimeUnavailableMessage("Local device support") + : error._tag === "DeviceHostTimeoutError" + ? `Agent tools did not start within ${error.timeoutMs} ms.` + : error._tag === "AgentDeviceStartError" + ? error.message + : (actionableHostFailure(error) ?? + `Device support failed during ${error.step}.`), + cause: error, + }), ), + stopIfRevoked(host, true), ); - const hostSummaries = yield* Effect.forEach(hosts.values(), (candidate) => candidate.summary); - yield* publish((state) => ({ ...state, hosts: hostSummaries })); - yield* setHostStatus(host.id, { status: "ready" }); - return { hostId: host.id, ...ready }; - }, lifecycleLock.withPermit); + const summarized = [...hosts.values()]; + const hostSummaries = yield* Effect.forEach(summarized, (candidate) => candidate.summary); + return yield* lifecycleLock.withPermit( + Effect.gen(function* () { + if (!(yield* confirmStarted(host, true))) return null; + // A reconcile while the summaries ran has already published the newer host list. + if ( + summarized.length === hosts.size && + summarized.every((candidate) => hosts.get(candidate.id) === candidate) + ) + yield* publish((state) => ({ ...state, hosts: hostSummaries })); + yield* setHostStatus(host.id, { status: "ready" }); + return { hostId: host.id, ...ready }; + }), + ); + }); const currentReadiness: DeviceService["Service"]["currentReadiness"] = (hostId) => resolveHost(hostId).pipe( @@ -622,6 +712,21 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function* ); }); + // Checked under the lock configure publishes under, so a failure that ends + // after device support was turned off cannot mark the setting back on. + const publishStartFailure = Effect.fn("DeviceService.publishStartFailure")( + function* (host: DeviceHost.DeviceHost["Service"], error: DeviceError) { + if (hosts.get(host.id) !== host || !(yield* readDeviceSettings).enabled) return; + yield* setHostStatus(host.id, { + status: "failed", + detail: error._tag === "DeviceHostUnavailableError" ? error.reason : error.message, + }); + }, + lifecycleLock.withPermit, + // Effect.fn also passes the call's arguments, which Effect.ignore would read as options. + (effect) => Effect.ignore(effect), + ); + const list: DeviceService["Service"]["list"] = Effect.gen(function* () { if (!(yield* readDeviceSettings).enabled) return (yield* SynchronizedRef.get(stateRef)).state; yield* Effect.forEach( @@ -630,14 +735,7 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function* Effect.gen(function* () { const ready = yield* readinessIfSupported(host.id); if (ready) yield* refresh(ready); - }).pipe( - Effect.catch((error) => - setHostStatus(host.id, { - status: "failed", - detail: error._tag === "DeviceHostUnavailableError" ? error.reason : error.message, - }), - ), - ), + }).pipe(Effect.catch((error) => publishStartFailure(host, error))), { concurrency: 4 }, ); return (yield* SynchronizedRef.get(stateRef)).state; @@ -677,20 +775,23 @@ export const makeWithHosts = Effect.fn("DeviceService.makeWithHosts")(function* const retryHost: DeviceService["Service"]["retryHost"] = Effect.fn("DeviceService.retryHost")( function* (hostId) { - yield* resolveHost(hostId); + const host = yield* resolveHost(hostId); if (!(yield* readDeviceSettings).enabled) return (yield* SynchronizedRef.get(stateRef)).state; yield* Effect.gen(function* () { const ready = - (yield* agentReadinessIfSupported(hostId)) ?? (yield* readinessIfSupported(hostId)); + (yield* agentReadinessIfSupported(hostId).pipe( + Effect.catch((error) => + readDeviceSettings.pipe( + Effect.flatMap((current) => + current.enabled && current.agentAccessEnabled + ? Effect.fail(error) + : Effect.succeed(null), + ), + ), + ), + )) ?? (yield* readinessIfSupported(hostId)); if (ready) yield* refresh(ready); - }).pipe( - Effect.catch((error) => - setHostStatus(hostId, { - status: "failed", - detail: error._tag === "DeviceHostUnavailableError" ? error.reason : error.message, - }), - ), - ); + }).pipe(Effect.catch((error) => publishStartFailure(host, error))); return (yield* SynchronizedRef.get(stateRef)).state; }, ); @@ -1540,9 +1641,13 @@ export const make = Effect.gen(function* () { ), ), (status, detail) => - service - .setHostStatus(host.id, { status, ...(detail ? { detail } : {}) }) - .pipe(Effect.asVoid), + Effect.suspend(() => + configured.get(host.id)?.scope === hostScope + ? service + .setHostStatus(host.id, { status, ...(detail ? { detail } : {}) }) + .pipe(Effect.asVoid) + : Effect.void, + ), ).pipe(Effect.provideService(Scope.Scope, hostScope), Effect.provide(hostContext)); hosts.set(host.id, instance); configured.set(host.id, { config: host, scope: hostScope }); diff --git a/apps/server/src/device/EasDeviceHost.test.ts b/apps/server/src/device/EasDeviceHost.test.ts index 3c083cdcbb7f8..50f9e1bdffd81 100644 --- a/apps/server/src/device/EasDeviceHost.test.ts +++ b/apps/server/src/device/EasDeviceHost.test.ts @@ -310,6 +310,7 @@ describe("EAS Simulator host", () => { ).toBeNull(); yield* Scope.close(scope, Exit.void); + expect(yield* host.current).toBeNull(); expect(expo.named("T3StopEasSimulatorSession").map((call) => call.variables.id)).toEqual([ "session-1", ]); diff --git a/apps/server/src/device/EasDeviceHost.ts b/apps/server/src/device/EasDeviceHost.ts index 719cfbf435a8a..0a5ec5809014a 100644 --- a/apps/server/src/device/EasDeviceHost.ts +++ b/apps/server/src/device/EasDeviceHost.ts @@ -843,10 +843,13 @@ export const make = Effect.fn("EasDeviceHost.make")(function* (config: EasDevice "EasDeviceHost.ensureReady", )(function* () { if (ready) return ready; + const startedIn = epoch; yield* adoptRecorded; // Without a credential the host is still ready: opening a device then // says how to sign in, and listing reaps once one is there. yield* FiberSet.run(background)(api.pipe(Effect.flatMap(reapLeftovers), Effect.ignore)); + if (epoch !== startedIn) + return yield* unavailable(`${config.label} was stopped while readiness started.`); ready = { nodePath, hub: { origin: "https://api.expo.dev" }, @@ -874,6 +877,7 @@ export const make = Effect.fn("EasDeviceHost.make")(function* (config: EasDevice // and stays in the ledger for the next server if this one closes first. const stopAll = Effect.gen(function* () { epoch++; + ready = null; // Interrupted starts stop the sessions they created before this returns. yield* FiberSet.clear(starts); yield* FiberSet.clear(background); diff --git a/apps/server/src/device/SimlockDeviceHost.test.ts b/apps/server/src/device/SimlockDeviceHost.test.ts index 8726a7a7c03de..4fd366affa23f 100644 --- a/apps/server/src/device/SimlockDeviceHost.test.ts +++ b/apps/server/src/device/SimlockDeviceHost.test.ts @@ -273,7 +273,15 @@ describe("Simlock device host", () => { yield* ready.devices.shutdown({ id: leaseId, platform: "android" }); expect(pool.ran("close")).toHaveLength(1); expect((yield* ready.devices.list).devices.some((device) => device.booted)).toBe(false); + yield* host.stop; + expect(yield* host.current).toBeNull(); + const stoppedStart = yield* host.ensureReady(() => host.stop).pipe(Effect.result); + expect(stoppedStart._tag).toBe("Failure"); + expect(yield* host.current).toBeNull(); + yield* host.ensureReady(() => Effect.void); + expect(yield* host.current).not.toBeNull(); yield* Scope.close(scope, Exit.void); + expect(yield* host.current).toBeNull(); expect(pool.ran("close")).toHaveLength(1); }).pipe(Effect.scoped, Effect.provide(layer)), ); diff --git a/apps/server/src/device/SimlockDeviceHost.ts b/apps/server/src/device/SimlockDeviceHost.ts index 417399c9114c8..37ba8661ff693 100644 --- a/apps/server/src/device/SimlockDeviceHost.ts +++ b/apps/server/src/device/SimlockDeviceHost.ts @@ -645,10 +645,13 @@ export const make = Effect.fn("SimlockDeviceHost.make")(function* ( "SimlockDeviceHost.ensureReady", )(function* (onPhase) { if (ready) return ready; + const startedIn = stops; // A missing token or unreachable gateway does not stop readiness: listing // and opening devices say what is wrong instead. yield* onPhase("installing"); cli = yield* agentDeviceCli; + if (stops !== startedIn) + return yield* unavailable("Device host was stopped while it was starting."); const { nodePath } = cli; ready = { nodePath, @@ -669,6 +672,7 @@ export const make = Effect.fn("SimlockDeviceHost.make")(function* ( const stopAll = Effect.gen(function* () { stops++; + ready = null; yield* FiberMap.clear(keepAlive); yield* Effect.forEach([...leases.values()], release, { discard: true, concurrency: 4 }); }).pipe(Effect.ignore); diff --git a/apps/server/src/device/SshDeviceHost.test.ts b/apps/server/src/device/SshDeviceHost.test.ts index 8249afb1e51c1..ea652030a10dd 100644 --- a/apps/server/src/device/SshDeviceHost.test.ts +++ b/apps/server/src/device/SshDeviceHost.test.ts @@ -2,6 +2,8 @@ import { expect, it } from "@effect/vitest"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; import * as Net from "@t3tools/shared/Net"; +import * as Scope from "effect/Scope"; +import * as Exit from "effect/Exit"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; @@ -105,6 +107,8 @@ it.effect("preserves installed status after probes and cleans failed agent activ }); }), ); + const hostScope = yield* Scope.make(); + yield* Effect.addFinalizer(() => Scope.close(hostScope, Exit.void)); const host = yield* SshDeviceHost.make( { kind: "ssh", id: "test", label: "Test", target: "test.example" }, () => @@ -118,6 +122,7 @@ it.effect("preserves installed status after probes and cleans failed agent activ ) : Effect.void, ).pipe( + Effect.provideService(Scope.Scope, hostScope), Effect.provide(Layer.mergeAll(ServerConfig.layerTest(home, home), Net.layer)), Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), Effect.provideService( @@ -193,8 +198,19 @@ it.effect("preserves installed status after probes and cleans failed agent activ for (const args of commandArgs) { for (const option of multiplexingOff) expect(args).not.toContain(option); } - yield* host.stop; + const retired = yield* host + .ensureAgentReady(() => Scope.close(hostScope, Exit.void)) + .pipe(Effect.result); + expect(retired._tag).toBe("Failure"); expect(forwards).toBe(0); + const modesAfterClose = [...modes]; + const unavailable = yield* host.ensureReady(() => Effect.void).pipe(Effect.flip); + expect(unavailable).toMatchObject({ + _tag: "DeviceHostUnavailableError", + reason: "Host configuration changed. Retry the operation.", + }); + expect(unavailable.cause).toBeUndefined(); + expect(modes).toEqual(modesAfterClose); }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer))), ); diff --git a/apps/server/src/device/SshDeviceHost.ts b/apps/server/src/device/SshDeviceHost.ts index c1d57869ead95..80216fe654890 100644 --- a/apps/server/src/device/SshDeviceHost.ts +++ b/apps/server/src/device/SshDeviceHost.ts @@ -1,5 +1,6 @@ import { type DeviceHostSummary, + DeviceHostUnavailableError, DevicePlatformAvailability, DeviceToolVersions, deviceToolInstallMessage, @@ -215,6 +216,18 @@ export const make = Effect.fn("SshDeviceHost.make")(function* ( ); const lock = yield* Semaphore.make(1); let stopped = false; + let retired = false; + const requireActive = Effect.suspend(() => + retired + ? Effect.fail( + new DeviceHostUnavailableError({ + hostId: config.id, + hostLabel: config.label, + reason: "Host configuration changed. Retry the operation.", + }), + ) + : Effect.void, + ); let activated = false; let wantsAgent = false; let ready: @@ -500,6 +513,7 @@ export const make = Effect.fn("SshDeviceHost.make")(function* ( const ensureReady: DeviceHost.DeviceHost["Service"]["ensureReady"] = (onPhase) => lock.withPermit( Effect.gen(function* () { + yield* requireActive; stopped = false; if (ready) return ready; summary = yield* provide(probe(config, owner)); @@ -530,6 +544,7 @@ export const make = Effect.fn("SshDeviceHost.make")(function* ( const changeAgent = (enabled: boolean) => lock.withPermit( Effect.gen(function* () { + yield* requireActive; wantsAgent = enabled; if (enabled && ready?.agentDevice) return { ...ready, agentDevice: ready.agentDevice }; if (!enabled && !ready?.agentDevice) return null; @@ -551,7 +566,11 @@ export const make = Effect.fn("SshDeviceHost.make")(function* ( ); }), ); - yield* Effect.addFinalizer(() => stop); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + retired = true; + }).pipe(Effect.andThen(stop)), + ); return { id: config.id, label: config.label,