Worker: memory management improvements

.changeset/humble-ravens-send.md

@@ -0,0 +1,5 @@
+---
+'@openfn/logger': patch
+---
+
+Fix an issue where logger.proxy can throw for nil values

.changeset/odd-sloths-make.md

@@ -0,0 +1,6 @@
+---
+'@openfn/engine-multi': patch
+'@openfn/ws-worker': patch
+---
+
+Use streaming algorithm to calculate payload size

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

@@ -48,6 +48,16 @@ export const calculateSizeStream = async (
   value: any,
   limit?: number
 ): Promise<number> => {
+  // skip all primitives
+  if (
+    !value ||
+    typeof value === 'number' ||
+    typeof value === 'boolean' ||
+    typeof value === 'function'
+  ) {
+    return 1;
+  }
+
   let size_bytes = 0;
 
   const stream = new JsonStreamStringify(value);
@@ -69,11 +79,16 @@ export default async (payload: any, limit_mb: number = 10) => {
   const newPayload = { ...payload };
 
   for (const key of KEYS_TO_VERIFY) {
-    try {
-      await verify(payload[key], limit_mb);
-    } catch (e) {
-      Object.assign(newPayload[key], replacements[key] ?? replacements.default);
-      newPayload.redacted = true;
+    if (key in payload) {
+      try {
+        await verify(payload[key], limit_mb, 'stream');
+      } catch (e) {
+        Object.assign(
+          newPayload[key],
+          replacements[key] ?? replacements.default
+        );
+        newPayload.redacted = true;
+      }
     }
   }
 

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

@@ -180,7 +180,7 @@ test.serial('should emit a job-complete event', async (t) => {
   t.assert(!isNaN(event.duration));
 });
 
-test.serial('should emit a log event', async (t) => {
+test.serial.only('should emit a log event', async (t) => {
   let workflowLog: any;
   const plan = {
     id: 'y',

packages/logger/src/logger.ts

@@ -262,6 +262,8 @@ export default function (name?: string, options: LogOptions = {}): Logger {
     }
     j = j as JSONLog;
 
+    j.message ??= [];
+
     log(j.name, j.level, ...j.message);
     return [j.name, j.level, ...j.message];
   };