Release/worker (#1160)

* Log limits (#1156)

* engine: support payload limits per event

* add log limit tests

* tests

* type

* changeset

* formatting

* update changeset

* remove rubbish

* Engine: try to make calculating payload sizes more efficient (#1155)

* scaffolding: make ensure-payload-size async and add slots for two algorithms

* first swing at async traversal

This actually increases memory overhead

* new traversal algorithm with queue, rather than recuion

* add benchmarking util

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

* tidying

* optimisations

Still doesn't get close

* add streaming algo

* update benchmark

* remove traversal algorithm

* tidy

* tidy up

* tidy docs

* remove ignore

* remove old traverse stats

* typing

* restore await

* add a synchronous publish option

important  for process exit events

* revert debug code

---------

Co-authored-by: Claude <noreply@anthropic.com>

* Worker: support optional batching of log events (#1162)

* worker: re-write core event handler to allow more control of sequencing

* tidy tests and finish

* types

* add basic tests

* more testing

* Add batching and tests

* more tests

* options and more tests

* types

* copy new log stuff from other branch

* updates for live testing against lightning

Works great against new and legacy app, with or without batching

* options for interval and limit

* fix test

* copy across old test

* fix type

* add batch log test

* upate default batch size

* changeset

* run integration tests in batch mode

* update tests

* remove logging

* docs

* versions: worker@1.20.

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: josephjclark

integration-tests/worker/src/init.ts

@@ -47,6 +47,7 @@ export const initWorker = async (
     secret: crypto.randomUUID(),
     collectionsVersion: '1.0.0',
     messageTimeoutSeconds: 0.01,
+    batchLogs: true,
     ...workerArgs,
   });
 

integration-tests/worker/test/integration.test.ts

@@ -988,9 +988,11 @@ test.serial('Redact logs which exceed the payload limit', (t) => {
     };
 
     lightning.on('run:log', (evt) => {
-      if (evt.payload.source === 'JOB') {
-        t.regex(evt.payload.message[0], /redacted/i);
-      }
+      evt.payload.logs.forEach((log) => {
+        if (log.source === 'JOB') {
+          t.regex(log.message[0], /redacted/i);
+        }
+      });
     });
 
     lightning.enqueueRun(run);
@@ -1066,11 +1068,13 @@ test.serial(
       const rtLogs = [];
 
       lightning.on('run:log', (e) => {
-        if (e.payload.source === 'JOB') {
-          jobLogs.push(e.payload);
-        } else if (e.payload.source === 'R/T') {
-          rtLogs.push(e.payload);
-        }
+        e.payload.logs.forEach((log) => {
+          if (log.source === 'JOB') {
+            jobLogs.push(log);
+          } else if (log.source === 'R/T') {
+            rtLogs.push(log);
+          }
+        });
       });
 
       lightning.once('run:complete', () => {

packages/engine-multi/CHANGELOG.md

@@ -1,5 +1,16 @@
 # engine-multi
 
+## 1.9.0
+
+### Minor Changes
+
+- 5417e97: Accept log-payload-limit-mb (defaults to 1mb)
+
+### Patch Changes
+
+- Updated dependencies [5417e97]
+  - @openfn/lexicon@1.3.0
+
 ## 1.8.5
 
 ### Patch Changes

packages/engine-multi/memtest.js

@@ -1,6 +1,6 @@
 {
   "name": "@openfn/engine-multi",
-  "version": "1.8.5",
+  "version": "1.9.0",
   "description": "Multi-process runtime engine",
   "main": "dist/index.js",
   "type": "module",
@@ -20,7 +20,8 @@
     "@openfn/lexicon": "workspace:^",
     "@openfn/logger": "workspace:*",
     "@openfn/runtime": "workspace:*",
-    "fast-safe-stringify": "^2.1.1"
+    "fast-safe-stringify": "^2.1.1",
+    "json-stream-stringify": "^3.1.6"
   },
   "devDependencies": {
     "@types/node": "^18.19.127",

packages/engine-multi/package.json

@@ -16,6 +16,7 @@ import preloadCredentials from './preload-credentials';
 import { ExecutionError } from '../errors';
 import type { RunOptions } from '../worker/thread/run';
 import { COMPILE_COMPLETE, COMPILE_START, ExternalEvents } from '../events';
+import type { PayloadLimits } from '../worker/thread/runtime';
 
 const execute = async (context: ExecutionContext) => {
   const { state, callWorker, logger, options } = context;
@@ -47,9 +48,18 @@ const execute = async (context: ExecutionContext) => {
       profilePollInteval: context.options.profilePollInterval,
     } as RunOptions;
 
+    // Construct the payload limits object
+    const payloadLimits: PayloadLimits = {};
+    if (options.payloadLimitMb !== undefined) {
+      payloadLimits.default = options.payloadLimitMb;
+    }
+    if (options.logPayloadLimitMb !== undefined) {
+      payloadLimits[workerEvents.LOG] = options.logPayloadLimitMb;
+    }
+
     const workerOptions = {
       memoryLimitMb: options.memoryLimitMb,
-      payloadLimitMb: options.payloadLimitMb,
+      payloadLimits,
       timeout: options.runTimeoutMs,
     };
 

packages/engine-multi/src/api/execute.ts

@@ -74,6 +74,7 @@ export type EngineOptions = {
   maxWorkers?: number;
   memoryLimitMb?: number;
   payloadLimitMb?: number;
+  logPayloadLimitMb?: number;
   repoDir: string;
   resolvers?: LazyResolvers;
   runtimelogger?: Logger;
@@ -109,6 +110,7 @@ const createEngine = async (
   const defaultMemoryLimit = options.memoryLimitMb || DEFAULT_MEMORY_LIMIT_MB;
   const defaultPayloadLimit =
     options.payloadLimitMb || DEFAULT_PAYLOAD_LIMIT_MB;
+  const defaultLogPayloadLimit = options.logPayloadLimitMb;
   const defaultProfile = options.profile ?? DEFAULT_PROFILE;
   const defaultProfilePollInterval =
     options.profilePollInterval ?? DEFAULT_PROFILE_POLL_INTERVAL;
@@ -170,6 +172,7 @@ const createEngine = async (
         runTimeoutMs: opts.runTimeoutMs ?? defaultTimeout,
         memoryLimitMb: opts.memoryLimitMb ?? defaultMemoryLimit,
         payloadLimitMb: opts.payloadLimitMb ?? defaultPayloadLimit,
+        logPayloadLimitMb: opts.logPayloadLimitMb ?? defaultLogPayloadLimit,
         jobLogLevel: opts.jobLogLevel,
         profile: defaultProfile,
         profilePollInterval: defaultProfilePollInterval,

packages/engine-multi/src/engine.ts

@@ -46,6 +46,7 @@ export type ExecutionContextConstructor = {
 
 export type ExecuteOptions = {
   payloadLimitMb?: number;
+  logPayloadLimitMb?: number;
   memoryLimitMb?: number;
   resolvers?: LazyResolvers;
   runTimeoutMs?: number;

packages/engine-multi/src/types.ts

@@ -1,3 +1,5 @@
+import { JsonStreamStringify } from 'json-stream-stringify';
+
 // This specifies which keys of an event payload to potentially redact
 // if they are too big
 const KEYS_TO_VERIFY = ['state', 'final_state', 'log'];
@@ -11,18 +13,22 @@ const replacements: Record<string, any> = {
   },
 };
 
-export const verify = (value: any, limit_mb: number = 10) => {
+export const verify = async (
+  value: any,
+  limit_mb: number = 10,
+  algo: 'stringify' | 'stream' = 'stringify'
+) => {
   if (value && !isNaN(limit_mb)) {
-    let size_mb = 0;
-    try {
-      const str = typeof value === 'string' ? value : JSON.stringify(value);
-      const size_bytes = Buffer.byteLength(str, 'utf8');
-      size_mb = size_bytes / 1024 / 1024;
-    } catch (e) {
-      // do nothing
+    const limitBytes = limit_mb * 1024 * 1024;
+
+    let sizeBytes: number;
+    if (algo === 'stream') {
+      sizeBytes = await calculateSizeStream(value, limitBytes);
+    } else {
+      sizeBytes = calculateSizeStringify(value);
     }
 
-    if (size_mb > limit_mb) {
+    if (sizeBytes > limitBytes) {
       const e = new Error();
       // @ts-ignore
       e.name = 'PAYLOAD_TOO_LARGE';
@@ -32,12 +38,39 @@ export const verify = (value: any, limit_mb: number = 10) => {
   }
 };
 
-export default (payload: any, limit_mb: number = 10) => {
+export const calculateSizeStringify = (value: any): number => {
+  const str = typeof value === 'string' ? value : JSON.stringify(value);
+  const size_bytes = Buffer.byteLength(str, 'utf8');
+  return size_bytes;
+};
+
+export const calculateSizeStream = async (
+  value: any,
+  limit?: number
+): Promise<number> => {
+  let size_bytes = 0;
+
+  const stream = new JsonStreamStringify(value);
+
+  for await (const chunk of stream) {
+    // Each chunk is a string token from the JSON output
+    size_bytes += Buffer.byteLength(chunk, 'utf8');
+
+    if (limit !== undefined && size_bytes > limit) {
+      break;
+    }
+  }
+  stream.destroy();
+
+  return size_bytes;
+};
+
+export default async (payload: any, limit_mb: number = 10) => {
   const newPayload = { ...payload };
 
   for (const key of KEYS_TO_VERIFY) {
     try {
-      verify(payload[key], limit_mb);
+      await verify(payload[key], limit_mb);
     } catch (e) {
       Object.assign(newPayload[key], replacements[key] ?? replacements.default);
       newPayload.redacted = true;

packages/engine-multi/src/util/ensure-payload-size.ts

@@ -12,6 +12,7 @@ import {
 } from './events';
 import { HANDLED_EXIT_CODE } from '../events';
 import { Logger } from '@openfn/logger';
+import type { PayloadLimits } from './thread/runtime';
 
 export type PoolOptions = {
   capacity?: number; // defaults to 5
@@ -34,7 +35,7 @@ export type ExecOpts = {
   timeout?: number; // ms
 
   memoryLimitMb?: number;
-  payloadLimitMb?: number;
+  payloadLimits?: PayloadLimits;
 };
 
 export type ChildProcessPool = Array<ChildProcess | false>;
@@ -212,7 +213,7 @@ function createPool(script: string, options: PoolOptions = {}, logger: Logger) {
           args,
           options: {
             memoryLimitMb: opts.memoryLimitMb,
-            payloadLimitMb: opts.payloadLimitMb,
+            payloadLimits: opts.payloadLimits,
           },
         } as RunTaskEvent);
       } catch (e) {