diff --git a/apps/cli/src/legacy/commands/encryption/update-root-key/update-root-key.command.ts b/apps/cli/src/legacy/commands/encryption/update-root-key/update-root-key.command.ts index 8a0896cea7..f34ccf9b4a 100644 --- a/apps/cli/src/legacy/commands/encryption/update-root-key/update-root-key.command.ts +++ b/apps/cli/src/legacy/commands/encryption/update-root-key/update-root-key.command.ts @@ -1,4 +1,3 @@ -import { BunServices } from "@effect/platform-bun"; import { Layer } from "effect"; import { Command, Flag } from "effect/unstable/cli"; import type * as CliCommand from "effect/unstable/cli/Command"; @@ -20,11 +19,11 @@ const config = { export type LegacyEncryptionUpdateRootKeyFlags = CliCommand.Command.Config.Infer; // `Stdin` is new production wiring for this command. Provide it explicitly -// (along with its `Tty` + `Stdio` deps) so the command's layer is self-contained -// and does not rely on sibling-layer leakage inside `Layer.mergeAll`. +// (along with its `Tty` dep) so the command's layer is self-contained and does +// not rely on sibling-layer leakage inside `Layer.mergeAll`. const updateRuntime = Layer.mergeAll( legacyManagementApiRuntimeLayer(["encryption", "update-root-key"]), - stdinLayer.pipe(Layer.provide(ttyLayer), Layer.provide(BunServices.layer)), + stdinLayer.pipe(Layer.provide(ttyLayer)), ); export const legacyEncryptionUpdateRootKeyCommand = Command.make("update-root-key", config).pipe( diff --git a/apps/cli/src/legacy/commands/logout/logout.layers.ts b/apps/cli/src/legacy/commands/logout/logout.layers.ts index 775cd58b05..e07e1ee126 100644 --- a/apps/cli/src/legacy/commands/logout/logout.layers.ts +++ b/apps/cli/src/legacy/commands/logout/logout.layers.ts @@ -18,7 +18,7 @@ import { stdinLayer } from "../../../shared/runtime/stdin.layer.ts"; * legacy CLAUDE.md item 5). `Analytics`, `Output`, `Stdio`, `Tty`, `FileSystem`, * `Path`, `TelemetryRuntime`, and `LegacyYesFlag` come from the root layer; * `stdinLayer` (the shared piped-stdin reader for the logout confirm) builds its - * `Stdin` from the root `Stdio`/`Tty`, like the migration runtimes. + * `Stdin` from the root `Tty`, like the migration runtimes. */ const cliSettings = legacyCliSettingsLayer.pipe(Layer.provide(legacyDebugLoggerLayer)); const credentials = legacyCredentialsLayer.pipe( diff --git a/apps/cli/src/shared/runtime/stdin.integration.test.ts b/apps/cli/src/shared/runtime/stdin.integration.test.ts index 89e851e2a4..806eefb013 100644 --- a/apps/cli/src/shared/runtime/stdin.integration.test.ts +++ b/apps/cli/src/shared/runtime/stdin.integration.test.ts @@ -1,30 +1,26 @@ +import { fileURLToPath } from "node:url"; import { describe, expect, it } from "@effect/vitest"; -import { Deferred, Duration, Effect, Fiber, Layer, Option, Queue, Stdio, Stream } from "effect"; +import { Duration, Effect, Fiber, Layer, Option, Queue, Ref, Stream } from "effect"; +import { systemError, type PlatformError } from "effect/PlatformError"; import { TestClock } from "effect/testing"; import { mockTty } from "../../../tests/helpers/mocks.ts"; import { Stdin } from "./stdin.service.ts"; -import { stdinLayer } from "./stdin.layer.ts"; +import { stdinLayerFrom } from "./stdin.layer.ts"; const enc = (s: string) => new TextEncoder().encode(s); -const dec = (bytes: Uint8Array) => new TextDecoder().decode(bytes); // Exercises the real `stdinLayer` (its persistent, lazily-opened line reader) over a -// controllable `Stdio` stream, instead of the array-indexing `mockStdin` double. -// `Stdio.layerTest` lets us drive stdin as a byte Stream with deliberate chunking / -// delays; `stdinLayer` also needs `Tty`, satisfied by `mockTty`. -const withStdin = (stdin: Stream.Stream, stdinIsTty = false) => - stdinLayer.pipe( - Layer.provide(Stdio.layerTest({ stdin })), - Layer.provide(mockTty({ stdinIsTty, stdoutIsTty: false })), - ); +// controllable byte stream, instead of the array-indexing `mockStdin` double, so stdin +// can be driven with deliberate chunking / delays; `Tty` is satisfied by `mockTty`. +const withStdin = (stdin: Stream.Stream, stdinIsTty = false) => + stdinLayerFrom(stdin).pipe(Layer.provide(mockTty({ stdinIsTty, stdoutIsTty: false }))); -describe("stdinLayer readLine", () => { +describe("stdinLayer", () => { it.live("dispenses successive lines across calls, buffering multi-line chunks", () => { - // Two chunks; the second carries two lines. A persistent reader must return a, b, - // c across successive calls (the second call pulls a fresh chunk, the third is - // served from the buffered remainder) — one bufio.Scanner, not a fresh read each - // time. A final call on the exhausted stream yields None (the prompt default). + // Two chunks, the second carrying two lines: one persistent reader returns a, b, c + // across successive calls, holding the rest of a chunk for the next call instead of + // starting over. A final call on the exhausted stream yields None (the prompt default). const layer = withStdin(Stream.fromIterable([enc("a\n"), enc("b\nc\n")])); return Effect.gen(function* () { const stdin = yield* Stdin; @@ -35,6 +31,19 @@ describe("stdinLayer readLine", () => { }).pipe(Effect.provide(layer)); }); + it.live("accepts CRLF and bare CR line endings and a final line without one", () => { + // Answers piped from Windows tooling (`\r\n`) or an old Mac convention (`\r`), plus + // `printf y` with no trailing newline, all read as whole lines before EOF. + const layer = withStdin(Stream.fromIterable([enc("a\r\nb\rc")])); + return Effect.gen(function* () { + const stdin = yield* Stdin; + expect(yield* stdin.readLine(10_000)).toStrictEqual(Option.some("a")); + expect(yield* stdin.readLine(10_000)).toStrictEqual(Option.some("b")); + expect(yield* stdin.readLine(10_000)).toStrictEqual(Option.some("c")); + expect(yield* stdin.readLine(10_000)).toStrictEqual(Option.none()); + }).pipe(Effect.provide(layer)); + }); + it.live("preserves interior blank lines so answers stay aligned", () => { // splitLines keeps blank interior lines: a caller that pipes "\ny\n" sees the // blank line first (→ prompt default) and the y second, not y first. @@ -47,9 +56,8 @@ describe("stdinLayer readLine", () => { }); it.live("times out to None when no line arrives within the window", () => { - // A pipe that stays open without a newline (Go's non-TTY `ReadLine` timeout, - // console.go:36): readLine must give up with None so the prompt takes its default - // instead of blocking on EOF. + // A pipe that stays open without sending a line: readLine must give up with None so + // the prompt takes its default instead of waiting for EOF. const layer = withStdin(Stream.never); return Effect.gen(function* () { const stdin = yield* Stdin; @@ -57,37 +65,198 @@ describe("stdinLayer readLine", () => { }).pipe(Effect.provide(layer)); }); - it.live("keeps a piped stdin consumed once the reader is open", () => + it.live("waits for a non-blocking pipe that has nothing to read yet", () => + Effect.gen(function* () { + // A non-blocking fd 0 with nothing to read yet fails the read with `WouldBlock` (how the + // layer reports `EAGAIN`, pinned over a real fd 0 below). That is "nothing yet", not EOF: + // the reader keeps asking until the answer lands, instead of taking the default for this + // prompt and every one after it. + let attempts = 0; + const layer = withStdin( + Stream.suspend(() => { + attempts += 1; + return attempts < 3 + ? Stream.fail(systemError({ module: "Stdin", method: "read", _tag: "WouldBlock" })) + : Stream.make(enc("y\n")); + }), + ); + yield* Effect.gen(function* () { + const stdin = yield* Stdin; + expect(yield* stdin.readLine(10_000)).toStrictEqual(Option.some("y")); + expect(attempts).toBe(3); + }).pipe(Effect.provide(layer)); + }), + ); + + it.effect("keeps waiting on a non-blocking pipe across a prompt that gave up", () => Effect.gen(function* () { - const pipe = yield* Queue.unbounded(); - const taken = yield* Deferred.make(); + // The first prompt times out between two looks, interrupting the wait. The next prompt + // must take the wait back up and see the answer once it lands, instead of inheriting the + // failed read as the reader's last word. + const ready = yield* Ref.make(false); const layer = withStdin( - Stream.fromQueue(pipe).pipe( - Stream.tap((chunk) => - dec(chunk) === "b\n" ? Deferred.succeed(taken, true) : Effect.succeed(false), + Stream.unwrap( + Ref.get(ready).pipe( + Effect.map((isReady) => + isReady + ? Stream.make(enc("y\n")) + : Stream.fail(systemError({ module: "Stdin", method: "read", _tag: "WouldBlock" })), + ), ), ), ); yield* Effect.gen(function* () { const stdin = yield* Stdin; - yield* Queue.offer(pipe, enc("a\n")); - expect(yield* stdin.readLine(10_000)).toStrictEqual(Option.some("a")); + const gaveUp = yield* Effect.forkChild(stdin.readLine(100)); + yield* TestClock.adjust(Duration.millis(100)); + expect(yield* Fiber.join(gaveUp)).toStrictEqual(Option.none()); + yield* Ref.set(ready, true); + const answered = yield* Effect.forkChild(stdin.readLine(10_000)); + yield* TestClock.adjust(Duration.millis(10)); + expect(yield* Fiber.join(answered)).toStrictEqual(Option.some("y")); + }).pipe(Effect.provide(layer)); + }), + ); + + it.effect("collects a pipe across a non-blocking read that had nothing yet", () => + Effect.gen(function* () { + // The whole-pipe collects wait a non-blocking fd 0 out the same way, and a fresh reader + // over the still-open descriptor carries on where the last one stopped, so what came + // before the empty read and what comes after it read as one stream. + let attempts = 0; + const layer = withStdin( + Stream.suspend(() => { + attempts += 1; + return attempts === 1 + ? Stream.concat( + Stream.make(enc("ab")), + Stream.fail(systemError({ module: "Stdin", method: "read", _tag: "WouldBlock" })), + ) + : Stream.make(enc("cd")); + }), + ); + yield* Effect.gen(function* () { + const stdin = yield* Stdin; + const collected = yield* Effect.forkChild(stdin.readPipedText); + yield* TestClock.adjust(Duration.millis(10)); + expect(yield* Fiber.join(collected)).toStrictEqual(Option.some("abcd")); + expect(attempts).toBe(2); + }).pipe(Effect.provide(layer)); + }), + ); + + it.effect("keeps reading after a prompt times out, finishing the line it was waiting on", () => + Effect.gen(function* () { + // A slow producer: the first prompt times out holding a partial line, and the bytes + // that complete it must still reach the next prompt. + const queue = yield* Queue.unbounded(); + const layer = withStdin(Stream.fromQueue(queue)); + yield* Effect.gen(function* () { + const stdin = yield* Stdin; + yield* Queue.offer(queue, enc("ab")); + const reading = yield* Effect.forkChild(stdin.readLine(100)); + yield* TestClock.adjust(Duration.millis(100)); + expect(yield* Fiber.join(reading)).toStrictEqual(Option.none()); + yield* Queue.offer(queue, enc("c\nd\n")); + expect(yield* stdin.readLine(10_000)).toStrictEqual(Option.some("abc")); + expect(yield* stdin.readLine(10_000)).toStrictEqual(Option.some("d")); + }).pipe(Effect.provide(layer)); + }), + ); - yield* Queue.offer(pipe, enc("b\n")); - yield* Deferred.await(taken); + it.live("lets one prompt at a time pull from the pipe", () => + Effect.gen(function* () { + // Two prompts wait at once. The second must get `2`, held back from the chunk the first + // one pulled, instead of pulling a chunk of its own and skipping it. Each pull yields + // once, so a second pull could slip in while the first is in flight. + let pulls = 0; + const layer = withStdin( + Stream.fromEffectRepeat( + Effect.suspend(() => { + pulls += 1; + return Effect.yieldNow.pipe(Effect.as(enc(`${2 * pulls - 1}\n${2 * pulls}\n`))); + }), + ), + ); + yield* Effect.gen(function* () { + const stdin = yield* Stdin; + const answers = yield* Effect.all([stdin.readLine(10_000), stdin.readLine(10_000)], { + concurrency: "unbounded", + }); + expect(answers.map(Option.getOrThrow).sort()).toStrictEqual(["1", "2"]); + expect(yield* stdin.readLine(10_000)).toStrictEqual(Option.some("3")); + }).pipe(Effect.provide(layer)); + }), + ); - expect(yield* stdin.readLine(10_000)).toStrictEqual(Option.some("b")); + it.live("reads a pipe only while a prompt is waiting", () => + Effect.gen(function* () { + // An endless producer, counted per chunk: nothing is pulled before the first prompt + // or between prompts, so whatever the prompts do not ask for stays in the pipe. + const pulled = yield* Ref.make(0); + const layer = withStdin( + Stream.fromEffectRepeat( + Ref.updateAndGet(pulled, (n) => n + 1).pipe(Effect.map((n) => enc(`line-${n}\n`))), + ), + ); + yield* Effect.gen(function* () { + const stdin = yield* Stdin; + expect(yield* Ref.get(pulled)).toBe(0); + expect(yield* stdin.readLine(10_000)).toStrictEqual(Option.some("line-1")); + expect(yield* Ref.get(pulled)).toBe(1); + expect(yield* stdin.readLine(10_000)).toStrictEqual(Option.some("line-2")); + expect(yield* Ref.get(pulled)).toBe(2); }).pipe(Effect.provide(layer)); }), ); - it.live("answers prompts from the earliest lines when a producer floods the pipe", () => { - const flood = Array.from({ length: 5_000 }, (_, index) => enc(`line-${index}\n`)); + it.live("answers every prompt in order when a producer floods the pipe", () => { + // The 10,000th prompt still gets the 10,000th line, and an unbounded producer is only + // read as far as the prompts ask. The lines total well over 64 KiB, so the pending-line + // bound must reset at each line break. + const flood = Array.from({ length: 10_000 }, (_, index) => enc(`line-${index}\n`)); const layer = withStdin(Stream.fromIterable(flood).pipe(Stream.concat(Stream.never))); return Effect.gen(function* () { const stdin = yield* Stdin; - expect(yield* stdin.readLine(10_000)).toStrictEqual(Option.some("line-0")); - expect(yield* stdin.readLine(10_000)).toStrictEqual(Option.some("line-1")); + for (let index = 0; index < 10_000; index++) { + expect(yield* stdin.readLine(10_000)).toStrictEqual(Option.some(`line-${index}`)); + } + }).pipe(Effect.provide(layer)); + }); + + it.live("gives up on a line that never ends instead of buffering it", () => + Effect.gen(function* () { + // A producer that never sends a newline (`yes | tr -d '\n'`), counted per 16 KiB + // chunk: the reader stops pulling once the pending line outgrows its 64 KiB bound + // (the fifth chunk), and every prompt from then on takes its default. + const pulled = yield* Ref.make(0); + const chunk = enc("y".repeat(16 * 1024)); + const layer = withStdin( + Stream.fromEffectRepeat(Ref.update(pulled, (n) => n + 1).pipe(Effect.as(chunk))), + ); + yield* Effect.gen(function* () { + const stdin = yield* Stdin; + expect(yield* stdin.readLine(10_000)).toStrictEqual(Option.none()); + expect(yield* Ref.get(pulled)).toBe(5); + expect(yield* stdin.readLine(10_000)).toStrictEqual(Option.none()); + expect(yield* Ref.get(pulled)).toBe(5); + }).pipe(Effect.provide(layer)); + }), + ); + + it.live("answers the lines ahead of a runaway tail that shares their chunk", () => { + // `{ printf 'y\n'; cat blob-without-newline; } | …` can land the answer and the start + // of the blob in one pull. The answer is still delivered; the tail then trips the bound, + // so the `n` behind it is never read and every later prompt takes its default. + const layer = withStdin( + Stream.fromArray([enc("y\n"), enc("z".repeat(64 * 1024 + 1))]).pipe( + Stream.concat(Stream.make(enc("n\n"))), + ), + ); + return Effect.gen(function* () { + const stdin = yield* Stdin; + expect(yield* stdin.readLine(10_000)).toStrictEqual(Option.some("y")); + expect(yield* stdin.readLine(10_000)).toStrictEqual(Option.none()); }).pipe(Effect.provide(layer)); }); @@ -112,3 +281,117 @@ describe("stdinLayer readLine", () => { }).pipe(Effect.provide(layer)); }); }); + +describe("stdinLayer over fd 0", () => { + it("waits out a non-blocking fd 0 until the answer lands", async () => { + // A parent that hands fd 0 down in non-blocking mode: perl flips `O_NONBLOCK` on the pipe + // (Bun cannot), confirms the mode on stderr and execs the reader. The first prompt finds + // the pipe empty and must run out its window to None instead of taking the empty read as a + // dead descriptor; the second must read the answer written once that window has closed. + const bun = Bun.which("bun"); + const perl = Bun.which("perl"); + if (!bun || !perl) throw new Error("bun and perl executables not found"); + const here = (file: string) => JSON.stringify(fileURLToPath(new URL(file, import.meta.url))); + const child = Bun.spawn( + [ + perl, + "-e", + `use Fcntl; + fcntl(STDIN, F_SETFL, O_NONBLOCK) or die "fcntl: $!"; + print STDERR ((fcntl(STDIN, F_GETFL, 0) & O_NONBLOCK) ? "nonblock\\n" : "block\\n"); + exec @ARGV or die "exec: $!";`, + bun, + "-e", + `import { Effect, Layer, Option } from "effect"; + import { Stdin } from ${here("./stdin.service.ts")}; + import { stdinLayer } from ${here("./stdin.layer.ts")}; + import { ttyLayer } from ${here("./tty.layer.ts")}; + const program = Effect.gen(function* () { + const stdin = yield* Stdin; + console.log(Option.getOrElse(yield* stdin.readLine(300), () => "")); + console.log(Option.getOrElse(yield* stdin.readLine(5_000), () => "")); + }); + Effect.runPromise(program.pipe(Effect.provide(stdinLayer.pipe(Layer.provide(ttyLayer))))).then( + () => process.exit(0), + );`, + ], + { cwd: import.meta.dirname, stdin: "pipe", stdout: "pipe", stderr: "pipe", timeout: 20_000 }, + ); + const stdout = child.stdout.pipeThrough(new TextDecoderStream()).getReader(); + let buffered = ""; + const nextLine = async () => { + while (!buffered.includes("\n")) { + const { value, done } = await stdout.read(); + if (done) throw new Error(`child exited early: ${await new Response(child.stderr).text()}`); + buffered += value; + } + const [line, ...rest] = buffered.split("\n"); + buffered = rest.join("\n"); + return line; + }; + try { + expect(await nextLine()).toBe(""); + await child.stdin.write("y\n"); + await child.stdin.flush(); + expect(await nextLine()).toBe("y"); + await child.stdin.end(); + const [exitCode, stderr] = await Promise.all([ + child.exited, + new Response(child.stderr).text(), + ]); + expect(exitCode, stderr).toBe(0); + expect(stderr).toContain("nonblock"); + } finally { + // A failed assertion must not leave the child waiting on its second prompt. + child.kill(); + } + }, 30_000); + + it("answers prompts from a flooded pipe and leaves the rest for a child inheriting fd 0", async () => { + // The production adapter in a real process: 2 MiB of lines are piped in, three prompts + // take the first three, then a child inheriting fd 0 counts what is left in the pipe. + // A reader that drained stdin would leave it nothing; this one reads a chunk ahead. + const bun = Bun.which("bun"); + if (!bun) throw new Error("Bun executable not found"); + const here = (file: string) => JSON.stringify(fileURLToPath(new URL(file, import.meta.url))); + const payload = enc(Array.from({ length: 200_000 }, (_, index) => `line-${index}\n`).join("")); + const child = Bun.spawn( + [ + bun, + "-e", + `import { Effect, Layer, Option } from "effect"; + import { Stdin } from ${here("./stdin.service.ts")}; + import { stdinLayer } from ${here("./stdin.layer.ts")}; + import { ttyLayer } from ${here("./tty.layer.ts")}; + const program = Effect.gen(function* () { + const stdin = yield* Stdin; + const answers = []; + for (let index = 0; index < 3; index++) { + answers.push(Option.getOrElse(yield* stdin.readLine(5_000), () => "")); + } + console.log(answers.join(" ")); + const rest = Bun.spawn( + [process.execPath, "-e", "let n = 0; for await (const c of Bun.stdin.stream()) n += c.length; console.log(n);"], + { stdin: "inherit", stdout: "pipe" }, + ); + console.log(yield* Effect.promise(() => new Response(rest.stdout).text())); + }); + Effect.runPromise(program.pipe(Effect.provide(stdinLayer.pipe(Layer.provide(ttyLayer))))).then( + () => process.exit(0), + );`, + ], + // Prompts give up after 3 x 5 s; a child that hangs anyway is killed at 20 s, ahead of + // vitest's 30 s guard, so the failure still carries its stderr. + { cwd: import.meta.dirname, stdin: payload, stdout: "pipe", stderr: "pipe", timeout: 20_000 }, + ); + const [exitCode, stdout, stderr] = await Promise.all([ + child.exited, + new Response(child.stdout).text(), + new Response(child.stderr).text(), + ]); + expect(exitCode, stderr).toBe(0); + const [answers, left] = stdout.trim().split("\n"); + expect(answers).toBe("line-0 line-1 line-2"); + expect(payload.length - Number(left)).toBeLessThanOrEqual(256 * 1024); + }, 30_000); +}); diff --git a/apps/cli/src/shared/runtime/stdin.layer.ts b/apps/cli/src/shared/runtime/stdin.layer.ts index dee28f3e21..b7891e8861 100644 --- a/apps/cli/src/shared/runtime/stdin.layer.ts +++ b/apps/cli/src/shared/runtime/stdin.layer.ts @@ -1,62 +1,119 @@ -import { Duration, Effect, Layer, Option, Pull, Queue, Ref, Scope, Stdio, Stream } from "effect"; +import { createReadStream } from "node:fs"; +import { BunStream } from "@effect/platform-bun"; +import { + Channel, + Duration, + Effect, + Layer, + Option, + Predicate, + Schedule, + Scope, + Stream, +} from "effect"; +import { systemError, type PlatformError } from "effect/PlatformError"; import { Tty } from "./tty.service.ts"; import { Stdin } from "./stdin.service.ts"; -const makeStdin = Effect.gen(function* () { - const stdio = yield* Stdio.Stdio; +// A parent that put the descriptor it hands down as fd 0 in non-blocking mode hands that mode +// down with it (`O_NONBLOCK` belongs to the shared open file, not to each process's descriptor), +// and an empty read then fails with `EAGAIN` instead of waiting for data. +const isEagain = (cause: unknown) => + cause instanceof Error && "code" in cause && cause.code === "EAGAIN"; + +// Bun's `process.stdin` cannot be throttled: one prompt is enough to start a read of the +// pipe that `pause`, `destroy` and detaching the listener all fail to stop, so an unbounded +// producer (`yes | supabase db push`) turns into an OOM kill. A file stream over fd 0 (the +// path is ignored once `fd` is given) honours backpressure: it reads at most one chunk ahead +// and leaves the rest in the pipe for a child inheriting fd 0, which `autoClose` keeps open +// when the stream is destroyed. Destroying it is what a reader's scope does on the way out: +// left alive, a stream whose listeners are gone would raise its next `EAGAIN` as an uncaught +// error. Clearing `O_NONBLOCK` would clear it for the parent too, so fd 0 is taken as it comes +// and `EAGAIN` is reported as `WouldBlock` for the reader to wait out. +// Ref: https://github.com/supabase/cli/issues/6287 +const processStdin: Stream.Stream = BunStream.fromReadable({ + evaluate: () => createReadStream("", { fd: 0, autoClose: false }), + onError: (cause) => + systemError({ + module: "Stdin", + method: "read", + _tag: isEagain(cause) ? "WouldBlock" : "Unknown", + description: cause instanceof Error ? cause.message : String(cause), + cause, + }), +}); + +// `splitLines` holds a partial line until its terminator arrives, so a producer that never +// sends one (`yes | tr -d '\n' | …`) would grow that buffer for as long as a prompt keeps +// pulling. This bounds it: once more than this many bytes are pending since the last line +// break, the next pull fails instead of reading further, and every prompt from then on takes +// its default. The check runs once per pull, so the buffer runs at most one pull, a chunk or +// two, past this bound before it trips. +const MAX_PENDING_LINE_BYTES = 64 * 1024; + +const boundPendingLine = (bytes: Stream.Stream) => + Stream.transformPull(bytes, (pull) => + Effect.sync(() => { + let pending = 0; + const tooLong = Effect.fail( + systemError({ + module: "Stdin", + method: "readLine", + _tag: "InvalidData", + description: `unterminated line exceeds ${MAX_PENDING_LINE_BYTES} bytes`, + }), + ); + return Effect.suspend(() => + pending > MAX_PENDING_LINE_BYTES + ? tooLong + : Effect.map(pull, (chunk) => { + for (const part of chunk) { + const lineEnd = Math.max(part.lastIndexOf(10), part.lastIndexOf(13)); + pending = lineEnd === -1 ? pending + part.length : part.length - lineEnd - 1; + } + return chunk; + }), + ); + }), + ); + +const makeStdin = Effect.fnUntraced(function* (stdin: Stream.Stream) { const tty = yield* Tty; const textDecoder = new TextDecoder(); - const scope = yield* Effect.scope; - const lineStream = stdio.stdin.pipe(Stream.decodeText(), Stream.splitLines); - - // A TTY answers at human speed, so it keeps the on-demand pull: nothing is read - // between prompts, leaving the keyboard to whatever reads stdin next. - const ttyLineReader = Effect.gen(function* () { - const pull = yield* Stream.toPull(lineStream).pipe(Scope.provide(scope)); - // Leftover lines from the last pulled chunk (a single pull may yield several). - const bufferRef = yield* Ref.make>([]); - return Effect.gen(function* () { - const buffered = yield* Ref.get(bufferRef); - if (buffered.length > 0) { - yield* Ref.set(bufferRef, buffered.slice(1)); - return Option.some(buffered[0] ?? ""); - } - return yield* Pull.matchEffect(pull, { - onSuccess: (chunk) => - Ref.set(bufferRef, chunk.slice(1)).pipe(Effect.as(Option.some(chunk[0] ?? ""))), - onFailure: () => Effect.succeedNone, - onDone: () => Effect.succeedNone, - }); - }); - }); - - // A pipe is read ahead into a bounded queue instead, because Bun's `process.stdin` - // cannot be throttled: one prompt is enough to start a read of the pipe that `pause`, - // `destroy` and detaching the listener all fail to stop. Everything left unconsumed - // accumulates for the rest of the command, which an unbounded producer turns into an - // OOM kill; consuming as fast as the pipe fills keeps memory flat. `dropping` discards - // the newest overflow, so the lines prompts read stay in pipe order. - // Ref: https://github.com/supabase/cli/issues/6287 - const pipedLineReader = Stream.toQueue(lineStream, { - // The pump reads at pipe speed, so this is in effect how many piped answers one run - // can use. `seed buckets` and `storage rm` prompt once per bucket, so it is sized to - // a project's bucket count rather than to a fixed handful of confirmations. - capacity: 1024, - strategy: "dropping", - }).pipe( - Scope.provide(scope), - Effect.map((queue) => - // EOF and read errors arrive as typed failures and become the prompt's default; - // a defect or an interrupt propagates rather than silently answering a prompt. - Queue.take(queue).pipe( - Effect.map(Option.some), - Effect.orElseSucceed(() => Option.none()), - ), + // `WouldBlock` is a non-blocking source with nothing to read yet, not a broken one: put a + // fresh reader over the still-open descriptor and ask again until data or EOF arrives, so the + // wait ends the way a blocking read's would, or with the prompt's timeout, which interrupts + // the retry. A poll, since the throttleable fd reader has no readiness signal to wait on; + // every 10 ms gives a piped prompt's 100 ms window ten looks, and stays fixed because the + // schedule keeps its state until data arrives, so a backoff reached while one prompt waited + // would still be slowing the next one down. Each look keeps a retry frame (a few KB) for the + // reader's lifetime, so a producer that never writes or closes costs a few hundred kilobytes + // a second of waiting, where a blocking read would hang for free. + const source = Stream.retry(stdin, ($) => + $(Schedule.spaced("10 millis")).pipe( + Schedule.while(({ input }) => Predicate.isTagged(input.reason, "WouldBlock")), ), ); + const scope = yield* Effect.scope; + const lineStream = source.pipe(boundPendingLine, Stream.decodeText(), Stream.splitLines); + + const lineReader = Effect.gen(function* () { + // One line per pull: `flattenArray` holds the rest of a multi-line chunk for the next + // pull, and `toPull` serializes pulls, so prompts running at once still take turns. + const pull = yield* Channel.toPull(Channel.flattenArray(Stream.toChannel(lineStream))).pipe( + Scope.provide(scope), + ); + // EOF, read errors and the line bound arrive as typed failures and become the prompt's + // default; a defect or an interrupt propagates rather than silently answering a prompt. + return pull.pipe( + Effect.map(Option.some), + Effect.orElseSucceed(() => Option.none()), + ); + }); + // Persistent, lazily-opened line reader shared by every `readLine` call, so a // command issuing several prompts (config push, seed buckets) reads the *next* piped // line each time instead of restarting from the top of the pipe. Opening it is @@ -64,11 +121,14 @@ const makeStdin = Effect.gen(function* () { // touched until the first `readLine`, so a TTY command that only prompts via clack // never grabs the keyboard (no contention with clack's own stdin capture), and the // reader outlives individual prompts. `splitLines` preserves interior blank lines so - // answers stay aligned across prompts. - const nextLine = yield* Effect.cached(tty.stdinIsTty ? ttyLineReader : pipedLineReader); + // answers stay aligned across prompts. A failed read stays failed for the rest of the + // process: what can go wrong with an open fd 0 (closed, hung up, not readable) does not + // mend on its own, so later prompts take their default instead of retrying a dead + // descriptor; the one transient failure, `WouldBlock`, is waited out upstream. + const nextLine = yield* Effect.cached(lineReader); const readPipedBytes = Effect.gen(function* () { - const chunks = yield* stdio.stdin.pipe(Stream.runCollect); + const chunks = yield* source.pipe(Stream.runCollect); const parts = Array.from(chunks); if (parts.length === 0) { return Option.none(); @@ -89,12 +149,11 @@ const makeStdin = Effect.gen(function* () { return Option.some(bytes); }).pipe(Effect.orElseSucceed(() => Option.none())); - // Read the next line (trimmed), bounded by `timeoutMillis`, from the persistent - // reader above. Mirrors Go's `Console.ReadLine` (`internal/utils/console.go:38-61`): - // successive calls return successive lines, and a timeout, EOF, or read error all - // collapse to `None` (Go returns "" — i.e. the prompt default — for each). The - // timeout bounds an open pipe that yields no newline (e.g. `yes y | …`) so it takes - // the default instead of blocking on EOF. + // Read the next line (trimmed), bounded by `timeoutMillis`, from the persistent reader + // above: successive calls return successive lines, and a timeout, EOF, or read error all + // collapse to `None`, the prompt's default. The timeout bounds a pipe that stays open + // without sending a line, and a user who never answers, so the prompt takes its default + // instead of waiting for EOF. const readLine = (timeoutMillis: number): Effect.Effect> => Effect.gen(function* () { const take = yield* nextLine; @@ -105,10 +164,10 @@ const makeStdin = Effect.gen(function* () { }); // Stream piped stdin without collecting it (constant memory). Read errors PROPAGATE on - // the error channel (unlike `readPipedBytes`'s `orElseSucceed(none)` swallow): Go's - // `io.Copy` returns `failed to copy from stdin` and exits non-zero rather than writing a - // truncated migration file, so the streaming consumer must surface the failure. - const pipedBytesStream = stdio.stdin; + // the error channel (unlike `readPipedBytes`'s `orElseSucceed(none)` swallow), once + // `WouldBlock` has been waited out above: a consumer writing the bytes to a file must fail + // rather than leave a truncated file behind. + const pipedBytesStream = source; return Stdin.of({ isTTY: tty.stdinIsTty, @@ -127,4 +186,8 @@ const makeStdin = Effect.gen(function* () { }); }); -export const stdinLayer = Layer.effect(Stdin, makeStdin); +/** `Stdin` over an arbitrary byte source, so tests can drive it with a controlled stream. */ +export const stdinLayerFrom = (stdin: Stream.Stream) => + Layer.effect(Stdin, makeStdin(stdin)); + +export const stdinLayer = stdinLayerFrom(processStdin); diff --git a/apps/cli/src/shared/runtime/stdin.layer.unit.test.ts b/apps/cli/src/shared/runtime/stdin.layer.unit.test.ts index 5c52a669f2..3b0d35246b 100644 --- a/apps/cli/src/shared/runtime/stdin.layer.unit.test.ts +++ b/apps/cli/src/shared/runtime/stdin.layer.unit.test.ts @@ -1,29 +1,30 @@ import { describe, expect, it } from "@effect/vitest"; -import { Effect, Layer, Option, Stdio, Stream } from "effect"; +import { Cause, Effect, Exit, Layer, Option, Stream } from "effect"; +import { systemError, type PlatformError } from "effect/PlatformError"; import { mockTty } from "../../../tests/helpers/mocks.ts"; import { Stdin } from "./stdin.service.ts"; -import { stdinLayer } from "./stdin.layer.ts"; +import { stdinLayerFrom } from "./stdin.layer.ts"; const encoder = new TextEncoder(); -function makeStdioLayer(stdin: Stream.Stream) { - return Layer.succeed( - Stdio.Stdio, - Stdio.make({ - args: Effect.succeed([]), - stdin, - stdout: { stream: Stream.empty, sink: { stream: Stream.empty } } as any, - stderr: { stream: Stream.empty, sink: { stream: Stream.empty } } as any, - }), - ); -} +// A read failure the layer keeps: `EAGAIN` would be waited out as `WouldBlock` instead. +const readError = Stream.fail( + systemError({ + module: "Stdin", + method: "read", + _tag: "Unknown", + description: "EIO: i/o error, read", + cause: new Error("EIO: i/o error, read"), + }), +); + +const withStdin = (stdin: Stream.Stream, stdinIsTty = false) => + stdinLayerFrom(stdin).pipe(Layer.provide(mockTty({ stdinIsTty }))); describe("Stdin", () => { describe("isTTY", () => { it.effect("returns true when Tty.stdinIsTty is true", () => { - const layer = stdinLayer.pipe( - Layer.provide(Layer.mergeAll(makeStdioLayer(Stream.empty), mockTty({ stdinIsTty: true }))), - ); + const layer = withStdin(Stream.empty, true); return Effect.gen(function* () { const { isTTY } = yield* Stdin; expect(isTTY).toBe(true); @@ -31,9 +32,7 @@ describe("Stdin", () => { }); it.effect("returns false when Tty.stdinIsTty is false", () => { - const layer = stdinLayer.pipe( - Layer.provide(Layer.mergeAll(makeStdioLayer(Stream.empty), mockTty({ stdinIsTty: false }))), - ); + const layer = withStdin(Stream.empty); return Effect.gen(function* () { const { isTTY } = yield* Stdin; expect(isTTY).toBe(false); @@ -45,9 +44,7 @@ describe("Stdin", () => { it.effect("returns Some(bytes) for valid input", () => { const expected = encoder.encode(" my-token-123 \n"); const stdin = Stream.fromIterable([expected]); - const layer = stdinLayer.pipe( - Layer.provide(Layer.mergeAll(makeStdioLayer(stdin), mockTty({ stdinIsTty: false }))), - ); + const layer = withStdin(stdin); return Effect.gen(function* () { const { readPipedBytes } = yield* Stdin; const result = yield* readPipedBytes; @@ -56,9 +53,7 @@ describe("Stdin", () => { }); it.effect("returns None for empty stream", () => { - const layer = stdinLayer.pipe( - Layer.provide(Layer.mergeAll(makeStdioLayer(Stream.empty), mockTty({ stdinIsTty: false }))), - ); + const layer = withStdin(Stream.empty); return Effect.gen(function* () { const { readPipedBytes } = yield* Stdin; const result = yield* readPipedBytes; @@ -67,10 +62,7 @@ describe("Stdin", () => { }); it.effect("returns None on stream error", () => { - const stdin = Stream.fail(new Error("read error")) as unknown as Stream.Stream; - const layer = stdinLayer.pipe( - Layer.provide(Layer.mergeAll(makeStdioLayer(stdin), mockTty({ stdinIsTty: false }))), - ); + const layer = withStdin(readError); return Effect.gen(function* () { const { readPipedBytes } = yield* Stdin; const result = yield* readPipedBytes; @@ -85,9 +77,7 @@ describe("Stdin", () => { encoder.encode("-chunk2"), encoder.encode("-chunk3"), ]); - const layer = stdinLayer.pipe( - Layer.provide(Layer.mergeAll(makeStdioLayer(stdin), mockTty({ stdinIsTty: false }))), - ); + const layer = withStdin(stdin); return Effect.gen(function* () { const { readPipedBytes } = yield* Stdin; const result = yield* readPipedBytes; @@ -98,9 +88,7 @@ describe("Stdin", () => { it.effect("preserves whitespace-only input", () => { const expected = encoder.encode(" \n \t "); const stdin = Stream.fromIterable([expected]); - const layer = stdinLayer.pipe( - Layer.provide(Layer.mergeAll(makeStdioLayer(stdin), mockTty({ stdinIsTty: false }))), - ); + const layer = withStdin(stdin); return Effect.gen(function* () { const { readPipedBytes } = yield* Stdin; const result = yield* readPipedBytes; @@ -112,9 +100,7 @@ describe("Stdin", () => { describe("readPipedText", () => { it.effect("returns Some(trimmed) for valid input", () => { const stdin = Stream.fromIterable([encoder.encode(" my-token-123 \n")]); - const layer = stdinLayer.pipe( - Layer.provide(Layer.mergeAll(makeStdioLayer(stdin), mockTty({ stdinIsTty: false }))), - ); + const layer = withStdin(stdin); return Effect.gen(function* () { const { readPipedText } = yield* Stdin; const result = yield* readPipedText; @@ -123,9 +109,7 @@ describe("Stdin", () => { }); it.effect("returns None for empty stream", () => { - const layer = stdinLayer.pipe( - Layer.provide(Layer.mergeAll(makeStdioLayer(Stream.empty), mockTty({ stdinIsTty: false }))), - ); + const layer = withStdin(Stream.empty); return Effect.gen(function* () { const { readPipedText } = yield* Stdin; const result = yield* readPipedText; @@ -134,10 +118,7 @@ describe("Stdin", () => { }); it.effect("returns None on stream error", () => { - const stdin = Stream.fail(new Error("read error")) as unknown as Stream.Stream; - const layer = stdinLayer.pipe( - Layer.provide(Layer.mergeAll(makeStdioLayer(stdin), mockTty({ stdinIsTty: false }))), - ); + const layer = withStdin(readError); return Effect.gen(function* () { const { readPipedText } = yield* Stdin; const result = yield* readPipedText; @@ -151,9 +132,7 @@ describe("Stdin", () => { encoder.encode("-chunk2"), encoder.encode("-chunk3"), ]); - const layer = stdinLayer.pipe( - Layer.provide(Layer.mergeAll(makeStdioLayer(stdin), mockTty({ stdinIsTty: false }))), - ); + const layer = withStdin(stdin); return Effect.gen(function* () { const { readPipedText } = yield* Stdin; const result = yield* readPipedText; @@ -163,9 +142,7 @@ describe("Stdin", () => { it.effect("returns None for whitespace-only input", () => { const stdin = Stream.fromIterable([encoder.encode(" \n \t ")]); - const layer = stdinLayer.pipe( - Layer.provide(Layer.mergeAll(makeStdioLayer(stdin), mockTty({ stdinIsTty: false }))), - ); + const layer = withStdin(stdin); return Effect.gen(function* () { const { readPipedText } = yield* Stdin; const result = yield* readPipedText; @@ -173,4 +150,35 @@ describe("Stdin", () => { }).pipe(Effect.provide(layer)); }); }); + + describe("readLine", () => { + it.effect("returns None at EOF", () => { + const layer = withStdin(Stream.empty); + return Effect.gen(function* () { + const { readLine } = yield* Stdin; + const result = yield* readLine(10_000); + expect(result).toEqual(Option.none()); + }).pipe(Effect.provide(layer)); + }); + + it.effect("returns None on a read error, for every prompt", () => { + const layer = withStdin(readError); + return Effect.gen(function* () { + const { readLine } = yield* Stdin; + const first = yield* readLine(10_000); + const second = yield* readLine(10_000); + expect(first).toEqual(Option.none()); + expect(second).toEqual(Option.none()); + }).pipe(Effect.provide(layer)); + }); + + it.effect("propagates a defect instead of answering the prompt", () => { + const layer = withStdin(Stream.die(new Error("boom"))); + return Effect.gen(function* () { + const { readLine } = yield* Stdin; + const exit = yield* readLine(10_000).pipe(Effect.exit); + expect(Exit.isFailure(exit) && Cause.hasDies(exit.cause)).toBe(true); + }).pipe(Effect.provide(layer)); + }); + }); }); diff --git a/apps/cli/src/shared/runtime/stdin.service.ts b/apps/cli/src/shared/runtime/stdin.service.ts index f2b6084ba2..2015aae6c5 100644 --- a/apps/cli/src/shared/runtime/stdin.service.ts +++ b/apps/cli/src/shared/runtime/stdin.service.ts @@ -2,17 +2,23 @@ import type { Effect, Option, Stream } from "effect"; import { Context } from "effect"; import type { PlatformError } from "effect/PlatformError"; +/** + * The process's stdin. `readPipedBytes`, `readPipedText`, `pipedBytesStream` and `readLine` + * each read fd 0 through a buffered reader of their own, so bytes one of them has read ahead + * are gone for the others: a command must use only one of them per invocation (`readLine` may + * be called repeatedly; its calls share a single reader). + */ interface StdinShape { readonly isTTY: boolean; readonly readPipedBytes: Effect.Effect>; /** * Piped stdin as a byte stream, for consumers that must avoid buffering the whole - * pipe (e.g. `migration new` seeding a file from a large `pg_dump`, matching Go's - * `io.Copy` streaming). Unlike {@link readPipedBytes}, read errors PROPAGATE on the - * error channel — Go's `io.Copy` returns `failed to copy from stdin` and exits - * non-zero rather than writing a truncated file, so the caller must map the failure. - * Emits nothing for an empty pipe; callers gate on {@link isTTY} themselves (a TTY - * should not be drained). + * pipe (e.g. `migration new` seeding a file from a large `pg_dump`). Unlike + * {@link readPipedBytes}, read errors PROPAGATE on the error channel (once a non-blocking + * fd 0 with nothing to read yet has been waited out): a caller writing the bytes to a file + * must fail rather than leave a truncated file behind, so it maps the failure itself. Emits + * nothing for an empty pipe; callers gate on {@link isTTY} themselves (a TTY should not be + * drained). */ readonly pipedBytesStream: Stream.Stream; readonly readPipedText: Effect.Effect>; @@ -27,9 +33,13 @@ interface StdinShape { * {@link readPipedText} (a whole-stream collect), this reads line by line, so it * works for an interactive terminal as well as a pipe. * - * On a PIPE lines are read ahead into a bounded buffer rather than on demand, so only - * the first N piped lines are answerable and anything past them is dropped. See - * `stdin.layer.ts` for N and for why reading ahead is required at all. + * stdin is pulled a chunk (64 KiB) at a time and only when a prompt needs a line, so a + * producer that outruns the prompts stays in the pipe apart from the chunk or two the reader + * holds for later prompts. A read error ends line reading the way EOF does: that prompt and + * every later one get `None`. So does a line that runs past 64 KiB without a line break + * (`MAX_PENDING_LINE_BYTES` in `stdin.layer.ts`): the reader stops a chunk or two past that + * bound and never reaches whatever follows. A non-blocking fd 0 with nothing to read yet is + * waited on, not treated as a read error. */ readonly readLine: (timeoutMillis: number) => Effect.Effect>; }