# Files changed · PR \#23870 · pingdotgg/t3code [View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/23870/diffs) Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code) Visibility: public [feat: choose updates and restart this device first](https://git.cafe/pingdotgg/t3code/pull/23870.md) Version: 57 Head commit: 59debb64aaea40f8f4efbc0fa95d43f50172eab5 Base commit: 11adcb0e25a04785f70d01c3f2a858889683007e Comparison base: 11adcb0e25a04785f70d01c3f2a858889683007e Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## apps/desktop/src/backend/DesktopBackendConfiguration.test.ts Status: modified; +2; −2 ```diff @@ -516,7 +516,7 @@ ); }); - it.effect("resolveWsl retires a staged runtime whose executable does not start", () => { + it.effect("resolveWsl retires a staged runtime when its probe fails", () => { const archiveHash = "c".repeat(64); const stagedAppRoot = `/home/test/.t3/wsl-runtime/sha256-${archiveHash}`; const observedProbeRoots: string[] = []; @@ -533,7 +533,7 @@ }), probeRuntime: (_distro, root) => { observedProbeRoots.push(root); - return { ok: false, reason: `${root}/t3 --version failed (exit 127)` }; + return { ok: false, reason: "t3 is not executable" }; }, ensureNodePty: (_distro, root) => { observedNodePtyRoots.push(root); ``` ## apps/desktop/src/backend/DesktopBackendManager.test.ts Status: modified; +5; −1 ```diff @@ -159,7 +159,9 @@ Layer.succeed(DesktopTelemetryPublisher.DesktopTelemetryPublisher, { latest: Effect.succeedNone, changes: Stream.empty, + encoded: () => input.desktopTelemetryStream ?? Stream.empty, + requestUpdateRestartPolicy: () => Effect.succeed(true), - encoded: input.desktopTelemetryStream ?? Stream.empty, + clearUpdateRestartPolicy: () => Effect.void, handleControlForSource: () => Effect.void, removeControlSource: () => Effect.void, publishUpdateReport: () => Effect.void, @@ -360,6 +362,8 @@ const instance = yield* makeTestInstance({ spawnerLayer: layerSpawner, desktopTelemetryPublisher: { + requestUpdateRestartPolicy: () => Effect.succeed(true), + clearUpdateRestartPolicy: () => Effect.void, handleControlForSource: (_sourceId, message) => message.type === "setDiagnosticsDemand" ? Deferred.succeed(handled, message.enabled).pipe(Effect.asVoid) ``` ## apps/desktop/src/backend/DesktopBackendManager.ts Status: modified; +1; −1 ```diff @@ -881,7 +881,7 @@ const program = runBackendProcess({ ...config.value, - desktopTelemetryStream: desktopTelemetryPublisher.encoded, + desktopTelemetryStream: desktopTelemetryPublisher.encoded(spec.id), desktopBrowserStream: desktopBrowserHost.events, onDesktopBrowserCommand: desktopBrowserHost.handleCommandLine, onDesktopTelemetryControl: (message) => ``` ## apps/desktop/src/backend/DesktopBackendPool.test.ts Status: modified; +3; −1 ```diff @@ -46,7 +46,9 @@ Layer.succeed(DesktopTelemetryPublisher.DesktopTelemetryPublisher, { latest: Effect.succeedNone, changes: Stream.empty, - encoded: Stream.empty, + encoded: () => Stream.empty, + requestUpdateRestartPolicy: () => Effect.succeed(true), + clearUpdateRestartPolicy: () => Effect.void, handleControlForSource: () => Effect.void, removeControlSource: () => Effect.void, publishUpdateReport: () => Effect.void, ``` ## apps/desktop/src/ipc/methods/updates.test.ts Status: added; +57; −0 ```diff @@ -1,0 +1,57 @@ +import { assert, describe, it } from "@effect/vitest"; +import { DesktopUpdateActionResultSchema } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import * as TestClock from "effect/testing/TestClock"; + +import * as DesktopUpdates from "../../updates/DesktopUpdates.ts"; +import { flushCallbacks, makeHarness } from "../../updates/updatesTestHarness.ts"; +import { installUpdate } from "./updates.ts"; + +describe("update install IPC", () => { + it.effect.each([ + { expectedVersion: "1.2.4", accepted: false }, + { expectedVersion: "1.2.5", accepted: true }, + { expectedVersion: undefined, accepted: true }, + { + expectedVersion: { expectedVersion: "1.2.4", continueRunningThreads: true }, + accepted: false, + }, + { expectedVersion: { expectedVersion: "1.2.5", continueRunningThreads: true }, accepted: true }, + { expectedVersion: { continueRunningThreads: true }, accepted: true }, + { + expectedVersion: { expectedVersion: "1.2.5", continueRunningThreads: false }, + accepted: true, + }, + { + expectedVersion: { expectedVersion: "1.2.4", continueRunningThreads: false }, + accepted: false, + }, + { expectedVersion: { continueRunningThreads: false }, accepted: true }, + ])("guards the selected version $expectedVersion", ({ expectedVersion, accepted }) => { + const harness = makeHarness(); + return Effect.scoped( + Effect.gen(function* () { + const updates = yield* DesktopUpdates.DesktopUpdates; + yield* updates.configure; + harness.emit("update-downloaded", { version: "1.2.5" }); + yield* flushCallbacks; + + const result = yield* installUpdate + .handler(expectedVersion) + .pipe(Effect.flatMap(Schema.decodeUnknownEffect(DesktopUpdateActionResultSchema))); + assert.equal(result.accepted, accepted); + assert.equal(harness.quitAndInstalls(), accepted ? 1 : 0); + assert.equal( + harness.restartPolicies.length, + typeof expectedVersion === "object" && + expectedVersion.continueRunningThreads === false && + accepted + ? 1 + : 0, + ); + }), + ).pipe(Effect.provide(Layer.merge(TestClock.layer(), harness.layer))); + }); +}); ``` ## apps/desktop/src/ipc/methods/updates.ts Status: modified; +14; −3 ```diff @@ -43,11 +43,22 @@ export const installUpdate = DesktopIpc.makeIpcMethod({ channel: IpcChannels.UPDATE_INSTALL_CHANNEL, + payload: Schema.Union([ + Schema.UndefinedOr(Schema.String), + Schema.Struct({ + expectedVersion: Schema.optional(Schema.String), + continueRunningThreads: Schema.optional(Schema.Boolean), - payload: Schema.Void, + }), + ]), result: DesktopUpdateActionResultSchema, + handler: Effect.fn("desktop.ipc.updates.install")(function* (input) { - handler: Effect.fn("desktop.ipc.updates.install")(function* () { + const expectedVersion = typeof input === "object" ? input.expectedVersion : input; + const continueRunningThreads = + typeof input === "object" ? input.continueRunningThreads : undefined; const updates = yield* DesktopUpdates.DesktopUpdates; - return yield* updates.install; + return yield* expectedVersion === undefined && continueRunningThreads === undefined + ? updates.install + : updates.installPrepared(expectedVersion, continueRunningThreads); }), }); ``` ## apps/desktop/src/preload.ts Status: modified; +8; −1 ```diff @@ -270,7 +270,14 @@ ipcRenderer.invoke(IpcChannels.UPDATE_SET_CHANNEL_CHANNEL, channel), checkForUpdate: () => ipcRenderer.invoke(IpcChannels.UPDATE_CHECK_CHANNEL), downloadUpdate: () => ipcRenderer.invoke(IpcChannels.UPDATE_DOWNLOAD_CHANNEL), + supportsUpdateThreadContinuationControl: true, + installUpdate: (expectedVersion, continueRunningThreads) => + ipcRenderer.invoke( + IpcChannels.UPDATE_INSTALL_CHANNEL, + continueRunningThreads === undefined - installUpdate: () => ipcRenderer.invoke(IpcChannels.UPDATE_INSTALL_CHANNEL), + ? expectedVersion + : { expectedVersion, continueRunningThreads }, + ), cliCommand: { getState: () => ipcRenderer.invoke(IpcChannels.CLI_COMMAND_GET_STATE_CHANNEL), install: () => ipcRenderer.invoke(IpcChannels.CLI_COMMAND_INSTALL_CHANNEL), ``` ## apps/desktop/src/telemetry/DesktopTelemetryPublisher.test.ts Status: modified; +214; −0 ```diff @@ -3,9 +3,13 @@ import * as Deferred from "effect/Deferred"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; import * as TestClock from "effect/testing/TestClock"; @@ -56,7 +60,217 @@ } satisfies ElectronApp.ElectronApp["Service"]); } +const layerPolicyPublisher = DesktopTelemetryPublisher.layer.pipe( + Layer.provide( + Layer.mergeAll( + layerElectronApp([]), + layerHistory, + Layer.succeed(ElectronPowerMonitor.ElectronPowerMonitor, { + isOnBatteryPower: Effect.succeed(false), + getSystemIdleTime: Effect.succeed(0), + getSystemIdleState: () => Effect.succeed("active"), + getCurrentThermalState: Effect.succeed("nominal" as const), + onSimpleEvent: () => Effect.void, + onThermalStateChange: () => Effect.void, + onSpeedLimitChange: () => Effect.void, + }), + ), + ), +); + describe("DesktopTelemetryPublisher", () => { + it.effect("requires matching policy ACKs from every targeted backend and targets rollback", () => + Effect.scoped( + Effect.gen(function* () { + const publisher = yield* DesktopTelemetryPublisher.DesktopTelemetryPublisher; + const primaryMessages = yield* Queue.unbounded>(); + const wslMessages = yield* Queue.unbounded>(); + for (const [sourceId, messages] of [ + ["primary", primaryMessages], + ["wsl:Ubuntu", wslMessages], + ] as const) { + yield* Stream.runForEach(publisher.encoded(sourceId), (bytes) => + Queue.offer(messages, JSON.parse(new TextDecoder().decode(bytes))), + ).pipe(Effect.forkScoped); + assert.equal((yield* Queue.take(messages)).type, "desktopTelemetryHello"); + } + const policy = { + version: 1, + type: "desktopUpdateRestartPolicy", + requestId: "install-1", + continueRunningThreads: false, + } as const; + const preparing = yield* publisher + .requestUpdateRestartPolicy(["primary", "wsl:Ubuntu"], policy) + .pipe(Effect.forkScoped); + for (const messages of [primaryMessages, wslMessages]) { + const command = yield* Stream.fromQueue(messages).pipe( + Stream.filter((message) => message.type === policy.type), + Stream.runHead, + ); + assert.deepEqual(Option.getOrThrow(command), policy); + } + const result = { + ...policy, + type: "desktopUpdateRestartPolicyResult", + accepted: true, + } as const; + yield* publisher.handleControlForSource("unrelated", result); + yield* publisher.handleControlForSource("primary", { + ...result, + continueRunningThreads: true, + }); + assert.isUndefined(preparing.pollUnsafe()); + yield* publisher.handleControlForSource("primary", result); + assert.isUndefined(preparing.pollUnsafe()); + yield* publisher.handleControlForSource("wsl:Ubuntu", result); + assert.isTrue(yield* Fiber.join(preparing)); + const rollback = yield* publisher + .requestUpdateRestartPolicy(["wsl:Ubuntu"], { ...policy, continueRunningThreads: true }) + .pipe(Effect.forkScoped); + const rollbackMessage = yield* Stream.fromQueue(wslMessages).pipe( + Stream.filter((message) => message.type === policy.type), + Stream.runHead, + ); + assert.equal(Option.getOrThrow(rollbackMessage).continueRunningThreads, true); + yield* publisher.handleControlForSource("wsl:Ubuntu", { + ...result, + continueRunningThreads: true, + }); + assert.isTrue(yield* Fiber.join(rollback)); + const pendingPrimary = yield* Queue.takeBetween( + primaryMessages, + 0, + Number.POSITIVE_INFINITY, + ); + assert.isFalse( + Array.from(pendingPrimary).some( + (message) => message.type === policy.type && message.continueRunningThreads === true, + ), + ); + yield* publisher.clearUpdateRestartPolicy(policy.requestId); + }), + ).pipe(Effect.provide(layerPolicyPublisher)), + ); + + it.effect.each(["queued", "initial", "cleared initial"] as const)( + "drops %s restart policies replaced while a source consumer is paused", + (policyState) => + Effect.scoped( + Effect.gen(function* () { + const publisher = yield* DesktopTelemetryPublisher.DesktopTelemetryPublisher; + const subscribe = yield* Deferred.make(); + const helloRead = yield* Deferred.make(); + const resume = yield* Deferred.make(); + const observerReady = yield* Deferred.make(); + const observedPolicies = yield* Queue.unbounded>(); + yield* Stream.runForEach(publisher.encoded("observer"), (bytes) => { + const message = JSON.parse(new TextDecoder().decode(bytes)) as Record; + return message.type === "desktopTelemetryHello" + ? Deferred.succeed(observerReady, undefined) + : message.type === "desktopUpdateRestartPolicy" + ? Queue.offer(observedPolicies, message) + : Effect.void; + }).pipe(Effect.forkScoped); + yield* Deferred.await(observerReady); + const nextPolicy = yield* publisher.encoded("paused").pipe( + Stream.map( + (bytes) => JSON.parse(new TextDecoder().decode(bytes)) as Record, + ), + Stream.tap((message) => + message.type === "desktopTelemetryHello" + ? Deferred.succeed(helloRead, undefined).pipe( + Effect.andThen(Deferred.await(resume)), + ) + : Effect.void, + ), + Stream.filter((message) => message.type === "desktopUpdateRestartPolicy"), + Stream.runHead, + (consume) => Deferred.await(subscribe).pipe(Effect.andThen(consume)), + Effect.forkScoped, + ); + if (policyState === "queued") { + yield* Deferred.succeed(subscribe, undefined); + yield* Deferred.await(helloRead); + } + for (const requestId of ["install-A", "install-B"]) { + const policy = { + version: 1, + type: "desktopUpdateRestartPolicy", + requestId, + continueRunningThreads: false, + } as const; + const preparing = yield* publisher + .requestUpdateRestartPolicy(["paused", "observer"], policy) + .pipe(Effect.forkScoped); + assert.deepEqual(yield* Queue.take(observedPolicies), policy); + for (const sourceId of ["paused", "observer"]) { + yield* publisher.handleControlForSource(sourceId, { + ...policy, + type: "desktopUpdateRestartPolicyResult", + accepted: true, + }); + } + assert.isTrue(yield* Fiber.join(preparing)); + if (requestId === "install-A" && policyState !== "queued") { + yield* Deferred.succeed(subscribe, undefined); + yield* Deferred.await(helloRead); + if (policyState === "cleared initial") + yield* publisher.clearUpdateRestartPolicy(requestId); + } + } + yield* Deferred.succeed(resume, undefined); + assert.equal(Option.getOrThrow(yield* Fiber.join(nextPolicy)).requestId, "install-B"); + }), + ).pipe(Effect.provide(layerPolicyPublisher)), + ); + + it.effect.each(["refused", "disconnected", "missing"] as const)( + "refuses an update when a backend policy ACK is %s", + (failure) => + Effect.scoped( + Effect.gen(function* () { + const publisher = yield* DesktopTelemetryPublisher.DesktopTelemetryPublisher; + const policy = { + version: 1, + type: "desktopUpdateRestartPolicy", + requestId: "failed-1", + continueRunningThreads: false, + } as const; + const pending = yield* publisher + .requestUpdateRestartPolicy(["primary", "wsl:Ubuntu"], policy) + .pipe(Effect.forkScoped); + const command = yield* publisher.encoded("primary").pipe( + Stream.map( + (bytes) => JSON.parse(new TextDecoder().decode(bytes)) as Record, + ), + Stream.filter((message) => message.type === policy.type), + Stream.runHead, + ); + assert.deepEqual(Option.getOrThrow(command), policy); + const result = { + ...policy, + type: "desktopUpdateRestartPolicyResult", + accepted: true, + } as const; + yield* publisher.handleControlForSource("primary", result); + if (failure === "refused") + yield* publisher.handleControlForSource("wsl:Ubuntu", { ...result, accepted: false }); + else if (failure === "disconnected") yield* publisher.removeControlSource("primary"); + else yield* TestClock.adjust("15 seconds"); + assert.isFalse(yield* Fiber.join(pending)); + yield* publisher.clearUpdateRestartPolicy(policy.requestId); + const hello = Option.getOrThrow(yield* publisher.encoded("primary").pipe(Stream.runHead)); + assert.deepEqual(JSON.parse(new TextDecoder().decode(hello)), { + version: 1, + type: "desktopTelemetryHello", + electronPid: process.pid, + supportsUpdateThreadContinuationControl: true, + }); + }), + ).pipe(Effect.provide(layerPolicyPublisher)), + ); + it.effect("stops when its scope closes during an Electron telemetry sample", () => Effect.gen(function* () { const pollStarted = yield* Deferred.make(); ``` ## apps/desktop/src/telemetry/DesktopTelemetryPublisher.ts Status: modified; +114; −13 ```diff @@ -6,6 +6,8 @@ type DesktopTelemetryCommitDesktopUpdate, type DesktopTelemetryRequestDesktopUpdate, type DesktopUpdateStatusReport, + type DesktopUpdateRestartPolicy, + type DesktopUpdateRestartPolicyResult, type HostPowerSnapshot, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; @@ -63,7 +65,12 @@ { readonly latest: Effect.Effect>; readonly changes: Stream.Stream; - readonly encoded: Stream.Stream; + readonly encoded: (sourceId: string) => Stream.Stream; + readonly requestUpdateRestartPolicy: ( + sourceIds: readonly string[], + policy: DesktopUpdateRestartPolicy, + ) => Effect.Effect; + readonly clearUpdateRestartPolicy: (requestId: string) => Effect.Effect; readonly handleControlForSource: ( sourceId: string, message: DesktopTelemetryControlMessage, @@ -176,6 +183,58 @@ const updateCommitQueue = yield* Queue.unbounded(); const updateCancellationQueue = yield* Queue.unbounded(); + const restartPolicies = yield* Ref.make>( + new Map(), + ); + const restartPolicyChanges = yield* PubSub.unbounded<{ + readonly sourceIds: readonly string[]; + readonly policy: DesktopUpdateRestartPolicy; + }>(); + const restartPolicyResults = yield* PubSub.unbounded<{ + readonly sourceId: string; + readonly result: DesktopUpdateRestartPolicyResult; + }>(); + + const requestUpdateRestartPolicy = Effect.fn("desktop.telemetry.requestUpdateRestartPolicy")( + function* (sourceIds: readonly string[], policy: DesktopUpdateRestartPolicy) { + if (sourceIds.length === 0) return true; + return yield* Effect.scoped( + Effect.gen(function* () { + const results = yield* PubSub.subscribe(restartPolicyResults); + yield* Ref.update(restartPolicies, (current) => { + const next = new Map(current); + for (const sourceId of sourceIds) next.set(sourceId, policy); + return next; + }); + yield* PubSub.publish(restartPolicyChanges, { sourceIds, policy }); + const pending = new Set(sourceIds); + while (pending.size > 0) { + const { sourceId, result } = yield* PubSub.take(results); + if ( + !sourceIds.includes(sourceId) || + result.requestId !== policy.requestId || + result.continueRunningThreads !== policy.continueRunningThreads + ) + continue; + if (!result.accepted) return false; + pending.delete(sourceId); + } + return true; + }), + ).pipe(Effect.timeoutOption(Duration.seconds(15)), Effect.map(Option.getOrElse(() => false))); + }, + ); + + const clearUpdateRestartPolicy = Effect.fn("desktop.telemetry.clearUpdateRestartPolicy")( + function* (requestId: string) { + yield* Ref.update( + restartPolicies, + (current) => + new Map(Array.from(current).filter(([, policy]) => policy.requestId !== requestId)), + ); + }, + ); + const offer = (event: PowerEvent): void => { Queue.offerUnsafe(powerEvents, event); }; @@ -322,6 +381,10 @@ message, ) => { switch (message.type) { + case "desktopUpdateRestartPolicyResult": + return PubSub.publish(restartPolicyResults, { sourceId, result: message }).pipe( + Effect.asVoid, + ); case "setDiagnosticsDemand": return Ref.modify(diagnosticsDemandSources, (sources) => { const previous = sources.size > 0; @@ -352,10 +415,21 @@ return Queue.offer(updateCancellationQueue, message).pipe(Effect.asVoid); } }; - const removeControlSource: DesktopTelemetryPublisher["Service"]["removeControlSource"] = ( - sourceId, - ) => - Ref.modify(diagnosticsDemandSources, (sources) => { + const removeControlSource = Effect.fn("desktop.telemetry.removeControlSource")(function* ( + sourceId: string, + ) { + const policy = (yield* Ref.get(restartPolicies)).get(sourceId); + if (policy !== undefined) + yield* PubSub.publish(restartPolicyResults, { + sourceId, + result: { + ...policy, + type: "desktopUpdateRestartPolicyResult", + accepted: false, + reason: "The backend disconnected while preparing the update.", + }, + }); + yield* Ref.modify(diagnosticsDemandSources, (sources) => { const previous = sources.size > 0; const next = new Set(sources); next.delete(sourceId); @@ -367,6 +441,7 @@ : Queue.offer(sampleTriggers, undefined).pipe(Effect.asVoid), ), ); + }); const snapshots = Stream.unwrap( Effect.gen(function* () { @@ -394,14 +469,38 @@ ); }), ); + const encoded = (sourceId: string) => + Stream.unwrap( + Effect.gen(function* () { + const subscription = yield* PubSub.subscribe(restartPolicyChanges); + const initialPolicy = (yield* Ref.get(restartPolicies)).get(sourceId); + const policies = Stream.fromSubscription(subscription).pipe( + Stream.filter((change) => change.sourceIds.includes(sourceId)), + Stream.map((change) => change.policy), + ); + return Stream.concat( + Stream.make({ + version: 1, + type: "desktopTelemetryHello", + electronPid: process.pid, + supportsUpdateThreadContinuationControl: true, + } as const), + Stream.concat( + initialPolicy === undefined ? Stream.empty : Stream.make(initialPolicy), + Stream.merge(Stream.merge(snapshots, updateReports), policies), + ), + ); + }), + ).pipe( + Stream.filterEffect((message) => + message.type === "desktopUpdateRestartPolicy" + ? Ref.get(restartPolicies).pipe( + Effect.map((current) => current.get(sourceId) === message), - const encoded = Stream.concat( - Stream.make({ - version: 1, + ) + : Effect.succeed(true), + ), + Stream.map((message) => textEncoder.encode(`${encodeMessage(message)}\n`)), + ); - type: "desktopTelemetryHello", - electronPid: process.pid, - } as const), - Stream.merge(snapshots, updateReports), - ).pipe(Stream.map((message) => textEncoder.encode(`${encodeMessage(message)}\n`))); const publishUpdateReport: DesktopTelemetryPublisher["Service"]["publishUpdateReport"] = ( report, @@ -415,6 +514,8 @@ latest: Ref.get(latest), changes: Stream.fromPubSub(changes), encoded, + requestUpdateRestartPolicy, + clearUpdateRestartPolicy, handleControlForSource, removeControlSource, publishUpdateReport, ``` ## apps/desktop/src/updates/DesktopRemoteUpdates.test.ts Status: modified; +87; −71 ```diff @@ -56,7 +56,9 @@ const publisher = DesktopTelemetryPublisher.DesktopTelemetryPublisher.of({ latest: Effect.succeedNone, changes: Stream.empty, + encoded: () => Stream.empty, + requestUpdateRestartPolicy: () => Effect.succeed(true), + clearUpdateRestartPolicy: () => Effect.void, - encoded: Stream.empty, handleControlForSource: () => Effect.void, removeControlSource: () => Effect.void, publishUpdateReport: (report) => @@ -84,42 +86,50 @@ } describe("DesktopRemoteUpdates", () => { - it.effect("drives check, download, and install with no local confirmation", () => { + it.effect.each([false, true, undefined])( + "retains the remote resume choice %s through install", + (continueRunningThreads) => { - const harness = makeHarness(); + const harness = makeHarness(); - return runRemoteUpdatesTest(harness, ({ reports, requests, commits }) => - Effect.gen(function* () { - yield* Queue.offer(requests, request("req-1")); - yield* settle; - assert.equal(harness.checkCount(), 1); + return runRemoteUpdatesTest(harness, ({ reports, requests, commits }) => + Effect.gen(function* () { + yield* Queue.offer(requests, request("req-1")); + yield* settle; + assert.equal(harness.checkCount(), 1); - harness.emit("update-available", { version: "1.2.4" }); + harness.emit("update-available", { version: "1.2.4" }); - yield* settle; - harness.emit("update-downloaded", { version: "1.2.4" }); - yield* settle; + yield* settle; + harness.emit("update-downloaded", { version: "1.2.4" }); + yield* settle; - const terminals = terminalReports(reports); - assert.equal(terminals.length, 1); + const terminals = terminalReports(reports); + assert.equal(terminals.length, 1); - assert.equal(terminals[0]?.outcome, "ready-to-install"); - assert.equal(terminals[0]?.requestId, "req-1"); - assert.equal(harness.quitAndInstalls(), 0); + assert.equal(terminals[0]?.outcome, "ready-to-install"); + assert.equal(terminals[0]?.requestId, "req-1"); + assert.equal(harness.quitAndInstalls(), 0); + yield* Queue.offer(commits, { + version: 1, - yield* Queue.offer(commits, { - version: 1, - type: "commitDesktopUpdate", - requestId: "req-1", - }); + type: "commitDesktopUpdate", + requestId: "req-1", + ...(continueRunningThreads === undefined ? {} : { continueRunningThreads }), - yield* settle; - assert.equal(harness.quitAndInstalls(), 1); + }); + yield* settle; + assert.equal(harness.quitAndInstalls(), 1); + assert.deepEqual( + harness.restartPolicies.map(({ policy }) => policy.continueRunningThreads), + continueRunningThreads === false ? [false] : [], + ); + const statuses = reports + .filter((report) => report.requestId === "req-1") + .map((report) => report.state.status); - const statuses = reports + assert.include(statuses, "available"); - .filter((report) => report.requestId === "req-1") - .map((report) => report.state.status); - assert.include(statuses, "available"); + assert.include(statuses, "downloaded"); + }), + ); + }, + ); - assert.include(statuses, "downloaded"); - }), - ); - }); it.effect("reports a failed outcome when quitAndInstall fails", () => { const harness = makeHarness({ @@ -159,49 +169,55 @@ ); }); - it.effect("joins an in-progress install on commit instead of failing the token", () => { - const installStarted = Deferred.makeUnsafe(); - const releaseInstall = Deferred.makeUnsafe(); - const harness = makeHarness({ - stopBackend: Deferred.succeed(installStarted, undefined).pipe( + it.effect.each([undefined, false])( + "joins an active install only when the resume choice allows it: %s", - Effect.andThen(Deferred.await(releaseInstall)), - ), - }); + (continueRunningThreads) => { + const installStarted = Deferred.makeUnsafe(); + const releaseInstall = Deferred.makeUnsafe(); + const harness = makeHarness({ + stopBackend: Deferred.succeed(installStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseInstall)), + ), + }); - return runRemoteUpdatesTest(harness, ({ reports, requests, commits }) => - Effect.gen(function* () { - const updates = yield* DesktopUpdates.DesktopUpdates; - yield* Queue.offer(requests, request("req-join-commit")); - yield* settle; - harness.emit("update-available", { version: "1.2.4" }); - yield* settle; - harness.emit("update-downloaded", { version: "1.2.4" }); - yield* settle; - assert.deepEqual( - terminalReports(reports).map((report) => report.outcome), - ["ready-to-install"], + return runRemoteUpdatesTest(harness, ({ reports, requests, commits }) => + Effect.gen(function* () { + const updates = yield* DesktopUpdates.DesktopUpdates; - ); + yield* Queue.offer(requests, request("req-join-commit")); + yield* settle; + harness.emit("update-available", { version: "1.2.4" }); + yield* settle; + harness.emit("update-downloaded", { version: "1.2.4" }); + yield* settle; + assert.deepEqual( + terminalReports(reports).map((report) => report.outcome), + ["ready-to-install"], + ); + const localInstall = yield* updates.install.pipe(Effect.forkChild); - const localInstall = yield* updates.install.pipe(Effect.forkChild); - yield* Deferred.await(installStarted); - yield* Queue.offer(commits, { - version: 1, - type: "commitDesktopUpdate", + yield* Deferred.await(installStarted); + yield* Queue.offer(commits, { - requestId: "req-join-commit", - }); - yield* settle; + version: 1, + type: "commitDesktopUpdate", + requestId: "req-join-commit", + ...(continueRunningThreads === undefined ? {} : { continueRunningThreads }), + }); + yield* settle; + assert.deepEqual( + terminalReports(reports).map((report) => report.outcome), + continueRunningThreads === false + ? ["ready-to-install", "failed"] - assert.deepEqual( - terminalReports(reports).map((report) => report.outcome), - ["ready-to-install"], - ); - assert.equal(harness.quitAndInstalls(), 0); + : ["ready-to-install"], + ); + assert.equal(harness.quitAndInstalls(), 0); + yield* Deferred.succeed(releaseInstall, undefined); + yield* Fiber.join(localInstall); + }), + ); + }, - yield* Deferred.succeed(releaseInstall, undefined); + ); - yield* Fiber.join(localInstall); - }), - ); - }); it.effect("does not join a normal app quit as an update install", () => { const harness = makeHarness(); ``` ## apps/desktop/src/updates/DesktopRemoteUpdates.ts Status: modified; +5; −1 ```diff @@ -410,9 +410,13 @@ } return; } + const result = yield* updates.installPrepared( + claim.prepared.downloadedVersion, - const result = yield* updates.installPrepared(claim.prepared.downloadedVersion); + commit.continueRunningThreads, + ); if ( !result.accepted && + commit.continueRunningThreads !== false && result.state.downloadedVersion === claim.prepared.downloadedVersion && (yield* updates.isInstallActive) ) { ``` ## apps/desktop/src/updates/DesktopUpdates.test.ts Status: modified; +397; −0 ```diff @@ -498,6 +498,403 @@ }), ); + it.effect.each([true, undefined])("preserves update restart preferences for %s", (choice) => { + const harness = makeHarness(); + return Effect.scoped( + Effect.gen(function* () { + const updates = yield* DesktopUpdates.DesktopUpdates; + yield* updates.configure; + harness.emit("update-downloaded", { version: "1.2.4" }); + yield* flushCallbacks; + assert.isTrue((yield* updates.installPrepared("1.2.4", choice)).accepted); + assert.deepEqual(harness.restartPolicies, []); + }), + ).pipe(Effect.provide(Layer.merge(TestClock.layer(), harness.layer))); + }); + + it.effect("requires every backend to acknowledge OFF before any backend stops", () => + Effect.gen(function* () { + const preparing = yield* Deferred.make(); + const accepted = yield* Deferred.make(); + const stopped: string[] = []; + const instances = [ + DesktopBackendPool.PRIMARY_INSTANCE_ID, + DesktopBackendPool.BackendInstanceId("wsl:Ubuntu"), + ].map((id, index) => ({ + id, + label: Effect.succeed(id), + start: Effect.void, + stop: () => + Effect.sync(() => { + stopped.push(id); + }), + currentConfig: Effect.succeedNone, + snapshot: Effect.succeed({ + desiredRunning: true, + ready: index === 0, + activePid: Option.some(123 + index), + restartAttempt: 0, + restartScheduled: false, + }), + waitForReady: () => Effect.succeed(true), + })); + const harness = makeHarness({ + backendInstances: instances, + requestUpdateRestartPolicy: () => + Deferred.succeed(preparing, undefined).pipe(Effect.andThen(Deferred.await(accepted))), + }); + yield* Effect.scoped( + Effect.gen(function* () { + const updates = yield* DesktopUpdates.DesktopUpdates; + yield* updates.configure; + harness.emit("update-downloaded", { version: "1.2.4" }); + yield* flushCallbacks; + const installing = yield* updates.installPrepared("1.2.4", false).pipe(Effect.forkScoped); + yield* Deferred.await(preparing); + assert.deepEqual(stopped, []); + assert.equal(harness.quitAndInstalls(), 0); + assert.sameMembers( + [...harness.restartPolicies[0]!.sourceIds], + instances.map((instance) => instance.id), + ); + yield* Deferred.succeed(accepted, true); + assert.isTrue((yield* Fiber.join(installing)).accepted); + assert.sameMembers( + stopped, + instances.map((instance) => instance.id), + ); + assert.deepEqual(harness.clearedRestartPolicies, [ + harness.restartPolicies[0]!.policy.requestId, + ]); + }), + ).pipe(Effect.provide(Layer.merge(TestClock.layer(), harness.layer))); + }), + ); + + it.effect("disarms untouched backends when OFF preparation is refused", () => { + const harness = makeHarness({ + requestUpdateRestartPolicy: (_sourceIds, policy) => + Effect.succeed(policy.continueRunningThreads), + }); + return Effect.scoped( + Effect.gen(function* () { + const updates = yield* DesktopUpdates.DesktopUpdates; + const desktopState = yield* DesktopState.DesktopState; + yield* updates.configure; + harness.emit("update-downloaded", { version: "1.2.4" }); + yield* flushCallbacks; + const result = yield* updates.installPrepared("1.2.4", false); + assert.isTrue(result.failed); + assert.equal( + result.state.message, + "Every running backend must accept the desktop update restart policy before installing.", + ); + assert.equal(harness.quitAndInstalls(), 0); + assert.deepEqual( + harness.restartPolicies.map(({ policy }) => policy.continueRunningThreads), + [false, true], + ); + assert.equal( + harness.restartPolicies[0]!.policy.requestId, + harness.restartPolicies[1]!.policy.requestId, + ); + assert.isFalse(yield* Ref.get(desktopState.quitting)); + assert.deepEqual(harness.installSteps, []); + }), + ).pipe(Effect.provide(Layer.merge(TestClock.layer(), harness.layer))); + }); + + it.effect.each([true, false, undefined])( + "retries a refused rollback before another install with continuation %s", + (choice) => + Effect.gen(function* () { + const retrying = yield* Deferred.make(); + const accepted = yield* Deferred.make(); + let acceptRetry = false; + let stops = 0; + const harness = makeHarness({ + stopBackend: Effect.sync(() => { + stops += 1; + }), + requestUpdateRestartPolicy: (_sourceIds, policy) => + acceptRetry && policy.continueRunningThreads + ? Deferred.succeed(retrying, undefined).pipe(Effect.andThen(Deferred.await(accepted))) + : Effect.succeed(acceptRetry), + }); + + yield* Effect.scoped( + Effect.gen(function* () { + const updates = yield* DesktopUpdates.DesktopUpdates; + const desktopState = yield* DesktopState.DesktopState; + yield* updates.configure; + harness.emit("update-downloaded", { version: "1.2.4" }); + yield* flushCallbacks; + + assert.isTrue((yield* updates.installPrepared("1.2.4", false)).failed); + const requestId = harness.restartPolicies[0]!.policy.requestId; + assert.deepEqual(harness.clearedRestartPolicies, []); + + for (let attempt = 0; attempt < 2; attempt += 1) { + const result = yield* updates.installPrepared("1.2.4", choice); + assert.isTrue(result.failed); + assert.equal( + result.state.message, + "Could not clear the previous desktop update restart policy. Try again.", + ); + assert.isFalse(yield* Ref.get(desktopState.quitting)); + assert.equal(stops, 0); + assert.equal(harness.quitAndInstalls(), 0); + assert.deepEqual(harness.clearedRestartPolicies, []); + assert.deepEqual( + harness.restartPolicies.map(({ policy }) => policy.continueRunningThreads), + [false, ...Array(attempt + 2).fill(true)], + ); + assert.isTrue( + harness.restartPolicies.every(({ policy }) => policy.requestId === requestId), + ); + } + + acceptRetry = true; + const installing = yield* updates + .installPrepared("1.2.4", choice) + .pipe(Effect.forkScoped); + yield* Deferred.await(retrying); + assert.equal(stops, 0); + assert.equal(harness.quitAndInstalls(), 0); + assert.deepEqual(harness.clearedRestartPolicies, []); + + yield* Deferred.succeed(accepted, true); + const result = yield* Fiber.join(installing); + assert.isTrue(result.accepted); + assert.isFalse(result.failed); + assert.equal(stops, 1); + assert.equal(harness.quitAndInstalls(), 1); + assert.equal(harness.clearedRestartPolicies[0], requestId); + assert.deepEqual( + harness.restartPolicies.map(({ policy }) => policy.continueRunningThreads), + choice === false + ? [false, true, true, true, true, false] + : [false, true, true, true, true], + ); + }), + ).pipe(Effect.provide(Layer.merge(TestClock.layer(), harness.layer))); + }), + ); + + it.effect.each(["removed", "replaced", "remaining"] as const)( + "reconciles a refused rollback with current backend instances: %s", + (state) => { + const stopped: string[] = []; + let acceptRollback = false; + const original: DesktopBackendPool.DesktopBackendInstance = { + id: DesktopBackendPool.BackendInstanceId("wsl:Ubuntu"), + label: Effect.succeed("WSL (Ubuntu)"), + start: Effect.void, + stop: () => + Effect.sync(() => { + stopped.push("original"); + }), + currentConfig: Effect.succeedNone, + snapshot: Effect.succeed({ + desiredRunning: true, + ready: true, + activePid: Option.some(123), + restartAttempt: 0, + restartScheduled: false, + }), + waitForReady: () => Effect.succeed(true), + }; + const primary = { + ...original, + id: DesktopBackendPool.PRIMARY_INSTANCE_ID, + stop: () => + Effect.sync(() => { + stopped.push("primary"); + }), + snapshot: Effect.succeed({ + desiredRunning: state === "remaining", + ready: state === "remaining", + activePid: state === "remaining" ? Option.some(456) : Option.none(), + restartAttempt: 0, + restartScheduled: false, + }), + }; + const replacement = { + ...original, + stop: () => + Effect.sync(() => { + stopped.push("replacement"); + }), + }; + let currentInstances = [primary, original]; + const harness = makeHarness({ + backendInstances: currentInstances, + backendList: Effect.sync(() => currentInstances), + requestUpdateRestartPolicy: () => Effect.sync(() => acceptRollback), + }); + + return Effect.scoped( + Effect.gen(function* () { + const updates = yield* DesktopUpdates.DesktopUpdates; + const desktopState = yield* DesktopState.DesktopState; + yield* updates.configure; + harness.emit("update-downloaded", { version: "1.2.4" }); + yield* flushCallbacks; + + assert.isTrue((yield* updates.installPrepared("1.2.4", false)).failed); + const requestId = harness.restartPolicies[0]!.policy.requestId; + assert.equal(harness.restartPolicies.length, 2); + assert.deepEqual(harness.clearedRestartPolicies, []); + assert.deepEqual(stopped, []); + + currentInstances = state === "replaced" ? [primary, replacement] : [primary]; + const result = yield* updates.installPrepared("1.2.4", true); + if (state === "remaining") { + assert.isTrue(result.failed); + assert.equal( + result.state.message, + "Could not clear the previous desktop update restart policy. Try again.", + ); + assert.isFalse(yield* Ref.get(desktopState.quitting)); + assert.deepEqual(stopped, []); + assert.equal(harness.quitAndInstalls(), 0); + assert.deepEqual(harness.clearedRestartPolicies, []); + assert.deepEqual(harness.restartPolicies[2]!.sourceIds, [primary.id]); + acceptRollback = true; + assert.isFalse((yield* updates.installPrepared("1.2.4", true)).failed); + assert.deepEqual(harness.restartPolicies[3]!.sourceIds, [primary.id]); + } else { + assert.isTrue(result.accepted); + assert.isFalse(result.failed); + assert.equal(harness.restartPolicies.length, 2); + } + assert.equal(harness.quitAndInstalls(), 1); + assert.deepEqual( + stopped, + state === "replaced" ? ["primary", "replacement"] : ["primary"], + ); + assert.deepEqual(harness.clearedRestartPolicies, [requestId]); + assert.isTrue( + harness.restartPolicies.every(({ policy }) => policy.requestId === requestId), + ); + }), + ).pipe(Effect.provide(Layer.merge(TestClock.layer(), harness.layer))); + }, + ); + + it.effect( + "skips policy for an already stopped backend and refuses a stale selected version", + () => { + const harness = makeHarness({ backendDesiredRunning: false }); + return Effect.scoped( + Effect.gen(function* () { + const updates = yield* DesktopUpdates.DesktopUpdates; + yield* updates.configure; + harness.emit("update-downloaded", { version: "1.2.4" }); + yield* flushCallbacks; + assert.isFalse((yield* updates.installPrepared("1.2.3", false)).accepted); + assert.isTrue((yield* updates.installPrepared("1.2.4", false)).accepted); + assert.deepEqual(harness.restartPolicies, []); + }), + ).pipe(Effect.provide(Layer.merge(TestClock.layer(), harness.layer))); + }, + ); + + it.effect.each(["changed", "missing", "membership"] as const)( + "refuses OFF when backend state changes: %s", + (state) => { + let pid = state === "missing" ? Option.none() : Option.some(123); + let stops = 0; + let prepared = false; + const instance: DesktopBackendPool.DesktopBackendInstance = { + id: DesktopBackendPool.PRIMARY_INSTANCE_ID, + label: Effect.succeed("Windows"), + start: Effect.void, + stop: () => + Effect.sync(() => { + stops += 1; + }), + currentConfig: Effect.succeedNone, + snapshot: Effect.sync(() => ({ + desiredRunning: true, + ready: true, + activePid: pid, + restartAttempt: 0, + restartScheduled: false, + })), + waitForReady: () => Effect.succeed(true), + }; + const harness = makeHarness({ + backendInstances: [instance], + backendList: Effect.sync(() => + state === "membership" && prepared + ? [instance, { ...instance, id: DesktopBackendPool.BackendInstanceId("wsl:Ubuntu") }] + : [instance], + ), + requestUpdateRestartPolicy: () => + Effect.sync(() => { + prepared = true; + if (state === "changed") pid = Option.some(456); + return true; + }), + }); + return Effect.scoped( + Effect.gen(function* () { + const updates = yield* DesktopUpdates.DesktopUpdates; + yield* updates.configure; + harness.emit("update-downloaded", { version: "1.2.4" }); + yield* flushCallbacks; + const result = yield* updates.installPrepared("1.2.4", false); + assert.isTrue(result.failed); + assert.equal( + result.state.message, + state === "missing" + ? "Wait for every desktop backend to start before disabling thread continuation." + : state === "membership" + ? "Desktop backends changed while preparing the update. Try again." + : "A desktop backend restarted while preparing the update. Try again.", + ); + assert.equal(stops, 0); + assert.equal(harness.quitAndInstalls(), 0); + assert.deepEqual( + harness.restartPolicies.map(({ policy }) => policy.continueRunningThreads), + state === "missing" ? [] : [false, true], + ); + }), + ).pipe(Effect.provide(Layer.merge(TestClock.layer(), harness.layer))); + }, + ); + + it.effect("keeps OFF for stopped backends when the native installer throws", () => { + const harness = makeHarness({ + quitAndInstall: Effect.fail( + new ElectronUpdater.ElectronUpdaterQuitAndInstallError({ + channel: "latest", + isSilent: true, + isForceRunAfter: true, + cause: new Error("installer refused"), + }), + ), + }); + return Effect.scoped( + Effect.gen(function* () { + const updates = yield* DesktopUpdates.DesktopUpdates; + yield* updates.configure; + harness.emit("update-downloaded", { version: "1.2.4" }); + yield* flushCallbacks; + assert.isTrue((yield* updates.installPrepared("1.2.4", false)).failed); + assert.deepEqual( + harness.restartPolicies.map(({ policy }) => policy.continueRunningThreads), + [false], + ); + assert.deepEqual(harness.clearedRestartPolicies, [ + harness.restartPolicies[0]!.policy.requestId, + ]); + assert.deepEqual(harness.installSteps, ["quitAndInstall", "startBackend"]); + }), + ).pipe(Effect.provide(Layer.merge(TestClock.layer(), harness.layer))); + }); + it.effect("keeps windows and restarts backends when quitAndInstall fails", () => { const harness = makeHarness({ quitAndInstall: Effect.fail( ``` ## apps/desktop/src/updates/DesktopUpdates.ts Status: modified; +150; −8 ```diff @@ -10,6 +10,7 @@ import { errorTag } from "@t3tools/shared/observability"; import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; @@ -33,6 +34,7 @@ import * as ElectronWindow from "../electron/ElectronWindow.ts"; import * as IpcChannels from "../ipc/channels.ts"; import * as DesktopAppSettings from "../settings/DesktopAppSettings.ts"; +import * as DesktopTelemetryPublisher from "../telemetry/DesktopTelemetryPublisher.ts"; import { normalizeDesktopUpdateReleaseNotes } from "./releaseNotes.ts"; import { resolveDefaultDesktopUpdateChannel } from "./updateChannels.ts"; import { @@ -158,6 +160,11 @@ } } +class DesktopUpdatePreparationError extends Schema.TaggedError()( + "DesktopUpdatePreparationError", + { message: Schema.String }, +) {} + export type DesktopUpdateConfigureError = never; export const DesktopUpdateSetChannelError = Schema.Union([ @@ -190,7 +197,8 @@ readonly download: Effect.Effect; readonly install: Effect.Effect; readonly installPrepared: ( + expectedVersion?: string, - expectedVersion: string, + continueRunningThreads?: boolean, ) => Effect.Effect; } >()("@t3tools/desktop/updates/DesktopUpdates") {} @@ -345,12 +353,21 @@ const environment = yield* DesktopEnvironment.DesktopEnvironment; const fileSystem = yield* FileSystem.FileSystem; const desktopSettings = yield* DesktopAppSettings.DesktopAppSettings; + const telemetry = yield* DesktopTelemetryPublisher.DesktopTelemetryPublisher; + const crypto = yield* Crypto.Crypto; const appUpdateYmlConfigRef = yield* Ref.make>(Option.none()); const activeUpdateActionRef = yield* Ref.make>(Option.none()); const installRunningBackendsRef = yield* Ref.make< ReadonlyArray >([]); + const installRestartPolicyRef = yield* Ref.make< + Option.Option<{ + readonly requestId: string; + readonly instances: readonly DesktopBackendPool.DesktopBackendInstance[]; + readonly stopped: ReadonlySet; + }> + >(Option.none()); const finishedUpdateActions = yield* PubSub.unbounded(); const updaterConfiguredRef = yield* Ref.make(false); const lastLoggedDownloadMilestoneRef = yield* Ref.make(-1); @@ -642,6 +659,38 @@ })), ); + const releaseInstallRestartPolicy = Effect.fn("desktop.updates.releaseInstallRestartPolicy")( + function* (disarmUntouched: boolean) { + const policy = yield* Ref.get(installRestartPolicyRef); + if (Option.isNone(policy)) return true; + const { requestId, instances, stopped } = policy.value; + const currentInstances = yield* pool.list; + const untouched = instances.filter( + (instance) => !stopped.has(instance) && currentInstances.includes(instance), + ); + if (disarmUntouched && untouched.length > 0) { + const accepted = yield* telemetry.requestUpdateRestartPolicy( + untouched.map((instance) => instance.id), + { + version: 1, + type: "desktopUpdateRestartPolicy", + requestId, + continueRunningThreads: true, + }, + ); + if (!accepted) { + yield* logUpdaterError( + "Desktop update could not clear the restart policy on every untouched backend.", + ); + return false; + } + } + yield* telemetry.clearUpdateRestartPolicy(requestId); + yield* Ref.set(installRestartPolicyRef, Option.none()); + return true; + }, + ); + const resetInstallAction = Effect.all( [ finishUpdateAction("install"), @@ -663,6 +712,7 @@ ); if (!ownsRecovery) return; + yield* releaseInstallRestartPolicy(true); yield* Ref.set(desktopState.quitting, false); yield* removeUpdateRestartMarker; yield* removeInstallRecord; @@ -687,7 +737,7 @@ ); }); + const installDownloadedUpdate = (expectedVersion?: string, continueRunningThreads?: boolean) => - const installDownloadedUpdate = (expectedVersion?: string) => Effect.scoped( Effect.gen(function* () { const actionCompletions = yield* PubSub.subscribe(finishedUpdateActions); @@ -739,6 +789,16 @@ yield* Ref.set(installRunningBackendsRef, []); return yield* Effect.gen(function* () { + if (!(yield* releaseInstallRestartPolicy(true))) { + yield* updateState((current) => + reduceDesktopUpdateStateOnInstallFailure( + current, + "Could not clear the previous desktop update restart policy. Try again.", + ), + ); + yield* resetInstallAction; + return { accepted: true, completed: false, failed: true }; + } yield* writeUpdateRestartMarker; const { downloadedVersion } = yield* Ref.get(updateStateRef); if (downloadedVersion !== null) yield* writeInstallRecord(downloadedVersion); @@ -749,22 +809,100 @@ // hard-killed by the OS instead of receiving SIGTERM + grace. const instances = yield* pool.list; // stop clears desiredRunning, so retain the recovery set beforehand. - const runningInstances = yield* Effect.filter(instances, (instance) => + const backendSnapshots = yield* Effect.forEach(instances, (instance) => + instance.snapshot.pipe(Effect.map((snapshot) => ({ instance, snapshot }))), - instance.snapshot.pipe(Effect.map((snapshot) => snapshot.desiredRunning)), ); + const runningInstances = backendSnapshots + .filter(({ snapshot }) => snapshot.desiredRunning) + .map(({ instance }) => instance); + if (continueRunningThreads === false) { + const affected = backendSnapshots.filter( + ({ snapshot }) => snapshot.desiredRunning || Option.isSome(snapshot.activePid), + ); + if (affected.some(({ snapshot }) => Option.isNone(snapshot.activePid))) { + return yield* new DesktopUpdatePreparationError({ + message: + "Wait for every desktop backend to start before disabling thread continuation.", + }); + } + if (affected.length > 0) { + const requestId = yield* crypto.randomUUIDv4().pipe(Effect.orDie); + yield* Ref.set( + installRestartPolicyRef, + Option.some({ + requestId, + instances: affected.map(({ instance }) => instance), + stopped: new Set(), + }), + ); + const accepted = yield* telemetry.requestUpdateRestartPolicy( + affected.map(({ instance }) => instance.id), + { + version: 1, + type: "desktopUpdateRestartPolicy", + requestId, + continueRunningThreads: false, + }, + ); + if (!accepted) + return yield* new DesktopUpdatePreparationError({ + message: + "Every running backend must accept the desktop update restart policy before installing.", + }); + } + const currentInstances = yield* pool.list; + if ( + currentInstances.length !== instances.length || + currentInstances.some((instance) => !instances.includes(instance)) + ) { + return yield* new DesktopUpdatePreparationError({ + message: "Desktop backends changed while preparing the update. Try again.", + }); + } + for (const { instance, snapshot: before } of backendSnapshots) { + const after = yield* instance.snapshot; + if ( + Option.getOrUndefined(before.activePid) !== + Option.getOrUndefined(after.activePid) || + before.desiredRunning !== after.desiredRunning + ) { + return yield* new DesktopUpdatePreparationError({ + message: "A desktop backend restarted while preparing the update. Try again.", + }); + } + } + } + const activeAction = yield* Ref.get(activeUpdateActionRef); + if (Option.isNone(activeAction) || activeAction.value !== "install") + return { accepted: true, completed: false, failed: true }; yield* Ref.set(installRunningBackendsRef, runningInstances); yield* Effect.forEach( instances, + (instance) => + Ref.update( + installRestartPolicyRef, + Option.map((policy) => ({ + ...policy, + stopped: new Set([...policy.stopped, instance]), + })), + ).pipe(Effect.andThen(instance.stop({ timeout: Duration.seconds(5) }))), - (instance) => instance.stop({ timeout: Duration.seconds(5) }), { concurrency: "unbounded" }, ); yield* electronUpdater.quitAndInstall({ isSilent: true, isForceRunAfter: true, }); + yield* releaseInstallRestartPolicy(false); return { accepted: true, completed: false, failed: false }; }).pipe( Effect.catchTags({ + DesktopUpdatePreparationError: Effect.fn("desktop.updates.handlePreparationFailure")( + function* (error) { + yield* recoverFailedInstall(error.message); + yield* logUpdaterError(error.message, { errorTag: error._tag }); + return { accepted: true, completed: false, failed: true }; + }, + ), ElectronUpdaterQuitAndInstallError: Effect.fn("desktop.updates.handleInstallFailure")( function* (error) { yield* recoverFailedInstall(error.message); @@ -779,7 +917,9 @@ }, ), }), + Effect.onInterrupt(() => + releaseInstallRestartPolicy(true).pipe(Effect.andThen(resetInstallAction)), + ), - Effect.onInterrupt(() => resetInstallAction), Effect.catchCause((cause) => Effect.gen(function* () { if (Cause.hasInterruptsOnly(cause)) { @@ -801,6 +941,7 @@ const installWithExpectedVersion = Effect.fn("desktop.updates.install")(function* ( expectedVersion?: string, + continueRunningThreads?: boolean, ) { if (yield* Ref.get(desktopState.quitting)) { return { @@ -810,7 +951,7 @@ state: yield* Ref.get(updateStateRef), }; } + const result = yield* installDownloadedUpdate(expectedVersion, continueRunningThreads); - const result = yield* installDownloadedUpdate(expectedVersion); return { accepted: result.accepted, completed: result.completed, @@ -1168,7 +1309,8 @@ install: installWithExpectedVersion().pipe( Effect.map(({ accepted, completed, state }) => ({ accepted, completed, state })), ), - installPrepared: (expectedVersion) => installWithExpectedVersion(expectedVersion), + installPrepared: (expectedVersion, continueRunningThreads) => + installWithExpectedVersion(expectedVersion, continueRunningThreads), }); }); ``` ## apps/desktop/src/updates/updatesTestHarness.ts Status: modified; +53; −3 ```diff @@ -1,5 +1,5 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; -import type { DesktopUpdateState } from "@t3tools/contracts"; +import type { DesktopUpdateState, DesktopUpdateRestartPolicy } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as PlatformError from "effect/PlatformError"; @@ -14,6 +14,8 @@ import * as DesktopAppSettings from "../settings/DesktopAppSettings.ts"; import * as DesktopState from "../app/DesktopState.ts"; import * as DesktopUpdates from "./DesktopUpdates.ts"; +import * as Stream from "effect/Stream"; +import * as DesktopTelemetryPublisher from "../telemetry/DesktopTelemetryPublisher.ts"; export const flushCallbacks = Effect.yieldNow; @@ -28,9 +30,14 @@ readonly downloadUpdate?: Effect.Effect; readonly quitAndInstall?: Effect.Effect; readonly stopBackend?: Effect.Effect; + readonly requestUpdateRestartPolicy?: ( + sourceIds: readonly string[], + policy: DesktopUpdateRestartPolicy, + ) => Effect.Effect; readonly startBackend?: Effect.Effect; readonly backendDesiredRunning?: boolean; readonly backendInstances?: ReadonlyArray; + readonly backendList?: Effect.Effect; readonly env?: Record; readonly platform?: NodeJS.Platform; readonly packageType?: string | undefined; @@ -142,15 +149,26 @@ snapshot: Effect.sync(() => ({ desiredRunning: backendDesiredRunning, ready: false, - activePid: Option.none(), + activePid: backendDesiredRunning ? Option.some(123) : Option.none(), restartAttempt: 0, restartScheduled: false, })), waitForReady: () => Effect.succeed(true), }; - const layerBackend = DesktopBackendPool.layerTest( + const layerBackendBase = DesktopBackendPool.layerTest( options.backendInstances ?? [stubBackendInstance], ); + const backendList = options.backendList; + const layerBackend = + backendList === undefined + ? layerBackendBase + : Layer.effect( + DesktopBackendPool.DesktopBackendPool, + Effect.map(DesktopBackendPool.DesktopBackendPool, (pool) => ({ + ...pool, + list: backendList, + })), + ).pipe(Layer.provide(layerBackendBase)); const layerEnvironment = DesktopEnvironment.layer({ dirname: "/repo/apps/desktop/src", @@ -249,7 +267,37 @@ }), }); + const restartPolicies: Array<{ + sourceIds: readonly string[]; + policy: DesktopUpdateRestartPolicy; + }> = []; + const clearedRestartPolicies: string[] = []; + const layerPublisher = Layer.succeed(DesktopTelemetryPublisher.DesktopTelemetryPublisher, { + latest: Effect.succeedNone, + changes: Stream.empty, + encoded: () => Stream.empty, + handleControlForSource: () => Effect.void, + removeControlSource: () => Effect.void, + publishUpdateReport: () => Effect.void, + updateRequests: Stream.empty, + updateCommits: Stream.empty, + updateCancellations: Stream.empty, + requestUpdateRestartPolicy: (sourceIds, policy) => + Effect.sync(() => { + restartPolicies.push({ sourceIds, policy }); + }).pipe( + Effect.andThen( + options.requestUpdateRestartPolicy?.(sourceIds, policy) ?? Effect.succeed(true), + ), + ), + clearUpdateRestartPolicy: (requestId) => + Effect.sync(() => { + clearedRestartPolicies.push(requestId); + }), + }); + const layer = DesktopUpdates.layer.pipe( + Layer.provideMerge(layerPublisher), Layer.provide(layerFileSystem), Layer.provideMerge(layerUpdater), Layer.provideMerge(layerWindow), @@ -273,6 +321,8 @@ checkCount: () => checkCount, quitAndInstalls: () => quitAndInstallCount, installSteps, + restartPolicies, + clearedRestartPolicies, updateRestartMarkers, installRecords, previousInstallRecord: () => previousInstallRecord, ``` ## apps/desktop/src/wsl/DesktopWslEnvironment.test.ts Status: modified; +59; −0 ```diff @@ -163,6 +163,65 @@ ].join("\n"), ); + it("checks the final executable once on each cold and warm startup", () => { + const fixture = createFixture( + '#!/bin/sh\n[ "$1" = "--version" ] || exit 1\nprintf "%s\\n" "$0" >> "$HOME/runtime-executions"\n', + ); + + for (const expected of [1, 2]) { + const installed = fixture.install(); + expect(installed.status, installed.stderr).toBe(0); + const probe = probeFixture(fixture); + expect(probe.status, probe.stderr).toBe(0); + expect(runShell(`cat ${sh(`${fixture.work}/home/runtime-executions`)}`).stdout).toBe( + `${fixture.serverEntry}\n`.repeat(expected), + ); + } + }); + + it("never marks a runtime ready when its executable fails at the final path", () => { + const fixture = createFixture( + '#!/bin/sh\ncase "$0" in *.tmp.*) exit 0 ;; esac\nprintf "final-path loader failure\\n" >&2\nexit 42\n', + ); + + const failed = fixture.install(); + + expect(failed.status).not.toBe(0); + expect(failed.stderr).toContain("final-path loader failure"); + expect(parseWslRuntimeRoot(failed.stdout)).toBeNull(); + expect(runShell(`[ ! -e ${sh(fixture.runtimeRoot)} ]`).status).toBe(0); + }); + + it("rejects an executable removed after runtime preparation", () => { + const fixture = createFixture(); + expect(fixture.install().status).toBe(0); + expect(runShell(`rm ${sh(fixture.serverEntry)}`).status).toBe(0); + + expect(probeFixture(fixture).status).not.toBe(0); + }); + + it("retains a colliding cache when the install did not promote it", () => { + const fixture = createFixture( + '#!/bin/sh\n[ ! -e "$HOME/checked" ] || exit 42\ntouch "$HOME/checked"\n', + ); + const failed = runShell( + [ + "mv() {", + ' mkdir "$3"', + ' cp -R "$2/." "$3"', + ' printf borrowed > "$3/owner"', + ' sha256sum "$3/t3" | cut -d " " -f 1 > "$3/.t3code-wsl-runtime-ready"', + " return 1", + "}", + fixture.installScript(), + ].join("\n"), + ); + + expect(failed.status).not.toBe(0); + expect(parseWslRuntimeRoot(failed.stdout)).toBeNull(); + expect(runShell(`cat ${sh(`${fixture.runtimeRoot}/owner`)}`).stdout).toBe("borrowed"); + }); + it("discovers version-managed Node for providers with a standalone runtime", () => { const fixture = createFixture(); expect(fixture.install().status).toBe(0); ``` ## apps/desktop/src/wsl/DesktopWslEnvironment.ts Status: modified; +21; −14 ```diff @@ -334,29 +334,36 @@ " fi", "fi", `runtime_tmp=$(mktemp -d "$runtime_parent/.${safeRuntimeId}.tmp.XXXXXX")`, + "runtime_promoted=0", + "cleanup_runtime_install() {", + ' rm -rf "$runtime_tmp"', + ' if [ "$runtime_promoted" = 1 ]; then', + ' rm -rf "$runtime_root"', - 'cleanup_runtime_install() { rm -rf "$runtime_tmp"; }', + " fi", + "}", "trap cleanup_runtime_install EXIT", `tar -xzf ${shellQuote(linuxArchivePath)} -C "$runtime_tmp" --strip-components=1`, - 'if ! runtime_entry_runs "$runtime_tmp"; then', + 'if mv -T "$runtime_tmp" "$runtime_root" 2>/dev/null; then', + " runtime_promoted=1", + "elif runtime_is_ready; then", + ' rm -rf "$runtime_tmp"', + "else", + ` printf 'Could not promote WSL runtime cache at %s\\n' "$runtime_root" >&2`, + " exit 1", + "fi", + 'if ! runtime_entry_runs "$runtime_root"; then', " printf 'WSL runtime archive does not contain a working t3 executable\\n' >&2", " exit 1", "fi", - 'installed_entry_digest=$(runtime_server_entry_digest "$runtime_tmp")', + 'installed_entry_digest=$(runtime_server_entry_digest "$runtime_root")', 'if [ -z "$installed_entry_digest" ]; then', " printf 'Could not hash the WSL runtime server entry\\n' >&2", " exit 1", "fi", + `printf '%s\\n' "$installed_entry_digest" > "$ready_marker"`, - `printf '%s\\n' "$installed_entry_digest" > "$runtime_tmp/${WSL_RUNTIME_READY_MARKER}"`, - 'if mv -T "$runtime_tmp" "$runtime_root" 2>/dev/null; then', - " :", - "elif runtime_is_ready; then", - ' rm -rf "$runtime_tmp"', - "else", - ` printf 'Could not promote WSL runtime cache at %s\\n' "$runtime_root" >&2`, - " exit 1", - "fi", `touch "$runtime_root/${WSL_RUNTIME_SELECTED_MARKER}"`, `printf 'runtimeRoot:%s\\n' "$runtime_root"`, + "runtime_promoted=0", ].join("\n"); }; @@ -485,7 +492,7 @@ export const buildWslRuntimeProbeScript = (linuxAppRoot: string) => [ `bash -lc ${shellQuote(`${buildWslNodeEnvPreamble()}${RESOLVED_PATH_LINE}`)} 2>/dev/null || ${RESOLVED_PATH_LINE}`, - `${shellQuote(`${linuxAppRoot}/t3`)} --version >/dev/null 2>&1`, + `[ -x ${shellQuote(`${linuxAppRoot}/t3`)} ]`, ].join("\n"); const TOOLCHAIN_CHECK_SCRIPT = [ @@ -622,7 +629,7 @@ const trimmedTail = probe.stderr.trim().slice(-500); return { ok: false, - reason: `${linuxAppRoot}/t3 --version failed (exit ${probe.exitCode})${trimmedTail ? `: ${trimmedTail}` : ""}`, + reason: `${linuxAppRoot}/t3 is not executable (exit ${probe.exitCode})${trimmedTail ? `: ${trimmedTail}` : ""}`, } as const; } const resolvedPath = parseResolvedPath(probe.stdout); ``` ## apps/server/src/cloud/pinnedRuntime.test.ts Status: modified; +126; −2 ```diff @@ -67,6 +67,130 @@ }); it.layer(NodeServices.layer)("ensurePinnedRuntimeInstalled", (it) => { + it.effect("downloads the archive while release checksums are pending", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const baseDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pinned-overlap-" }); + const checksumsRequested = yield* Deferred.make(); + const archiveRequested = yield* Deferred.make(); + const releaseChecksums = yield* Deferred.make(); + const checksums = yield* validChecksums; + const commands: string[] = []; + const client = HttpClient.make((request) => + Effect.gen(function* () { + const isChecksums = request.url.endsWith("/SHA256SUMS"); + yield* Deferred.succeed(isChecksums ? checksumsRequested : archiveRequested, undefined); + if (isChecksums) yield* Deferred.await(releaseChecksums); + return HttpClientResponse.fromWeb( + request, + new Response(isChecksums ? checksums : archiveBytes), + ); + }), + ); + const install = yield* ensurePinnedRuntimeInstalled({ + baseDir, + version, + fs, + path, + platform: "linux", + arch: "x64", + httpClient: client, + runner: extractingRunner(fs, path, commands), + validate: () => Effect.void, + }).pipe(Effect.forkScoped); + yield* Deferred.await(checksumsRequested); + yield* TestClock.adjust(Duration.zero); + assert.isTrue(yield* Deferred.isDone(archiveRequested)); + assert.deepEqual(commands, []); + const [staging] = yield* fs.readDirectory(path.join(baseDir, "runtime", "versions")); + assert.isDefined(staging); + assert.deepEqual( + yield* fs.readDirectory(path.join(baseDir, "runtime", "versions", staging)), + [], + ); + yield* Deferred.succeed(releaseChecksums, undefined); + const installed = yield* Fiber.join(install); + assert.equal(yield* fs.readFileString(installed.sentinelPath), `${version}\n`); + assert.deepEqual(commands, ["tar"]); + }), + ); + + it.effect.each([ + { + name: "the checksums request fails", + failedAsset: "checksums", + status: 503, + step: "downloading the t3 release checksums", + }, + { + name: "the archive request fails", + failedAsset: "archive", + status: 503, + step: "downloading the t3 release archive", + }, + { + name: "the checksum entry is missing", + failedAsset: "checksums", + status: 200, + step: `finding ${archiveName} in the t3 release checksums`, + }, + ])("cancels the other download when $name", ({ failedAsset, status, step }) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const baseDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pinned-download-failed-" }); + const bodyStarted = yield* Deferred.make(); + let cancelled = false; + const client = HttpClient.make((request) => { + const asset = request.url.endsWith("/SHA256SUMS") ? "checksums" : "archive"; + if (asset === failedAsset) { + return Deferred.await(bodyStarted).pipe( + Effect.as( + HttpClientResponse.fromWeb( + request, + new Response(status === 200 ? "" : null, { status }), + ), + ), + ); + } + return Effect.succeed( + HttpClientResponse.fromWeb( + request, + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(archiveBytes.slice(0, 1)); + }, + pull() { + Deferred.doneUnsafe(bodyStarted, Effect.void); + }, + cancel() { + cancelled = true; + }, + }), + ), + ), + ); + }); + const error = yield* ensurePinnedRuntimeInstalled({ + baseDir, + version, + fs: { ...fs, writeFile: () => Effect.die("must not write an incomplete archive") }, + path, + platform: "linux", + arch: "x64", + httpClient: client, + runner: { run: () => Effect.die("must not extract an incomplete archive") }, + validate: () => Effect.die("must not validate an incomplete archive"), + }).pipe(Effect.flip); + assert.instanceOf(error, PinnedRuntimeInstallError); + assert.equal(error.step, step); + assert.isTrue(cancelled); + assert.deepEqual(yield* fs.readDirectory(path.join(baseDir, "runtime", "versions")), []); + }), + ); + it.effect("installs the verified release archive as the runtime executable", () => Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; @@ -155,7 +279,7 @@ }), ); - it.effect("refuses an archive whose checksum does not match the release", () => + it.effect("refuses to write or extract an archive whose checksum does not match", () => Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; const path = yield* Path.Path; @@ -164,7 +288,7 @@ const error = yield* ensurePinnedRuntimeInstalled({ baseDir, version, - fs, + fs: { ...fs, writeFile: () => Effect.die("must not write an unverified archive") }, path, platform: "linux", arch: "x64", ``` ## apps/server/src/cloud/pinnedRuntime.ts Status: modified; +22; −16 ```diff @@ -187,26 +187,32 @@ const fileName = cliArchiveFileName(input.version, platformKey); input.onProgress?.({ stage: "download", received: 0, total: undefined }); + const [expected, archive] = yield* Effect.all( + [ - const checksums = parseChecksums( - new TextDecoder().decode( - yield* fetchReleaseAsset( + fetchReleaseAsset( httpClient, `${baseUrl}/${CLI_RELEASE_CHECKSUMS_FILE}`, "downloading the t3 release checksums", + ).pipe( + Effect.flatMap((bytes) => { + const expected = parseChecksums(new TextDecoder().decode(bytes)).get(fileName); + return expected === undefined + ? Effect.fail( + new PinnedRuntimeInstallError({ + step: `finding ${fileName} in the t3 release checksums`, + }), + ) + : Effect.succeed(expected); + }), ), - ), - ); + fetchReleaseAsset( + httpClient, + `${baseUrl}/${fileName}`, - const expected = checksums.get(fileName); - if (expected === undefined) { + "downloading the t3 release archive", + input.onProgress, + ), + ], + { concurrency: "unbounded" }, - return yield* new PinnedRuntimeInstallError({ - step: `finding ${fileName} in the t3 release checksums`, - }); - } - const archive = yield* fetchReleaseAsset( - httpClient, - `${baseUrl}/${fileName}`, - "downloading the t3 release archive", - input.onProgress, ); input.onProgress?.({ stage: "verify" }); const digest = yield* Effect.tryPromise({ ``` ## apps/server/src/cloud/selfUpdate.test.ts Status: modified; +42; −0 ```diff @@ -20,8 +20,10 @@ import * as ServiceLauncherClient from "./serviceLauncherClient.ts"; import { SERVICE_LAUNCHER_PROTOCOL } from "./serviceProtocol.ts"; import * as ServerSelfUpdate from "./selfUpdate.ts"; +import { UpdateRestartPolicy } from "../updateRestartPolicy.ts"; interface HarnessOptions { + readonly policy?: Effect.Success>; readonly mode?: "web" | "desktop"; readonly managed?: boolean; readonly preflight?: "ready" | "blocked"; @@ -111,7 +113,9 @@ const config = yield* ServerConfig.ServerConfig.pipe( Effect.provide(ServerConfig.layerTest(process.cwd(), baseDir)), ); + const policy = yield* UpdateRestartPolicy; const selfUpdate = yield* ServerSelfUpdate.make().pipe( + Effect.provideService(UpdateRestartPolicy, options.policy ?? policy), Effect.provideService(ProcessRunner.ProcessRunner, runner), Effect.provideService(ServiceLauncherClient.ServiceLauncherClient, launcher), Effect.provideService( @@ -130,6 +134,44 @@ }); it.layer(NodeServices.layer)("server self update", (it) => { + it.effect("persists OFF before handoff and clears a refused handoff", () => + Effect.gen(function* () { + const events: boolean[] = []; + const { selfUpdate } = yield* makeHarness({ + policy: { + set: (_id, enabled) => Effect.sync(() => void events.push(enabled)), + capture: () => Effect.void, + isVetoed: () => Effect.succeed(false), + }, + requestUpdate: () => + Effect.sync(() => { + expect(events).toEqual([false]); + return "accepted"; + }), + }); + yield* selfUpdate.update({ targetVersion: "1.1.0", continueRunningThreads: false }); + expect(events).toEqual([false]); + const refused = yield* makeHarness({ + policy: { + set: (_id, enabled) => Effect.sync(() => void events.push(enabled)), + capture: () => Effect.void, + isVetoed: () => Effect.succeed(false), + }, + requestUpdate: () => + Effect.fail( + new ServiceLauncherClient.ServiceLauncherRejectedError({ + targetVersion: "1.1.0", + reason: "refused", + }), + ), + }); + yield* refused.selfUpdate + .update({ targetVersion: "1.1.0", continueRunningThreads: false }) + .pipe(Effect.flip); + expect(events).toEqual([false, false, true]); + }), + ); + it.effect("marks running threads at the boot-service handoff", () => Effect.gen(function* () { const events: string[] = []; ``` ## apps/server/src/cloud/selfUpdate.ts Status: modified; +46; −4 ```diff @@ -11,6 +11,7 @@ import * as Cause from "effect/Cause"; import * as Config from "effect/Config"; import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as HashSet from "effect/HashSet"; @@ -29,6 +30,7 @@ import * as ServerConfig from "../config.ts"; import * as DesktopAppUpdate from "../desktopUpdate/DesktopAppUpdate.ts"; import * as ProcessRunner from "../processRunner.ts"; +import { UpdateRestartPolicy } from "../updateRestartPolicy.ts"; import { ensurePinnedRuntimeInstalled, pinnedRuntimeCommand, @@ -216,6 +218,8 @@ yield* Config.String(CLI_RELEASE_BASE_URL_ENV).pipe(Config.option), ); const inFlight = yield* Ref.make(false); + const restartPolicy = yield* UpdateRestartPolicy; + const crypto = yield* Crypto.Crypto; yield* pruneRuntimesOnStart({ baseDir: serverConfig.baseDir, managed: launcher.managed }).pipe( Effect.forkScoped, @@ -233,7 +237,7 @@ )(function* (input, reportProgress = () => Effect.void, onHandoffAccepted = () => Effect.void) { if (capability === "desktop-managed") { if (desktopAppUpdate.available) { - return yield* desktopAppUpdate.run(reportProgress); + return yield* desktopAppUpdate.run(reportProgress, input.continueRunningThreads); } return yield* failWith( "This server is managed by the T3 Code desktop app on its machine; update the desktop app to update it.", @@ -253,6 +257,8 @@ return yield* failWith("A server update is already in progress."); } + let policyRequestId: string | undefined; + let handoffAccepted = false; return yield* Effect.gen(function* () { yield* reportProgress("downloading"); const paths = yield* ensurePinnedRuntimeInstalled({ @@ -344,7 +350,25 @@ yield* reportProgress("installing"); const updateId = yield* Effect.uninterruptible( + Effect.gen(function* () { + if (input.continueRunningThreads === false) { + policyRequestId = yield* crypto + .randomUUIDv4() + .pipe( + Effect.mapError((cause) => + failWith("Could not prepare the update resume choice.", cause), - launcher.requestUpdate({ targetVersion, dbPath: serverConfig.dbPath }).pipe( + ), + ); + yield* restartPolicy + .set(policyRequestId, false) + .pipe( + Effect.mapError((cause) => + failWith("Could not persist the update resume choice.", cause), + ), + ); + } + return yield* launcher.requestUpdate({ targetVersion, dbPath: serverConfig.dbPath }); + }).pipe( Effect.mapError((error) => failWith( error._tag === "ServiceLauncherRejectedError" @@ -353,7 +377,10 @@ error, ), ), + Effect.tap(() => { + handoffAccepted = true; + return onHandoffAccepted(); + }), - Effect.tap(() => onHandoffAccepted()), ), ); @@ -363,7 +390,22 @@ runtimePath: paths.entryPath, }); return { targetVersion, method: "boot-service" as const, updateId }; + }).pipe( + Effect.catchCause((cause) => + Effect.gen(function* () { + if (policyRequestId !== undefined && !handoffAccepted) { + yield* restartPolicy + .set(policyRequestId, true) + .pipe( - }).pipe(Effect.onError(() => Ref.set(inFlight, false))); + Effect.mapError((cause) => + failWith("Could not clear the update resume choice.", cause), + ), + ); + } + return yield* Effect.failCause(cause); + }).pipe(Effect.ensuring(Ref.set(inFlight, false))), + ), + ); }); return ServerSelfUpdate.of({ ``` [Next page](https://git.cafe/pingdotgg/t3code/pull/23870/diffs?after=ImFwcHMvc2VydmVyL3NyYy9jbG91ZC9zZWxmVXBkYXRlLnRzIg&expectedVersion=57&format=markdown)