| author | |
| committer | |
| log | 803c2b77387bbaf5f10319f8ac6e5742b68d9612 |
| tree | 3d5055e5fdfc9979cbe2db3a993d8a8fe9bcc099 |
| parent | 28e7c9ae17affc31e028a06cf33911fabe84d2cd |
| signature | Signed by PGP key B5690EEEBB952194 |
Co-authored-by: codex <codex@users.noreply.github.com>2 files changed, 178 insertions(+), 8 deletions(-)
apps/desktop/src/backend/DesktopBackendManager.test.ts+89-2| ... | @@ -59,8 +59,8 @@ const configWithObservability: DesktopBackendBootstrapValue = { | ... | @@ -59,8 +59,8 @@ const configWithObservability: DesktopBackendBootstrapValue = { |
| 59 | }; | 59 | }; |
| 60 | 60 | ||
| 61 | function makeProcess(options?: { | 61 | function makeProcess(options?: { |
| 62 | readonly stdout?: Stream.Stream<Uint8Array>; | 62 | readonly stdout?: Stream.Stream<Uint8Array, PlatformError.PlatformError>; |
| 63 | readonly stderr?: Stream.Stream<Uint8Array>; | 63 | readonly stderr?: Stream.Stream<Uint8Array, PlatformError.PlatformError>; |
| 64 | readonly exitCode?: Effect.Effect<ChildProcessSpawner.ExitCode, PlatformError.PlatformError>; | 64 | readonly exitCode?: Effect.Effect<ChildProcessSpawner.ExitCode, PlatformError.PlatformError>; |
| 65 | readonly kill?: ChildProcessSpawner.ChildProcessHandle["kill"]; | 65 | readonly kill?: ChildProcessSpawner.ChildProcessHandle["kill"]; |
| 66 | }): ChildProcessSpawner.ChildProcessHandle { | 66 | }): ChildProcessSpawner.ChildProcessHandle { |
| ... | @@ -389,6 +389,93 @@ describe("DesktopBackendManager", () => { | ... | @@ -389,6 +389,93 @@ describe("DesktopBackendManager", () => { |
| 389 | }), | 389 | }), |
| 390 | ); | 390 | ); |
| 391 | 391 | ||
| 392 | it.effect("reports output stream failures with process and stream context", () => | ||
| 393 | Effect.gen(function* () { | ||
| 394 | const outputCause = PlatformError.systemError({ | ||
| 395 | _tag: "BadResource", | ||
| 396 | module: "ChildProcess", | ||
| 397 | method: "stdout", | ||
| 398 | description: "output-stream-secret-sentinel", | ||
| 399 | }); | ||
| 400 | const reported = yield* Deferred.make<DesktopBackendManager.BackendProcessOutputError>(); | ||
| 401 | const spawnerLayer = Layer.succeed( | ||
| 402 | ChildProcessSpawner.ChildProcessSpawner, | ||
| 403 | ChildProcessSpawner.make(() => | ||
| 404 | Effect.succeed( | ||
| 405 | makeProcess({ | ||
| 406 | stdout: Stream.fail(outputCause), | ||
| 407 | exitCode: Deferred.await(reported).pipe(Effect.as(ChildProcessSpawner.ExitCode(0))), | ||
| 408 | }), | ||
| 409 | ), | ||
| 410 | ), | ||
| 411 | ); | ||
| 412 | |||
| 413 | const exit = yield* DesktopBackendManager.runBackendProcess({ | ||
| 414 | ...baseConfig, | ||
| 415 | onOutputFailure: (error) => Deferred.succeed(reported, error).pipe(Effect.asVoid), | ||
| 416 | }).pipe(Effect.scoped, Effect.provide(Layer.merge(spawnerLayer, healthyHttpClientLayer))); | ||
| 417 | const error = yield* Deferred.await(reported); | ||
| 418 | |||
| 419 | assert.equal(exit.code.pipe(Option.getOrUndefined), 0); | ||
| 420 | if (error._tag !== "BackendProcessOutputReadError") { | ||
| 421 | return assert.fail(`Expected output read error, received ${error._tag}`); | ||
| 422 | } | ||
| 423 | assert.equal(error.executablePath, "/electron"); | ||
| 424 | assert.equal(error.entryPath, "/server/bin.mjs"); | ||
| 425 | assert.equal(error.cwd, "/server"); | ||
| 426 | assert.equal(error.httpBaseUrl.href, "http://127.0.0.1:3773/"); | ||
| 427 | assert.equal(error.pid, 123); | ||
| 428 | assert.equal(error.streamName, "stdout"); | ||
| 429 | assert.strictEqual(error.cause, outputCause); | ||
| 430 | assert.equal(error.message, "Failed to read stdout from desktop backend process 123."); | ||
| 431 | assert.notInclude(error.message, "output-stream-secret-sentinel"); | ||
| 432 | }), | ||
| 433 | ); | ||
| 434 | |||
| 435 | it.effect("reports output handler failures separately from stream read failures", () => | ||
| 436 | Effect.gen(function* () { | ||
| 437 | const chunk = new TextEncoder().encode("backend output"); | ||
| 438 | const outputCause = new Error("output-handler-secret-sentinel"); | ||
| 439 | const reported = yield* Deferred.make<DesktopBackendManager.BackendProcessOutputError>(); | ||
| 440 | const spawnerLayer = Layer.succeed( | ||
| 441 | ChildProcessSpawner.ChildProcessSpawner, | ||
| 442 | ChildProcessSpawner.make(() => | ||
| 443 | Effect.succeed( | ||
| 444 | makeProcess({ | ||
| 445 | stdout: Stream.make(chunk), | ||
| 446 | exitCode: Deferred.await(reported).pipe(Effect.as(ChildProcessSpawner.ExitCode(0))), | ||
| 447 | }), | ||
| 448 | ), | ||
| 449 | ), | ||
| 450 | ); | ||
| 451 | |||
| 452 | const exit = yield* DesktopBackendManager.runBackendProcess({ | ||
| 453 | ...baseConfig, | ||
| 454 | onOutput: () => Effect.fail(outputCause), | ||
| 455 | onOutputFailure: (error) => Deferred.succeed(reported, error).pipe(Effect.asVoid), | ||
| 456 | }).pipe(Effect.scoped, Effect.provide(Layer.merge(spawnerLayer, healthyHttpClientLayer))); | ||
| 457 | const error = yield* Deferred.await(reported); | ||
| 458 | |||
| 459 | assert.equal(exit.code.pipe(Option.getOrUndefined), 0); | ||
| 460 | if (error._tag !== "BackendProcessOutputHandlingError") { | ||
| 461 | return assert.fail(`Expected output handling error, received ${error._tag}`); | ||
| 462 | } | ||
| 463 | assert.equal(error.executablePath, "/electron"); | ||
| 464 | assert.equal(error.entryPath, "/server/bin.mjs"); | ||
| 465 | assert.equal(error.cwd, "/server"); | ||
| 466 | assert.equal(error.httpBaseUrl.href, "http://127.0.0.1:3773/"); | ||
| 467 | assert.equal(error.pid, 123); | ||
| 468 | assert.equal(error.streamName, "stdout"); | ||
| 469 | assert.equal(error.chunkByteLength, chunk.byteLength); | ||
| 470 | assert.strictEqual(error.cause, outputCause); | ||
| 471 | assert.equal( | ||
| 472 | error.message, | ||
| 473 | `Failed to handle ${chunk.byteLength} bytes from stdout of desktop backend process 123.`, | ||
| 474 | ); | ||
| 475 | assert.notInclude(error.message, "output-handler-secret-sentinel"); | ||
| 476 | }), | ||
| 477 | ); | ||
| 478 | |||
| 392 | it.effect("retries HTTP readiness before reporting the backend ready", () => | 479 | it.effect("retries HTTP readiness before reporting the backend ready", () => |
| 393 | Effect.gen(function* () { | 480 | Effect.gen(function* () { |
| 394 | const requestUrls: Array<string> = []; | 481 | const requestUrls: Array<string> = []; |
apps/desktop/src/backend/DesktopBackendManager.ts+89-6| ... | @@ -105,6 +105,39 @@ export class BackendProcessSpawnError extends Schema.TaggedErrorClass<BackendPro | ... | @@ -105,6 +105,39 @@ export class BackendProcessSpawnError extends Schema.TaggedErrorClass<BackendPro |
| 105 | } | 105 | } |
| 106 | } | 106 | } |
| 107 | 107 | ||
| 108 | export class BackendProcessOutputReadError extends Schema.TaggedErrorClass<BackendProcessOutputReadError>()( | ||
| 109 | "BackendProcessOutputReadError", | ||
| 110 | { | ||
| 111 | ...backendProcessContextSchema, | ||
| 112 | pid: Schema.Number, | ||
| 113 | streamName: Schema.Literals(["stdout", "stderr"]), | ||
| 114 | cause: Schema.Defect(), | ||
| 115 | }, | ||
| 116 | ) { | ||
| 117 | override get message(): string { | ||
| 118 | return `Failed to read ${this.streamName} from desktop backend process ${this.pid}.`; | ||
| 119 | } | ||
| 120 | } | ||
| 121 | |||
| 122 | export class BackendProcessOutputHandlingError extends Schema.TaggedErrorClass<BackendProcessOutputHandlingError>()( | ||
| 123 | "BackendProcessOutputHandlingError", | ||
| 124 | { | ||
| 125 | ...backendProcessContextSchema, | ||
| 126 | pid: Schema.Number, | ||
| 127 | streamName: Schema.Literals(["stdout", "stderr"]), | ||
| 128 | chunkByteLength: Schema.Number, | ||
| 129 | cause: Schema.Defect(), | ||
| 130 | }, | ||
| 131 | ) { | ||
| 132 | override get message(): string { | ||
| 133 | return `Failed to handle ${this.chunkByteLength} bytes from ${this.streamName} of desktop backend process ${this.pid}.`; | ||
| 134 | } | ||
| 135 | } | ||
| 136 | |||
| 137 | export type BackendProcessOutputError = | ||
| 138 | | BackendProcessOutputReadError | ||
| 139 | | BackendProcessOutputHandlingError; | ||
| 140 | |||
| 108 | export class BackendProcessExitStatusError extends Schema.TaggedErrorClass<BackendProcessExitStatusError>()( | 141 | export class BackendProcessExitStatusError extends Schema.TaggedErrorClass<BackendProcessExitStatusError>()( |
| 109 | "BackendProcessExitStatusError", | 142 | "BackendProcessExitStatusError", |
| 110 | { | 143 | { |
| ... | @@ -146,7 +179,8 @@ interface RunBackendProcessOptions extends DesktopBackendStartConfig { | ... | @@ -146,7 +179,8 @@ interface RunBackendProcessOptions extends DesktopBackendStartConfig { |
| 146 | readonly onOutput?: ( | 179 | readonly onOutput?: ( |
| 147 | streamName: BackendProcessOutputStream, | 180 | streamName: BackendProcessOutputStream, |
| 148 | chunk: Uint8Array, | 181 | chunk: Uint8Array, |
| 149 | ) => Effect.Effect<void>; | 182 | ) => Effect.Effect<void, Error>; |
| 183 | readonly onOutputFailure?: (error: BackendProcessOutputError) => Effect.Effect<void>; | ||
| 150 | } | 184 | } |
| 151 | 185 | ||
| 152 | export interface DesktopBackendSnapshot { | 186 | export interface DesktopBackendSnapshot { |
| ... | @@ -254,13 +288,41 @@ export const waitForHttpReady = Effect.fn("desktop.backendManager.waitForHttpRea | ... | @@ -254,13 +288,41 @@ export const waitForHttpReady = Effect.fn("desktop.backendManager.waitForHttpRea |
| 254 | }); | 288 | }); |
| 255 | 289 | ||
| 256 | function drainBackendOutput( | 290 | function drainBackendOutput( |
| 291 | context: BackendProcessContext & { readonly pid: number }, | ||
| 257 | streamName: BackendProcessOutputStream, | 292 | streamName: BackendProcessOutputStream, |
| 258 | stream: Stream.Stream<Uint8Array, PlatformError.PlatformError>, | 293 | stream: Stream.Stream<Uint8Array, PlatformError.PlatformError>, |
| 259 | onOutput: (streamName: BackendProcessOutputStream, chunk: Uint8Array) => Effect.Effect<void>, | 294 | onOutput: ( |
| 295 | streamName: BackendProcessOutputStream, | ||
| 296 | chunk: Uint8Array, | ||
| 297 | ) => Effect.Effect<void, Error>, | ||
| 298 | onOutputFailure: (error: BackendProcessOutputError) => Effect.Effect<void>, | ||
| 260 | ): Effect.Effect<void> { | 299 | ): Effect.Effect<void> { |
| 261 | return stream.pipe( | 300 | return stream.pipe( |
| 262 | Stream.runForEach((chunk) => onOutput(streamName, chunk)), | 301 | Stream.mapError( |
| 263 | Effect.ignore, | 302 | (cause) => |
| 303 | new BackendProcessOutputReadError({ | ||
| 304 | ...context, | ||
| 305 | streamName, | ||
| 306 | cause, | ||
| 307 | }), | ||
| 308 | ), | ||
| 309 | Stream.runForEach((chunk) => | ||
| 310 | onOutput(streamName, chunk).pipe( | ||
| 311 | Effect.mapError( | ||
| 312 | (cause) => | ||
| 313 | new BackendProcessOutputHandlingError({ | ||
| 314 | ...context, | ||
| 315 | streamName, | ||
| 316 | chunkByteLength: chunk.byteLength, | ||
| 317 | cause, | ||
| 318 | }), | ||
| 319 | ), | ||
| 320 | ), | ||
| 321 | ), | ||
| 322 | Effect.catchTags({ | ||
| 323 | BackendProcessOutputReadError: onOutputFailure, | ||
| 324 | BackendProcessOutputHandlingError: onOutputFailure, | ||
| 325 | }), | ||
| 264 | ); | 326 | ); |
| 265 | } | 327 | } |
| 266 | 328 | ||
| ... | @@ -321,8 +383,28 @@ export const runBackendProcess = Effect.fn("runBackendProcess")(function* ( | ... | @@ -321,8 +383,28 @@ export const runBackendProcess = Effect.fn("runBackendProcess")(function* ( |
| 321 | 383 | ||
| 322 | yield* options.onStarted?.(handle.pid) ?? Effect.void; | 384 | yield* options.onStarted?.(handle.pid) ?? Effect.void; |
| 323 | if (options.captureOutput) { | 385 | if (options.captureOutput) { |
| 324 | yield* drainBackendOutput("stdout", handle.stdout, onOutput).pipe(Effect.forkScoped); | 386 | const outputContext = { |
| 325 | yield* drainBackendOutput("stderr", handle.stderr, onOutput).pipe(Effect.forkScoped); | 387 | executablePath: options.executablePath, |
| 388 | entryPath: options.entryPath, | ||
| 389 | cwd: options.cwd, | ||
| 390 | httpBaseUrl: options.httpBaseUrl, | ||
| 391 | pid: Number(handle.pid), | ||
| 392 | }; | ||
| 393 | const onOutputFailure = options.onOutputFailure ?? (() => Effect.void); | ||
| 394 | yield* drainBackendOutput( | ||
| 395 | outputContext, | ||
| 396 | "stdout", | ||
| 397 | handle.stdout, | ||
| 398 | onOutput, | ||
| 399 | onOutputFailure, | ||
| 400 | ).pipe(Effect.forkScoped); | ||
| 401 | yield* drainBackendOutput( | ||
| 402 | outputContext, | ||
| 403 | "stderr", | ||
| 404 | handle.stderr, | ||
| 405 | onOutput, | ||
| 406 | onOutputFailure, | ||
| 407 | ).pipe(Effect.forkScoped); | ||
| 326 | } | 408 | } |
| 327 | yield* waitForHttpReady({ | 409 | yield* waitForHttpReady({ |
| 328 | executablePath: options.executablePath, | 410 | executablePath: options.executablePath, |
| ... | @@ -560,6 +642,7 @@ export const make = Effect.gen(function* () { | ... | @@ -560,6 +642,7 @@ export const make = Effect.gen(function* () { |
| 560 | error, | 642 | error, |
| 561 | }), | 643 | }), |
| 562 | onOutput: (streamName, chunk) => backendOutputLog.writeOutputChunk(streamName, chunk), | 644 | onOutput: (streamName, chunk) => backendOutputLog.writeOutputChunk(streamName, chunk), |
| 645 | onOutputFailure: (error) => logBackendManagerError(error.message, { error }), | ||
| 563 | }).pipe( | 646 | }).pipe( |
| 564 | Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), | 647 | Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), |
| 565 | Effect.provideService(HttpClient.HttpClient, httpClient), | 648 | Effect.provideService(HttpClient.HttpClient, httpClient), |