diff --git a/src/persist/activation_repo.lua b/src/persist/activation_repo.lua index 67669a4..626d822 100644 --- a/src/persist/activation_repo.lua +++ b/src/persist/activation_repo.lua @@ -156,7 +156,12 @@ local function normalize_row(row: any) }, nil end -local function lock_workflow_status_tx(tx, dataflow_id) +-- The workflow row is the first lock in every transaction that also mutates +-- activation or wake rows. PostgreSQL foreign-key checks can hold KEY SHARE on +-- this parent row, so acquiring a weaker UPDATE lock and upgrading it later can +-- deadlock with a concurrent commit. Callers that cross the workflow/lifecycle +-- boundary must establish this lock before either side is changed. +function activation_repo.lock_workflow_tx(tx, dataflow_id) local db_type, type_err = tx:db_type() if type_err then return nil, type_err end if db_type ~= sql.type.POSTGRES and db_type ~= "postgres" then @@ -271,7 +276,7 @@ function activation_repo.request_activation_tx(tx, dataflow_id, launch_args, now if not valid then return nil, id_err end valid, id_err = validate_timestamp(now_value, "requested_at") if not valid then return nil, id_err end - local status, status_err = lock_workflow_status_tx(tx, dataflow_id) + local status, status_err = activation_repo.lock_workflow_tx(tx, dataflow_id) if status_err then return nil, status_err end local terminal = terminal_result_from_status(status) if terminal then return terminal, nil end @@ -290,7 +295,7 @@ function activation_repo.activate_for_signal_tx(tx, dataflow_id, wake_key, wake_ valid, validation_err = validate_timestamp(now_value, "requested_at") if not valid then return nil, validation_err end - local status, status_err = lock_workflow_status_tx(tx, dataflow_id) + local status, status_err = activation_repo.lock_workflow_tx(tx, dataflow_id) if status_err then return nil, status_err end local terminal = terminal_result_from_status(status) if terminal then @@ -348,7 +353,7 @@ function activation_repo.activate_due_tx(tx, dataflow_id, wake_key, now_value) valid, validation_err = validate_timestamp(now_value, "now") if not valid then return nil, validation_err end - local status, status_err = lock_workflow_status_tx(tx, dataflow_id) + local status, status_err = activation_repo.lock_workflow_tx(tx, dataflow_id) if status_err then return nil, status_err end local terminal = terminal_result_from_status(status) if terminal then @@ -423,7 +428,7 @@ function activation_repo.release_if_generation_tx(tx, dataflow_id, generation, n valid, validation_err = validate_timestamp(now_value, "updated_at") if not valid then return nil, validation_err end - local status, status_err = lock_workflow_status_tx(tx, dataflow_id) + local status, status_err = activation_repo.lock_workflow_tx(tx, dataflow_id) if status_err then return nil, status_err end local terminal = terminal_result_from_status(status) if terminal then @@ -477,7 +482,7 @@ function activation_repo.claim_epoch_tx( valid, validation_err = validate_timestamp(now_value, "updated_at") if not valid then return nil, validation_err end - local status, status_err = lock_workflow_status_tx(tx, dataflow_id) + local status, status_err = activation_repo.lock_workflow_tx(tx, dataflow_id) if status_err then return nil, status_err end local terminal = terminal_result_from_status(status) if terminal then @@ -515,7 +520,7 @@ function activation_repo.consume_wake_tx(tx, dataflow_id, wake_key, generation) if not valid then return nil, validation_err end if type(wake_key) ~= "string" or wake_key == "" then return nil, "wake_key is required" end - local status, status_err = lock_workflow_status_tx(tx, dataflow_id) + local status, status_err = activation_repo.lock_workflow_tx(tx, dataflow_id) if status_err then return nil, status_err end local terminal = terminal_result_from_status(status) if terminal then @@ -568,7 +573,7 @@ function activation_repo.disable_terminal_tx(tx, dataflow_id, now_value) if not valid then return nil, validation_err end valid, validation_err = validate_timestamp(now_value, "updated_at") if not valid then return nil, validation_err end - local status, status_err = lock_workflow_status_tx(tx, dataflow_id) + local status, status_err = activation_repo.lock_workflow_tx(tx, dataflow_id) if status_err then return nil, status_err end if not TERMINAL_STATUS[status] then return nil, "dataflow is not terminal" end return cleanup_terminal_tx(tx, dataflow_id, status, now_value) diff --git a/src/persist/ops.lua b/src/persist/ops.lua index acf4c70..407a5ad 100644 --- a/src/persist/ops.lua +++ b/src/persist/ops.lua @@ -794,6 +794,24 @@ handlers[constants.COMMAND_TYPES.UPDATE_WORKFLOW] = function(tx, dataflow_id, op local payload = command.payload or {} local wf_id_to_update = payload.dataflow_id or dataflow_id + local terminal = payload.status == constants.STATUS.COMPLETED_SUCCESS or + payload.status == constants.STATUS.COMPLETED_FAILURE or + payload.status == constants.STATUS.CANCELLED or + payload.status == constants.STATUS.TERMINATED + + -- A terminal update crosses from the workflow row into activation and wake + -- rows. Establish the canonical parent-first lock order before UPDATE takes + -- PostgreSQL's weaker NO KEY UPDATE lock; upgrading that lock afterwards can + -- deadlock with a concurrent commit holding a foreign-key KEY SHARE lock. + if terminal then + local _, lock_err = activation_repo.lock_workflow_tx(tx, wf_id_to_update) + if lock_err then + if tostring(lock_err) == "dataflow not found" then + return nil, "Workflow not found or no changes applied" + end + return nil, "Failed to lock workflow lifecycle: " .. tostring(lock_err) + end + end -- Metadata merge configuration - default is merge=true for consistency local merge_metadata = payload.merge_metadata @@ -928,10 +946,6 @@ handlers[constants.COMMAND_TYPES.UPDATE_WORKFLOW] = function(tx, dataflow_id, op return nil, "Workflow not found or no changes applied" end - local terminal = payload.status == constants.STATUS.COMPLETED_SUCCESS or - payload.status == constants.STATUS.COMPLETED_FAILURE or - payload.status == constants.STATUS.CANCELLED or - payload.status == constants.STATUS.TERMINATED if terminal then local _, projection_err = activation_repo.disable_terminal_tx(tx, wf_id_to_update, now_ts) if projection_err then diff --git a/src/persist/ops_test.lua b/src/persist/ops_test.lua index 6ba6364..176cdd2 100644 --- a/src/persist/ops_test.lua +++ b/src/persist/ops_test.lua @@ -992,6 +992,37 @@ local function define_tests() test.is_false(db_bool(activations[1].desired_active)) end) + it("locks the workflow before a terminal status update", function() + local resources = setup_test_resources() + local tx = get_test_transaction() + local observed_statuses = {} + local original_lock = activation_repo.lock_workflow_tx + activation_repo.lock_workflow_tx = function(lock_tx, dataflow_id) + local rows, query_err = txq(lock_tx, + "SELECT status FROM dataflows WHERE dataflow_id = ?", + { dataflow_id }) + if query_err then return nil, query_err end + observed_statuses[#observed_statuses + 1] = rows[1].status + return original_lock(lock_tx, dataflow_id) + end + + local execute_result + local execute_err + local called, call_err = pcall(function() + execute_result, execute_err = ops.execute(tx, resources.dataflow_id, nil, { + type = ops.COMMAND_TYPES.UPDATE_WORKFLOW, + payload = { status = ops.STATUS.CANCELLED }, + }) + end) + activation_repo.lock_workflow_tx = original_lock + if not called then error(call_err) end + + test.is_nil(execute_err) + test.not_nil(execute_result) + test.eq(observed_statuses[1], "active") + test.eq(observed_statuses[2], ops.STATUS.CANCELLED) + end) + it("rejects stale completion after a newer signal activation", function() local resources = setup_test_resources() local tx = get_test_transaction()