diff --git a/packages/features/src/index.ts b/packages/features/src/index.ts index bc5029955d..29b5253a7d 100644 --- a/packages/features/src/index.ts +++ b/packages/features/src/index.ts @@ -125,6 +125,7 @@ export { export { HostedMacCommandClient, type HostedMacCommandClientOptions, + MAC_COMMAND_NOT_CLAIMED, } from "./mac-command-client.js"; export { automationPath, diff --git a/packages/features/src/mac-command-client.test.ts b/packages/features/src/mac-command-client.test.ts index c1899e4965..e8ead8b4fa 100644 --- a/packages/features/src/mac-command-client.test.ts +++ b/packages/features/src/mac-command-client.test.ts @@ -1,28 +1,21 @@ /** * mac-command-client.test.ts -- the Mac's side of the commands its account's agents queue, as the Mac observes it. * - * The held read names the kinds this Mac runs and answers the commands the - * service claimed for it, a command of a kind this build does not know - * read as its id and kind alone rather than spoiling the read, and it - * outlasts the service's hold before it gives up as nothing; a result is - * posted as the service reads it, and is taken or not. + * The live read names the kinds this Mac runs and the account it is for, + * and reads a refused session as signed out; a claim answers the command + * granted or nothing; a result is posted as the service reads it, and is + * taken or not. * * Synthetic bearers, ids, and commands throughout. */ import assert from "node:assert/strict"; import { it } from "@effect/vitest"; -import { - fakeCloudApi, - HTTP_STATUS, - type RecordedRequest, - recordingHttpClient, -} from "@luke/wire/testing"; -import { Effect, Fiber } from "effect"; -import { TestClock } from "effect/testing"; +import { fakeCloudApi, HTTP_STATUS } from "@luke/wire/testing"; +import { Effect } from "effect"; import { accountBearer } from "./account-call.js"; import { LIVE_SIGNED_OUT } from "./live-wire.js"; -import { HostedMacCommandClient } from "./mac-command-client.js"; +import { HostedMacCommandClient, MAC_COMMAND_NOT_CLAIMED } from "./mac-command-client.js"; import { MAC_COMMAND_KIND, type MacCommand, type MacCommandResult } from "./mac-command-wire.js"; import { HOSTED_API_ERROR } from "./service-wire.js"; @@ -36,8 +29,6 @@ const COMMAND: MacCommand = { timeoutSeconds: 60, }; -const NEWER_ID = "22222222-2222-4222-8222-222222222222"; - const RESULT: MacCommandResult = { exitCode: 0, stdout: "nothing to commit\n", @@ -57,60 +48,6 @@ function client(): HostedMacCommandClient { }); } -/** Lets the client's promise steps run until it has asked, so the deadline is armed before the clock moves. */ -function untilAsked(requests: readonly RecordedRequest[]): Effect.Effect { - return Effect.gen(function* () { - while (requests.length === 0) yield* Effect.yieldNow; - }); -} - -it.effect("a held read names this Mac's kinds and answers the commands claimed for it", () => - Effect.gen(function* () { - const newer = { id: NEWER_ID, kind: "hologram", timeoutSeconds: 60, beam: true }; - const api = fakeCloudApi({ - "GET /api/mac-commands": { answer: () => ({ commands: [COMMAND, newer] }) }, - }); - - const kinds = [MAC_COMMAND_KIND.SHELL, MAC_COMMAND_KIND.BROWSER]; - const waiting = yield* Effect.provide(client().waiting(kinds), api.layer); - - assert.deepEqual(waiting, [COMMAND, { id: NEWER_ID, kind: "hologram" }]); - assert.deepEqual(api.credentials(), ["token-1"]); - assert.equal(api.requests()[0]?.searchParams.get("kinds"), "shell,browser"); - }), -); - -it.effect( - "a shell command from a service before kinds, which names none, is read as a shell command", - () => - Effect.gen(function* () { - const { kind: _kind, ...older } = COMMAND; - const api = fakeCloudApi({ - "GET /api/mac-commands": { answer: () => ({ commands: [older] }) }, - }); - - const waiting = yield* Effect.provide(client().waiting([MAC_COMMAND_KIND.SHELL]), api.layer); - - assert.deepEqual(waiting, [COMMAND]); - }), -); - -it.effect("a held read outlasts the service's hold, then gives up as nothing", () => - Effect.gen(function* () { - const silent = recordingHttpClient(() => new Promise(() => undefined)); - - const read = yield* Effect.forkChild( - Effect.provide(client().waiting([MAC_COMMAND_KIND.SHELL]), silent.layer), - ); - yield* untilAsked(silent.requests); - yield* TestClock.adjust("34 seconds"); - assert.equal(read.pollUnsafe(), undefined, "still held short of its deadline"); - yield* TestClock.adjust("1 second"); - - assert.equal(yield* Fiber.join(read), undefined); - }), -); - it.effect("a result is posted as the service reads it, and is taken or not", () => Effect.gen(function* () { const taken = fakeCloudApi({ "POST /api/mac-commands": { answer: () => ({ settled: true }) } }); @@ -160,20 +97,26 @@ it.effect( }), ); -it.effect("a claim answers the command granted to this Mac, and nothing where it was not", () => - Effect.gen(function* () { - const granted = fakeCloudApi({ - "POST /api/mac-commands/claim": { answer: () => ({ command: COMMAND }) }, - }); - assert.deepEqual(yield* Effect.provide(client().claim(COMMAND_ID), granted.layer), COMMAND); - assert.deepEqual(JSON.parse(granted.requests()[0]?.body ?? ""), { id: COMMAND_ID }); +it.effect( + "a claim answers the command granted to this Mac, and not claimed where the service refused it", + () => + Effect.gen(function* () { + const granted = fakeCloudApi({ + "POST /api/mac-commands/claim": { answer: () => ({ command: COMMAND }) }, + }); + assert.deepEqual(yield* Effect.provide(client().claim(COMMAND_ID), granted.layer), COMMAND); + assert.deepEqual(JSON.parse(granted.requests()[0]?.body ?? ""), { id: COMMAND_ID }); - const taken = fakeCloudApi({ - "POST /api/mac-commands/claim": { - answer: () => ({ error: HOSTED_API_ERROR.NOT_FOUND }), - status: HTTP_STATUS.NOT_FOUND, - }, - }); - assert.equal(yield* Effect.provide(client().claim(COMMAND_ID), taken.layer), undefined); - }), + const taken = fakeCloudApi({ + "POST /api/mac-commands/claim": { + answer: () => ({ error: HOSTED_API_ERROR.NOT_FOUND }), + status: HTTP_STATUS.NOT_FOUND, + }, + }); + assert.deepEqual( + yield* Effect.provide(client().claim(COMMAND_ID), taken.layer), + MAC_COMMAND_NOT_CLAIMED, + "a refused claim is told apart from one never answered", + ); + }), ); diff --git a/packages/features/src/mac-command-client.ts b/packages/features/src/mac-command-client.ts index 493842fc89..07b8938bd6 100644 --- a/packages/features/src/mac-command-client.ts +++ b/packages/features/src/mac-command-client.ts @@ -1,41 +1,36 @@ import { Effect } from "effect"; import type * as HttpClient from "effect/http/HttpClient"; import type { CallCredential } from "./account-call.js"; -import { - askHosted, - askLive, - HOSTED_DEADLINE, - type HostedClient, - hostedClient, -} from "./hosted-link.js"; +import { askHosted, askLive, type HostedClient, hostedClient } from "./hosted-link.js"; import { LIVE_SIGNED_OUT, LIVE_USER_QUERY, type MacCommandsLiveAnswer } from "./live-wire.js"; import { type MacBrowserResult, type MacCommand, type MacCommandKind, - type MacCommandRead, type MacCommandSettleResult, macCommandKindsQuery, } from "./mac-command-wire.js"; import { HOSTED_API_ERROR } from "./service-wire.js"; /** - * mac-command-client.ts -- the Mac's side of the commands the account's agents queue for it: the held read that takes them, and each result posted back. + * mac-command-client.ts -- the Mac's side of the commands the account's agents queue for it: the unclaimed ones of its kinds followed live, each claimed by its id, that the Mac listens, and each result posted back. * - * Both are procedures of the hosted contract (`contract/mac-commands.ts`) - * over the one account call, so the bearer is read fresh per attempt and a - * 401 is renewed and retried once. The read is held by the service while - * nothing waits, so it is asked under the held deadline, which outlasts the - * hold. Neither call fails: a read the service did not answer is nothing, - * which the caller rests after, and a result it did not take is false, - * since its call has already given up on it. A browser command's result, - * which may carry an image, is posted to its own route. A Mac that follows - * its commands live asks for the unclaimed ones of its kinds sealed (or, - * polling, inline), claims each it hears of by its id, and says it listens; - * a seal the service did not answer is a failure the caller asks again, a - * claim not granted is nothing, and a presence not taken is false. + * Each is a procedure of the hosted contract (`contract/live.ts`, + * `contract/mac-commands.ts`) over the one account call, so the bearer is + * read fresh per attempt and a 401 is renewed and retried once. The Mac + * asks for the unclaimed commands of its kinds sealed (or, polling, inline + * and held until they move) for the account it started for, claims each it + * hears of by its id, and says it listens. No call fails: a read refused as + * signed out is `LIVE_SIGNED_OUT` and any other unanswered a failure the + * caller asks again, a claim refused is `MAC_COMMAND_NOT_CLAIMED` and one + * unanswered nothing, and a presence or a result not taken is false, its + * call having given up on it. A browser + * command's result, which may carry an image, is posted to its own route. */ +/** What a claim the service refused answers: another Mac took the command, or it waited past the claim window. */ +export const MAC_COMMAND_NOT_CLAIMED = { claimed: false } as const; + export interface HostedMacCommandClientOptions { /** What authorizes each call on the signed-in account, read fresh for every attempt. */ credential: CallCredential; @@ -53,20 +48,6 @@ export class HostedMacCommandClient { }); } - /** - * The commands of the kinds named claimed for this Mac, once one waits or - * the service's hold ends; nothing where the read was not answered. - */ - waiting( - kinds: readonly MacCommandKind[], - ): Effect.Effect { - const input = { kinds: macCommandKindsQuery(kinds) }; - return askHosted(this.#hosted.macCommands.waiting, input, HOSTED_DEADLINE.HELD).pipe( - Effect.map((answer): readonly MacCommandRead[] | undefined => answer.commands), - Effect.orElseSucceed(() => undefined), - ); - } - /** One command's result posted back; whether the service took it. */ settle( id: string, @@ -122,11 +103,24 @@ export class HostedMacCommandClient { ); } - /** One command heard of live, claimed for this Mac; nothing where another took it, it waited too long, or the claim was not answered. */ - claim(id: string): Effect.Effect { + /** + * One command heard of live, claimed for this Mac; `MAC_COMMAND_NOT_CLAIMED` + * where the service refused it (another took it, or it waited too long), + * and nothing where the claim was not answered, which the caller may ask + * again. + */ + claim( + id: string, + ): Effect.Effect< + MacCommand | typeof MAC_COMMAND_NOT_CLAIMED | undefined, + never, + HttpClient.HttpClient + > { return askHosted(this.#hosted.macCommands.claim, { id }).pipe( - Effect.map((answer): MacCommand | undefined => answer.command), - Effect.orElseSucceed(() => undefined), + Effect.map((answer): MacCommand => answer.command), + Effect.catch((failed) => + Effect.succeed(failed._tag === "HostedRefused" ? MAC_COMMAND_NOT_CLAIMED : undefined), + ), ); } diff --git a/packages/host/src/compose-host.ts b/packages/host/src/compose-host.ts index d42e686269..0ef670fca3 100644 --- a/packages/host/src/compose-host.ts +++ b/packages/host/src/compose-host.ts @@ -227,6 +227,7 @@ export function hostAssemblyLayer( ? yield* composeMacCommands({ kernel, account, + accountId, enabled: Effect.map( Effect.orDie(settings.store.snapshot()), (snapshot) => snapshot.stored.macCommands ?? APP_SETTING_SCHEMA.macCommands.default, diff --git a/packages/host/src/compose-mac-commands.test.ts b/packages/host/src/compose-mac-commands.test.ts index 00328a5dc1..635895bbba 100644 --- a/packages/host/src/compose-mac-commands.test.ts +++ b/packages/host/src/compose-mac-commands.test.ts @@ -1,14 +1,19 @@ import assert from "node:assert/strict"; import { it } from "@effect/vitest"; +import { MAC_COMMAND_NOT_CLAIMED } from "@luke/features"; +import { + LIVE_POLL_URL, + LIVE_SIGNED_OUT, + type MacCommandsLiveAnswer, +} from "@luke/features/live-wire"; import { MAC_COMMAND_KIND, + type MacCommand, type MacCommandKind, - type MacCommandRead, type MacCommandResult, - macCommandsAnswerSchema, } from "@luke/features/mac-command-wire"; import { noNetwork } from "@luke/wire/testing"; -import { ConfigProvider, Effect, Layer, Ref, Schema } from "effect"; +import { ConfigProvider, Duration, Effect, Layer } from "effect"; import { TestClock } from "effect/testing"; import { composeMacCommands, type MacCommandClient } from "./compose-mac-commands.js"; import { Environment, reporterLayer } from "./effect/seams.js"; @@ -16,61 +21,140 @@ import { scriptedSpawner } from "./testing/scripted-spawner.js"; /** * The concern that runs on this Mac what the account's agents queue for it, - * against a scripted service and spawner: with the setting off, or no - * account, the service is never read; each read names the kinds this Mac - * runs, and each command it takes is run or refused by its kind. + * against a scripted service and spawner, with Realtime off, as the + * deployment runs without `LUKE_REALTIME`: the live read answered polled and + * held until the rows move. With the setting off, or no account, the + * service is never read; otherwise every read names the account and the + * kinds this Mac runs, each command waiting is claimed by its id, the + * earliest first, and run or refused by its kind, one queued later is heard + * within the poll's hold, and the Mac says it listens every fifteen + * seconds until the setting is switched off. + * + * Synthetic commands and accounts throughout. */ -/** Commands as this Mac reads them off the wire: a shell command, a browser command, and one of a kind a newer service queued. */ -const [COMMAND, BROWSER, NEWER] = Schema.decodeUnknownSync(macCommandsAnswerSchema)({ - commands: [ - { - id: "2f6a1c7e-9b3d-4e8f-a5c1-7d2e9b4f6a03", - kind: MAC_COMMAND_KIND.SHELL, - command: "node --version", - timeoutSeconds: 60, - }, - { - id: "7c1d2e3f-4a5b-4c6d-8e7f-9a0b1c2d3e4f", - kind: MAC_COMMAND_KIND.BROWSER, - params: { act: "open", address: "http://localhost:3000" }, - timeoutSeconds: 30, - }, - { id: "0e9d8c7b-6a5f-4e3d-9c2b-1a0f9e8d7c6b", kind: "hologram", timeoutSeconds: 30 }, - ], -}).commands as readonly [MacCommandRead, MacCommandRead, MacCommandRead]; +/** The account signed in, which every live read names. */ +const ACCOUNT = "user-mac"; + +const NOW = 1_800_000_000_000; + +const COMMAND: MacCommand = { + id: "2f6a1c7e-9b3d-4e8f-a5c1-7d2e9b4f6a03", + kind: MAC_COMMAND_KIND.SHELL, + command: "node --version", + timeoutSeconds: 60, +}; + +const BROWSER: MacCommand = { + id: "7c1d2e3f-4a5b-4c6d-8e7f-9a0b1c2d3e4f", + kind: MAC_COMMAND_KIND.BROWSER, + params: { act: "open", address: "http://localhost:3000" }, + timeoutSeconds: 30, +}; + +const LATER: MacCommand = { + id: "0e9d8c7b-6a5f-4e3d-9c2b-1a0f9e8d7c6b", + kind: MAC_COMMAND_KIND.SHELL, + command: "git status", + timeoutSeconds: 30, +}; + +/** How often the scripted service looks again while it holds a poll, as the service does. */ +const HOLD_LOOK = Duration.seconds(1); + +/** Let every fiber the step woke run to where it waits. */ +const settled = Effect.repeat(Effect.yieldNow, { times: 200 }); /** - * A service that hands the first read its commands and answers every read - * after it with none once its hold of twenty seconds ends, keeping the - * kinds each read named and every result posted to it. + * The service with Realtime off, as a script: the commands still waiting, + * each with when it was queued, answered polled, and a poll past where they + * stood held until they move; a claim granted once per command; and every + * read, presence, and result the Mac sent, as the service would observe it. */ -function scriptedService(commands: readonly MacCommandRead[]) { - return Effect.gen(function* () { - const reads = yield* Ref.make([]); - const settled = yield* Ref.make>(new Map()); - const client: MacCommandClient = { - waiting: (kinds) => - Effect.flatMap( - Ref.getAndUpdate(reads, (named) => [...named, kinds]), - (named) => - named.length === 0 - ? Effect.succeed(commands) - : Effect.as(Effect.sleep("20 seconds"), []), - ), - settle: (id, result) => - Effect.as( - Ref.update(settled, (held) => new Map([...held, [id, result]])), - true, - ), - settleBrowser: (id, result) => - Effect.as( - Ref.update(settled, (held) => new Map([...held, [id, result]])), - true, - ), - }; - return { client, reads, settled }; +function scriptedService( + options: { + readonly signedOut?: boolean; + readonly unansweredClaims?: number; + /** What happens as a claim is made, before the service grants it, such as the setting being switched off. */ + readonly duringClaim?: Effect.Effect; + } = {}, +) { + let unanswered = options.unansweredClaims ?? 0; + const waiting: (MacCommand & { readonly queuedAt: number })[] = []; + let version = 0; + const reads: { readonly userId: string; readonly kinds: readonly MacCommandKind[] }[] = []; + const claims: string[] = []; + const beats: (readonly MacCommandKind[])[] = []; + const settledResults = new Map(); + const answer = (rows: boolean): MacCommandsLiveAnswer => ({ + url: LIVE_POLL_URL, + queries: { + commands: { + capability: "", + queryFingerprint: "", + expiresAt: NOW + 3_600_000, + digest: String(version), + ...(rows ? { rows: [...waiting] } : undefined), + }, + }, }); + const client: MacCommandClient = { + live: (kinds, userId, _poll, after) => + Effect.gen(function* () { + reads.push({ userId, kinds }); + if (options.signedOut === true) return LIVE_SIGNED_OUT; + for (let held = 0; after === String(version) && held < 25; held += 1) { + yield* Effect.sleep(HOLD_LOOK); + } + return answer(after !== String(version)); + }), + claim: (id) => + Effect.andThen( + options.duringClaim ?? Effect.void, + Effect.sync((): MacCommand | typeof MAC_COMMAND_NOT_CLAIMED | undefined => { + claims.push(id); + // A claim lost on its way, as a network fault loses one, reaches nothing. + if (unanswered > 0) { + unanswered -= 1; + return undefined; + } + const at = waiting.findIndex((command) => command.id === id); + if (at < 0) return MAC_COMMAND_NOT_CLAIMED; + const [claimed] = waiting.splice(at, 1); + version += 1; + if (claimed === undefined) return MAC_COMMAND_NOT_CLAIMED; + const { queuedAt: _queuedAt, ...command } = claimed; + return command; + }), + ), + listening: (kinds) => + Effect.sync(() => { + beats.push(kinds); + return true; + }), + settle: (id, result) => + Effect.sync(() => { + settledResults.set(id, result); + return true; + }), + settleBrowser: (id, result) => + Effect.sync(() => { + settledResults.set(id, result); + return true; + }), + }; + return { + client, + reads, + claims, + beats, + settledResults, + /** A command queued for the account, as an agent's call queues it. */ + queue: (command: MacCommand, queuedAt: number) => { + waiting.push({ ...command, queuedAt }); + version += 1; + }, + }; } const machine = (spawner: ReturnType) => @@ -84,6 +168,20 @@ const machine = (spawner: ReturnType) => noNetwork, ); +/** The concern composed over the scripted service, the gate as `gate` holds it. */ +const subject = ( + service: ReturnType, + spawner: ReturnType, + gate: { enabled: boolean; signedIn: boolean; account?: string }, +) => + composeMacCommands({ + kernel: { runMode: { sendsNetwork: true }, now: () => NOW }, + account: { capabilitiesActive: () => gate.signedIn }, + accountId: () => gate.account ?? ACCOUNT, + enabled: Effect.sync(() => gate.enabled), + client: service.client, + }).pipe(Effect.provide(machine(spawner))); + it.effect("with the setting off, or no account signed in, the service is never read", () => Effect.gen(function* () { for (const gate of [ @@ -91,55 +189,191 @@ it.effect("with the setting off, or no account signed in, the service is never r { enabled: true, signedIn: false }, ]) { const spawner = scriptedSpawner(() => ({ exitCode: 0 })); - const service = yield* scriptedService([COMMAND]); - const composer = yield* composeMacCommands({ - kernel: { runMode: { sendsNetwork: true } }, - account: { capabilitiesActive: () => gate.signedIn }, - enabled: Effect.succeed(gate.enabled), - client: service.client, - }).pipe(Effect.provide(machine(spawner))); + const service = scriptedService(); + service.queue(COMMAND, NOW); + const composer = yield* subject(service, spawner, gate); yield* Effect.scoped(Effect.andThen(composer.lifetime, TestClock.adjust("1 minute"))); - assert.deepEqual(yield* Ref.get(service.reads), [], JSON.stringify(gate)); + assert.deepEqual(service.reads, [], JSON.stringify(gate)); + assert.deepEqual(service.beats, [], JSON.stringify(gate)); assert.deepEqual(spawner.spawned, []); } }), ); it.effect( - "each read names shell alone, and a command of a kind this Mac cannot run is refused while the shell one runs", + "over the held poll, each waiting command is claimed by its id, the earliest first, and run or refused by its kind, one queued later is heard within a second or so, and every read names the account and shell alone", () => Effect.gen(function* () { const spawner = scriptedSpawner(() => ({ stdout: "v24.0.0\n", exitCode: 0 })); - const service = yield* scriptedService([BROWSER, COMMAND, NEWER]); - const composer = yield* composeMacCommands({ - kernel: { runMode: { sendsNetwork: true } }, - account: { capabilitiesActive: () => true }, - enabled: Effect.succeed(true), - client: service.client, - }).pipe(Effect.provide(machine(spawner))); - - yield* Effect.scoped(Effect.andThen(composer.lifetime, TestClock.adjust("30 seconds"))); - - const reads = yield* Ref.get(service.reads); - assert.ok(reads.length >= 2, "the reads go on after the refusals"); - for (const kinds of reads) assert.deepEqual(kinds, [MAC_COMMAND_KIND.SHELL]); - const settled = yield* Ref.get(service.settled); - assert.deepEqual(settled.get(COMMAND.id), { + const service = scriptedService(); + service.queue(BROWSER, NOW - 1_000); + service.queue(COMMAND, NOW - 2_000); + const composer = yield* subject(service, spawner, { enabled: true, signedIn: true }); + + yield* Effect.scoped( + Effect.gen(function* () { + yield* composer.lifetime; + yield* settled; + assert.deepEqual(service.claims, [COMMAND.id, BROWSER.id]); + + service.queue(LATER, NOW); + yield* TestClock.adjust("2 seconds"); + yield* settled; + assert.deepEqual(service.claims, [COMMAND.id, BROWSER.id, LATER.id]); + }), + ); + + assert.ok(service.reads.length > 0); + for (const read of service.reads) { + assert.deepEqual(read, { userId: ACCOUNT, kinds: [MAC_COMMAND_KIND.SHELL] }); + } + assert.deepEqual(service.settledResults.get(COMMAND.id), { exitCode: 0, stdout: "v24.0.0\n", stderr: "", truncated: false, timedOut: false, }); - for (const refused of [BROWSER, NEWER]) { - const result = settled.get(refused.id); - assert.ok(result !== undefined && "refused" in result, refused.kind); - } - assert.deepEqual( - spawner.spawned.map((process) => process.args), - [["-l", "-c", "node --version"]], + const refused = service.settledResults.get(BROWSER.id); + assert.ok(refused !== undefined && "refused" in refused, "a browser command is refused"); + assert.ok(service.settledResults.has(LATER.id), "the later command ran"); + }), +); + +it.effect( + "the Mac says it listens for shell at once and every fifteen seconds, well inside the service's forty-five, and once the setting is switched off it stops", + () => + Effect.gen(function* () { + const spawner = scriptedSpawner(() => ({ exitCode: 0 })); + const service = scriptedService(); + const gate = { enabled: true, signedIn: true }; + const composer = yield* subject(service, spawner, gate); + + yield* Effect.scoped( + Effect.gen(function* () { + yield* composer.lifetime; + yield* settled; + assert.equal(service.beats.length, 1, "said at once"); + yield* TestClock.adjust("45 seconds"); + yield* settled; + assert.equal(service.beats.length, 4); + for (const kinds of service.beats) assert.deepEqual(kinds, [MAC_COMMAND_KIND.SHELL]); + + gate.enabled = false; + yield* TestClock.adjust("15 seconds"); + yield* settled; + const reads = service.reads.length; + yield* TestClock.adjust("1 minute"); + yield* settled; + assert.equal(service.beats.length, 4, "a Mac switched off says nothing"); + assert.equal(service.reads.length, reads, "nor reads"); + }), ); }), ); + +it.effect("a read the service signs out follows nothing, and is asked again after a pause", () => + Effect.gen(function* () { + const spawner = scriptedSpawner(() => ({ exitCode: 0 })); + const service = scriptedService({ signedOut: true }); + service.queue(COMMAND, NOW); + const composer = yield* subject(service, spawner, { enabled: true, signedIn: true }); + + yield* Effect.scoped( + Effect.gen(function* () { + yield* composer.lifetime; + yield* settled; + assert.equal(service.reads.length, 1); + yield* TestClock.adjust("5 seconds"); + yield* settled; + assert.equal(service.reads.length, 2); + }), + ); + + assert.deepEqual(service.claims, []); + assert.deepEqual(service.beats, []); + }), +); + +it.effect("a claim the service never heard is asked again, and the command runs", () => + Effect.gen(function* () { + const spawner = scriptedSpawner(() => ({ exitCode: 0 })); + const service = scriptedService({ unansweredClaims: 2 }); + service.queue(COMMAND, NOW); + const composer = yield* subject(service, spawner, { enabled: true, signedIn: true }); + + yield* Effect.scoped( + Effect.gen(function* () { + yield* composer.lifetime; + yield* settled; + yield* TestClock.adjust("3 seconds"); + yield* settled; + }), + ); + + assert.deepEqual(service.claims, [COMMAND.id, COMMAND.id, COMMAND.id]); + assert.ok(service.settledResults.has(COMMAND.id), "it ran once the claim got through"); + }), +); + +it.effect( + "a command granted as the setting is switched off is answered as not run, though the follow ends mid-claim", + () => + Effect.gen(function* () { + const spawner = scriptedSpawner(() => ({ exitCode: 0 })); + const gate = { enabled: true, signedIn: true }; + const service = scriptedService({ + // The setting goes off as the claim leaves, and the service is slow to grant it, so a beat sees the gate shut first. + duringClaim: Effect.andThen( + Effect.sync(() => { + gate.enabled = false; + }), + Effect.sleep("20 seconds"), + ), + }); + service.queue(COMMAND, NOW); + const composer = yield* subject(service, spawner, gate); + + yield* Effect.scoped( + Effect.gen(function* () { + yield* composer.lifetime; + yield* settled; + yield* TestClock.adjust("30 seconds"); + yield* settled; + }), + ); + + const answered = service.settledResults.get(COMMAND.id); + assert.ok(answered !== undefined && "stderr" in answered, "the call hears it was not run"); + assert.match(answered.stderr, /switched off/u); + assert.deepEqual(spawner.spawned, []); + }), +); + +it.effect( + "an account switched between beats ends the follow, and the next names the new account", + () => + Effect.gen(function* () { + const spawner = scriptedSpawner(() => ({ exitCode: 0 })); + const service = scriptedService(); + const gate = { enabled: true, signedIn: true, account: ACCOUNT }; + const composer = yield* subject(service, spawner, gate); + + yield* Effect.scoped( + Effect.gen(function* () { + yield* composer.lifetime; + yield* settled; + gate.account = "user-other"; + yield* TestClock.adjust("15 seconds"); + yield* settled; + yield* TestClock.adjust("5 seconds"); + yield* settled; + }), + ); + + assert.equal(service.reads[0]?.userId, ACCOUNT); + assert.ok(service.reads.some((read) => read.userId === "user-other")); + }), +); diff --git a/packages/host/src/compose-mac-commands.ts b/packages/host/src/compose-mac-commands.ts index d68fe2b1cc..d66ec9da20 100644 --- a/packages/host/src/compose-mac-commands.ts +++ b/packages/host/src/compose-mac-commands.ts @@ -1,4 +1,9 @@ import type { HostedMacCommandClient } from "@luke/features"; +import { + type MacCommandsLiveAnswer, + macCommandsLiveRowsSchema, + macCommandsOfLiveRows, +} from "@luke/features/live-wire"; import { isMacShellCommand, MAC_COMMAND_KIND, @@ -8,13 +13,15 @@ import { type MacCommandResult, type MacShellResult, } from "@luke/features/mac-command-wire"; -import { Config, Duration, Effect, FiberSet, Option } from "effect"; +import { Config, Duration, Effect, FiberSet, Option, Schedule, Schema, Stream } from "effect"; import * as HttpClient from "effect/http/HttpClient"; import type { ChildProcessSpawner } from "effect/process"; import type { AccountComposer } from "./compose-account.js"; import type { Composer } from "./composer.js"; import { Environment, Reporter } from "./effect/seams.js"; import type { HOST_PLATFORM_CONCERN, PlatformConcernMethods } from "./host-concerns.js"; +import type { HostKernel } from "./host-kernel.js"; +import { LiveBroken, type LiveClient, liveRows, openRealtimeClient } from "./live-follow.js"; import { type MacCommandSettleClient, settleMacCommandResult } from "./mac-browser-result.js"; import { type MacShell, runMacCommand } from "./mac-command-runner.js"; import type { RunMode } from "./run-mode.js"; @@ -23,26 +30,43 @@ import type { RunMode } from "./run-mode.js"; * compose-mac-commands.ts -- runs on this Mac the commands the account's agents queue for it, for as long as an account is signed in and Settings lets agents run commands here. * * Nothing asks the developer first: Dean's call, with the setting the one - * brake, on by default. While the gate stands, one held read after another - * takes every command waiting for the account of the kinds this Mac runs, - * which each read names (`HostedMacCommandClient`); each runs on a fiber of - * its own in the concern's scope, by its kind: a shell command in the - * login shell from the home folder (`mac-command-runner.ts`), and a - * command of a kind this Mac cannot run answered with a refusal rather - * than spoiling the read. Its result is posted back by the route its kind + * brake, on by default. While the gate stands, the commands no Mac has + * claimed, of the kinds this Mac runs, are followed through the live layer + * (`live-follow.ts`: over Neon Realtime where the deployment turns it on, + * otherwise a poll the service holds until the rows move), and each one + * heard of is claimed by its id; the service grants a claim once, so when + * two Macs of one account follow, only the one it granted runs it. Each + * claimed command runs on a fiber of its own in the concern's scope, by its + * kind: a shell command in the login shell from the home folder + * (`mac-command-runner.ts`), and a command of a kind this Mac cannot run + * answered with a refusal. Its result is posted back by the route its kind * takes (`mac-browser-result.ts`), so a long command holds up neither the - * next read nor the others. - * A Mac holding the read is what the service counts as listening, so the - * setting switched off, a sign-out, or a run that sends nothing stops the - * reads and the agents' calls are told no Mac listens, and stops every - * command still running, its process group killed. When two Macs of one - * account hold the read, whichever read takes a command runs it, and only - * that one: the service claims it once. Quitting closes the scope, which - * kills every command still running with its process group. + * follow nor the others. + * + * A live read writes nothing, so the Mac says it listens for its kinds + * itself, every `LISTENING_EVERY`, well inside the service's freshness + * window: the one timer that remains, because presence is a fact about this + * process that no row change can carry. Each beat reads the gate first, so + * the setting switched off, a sign-out, or a run that sends nothing ends the + * follow and the beats, the agents' calls are told no Mac listens, and every + * command still running is stopped, its process group killed. Every read + * names the account the follow started for, so one the service signs out + * ends the follow. Quitting closes the scope, which kills every command + * still running with its process group. */ -/** How long the reads rest while the gate is shut, or after one that went unanswered or died. */ +/** How long the follow rests while the gate is shut, or after a turn that died. */ const IDLE = Duration.seconds(5); +/** How often a Mac following its commands says it listens: a third of the service's forty-five-second freshness window. */ +const LISTENING_EVERY = Duration.seconds(15); +/** How long after a read went unanswered, or a follow stopped for good, it is asked again. */ +const LIVE_RESTART = Duration.seconds(5); +/** + * How a claim the service did not answer is asked again: a second apart, a + * few times, well inside the service's thirty-second claim window, since + * the rows that named it may not change again to say so. + */ +const CLAIM_RETRY = { every: Duration.seconds(1), times: 5 } as const; /** The login shell when the environment names none: macOS's own default. */ const DEFAULT_SHELL = "/bin/zsh"; @@ -53,14 +77,19 @@ const VARIABLE = { HOME: "HOME", SHELL: "SHELL" } as const; const RUNS: readonly MacCommandKind[] = [MAC_COMMAND_KIND.SHELL]; /** The service's side of the commands, as this concern asks it. */ -export type MacCommandClient = Pick & MacCommandSettleClient; +export type MacCommandClient = Pick & + MacCommandSettleClient; export interface MacCommandsDependencies { - kernel: { runMode: Pick }; + kernel: Pick & { runMode: Pick }; account: Pick; - /** Whether Settings lets agents run commands on this Mac, read as each read begins. */ + /** The account signed in now, by its id, which every live read names; none where nobody is. */ + accountId: () => string | undefined; + /** Whether Settings lets agents run commands on this Mac, read as each follow begins and on every beat. */ enabled: Effect.Effect; client: MacCommandClient; + /** Realtime's client at an address, where the deployment follows over Realtime; Realtime's own outside a test. */ + openLive?: (url: string) => LiveClient | PromiseLike; } export type MacCommandsComposer = Composer; @@ -113,6 +142,8 @@ export function composeMacCommands( const services = yield* Effect.context(); const http = yield* HttpClient.HttpClient; const { kernel, account, enabled, client } = dependencies; + const openLive = dependencies.openLive ?? openRealtimeClient; + const decode = Schema.decodeUnknownOption(macCommandsLiveRowsSchema); const asked = (call: Effect.Effect) => Effect.provideService(call, HttpClient.HttpClient, http); @@ -139,27 +170,107 @@ export function composeMacCommands( ); /** - * One turn of the loop: a held read and a fiber per command it took, or, - * with the gate shut, every command still running stopped (its process - * group killed as its scope closes) and a rest. Note that the gate is read - * again once the read answers, because it may have shut while the read was - * held, and a command taken then is answered as not run rather than run. + * One command heard of, claimed by its id and, once granted, run on a + * fiber of its own; with the gate shut since, answered as not run + * instead. A claim the service did not answer is asked again a few times + * (`CLAIM_RETRY`), since the rows that named it may not change again to + * say so. Note that the claim and its answer are uninterruptible and run + * on the concern's own set rather than the follow's fiber, because a + * command the service granted is this Mac's alone to answer, and a follow + * ended mid-claim would otherwise leave its call waiting out its time. */ - const turn = (running: FiberSet.FiberSet) => + const claimed = (running: FiberSet.FiberSet, command: MacCommandRead) => Effect.gen(function* () { + const answer = yield* asked(client.claim(command.id)).pipe( + Effect.repeat({ + schedule: Schedule.spaced(CLAIM_RETRY.every), + times: CLAIM_RETRY.times, + until: (granted) => granted !== undefined, + }), + ); + // A claim refused, or never answered, runs nothing: the command is another Mac's, or its call has given up. + if (answer === undefined || !("id" in answer)) return; if (!(yield* gateOpen)) { - yield* FiberSet.clear(running); - return yield* Effect.sleep(IDLE); + yield* asked(client.settle(answer.id, SWITCHED_OFF)); + return; + } + yield* FiberSet.run(running, run(answer)); + }).pipe(Effect.uninterruptible); + + /** + * That this Mac listens for its kinds, said every `LISTENING_EVERY` for + * as long as the gate stands and the account signed in is the one the + * follow is for; it returns once either changes, which ends the follow. + */ + const presence = (userId: string) => + Effect.gen(function* () { + while ((yield* gateOpen) && dependencies.accountId() === userId) { + yield* asked(client.listening(RUNS)); + yield* Effect.sleep(LISTENING_EVERY); } - const commands = yield* asked(client.waiting(RUNS)); - // Note that an unanswered read rests before the next, so a service that is down is not asked in a tight loop. - if (commands === undefined) return yield* Effect.sleep(IDLE); + }); + + /** The unclaimed commands of this Mac's kinds, sealed or with `poll` inline, for the account `userId` names. */ + const seal = (userId: string, poll: boolean, after?: string) => + asked(client.live(RUNS, userId, poll, after)); + + /** + * The unclaimed commands followed (`live-follow.ts`, which polls the + * service, held, where Realtime is off or fails) until the gate shuts or + * the session signs out, each new one claimed by its id, the earliest + * queued first (`claimed`). A command is claimed once per follow, and + * forgotten once it leaves the rows, so the ids this keeps are only those + * still waiting. A follow ends where the gate shuts, the session signs + * out, or the account signed in changes. + */ + const followed = (running: FiberSet.FiberSet, userId: string, first: MacCommandsLiveAnswer) => + Effect.suspend(() => { + let tried = new Set(); + const claimNew = (commands: readonly MacCommandRead[]) => + Effect.gen(function* () { + const fresh = commands.filter((command) => !tried.has(command.id)); + tried = new Set(commands.map((command) => command.id)); + for (const command of fresh) yield* FiberSet.run(running, claimed(running, command)); + }); + const rows = liveRows( + { + openLive, + reseal: () => seal(userId, false), + poll: (after) => seal(userId, true, after), + now: kernel.now, + }, + first, + ); + const following = Stream.runForEach(rows, (delivered) => + Option.match(decode(delivered), { + onNone: () => Effect.fail(new LiveBroken({ reason: "a row did not decode" })), + onSome: (read) => claimNew(macCommandsOfLiveRows(read)), + }), + ); + return Effect.raceFirst(following, presence(userId)).pipe( + Effect.catch(() => Effect.sleep(LIVE_RESTART)), + ); + }); + + /** + * One turn of the loop: the commands followed until the gate shuts, the + * read asked again after a pause where the service did not answer it or + * signed the session out; with the gate shut, every command still + * running stopped (its process group killed as its scope closes) and a + * rest. + */ + const turn = (running: FiberSet.FiberSet) => + Effect.gen(function* () { if (!(yield* gateOpen)) { yield* FiberSet.clear(running); - for (const command of commands) yield* asked(client.settle(command.id, SWITCHED_OFF)); - return; + return yield* Effect.sleep(IDLE); } - for (const command of commands) yield* FiberSet.run(running, run(command)); + // The account the follow starts for, which every read names, so one under another's session is refused as signed out. + const userId = dependencies.accountId(); + if (userId === undefined) return yield* Effect.sleep(IDLE); + const sealed = yield* seal(userId, false); + if ("failure" in sealed) return yield* Effect.sleep(LIVE_RESTART); + yield* followed(running, userId, sealed); }); const lifetime = Effect.gen(function* () {