diff --git a/packages/admin/src/lib/functions/on-call-stream.spec.ts b/packages/admin/src/lib/functions/on-call-stream.spec.ts index a8005f2..1af8c52 100644 --- a/packages/admin/src/lib/functions/on-call-stream.spec.ts +++ b/packages/admin/src/lib/functions/on-call-stream.spec.ts @@ -1,7 +1,25 @@ -import { Effect, ManagedRuntime, Layer, Schema, Stream } from 'effect'; -import { describe, expect, it } from '@effect/vitest'; +import { + Data, + Effect, + ErrorReporter, + Layer, + ManagedRuntime, + Schema, + Stream, +} from 'effect'; +import { + afterEach, + beforeEach, + describe, + expect, + it, + vi, +} from '@effect/vitest'; +import { logger } from 'firebase-functions'; +import { HttpsError } from 'firebase-functions/https'; import { onCallStreamEffect } from './on-call-stream.js'; import { onCallEffect } from './on-call.js'; +import { FunctionSetupError } from './setup-error.js'; import { makeCallableRequest, runCallable, @@ -132,14 +150,202 @@ describe('onCallStreamEffect', () => { }); it('fails when the input does not match the schema', async () => { + let handlerRan = false; + const fn = onCallStreamEffect( + { runtime, inputSchema: Schema.Struct({ count: Schema.Number }) }, + (input) => { + handlerRan = true; + return Stream.make(input.count); + }, + ); + + const error = await streamCallable(fn, { count: 'nope' } as never) + .data.then(() => undefined) + .catch((e: unknown) => e); + + expect(error).toBeInstanceOf(HttpsError); + expect((error as HttpsError).code).toBe('invalid-argument'); + expect((error as HttpsError).message).toContain('count'); + expect(handlerRan).toBe(false); + }); +}); + +describe('onCallStreamEffect setup-error recovery', () => { + it('rejects with an invalid-argument HttpsError carrying the decode message', async () => { const fn = onCallStreamEffect( { runtime, inputSchema: Schema.Struct({ count: Schema.Number }) }, (input) => Stream.make(input.count), ); - await expect( - streamCallable(fn, { count: 'nope' } as never).data, - ).rejects.toThrow(); + const error = await streamCallable(fn, { count: 'nope' } as never) + .data.then(() => undefined) + .catch((e: unknown) => e); + + expect(error).toBeInstanceOf(HttpsError); + expect((error as HttpsError).code).toBe('invalid-argument'); + expect((error as HttpsError).message).toMatch(/Expected number/i); + expect((error as HttpsError).message).toContain('count'); + }); + + it('rejects with an internal HttpsError when a chunk fails chunkSchema', async () => { + const fn = onCallStreamEffect( + { + runtime, + inputSchema: Schema.Struct({ name: Schema.String }), + chunkSchema: Schema.Struct({ name: Schema.NonEmptyString }), + }, + (input) => Stream.succeed({ name: input.name }), + ); + + // '' passes inputSchema (String) but is rejected by NonEmptyString at the + // encode boundary. + const error = await streamCallable(fn, { name: '' } as never) + .data.then(() => undefined) + .catch((e: unknown) => e); + + expect(error).toBeInstanceOf(HttpsError); + expect((error as HttpsError).code).toBe('internal'); + expect((error as HttpsError).message).toBe( + 'Failed to encode function output', + ); + }); + + it('fails the stream at the first unencodable chunk, keeping the chunks sent before it', async () => { + const fn = onCallStreamEffect( + { + runtime, + chunkSchema: Schema.Struct({ name: Schema.NonEmptyString }), + }, + () => Stream.make({ name: 'ok' }, { name: '' }, { name: 'also-ok' }), + ); + + const { stream, data } = streamCallable(fn, null); + const seen: unknown[] = []; + for await (const chunk of stream) seen.push(chunk); + + expect(seen).toEqual([{ name: 'ok' }]); + const error = await data.then(() => undefined).catch((e: unknown) => e); + expect(error).toBeInstanceOf(HttpsError); + expect((error as HttpsError).code).toBe('internal'); + expect((error as HttpsError).message).toBe( + 'Failed to encode function output', + ); + }); + + it('propagates an HttpsError raised by the handler verbatim', async () => { + const fn = onCallStreamEffect( + { runtime, inputSchema: Schema.Struct({}) }, + () => Stream.fail(new HttpsError('permission-denied', 'no access')), + ); + + const error = await streamCallable(fn, {}) + .data.then(() => undefined) + .catch((e: unknown) => e); + + expect(error).toBeInstanceOf(HttpsError); + expect((error as HttpsError).code).toBe('permission-denied'); + expect((error as HttpsError).message).toBe('no access'); + }); + + it('does not route a FunctionSetupError raised by the handler through setup recovery', async () => { + const fn = onCallStreamEffect( + { runtime, inputSchema: Schema.Struct({}) }, + () => + Stream.fail( + new FunctionSetupError({ + phase: 'decode-input', + cause: new Error('raised by the handler, not the boundary'), + }), + ), + ); + + const error = await streamCallable(fn, {}) + .data.then(() => undefined) + .catch((e: unknown) => e); + + expect(error).toBeInstanceOf(FunctionSetupError); + expect((error as FunctionSetupError).phase).toBe('decode-input'); + }); +}); + +class QuietError extends Data.TaggedError('QuietError')<{ + readonly reason: string; +}> { + readonly [ErrorReporter.ignore] = true; +} + +class LoudError extends Data.TaggedError('LoudError')<{ + readonly reason: string; +}> {} + +describe('onCallStreamEffect defect logging', () => { + let errorSpy: ReturnType; + + beforeEach(() => { + errorSpy = vi.spyOn(logger, 'error').mockImplementation(() => undefined); + }); + + afterEach(() => { + errorSpy.mockRestore(); + }); + + it('does not log a decode-input failure (now an HttpsError) as a defect', async () => { + const fn = onCallStreamEffect( + { runtime, inputSchema: Schema.Struct({ count: Schema.Number }) }, + (input) => Stream.make(input.count), + ); + + const error = await streamCallable(fn, { count: 'nope' } as never) + .data.then(() => undefined) + .catch((e: unknown) => e); + + expect(error).toBeInstanceOf(HttpsError); + expect(errorSpy).not.toHaveBeenCalled(); + }); + + it('does not log an HttpsError raised by the handler as a defect', async () => { + const fn = onCallStreamEffect( + { runtime, inputSchema: Schema.Struct({}) }, + () => Stream.fail(new HttpsError('not-found', 'gone')), + ); + + const error = await streamCallable(fn, {}) + .data.then(() => undefined) + .catch((e: unknown) => e); + + expect(error).toBeInstanceOf(HttpsError); + expect(errorSpy).not.toHaveBeenCalled(); + }); + + it('does not log an error annotated with ErrorReporter.ignore', async () => { + const fn = onCallStreamEffect( + { runtime, inputSchema: Schema.Struct({}) }, + () => Stream.fail(new QuietError({ reason: 'expected' })), + ); + + const error = await streamCallable(fn, {}) + .data.then(() => undefined) + .catch((e: unknown) => e); + + expect(error).toBeInstanceOf(QuietError); + expect(errorSpy).not.toHaveBeenCalled(); + }); + + it('logs an unannotated handler error as a defect, and rethrows it', async () => { + const fn = onCallStreamEffect( + { runtime, inputSchema: Schema.Struct({}) }, + () => Stream.fail(new LoudError({ reason: 'unexpected' })), + ); + + const error = await streamCallable(fn, {}) + .data.then(() => undefined) + .catch((e: unknown) => e); + + expect(error).toBeInstanceOf(LoudError); + expect(errorSpy).toHaveBeenCalledWith( + 'Defect in onCallStream', + expect.objectContaining({ inner: expect.any(LoudError) }), + ); }); }); diff --git a/packages/admin/src/lib/functions/on-call-stream.ts b/packages/admin/src/lib/functions/on-call-stream.ts index 1376b24..97224a1 100644 --- a/packages/admin/src/lib/functions/on-call-stream.ts +++ b/packages/admin/src/lib/functions/on-call-stream.ts @@ -1,4 +1,4 @@ -import { Effect, pipe, Schema, Stream } from 'effect'; +import { Cause, Effect, Exit, pipe, Schema, Stream } from 'effect'; import { onCall, CallableFunction, @@ -6,13 +6,16 @@ import { CallableRequest, CallableResponse, } from 'firebase-functions/https'; -import { run, Runtime } from './run.js'; +import { runExit, Runtime } from './run.js'; import { logger } from 'firebase-functions'; import { CallableContext, decodeInput, extractContext, } from './on-call-helpers.js'; +import { FunctionSetupError } from './setup-error.js'; +import { isExpectedRejection } from './report.js'; +import { defaultSetupErrorResponse } from './recover-callable-setup-error.js'; interface CallStreamEffectOptions extends CallableOptions { runtime: Runtime; @@ -147,55 +150,84 @@ export function onCallStreamEffect( return onCall(options, async (request, response) => { const context = extractContext(request, response); - const effect = pipe( - // Step 1: Decode input if schema provided - (inputSchema - ? decodeInput(inputSchema)(request) - : Effect.succeed(request)) as Effect.Effect< - unknown, - Schema.SchemaError - >, - - // Step 2: Run handler to obtain the stream, then drain it - Effect.andThen((inputOrRequest) => - pipe( - handler(inputOrRequest, context), - - // Step 3: Encode each chunk if schema provided - Stream.mapEffect( - (chunk): Effect.Effect => - chunkSchema - ? (Schema.encodeUnknownEffect(chunkSchema)( - chunk, - ) as Effect.Effect) - : Effect.succeed(chunk), + // Boundary step 1: decode the input. Its only failure is a setup error, + // recovered below so the handler and stream never run on bad input. + const decoded = inputSchema + ? decodeInput(inputSchema)(request).pipe( + Effect.mapError( + (cause) => new FunctionSetupError({ phase: 'decode-input', cause }), ), + ) + : Effect.succeed(request); - // Step 4: Forward each encoded chunk to the client - Stream.mapEffect((encoded) => - pipe(sendChunk(response, encoded), Effect.as(encoded)), - ), + // Boundary step 2: encode each chunk. Likewise a setup error, recovered + // so a chunk that violates its schema fails the stream with an HttpsError. + const encodeChunk = (chunk: unknown) => + chunkSchema + ? ( + Schema.encodeUnknownEffect(chunkSchema)(chunk) as Effect.Effect< + unknown, + Schema.SchemaError + > + ).pipe( + Effect.mapError( + (cause) => + new FunctionSetupError({ phase: 'encode-output', cause }), + ), + Effect.catch(defaultSetupErrorResponse), + ) + : Effect.succeed(chunk); + + const effect = decoded.pipe( + Effect.matchEffect({ + // Decoding failed, so the handler and stream never run. + onFailure: (error) => defaultSetupErrorResponse(error), + // Recovery is deliberately NOT wrapped around the handler: a handler + // failure is the handler's own error, even when it happens to be a + // FunctionSetupError, and must reach the caller unchanged. + onSuccess: (inputOrRequest) => + pipe( + handler(inputOrRequest, context), + + // Step 3: Encode each chunk if a schema is provided + Stream.mapEffect(encodeChunk), - // Step 5: Interrupt the stream (including an in-flight pull) when - // the client disconnects; chunks collected so far are kept - Stream.interruptWhen(clientDisconnected(response)), + // Step 4: Forward each encoded chunk to the client + Stream.mapEffect((encoded) => + pipe(sendChunk(response, encoded), Effect.as(encoded)), + ), - // Step 6: Collect all chunks as the final result - Stream.runCollect, - ), - ), - ).pipe(Effect.withSpan('onCallStreamEffect')); + // Step 5: Interrupt the stream (including an in-flight pull) when + // the client disconnects; chunks collected so far are kept + Stream.interruptWhen(clientDisconnected(response)), + + // Step 6: Collect all chunks as the final result + Stream.runCollect, + ), + }), + Effect.withSpan('onCallStreamEffect'), + ); - return await run( + const exit = await runExit( options.runtime, - effect as unknown as Effect.Effect, never, R>, - ).catch((error) => { + effect as Effect.Effect, + ); + + if (Exit.isSuccess(exit)) { + return exit.value as ReadonlyArray; + } + + const error = Cause.squash(exit.cause); + // Expected rejections (an HttpsError, or any error annotated with + // ErrorReporter.ignore) are rethrown for Firebase to serialize, so the + // client receives their code and message, but are not logged as defects. + if (!isExpectedRejection(error)) { logger.error('Defect in onCallStream', { inner: error, stack: error instanceof Error ? error.stack : undefined, }); - throw error; - }); + } + throw error; }); } diff --git a/packages/admin/src/lib/functions/on-call.ts b/packages/admin/src/lib/functions/on-call.ts index f938d42..a946cb1 100644 --- a/packages/admin/src/lib/functions/on-call.ts +++ b/packages/admin/src/lib/functions/on-call.ts @@ -17,6 +17,7 @@ import { } from './on-call-helpers.js'; import { FunctionSetupError } from './setup-error.js'; import { isExpectedRejection } from './report.js'; +import { defaultSetupErrorResponse } from './recover-callable-setup-error.js'; interface CallEffectOptions extends CallableOptions { runtime: Runtime; @@ -60,19 +61,6 @@ interface CallEffectOptionsWithBoth< outputSchema: O; } -/** - * Default recovery: reject the call with an HttpsError that reflects the - * setup phase that failed. - */ -const defaultSetupErrorResponse = ( - error: FunctionSetupError, -): Effect.Effect => - Effect.fail( - error.phase === 'decode-input' - ? new HttpsError('invalid-argument', error.cause.message) - : new HttpsError('internal', 'Failed to encode function output'), - ); - /** * Create a Firebase Functions callable trigger that runs an effect. * diff --git a/packages/admin/src/lib/functions/recover-callable-setup-error.ts b/packages/admin/src/lib/functions/recover-callable-setup-error.ts new file mode 100644 index 0000000..d2b5930 --- /dev/null +++ b/packages/admin/src/lib/functions/recover-callable-setup-error.ts @@ -0,0 +1,24 @@ +import { Effect } from 'effect'; +import { HttpsError } from 'firebase-functions/https'; +import { FunctionSetupError } from './setup-error.js'; + +/** + * @internal Not part of the package's public API. + * + * Default recovery for a callable setup error: reject the call with an + * `HttpsError` whose code and message reflect the setup phase that failed. + * + * Shared by the callable wrappers (`onCallEffect`, `onCallStreamEffect`), + * which both reject instead of dying: an input decode failure becomes an + * `invalid-argument` rejection carrying the schema error message, and any + * other setup failure (notably output or chunk encoding) becomes an `internal` + * rejection with a stable `'Failed to encode function output'` message. + */ +export const defaultSetupErrorResponse = ( + error: FunctionSetupError, +): Effect.Effect => + Effect.fail( + error.phase === 'decode-input' + ? new HttpsError('invalid-argument', error.cause.message) + : new HttpsError('internal', 'Failed to encode function output'), + );