Skip to content

fix(persist): order terminal lifecycle locks #52

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wolfy-j wants to merge 1 commit into
base: master
Choose a base branch
from
Open

fix(persist): order terminal lifecycle locks #52

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 13 additions & 8 deletions src/persist/activation_repo.lua
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down
22 changes: 18 additions & 4 deletions src/persist/ops.lua
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
31 changes: 31 additions & 0 deletions src/persist/ops_test.lua
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down