diff --git a/packages/core/src/cross-spawn-spawner.ts b/packages/core/src/cross-spawn-spawner.ts index 6ea9022acf1d..80de63261ba0 100644 --- a/packages/core/src/cross-spawn-spawner.ts +++ b/packages/core/src/cross-spawn-spawner.ts @@ -228,12 +228,42 @@ export const make = Effect.gen(function* () { Effect.suspend(() => { let sink: Sink.Sink = Sink.drain if (Predicate.isNotNull(proc.stdin)) { + const writable = proc.stdin + const failed = Deferred.makeUnsafe() + const onError = (err: Error) => + Deferred.doneUnsafe(failed, Effect.fail(toPlatformError("fromWritable(stdin)", err, command))) + // NodeSink's input listener ends before writable.end(); keep late pipe errors owned until close. + writable.on("error", onError) + writable.once("close", () => writable.off("error", onError)) sink = NodeSink.fromWritable({ - evaluate: () => proc.stdin!, + evaluate: () => writable, onError: (err) => toPlatformError("fromWritable(stdin)", toError(err), command), - endOnDone: cfg.endOnDone, + endOnDone: false, encoding: cfg.encoding, }) + if (cfg.endOnDone) { + sink = sink.pipe( + Sink.mapEffect(() => + Effect.callback((resume) => { + if (writable.errored) + return resume(Effect.fail(toPlatformError("end(stdin)", writable.errored, command))) + if (writable.writableFinished || writable.closed) return resume(Effect.void) + const finish = () => resume(Effect.void) + const close = () => + resume( + Effect.fail(toPlatformError("end(stdin)", new Error("stdin closed before finishing"), command)), + ) + writable.once("finish", finish) + writable.once("close", close) + writable.end() + return Effect.sync(() => { + writable.off("finish", finish) + writable.off("close", close) + }) + }).pipe(Effect.raceFirst(Deferred.await(failed))), + ), + ) + } } if (Stream.isStream(cfg.stream)) return Effect.as(Effect.forkScoped(Stream.run(cfg.stream, sink)), sink) return Effect.succeed(sink) diff --git a/packages/core/test/process/process.test.ts b/packages/core/test/process/process.test.ts index 82b6dc4717a3..359f30a667f1 100644 --- a/packages/core/test/process/process.test.ts +++ b/packages/core/test/process/process.test.ts @@ -292,6 +292,89 @@ describe("AppProcess", () => { ) }) + describe("stdin lifecycle", () => { + it.live( + "preserves early child exit diagnostics when a small stdin pipe closes", + Effect.gen(function* () { + const svc = yield* AppProcess.Service + const command = + process.platform === "win32" + ? ChildProcess.make("cmd.exe", ["/d", "/s", "/c", "echo rejected 1>&2 & exit /b 128"]) + : ChildProcess.make("/bin/sh", ["-c", "printf rejected >&2; exit 128"]) + for (let attempt = 0; attempt < 50; attempt++) { + const result = yield* svc.run(command, { stdin: "x" }) + expect(result.exitCode).toBe(128) + expect(result.stderr.toString("utf8").trim()).toBe("rejected") + } + }), + ) + + it.live( + "keeps caller-managed stdin open across multiple writes", + Effect.gen(function* () { + const svc = yield* AppProcess.Service + const handle = yield* svc.spawn( + ChildProcess.make( + NODE, + [ + "-e", + ` + let text = "" + process.stdin.on("data", (chunk) => { + text += chunk + if (text.length === 2) { + process.stdout.write(text) + process.exit(0) + } + }) + process.stdin.on("end", () => process.exit(1)) + `, + ], + { stdin: { stream: "pipe", endOnDone: false } }, + ), + ) + yield* Stream.run(Stream.make(Buffer.from("a")), handle.stdin) + yield* Stream.run(Stream.make(Buffer.from("b")), handle.stdin) + const out = yield* Stream.runCollect(handle.stdout) + expect(Buffer.concat(Array.from(out)).toString()).toBe("ab") + expect(Number(yield* handle.exitCode)).toBe(0) + }), + ) + + it.live( + "reports a closed child stdin as a typed write failure without losing its diagnostics", + Effect.gen(function* () { + const svc = yield* AppProcess.Service + const handle = yield* svc.spawn(cmd("-e", "process.stderr.write('rejected'); process.exit(128)")) + const exit = yield* Stream.run(Stream.make(Buffer.alloc(1024 * 1024)), handle.stdin).pipe(Effect.exit) + expect(Exit.isFailure(exit)).toBe(true) + if (Exit.isFailure(exit)) { + expect(exit.cause.reasons[0]).toMatchObject({ + _tag: "Fail", + error: { _tag: "PlatformError", reason: { module: "ChildProcess", method: "fromWritable(stdin)" } }, + }) + } + const stderr = yield* Stream.runCollect(handle.stderr) + expect(Buffer.concat(Array.from(stderr)).toString()).toBe("rejected") + expect(Number(yield* handle.exitCode)).toBe(128) + }), + ) + + it.live( + "preserves failures from the stdin producer", + Effect.gen(function* () { + const svc = yield* AppProcess.Service + const handle = yield* svc.spawn(cmd("-e", "process.stdin.resume()")) + const error = new Error("stdin producer failed") + const exit = yield* Stream.run(Stream.fail(error), handle.stdin).pipe(Effect.exit) + expect(Exit.isFailure(exit)).toBe(true) + if (Exit.isFailure(exit)) { + expect(exit.cause.reasons[0]).toMatchObject({ _tag: "Fail", error }) + } + }), + ) + }) + describe("runStream", () => { it.live( "emits lines incrementally and ends cleanly on exit 0", diff --git a/packages/opencode/src/snapshot/index.ts b/packages/opencode/src/snapshot/index.ts index 4da9bc3ca864..4b5043900c0e 100644 --- a/packages/opencode/src/snapshot/index.ts +++ b/packages/opencode/src/snapshot/index.ts @@ -80,15 +80,31 @@ const layer: Layer.Layer; stdin?: string }) { - const result = yield* appProcess.run( - ChildProcess.make("git", cmd, { cwd: opts?.cwd, env: opts?.env, extendEnv: true }), - { stdin: opts?.stdin }, - ) - return { - code: ChildProcessSpawner.ExitCode(result.exitCode), - text: result.stdout.toString("utf8"), - stderr: result.stderr.toString("utf8"), - } satisfies GitResult + for (let attempt = 0; ; attempt++) { + const result = yield* appProcess.run( + ChildProcess.make("git", cmd, { + cwd: opts?.cwd, + env: { ...opts?.env, LC_ALL: "C" }, + extendEnv: true, + }), + { stdin: opts?.stdin }, + ) + const stderr = result.stderr.toString("utf8") + // Other OpenCode processes share this index; their locks must be released by their owners. + if ( + result.exitCode !== 128 || + !stderr.includes("index.lock") || + !stderr.includes("File exists") || + attempt === 6 + ) { + return { + code: ChildProcessSpawner.ExitCode(result.exitCode), + text: result.stdout.toString("utf8"), + stderr, + } satisfies GitResult + } + yield* Effect.sleep(Duration.millis(50 * 2 ** attempt)) + } }, Effect.catch((err) => Effect.succeed({ @@ -144,7 +160,7 @@ const layer: Layer.Layer fs.exists(file).pipe(Effect.orDie) @@ -252,13 +269,13 @@ const layer: Layer.Layer !ignored.has(item)) - if (!allow.length) return + if (!allow.length) return true const large = new Set( (yield* Effect.all( @@ -294,7 +311,7 @@ const layer: Layer.Layer large.has(item))) yield* sync(Array.from(block)) // Stage only the allowed candidate paths so snapshot updates stay scoped. - yield* stage(allow.filter((item) => !block.has(item))) + return yield* stage(allow.filter((item) => !block.has(item))) }) const cleanup = Effect.fnUntraced(function* () { @@ -337,8 +354,15 @@ const layer: Layer.Layer + Bun.spawn(["git", "--git-dir", gitdir, "update-index", "--index-info"], { + cwd, + stdin: "pipe", + stdout: "ignore", + stderr: "pipe", + }), + ), + (proc) => + Effect.promise(async () => { + if (proc.exitCode === null) proc.stdin.end() + await proc.exited + }), + ) + yield* Effect.promise(async () => { + const deadline = Date.now() + 2000 + while (!(await Bun.file(path.join(gitdir, "index.lock")).exists())) { + if (Date.now() > deadline) throw new Error("git did not acquire the snapshot index lock") + await Bun.sleep(10) + } + }) + return holder +}) + +it.instance( + "track retries an index lock held by another git process without losing new files", + Effect.gen(function* () { + const tmp = yield* bootstrap() + const snapshot = yield* Snapshot.Service + const before = yield* snapshot.track() + const ctx = yield* InstanceState.context + const gitdir = path.join(Global.Path.data, "snapshot", ctx.project.id, Hash.fast(ctx.worktree)) + yield* write(`${tmp.path}/contended.txt`, "captured after contention") + + const holder = yield* holdIndex(gitdir, tmp.path) + const pending = yield* snapshot.track().pipe(Effect.forkChild) + yield* Effect.sleep("250 millis") + holder.stdin.end() + expect(yield* Effect.promise(() => holder.exited)).toBe(0) + const after = yield* Fiber.join(pending) + expect(after).toMatch(/^[0-9a-f]{40}$/) + expect(after).not.toBe(before) + const proc = Bun.spawn(["git", "--git-dir", gitdir, "show", `${after}:contended.txt`], { + stdout: "pipe", + stderr: "pipe", + }) + expect(yield* Effect.promise(() => new Response(proc.stdout).text())).toBe("captured after contention") + expect(yield* Effect.promise(() => proc.exited)).toBe(0) + }), + { git: true }, +) + +it.instance( + "track stops retrying a persistent index collision and leaves the other owner alive", + Effect.gen(function* () { + const tmp = yield* bootstrap() + const snapshot = yield* Snapshot.Service + yield* snapshot.track() + const ctx = yield* InstanceState.context + const gitdir = path.join(Global.Path.data, "snapshot", ctx.project.id, Hash.fast(ctx.worktree)) + yield* write(`${tmp.path}/contended.txt`, "not snapshotted while the index is locked") + const holder = yield* holdIndex(gitdir, tmp.path) + expect(yield* snapshot.track().pipe(Effect.timeout("12 seconds"))).toBeUndefined() + expect(yield* exists(path.join(gitdir, "index.lock"))).toBe(true) + expect(holder.exitCode).toBeNull() + }), + { git: true }, +) + +it.instance( + "track does not return a stale tree or retry an unrelated fatal git staging error", + Effect.gen(function* () { + const tmp = yield* bootstrap() + const snapshot = yield* Snapshot.Service + yield* snapshot.track() + const ctx = yield* InstanceState.context + const gitdir = path.join(Global.Path.data, "snapshot", ctx.project.id, Hash.fast(ctx.worktree)) + yield* exec(tmp.path, [ + "git", + "--git-dir", + gitdir, + "config", + "filter.reject.clean", + "printf x >> filter-count; false", + ]) + yield* exec(tmp.path, ["git", "--git-dir", gitdir, "config", "filter.reject.required", "true"]) + yield* write(`${tmp.path}/.gitattributes`, "*.fail filter=reject\n") + yield* write(`${tmp.path}/new.fail`, "must not be claimed as captured") + expect(yield* snapshot.track()).toBeUndefined() + expect(yield* readText(`${tmp.path}/filter-count`)).toBe("x") + }), + { git: true }, +)