Suggestion - Don't break Zeebe gRPC connection if invalid variable payload is detected in parseVariablesAndCustomHeadersToJSON

[ryanelee]

This is a suggestion for improve the parseVariablesAndCustomHeadersToJSON method in Zeebe lib.
If an invalid job variable payload is sent in and cannot be parsed out by the method, can we still return the response and don't break the entire gPRC connection?

The benefits are:

• The gRPC connection would not be broken, and the other jobs can be continually pulled up
• The invalid job variable payload can be returned and easily for analyzing

SDK Component

Zeebe gRPC client

Expected Behavior

The dirty job can be handled gracefully without breaking the entire connection

Current Behavior

One invalid job variable will interrupt the entire gRPC connection

Possible Solution

Add a try..catch on parseVariablesAndCustomHeadersToJSON and handle the error gracefully without throw out the breaking error

Change method parseVariablesAndCustomHeadersToJSON from

function parseVariablesAndCustomHeadersToJSON(response,
 inputVariableDto, 
 // eslint-disable-next-line @typescript-eslint/no-explicit-any
 customHeadersDto) {
  return Object.assign({}, response, {
    customHeaders: (0, lib_1.losslessParse)(response.customHeaders, customHeadersDto),
    variables: (0, lib_1.losslessParse)(response.variables, inputVariableDto),
  });
 }

to:

function parseVariablesAndCustomHeadersToJSON(response,
 inputVariableDto, 
 // eslint-disable-next-line @typescript-eslint/no-explicit-any
 customHeadersDto) {
  let res;

    try {
        res = Object.assign({}, response, {
            customHeaders: (0, lib_1.losslessParse)(response.customHeaders, customHeadersDto),
            variables: (0, lib_1.losslessParse)(response.variables, inputVariableDto),
        });
    } catch (error) {
        console.error({message: 'Camunda8/sdk -> Error parsing response:', response, error});
        return response;
    }

    return res;
 }

Steps to Reproduce

1. Add a unit test for stringifyVariables.parseVariablesAndCustomHeadersToJSON

test('parseVariablesAndCustomHeadersToJSON gracefully handle invalid JSON variable payload', () => {
    const invalidJobVariableString = `{"x": 1,"x":2 }`; //duplicate key

    const activateJobs = {
            jobs: [
                {
                    key: '1234567',
                    bpmnProcessId: 'test-1',
                    processDefinitionVersion: 1,
                    processInstanceKey: '1234567',
                    elementId: 'Task_1q0q0m6',
                    variables: invalidJobVariableString,
                    type: '',
                    processDefinitionKey: '',
                    elementInstanceKey: '',
                    customHeaders: JSON.stringify({
                        workerType: 'external',
                    }),
                    worker: '',
                    retries: 0,
                    deadline: '',
                    tenantId: '',
                },
            ],
        };

        const res = parseVariablesAndCustomHeadersToJSON(
            activateJobs.jobs[0],
            null,
            null,
        );
        expect(res.variables).toEqual(invalidJobVariableString);
});

Context (Environment)

This was validated in Camunda self-managed version

[jwulf]

Thanks for this.

Is the scenario that an invalid payload comes from Zeebe?

This is something that I haven't come across, or thought about.

UPDATE: OK, I thought about it and looked at the code.

I see.

We should catch the failure at the call site (rather than inside the function) and fail the job, so that an incident is raised. We can also log an error.

If we catch it inside the function and basically throw the error away (aside from a console log) then whatever happens after that in the worker code is unpredictable.

In principle, the system is in an invalid state due to an undetected bug upstream, so we need to throw. We'll do that with a failure response for the job and an error message logged out locally, and make sure that we don't blow up inside the code consuming the job stream.

So we'll limit the exception to the scope of that job, rather than the stream itself.

I'll implement a solution tomorrow.

[jwulf]

I wrapped the parsing in a Promise, like this:

export function parseVariablesAndCustomHeadersToJSON<Variables, CustomHeaders>(
	response: ActivatedJob,
	inputVariableDto: new (...args: any[]) => Readonly<Variables>,
	customHeadersDto: new (...args: any[]) => Readonly<CustomHeaders>
): Promise<Job<Variables, CustomHeaders>> {
	return new Promise((resolve, reject) => {
		try {
			resolve(
				Object.assign({}, response, {
					customHeaders: losslessParse(
						response.customHeaders,
						customHeadersDto
					) as CustomHeaders,
					variables: losslessParse(
						response.variables,
						inputVariableDto
					) as Readonly<Variables>,
				}) as Job<Variables, CustomHeaders>
			)
		} catch (e) {
			console.error(`Error parsing variables ${e}`)
			console.error('Job', response)
			reject(e)
		}
	})
}

This makes the calling code (it gets called in several places) responsible for resolving the promise to get the value out. And it can catch and fail the job.