From 7398d415227515e67ea05f04b147479635775a1a Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Wed, 26 Aug 2026 15:34:27 +0100 Subject: [PATCH 1/3] Worker: better reporting on socket errors (#1505) * Worker: attribute sentry reports to the run that produced them Each run already opened a sentry isolation scope, but engine and socket callbacks fire outside its async context, so breadcrumbs and errors were landing on the global scope and interleaving across concurrent runs. The scope is now held on the run context and re-entered where the reporting actually happens. Co-Authored-By: Claude Opus 5 (1M context) * carry payload size on step-complete events for better reporting * type fix * add payload size to step complete event * report payload size to sentry * changeset --------- Co-authored-by: Claude Opus 5 (1M context) --- .changeset/calm-adults-own.md | 5 + .changeset/calm-otters-sing.md | 5 + .changeset/eight-buses-jump.md | 5 + .changeset/lucky-owls-report.md | 5 + packages/engine-multi/src/api/lifecycle.ts | 12 +- packages/engine-multi/src/events.ts | 5 +- .../src/util/ensure-payload-size.ts | 27 +++- packages/engine-multi/src/worker/events.ts | 4 + .../engine-multi/src/worker/thread/runtime.ts | 3 +- .../engine-multi/test/api/lifecycle.test.ts | 37 ++++++ .../test/util/ensure-payload-size.test.ts | 49 ++++++- packages/lexicon/lightning.d.ts | 1 + packages/ws-worker/src/api/execute.ts | 12 +- packages/ws-worker/src/api/process-events.ts | 11 +- .../ws-worker/src/events/step-complete.ts | 18 ++- packages/ws-worker/src/util/send-event.ts | 62 +++++++-- .../test/events/step-complete.test.ts | 33 +++++ .../ws-worker/test/util/send-event.test.ts | 121 ++++++++++++++++++ 18 files changed, 387 insertions(+), 28 deletions(-) create mode 100644 .changeset/calm-adults-own.md create mode 100644 .changeset/calm-otters-sing.md create mode 100644 .changeset/eight-buses-jump.md create mode 100644 .changeset/lucky-owls-report.md diff --git a/.changeset/calm-adults-own.md b/.changeset/calm-adults-own.md new file mode 100644 index 000000000..65368aad0 --- /dev/null +++ b/.changeset/calm-adults-own.md @@ -0,0 +1,5 @@ +--- +'@openfn/lexicon': patch +--- + +Support dataclipSize diff --git a/.changeset/calm-otters-sing.md b/.changeset/calm-otters-sing.md new file mode 100644 index 000000000..4c8d88be3 --- /dev/null +++ b/.changeset/calm-otters-sing.md @@ -0,0 +1,5 @@ +--- +'@openfn/engine-multi': patch +--- + +Report the size of a job's output state alongside the existing payload redaction check diff --git a/.changeset/eight-buses-jump.md b/.changeset/eight-buses-jump.md new file mode 100644 index 000000000..304e20e52 --- /dev/null +++ b/.changeset/eight-buses-jump.md @@ -0,0 +1,5 @@ +--- +'@openfn/ws-worker': patch +--- + +Include dataclip size in sentry reports diff --git a/.changeset/lucky-owls-report.md b/.changeset/lucky-owls-report.md new file mode 100644 index 000000000..2f03d8faa --- /dev/null +++ b/.changeset/lucky-owls-report.md @@ -0,0 +1,5 @@ +--- +'@openfn/ws-worker': patch +--- + +Attribute Sentry reports to the run that produced them diff --git a/packages/engine-multi/src/api/lifecycle.ts b/packages/engine-multi/src/api/lifecycle.ts index bd1a6254f..7da4c09c2 100644 --- a/packages/engine-multi/src/api/lifecycle.ts +++ b/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(), }); diff --git a/packages/engine-multi/src/events.ts b/packages/engine-multi/src/events.ts index 7d62230b2..f9c95370f 100644 --- a/packages/engine-multi/src/events.ts +++ b/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 { diff --git a/packages/engine-multi/src/util/ensure-payload-size.ts b/packages/engine-multi/src/util/ensure-payload-size.ts index db6dc6ee3..7cb29f84d 100644 --- a/packages/engine-multi/src/util/ensure-payload-size.ts +++ b/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 => { 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 => { + 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; + } } } diff --git a/packages/engine-multi/src/worker/events.ts b/packages/engine-multi/src/worker/events.ts index 495ee1b6a..4ccc26f7f 100644 --- a/packages/engine-multi/src/worker/events.ts +++ b/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 {} diff --git a/packages/engine-multi/src/worker/thread/runtime.ts b/packages/engine-multi/src/worker/thread/runtime.ts index b40059dbe..92d8fa49d 100644 --- a/packages/engine-multi/src/worker/thread/runtime.ts +++ b/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, diff --git a/packages/engine-multi/test/api/lifecycle.test.ts b/packages/engine-multi/test/api/lifecycle.test.ts index f8656ba00..dc05f6fb3 100644 --- a/packages/engine-multi/test/api/lifecycle.test.ts +++ b/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'; diff --git a/packages/engine-multi/test/util/ensure-payload-size.test.ts b/packages/engine-multi/test/util/ensure-payload-size.test.ts index 6babd7f7b..7e474e06c 100644 --- a/packages/engine-multi/test/util/ensure-payload-size.test.ts +++ b/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 => + 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) => { diff --git a/packages/lexicon/lightning.d.ts b/packages/lexicon/lightning.d.ts index 46b4f03c5..0a279f3f7 100644 --- a/packages/lexicon/lightning.d.ts +++ b/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'; diff --git a/packages/ws-worker/src/api/execute.ts b/packages/ws-worker/src/api/execute.ts index 7d348438c..cb9a6ea60 100644 --- a/packages/ws-worker/src/api/execute.ts +++ b/packages/ws-worker/src/api/execute.ts @@ -42,6 +42,9 @@ export type Context = { options: WorkerRunOptions; onFinish: (result: any) => void; + // This run's sentry isolation scope + sentryScope?: Sentry.Scope; + // maybe its better for version numbers to be scribbled here as we go? }; @@ -62,6 +65,10 @@ export function execute( const state = createRunState(plan, input); + // Ensure that each execute call is in its own sentry isolated scope + const sentryScope = Sentry.getIsolationScope().clone(); + sentryScope.setTag('run_id', plan.id!); + const context: Context = { id: plan.id!, channel, @@ -70,11 +77,10 @@ export function execute( engine, options, onFinish, + sentryScope, }; - // Ensure that each execute call is in its own sentry isolated scope - // Because I don't trust it to automatically scope each request properly - Sentry.withIsolationScope(async () => { + Sentry.withIsolationScope(sentryScope, async () => { Sentry.addBreadcrumb({ category: 'run', message: 'Executing run: loading metadata', diff --git a/packages/ws-worker/src/api/process-events.ts b/packages/ws-worker/src/api/process-events.ts index e84c0ca1a..898dd6649 100644 --- a/packages/ws-worker/src/api/process-events.ts +++ b/packages/ws-worker/src/api/process-events.ts @@ -78,7 +78,7 @@ export function eventProcessor( callbacks: Record, options: EventProcessorOptions = {} ) { - const { id: planId, logger } = context; + const { id: planId, logger, sentryScope } = context; const { batchLimit = DEFAULT_BATCH_LIMIT, batchInterval = DEFAULT_BATCH_INTERVAL, @@ -175,7 +175,12 @@ export function eventProcessor( } } catch (e: any) { if (!e.reportedToSentry) { - Sentry.captureException(e); + // Engine events fire outside the async context which created this + // processor, so the run's scope has to be re-entered to pick up its + // breadcrumb trail + Sentry.withIsolationScope(sentryScope, () => + Sentry.captureException(e) + ); logger.error(e); } // Do nothing else here: the error should have been handled @@ -196,7 +201,7 @@ export function eventProcessor( trace('process', name); // TODO this actually shouldn't be here - should be done separately if (name !== 'workflow-log') { - Sentry.addBreadcrumb({ + sentryScope?.addBreadcrumb({ category: 'event', message: name, level: 'info', diff --git a/packages/ws-worker/src/events/step-complete.ts b/packages/ws-worker/src/events/step-complete.ts index 8218faebe..18288e7ef 100644 --- a/packages/ws-worker/src/events/step-complete.ts +++ b/packages/ws-worker/src/events/step-complete.ts @@ -64,6 +64,12 @@ export default async function onStepComplete( duration: event.duration, thread_id: event.threadId, timestamp: timeInMicroseconds(event.time), + // toPrecision (not toFixed) so small dataclips don't round to "0.00" - + // this needs to read sensibly from a few KB up to the ~10mb redaction + // limit, not just near the limit + dataclip_size_mb: event.payloadSize_b + ? (event.payloadSize_b / (1024 * 1024)).toPrecision(3) + : undefined, } as StepCompletePayload; // Feed through the webhook response if it's on state @@ -106,10 +112,18 @@ export default async function onStepComplete( const { output_dataclip, ...eventWithoutDataclip } = evt; context.logger?.debug( - `${context.id} step-complete payload: ${JSON.stringify( + `${context.id} step-complete (without dataclip): ${JSON.stringify( eventWithoutDataclip )}` ); - return sendEvent(context, STEP_COMPLETE, evt); + context.logger?.debug( + `${context.id} step-complete payload is ${evt.dataclip_size_mb}mb` + ); + + return sendEvent(context, STEP_COMPLETE, evt, { + // Raw bytes, not the formatted evt.dataclip_size_mb string - kept out of + // the Lightning-bound payload, only surfaced if this push errors or times out + sentryExtras: { payloadSize_b: event.payloadSize_b }, + }); } diff --git a/packages/ws-worker/src/util/send-event.ts b/packages/ws-worker/src/util/send-event.ts index f4cad446f..bcbb70ee8 100644 --- a/packages/ws-worker/src/util/send-event.ts +++ b/packages/ws-worker/src/util/send-event.ts @@ -6,19 +6,43 @@ import { LightningSocketError, LightningTimeoutError } from '../errors'; // See https://github.com/OpenFn/kit/issues/1137 const allowRetryOntimeout = false; +// channel.socket is not part of our Channel type (or of @types/phoenix's), +// but it exists on the real phoenix Channel instance - reach for it +// defensively so a mock channel in tests, or a future phoenix version, +// cannot turn this into a reporting-path crash +const getSocketState = (channel: any): string | undefined => { + try { + return channel?.socket?.connectionState?.(); + } catch { + return undefined; + } +}; + +export type SentryExtras = Record; + +export type SendEventOptions = { + attempts?: number; + // Extra data to report to sentry + sentryExtras?: SentryExtras; +}; + export const sendEvent = ( - context: Pick, + context: Pick< + Context, + 'logger' | 'channel' | 'id' | 'options' | 'sentryScope' + >, event: string, payload?: any, - attempts?: number + opts: SendEventOptions = {} ) => { // Low defaults here are better for unit tests const { timeoutRetryCount = 1, timeoutRetryDelay = 1 } = context.options ?? {}; + const { attempts, sentryExtras } = opts; const thisAttempt = attempts ?? 1; - const { channel, logger, id: runId = '' } = context; + const { channel, logger, id: runId = '', sentryScope } = context; return new Promise((resolve, reject) => { const report = (error: any) => { @@ -28,16 +52,33 @@ export const sendEvent = ( run_id: runId, event: event, }; - const extras: any = {}; + const extras: SentryExtras = { + // Distinguishes a genuine timeout/error on a healthy channel from + // collateral damage while the channel is mid-rejoin after a drop + channel_state: channel.state, + socket_state: getSocketState(channel), + ...sentryExtras, + }; if (error.rejectMessage) { extras.rejection_reason = error.rejectMessage; } - Sentry.captureException(error, (scope) => { - scope.setContext('run', context); - scope.setExtras(extras); - return scope; + // report() is invoked from a phoenix receive callback, ie off the + // socket's async chain, so the run's scope must be re-entered by hand + Sentry.withIsolationScope(sentryScope, () => { + Sentry.captureException(error, (scope) => { + scope.setTag('run_id', runId); + scope.setTag('lightning_event', event); + // Every timeout (or every socket error) currently collapses into a + // single sentry issue regardless of which event caused it. Splitting + // the fingerprint by event name is what would have made this + // pattern visible without needing to dig through raw events + scope.setFingerprint([error.name, event]); + scope.setContext('run', context); + scope.setExtras(extras); + return scope; + }); }); // Mark that we've reported this to downstream handlers @@ -67,7 +108,10 @@ export const sendEvent = ( ); setTimeout(() => { - sendEvent(context, event, payload, thisAttempt + 1) + sendEvent(context, event, payload, { + attempts: thisAttempt + 1, + sentryExtras, + }) .then(resolve) .catch(reject); }, timeoutRetryDelay); diff --git a/packages/ws-worker/test/events/step-complete.test.ts b/packages/ws-worker/test/events/step-complete.test.ts index a80e7a217..7f53465de 100644 --- a/packages/ws-worker/test/events/step-complete.test.ts +++ b/packages/ws-worker/test/events/step-complete.test.ts @@ -157,6 +157,39 @@ test('send a step:complete event', async (t) => { await handleStepComplete({ channel, state } as any, event); }); +test('does not put payloadSize_b on the wire, only dataclip_size_mb', async (t) => { + const plan = createPlan(); + const jobId = 'job-1'; + const result = { x: 10 }; + + const state = createRunState(plan); + state.activeJob = jobId; + state.activeStep = 'b'; + + const channel = mockChannel({ + [STEP_COMPLETE]: (evt: StepCompletePayload) => { + // payloadSize_b is raw bytes for sentry diagnostics on failure - it + // should never reach the actual Lightning payload, only the derived, + // formatted dataclip_size_mb should + t.false('payloadSize_b' in evt); + t.is(evt.dataclip_size_mb, (1536 / 1024 / 1024).toPrecision(3)); + }, + }); + + const event = { + jobId, + workflowId: plan.id, + state: result, + next: ['a'], + mem: { job: 1, system: 10 }, + duration: 61, + thread_id: 'abc', + time: BigInt(123), + payloadSize_b: 1536, + } as JobCompletePayload; + await handleStepComplete({ channel, state } as any, event); +}); + test('do not include dataclips in step:complete if output_dataclip is false', async (t) => { const plan = createPlan(); const jobId = 'job-1'; diff --git a/packages/ws-worker/test/util/send-event.test.ts b/packages/ws-worker/test/util/send-event.test.ts index e1bd9b9ac..67256a517 100644 --- a/packages/ws-worker/test/util/send-event.test.ts +++ b/packages/ws-worker/test/util/send-event.test.ts @@ -1,4 +1,6 @@ import test from 'ava'; +import { EventEmitter } from 'node:events'; +import * as Sentry from '@sentry/node'; import { createMockLogger } from '@openfn/logger'; import { mockChannel } from '../../src/mock/sockets'; @@ -268,4 +270,123 @@ test.serial('should report to sentry if the event timesout', async (t) => { } const reports = await waitForSentryReport(testkit); t.is(reports[0].error?.name, 'LightningTimeoutError'); + + // Tags are indexed (unlike the run context below), so these are what make + // it possible to ask sentry "is it only step:complete that times out?" + t.is(reports[0].tags.run_id, 'x'); + t.is(reports[0].tags.lightning_event, EVENT_NAME); +}); + +test.serial('should fingerprint sentry reports by error type and event name', async (t) => { + // Without this, every timeout for every event collapses into one sentry + // issue - this is the change that would have made the step:complete + // pattern visible without digging through raw events. Each event is + // checked against a fresh testkit so the two reports cannot be confused + // with each other or raced against waitForSentryReport's "at least one" + // polling. + const channelA = mockChannel({}); + await t.throwsAsync(() => + sendEvent({ id: 'x', channel: channelA, logger, options: {} }, 'step:complete', {}) + ); + const [stepReport] = await waitForSentryReport(testkit); + t.deepEqual(stepReport.originalReport.fingerprint, [ + 'LightningTimeoutError', + 'step:complete', + ]); + + testkit.reset(); + + const channelB = mockChannel({}); + await t.throwsAsync(() => + sendEvent({ id: 'x', channel: channelB, logger, options: {} }, 'run:complete', {}) + ); + const [runReport] = await waitForSentryReport(testkit); + t.deepEqual(runReport.originalReport.fingerprint, [ + 'LightningTimeoutError', + 'run:complete', + ]); +}); + +test.serial('should report channel and socket state alongside a failed event', async (t) => { + // Distinguishes a genuine failure on a healthy channel from collateral + // damage while the channel is mid-rejoin after a drop + const channel = { + ...mockChannel({}), + state: 'errored', + socket: { connectionState: () => 'connecting' }, + }; + + await t.throwsAsync(() => + sendEvent({ id: 'x', channel, logger, options: {} }, 'step:complete', {}) + ); + + const reports = await waitForSentryReport(testkit); + t.is(reports[0].extra?.channel_state, 'errored'); + t.is(reports[0].extra?.socket_state, 'connecting'); +}); + +// The real phoenix channel invokes its receive callbacks from the socket's +// message chain, which is not the chain that called push(). mockChannel defers +// with a setTimeout created inside push, so async context leaks through it and +// it cannot exercise this. This mock replies from a pump created up-front, so +// the callback runs with no inherited context, exactly like the real socket +const mockDetachedChannel = () => { + const bus = new EventEmitter(); + const pump = setInterval(() => bus.emit('reply'), 1); + + return { + stop: () => clearInterval(pump), + channel: { + push: () => { + const responses = {} as Record void>; + bus.once('reply', () => responses.error?.('detached')); + + const receive = { + receive: (status: string, callback: (e?: any) => void) => { + responses[status] = callback; + return receive; + }, + }; + return receive; + }, + } as any, + }; +}; + +test.serial('should report to sentry against the run scope', async (t) => { + const sentryScope = Sentry.getIsolationScope().clone(); + sentryScope.setTag('run_id', 'run-1'); + sentryScope.addBreadcrumb({ category: 'event', message: 'job-complete' }); + + const { channel, stop } = mockDetachedChannel(); + const context = { id: 'run-1', channel, logger, options: {}, sentryScope }; + + await t.throwsAsync(() => sendEvent(context, 'step:complete', {})); + stop(); + + const reports = await waitForSentryReport(testkit); + t.is(reports[0].error?.name, 'LightningSocketError'); + t.is(reports[0].tags.run_id, 'run-1'); + + // The run's breadcrumb trail must survive too - this is why the capture + // re-enters the scope rather than passing it to captureException, which + // merges tags but drops breadcrumbs + const trail = reports[0].originalReport?.breadcrumbs ?? []; + t.true(trail.some((b: any) => b.message === 'job-complete')); +}); + +test.serial('should report caller-supplied sentryExtras alongside a failed event', async (t) => { + const EVENT_NAME = 'test'; + const channel = { ...mockChannel({}), state: 'joined' }; + + const context = { id: 'x', channel, logger, options: {} }; + + await t.throwsAsync(() => + sendEvent(context, EVENT_NAME, {}, { sentryExtras: { payloadSize_b: 1536 } }) + ); + + const reports = await waitForSentryReport(testkit); + t.is(reports[0].extra?.payloadSize_b, 1536); + // sentryExtras must not crowd out the fields send-event already reports + t.is(reports[0].extra?.channel_state, 'joined'); }); From c433bcf6faca2936e6355a872b8007162b78e966 Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Wed, 26 Aug 2026 16:17:05 +0100 Subject: [PATCH 2/3] Worker: Fix dataclip serialization (#1507) * Worker: attribute sentry reports to the run that produced them Each run already opened a sentry isolation scope, but engine and socket callbacks fire outside its async context, so breadcrumbs and errors were landing on the global scope and interleaving across concurrent runs. The scope is now held on the run context and re-entered where the reporting actually happens. Co-Authored-By: Claude Opus 5 (1M context) * carry payload size on step-complete events for better reporting * type fix * report better errors out of sentry * Gate double-encoding fix for output dataclips behind WORKER_NO_STRINGIFY_STATE The worker double-JSON-encodes step output dataclips before sending them to Lightning: once via a manual stringify, then again when phoenix serializes the envelope. That double-encoding is what bloats large dataclips past Lightning's websocket frame limit and kills the connection mid-run. The runtime already sanitizes/clones state before it reaches this point, so the manual stringify is redundant - but skipping it changes the wire format, so it's opt-in via --stringify-state/--no-stringify-state (CLI) or WORKER_NO_STRINGIFY_STATE (env), default false to preserve current behaviour until the matching Lightning-side support (2.19+) is widely deployed. * docs and changelog --------- Co-authored-by: Claude Opus 5 (1M context) --- .changeset/tame-hats-drum.md | 5 + .changeset/tidy-plums-obey.md | 5 + packages/ws-worker/README.md | 6 + packages/ws-worker/src/api/execute.ts | 10 + packages/ws-worker/src/channels/run.ts | 8 +- .../ws-worker/src/channels/worker-queue.ts | 12 +- packages/ws-worker/src/errors.ts | 21 ++ .../ws-worker/src/events/step-complete.ts | 10 +- packages/ws-worker/src/mock/sockets.ts | 19 +- packages/ws-worker/src/server.ts | 23 ++- packages/ws-worker/src/start.ts | 1 + packages/ws-worker/src/util/cli.ts | 16 +- .../src/util/convert-lightning-plan.ts | 3 + packages/ws-worker/test/api/execute.test.ts | 26 +++ packages/ws-worker/test/channels/run.test.ts | 18 ++ .../test/channels/worker-queue.test.ts | 26 +++ .../test/events/step-complete.test.ts | 80 ++++++++ packages/ws-worker/test/util/cli.test.ts | 45 +++++ .../ws-worker/test/util/send-event.test.ts | 190 ++++++++++-------- 19 files changed, 427 insertions(+), 97 deletions(-) create mode 100644 .changeset/tame-hats-drum.md create mode 100644 .changeset/tidy-plums-obey.md diff --git a/.changeset/tame-hats-drum.md b/.changeset/tame-hats-drum.md new file mode 100644 index 000000000..9b4cb187e --- /dev/null +++ b/.changeset/tame-hats-drum.md @@ -0,0 +1,5 @@ +--- +'@openfn/ws-worker': patch +--- + +Capture more diagnostic detail when the connection to Lightning drops unexpectedly diff --git a/.changeset/tidy-plums-obey.md b/.changeset/tidy-plums-obey.md new file mode 100644 index 000000000..4f6f08c4d --- /dev/null +++ b/.changeset/tidy-plums-obey.md @@ -0,0 +1,5 @@ +--- +'@openfn/ws-worker': patch +--- + +Reduce dataclip bloat when sending step results to Lightning diff --git a/packages/ws-worker/README.md b/packages/ws-worker/README.md index 6d0cbfb2c..b11b611eb 100644 --- a/packages/ws-worker/README.md +++ b/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). diff --git a/packages/ws-worker/src/api/execute.ts b/packages/ws-worker/src/api/execute.ts index cb9a6ea60..9106c221a 100644 --- a/packages/ws-worker/src/api/execute.ts +++ b/packages/ws-worker/src/api/execute.ts @@ -80,6 +80,16 @@ export function execute( sentryScope, }; + // Log pheonix channel errors to sentry + channel.onError((...args: any) => { + sentryScope.addBreadcrumb({ + category: 'channel', + message: 'Channel error', + level: 'warning', + data: { state: channel.state, args }, + }); + }); + Sentry.withIsolationScope(sentryScope, async () => { Sentry.addBreadcrumb({ category: 'run', diff --git a/packages/ws-worker/src/channels/run.ts b/packages/ws-worker/src/channels/run.ts index b99a2ce5f..2c2a95ec7 100644 --- a/packages/ws-worker/src/channels/run.ts +++ b/packages/ws-worker/src/channels/run.ts @@ -63,7 +63,13 @@ const joinRunChannel = ( channel.onError((...args: any) => { // Error occurred on the channel // (the socket will try to reconnect with backoff) - logger.debug(`Critical error in channel ${channelName}`, args); + // Note we don't report to sentry here - the socket error handler does that + logger.error( + `Critical error in channel ${channelName}`, + args, + 'state:', + channel.state + ); }); }); }; diff --git a/packages/ws-worker/src/channels/worker-queue.ts b/packages/ws-worker/src/channels/worker-queue.ts index 4e70d4460..4c1694c02 100644 --- a/packages/ws-worker/src/channels/worker-queue.ts +++ b/packages/ws-worker/src/channels/worker-queue.ts @@ -103,9 +103,15 @@ const connectToWorkerQueue = ( // On close, the socket will try and reconnect itself // Forever, so far as I can tell - socket.onClose((_e: any) => { - logger.debug('queue socket closed'); - events.emit('disconnect'); + socket.onClose((e: any) => { + logger.warn( + `queue socket closed: code=${e?.code} reason=${e?.reason} clean=${e?.wasClean}` + ); + events.emit('disconnect', { + code: e?.code, + reason: e?.reason, + wasClean: e?.wasClean, + }); }); // if we fail to connect, the socket will try to reconnect diff --git a/packages/ws-worker/src/errors.ts b/packages/ws-worker/src/errors.ts index c3ee3ea73..522e4ad8f 100644 --- a/packages/ws-worker/src/errors.ts +++ b/packages/ws-worker/src/errors.ts @@ -35,3 +35,24 @@ export class LightningTimeoutError extends Error { super(`[${event}] timeout`); } } + +export type SocketCloseDetails = { + code?: number; + reason?: string; + wasClean?: boolean; +}; + +export class LightningSocketClosedError extends Error { + name = 'LightningSocketClosedError'; + code?: number; + reason?: string; + wasClean?: boolean; + constructor({ code, reason, wasClean }: SocketCloseDetails = {}) { + super( + `Lightning socket closed: code=${code} reason=${reason ?? 'unknown'}` + ); + this.code = code; + this.reason = reason; + this.wasClean = wasClean; + } +} diff --git a/packages/ws-worker/src/events/step-complete.ts b/packages/ws-worker/src/events/step-complete.ts index 18288e7ef..4ec2ddb8f 100644 --- a/packages/ws-worker/src/events/step-complete.ts +++ b/packages/ws-worker/src/events/step-complete.ts @@ -98,10 +98,14 @@ export default async function onStepComplete( ]); } else { evt.output_dataclip_id = dataclipId; + // Write the dataclip if it's not too big if (!options || options.outputDataclips !== false) { - const payload = stringify(outputState); - // Write the dataclip if it's not too big - evt.output_dataclip = payload; + // For back compatibility, stringify the the state object before sending + // Note that this causes payloads to bloat + // In a major version soon, we should remove the option and never stringify + evt.output_dataclip = options?.noStringifyState + ? outputState + : stringify(outputState); } } diff --git a/packages/ws-worker/src/mock/sockets.ts b/packages/ws-worker/src/mock/sockets.ts index 9d2ac18b2..534106ee0 100644 --- a/packages/ws-worker/src/mock/sockets.ts +++ b/packages/ws-worker/src/mock/sockets.ts @@ -4,6 +4,8 @@ type EventHandler = (evt?: any) => void; export const mockChannel = ( callbacks: Record = {} ): any => { + const closeCallbacks: EventHandler[] = []; + const errorCallbacks: EventHandler[] = []; const c = { on: (event: string, fn: EventHandler) => { // TODO support multiple callbacks @@ -71,8 +73,21 @@ export const mockChannel = ( return receive; }, leave: () => {}, - onClose: () => {}, - onError: () => {}, + // Real phoenix channels support multiple onClose/onError bindings (each + // call pushes onto an array), which is now relied on in production - + // run.ts and execute.ts both bind onError on the same channel. So this + // collects every registered callback rather than keeping only the last + onClose: (fn: EventHandler) => { + closeCallbacks.push(fn); + }, + onError: (fn: EventHandler) => { + errorCallbacks.push(fn); + }, + // test helpers: fire every registered callback, as the real socket would + _triggerClose: (...args: any[]) => + closeCallbacks.forEach((fn) => fn(...args)), + _triggerError: (...args: any[]) => + errorCallbacks.forEach((fn) => fn(...args)), }; return c; }; diff --git a/packages/ws-worker/src/server.ts b/packages/ws-worker/src/server.ts index 25ea9d220..3c95d2f0b 100644 --- a/packages/ws-worker/src/server.ts +++ b/packages/ws-worker/src/server.ts @@ -31,6 +31,7 @@ import { convertRun } from './util'; import parseWorkloops from './util/parse-workloops'; import getDefaultWorkloopConfig from './util/get-default-workloop-config'; import { matchesIgnoredError } from './util/ignored-errors'; +import { LightningSocketClosedError, SocketCloseDetails } from './errors'; const exec = promisify(_exec); @@ -61,6 +62,7 @@ export type ServerOptions = { claimTimeoutSeconds?: number; payloadLimitMb?: number; // max memory limit for socket payload (ie, step:complete, log) logPayloadLimitMb?: number; // max memory limit for log payloads specifically + noStringifyState?: boolean; // send output dataclips as native JSON instead of a pre-stringified string. Requires lightning support collectionsVersion?: string; collectionsUrl?: string; monorepoDir?: string; @@ -137,18 +139,32 @@ function connect(app: ServerApp, logger: Logger, options: ServerOptions = {}) { }; // We were disconnected from the queue - const onDisconnect = () => { + const onDisconnect = (details: SocketCloseDetails = {}) => { for (const w of app.workloops) { if (!w.isStopped()) { w.stop('Socket disconnected unexpectedly'); } } if (!app.destroyed) { - logger.info('Connection to lightning lost'); + logger.info( + `Connection to lightning lost (code=${details.code} reason=${details.reason} clean=${details.wasClean})` + ); logger.info( 'Worker will automatically reconnect when lightning is back online' ); - // So far as I know, the socket will try and reconnect in the background forever + Sentry.captureException( + new LightningSocketClosedError(details), + (scope) => { + scope.setFingerprint([ + 'LightningSocketClosedError', + String(details.code), + details.reason ?? '', + ]); + scope.setTag('close_code', String(details.code)); + scope.setExtras(details); + return scope; + } + ); } }; @@ -353,6 +369,7 @@ function createServer(engine: RuntimeEngine, options: ServerOptions = {}) { options.logPayloadLimitMb = app.options.logPayloadLimitMb; } + options.noStringifyState = app.options.noStringifyState; options.timeoutRetryCount = app.options.timeoutRetryCount; options.timeoutRetryDelay = app.options.timeoutRetryDelayMs ?? app.options.socketTimeoutSeconds; diff --git a/packages/ws-worker/src/start.ts b/packages/ws-worker/src/start.ts index e8650ee72..9ee722f8f 100644 --- a/packages/ws-worker/src/start.ts +++ b/packages/ws-worker/src/start.ts @@ -59,6 +59,7 @@ function engineReady(engine: any) { maxWorkflows: effectiveCapacity, workloopConfigs, payloadLimitMb: args.payloadMemory, + noStringifyState: args.noStringifyState, logPayloadLimitMb: args.logPayloadMemory ?? 1, // Default to 1MB collectionsVersion: args.collectionsVersion, collectionsUrl: args.collectionsUrl, diff --git a/packages/ws-worker/src/util/cli.ts b/packages/ws-worker/src/util/cli.ts index e6bd9d374..e843e0e3c 100644 --- a/packages/ws-worker/src/util/cli.ts +++ b/packages/ws-worker/src/util/cli.ts @@ -33,6 +33,7 @@ type Args = { messageTimeoutSeconds?: number; mock?: boolean; monorepoDir?: string; + noStringifyState?: boolean; payloadMemory?: number; port?: number; profile?: boolean; @@ -96,6 +97,7 @@ export default function parseArgs(argv: string[]): Args { WORKER_MAX_RUN_MEMORY_MB, WORKER_MAX_STATE_MEMORY_MB, WORKER_MESSAGE_TIMEOUT_SECONDS, + WORKER_NO_STRINGIFY_STATE, WORKER_PORT, WORKER_PROFILE_POLL_INTERVAL_MS, WORKER_PROFILE, @@ -233,6 +235,11 @@ export default function parseArgs(argv: string[]): Args { 'Maximum memory allocated to a single run, in mb. Env: WORKER_MAX_PAYLOAD_MB', type: 'number', }) + .option('stringify-state', { + description: + 'Pass --no-stringify-state or set WORKER_NO_STRINGIFY_STATE to optimize stateful payloads sent to lightning. Not back compatible with lightning versions older than 2.19.', + type: 'boolean', + }) .option('cgroup', { alias: ['enable-cgroup-enforcement', 'cgroups'], description: @@ -298,7 +305,7 @@ export default function parseArgs(argv: string[]): Args { 'production start configuration with 1 fast lane workloop (capacity 1) and a second workloop with capacity 4' ); - const args = parser.parse() as Args; + const args = parser.parse() as Args & { stringifyState?: boolean }; const resolvedWorkloops = setArg(args.workloops, WORKER_WORKLOOPS) as | string @@ -348,6 +355,13 @@ export default function parseArgs(argv: string[]): Args { ? parseInt(WORKER_MAX_STATE_MEMORY_MB, 10) : undefined), payloadMemory: setArg(args.payloadMemory, WORKER_MAX_PAYLOAD_MB, 10), + // args.stringifyState is positively framed (see the --stringify-state + // option above); everything downstream of parseArgs uses the negatively + // framed noStringifyState, matching WORKER_NO_STRINGIFY_STATE + noStringifyState: + args.stringifyState !== undefined + ? !args.stringifyState + : setArg(undefined, WORKER_NO_STRINGIFY_STATE, false), logPayloadMemory: setArg( args.logPayloadMemory, WORKER_MAX_LOG_PAYLOAD_MB, diff --git a/packages/ws-worker/src/util/convert-lightning-plan.ts b/packages/ws-worker/src/util/convert-lightning-plan.ts index df595a94b..e59e5824d 100644 --- a/packages/ws-worker/src/util/convert-lightning-plan.ts +++ b/packages/ws-worker/src/util/convert-lightning-plan.ts @@ -33,6 +33,9 @@ export type WorkerRunOptions = ExecuteOptions & { outputDataclips?: boolean; payloadLimitMb?: number; logPayloadLimitMb?: number; + // Send the output dataclip as a native JSON value instead of a + // pre-stringified string. Defaults to false (old behaviour) + noStringifyState?: boolean; jobLogLevel?: LogLevel; timeoutRetryCount?: number; timeoutRetryDelay?: number; diff --git a/packages/ws-worker/test/api/execute.test.ts b/packages/ws-worker/test/api/execute.test.ts index 1e33e615d..dbce02066 100644 --- a/packages/ws-worker/test/api/execute.test.ts +++ b/packages/ws-worker/test/api/execute.test.ts @@ -402,6 +402,32 @@ test('execute should return a context object', async (t) => { }); }); +test('execute should breadcrumb a channel error onto the run scope', async (t) => { + const channel = mockChannel(mockEventHandlers); + const engine = await createMockRTE(); + const logger = createMockLogger(); + + const plan = { + id: 'a', + workflow: { + steps: [ + { + expression: 'fn(() => ({ done: true }))', + }, + ], + }, + } as ExecutionPlan; + + const context = execute(channel, engine, logger, plan, {}, {}, () => {}); + + channel._triggerError('boom'); + + const breadcrumbs = context.sentryScope!.getScopeData().breadcrumbs; + const found = breadcrumbs.find((b: any) => b.message === 'Channel error'); + t.truthy(found); + t.is(found!.category, 'channel'); +}); + // TODO this is more of an engine test really, but worth having I suppose test('execute should lazy-load a credential', async (t) => { const logger = createMockLogger(); diff --git a/packages/ws-worker/test/channels/run.test.ts b/packages/ws-worker/test/channels/run.test.ts index 14ba80672..0f631ed1a 100644 --- a/packages/ws-worker/test/channels/run.test.ts +++ b/packages/ws-worker/test/channels/run.test.ts @@ -43,3 +43,21 @@ test('should fail to join an run channel with an invalid token', async (t) => { t.pass(); } }); + +test('should log an error including channel state when the channel errors', async (t) => { + const logger = createMockLogger(); + const channel = mockChannel({ + join: () => ({ status: 'ok' }), + [GET_PLAN]: () => runs['run-1'], + }); + const socket = new MockSocket('www', { 'run:a': channel }); + + await joinRunChannel(socket, 'x.y.z', 'a', logger); + + channel.state = 'errored'; + channel._triggerError('boom'); + + const log = logger._find('error', /Critical error in channel run:a/); + t.truthy(log); + t.regex(log!.message as string, /errored/); +}); diff --git a/packages/ws-worker/test/channels/worker-queue.test.ts b/packages/ws-worker/test/channels/worker-queue.test.ts index d8d640d41..6bfd7a86a 100644 --- a/packages/ws-worker/test/channels/worker-queue.test.ts +++ b/packages/ws-worker/test/channels/worker-queue.test.ts @@ -24,6 +24,32 @@ test('should connect', (t) => { }); }); +test('should emit disconnect with the close code, reason and wasClean', (t) => { + return new Promise((done) => { + connectToWorkerQueue('www', 'a', 'secret', logger, { + SocketConstructor: MockSocket as any, + }) + .on('connect', ({ socket }) => { + // Real phoenix sockets invoke onClose with a CloseEvent-like object - + // MockSocket doesn't drive this itself, so trigger it directly + // @ts-ignore accessing test-only internals + socket.callbacks.onClose({ + code: 1009, + reason: 'message too big', + wasClean: false, + }); + }) + .on('disconnect', (details) => { + t.deepEqual(details, { + code: 1009, + reason: 'message too big', + wasClean: false, + }); + done(); + }); + }); +}); + test('should connect with an auth token', async (t) => { return new Promise((done) => { const workerId = 'x'; diff --git a/packages/ws-worker/test/events/step-complete.test.ts b/packages/ws-worker/test/events/step-complete.test.ts index 7f53465de..f317f94bd 100644 --- a/packages/ws-worker/test/events/step-complete.test.ts +++ b/packages/ws-worker/test/events/step-complete.test.ts @@ -157,6 +157,86 @@ test('send a step:complete event', async (t) => { await handleStepComplete({ channel, state } as any, event); }); +test('stringifies the output dataclip by default (noStringifyState unset)', async (t) => { + const plan = createPlan(); + const jobId = 'job-1'; + const result = { x: 10 }; + + const state = createRunState(plan); + state.activeJob = jobId; + state.activeStep = 'b'; + + const channel = mockChannel({ + [STEP_COMPLETE]: (evt: StepCompletePayload) => { + t.is(evt.output_dataclip, JSON.stringify(result)); + }, + }); + + const event = { + jobId, + workflowId: plan.id, + state: result, + next: ['a'], + time: BigInt(123), + } as JobCompletePayload; + await handleStepComplete({ channel, state } as any, event); +}); + +test('stringifies the output dataclip when noStringifyState is false', async (t) => { + const plan = createPlan(); + const jobId = 'job-1'; + const result = { x: 10 }; + + const state = createRunState(plan); + state.activeJob = jobId; + state.activeStep = 'b'; + + const options = { noStringifyState: false }; + + const channel = mockChannel({ + [STEP_COMPLETE]: (evt: StepCompletePayload) => { + t.is(evt.output_dataclip, JSON.stringify(result)); + }, + }); + + const event = { + jobId, + workflowId: plan.id, + state: result, + next: ['a'], + time: BigInt(123), + } as JobCompletePayload; + await handleStepComplete({ channel, state, options } as any, event); +}); + +test('sends the output dataclip as a native object when noStringifyState is true', async (t) => { + const plan = createPlan(); + const jobId = 'job-1'; + const result = { x: 10 }; + + const state = createRunState(plan); + state.activeJob = jobId; + state.activeStep = 'b'; + + const options = { noStringifyState: true }; + + const channel = mockChannel({ + [STEP_COMPLETE]: (evt: StepCompletePayload) => { + t.deepEqual(evt.output_dataclip, result as any); + t.not(evt.output_dataclip, JSON.stringify(result) as any); + }, + }); + + const event = { + jobId, + workflowId: plan.id, + state: result, + next: ['a'], + time: BigInt(123), + } as JobCompletePayload; + await handleStepComplete({ channel, state, options } as any, event); +}); + test('does not put payloadSize_b on the wire, only dataclip_size_mb', async (t) => { const plan = createPlan(); const jobId = 'job-1'; diff --git a/packages/ws-worker/test/util/cli.test.ts b/packages/ws-worker/test/util/cli.test.ts index be156dd72..f1930b927 100644 --- a/packages/ws-worker/test/util/cli.test.ts +++ b/packages/ws-worker/test/util/cli.test.ts @@ -68,6 +68,51 @@ test('cli should set default values for unspecified options', (t) => { t.is(args.engineValidationTimeoutMs, 5000); t.is(args.profile, false); t.is(args.profilePollIntervalMs, 10); + t.is(args.noStringifyState, false); +}); + +test('cli should default noStringifyState to false', (t) => { + const argv = 'pnpm start'.split(' '); + const args = cli(argv); + + t.is(args.noStringifyState, false); +}); + +test('cli should enable noStringifyState via --no-stringify-state', (t) => { + const argv = 'pnpm start --no-stringify-state'.split(' '); + const args = cli(argv); + + t.is(args.noStringifyState, true); +}); + +test('cli should enable noStringifyState via --stringify-state false', (t) => { + const argv = 'pnpm start --stringify-state false'.split(' '); + const args = cli(argv); + + t.is(args.noStringifyState, true); +}); + +test('cli should disable noStringifyState via --stringify-state true', (t) => { + const argv = 'pnpm start --stringify-state true'.split(' '); + const args = cli(argv); + + t.is(args.noStringifyState, false); +}); + +test('cli should enable noStringifyState via WORKER_NO_STRINGIFY_STATE', (t) => { + process.env.WORKER_NO_STRINGIFY_STATE = 'true'; + const argv = 'pnpm start'.split(' '); + const args = cli(argv); + + t.is(args.noStringifyState, true); +}); + +test('cli --no-stringify-state should override WORKER_NO_STRINGIFY_STATE', (t) => { + process.env.WORKER_NO_STRINGIFY_STATE = 'true'; + const argv = 'pnpm start --stringify-state true'.split(' '); + const args = cli(argv); + + t.is(args.noStringifyState, false); }); test('cli should handle boolean options correctly', (t) => { diff --git a/packages/ws-worker/test/util/send-event.test.ts b/packages/ws-worker/test/util/send-event.test.ts index 67256a517..8ff85230e 100644 --- a/packages/ws-worker/test/util/send-event.test.ts +++ b/packages/ws-worker/test/util/send-event.test.ts @@ -12,6 +12,34 @@ const testkit = initSentry(); const logger = createMockLogger(undefined, { json: true }); +// The real phoenix channel invokes its receive callbacks from the socket's +// message chain, which is not the chain that called push(). mockChannel defers +// with a setTimeout created inside push, so async context leaks through it and +// it cannot exercise this. This mock replies from a pump created up-front, so +// the callback runs with no inherited context, exactly like the real socket +const mockDetachedChannel = () => { + const bus = new EventEmitter(); + const pump = setInterval(() => bus.emit('reply'), 1); + + return { + stop: () => clearInterval(pump), + channel: { + push: () => { + const responses = {} as Record void>; + bus.once('reply', () => responses.error?.('detached')); + + const receive = { + receive: (status: string, callback: (e?: any) => void) => { + responses[status] = callback; + return receive; + }, + }; + return receive; + }, + } as any, + }; +}; + test.beforeEach(() => { testkit.reset(); logger._reset(); @@ -277,81 +305,67 @@ test.serial('should report to sentry if the event timesout', async (t) => { t.is(reports[0].tags.lightning_event, EVENT_NAME); }); -test.serial('should fingerprint sentry reports by error type and event name', async (t) => { - // Without this, every timeout for every event collapses into one sentry - // issue - this is the change that would have made the step:complete - // pattern visible without digging through raw events. Each event is - // checked against a fresh testkit so the two reports cannot be confused - // with each other or raced against waitForSentryReport's "at least one" - // polling. - const channelA = mockChannel({}); - await t.throwsAsync(() => - sendEvent({ id: 'x', channel: channelA, logger, options: {} }, 'step:complete', {}) - ); - const [stepReport] = await waitForSentryReport(testkit); - t.deepEqual(stepReport.originalReport.fingerprint, [ - 'LightningTimeoutError', - 'step:complete', - ]); - - testkit.reset(); - - const channelB = mockChannel({}); - await t.throwsAsync(() => - sendEvent({ id: 'x', channel: channelB, logger, options: {} }, 'run:complete', {}) - ); - const [runReport] = await waitForSentryReport(testkit); - t.deepEqual(runReport.originalReport.fingerprint, [ - 'LightningTimeoutError', - 'run:complete', - ]); -}); - -test.serial('should report channel and socket state alongside a failed event', async (t) => { - // Distinguishes a genuine failure on a healthy channel from collateral - // damage while the channel is mid-rejoin after a drop - const channel = { - ...mockChannel({}), - state: 'errored', - socket: { connectionState: () => 'connecting' }, - }; - - await t.throwsAsync(() => - sendEvent({ id: 'x', channel, logger, options: {} }, 'step:complete', {}) - ); - - const reports = await waitForSentryReport(testkit); - t.is(reports[0].extra?.channel_state, 'errored'); - t.is(reports[0].extra?.socket_state, 'connecting'); -}); +test.serial( + 'should fingerprint sentry reports by error type and event name', + async (t) => { + // Without this, every timeout for every event collapses into one sentry + // issue - this is the change that would have made the step:complete + // pattern visible without digging through raw events. Each event is + // checked against a fresh testkit so the two reports cannot be confused + // with each other or raced against waitForSentryReport's "at least one" + // polling. + const channelA = mockChannel({}); + await t.throwsAsync(() => + sendEvent( + { id: 'x', channel: channelA, logger, options: {} }, + 'step:complete', + {} + ) + ); + const [stepReport] = await waitForSentryReport(testkit); + t.deepEqual(stepReport.originalReport.fingerprint, [ + 'LightningTimeoutError', + 'step:complete', + ]); + + testkit.reset(); + + const channelB = mockChannel({}); + await t.throwsAsync(() => + sendEvent( + { id: 'x', channel: channelB, logger, options: {} }, + 'run:complete', + {} + ) + ); + const [runReport] = await waitForSentryReport(testkit); + t.deepEqual(runReport.originalReport.fingerprint, [ + 'LightningTimeoutError', + 'run:complete', + ]); + } +); -// The real phoenix channel invokes its receive callbacks from the socket's -// message chain, which is not the chain that called push(). mockChannel defers -// with a setTimeout created inside push, so async context leaks through it and -// it cannot exercise this. This mock replies from a pump created up-front, so -// the callback runs with no inherited context, exactly like the real socket -const mockDetachedChannel = () => { - const bus = new EventEmitter(); - const pump = setInterval(() => bus.emit('reply'), 1); +test.serial( + 'should report channel and socket state alongside a failed event', + async (t) => { + // Distinguishes a genuine failure on a healthy channel from collateral + // damage while the channel is mid-rejoin after a drop + const channel = { + ...mockChannel({}), + state: 'errored', + socket: { connectionState: () => 'connecting' }, + }; - return { - stop: () => clearInterval(pump), - channel: { - push: () => { - const responses = {} as Record void>; - bus.once('reply', () => responses.error?.('detached')); + await t.throwsAsync(() => + sendEvent({ id: 'x', channel, logger, options: {} }, 'step:complete', {}) + ); - const receive = { - receive: (status: string, callback: (e?: any) => void) => { - responses[status] = callback; - return receive; - }, - }; - return receive; - }, - } as any, - }; -}; + const reports = await waitForSentryReport(testkit); + t.is(reports[0].extra?.channel_state, 'errored'); + t.is(reports[0].extra?.socket_state, 'connecting'); + } +); test.serial('should report to sentry against the run scope', async (t) => { const sentryScope = Sentry.getIsolationScope().clone(); @@ -375,18 +389,26 @@ test.serial('should report to sentry against the run scope', async (t) => { t.true(trail.some((b: any) => b.message === 'job-complete')); }); -test.serial('should report caller-supplied sentryExtras alongside a failed event', async (t) => { - const EVENT_NAME = 'test'; - const channel = { ...mockChannel({}), state: 'joined' }; +test.serial( + 'should report caller-supplied sentryExtras alongside a failed event', + async (t) => { + const EVENT_NAME = 'test'; + const channel = { ...mockChannel({}), state: 'joined' }; - const context = { id: 'x', channel, logger, options: {} }; + const context = { id: 'x', channel, logger, options: {} }; - await t.throwsAsync(() => - sendEvent(context, EVENT_NAME, {}, { sentryExtras: { payloadSize_b: 1536 } }) - ); + await t.throwsAsync(() => + sendEvent( + context, + EVENT_NAME, + {}, + { sentryExtras: { payloadSize_b: 1536 } } + ) + ); - const reports = await waitForSentryReport(testkit); - t.is(reports[0].extra?.payloadSize_b, 1536); - // sentryExtras must not crowd out the fields send-event already reports - t.is(reports[0].extra?.channel_state, 'joined'); -}); + const reports = await waitForSentryReport(testkit); + t.is(reports[0].extra?.payloadSize_b, 1536); + // sentryExtras must not crowd out the fields send-event already reports + t.is(reports[0].extra?.channel_state, 'joined'); + } +); From c938db92b62f2411c41383b2c4083c66cb496cef Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Wed, 26 Aug 2026 16:21:44 +0100 Subject: [PATCH 3/3] versions --- .changeset/calm-adults-own.md | 5 ----- .changeset/calm-otters-sing.md | 5 ----- .changeset/eight-buses-jump.md | 5 ----- .changeset/lucky-owls-report.md | 5 ----- .changeset/tame-hats-drum.md | 5 ----- .changeset/tidy-plums-obey.md | 5 ----- packages/cli/CHANGELOG.md | 7 +++++++ packages/cli/package.json | 2 +- packages/engine-multi/CHANGELOG.md | 8 ++++++++ packages/engine-multi/package.json | 2 +- packages/lexicon/CHANGELOG.md | 6 ++++++ packages/lexicon/package.json | 2 +- packages/lightning-mock/CHANGELOG.md | 9 +++++++++ packages/lightning-mock/package.json | 2 +- packages/ws-worker/CHANGELOG.md | 13 +++++++++++++ packages/ws-worker/package.json | 2 +- 16 files changed, 48 insertions(+), 35 deletions(-) delete mode 100644 .changeset/calm-adults-own.md delete mode 100644 .changeset/calm-otters-sing.md delete mode 100644 .changeset/eight-buses-jump.md delete mode 100644 .changeset/lucky-owls-report.md delete mode 100644 .changeset/tame-hats-drum.md delete mode 100644 .changeset/tidy-plums-obey.md diff --git a/.changeset/calm-adults-own.md b/.changeset/calm-adults-own.md deleted file mode 100644 index 65368aad0..000000000 --- a/.changeset/calm-adults-own.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -'@openfn/lexicon': patch ---- - -Support dataclipSize diff --git a/.changeset/calm-otters-sing.md b/.changeset/calm-otters-sing.md deleted file mode 100644 index 4c8d88be3..000000000 --- a/.changeset/calm-otters-sing.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -'@openfn/engine-multi': patch ---- - -Report the size of a job's output state alongside the existing payload redaction check diff --git a/.changeset/eight-buses-jump.md b/.changeset/eight-buses-jump.md deleted file mode 100644 index 304e20e52..000000000 --- a/.changeset/eight-buses-jump.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -'@openfn/ws-worker': patch ---- - -Include dataclip size in sentry reports diff --git a/.changeset/lucky-owls-report.md b/.changeset/lucky-owls-report.md deleted file mode 100644 index 2f03d8faa..000000000 --- a/.changeset/lucky-owls-report.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -'@openfn/ws-worker': patch ---- - -Attribute Sentry reports to the run that produced them diff --git a/.changeset/tame-hats-drum.md b/.changeset/tame-hats-drum.md deleted file mode 100644 index 9b4cb187e..000000000 --- a/.changeset/tame-hats-drum.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -'@openfn/ws-worker': patch ---- - -Capture more diagnostic detail when the connection to Lightning drops unexpectedly diff --git a/.changeset/tidy-plums-obey.md b/.changeset/tidy-plums-obey.md deleted file mode 100644 index 4f6f08c4d..000000000 --- a/.changeset/tidy-plums-obey.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -'@openfn/ws-worker': patch ---- - -Reduce dataclip bloat when sending step results to Lightning diff --git a/packages/cli/CHANGELOG.md b/packages/cli/CHANGELOG.md index f585b61f7..71f3ce76f 100644 --- a/packages/cli/CHANGELOG.md +++ b/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 diff --git a/packages/cli/package.json b/packages/cli/package.json index d816436ed..47f79892f 100644 --- a/packages/cli/package.json +++ b/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", diff --git a/packages/engine-multi/CHANGELOG.md b/packages/engine-multi/CHANGELOG.md index d6c55f360..af1b5ebf1 100644 --- a/packages/engine-multi/CHANGELOG.md +++ b/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 diff --git a/packages/engine-multi/package.json b/packages/engine-multi/package.json index 4e6207eb2..0dbad7f29 100644 --- a/packages/engine-multi/package.json +++ b/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", diff --git a/packages/lexicon/CHANGELOG.md b/packages/lexicon/CHANGELOG.md index 22437b683..eb1ba3dd4 100644 --- a/packages/lexicon/CHANGELOG.md +++ b/packages/lexicon/CHANGELOG.md @@ -1,5 +1,11 @@ # lexicon +## 2.4.2 + +### Patch Changes + +- 7398d41: Support dataclipSize + ## 2.4.1 ### Patch Changes diff --git a/packages/lexicon/package.json b/packages/lexicon/package.json index 04c1de44d..ef4c8c404 100644 --- a/packages/lexicon/package.json +++ b/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 ", "license": "ISC", diff --git a/packages/lightning-mock/CHANGELOG.md b/packages/lightning-mock/CHANGELOG.md index df5a907cd..e2e1d2df8 100644 --- a/packages/lightning-mock/CHANGELOG.md +++ b/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 diff --git a/packages/lightning-mock/package.json b/packages/lightning-mock/package.json index 857fad3dc..c3a507b12 100644 --- a/packages/lightning-mock/package.json +++ b/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", diff --git a/packages/ws-worker/CHANGELOG.md b/packages/ws-worker/CHANGELOG.md index 6f8f92f2d..d60ce117b 100644 --- a/packages/ws-worker/CHANGELOG.md +++ b/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 diff --git a/packages/ws-worker/package.json b/packages/ws-worker/package.json index 2bd5a6534..1708ebd1b 100644 --- a/packages/ws-worker/package.json +++ b/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",