[DCP Ingestion] Add preprocessing stage to workflow

[gmechali]

No description provided.

[gemini-code-assist]

Code Review

This pull request updates the ingestion workflow by changing the initial stage to "preprocessing" and adding a step to update the ingestion history status to "RUNNING" for this stage. However, the feedback highlights a critical issue: because these preprocessing steps run outside of a try-catch block, any failure during this stage will leave the workflow permanently stuck in the "RUNNING" state. It is recommended to wrap the preprocessing and handshake steps in a try-catch block to properly handle failures and update the status to "FAILURE".

[gemini-code-assist]

Issue: Missing Error Handling for Preprocessing Stage

The new update_history_preprocessing step sets the ingestion history status to RUNNING for the preprocessing stage. However, the subsequent run_preprocessing step (and other setup steps like read_handshake_file) are executed outside of any try-catch block.

If the preprocessing Cloud Run job fails or if reading the handshake file fails, the workflow will terminate immediately. Because the global except block is only defined inside the process_ingestion step (which is executed much later, after lock acquisition), the workflow will never call update_history to mark the status as FAILURE. This will leave the ingestion permanently stuck in the RUNNING state in the history database and UI.

Recommendation

Wrap the preprocessing and handshake steps in a try-catch block to ensure that any failures during this stage are caught and the history is updated to FAILURE.

For example:

    - preprocessing_stage:
        try:
          steps:
            - update_history_preprocessing:
                call: update_history
                args:
                  workflow_id: '$${workflow_id}'
                  status: 'RUNNING'
                  stage: 'preprocessing'
                  job_id: null
                  import_list: '$${imports_history_list}'

            - run_preprocessing:
                call: launch_and_poll_preprocessing_job
                args:
                  job_name: '$${prep_job_name}'
                  workflow_id: '$${workflow_id}'
                  temp_location: '$${input.tempLocation}'
                  imports_arg: '$${comma_separated_imports}'
                  region: '$${input.region}'
                  project_id: '${project_id}'
                result: prep_res

            - read_handshake_file:
                call: googleapis.storage.v1.objects.get
                args:
                  bucket: '$${bucket_name}'
                  object: '$${text.url_encode(temp_path_prefix + "/datacommons/ingestion_records/" + workflow_id + ".json")}'
                  alt: 'media'
                result: handshake_data

            - update_launch_params:
                assign:
                  - launch_params.importList: '$${handshake_data.importList}'
                  - decoded_imports: '$${json.decode(handshake_data.importList)}'
        except:
          as: e
          steps:
            - update_history_preprocessing_failure:
                call: update_history
                args:
                  workflow_id: '$${workflow_id}'
                  status: 'FAILURE'
                  stage: 'preprocessing'
                  job_id: null
                  import_list: '$${imports_history_list}'
            - raise_preprocessing_error:
                raise: $${e}

[gemini-code-assist]

Initializing current_stage to "preprocessing" is logical, but please note that the except block that uses current_stage to report failures is defined inside the process_ingestion step (lines 173-294). Since process_ingestion is only executed after lock acquisition, any failure during the actual preprocessing stage (which runs before lock acquisition) will not trigger that except block. Therefore, initializing current_stage here does not automatically protect the preprocessing stage from leaving the history in a RUNNING state upon failure.