Improve deserialization error handling

[loicmathieu]

This PR provides safe deserialization for:

• A WorkerTrigger
• A WorkerTriggerResult
• A WorkerTask
• A Flow

You can QA with a flow that uses a task from a plugin, launching it thousands of times, hard killing Kestra and removing the plugin so you have pending WorkerTask and WorkerTask result to verify that it correctly ends the flow.

You can also use the following instructions to QA features one by one.

To QA an invalid WorkerTrigger, you can use the following INSERT statement, there should be error logs but no stacktraces or crashes.

INSERT INTO queues ("type", "key", value, updated)
VALUES('io.kestra.core.runners.WorkerJob'::"queue_type", 'company.team_hello-world_trigger', '{"type": "trigger", "trigger": {"id": "trigger", "sql": "select CURRENT_TIMESTAMP", "url": "jdbc:postgresql://localhost:5432/kestra", "type": "io.kestra.unknown.Invalid", "password": "k3str4", "username": "kestra"}, "triggerContext": {"date": "2023-12-05T13:29:47.455480928Z", "flowId": "hello-world", "namespace": "company.team", "triggerId": "trigger", "flowRevision": 3}, "conditionContext": {"flow": {"id": "hello-world", "tasks": [{"id": "hello", "type": "io.kestra.core.tasks.log.Log", "message": "Kestra team wishes you a great day! 👋"}], "deleted": false, "disabled": false, "revision": 3, "triggers": [{"id": "trigger", "sql": "select CURRENT_TIMESTAMP", "url": "jdbc:postgresql://localhost:5432/kestra", "type": "io.kestra.plugin.jdbc.postgresql.Trigger", "password": "k3str4", "username": "kestra"}], "namespace": "company.team"}, "runContext": {"variables": {"envs": {"plugins_path": "/home/loic/dev/kestra-plugins"}, "flow": {"id": "hello-world", "revision": 3, "namespace": "company.team"}, "trigger": {"id": "trigger", "type": "io.kestra.plugin.jdbc.postgresql.Trigger"}}}}}'::jsonb, '2023-12-05 13:29:47.942');

To QA an invalid WorkerTriggerResult, you can use the following INSERT statement, there should be error logs but no stacktraces or crashes.

INSERT INTO queues("type", "key", value, updated)
VALUES('io.kestra.core.runners.WorkerTriggerResult'::"queue_type", 'company.team_hello-world_trigger', '{"success": true, "trigger": {"id": "trigger", "sql": "select CURRENT_TIMESTAMP", "url": "jdbc:postgresql://localhost:5432/kestra", "type": "io.kestra.unknown.Invalid", "fetchOne": true, "password": "k3str4", "username": "kestra"}, "execution": {"id": "2xpPLaPFjLsfG6rd5N4sWC", "state": {"current": "CREATED", "duration": 0.000485197, "histories": [{"date": "2023-12-05T13:30:49.110397416Z", "state": "CREATED"}], "startDate": "2023-12-05T13:30:49.110397416Z"}, "flowId": "hello-world", "deleted": false, "trigger": {"id": "trigger", "type": "io.kestra.plugin.jdbc.postgresql.Trigger", "variables": {"row": {"current_timestamp": "2023-12-05T13:30:49.101707Z"}, "size": 1}}, "namespace": "company.team", "originalId": "2xpPLaPFjLsfG6rd5N4sWC", "flowRevision": 4}, "triggerContext": {"date": "2023-12-05T13:30:48.453616306Z", "flowId": "hello-world", "namespace": "company.team", "triggerId": "trigger", "flowRevision": 4}}'::jsonb, '2023-12-05 13:30:49.123');

To QA an invalid WorkerTask, you can use the following INSERT statement, there should be error logs but no stacktraces or crashes.

INSERT INTO public.queues("type", "key", value, updated)
VALUES('io.kestra.core.runners.WorkerJob'::"queue_type", '2fKi81KTdBbHH1FbkvqntW', '{"task": {"id": "hello", "type": "io.kestra.unknown.Invalid", "message": "Kestra team wishes you a great day! 👋"}, "type": "task", "taskRun": {"id": "2fKi81KTdBbHH1FbkvqntW", "state": {"current": "CREATED", "duration": 0.161643766, "histories": [{"date": "2023-12-05T13:16:58.047066195Z", "state": "CREATED"}], "startDate": "2023-12-05T13:16:58.047066195Z"}, "flowId": "hello-world", "taskId": "hello", "namespace": "company.team", "executionId": "5ojop3jqsa32c5YPy8en65"}, "runContext": {"variables": {"envs": {"plugins_path": "/home/loic/dev/kestra-plugins"}, "flow": {"id": "hello-world", "revision": 1, "namespace": "company.team"}, "task": {"id": "hello", "type": "io.kestra.core.tasks.log.Log"}, "taskrun": {"id": "2fKi81KTdBbHH1FbkvqntW", "startDate": "2023-12-05T13:16:58.047066195Z", "attemptsCount": 0}, "execution": {"id": "5ojop3jqsa32c5YPy8en65", "startDate": "2023-12-05T13:16:57.763Z", "originalId": "5ojop3jqsa32c5YPy8en65"}}, "storageOutputPrefix": "///company/team/hello-world/executions/5ojop3jqsa32c5YPy8en65/tasks/hello/2fKi81KTdBbHH1FbkvqntW"}}'::jsonb, '2023-12-05 13:16:58.236');

To QA an invalid Flow, you can use the following INSERT statement, there should be error logs but no stacktraces or crashes.

INSERT INTO queues("type", "key", value, updated)
VALUES('io.kestra.core.models.flows.Flow'::"queue_type", 'company.team_hello-world_1', '{"id": "hello-world", "tasks": [{"id": "hello", "type": "io.kestra.unknown.Invalid", "message": "Kestra team wishes you a great day! 👋"}], "deleted": false, "disabled": false, "revision": 1, "namespace": "company.team"}'::jsonb, '2023-12-05 13:16:54.836');

[loicmathieu]

This PR is in draft because it needs #2601 to be merged before.

[brian-mulier-p]

small changes

[loicmathieu]

@brian-mulier-p I applied all your suggestions.

[brian-mulier-p]

LGTM 👍

[sonarcloud]

Quality Gate failed

Failed conditions

69.8% Coverage on New Code (required ≥ 80%)
5.3% Duplication on New Code (required ≤ 3%)

See analysis details on SonarCloud