fix: handle big payload across the stack (NangoHQ#2474)

## Describe your changes

Fixes
https://linear.app/nango/issue/NAN-1363/big-payload-is-crashing-the-action-execution

A big output from an action would break logging, orchestrator and the UI
(all for different reasons).
The error management being still not good it's messy to handle and
propagate the correct error but I think I managed something not too
ugly.

- Runner: Handle big payload validation 
- Runner: Better handle JSON.stringification when logging and avoid
error by truncating before sending
- Orch: stop stringifying errors
- Orch: return json response in the payload as json
- Orch: handle big payload (> 10mb) preventing task to be marked as
completed
- UI: handle big payload in the message panel


## Test

```ts
import type { NangoSync } from '../../models';

export default async function runAction(nango: NangoSync): Promise<string> {
  return '1'.repeat(1_999_999_999);
}
```

<img width="1173" alt="Screenshot 2024-07-05 at 18 37 41"
src="https://github.com/NangoHQ/nango/assets/1637651/c077c47b-edd7-4f02-ab46-988b8fc04616">

Author: bodinsamuel

package-lock.json

@@ -9,7 +9,7 @@ import { route as getOutputRoute } from '../routes/v1/tasks/taskId/getOutput.js'
 import { route as putTaskRoute } from '../routes/v1/tasks/putTaskId.js';
 import { route as postHeartbeatRoute } from '../routes/v1/tasks/taskId/postHeartbeat.js';
 import type { Result, Route } from '@nangohq/utils';
-import { Ok, Err, routeFetch, stringifyError, getLogger } from '@nangohq/utils';
+import { Ok, Err, routeFetch, getLogger } from '@nangohq/utils';
 import type { Endpoint } from '@nangohq/types';
 import type {
     ClientError,
@@ -54,8 +54,8 @@ export class OrchestratorClient {
         if ('error' in res) {
             return Err({
                 name: res.error.code,
-                message: res.error.message || `Error scheduling  immediate task`,
-                payload: props
+                message: res.error.message || `Error scheduling immediate task`,
+                payload: { ...props, response: res.error.payload as any }
             });
         } else {
             return Ok(res);
@@ -80,7 +80,7 @@ export class OrchestratorClient {
             return Err({
                 name: res.error.code,
                 message: res.error.message || `Error creating recurring schedule`,
-                payload: { ...props, startsAt }
+                payload: { ...props, startsAt, response: res.error.payload as any }
             });
         } else {
             return Ok(res);
@@ -107,7 +107,7 @@ export class OrchestratorClient {
             return Err({
                 name: res.error.code,
                 message: res.error.message || `Error setting schedule state`,
-                payload: { scheduleName, state }
+                payload: { scheduleName, state, response: res.error.payload as any }
             });
         } else {
             return Ok(undefined);
@@ -122,7 +122,7 @@ export class OrchestratorClient {
             return Err({
                 name: res.error.code,
                 message: res.error.message || `Error updating schedule frequency`,
-                payload: { scheduleName, frequencyMs }
+                payload: { scheduleName, frequencyMs, response: res.error.payload as any }
             });
         } else {
             return Ok(undefined);
@@ -139,7 +139,7 @@ export class OrchestratorClient {
             return Err({
                 name: res.error.code,
                 message: res.error.message || `Error creating recurring schedule`,
-                payload: props
+                payload: { ...props, response: res.error.payload as any }
             });
         } else {
             return Ok(undefined);
@@ -162,7 +162,7 @@ export class OrchestratorClient {
             return Err({
                 name: getOutput.error.code,
                 message: getOutput.error.message || `Error fetching task '${taskId}' output`,
-                payload: {}
+                payload: { response: getOutput.error.payload as any }
             });
         } else {
             switch (getOutput.state) {
@@ -261,7 +261,7 @@ export class OrchestratorClient {
             return Err({
                 name: res.error.code,
                 message: res.error.message || `Error listing tasks`,
-                payload: body
+                payload: { ...body, response: res.error.payload as any }
             });
         } else {
             const tasks = res.flatMap((task) => {
@@ -284,7 +284,7 @@ export class OrchestratorClient {
             return Err({
                 name: res.error.code,
                 message: res.error.message || `Error listing schedules`,
-                payload: { scheduleNames }
+                payload: { scheduleNames, response: res.error.payload as any }
             });
         } else {
             const schedules = res.flatMap((schedule) => {
@@ -319,7 +319,7 @@ export class OrchestratorClient {
             return Err({
                 name: res.error.code,
                 message: res.error.message || `Error dequeueing tasks`,
-                payload: { groupKey, limit }
+                payload: { groupKey, limit, response: res.error.payload as any }
             });
         } else {
             const dequeuedTasks = res.flatMap((task) => {
@@ -342,7 +342,7 @@ export class OrchestratorClient {
             return Err({
                 name: res.error.code,
                 message: res.error.message || `Error heartbeating task '${taskId}'`,
-                payload: { taskId }
+                payload: { taskId, response: res.error.payload as any }
             });
         } else {
             return Ok(undefined);
@@ -358,21 +358,21 @@ export class OrchestratorClient {
             return Err({
                 name: res.error.code,
                 message: res.error.message || `Error succeeding task '${taskId}'`,
-                payload: { taskId, output }
+                payload: { taskId, output, response: res.error.payload as any }
             });
         } else {
             return validateTask(res).mapError((err) => ({
                 name: 'succeed_failed',
-                message: `Failed to mark task ${taskId} as succeeded: ${stringifyError(err)}`,
-                payload: { taskId, output }
+                message: `Failed to mark task ${taskId} as succeeded`,
+                payload: { taskId, output, err: err as any }
             }));
         }
     }
 
     public async failed({ taskId, error }: { taskId: string; error: Error }): Promise<Result<OrchestratorTask, ClientError>> {
         const output = {
             name: error.name,
-            type: 'type' in error ? (error.type as string) : null,
+            type: 'type' in error ? (error.type as string) : 'unknown_error',
             message: error.message,
             payload: 'payload' in error ? (error.payload as any) : null
         };
@@ -384,13 +384,13 @@ export class OrchestratorClient {
             return Err({
                 name: res.error.code,
                 message: res.error.message || `Error failing task '${taskId}'`,
-                payload: { taskId, error: output }
+                payload: { taskId, error: output, response: res.error.payload as any }
             });
         } else {
             return validateTask(res).mapError((err) => ({
                 name: 'failed_failed',
-                message: `Failed to mark task ${taskId} as failed: ${stringifyError(err)}`,
-                payload: { taskId, error: output }
+                message: `Failed to mark task ${taskId} as failed`,
+                payload: { taskId, error: output, err: err as any }
             }));
         }
     }
@@ -404,13 +404,13 @@ export class OrchestratorClient {
             return Err({
                 name: res.error.code,
                 message: res.error.message || `Error cancelling task '${taskId}'`,
-                payload: { taskId, error: reason }
+                payload: { taskId, error: reason, response: res.error.payload as any }
             });
         } else {
             return validateTask(res).mapError((err) => ({
-                name: 'cacel_failed',
-                message: `Failed to mark task ${taskId} as cancelled: ${stringifyError(err)}`,
-                payload: { taskId, error: reason }
+                name: 'cancel_failed',
+                message: `Failed to mark task ${taskId} as cancelled`,
+                payload: { taskId, error: reason, err: err as any }
             }));
         }
     }

packages/orchestrator/lib/clients/client.ts

@@ -1,10 +1,11 @@
 import type { Result } from '@nangohq/utils';
 import { Err, stringifyError, getLogger } from '@nangohq/utils';
 import type { OrchestratorClient } from './client.js';
-import type { OrchestratorTask } from './types.js';
+import type { ClientError, OrchestratorTask } from './types.js';
 import type { JsonValue } from 'type-fest';
 import PQueue from 'p-queue';
 import type { Tracer } from 'dd-trace';
+import type { ApiError } from '@nangohq/types';
 
 const logger = getLogger('orchestrator.clients.processor');
 
@@ -121,26 +122,29 @@ export class OrchestratorProcessor {
                         // task was aborted. No need to set it as failed
                         return;
                     }
+
                     const setFailed = await this.orchestratorClient.failed({ taskId: task.id, error: res.error });
                     if (setFailed.isErr()) {
-                        logger.error(`failed to set task ${task.id} as failed: ${stringifyError(setFailed.error)}`);
+                        throwIfPayloadTooBig(setFailed.error);
+                        logger.error(`failed to set task ${task.id} as failed`, setFailed.error);
                         span.setTag('error', setFailed);
                     } else {
                         span.setTag('error', res.error);
                     }
                 } else {
                     const setSucceed = await this.orchestratorClient.succeed({ taskId: task.id, output: res.value });
                     if (setSucceed.isErr()) {
-                        logger.error(`failed to set task ${task.id} as succeeded: ${stringifyError(setSucceed.error)}`);
+                        throwIfPayloadTooBig(setSucceed.error);
+                        logger.error(`failed to set task ${task.id} as succeeded`, setSucceed.error);
                         span.setTag('error', setSucceed);
                     }
                 }
             } catch (err: unknown) {
-                const error = new Error(stringifyError(err));
-                logger.error(`Failed to process task ${task.id}: ${stringifyError(error)}`);
+                const error = err instanceof Error ? err : new Error(stringifyError(err));
+                logger.error(`Failed to process task ${task.id}`, error);
                 const setFailed = await this.orchestratorClient.failed({ taskId: task.id, error });
                 if (setFailed.isErr()) {
-                    logger.error(`failed to set task ${task.id} as failed. Unknown error: ${stringifyError(setFailed.error)}`);
+                    logger.error(`failed to set task ${task.id} as failed. Unknown error`, setFailed.error);
                     span.setTag('error', setFailed);
                 } else {
                     span.setTag('error', error);
@@ -157,8 +161,23 @@ export class OrchestratorProcessor {
         return setInterval(async () => {
             const res = await this.orchestratorClient.heartbeat({ taskId: task.id });
             if (res.isErr()) {
-                logger.error(`failed to send heartbeat for task ${task.id}: ${stringifyError(res.error)}`);
+                logger.error(`failed to send heartbeat for task ${task.id}`, res.error);
             }
         }, 300_000);
     }
 }
+
+// We don't have access to NangoError so we have to create a temp error
+class PayloadTooBigError extends Error {
+    type = 'action_output_too_big';
+    override message = 'Output is too big';
+}
+
+function throwIfPayloadTooBig(err: ClientError) {
+    if (err.payload && typeof err.payload === 'object' && 'response' in err.payload && err.payload['response'] && typeof err.payload['response'] === 'object') {
+        const res = err.payload['response'] as unknown as ApiError<string>;
+        if (res.error.code === 'payload_too_big') {
+            throw new PayloadTooBigError();
+        }
+    }
+}

packages/orchestrator/lib/clients/processor.ts

@@ -42,16 +42,17 @@ export const getServer = (scheduler: Scheduler, eventEmmiter: EventEmitter): Exp
     createRoute(server, postHeartbeatHandler(scheduler));
     createRoute(server, postDequeueHandler(scheduler, eventEmmiter));
 
-    server.use((err: unknown, _req: Request, res: Response<ApiError<'server_error', any>>, next: NextFunction) => {
-        res.status(500).json({ error: { code: 'server_error', errors: err } });
-        next();
-    });
-
-    server.use((err: any, _req: Request, res: Response<ApiError<'invalid_json' | 'internal_error'>>, _next: any) => {
+    server.use((err: any, _req: Request, res: Response<ApiError<'invalid_json' | 'internal_error' | 'payload_too_big'>>, _next: NextFunction) => {
         if (err instanceof SyntaxError && 'body' in err && 'type' in err && err.type === 'entity.parse.failed') {
             res.status(400).send({ error: { code: 'invalid_json', message: err.message } });
             return;
+        } else if (err instanceof Error) {
+            if (err.message === 'request entity too large') {
+                res.status(400).json({ error: { code: 'payload_too_big', message: 'Payload is too big' } });
+                return;
+            }
         }
+
         res.status(500).send({ error: { code: 'internal_error', message: err.message } });
     });
 

packages/orchestrator/lib/server.ts

@@ -5,14 +5,15 @@ import paginateService from '../services/paginate.service.js';
 import proxyService from '../services/proxy.service.js';
 import type { AxiosInstance } from 'axios';
 import axios, { AxiosError } from 'axios';
-import { getPersistAPIUrl, safeStringify } from '../utils/utils.js';
+import { getPersistAPIUrl } from '../utils/utils.js';
 import type { IntegrationWithCreds } from '@nangohq/node';
 import type { UserProvidedProxyConfiguration } from '../models/Proxy.js';
 import { getLogger, httpRetryStrategy, metrics, retryWithBackoff } from '@nangohq/utils';
 import type { SyncConfig } from '../models/Sync.js';
 import type { RunnerFlags } from '../services/sync/run.utils.js';
 import { validateData } from './dataValidation.js';
 import { NangoError } from '../utils/error.js';
+import { stringifyAndTruncateLog } from './utils.js';
 
 const logger = getLogger('SDK');
 
@@ -657,7 +658,8 @@ export class NangoAction {
             return;
         }
 
-        const content = safeStringify(args);
+        const content = stringifyAndTruncateLog(args, 99_000);
+
         await this.sendLogToPersist(content, { level, timestamp: Date.now() });
     }
 

packages/shared/lib/sdk/sync.ts

@@ -0,0 +1,19 @@
+import safeStringify from 'fast-safe-stringify';
+
+export function stringifyAndTruncateLog(args: any[], maxSize: number = 99_000) {
+    let msg = '';
+
+    if (typeof args[0] === 'string') {
+        msg = args.shift();
+    }
+
+    if (args.length > 0) {
+        msg += ` ${args.map((arg) => safeStringify.stableStringify(arg, undefined, undefined, { depthLimit: 10, edgesLimit: 20 }))}`;
+    }
+
+    if (msg.length > maxSize) {
+        msg = `${msg.substring(0, maxSize)}... (truncated)`;
+    }
+
+    return msg;
+}

packages/shared/lib/sdk/utils.ts

@@ -0,0 +1,33 @@
+import { describe, expect, it } from 'vitest';
+import { stringifyAndTruncateLog } from './utils.js';
+
+describe('stringifyAndTruncateLog', () => {
+    it('should not break a small string', () => {
+        const str = stringifyAndTruncateLog(['hello']);
+        expect(str).toStrictEqual('hello');
+    });
+
+    it('should limit a string', () => {
+        const str = stringifyAndTruncateLog(['hello'], 2);
+        expect(str).toStrictEqual('he... (truncated)');
+    });
+
+    it('should limit an object', () => {
+        const str = stringifyAndTruncateLog(['hello', { foo: 'bar' }], 10);
+        expect(str).toStrictEqual('hello {"fo... (truncated)');
+    });
+
+    it('should not break empty args', () => {
+        const str = stringifyAndTruncateLog([]);
+        expect(str).toStrictEqual('');
+    });
+
+    it('should handle object', () => {
+        const str = stringifyAndTruncateLog([{ foo: 1 }]);
+        expect(str).toStrictEqual(' {"foo":1}');
+    });
+    it('should handle object + string', () => {
+        const str = stringifyAndTruncateLog(['hello', { foo: 1 }]);
+        expect(str).toStrictEqual('hello {"foo":1}');
+    });
+});

packages/shared/lib/sdk/utils.unit.test.ts

@@ -112,7 +112,7 @@ export class NangoError extends Error {
                 this.message = 'The API endpoint could not be found and returned a 404. Please ensure you have the endpoint specified and spelled correctly.';
                 break;
 
-            case 'fobidden':
+            case 'forbidden':
                 this.status = 403;
                 this.message = 'The API endpoint returned back a 403 error. Check the scopes requested to make sure proper access is requested to the API.';
                 break;
@@ -603,6 +603,11 @@ export class NangoError extends Error {
                 this.message = 'Failed to validate a record in batchSave';
                 break;
 
+            case 'action_output_too_big':
+                this.status = 400;
+                this.message = 'Action output is too big';
+                break;
+
             default:
                 this.status = 500;
                 this.type = 'unhandled_' + type;