Release: worker@1.29.2 (cli@1.39.5)

packages/cli/CHANGELOG.md

@@ -1,5 +1,12 @@
 # @openfn/cli
 
+## 1.39.5
+
+### Patch Changes
+
+- Updated dependencies [7398d41]
+  - @openfn/lexicon@2.4.2
+
 ## 1.39.4
 
 ### Patch Changes

packages/cli/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openfn/cli",
-  "version": "1.39.4",
+  "version": "1.39.5",
   "description": "CLI devtools for the OpenFn toolchain",
   "engines": {
     "node": ">=18",

packages/engine-multi/CHANGELOG.md

@@ -1,5 +1,13 @@
 # engine-multi
 
+## 1.13.1
+
+### Patch Changes
+
+- 7398d41: Report the size of a job's output state alongside the existing payload redaction check
+- Updated dependencies [7398d41]
+  - @openfn/lexicon@2.4.2
+
 ## 1.13.0
 
 ### Minor Changes

packages/engine-multi/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openfn/engine-multi",
-  "version": "1.13.0",
+  "version": "1.13.1",
   "description": "Multi-process runtime engine",
   "main": "dist/index.js",
   "type": "module",

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

@@ -83,7 +83,16 @@ export const jobComplete = (
   event: internalEvents.JobCompleteEvent
 ) => {
   const { logger, state: runState } = context;
-  const { threadId, state, duration, jobId, next, mem, redacted } = event;
+  const {
+    threadId,
+    state,
+    duration,
+    jobId,
+    next,
+    mem,
+    redacted,
+    payloadSize_b,
+  } = event;
   logger.debug(
     `${runState.id}: sending job complete (step complete): ${event.jobId}`
   );
@@ -95,6 +104,7 @@ export const jobComplete = (
     jobId,
     next,
     redacted,
+    payloadSize_b,
     mem,
     time: timestamp(),
   });

packages/engine-multi/src/events.ts

@@ -51,9 +51,12 @@ export type EventMap = {
 
 export type ExternalEvents = keyof EventMap;
 
-interface ExternalEvent {
+export interface ExternalEvent {
   threadId?: string;
   workflowId: UUID;
+  // Byte size of any large fields on this payload
+  // eg, dataclips, state objects, log objects
+  payloadSize_b?: number;
 }
 
 export interface WorkflowStartPayload extends ExternalEvent {

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

@@ -1,4 +1,5 @@
 import { JsonStreamStringify } from 'json-stream-stringify';
+import type { ExternalEvent } from '../events';
 
 // This specifies which keys of an event payload to potentially redact
 // if they are too big
@@ -17,7 +18,7 @@ export const verify = async (
   value: any,
   limit_mb: number = 10,
   algo: 'stringify' | 'stream' = 'stringify'
-) => {
+): Promise<number | undefined> => {
   if (value && !isNaN(limit_mb)) {
     const limitBytes = limit_mb * 1024 * 1024;
 
@@ -33,9 +34,15 @@ export const verify = async (
       // @ts-ignore
       e.name = 'PAYLOAD_TOO_LARGE';
       e.message = `The payload exceeded the size limit of ${limit_mb}mb`;
+      // @ts-ignore carry the size we already computed out to the caller
+      e.sizeBytes = sizeBytes;
       throw e;
     }
+
+    return sizeBytes;
   }
+
+  return undefined;
 };
 
 export const calculateSizeStringify = (value: any): number => {
@@ -65,15 +72,25 @@ export const calculateSizeStream = async (
   return size_bytes;
 };
 
-export default async (payload: any, limit_mb: number = 10) => {
-  const newPayload = { ...payload };
+export default async (
+  payload: ExternalEvent,
+  limit_mb: number = 10
+): Promise<ExternalEvent> => {
+  const newPayload: any = { ...payload };
+  const rawPayload = payload as any;
 
   for (const key of KEYS_TO_VERIFY) {
     try {
-      await verify(payload[key], limit_mb);
-    } catch (e) {
+      const sizeBytes = await verify(rawPayload[key], limit_mb);
+      if (key === 'state' && sizeBytes !== undefined) {
+        newPayload.payloadSize_b = sizeBytes;
+      }
+    } catch (e: any) {
       Object.assign(newPayload[key], replacements[key] ?? replacements.default);
       newPayload.redacted = true;
+      if (key === 'state') {
+        newPayload.payloadSize_b = e.sizeBytes;
+      }
     }
   }
 

packages/engine-multi/src/worker/events.ts

@@ -39,6 +39,10 @@ interface InternalEvent {
   type: WorkerEvents;
   workflowId: UUID;
   threadId: string;
+  // Byte size of whichever field this payload had checked against the
+  // redaction limit (state, final_state or log - see KEYS_TO_VERIFY in
+  // ensure-payload-size.ts)
+  payloadSize_b?: number;
 }
 
 export interface WorkflowStartEvent extends InternalEvent {}

packages/engine-multi/src/worker/thread/runtime.ts

@@ -55,7 +55,8 @@ export const publish = async (
   // Redact any payloads that are too large
   const limit =
     payloadLimits?.[type as keyof PayloadLimits] ?? payloadLimits?.default;
-  const safePayload = await ensurePayloadSize(payload, limit);
+
+  const safePayload = await ensurePayloadSize(payload as any, limit);
 
   parentPort!.postMessage({
     type,

packages/engine-multi/test/api/lifecycle.test.ts

@@ -194,6 +194,43 @@ test(`job-complete: emits ${e.JOB_COMPLETE} with key fields`, (t) => {
   });
 });
 
+test(`job-complete: forwards payloadSize_b`, (t) => {
+  return new Promise((done) => {
+    const workflowId = 'a';
+
+    const state = {
+      id: workflowId,
+      startTime: Date.now() - 1000,
+    } as WorkflowState;
+
+    const context = createContext(workflowId, state);
+
+    const event: w.JobCompleteEvent = {
+      type: w.JOB_COMPLETE,
+      workflowId,
+      threadId: '1',
+      jobId: 'j',
+      duration: 200,
+      state: 22,
+      redacted: true,
+      payloadSize_b: 12345,
+      next: [],
+      mem: { job: 100, system: 1000 },
+    };
+
+    context.on(e.JOB_COMPLETE, (evt) => {
+      // This is the number that lets a diagnostic downstream (eg the
+      // lightning worker's sentry reporting) see how big the state was even
+      // when it never tripped the redaction limit
+      t.is(evt.payloadSize_b, 12345);
+      t.true(evt.redacted);
+      done();
+    });
+
+    jobComplete(context, event);
+  });
+});
+
 test(`job-error: emits ${e.JOB_ERROR} with key fields`, (t) => {
   return new Promise((done) => {
     const workflowId = 'a';

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

@@ -5,6 +5,13 @@ import ensurePayloadSize, {
   calculateSizeStream,
 } from '../../src/util/ensure-payload-size';
 
+// ensurePayloadSize now takes/returns the full ExternalEvent envelope, but
+// these tests exercise its redaction/sizing behaviour in isolation against
+// bare state/log/final_state fixtures, not real events - so untyped is right
+// here rather than padding every fixture with a fake workflowId
+const check = (payload: any, limit?: number): Promise<any> =>
+  ensurePayloadSize(payload, limit);
+
 (['stringify', 'stream'] as const).forEach((algo) => {
   test(algo + ': throw limit 0, payload 1 byte', async (t) => {
     await t.throwsAsync(() => verify('x', 0, algo), {
@@ -60,7 +67,7 @@ import ensurePayloadSize, {
       },
     };
 
-    const newPayload = await ensurePayloadSize(payload, 1);
+    const newPayload = await check(payload, 1);
     t.deepEqual(newPayload.state, {
       data: '[REDACTED]',
     });
@@ -74,7 +81,7 @@ import ensurePayloadSize, {
       },
     };
 
-    const newPayload = await ensurePayloadSize(payload, 1);
+    const newPayload = await check(payload, 1);
     t.deepEqual(newPayload.log, {
       message: ['[REDACTED: Message length exceeds payload limit]'],
     });
@@ -88,12 +95,48 @@ import ensurePayloadSize, {
       },
     };
 
-    const newPayload = await ensurePayloadSize(payload, 1);
+    const newPayload = await check(payload, 1);
     t.deepEqual(newPayload.final_state, {
       data: '[REDACTED]',
     });
     t.true(newPayload.redacted);
   });
+
+  test(algo + ': attaches payloadSize_b when state is within limit', async (t) => {
+    const payload = { state: { data: 'hello world' } };
+
+    const newPayload = await check(payload, 1);
+    t.false(!!newPayload.redacted);
+    t.is(
+      newPayload.payloadSize_b,
+      calculateSizeStringify(payload.state)
+    );
+  });
+
+  test(algo + ': attaches payloadSize_b when state is redacted', async (t) => {
+    const payload = {
+      state: {
+        data: new Array(1024 * 1024).fill('z').join(''),
+      },
+    };
+    const rawSize = calculateSizeStringify(payload.state);
+
+    const newPayload = await check(payload, 1);
+    t.true(newPayload.redacted);
+    // The size must survive redaction - this is the number that explains a
+    // run which timed out sending its dataclip without tripping this limit
+    t.is(newPayload.payloadSize_b, rawSize);
+  });
+
+  test(algo + ': does not attach payloadSize_b for final_state or log', async (t) => {
+    const payload = {
+      final_state: { data: 'hello world' },
+      log: { message: ['hello world'] },
+    };
+
+    const newPayload = await check(payload, 1);
+    t.is(newPayload.payloadSize_b, undefined);
+  });
 });
 
 test('size estimation: null value', async (t) => {

packages/lexicon/CHANGELOG.md

@@ -1,5 +1,11 @@
 # lexicon
 
+## 2.4.2
+
+### Patch Changes
+
+- 7398d41: Support dataclipSize
+
 ## 2.4.1
 
 ### Patch Changes

packages/lexicon/lightning.d.ts

@@ -228,6 +228,7 @@ export type StepCompletePayload = ExitReason & {
   run_id?: string;
   job_id: string;
   step_id: string;
+  dataclip_size_mb?: string;
   output_dataclip?: string;
   output_dataclip_id?: string;
   output_dataclip_error?: 'DATACLIP_TOO_LARGE';

packages/lexicon/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openfn/lexicon",
-  "version": "2.4.1",
+  "version": "2.4.2",
   "description": "Central repo of names and type definitions",
   "author": "Open Function Group <admin@openfn.org>",
   "license": "ISC",

packages/lightning-mock/CHANGELOG.md

@@ -1,5 +1,14 @@
 # @openfn/lightning-mock
 
+## 2.4.28
+
+### Patch Changes
+
+- Updated dependencies [7398d41]
+- Updated dependencies [7398d41]
+  - @openfn/lexicon@2.4.2
+  - @openfn/engine-multi@1.13.1
+
 ## 2.4.27
 
 ### Patch Changes

packages/lightning-mock/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openfn/lightning-mock",
-  "version": "2.4.27",
+  "version": "2.4.28",
   "private": true,
   "description": "A mock Lightning server",
   "main": "dist/index.js",

packages/ws-worker/CHANGELOG.md

@@ -1,5 +1,18 @@
 # ws-worker
 
+## 1.29.2
+
+### Patch Changes
+
+- 7398d41: Include dataclip size in sentry reports
+- 7398d41: Attribute Sentry reports to the run that produced them
+- c433bcf: Capture more diagnostic detail when the connection to Lightning drops unexpectedly
+- c433bcf: Reduce dataclip bloat when sending step results to Lightning
+- Updated dependencies [7398d41]
+- Updated dependencies [7398d41]
+  - @openfn/lexicon@2.4.2
+  - @openfn/engine-multi@1.13.1
+
 ## 1.29.1
 
 ### Patch Changes

packages/ws-worker/README.md

@@ -51,6 +51,12 @@ Use `-l mock` to connect to a lightning mock server (on the default port).
 
 For a list of supported worker and engine options, see src/start.ts
 
+### Sending output dataclips without double-encoding
+
+By default, the worker JSON-stringifies each step's output dataclip before sending it to Lightning, and Lightning's own transport then re-encodes the whole envelope — this double-encoding bloats large dataclips on the wire. Pass `--no-stringify-state` or set `WORKER_NO_STRINGIFY_STATE` to send the dataclip as a native JSON value instead, avoiding that bloat.
+
+This is only compatible with Lightning 2.19 or later — do not enable it against older Lightning versions.
+
 ## Enforcing memory limits with cgroups
 
 Each run's memory limit is enforced by default through node's max-old-space-size, which only constrains heap size. Native and buffer allocations bypass this limit. This can cause the worker to consume more memory than it is technically allowed, which can in turn cause the whole worker process to be killed by its container (ie, kubernetes).

packages/ws-worker/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openfn/ws-worker",
-  "version": "1.29.1",
+  "version": "1.29.2",
   "description": "A Websocket Worker to connect Lightning to a Runtime Engine",
   "main": "dist/index.js",
   "type": "module",