From 3d38cfea3fea11efd4dc8f5ae527b349e0af8758 Mon Sep 17 00:00:00 2001 From: Wolfy-J Date: Fri, 25 Sep 2026 14:08:21 -0400 Subject: [PATCH 1/3] Add durable dataflow admission evidence --- src/activation_evidence_test.lua | 154 +++++++++++++++++++ src/client.lua | 52 +++++++ src/migrations/10_admission_evidence.lua | 65 ++++++++ src/persist/activation_repo.lua | 188 ++++++++++++++++++++++- 4 files changed, 454 insertions(+), 5 deletions(-) create mode 100644 src/activation_evidence_test.lua create mode 100644 src/migrations/10_admission_evidence.lua diff --git a/src/activation_evidence_test.lua b/src/activation_evidence_test.lua new file mode 100644 index 0000000..b989325 --- /dev/null +++ b/src/activation_evidence_test.lua @@ -0,0 +1,154 @@ +local test = require("test") +local sql = require("sql") +local uuid = require("uuid") +local time = require("time") +local activation_repo = require("activation_repo") +local consts = require("dataflow_consts") +local client = require("client") + +local function now() + return time.now():format(time.RFC3339NANO) +end + +local function with_tx(fn) + local db, db_err = sql.get("app:db") + test.is_nil(db_err) + local tx, begin_err = db:begin() + test.is_nil(begin_err) + local value, err = fn(tx) + if err then tx:rollback() else + local committed, commit_err = tx:commit() + test.is_true(committed) + test.is_nil(commit_err) + end + db:release() + return value, err +end + +local function create_flow() + local id = uuid.v7() + local db, db_err = sql.get("app:db") + test.is_nil(db_err) + local _, insert_err = sql.builder.insert("dataflows"):set_map({ + dataflow_id = id, actor_id = "activation-evidence-test", type = "test", + status = "pending", metadata = "{}", created_at = now(), updated_at = now(), + }):run_with(db):exec() + db:release() + test.is_nil(insert_err) + return id +end + +local function delete_flow(id) + local db, db_err = sql.get("app:db") + test.is_nil(db_err) + local _, delete_err = sql.builder.delete("dataflows"): + where("dataflow_id = ?", id):run_with(db):exec() + db:release() + test.is_nil(delete_err) +end + +local function define_tests() + test.describe("Dataflow admission evidence", function() + test.it("exposes absent and durable activation through the public client", function() + local actor_id = "activation-evidence-test" + local api, new_err = client.new({ + security = { + actor = function() return { id = function() return actor_id end } end, + scope = function() return {} end, + }, + dataflow_repo = { + get_by_user = function(id, owner) + if owner ~= actor_id then return nil, "access denied" end + return { dataflow_id = id, actor_id = owner, + actor_context = "test-context", status = consts.STATUS.PENDING }, nil + end, + }, + activation_repo = activation_repo, + commit = { notify_activation = function() return true, nil end }, + }) + test.is_nil(new_err) + local missing_id = uuid.v7() + local missing, missing_err = api:get_activation_evidence(missing_id, "run:missing") + test.is_nil(missing_err) + test.eq(missing.state, "absent") + local absent_activation, absent_err = api:ensure_activation(missing_id, "run:missing") + test.is_nil(absent_err) + test.eq(absent_activation.state, "absent") + local id = create_flow() + local key = "run:" .. uuid.v7() + local created, created_err = api:get_activation_evidence(id, key) + test.is_nil(created_err) + test.eq(created.state, "created") + local first, first_err = api:ensure_activation(id, key) + test.is_nil(first_err) + test.eq(first.state, "activated") + local second, second_err = api:ensure_activation(id, key) + test.is_nil(second_err) + test.eq(second.generation, first.generation) + delete_flow(id) + end) + + test.it("ensures the same admission without advancing generation and rejects another key", function() + local id = create_flow() + local key = "run:" .. uuid.v7() + local first, first_err = activation_repo.ensure_activation(id, key, now()) + test.is_nil(first_err) + test.eq(first.generation, 1) + local second, second_err = activation_repo.ensure_activation(id, key, now()) + test.is_nil(second_err) + test.eq(second.generation, first.generation) + local other, conflict = activation_repo.ensure_activation(id, key .. ":other", now()) + test.is_nil(other) + test.contains(conflict, "CONFLICT") + local wrong_read, read_conflict = activation_repo.get_activation_evidence( + id, key .. ":other") + test.is_nil(wrong_read) + test.contains(read_conflict, "CONFLICT") + local evidence, evidence_err = activation_repo.get_activation_evidence(id, key) + test.is_nil(evidence_err) + test.eq(evidence.generation, 1) + test.is_true(evidence.ever_activated) + delete_flow(id) + end) + + test.it("retains terminal status and outcome until matching acknowledgment", function() + local id = create_flow() + local key = "run:" .. uuid.v7() + local activation, activation_err = activation_repo.ensure_activation(id, key, now()) + test.is_nil(activation_err) + local terminal, terminal_err = with_tx(function(tx) + local db_type, type_err = tx:db_type() + if type_err then return nil, type_err end + local query = "UPDATE dataflows SET status = ?, metadata = ? WHERE dataflow_id = ?" + if db_type == "postgres" then query = "UPDATE dataflows SET status = $1, metadata = $2 WHERE dataflow_id = $3" end + local _, err = tx:execute(query, { consts.STATUS.COMPLETED_SUCCESS, '{"value":42}', id }) + if err then return nil, err end + return activation_repo.disable_terminal_tx(tx, id, now()) + end) + test.is_nil(terminal_err) + test.is_true(terminal.terminal) + local evidence, evidence_err = activation_repo.get_activation_evidence(id, key) + test.is_nil(evidence_err) + test.eq(evidence.state, "terminal") + test.eq(evidence.terminal_generation, activation.generation) + test.eq(evidence.terminal_outcome.value, 42) + test.is_nil(evidence.terminal_ack_at) + local wrong, wrong_err = activation_repo.ack_terminal(id, key, activation.generation + 1, now()) + test.is_nil(wrong) + test.contains(wrong_err, "CONFLICT") + local wrong_key, wrong_key_err = activation_repo.ack_terminal( + id, key .. ":other", activation.generation, now()) + test.is_nil(wrong_key) + test.contains(wrong_key_err, "CONFLICT") + local ack, ack_err = activation_repo.ack_terminal(id, key, activation.generation, now()) + test.is_nil(ack_err) + test.is_true(ack.acknowledged) + local repeat_ack, repeat_err = activation_repo.ack_terminal(id, key, activation.generation, now()) + test.is_nil(repeat_err) + test.eq(repeat_ack.terminal_ack_at, ack.terminal_ack_at) + delete_flow(id) + end) + end) +end + +return { run_tests = test.run_cases(define_tests) } diff --git a/src/client.lua b/src/client.lua index 1b00653..766af26 100644 --- a/src/client.lua +++ b/src/client.lua @@ -6,6 +6,7 @@ local consts = require("dataflow_consts") local function get_default_deps() return { dataflow_repo = require("dataflow_repo"), + activation_repo = require("activation_repo"), commit = require("commit"), data_reader = require("data_reader"), process = process, @@ -414,6 +415,57 @@ function methods:start(dataflow_id, options) return dataflow_id, nil end +-- Admission calls use one durable key for the lifetime of a workflow run. +-- The repository serializes them with ordinary activation and terminal writes. +function methods:ensure_activation(dataflow_id, admission_key) + if type(dataflow_id) ~= "string" or dataflow_id == "" then + return nil, "dataflow_id is required" + end + if type(admission_key) ~= "string" or admission_key == "" then + return nil, "admission_key is required" + end + local current, current_err = self._deps.activation_repo.get_activation_evidence( + dataflow_id, admission_key) + if current_err then return nil, current_err end + if current.state == "absent" then return current, nil end + local workflow, ownership_err = self:_owned_workflow(dataflow_id) + if not workflow then return nil, ownership_err end + if not TERMINAL_STATUS[workflow.status] then + workflow, ownership_err = self:_ensure_workflow_context(workflow) + if not workflow then return nil, ownership_err end + end + local evidence, err = self._deps.activation_repo.ensure_activation( + dataflow_id, admission_key, time.now():format(time.RFC3339NANO)) + if err then return nil, err end + if evidence.state == "activated" or evidence.state == "running" then + local _, notify_err = self._deps.commit.notify_activation(dataflow_id, evidence.generation) + if notify_err then return nil, notify_err end + end + return evidence, nil +end + +function methods:get_activation_evidence(dataflow_id, admission_key) + if type(dataflow_id) ~= "string" or dataflow_id == "" then + return nil, "dataflow_id is required" + end + if type(admission_key) ~= "string" or admission_key == "" then + return nil, "admission_key is required" + end + local evidence, err = self._deps.activation_repo.get_activation_evidence( + dataflow_id, admission_key) + if err or evidence.state == "absent" then return evidence, err end + local workflow, ownership_err = self:_owned_workflow(dataflow_id) + if not workflow then return nil, ownership_err end + return evidence, nil +end + +function methods:ack_terminal(dataflow_id, admission_key, generation) + local workflow, ownership_err = self:_owned_workflow(dataflow_id) + if not workflow then return nil, ownership_err end + return self._deps.activation_repo.ack_terminal( + dataflow_id, admission_key, generation, time.now():format(time.RFC3339NANO)) +end + -- Cancel workflow function methods:cancel(dataflow_id, timeout) if not dataflow_id or dataflow_id == "" then diff --git a/src/migrations/10_admission_evidence.lua b/src/migrations/10_admission_evidence.lua new file mode 100644 index 0000000..68e63f2 --- /dev/null +++ b/src/migrations/10_admission_evidence.lua @@ -0,0 +1,65 @@ +local function run(db, statements) + for _, statement in ipairs(statements) do + local _, err = db:execute(statement) + if err then error(err) end + end +end + +return require("migration").define(function() + migration("Retain dataflow admission and terminal evidence", function() + database("postgres", function() + up(function(db) + run(db, { + "ALTER TABLE dataflow_activations ADD COLUMN admission_key TEXT", + "ALTER TABLE dataflow_activations ADD COLUMN ever_activated BOOLEAN NOT NULL DEFAULT FALSE", + "ALTER TABLE dataflow_activations ADD COLUMN terminal_status TEXT", + "ALTER TABLE dataflow_activations ADD COLUMN terminal_outcome_json TEXT", + "ALTER TABLE dataflow_activations ADD COLUMN terminal_generation BIGINT", + "ALTER TABLE dataflow_activations ADD COLUMN terminal_ack_at TIMESTAMPTZ", + "CREATE UNIQUE INDEX uq_dataflow_activation_admission ON dataflow_activations(dataflow_id,admission_key) WHERE admission_key IS NOT NULL", + "CREATE INDEX idx_dataflow_terminal_ack ON dataflow_activations(terminal_ack_at,dataflow_id) WHERE terminal_status IS NOT NULL", + "UPDATE dataflow_activations SET ever_activated = TRUE WHERE generation > 0", + }) + end) + down(function(db) + run(db, { + "DROP INDEX idx_dataflow_terminal_ack", + "DROP INDEX uq_dataflow_activation_admission", + "ALTER TABLE dataflow_activations DROP COLUMN terminal_ack_at", + "ALTER TABLE dataflow_activations DROP COLUMN terminal_generation", + "ALTER TABLE dataflow_activations DROP COLUMN terminal_outcome_json", + "ALTER TABLE dataflow_activations DROP COLUMN terminal_status", + "ALTER TABLE dataflow_activations DROP COLUMN ever_activated", + "ALTER TABLE dataflow_activations DROP COLUMN admission_key", + }) + end) + end) + database("sqlite", function() + up(function(db) + run(db, { + "ALTER TABLE dataflow_activations ADD COLUMN admission_key TEXT", + "ALTER TABLE dataflow_activations ADD COLUMN ever_activated INTEGER NOT NULL DEFAULT 0 CHECK(ever_activated IN (0,1))", + "ALTER TABLE dataflow_activations ADD COLUMN terminal_status TEXT", + "ALTER TABLE dataflow_activations ADD COLUMN terminal_outcome_json TEXT", + "ALTER TABLE dataflow_activations ADD COLUMN terminal_generation INTEGER", + "ALTER TABLE dataflow_activations ADD COLUMN terminal_ack_at TEXT", + "CREATE UNIQUE INDEX uq_dataflow_activation_admission ON dataflow_activations(dataflow_id,admission_key) WHERE admission_key IS NOT NULL", + "CREATE INDEX idx_dataflow_terminal_ack ON dataflow_activations(terminal_ack_at,dataflow_id) WHERE terminal_status IS NOT NULL", + "UPDATE dataflow_activations SET ever_activated = 1 WHERE generation > 0", + }) + end) + down(function(db) + run(db, { + "DROP INDEX idx_dataflow_terminal_ack", + "DROP INDEX uq_dataflow_activation_admission", + "ALTER TABLE dataflow_activations DROP COLUMN terminal_ack_at", + "ALTER TABLE dataflow_activations DROP COLUMN terminal_generation", + "ALTER TABLE dataflow_activations DROP COLUMN terminal_outcome_json", + "ALTER TABLE dataflow_activations DROP COLUMN terminal_status", + "ALTER TABLE dataflow_activations DROP COLUMN ever_activated", + "ALTER TABLE dataflow_activations DROP COLUMN admission_key", + }) + end) + end) + end) +end) diff --git a/src/persist/activation_repo.lua b/src/persist/activation_repo.lua index 626d822..971f5c6 100644 --- a/src/persist/activation_repo.lua +++ b/src/persist/activation_repo.lua @@ -153,6 +153,12 @@ local function normalize_row(row: any) launch_args = launch_args, requested_at = tostring(row.requested_at), updated_at = tostring(row.updated_at), + admission_key = row.admission_key and tostring(row.admission_key) or nil, + ever_activated = row.ever_activated == true or tonumber(row.ever_activated) == 1, + terminal_status = row.terminal_status and tostring(row.terminal_status) or nil, + terminal_outcome_json = row.terminal_outcome_json, + terminal_generation = row.terminal_generation and tonumber(row.terminal_generation) or nil, + terminal_ack_at = row.terminal_ack_at and tostring(row.terminal_ack_at) or nil, }, nil end @@ -189,7 +195,8 @@ end local function get_tx(tx, dataflow_id) local rows, query_err = tx_query(tx, [[ SELECT dataflow_id, generation, desired_active, owner_epoch, - launch_args, requested_at, updated_at + launch_args, requested_at, updated_at, admission_key, ever_activated, + terminal_status, terminal_outcome_json, terminal_generation, terminal_ack_at FROM dataflow_activations WHERE dataflow_id = ? LIMIT 1 ]], { dataflow_id }) if query_err then return nil, query_err end @@ -207,11 +214,27 @@ end -- owns both durable activation intent and its wake index, so converge them in -- the same transaction before returning the terminal observation. local function cleanup_terminal_tx(tx, dataflow_id, status, now_value) + local flow_rows, flow_err = tx_query(tx, + "SELECT metadata FROM dataflows WHERE dataflow_id = ?", { dataflow_id }) + if flow_err then return nil, "failed to read terminal outcome: " .. tostring(flow_err) end + local outcome = flow_rows and flow_rows[1] and flow_rows[1].metadata or nil + if type(outcome) == "table" then + local encoded, encode_err = json.encode(outcome) + if encode_err then return nil, "failed to encode terminal outcome: " .. tostring(encode_err) end + outcome = encoded + end local activation_result, activation_err = tx_execute(tx, [[ UPDATE dataflow_activations - SET desired_active = ?, launch_args = NULL, updated_at = ? - WHERE dataflow_id = ? AND (desired_active = ? OR launch_args IS NOT NULL) - ]], { false, now_value, dataflow_id, true }) + SET desired_active = ?, launch_args = NULL, updated_at = ?, + terminal_status = CASE WHEN admission_key IS NOT NULL + THEN COALESCE(terminal_status, ?) ELSE terminal_status END, + terminal_outcome_json = CASE WHEN admission_key IS NOT NULL + THEN COALESCE(terminal_outcome_json, ?) ELSE terminal_outcome_json END, + terminal_generation = CASE WHEN admission_key IS NOT NULL + THEN COALESCE(terminal_generation, generation) ELSE terminal_generation END + WHERE dataflow_id = ? AND (desired_active = ? OR launch_args IS NOT NULL + OR (admission_key IS NOT NULL AND terminal_status IS NULL)) + ]], { false, now_value, status, outcome or sql.as.null(), dataflow_id, true }) if activation_err then return nil, "failed to disable terminal activation: " .. tostring(activation_err) end local wake_result, wake_err = tx_execute(tx, @@ -586,7 +609,8 @@ function activation_repo.get(dataflow_id) if db_err then return nil, db_err end local rows, query_err = db_query(db, [[ SELECT dataflow_id, generation, desired_active, owner_epoch, - launch_args, requested_at, updated_at + launch_args, requested_at, updated_at, admission_key, ever_activated, + terminal_status, terminal_outcome_json, terminal_generation, terminal_ack_at FROM dataflow_activations WHERE dataflow_id = ? LIMIT 1 ]], { dataflow_id }) db:release() @@ -594,6 +618,160 @@ function activation_repo.get(dataflow_id) return normalize_row(rows and rows[1] or nil) end +local function admission_key_valid(key) + return type(key) == "string" and key ~= "" +end + +local function evidence_from_row(row, status, key) + if not row then + return { state = TERMINAL_STATUS[status] and "terminal" or "created", + admission_key = key, ever_activated = false, + terminal_status = TERMINAL_STATUS[status] and status or nil }, nil + end + local outcome = row.terminal_outcome_json + if type(outcome) == "string" and outcome ~= "" then + local decoded, err = json.decode(outcome) + if err then return nil, "invalid terminal outcome: " .. tostring(err) end + outcome = decoded + end + local state = "activated" + if TERMINAL_STATUS[status] or row.terminal_status then + state = row.terminal_status and "terminal" or "unknown" + elseif status == consts.STATUS.RUNNING then state = "running" end + return { + state = state, admission_key = key, generation = row.generation, + ever_activated = row.ever_activated, terminal_status = row.terminal_status, + terminal_outcome = outcome, terminal_generation = row.terminal_generation, + terminal_ack_at = row.terminal_ack_at, + }, nil +end + +local function transaction(fn) + local db, db_err = sql.get(consts.APP_DB) + if db_err then return nil, db_err end + local tx, begin_err = db:begin() + if begin_err then db:release(); return nil, begin_err end + local value, operation_err = fn(tx) + if operation_err then tx:rollback(); db:release(); return nil, operation_err end + local committed, commit_err = tx:commit() + if not committed or commit_err then + tx:rollback(); db:release() + return nil, commit_err or "transaction did not commit" + end + db:release() + return value, nil +end + +function activation_repo.ensure_activation(dataflow_id, admission_key, now_value) + local valid, id_err = validate_id(dataflow_id) + if not valid then return nil, id_err end + if not admission_key_valid(admission_key) then return nil, "admission_key is required" end + valid, id_err = validate_timestamp(now_value, "requested_at") + if not valid then return nil, id_err end + return transaction(function(tx) + local status, lock_err = activation_repo.lock_workflow_tx(tx, dataflow_id) + if lock_err == "dataflow not found" then + return { state = "absent", admission_key = admission_key, + ever_activated = false }, nil + end + if lock_err then return nil, lock_err end + local row, row_err = get_tx(tx, dataflow_id) + if row_err then return nil, row_err end + if row and row.admission_key and row.admission_key ~= admission_key then + return nil, "CONFLICT: dataflow has another admission key" + end + if not row and TERMINAL_STATUS[status] then + return evidence_from_row(nil, status, admission_key) + end + if not row then + local result, insert_err = tx_execute(tx, [[ + INSERT INTO dataflow_activations(dataflow_id,generation,desired_active, + owner_epoch,launch_args,requested_at,updated_at,admission_key,ever_activated) + VALUES (?,1,?,NULL,NULL,?,?,?,?) + ]], { dataflow_id, true, now_value, now_value, admission_key, true }) + if insert_err then return nil, insert_err end + if not result or (result.rows_affected or 0) ~= 1 then + return nil, "activation insert made no change" + end + elseif not row.admission_key then + local _, update_err = tx_execute(tx, [[ + UPDATE dataflow_activations + SET admission_key = ?, ever_activated = ?, updated_at = ? + WHERE dataflow_id = ? AND admission_key IS NULL + ]], { admission_key, true, now_value, dataflow_id }) + if update_err then return nil, update_err end + end + if TERMINAL_STATUS[status] then + local _, cleanup_err = cleanup_terminal_tx(tx, dataflow_id, status, now_value) + if cleanup_err then return nil, cleanup_err end + end + row, row_err = get_tx(tx, dataflow_id) + if row_err then return nil, row_err end + return evidence_from_row(row, status, admission_key) + end) +end + +function activation_repo.get_activation_evidence(dataflow_id, admission_key) + local valid, id_err = validate_id(dataflow_id) + if not valid then return nil, id_err end + if not admission_key_valid(admission_key) then return nil, "admission_key is required" end + local db, db_err = sql.get(consts.APP_DB) + if db_err then return nil, db_err end + local rows, query_err = db_query(db, [[ + SELECT d.status, a.dataflow_id, a.generation, a.desired_active, + a.owner_epoch, a.launch_args, a.requested_at, a.updated_at, + a.admission_key, a.ever_activated, a.terminal_status, + a.terminal_outcome_json, a.terminal_generation, a.terminal_ack_at + FROM dataflows d LEFT JOIN dataflow_activations a ON a.dataflow_id = d.dataflow_id + WHERE d.dataflow_id = ? LIMIT 1 + ]], { dataflow_id }) + db:release() + if query_err then return nil, query_err end + local joined = rows and rows[1] or nil + if not joined then return { state = "absent", admission_key = admission_key, + ever_activated = false }, nil end + if joined.admission_key and tostring(joined.admission_key) ~= admission_key then + return nil, "CONFLICT: dataflow has another admission key" + end + local row, row_err = normalize_row(joined.dataflow_id and joined or nil) + if row_err then return nil, row_err end + return evidence_from_row(row, tostring(joined.status), admission_key) +end + +function activation_repo.ack_terminal(dataflow_id, admission_key, generation, now_value) + local valid, id_err = validate_id(dataflow_id) + if not valid then return nil, id_err end + if not admission_key_valid(admission_key) then return nil, "admission_key is required" end + generation = tonumber(generation) + if not generation or generation < 1 or generation % 1 ~= 0 then + return nil, "generation must be a positive integer" + end + valid, id_err = validate_timestamp(now_value, "terminal_ack_at") + if not valid then return nil, id_err end + return transaction(function(tx) + local _, lock_err = activation_repo.lock_workflow_tx(tx, dataflow_id) + if lock_err then return nil, lock_err end + local row, row_err = get_tx(tx, dataflow_id) + if row_err then return nil, row_err end + if not row or row.admission_key ~= admission_key or + row.terminal_generation ~= generation then + return nil, "CONFLICT: terminal admission or generation differs" + end + if not row.terminal_status then return nil, "terminal evidence is unavailable" end + if not row.terminal_ack_at then + local _, update_err = tx_execute(tx, [[ + UPDATE dataflow_activations SET terminal_ack_at = ? + WHERE dataflow_id = ? AND admission_key = ? + AND terminal_generation = ? AND terminal_ack_at IS NULL + ]], { now_value, dataflow_id, admission_key, generation }) + if update_err then return nil, update_err end + row, row_err = get_tx(tx, dataflow_id) + if row_err then return nil, row_err end + end + return { acknowledged = true, terminal_ack_at = row.terminal_ack_at }, nil + end) +end + function activation_repo.list_active() local db, db_err = sql.get(consts.APP_DB) if db_err then return nil, db_err end From 7e3b92d04cd9cf14971b049f23e2c05dbe01c992 Mon Sep 17 00:00:00 2001 From: Wolfy-J Date: Fri, 25 Sep 2026 14:27:27 -0400 Subject: [PATCH 2/3] Retain dataflow evidence across delete and register admission modules --- src/_index.yaml | 19 + src/activation_evidence_test.lua | 58 ++- src/client.lua | 318 +-------------- src/client/_index.yaml | 17 + src/client/admission.lua | 87 ++++ src/client/lifecycle.lua | 275 +++++++++++++ src/migrations/_index.yaml | 11 + src/persist/_index.yaml | 2 + src/persist/activation/_index.yaml | 11 + src/persist/activation/evidence.lua | 175 ++++++++ src/persist/activation/operations.lua | 358 ++++++++++++++++ src/persist/activation_repo.lua | 561 ++------------------------ src/persist/ops.lua | 31 +- 13 files changed, 1083 insertions(+), 840 deletions(-) create mode 100644 src/client/_index.yaml create mode 100644 src/client/admission.lua create mode 100644 src/client/lifecycle.lua create mode 100644 src/persist/activation/_index.yaml create mode 100644 src/persist/activation/evidence.lua create mode 100644 src/persist/activation/operations.lua diff --git a/src/_index.yaml b/src/_index.yaml index 8a248ab..351f39a 100644 --- a/src/_index.yaml +++ b/src/_index.yaml @@ -29,11 +29,30 @@ entries: - security - funcs imports: + activation_repo: userspace.dataflow.persist:activation_repo commit: userspace.dataflow.persist:commit dataflow_consts: userspace.dataflow:consts data_reader: userspace.dataflow.persist:data_reader dataflow_repo: userspace.dataflow.persist:dataflow_repo execution_frame: userspace.dataflow:execution_frame + client_admission: userspace.dataflow.client:admission + client_lifecycle: userspace.dataflow.client:lifecycle + + - name: activation_evidence_test + kind: function.lua + meta: + name: Dataflow Admission Evidence Tests + type: test + group: Workflow + source: file://activation_evidence_test.lua + modules: [sql, uuid, time] + imports: + activation_repo: userspace.dataflow.persist:activation_repo + client: userspace.dataflow:client + dataflow_consts: userspace.dataflow:consts + ops: userspace.dataflow.persist:ops + test: wippy.test:test + method: run_tests # userspace.dataflow:child_output_test - name: child_output_test diff --git a/src/activation_evidence_test.lua b/src/activation_evidence_test.lua index b989325..2f58694 100644 --- a/src/activation_evidence_test.lua +++ b/src/activation_evidence_test.lua @@ -5,6 +5,7 @@ local time = require("time") local activation_repo = require("activation_repo") local consts = require("dataflow_consts") local client = require("client") +local ops = require("ops") local function now() return time.now():format(time.RFC3339NANO) @@ -99,11 +100,11 @@ local function define_tests() test.eq(second.generation, first.generation) local other, conflict = activation_repo.ensure_activation(id, key .. ":other", now()) test.is_nil(other) - test.contains(conflict, "CONFLICT") + test.is_true(errors.is(conflict, errors.CONFLICT)) local wrong_read, read_conflict = activation_repo.get_activation_evidence( id, key .. ":other") test.is_nil(wrong_read) - test.contains(read_conflict, "CONFLICT") + test.is_true(errors.is(read_conflict, errors.CONFLICT)) local evidence, evidence_err = activation_repo.get_activation_evidence(id, key) test.is_nil(evidence_err) test.eq(evidence.generation, 1) @@ -116,6 +117,10 @@ local function define_tests() local key = "run:" .. uuid.v7() local activation, activation_err = activation_repo.ensure_activation(id, key, now()) test.is_nil(activation_err) + local premature_ack, premature_err = activation_repo.ack_terminal( + id, key, activation.generation, now()) + test.is_nil(premature_ack) + test.is_true(errors.is(premature_err, errors.UNAVAILABLE)) local terminal, terminal_err = with_tx(function(tx) local db_type, type_err = tx:db_type() if type_err then return nil, type_err end @@ -135,11 +140,11 @@ local function define_tests() test.is_nil(evidence.terminal_ack_at) local wrong, wrong_err = activation_repo.ack_terminal(id, key, activation.generation + 1, now()) test.is_nil(wrong) - test.contains(wrong_err, "CONFLICT") + test.is_true(errors.is(wrong_err, errors.CONFLICT)) local wrong_key, wrong_key_err = activation_repo.ack_terminal( id, key .. ":other", activation.generation, now()) test.is_nil(wrong_key) - test.contains(wrong_key_err, "CONFLICT") + test.is_true(errors.is(wrong_key_err, errors.CONFLICT)) local ack, ack_err = activation_repo.ack_terminal(id, key, activation.generation, now()) test.is_nil(ack_err) test.is_true(ack.acknowledged) @@ -148,6 +153,51 @@ local function define_tests() test.eq(repeat_ack.terminal_ack_at, ack.terminal_ack_at) delete_flow(id) end) + + test.it("fences DELETE_WORKFLOW until terminal evidence is acknowledged", function() + local id = create_flow() + local key = "run:" .. uuid.v7() + local activation, activation_err = activation_repo.ensure_activation(id, key, now()) + test.is_nil(activation_err) + local terminal, terminal_err = with_tx(function(tx) + local db_type, type_err = tx:db_type() + if type_err then return nil, type_err end + local query = "UPDATE dataflows SET status = ?, metadata = ? WHERE dataflow_id = ?" + if db_type == "postgres" then + query = "UPDATE dataflows SET status = $1, metadata = $2 WHERE dataflow_id = $3" + end + local _, update_err = tx:execute(query, + { consts.STATUS.COMPLETED_SUCCESS, '{"value":42}', id }) + if update_err then return nil, update_err end + return activation_repo.disable_terminal_tx(tx, id, now()) + end) + test.is_nil(terminal_err) + test.is_true(terminal.terminal) + + local command = { type = consts.COMMAND_TYPES.DELETE_WORKFLOW, payload = {} } + local deleted, delete_err = with_tx(function(tx) + return ops.execute(tx, id, nil, command) + end) + test.is_nil(deleted) + test.is_true(errors.is(delete_err, errors.CONFLICT)) + local evidence, evidence_err = activation_repo.get_activation_evidence(id, key) + test.is_nil(evidence_err) + test.eq(evidence.state, "terminal") + test.eq(evidence.terminal_outcome.value, 42) + test.is_nil(evidence.terminal_ack_at) + + local ack, ack_err = activation_repo.ack_terminal(id, key, activation.generation, now()) + test.is_nil(ack_err) + test.is_true(ack.acknowledged) + deleted, delete_err = with_tx(function(tx) + return ops.execute(tx, id, nil, command) + end) + test.is_nil(delete_err) + test.is_true(deleted.results[1].deleted) + local absent, absent_err = activation_repo.get_activation_evidence(id, key) + test.is_nil(absent_err) + test.eq(absent.state, "absent") + end) end) end diff --git a/src/client.lua b/src/client.lua index 766af26..7d76aae 100644 --- a/src/client.lua +++ b/src/client.lua @@ -17,7 +17,7 @@ local function get_default_deps() end local client = {} -local methods = {} +local methods: any = {} local mt = { __index = methods } local TERMINAL_STATUS = { @@ -415,317 +415,7 @@ function methods:start(dataflow_id, options) return dataflow_id, nil end --- Admission calls use one durable key for the lifetime of a workflow run. --- The repository serializes them with ordinary activation and terminal writes. -function methods:ensure_activation(dataflow_id, admission_key) - if type(dataflow_id) ~= "string" or dataflow_id == "" then - return nil, "dataflow_id is required" - end - if type(admission_key) ~= "string" or admission_key == "" then - return nil, "admission_key is required" - end - local current, current_err = self._deps.activation_repo.get_activation_evidence( - dataflow_id, admission_key) - if current_err then return nil, current_err end - if current.state == "absent" then return current, nil end - local workflow, ownership_err = self:_owned_workflow(dataflow_id) - if not workflow then return nil, ownership_err end - if not TERMINAL_STATUS[workflow.status] then - workflow, ownership_err = self:_ensure_workflow_context(workflow) - if not workflow then return nil, ownership_err end - end - local evidence, err = self._deps.activation_repo.ensure_activation( - dataflow_id, admission_key, time.now():format(time.RFC3339NANO)) - if err then return nil, err end - if evidence.state == "activated" or evidence.state == "running" then - local _, notify_err = self._deps.commit.notify_activation(dataflow_id, evidence.generation) - if notify_err then return nil, notify_err end - end - return evidence, nil -end - -function methods:get_activation_evidence(dataflow_id, admission_key) - if type(dataflow_id) ~= "string" or dataflow_id == "" then - return nil, "dataflow_id is required" - end - if type(admission_key) ~= "string" or admission_key == "" then - return nil, "admission_key is required" - end - local evidence, err = self._deps.activation_repo.get_activation_evidence( - dataflow_id, admission_key) - if err or evidence.state == "absent" then return evidence, err end - local workflow, ownership_err = self:_owned_workflow(dataflow_id) - if not workflow then return nil, ownership_err end - return evidence, nil -end - -function methods:ack_terminal(dataflow_id, admission_key, generation) - local workflow, ownership_err = self:_owned_workflow(dataflow_id) - if not workflow then return nil, ownership_err end - return self._deps.activation_repo.ack_terminal( - dataflow_id, admission_key, generation, time.now():format(time.RFC3339NANO)) -end - --- Cancel workflow -function methods:cancel(dataflow_id, timeout) - if not dataflow_id or dataflow_id == "" then - return false, "Workflow ID is required" - end - - timeout = timeout or "30s" - - -- Verify workflow exists and user has access - local workflow, err = self._deps.dataflow_repo.get_by_user(dataflow_id, self._actor_id) - if err then - return false, err - end - - if not workflow then - return false, "Workflow not found" - end - - -- Check if workflow can be cancelled - local cancellable_states = { - [consts.STATUS.PENDING] = true, - [consts.STATUS.RUNNING] = true, - [consts.STATUS.WAITING] = true - } - - if not cancellable_states[workflow.status] then - return false, "Workflow cannot be cancelled in current state: " .. workflow.status - end - - -- Persist the business outcome before touching the runtime process. A - -- process CANCEL is also used during application shutdown and therefore - -- cannot itself carry cancellation semantics. - local _result, update_err = self._deps.commit.execute(dataflow_id, uuid.v7(), { - { - type = consts.COMMAND_TYPES.UPDATE_WORKFLOW, - payload = { - status = consts.STATUS.CANCELLED, - metadata = { - cancelled_at = time.now():format(time.RFC3339), - cancelled_by = self._actor_id, - }, - }, - }, - }) - if update_err then return false, "Failed to cancel workflow: " .. update_err end - - local pid = self._deps.process.registry.lookup("dataflow." .. dataflow_id) - local process_cancelled = false - local cancel_error = nil - if pid then - local success, cancel_err = self._deps.process.cancel(pid, timeout) - process_cancelled = success == true - cancel_error = success and nil or tostring(cancel_err or "unknown error") - end - - return true, nil, { - dataflow_id = dataflow_id, - timeout = timeout, - process_cancelled = process_cancelled, - status_updated = true, - cancel_error = cancel_error, - message = pid and "Workflow cancelled; runtime stop requested" or - "Workflow cancelled without a live process", - } -end - --- Terminate workflow -function methods:terminate(dataflow_id) - if not dataflow_id or dataflow_id == "" then - return false, "Workflow ID is required" - end - - -- Verify workflow exists and user has access - local workflow, err = self._deps.dataflow_repo.get_by_user(dataflow_id, self._actor_id) - if err then - return false, err - end - - if not workflow then - return false, "Workflow not found" - end - - -- Check if workflow is already finished - local finished_states = { - [consts.STATUS.COMPLETED_SUCCESS] = true, - [consts.STATUS.COMPLETED_FAILURE] = true, - [consts.STATUS.CANCELLED] = true, - [consts.STATUS.TERMINATED] = true - } - - if finished_states[workflow.status] then - return false, "Workflow already finished with status: " .. workflow.status - end - - local info = { - dataflow_id = dataflow_id, - process_terminated = false, - status_updated = false - } - - -- Persist the terminal state first so an EXIT can never race the overseer - -- into projecting an operational failure over an administrative outcome. - local update_commands = { - { - type = consts.COMMAND_TYPES.UPDATE_WORKFLOW, - payload = { - status = consts.STATUS.TERMINATED, - metadata = { - terminated_at = time.now():format(time.RFC3339), - terminated_by = self._actor_id - } - } - } - } - - local result, update_err = self._deps.commit.execute(dataflow_id, uuid.v7(), update_commands) - if update_err then - return false, "Failed to update workflow status: " .. update_err, info - end - - info.status_updated = true - - local pid = self._deps.process.registry.lookup("dataflow." .. dataflow_id) - if pid then - local terminate_success, terminate_err = self._deps.process.terminate(pid) - if terminate_success then - info.process_terminated = true - else - info.terminate_error = terminate_err - end - end - return true, nil, info -end - --- Get workflow status -function methods:get_status(dataflow_id) - if not dataflow_id or dataflow_id == "" then - return nil, "Workflow ID is required" - end - - -- Get workflow with actor verification - local workflow, err = self._deps.dataflow_repo.get_by_user(dataflow_id, self._actor_id) - if err then - return nil, err - end - - if not workflow then - return nil, "Workflow not found" - end - - return workflow.status, nil -end - --- Send a signal to a waiting signal node. commit.submit atomically persists the --- signal activation and owns the post-commit overseer notification. -function methods:signal(dataflow_id, signal_id, data) - if not dataflow_id or dataflow_id == "" then - return nil, "Workflow ID is required" - end - if not signal_id or signal_id == "" then - return nil, "Signal ID is required" - end - - local workflow, get_err = self:_owned_workflow(dataflow_id) - if get_err or not workflow then - return nil, errors.new({ - message = "Cannot signal workflow: " .. (get_err or "not found"), - kind = "WorkflowNotFound", - details = { dataflow_id = dataflow_id } - }) - end - if TERMINAL_STATUS[workflow.status] then - return nil, errors.new({ - message = "Cannot signal workflow in terminal state: " .. tostring(workflow.status), - kind = "WorkflowTerminal", - details = { dataflow_id = dataflow_id, status = workflow.status } - }) - end - workflow, get_err = self:_ensure_workflow_context(workflow) - if not workflow then return nil, "Cannot signal workflow: " .. tostring(get_err) end - - -- 1. Write signal commit to outbox (durable, survives crashes) - local op_id = uuid.v7() - local result, err = self._deps.commit.submit(dataflow_id, op_id, { - { - type = consts.COMMAND_TYPES.CREATE_DATA, - payload = { - data_id = uuid.v7(), - data_type = consts.DATA_TYPE.NODE_SIGNAL, - content = data or {}, - content_type = consts.CONTENT_TYPE.JSON, - key = signal_id, - } - } - }) - - if err then - return nil, "Failed to send signal: " .. tostring(err) - end - - return result, nil -end - --- Ensure the desired activation exists and notify the overseer. A registered --- orchestrator PID remains observable for compatibility; a newly accepted --- activation is explicitly pending and is never represented as a PID. -function methods:revive(dataflow_id) - if not dataflow_id or dataflow_id == "" then - return nil, "Workflow ID is required" - end - local workflow, ownership_err = self:_owned_workflow(dataflow_id) - if not workflow then return nil, "Failed to authorize workflow revival: " .. tostring(ownership_err) end - if TERMINAL_STATUS[workflow.status] then - return nil, nil, { - accepted = false, - pending = false, - terminal = true, - status = workflow.status, - spawned = false, - } - end - - local pid = self._deps.process.registry.lookup("dataflow." .. dataflow_id) - if pid then - return pid, nil, { - accepted = true, - pending = false, - existing = true, - spawned = false, - } - end - - workflow, ownership_err = self:_ensure_workflow_context(workflow) - if not workflow then return nil, "Failed to prepare workflow revival: " .. tostring(ownership_err) end - - local activation, activation_err = self._deps.commit.request_activation(dataflow_id, { - dataflow_id = dataflow_id - }) - if activation_err then - return nil, activation_err - end - if type(activation) ~= "table" then - return nil, "invalid activation result" - end - if activation.terminal == true then - return nil, nil, { - accepted = false, - pending = false, - terminal = true, - status = activation.status, - spawned = false, - } - end - return nil, nil, { - accepted = true, - pending = true, - dataflow_id = dataflow_id, - generation = activation.generation, - spawned = false, - } -end +require("client_admission")(methods) +require("client_lifecycle")(methods) -return client +return client :: any diff --git a/src/client/_index.yaml b/src/client/_index.yaml new file mode 100644 index 0000000..09edfb6 --- /dev/null +++ b/src/client/_index.yaml @@ -0,0 +1,17 @@ +version: "1.0" +namespace: userspace.dataflow.client + +entries: + - name: admission + kind: library.lua + source: file://admission.lua + modules: [time] + imports: + dataflow_consts: userspace.dataflow:consts + + - name: lifecycle + kind: library.lua + source: file://lifecycle.lua + modules: [uuid, time] + imports: + dataflow_consts: userspace.dataflow:consts diff --git a/src/client/admission.lua b/src/client/admission.lua new file mode 100644 index 0000000..f693d41 --- /dev/null +++ b/src/client/admission.lua @@ -0,0 +1,87 @@ +local time = require("time") +local consts = require("dataflow_consts") + +return function(methods) + local function typed(kind, message: any) + return errors.new({ kind = kind, message = tostring(message) }) + end + local function owner_error(err) + local message = tostring(err or "workflow not found") + if message:find("access denied", 1, true) or + message:find("actor differs", 1, true) then + return typed(errors.PERMISSION_DENIED, message) + end + return typed(errors.NOT_FOUND, message) + end + local function available_error(err) + if type(err) == "userdata" then return err end + return typed(errors.UNAVAILABLE, err or "operation unavailable") + end + local TERMINAL_STATUS = { + [consts.STATUS.COMPLETED_SUCCESS] = true, + [consts.STATUS.COMPLETED_FAILURE] = true, + [consts.STATUS.CANCELLED] = true, + [consts.STATUS.TERMINATED] = true, + } + +-- Admission calls use one durable key for the lifetime of a workflow run. +-- The repository serializes them with ordinary activation and terminal writes. +function methods:ensure_activation(dataflow_id, admission_key) + if type(dataflow_id) ~= "string" or dataflow_id == "" then + return nil, typed(errors.INVALID, "dataflow_id is required") + end + if type(admission_key) ~= "string" or admission_key == "" then + return nil, typed(errors.INVALID, "admission_key is required") + end + local current, current_err = self._deps.activation_repo.get_activation_evidence( + dataflow_id, admission_key) + if current_err then return nil, current_err end + if current.state == "absent" then return current, nil end + local workflow, ownership_err = self:_owned_workflow(dataflow_id) + if not workflow then return nil, owner_error(ownership_err) end + if not TERMINAL_STATUS[workflow.status] then + workflow, ownership_err = self:_ensure_workflow_context(workflow) + if not workflow then return nil, available_error(ownership_err) end + end + local evidence, err = self._deps.activation_repo.ensure_activation( + dataflow_id, admission_key, time.now():format(time.RFC3339NANO)) + if err then return nil, err end + if evidence.state == "activated" or evidence.state == "running" then + local _, notify_err = self._deps.commit.notify_activation(dataflow_id, evidence.generation) + if notify_err then return nil, available_error(notify_err) end + end + return evidence, nil +end + +function methods:get_activation_evidence(dataflow_id, admission_key) + if type(dataflow_id) ~= "string" or dataflow_id == "" then + return nil, typed(errors.INVALID, "dataflow_id is required") + end + if type(admission_key) ~= "string" or admission_key == "" then + return nil, typed(errors.INVALID, "admission_key is required") + end + local evidence, err = self._deps.activation_repo.get_activation_evidence( + dataflow_id, admission_key) + if err or evidence.state == "absent" then return evidence, err end + local workflow, ownership_err = self:_owned_workflow(dataflow_id) + if not workflow then return nil, owner_error(ownership_err) end + return evidence, nil +end + +function methods:ack_terminal(dataflow_id, admission_key, generation) + if type(dataflow_id) ~= "string" or dataflow_id == "" then + return nil, typed(errors.INVALID, "dataflow_id is required") + end + if type(admission_key) ~= "string" or admission_key == "" then + return nil, typed(errors.INVALID, "admission_key is required") + end + if type(generation) ~= "number" or generation < 1 or generation % 1 ~= 0 then + return nil, typed(errors.INVALID, "generation must be a positive integer") + end + local workflow, ownership_err = self:_owned_workflow(dataflow_id) + if not workflow then return nil, owner_error(ownership_err) end + return self._deps.activation_repo.ack_terminal( + dataflow_id, admission_key, generation, time.now():format(time.RFC3339NANO)) +end + +end diff --git a/src/client/lifecycle.lua b/src/client/lifecycle.lua new file mode 100644 index 0000000..c0e15cc --- /dev/null +++ b/src/client/lifecycle.lua @@ -0,0 +1,275 @@ +local uuid = require("uuid") +local time = require("time") +local consts = require("dataflow_consts") + +return function(methods) + local TERMINAL_STATUS = { + [consts.STATUS.COMPLETED_SUCCESS] = true, + [consts.STATUS.COMPLETED_FAILURE] = true, + [consts.STATUS.CANCELLED] = true, + [consts.STATUS.TERMINATED] = true, + } + +-- Cancel workflow +function methods:cancel(dataflow_id, timeout) + if not dataflow_id or dataflow_id == "" then + return false, "Workflow ID is required" + end + + timeout = timeout or "30s" + + -- Verify workflow exists and user has access + local workflow, err = self._deps.dataflow_repo.get_by_user(dataflow_id, self._actor_id) + if err then + return false, err + end + + if not workflow then + return false, "Workflow not found" + end + + -- Check if workflow can be cancelled + local cancellable_states = { + [consts.STATUS.PENDING] = true, + [consts.STATUS.RUNNING] = true, + [consts.STATUS.WAITING] = true + } + + if not cancellable_states[workflow.status] then + return false, "Workflow cannot be cancelled in current state: " .. workflow.status + end + + -- Persist the business outcome before touching the runtime process. A + -- process CANCEL is also used during application shutdown and therefore + -- cannot itself carry cancellation semantics. + local _result, update_err = self._deps.commit.execute(dataflow_id, uuid.v7(), { + { + type = consts.COMMAND_TYPES.UPDATE_WORKFLOW, + payload = { + status = consts.STATUS.CANCELLED, + metadata = { + cancelled_at = time.now():format(time.RFC3339), + cancelled_by = self._actor_id, + }, + }, + }, + }) + if update_err then return false, "Failed to cancel workflow: " .. update_err end + + local pid = self._deps.process.registry.lookup("dataflow." .. dataflow_id) + local process_cancelled = false + local cancel_error = nil + if pid then + local success, cancel_err = self._deps.process.cancel(pid, timeout) + process_cancelled = success == true + cancel_error = success and nil or tostring(cancel_err or "unknown error") + end + + return true, nil, { + dataflow_id = dataflow_id, + timeout = timeout, + process_cancelled = process_cancelled, + status_updated = true, + cancel_error = cancel_error, + message = pid and "Workflow cancelled; runtime stop requested" or + "Workflow cancelled without a live process", + } +end + +-- Terminate workflow +function methods:terminate(dataflow_id) + if not dataflow_id or dataflow_id == "" then + return false, "Workflow ID is required" + end + + -- Verify workflow exists and user has access + local workflow, err = self._deps.dataflow_repo.get_by_user(dataflow_id, self._actor_id) + if err then + return false, err + end + + if not workflow then + return false, "Workflow not found" + end + + -- Check if workflow is already finished + local finished_states = { + [consts.STATUS.COMPLETED_SUCCESS] = true, + [consts.STATUS.COMPLETED_FAILURE] = true, + [consts.STATUS.CANCELLED] = true, + [consts.STATUS.TERMINATED] = true + } + + if finished_states[workflow.status] then + return false, "Workflow already finished with status: " .. workflow.status + end + + local info = { + dataflow_id = dataflow_id, + process_terminated = false, + status_updated = false + } + + -- Persist the terminal state first so an EXIT can never race the overseer + -- into projecting an operational failure over an administrative outcome. + local update_commands = { + { + type = consts.COMMAND_TYPES.UPDATE_WORKFLOW, + payload = { + status = consts.STATUS.TERMINATED, + metadata = { + terminated_at = time.now():format(time.RFC3339), + terminated_by = self._actor_id + } + } + } + } + + local result, update_err = self._deps.commit.execute(dataflow_id, uuid.v7(), update_commands) + if update_err then + return false, "Failed to update workflow status: " .. update_err, info + end + + info.status_updated = true + + local pid = self._deps.process.registry.lookup("dataflow." .. dataflow_id) + if pid then + local terminate_success, terminate_err = self._deps.process.terminate(pid) + if terminate_success then + info.process_terminated = true + else + info.terminate_error = terminate_err + end + end + return true, nil, info +end + +-- Get workflow status +function methods:get_status(dataflow_id) + if not dataflow_id or dataflow_id == "" then + return nil, "Workflow ID is required" + end + + -- Get workflow with actor verification + local workflow, err = self._deps.dataflow_repo.get_by_user(dataflow_id, self._actor_id) + if err then + return nil, err + end + + if not workflow then + return nil, "Workflow not found" + end + + return workflow.status, nil +end + +-- Send a signal to a waiting signal node. commit.submit atomically persists the +-- signal activation and owns the post-commit overseer notification. +function methods:signal(dataflow_id, signal_id, data) + if not dataflow_id or dataflow_id == "" then + return nil, "Workflow ID is required" + end + if not signal_id or signal_id == "" then + return nil, "Signal ID is required" + end + + local workflow, get_err = self:_owned_workflow(dataflow_id) + if get_err or not workflow then + return nil, errors.new({ + message = "Cannot signal workflow: " .. (get_err or "not found"), + kind = "WorkflowNotFound", + details = { dataflow_id = dataflow_id } + }) + end + if TERMINAL_STATUS[workflow.status] then + return nil, errors.new({ + message = "Cannot signal workflow in terminal state: " .. tostring(workflow.status), + kind = "WorkflowTerminal", + details = { dataflow_id = dataflow_id, status = workflow.status } + }) + end + workflow, get_err = self:_ensure_workflow_context(workflow) + if not workflow then return nil, "Cannot signal workflow: " .. tostring(get_err) end + + -- 1. Write signal commit to outbox (durable, survives crashes) + local op_id = uuid.v7() + local result, err = self._deps.commit.submit(dataflow_id, op_id, { + { + type = consts.COMMAND_TYPES.CREATE_DATA, + payload = { + data_id = uuid.v7(), + data_type = consts.DATA_TYPE.NODE_SIGNAL, + content = data or {}, + content_type = consts.CONTENT_TYPE.JSON, + key = signal_id, + } + } + }) + + if err then + return nil, "Failed to send signal: " .. tostring(err) + end + + return result, nil +end + +-- Ensure the desired activation exists and notify the overseer. A registered +-- orchestrator PID remains observable for compatibility; a newly accepted +-- activation is explicitly pending and is never represented as a PID. +function methods:revive(dataflow_id) + if not dataflow_id or dataflow_id == "" then + return nil, "Workflow ID is required" + end + local workflow, ownership_err = self:_owned_workflow(dataflow_id) + if not workflow then return nil, "Failed to authorize workflow revival: " .. tostring(ownership_err) end + if TERMINAL_STATUS[workflow.status] then + return nil, nil, { + accepted = false, + pending = false, + terminal = true, + status = workflow.status, + spawned = false, + } + end + + local pid = self._deps.process.registry.lookup("dataflow." .. dataflow_id) + if pid then + return pid, nil, { + accepted = true, + pending = false, + existing = true, + spawned = false, + } + end + + workflow, ownership_err = self:_ensure_workflow_context(workflow) + if not workflow then return nil, "Failed to prepare workflow revival: " .. tostring(ownership_err) end + + local activation, activation_err = self._deps.commit.request_activation(dataflow_id, { + dataflow_id = dataflow_id + }) + if activation_err then + return nil, activation_err + end + if type(activation) ~= "table" then + return nil, "invalid activation result" + end + if activation.terminal == true then + return nil, nil, { + accepted = false, + pending = false, + terminal = true, + status = activation.status, + spawned = false, + } + end + return nil, nil, { + accepted = true, + pending = true, + dataflow_id = dataflow_id, + generation = activation.generation, + spawned = false, + } +end + +end diff --git a/src/migrations/_index.yaml b/src/migrations/_index.yaml index 1fc6aed..341bd6d 100644 --- a/src/migrations/_index.yaml +++ b/src/migrations/_index.yaml @@ -198,3 +198,14 @@ entries: imports: migration: wippy.migration:migration method: migrate + + - name: 10_admission_evidence + kind: function.lua + meta: + type: migration + target_db: app:db + depends_on: [ns:wippy.migration] + timestamp: "2026-09-25T17:00:05Z" + source: file://10_admission_evidence.lua + imports: {migration: wippy.migration:migration} + method: migrate diff --git a/src/persist/_index.yaml b/src/persist/_index.yaml index 5eb5dad..dfb9e8c 100644 --- a/src/persist/_index.yaml +++ b/src/persist/_index.yaml @@ -10,6 +10,8 @@ entries: source: file://activation_repo.lua modules: [sql, json] imports: + activation_operations: userspace.dataflow.persist.activation:operations + activation_evidence: userspace.dataflow.persist.activation:evidence dataflow_consts: userspace.dataflow:consts - name: activation_repo_test diff --git a/src/persist/activation/_index.yaml b/src/persist/activation/_index.yaml new file mode 100644 index 0000000..b309150 --- /dev/null +++ b/src/persist/activation/_index.yaml @@ -0,0 +1,11 @@ +version: "1.0" +namespace: userspace.dataflow.persist.activation + +entries: + - name: operations + kind: library.lua + source: file://operations.lua + + - name: evidence + kind: library.lua + source: file://evidence.lua diff --git a/src/persist/activation/evidence.lua b/src/persist/activation/evidence.lua new file mode 100644 index 0000000..611ab17 --- /dev/null +++ b/src/persist/activation/evidence.lua @@ -0,0 +1,175 @@ +return function(activation_repo, shared) + local sql = shared.sql + local json = shared.json + local consts = shared.consts + local TERMINAL_STATUS = shared.TERMINAL_STATUS + local TERMINAL_VALUES = shared.TERMINAL_VALUES + local tx_query = shared.tx_query + local tx_execute = shared.tx_execute + local db_query = shared.db_query + local validate_id = shared.validate_id + local validate_timestamp = shared.validate_timestamp + local normalize_row = shared.normalize_row + local get_tx = shared.get_tx + local cleanup_terminal_tx = shared.cleanup_terminal_tx + local encode_launch_args = shared.encode_launch_args + local rebind = shared.rebind + local typed = shared.typed + +local function admission_key_valid(key) + return type(key) == "string" and key ~= "" +end + +local function evidence_from_row(row, status, key) + if not row then + return { state = TERMINAL_STATUS[status] and "terminal" or "created", + admission_key = key, ever_activated = false, + terminal_status = TERMINAL_STATUS[status] and status or nil }, nil + end + local outcome = row.terminal_outcome_json + if type(outcome) == "string" and outcome ~= "" then + local decoded, err = json.decode(outcome) + if err then return nil, typed(errors.INTERNAL, "invalid terminal outcome: " .. tostring(err)) end + outcome = decoded + end + local state = "activated" + if TERMINAL_STATUS[status] or row.terminal_status then + state = row.terminal_status and "terminal" or "unknown" + elseif status == consts.STATUS.RUNNING then state = "running" end + return { + state = state, admission_key = key, generation = row.generation, + ever_activated = row.ever_activated, terminal_status = row.terminal_status, + terminal_outcome = outcome, terminal_generation = row.terminal_generation, + terminal_ack_at = row.terminal_ack_at, + }, nil +end + +local function transaction(fn) + local db, db_err = sql.get(consts.APP_DB) + if db_err then return nil, db_err end + local tx, begin_err = db:begin() + if begin_err then db:release(); return nil, begin_err end + local value, operation_err = fn(tx) + if operation_err then tx:rollback(); db:release(); return nil, operation_err end + local committed, commit_err = tx:commit() + if not committed or commit_err then + tx:rollback(); db:release() + return nil, commit_err or typed(errors.UNAVAILABLE, "transaction did not commit") + end + db:release() + return value, nil +end + +function activation_repo.ensure_activation(dataflow_id, admission_key, now_value) + local valid, id_err = validate_id(dataflow_id) + if not valid then return nil, id_err end + if not admission_key_valid(admission_key) then return nil, typed(errors.INVALID, "admission_key is required") end + valid, id_err = validate_timestamp(now_value, "requested_at") + if not valid then return nil, id_err end + return transaction(function(tx) + local status, lock_err = activation_repo.lock_workflow_tx(tx, dataflow_id) + if lock_err and errors.is(lock_err, errors.NOT_FOUND) then + return { state = "absent", admission_key = admission_key, + ever_activated = false }, nil + end + if lock_err then return nil, lock_err end + local row, row_err = get_tx(tx, dataflow_id) + if row_err then return nil, row_err end + if row and row.admission_key and row.admission_key ~= admission_key then + return nil, typed(errors.CONFLICT, "dataflow has another admission key") + end + if not row and TERMINAL_STATUS[status] then + return evidence_from_row(nil, status, admission_key) + end + if not row then + local result, insert_err = tx_execute(tx, [[ + INSERT INTO dataflow_activations(dataflow_id,generation,desired_active, + owner_epoch,launch_args,requested_at,updated_at,admission_key,ever_activated) + VALUES (?,1,?,NULL,NULL,?,?,?,?) + ]], { dataflow_id, true, now_value, now_value, admission_key, true }) + if insert_err then return nil, insert_err end + if not result or (result.rows_affected or 0) ~= 1 then + return nil, typed(errors.INTERNAL, "activation insert made no change") + end + elseif not row.admission_key then + local _, update_err = tx_execute(tx, [[ + UPDATE dataflow_activations + SET admission_key = ?, ever_activated = ?, updated_at = ? + WHERE dataflow_id = ? AND admission_key IS NULL + ]], { admission_key, true, now_value, dataflow_id }) + if update_err then return nil, update_err end + end + if TERMINAL_STATUS[status] then + local _, cleanup_err = cleanup_terminal_tx(tx, dataflow_id, status, now_value) + if cleanup_err then return nil, cleanup_err end + end + row, row_err = get_tx(tx, dataflow_id) + if row_err then return nil, row_err end + return evidence_from_row(row, status, admission_key) + end) +end + +function activation_repo.get_activation_evidence(dataflow_id, admission_key) + local valid, id_err = validate_id(dataflow_id) + if not valid then return nil, id_err end + if not admission_key_valid(admission_key) then return nil, typed(errors.INVALID, "admission_key is required") end + local db, db_err = sql.get(consts.APP_DB) + if db_err then return nil, db_err end + local rows, query_err = db_query(db, [[ + SELECT d.status, a.dataflow_id, a.generation, a.desired_active, + a.owner_epoch, a.launch_args, a.requested_at, a.updated_at, + a.admission_key, a.ever_activated, a.terminal_status, + a.terminal_outcome_json, a.terminal_generation, a.terminal_ack_at + FROM dataflows d LEFT JOIN dataflow_activations a ON a.dataflow_id = d.dataflow_id + WHERE d.dataflow_id = ? LIMIT 1 + ]], { dataflow_id }) + db:release() + if query_err then return nil, query_err end + local joined = rows and rows[1] or nil + if not joined then return { state = "absent", admission_key = admission_key, + ever_activated = false }, nil end + if joined.admission_key and tostring(joined.admission_key) ~= admission_key then + return nil, typed(errors.CONFLICT, "dataflow has another admission key") + end + local row, row_err = normalize_row(joined.dataflow_id and joined or nil) + if row_err then return nil, row_err end + return evidence_from_row(row, tostring(joined.status), admission_key) +end + +function activation_repo.ack_terminal(dataflow_id, admission_key, generation, now_value) + local valid, id_err = validate_id(dataflow_id) + if not valid then return nil, id_err end + if not admission_key_valid(admission_key) then return nil, typed(errors.INVALID, "admission_key is required") end + generation = tonumber(generation) + if not generation or generation < 1 or generation % 1 ~= 0 then + return nil, typed(errors.INVALID, "generation must be a positive integer") + end + valid, id_err = validate_timestamp(now_value, "terminal_ack_at") + if not valid then return nil, id_err end + return transaction(function(tx) + local _, lock_err = activation_repo.lock_workflow_tx(tx, dataflow_id) + if lock_err then return nil, lock_err end + local row, row_err = get_tx(tx, dataflow_id) + if row_err then return nil, row_err end + if not row or row.admission_key ~= admission_key then + return nil, typed(errors.CONFLICT, "terminal admission or generation differs") + end + if not row.terminal_status then return nil, typed(errors.UNAVAILABLE, "terminal evidence is unavailable") end + if row.terminal_generation ~= generation then + return nil, typed(errors.CONFLICT, "terminal admission or generation differs") + end + if not row.terminal_ack_at then + local _, update_err = tx_execute(tx, [[ + UPDATE dataflow_activations SET terminal_ack_at = ? + WHERE dataflow_id = ? AND admission_key = ? + AND terminal_generation = ? AND terminal_ack_at IS NULL + ]], { now_value, dataflow_id, admission_key, generation }) + if update_err then return nil, update_err end + row, row_err = get_tx(tx, dataflow_id) + if row_err then return nil, row_err end + end + return { acknowledged = true, terminal_ack_at = row.terminal_ack_at }, nil + end) +end + +end diff --git a/src/persist/activation/operations.lua b/src/persist/activation/operations.lua new file mode 100644 index 0000000..498715d --- /dev/null +++ b/src/persist/activation/operations.lua @@ -0,0 +1,358 @@ +return function(activation_repo, shared) + local sql = shared.sql + local json = shared.json + local consts = shared.consts + local TERMINAL_STATUS = shared.TERMINAL_STATUS + local TERMINAL_VALUES = shared.TERMINAL_VALUES + local tx_query = shared.tx_query + local tx_execute = shared.tx_execute + local db_query = shared.db_query + local validate_id = shared.validate_id + local validate_timestamp = shared.validate_timestamp + local normalize_row = shared.normalize_row + local get_tx = shared.get_tx + local cleanup_terminal_tx = shared.cleanup_terminal_tx + local terminal_result_from_status = shared.terminal_result_from_status + local encode_launch_args = shared.encode_launch_args + local rebind = shared.rebind + local typed = shared.typed + +local function advance_activation_tx(tx, dataflow_id, launch_args: any, now_value, preserve_launch_args) + local encoded_args, encode_err = encode_launch_args(launch_args) + if encode_err then return nil, encode_err end + + local update_args = preserve_launch_args and "dataflow_activations.launch_args" or "excluded.launch_args" + local result, write_err = tx_execute(tx, ([[ + INSERT INTO dataflow_activations( + dataflow_id, generation, desired_active, owner_epoch, + launch_args, requested_at, updated_at + ) + SELECT ?, 1, ?, NULL, ?, ?, ? FROM dataflows + WHERE dataflow_id = ? AND status NOT IN (?, ?, ?, ?) + ON CONFLICT(dataflow_id) DO UPDATE SET + generation = dataflow_activations.generation + 1, + desired_active = excluded.desired_active, + owner_epoch = NULL, + launch_args = %s, + requested_at = excluded.requested_at, + updated_at = excluded.updated_at + WHERE EXISTS ( + SELECT 1 FROM dataflows + WHERE dataflow_id = excluded.dataflow_id AND status NOT IN (?, ?, ?, ?) + ) + ]]):format(update_args), { + dataflow_id, true, encoded_args or sql.as.null(), now_value, now_value, dataflow_id, + TERMINAL_VALUES[1], TERMINAL_VALUES[2], TERMINAL_VALUES[3], TERMINAL_VALUES[4], + TERMINAL_VALUES[1], TERMINAL_VALUES[2], TERMINAL_VALUES[3], TERMINAL_VALUES[4], + }) + if write_err then return nil, typed(errors.UNAVAILABLE, "failed to advance activation: " .. tostring(write_err)) end + if not result or (result.rows_affected or 0) == 0 then + return nil, typed(errors.INTERNAL, "activation request made no change") + end + + local row, row_err = get_tx(tx, dataflow_id) + if row_err then return nil, row_err end + if not row then return nil, typed(errors.INTERNAL, "activation row missing after advance") end + row.changed = true + row.terminal = false + return row, nil +end + +function activation_repo.request_activation_tx(tx, dataflow_id, launch_args, now_value) + if not tx then return nil, typed(errors.INVALID, "transaction is required") end + local valid, id_err = validate_id(dataflow_id) + 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 = 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 + return advance_activation_tx(tx, dataflow_id, launch_args, now_value, false) +end + +function activation_repo.activate_for_signal_tx(tx, dataflow_id, wake_key, wake_at, now_value) + if not tx then return nil, typed(errors.INVALID, "transaction is required") end + local valid, validation_err = validate_id(dataflow_id) + if not valid then return nil, validation_err end + if type(wake_key) ~= "string" or not wake_key:match("^signal:.+") then + return nil, typed(errors.INVALID, "signal wake_key is required") + end + valid, validation_err = validate_timestamp(wake_at, "wake_at") + if not valid then return nil, validation_err end + valid, validation_err = validate_timestamp(now_value, "requested_at") + if not valid then return nil, validation_err end + + 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 + terminal.wake_inserted = false + return terminal, nil + end + + local insert_result, insert_err = tx_execute(tx, [[ + INSERT INTO dataflow_wakes(dataflow_id, wake_key, wake_at, activation_generation) + SELECT ?, ?, ?, NULL FROM dataflows + WHERE dataflow_id = ? AND status NOT IN (?, ?, ?, ?) + ON CONFLICT(dataflow_id, wake_key) DO NOTHING + ]], { + dataflow_id, wake_key, wake_at, dataflow_id, + TERMINAL_VALUES[1], TERMINAL_VALUES[2], TERMINAL_VALUES[3], TERMINAL_VALUES[4], + }) + if insert_err then return nil, typed(errors.UNAVAILABLE, "failed to insert signal wake: " .. tostring(insert_err)) end + + if not insert_result or (insert_result.rows_affected or 0) == 0 then + local rows, row_err = tx_query(tx, [[ + SELECT activation_generation FROM dataflow_wakes + WHERE dataflow_id = ? AND wake_key = ? LIMIT 1 + ]], { dataflow_id, wake_key }) + if row_err then return nil, row_err end + return { + changed = false, + terminal = false, + wake_inserted = false, + generation = rows and rows[1] and tonumber(rows[1].activation_generation) or nil, + }, nil + end + + local activation, activation_err = advance_activation_tx(tx, dataflow_id, nil, now_value, true) + if activation_err then return nil, activation_err end + if activation.terminal then return nil, typed(errors.CONFLICT, "signal wake inserted for terminal dataflow") end + + local stamp_result, stamp_err = tx_execute(tx, [[ + UPDATE dataflow_wakes SET activation_generation = ? + WHERE dataflow_id = ? AND wake_key = ? AND activation_generation IS NULL + ]], { activation.generation, dataflow_id, wake_key }) + if stamp_err then return nil, typed(errors.UNAVAILABLE, "failed to fence signal wake: " .. tostring(stamp_err)) end + if not stamp_result or (stamp_result.rows_affected or 0) ~= 1 then + return nil, typed(errors.INTERNAL, "signal wake generation fence was not written") + end + + activation.wake_inserted = true + return activation, nil +end + +function activation_repo.activate_due_tx(tx, dataflow_id, wake_key, now_value) + if not tx then return nil, typed(errors.INVALID, "transaction is required") end + local valid, validation_err = validate_id(dataflow_id) + if not valid then return nil, validation_err end + if type(wake_key) ~= "string" or wake_key == "" then return nil, typed(errors.INVALID, "wake_key is required") end + valid, validation_err = validate_timestamp(now_value, "now") + if not valid then return nil, validation_err end + + 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 + local cleaned, cleanup_err = cleanup_terminal_tx(tx, dataflow_id, status, now_value) + if cleanup_err then return nil, cleanup_err end + cleaned.promoted = false + return cleaned, nil + end + + -- This conditional no-op update is the row lock/CAS. On PostgreSQL a + -- concurrent scanner waits and then rechecks activation_generation; on + -- SQLite it acquires the database writer lock before generation advances. + local lock_result, lock_err = tx_execute(tx, [[ + UPDATE dataflow_wakes SET wake_at = wake_at + WHERE dataflow_id = ? AND wake_key = ? AND wake_at <= ? + AND activation_generation IS NULL + AND EXISTS ( + SELECT 1 FROM dataflows + WHERE dataflow_id = ? AND status NOT IN (?, ?, ?, ?) + ) + ]], { + dataflow_id, wake_key, now_value, dataflow_id, + TERMINAL_VALUES[1], TERMINAL_VALUES[2], TERMINAL_VALUES[3], TERMINAL_VALUES[4], + }) + if lock_err then return nil, typed(errors.UNAVAILABLE, "failed to lock due wake: " .. tostring(lock_err)) end + + if lock_result and (lock_result.rows_affected or 0) > 0 then + local activation, activation_err = advance_activation_tx(tx, dataflow_id, nil, now_value, true) + if activation_err then return nil, activation_err end + if activation.terminal then return nil, typed(errors.CONFLICT, "due wake promoted for terminal dataflow") end + local stamp_result, stamp_err = tx_execute(tx, [[ + UPDATE dataflow_wakes SET activation_generation = ? + WHERE dataflow_id = ? AND wake_key = ? AND activation_generation IS NULL + ]], { activation.generation, dataflow_id, wake_key }) + if stamp_err then return nil, typed(errors.UNAVAILABLE, "failed to fence due wake: " .. tostring(stamp_err)) end + if not stamp_result or (stamp_result.rows_affected or 0) ~= 1 then + return nil, typed(errors.INTERNAL, "due wake generation fence was not written") + end + activation.promoted = true + return activation, nil + end + + local rows, row_err = tx_query(tx, [[ + SELECT wake_at, activation_generation FROM dataflow_wakes + WHERE dataflow_id = ? AND wake_key = ? LIMIT 1 + ]], { dataflow_id, wake_key }) + if row_err then return nil, row_err end + local row = rows and rows[1] or nil + if not row then + return { changed = false, terminal = false, promoted = false, missing = true }, nil + end + if row.activation_generation ~= nil then + return { + changed = false, + terminal = false, + promoted = false, + already_promoted = true, + generation = tonumber(row.activation_generation), + }, nil + end + return { changed = false, terminal = false, promoted = false, due = false }, nil +end + +function activation_repo.release_if_generation_tx(tx, dataflow_id, generation, now_value) + if not tx then return nil, typed(errors.INVALID, "transaction is required") end + local valid, validation_err = validate_id(dataflow_id) + if not valid then return nil, validation_err end + generation = tonumber(generation) + if not generation or generation < 1 or generation % 1 ~= 0 then + return nil, typed(errors.INVALID, "generation must be a positive integer") + end + valid, validation_err = validate_timestamp(now_value, "updated_at") + if not valid then return nil, validation_err end + + 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 + terminal.released = false + return terminal, nil + end + + local result, update_err = tx_execute(tx, [[ + UPDATE dataflow_activations + SET desired_active = ?, launch_args = NULL, updated_at = ? + WHERE dataflow_id = ? AND generation = ? AND desired_active = ? + AND EXISTS ( + SELECT 1 FROM dataflows + WHERE dataflow_id = ? AND status NOT IN (?, ?, ?, ?) + ) + ]], { + false, now_value, dataflow_id, generation, true, dataflow_id, + TERMINAL_VALUES[1], TERMINAL_VALUES[2], TERMINAL_VALUES[3], TERMINAL_VALUES[4], + }) + if update_err then return nil, typed(errors.UNAVAILABLE, "failed to release activation: " .. tostring(update_err)) end + if result and (result.rows_affected or 0) > 0 then + return { changed = true, released = true, generation = generation, terminal = false }, nil + end + + local current, current_err = get_tx(tx, dataflow_id) + if current_err then return nil, current_err end + return { + changed = false, + released = false, + terminal = false, + generation = current and current.generation or nil, + }, nil +end + +-- Fence process ownership before spawn. A generation can be claimed only from +-- the exact epoch observed by the overseer. The write happens before process +-- creation, so an overseer crash between claim and spawn is classified as a +-- same-runtime loss rather than retried into a process flood. +function activation_repo.claim_epoch_tx( + tx, dataflow_id, generation, observed_epoch, runtime_epoch, now_value) + if not tx then return nil, typed(errors.INVALID, "transaction is required") end + local valid, validation_err = validate_id(dataflow_id) + if not valid then return nil, validation_err end + generation = tonumber(generation) + if not generation or generation < 1 or generation % 1 ~= 0 then + return nil, typed(errors.INVALID, "generation must be a positive integer") + end + if type(runtime_epoch) ~= "string" or runtime_epoch == "" then + return nil, typed(errors.INVALID, "runtime_epoch is required") + end + valid, validation_err = validate_timestamp(now_value, "updated_at") + if not valid then return nil, validation_err end + + 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 + terminal.claimed = false + return terminal, nil + end + + local epoch_predicate = "owner_epoch IS NULL" + local params = { runtime_epoch, now_value, dataflow_id, generation, true } + if observed_epoch ~= nil then + if type(observed_epoch) ~= "string" or observed_epoch == "" then + return nil, typed(errors.INVALID, "observed_epoch must be nil or a non-empty string") + end + epoch_predicate = "owner_epoch = ?" + table.insert(params, observed_epoch) + end + local result, update_err = tx_execute(tx, [[ + UPDATE dataflow_activations + SET owner_epoch = ?, updated_at = ? + WHERE dataflow_id = ? AND generation = ? AND desired_active = ? + AND ]] .. epoch_predicate, params) + if update_err then return nil, typed(errors.UNAVAILABLE, "failed to claim activation epoch: " .. tostring(update_err)) end + + local current, current_err = get_tx(tx, dataflow_id) + if current_err then return nil, current_err end + if not current then return nil, typed(errors.INTERNAL, "activation row missing after epoch claim") end + current.claimed = result ~= nil and (result.rows_affected or 0) == 1 + current.terminal = false + return current, nil +end + +function activation_repo.consume_wake_tx(tx, dataflow_id, wake_key, generation) + if not tx then return nil, typed(errors.INVALID, "transaction is required") end + local valid, validation_err = validate_id(dataflow_id) + if not valid then return nil, validation_err end + if type(wake_key) ~= "string" or wake_key == "" then return nil, typed(errors.INVALID, "wake_key is required") end + + 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 + terminal.consumed = false + return terminal, nil + end + + local query = "DELETE FROM dataflow_wakes WHERE dataflow_id = ? AND wake_key = ?" + local params = { dataflow_id, wake_key } + if generation ~= nil then + generation = tonumber(generation) + if not generation or generation < 1 or generation % 1 ~= 0 then + return nil, typed(errors.INVALID, "generation must be a positive integer") + end + query = query .. " AND activation_generation = ?" + table.insert(params, generation) + end + local result, delete_err = tx_execute(tx, query, params) + if delete_err then return nil, typed(errors.UNAVAILABLE, "failed to consume wake: " .. tostring(delete_err)) end + return { changed = result and (result.rows_affected or 0) > 0, consumed = result and (result.rows_affected or 0) > 0 }, nil +end + +-- Register or re-arm a durable yield deadline. Reusing the same logical yield +-- is a new wait episode, so any activation fence left by the previous episode +-- must be cleared atomically with the new deadline. +function activation_repo.register_yield_wake_tx(tx, dataflow_id, yield_id, wake_at) + if not tx then return nil, typed(errors.INVALID, "transaction is required") end + local valid, validation_err = validate_id(dataflow_id) + if not valid then return nil, validation_err end + if type(yield_id) ~= "string" or yield_id == "" then return nil, typed(errors.INVALID, "yield_id is required") end + valid, validation_err = validate_timestamp(wake_at, "wake_at") + if not valid then return nil, validation_err end + + local result, write_err = tx_execute(tx, [[ + INSERT INTO dataflow_wakes(dataflow_id, wake_key, wake_at, activation_generation) + VALUES (?, ?, ?, NULL) + ON CONFLICT(dataflow_id, wake_key) DO UPDATE SET + wake_at = excluded.wake_at, + activation_generation = NULL + ]], { dataflow_id, "yield:" .. yield_id, wake_at }) + if write_err then return nil, typed(errors.UNAVAILABLE, "failed to register yield wake: " .. tostring(write_err)) end + return { + changed = result ~= nil and (result.rows_affected or 0) > 0, + }, nil +end + +end diff --git a/src/persist/activation_repo.lua b/src/persist/activation_repo.lua index 971f5c6..77fd4cc 100644 --- a/src/persist/activation_repo.lua +++ b/src/persist/activation_repo.lua @@ -2,7 +2,11 @@ local sql = require("sql") local json = require("json") local consts = require("dataflow_consts") -local activation_repo = {} +local activation_repo: any = {} + +local function typed(kind, message) + return errors.new({ kind = kind, message = tostring(message) }) +end local TERMINAL_STATUS = { [consts.STATUS.COMPLETED_SUCCESS] = true, @@ -47,14 +51,14 @@ end local function validate_id(dataflow_id) if type(dataflow_id) ~= "string" or dataflow_id == "" then - return nil, "dataflow_id is required" + return nil, typed(errors.INVALID, "dataflow_id is required") end return true, nil end local function validate_timestamp(value, field) if type(value) ~= "string" or value == "" then - return nil, field .. " is required" + return nil, typed(errors.INVALID, field .. " is required") end return true, nil end @@ -64,13 +68,13 @@ local function validate_json_value(value: any, seen: any, path: string) if kind == "nil" or kind == "string" or kind == "boolean" then return true, nil end if kind == "number" then if value ~= value or value == math.huge or value == -math.huge then - return nil, path .. " contains a non-finite number" + return nil, typed(errors.INVALID, path .. " contains a non-finite number") end return true, nil end - if kind ~= "table" then return nil, path .. " contains unsupported " .. kind end - if getmetatable(value) ~= nil then return nil, path .. " must not have a metatable" end - if seen[value] then return nil, path .. " contains a cycle" end + if kind ~= "table" then return nil, typed(errors.INVALID, path .. " contains unsupported " .. kind) end + if getmetatable(value) ~= nil then return nil, typed(errors.INVALID, path .. " must not have a metatable") end + if seen[value] then return nil, typed(errors.INVALID, path .. " contains a cycle") end seen[value] = true local key_kind = nil @@ -82,7 +86,7 @@ local function validate_json_value(value: any, seen: any, path: string) local array_index = tonumber(key) or 0 if array_index < 1 or array_index % 1 ~= 0 then seen[value] = nil - return nil, path .. " contains an invalid array index" + return nil, typed(errors.INVALID, path .. " contains an invalid array index") end max_index = math.max(max_index, array_index) current_kind = "array" @@ -90,11 +94,11 @@ local function validate_json_value(value: any, seen: any, path: string) current_kind = "object" else seen[value] = nil - return nil, path .. " contains an unsupported key" + return nil, typed(errors.INVALID, path .. " contains an unsupported key") end if key_kind and key_kind ~= current_kind then seen[value] = nil - return nil, path .. " mixes object and array keys" + return nil, typed(errors.INVALID, path .. " mixes object and array keys") end key_kind = current_kind count = count + 1 @@ -106,7 +110,7 @@ local function validate_json_value(value: any, seen: any, path: string) end seen[value] = nil if key_kind == "array" and max_index ~= count then - return nil, path .. " contains a sparse array" + return nil, typed(errors.INVALID, path .. " contains a sparse array") end return true, nil end @@ -114,15 +118,15 @@ end local function encode_launch_args(launch_args: any) if launch_args == nil then return nil, nil end if type(launch_args) ~= "table" or getmetatable(launch_args) ~= nil then - return nil, "launch_args must be a plain object" + return nil, typed(errors.INVALID, "launch_args must be a plain object") end for key in pairs(launch_args) do - if type(key) ~= "string" then return nil, "launch_args must be a plain object" end + if type(key) ~= "string" then return nil, typed(errors.INVALID, "launch_args must be a plain object") end end local valid, validation_err = validate_json_value(launch_args, {}, "launch_args") if not valid then return nil, validation_err end local encoded, encode_err = json.encode(launch_args) - if encode_err then return nil, "failed to encode launch_args: " .. tostring(encode_err) end + if encode_err then return nil, typed(errors.UNAVAILABLE, "failed to encode launch_args: " .. tostring(encode_err)) end return encoded, nil end @@ -132,11 +136,11 @@ local function decode_launch_args(value: any) if type(value) == "string" then local decode_err decoded, decode_err = json.decode(value) - if decode_err then return nil, "failed to decode launch_args: " .. tostring(decode_err) end + if decode_err then return nil, typed(errors.UNAVAILABLE, "failed to decode launch_args: " .. tostring(decode_err)) end end - if type(decoded) ~= "table" then return nil, "launch_args is not an object" end + if type(decoded) ~= "table" then return nil, typed(errors.INVALID, "launch_args is not an object") end for key in pairs(decoded) do - if type(key) ~= "string" then return nil, "launch_args is not an object" end + if type(key) ~= "string" then return nil, typed(errors.INVALID, "launch_args is not an object") end end return decoded, nil end @@ -179,7 +183,7 @@ function activation_repo.lock_workflow_tx(tx, dataflow_id) { dataflow_id }) if lock_err then return nil, lock_err end if not lock_result or (lock_result.rows_affected or 0) == 0 then - return nil, "dataflow not found" + return nil, typed(errors.NOT_FOUND, "dataflow not found") end end local query = "SELECT status FROM dataflows WHERE dataflow_id = ? LIMIT 1" @@ -188,7 +192,7 @@ function activation_repo.lock_workflow_tx(tx, dataflow_id) end local rows, query_err = tx:query(rebind(query, db_type), { dataflow_id }) if query_err then return nil, query_err end - if not rows or not rows[1] then return nil, "dataflow not found" end + if not rows or not rows[1] then return nil, typed(errors.NOT_FOUND, "dataflow not found") end return tostring(rows[1].status), nil end @@ -216,11 +220,11 @@ end local function cleanup_terminal_tx(tx, dataflow_id, status, now_value) local flow_rows, flow_err = tx_query(tx, "SELECT metadata FROM dataflows WHERE dataflow_id = ?", { dataflow_id }) - if flow_err then return nil, "failed to read terminal outcome: " .. tostring(flow_err) end + if flow_err then return nil, typed(errors.UNAVAILABLE, "failed to read terminal outcome: " .. tostring(flow_err)) end local outcome = flow_rows and flow_rows[1] and flow_rows[1].metadata or nil if type(outcome) == "table" then local encoded, encode_err = json.encode(outcome) - if encode_err then return nil, "failed to encode terminal outcome: " .. tostring(encode_err) end + if encode_err then return nil, typed(errors.UNAVAILABLE, "failed to encode terminal outcome: " .. tostring(encode_err)) end outcome = encoded end local activation_result, activation_err = tx_execute(tx, [[ @@ -235,11 +239,11 @@ local function cleanup_terminal_tx(tx, dataflow_id, status, now_value) WHERE dataflow_id = ? AND (desired_active = ? OR launch_args IS NOT NULL OR (admission_key IS NOT NULL AND terminal_status IS NULL)) ]], { false, now_value, status, outcome or sql.as.null(), dataflow_id, true }) - if activation_err then return nil, "failed to disable terminal activation: " .. tostring(activation_err) end + if activation_err then return nil, typed(errors.UNAVAILABLE, "failed to disable terminal activation: " .. tostring(activation_err)) end local wake_result, wake_err = tx_execute(tx, "DELETE FROM dataflow_wakes WHERE dataflow_id = ?", { dataflow_id }) - if wake_err then return nil, "failed to clear terminal wakes: " .. tostring(wake_err) end + if wake_err then return nil, typed(errors.UNAVAILABLE, "failed to clear terminal wakes: " .. tostring(wake_err)) end local activation_disabled = activation_result and (activation_result.rows_affected or 0) > 0 local wake_index_changed = wake_result and (wake_result.rows_affected or 0) > 0 @@ -252,353 +256,15 @@ local function cleanup_terminal_tx(tx, dataflow_id, status, now_value) }, nil end -local function advance_activation_tx(tx, dataflow_id, launch_args: any, now_value, preserve_launch_args) - local encoded_args, encode_err = encode_launch_args(launch_args) - if encode_err then return nil, encode_err end - - local update_args = preserve_launch_args and "dataflow_activations.launch_args" or "excluded.launch_args" - local result, write_err = tx_execute(tx, ([[ - INSERT INTO dataflow_activations( - dataflow_id, generation, desired_active, owner_epoch, - launch_args, requested_at, updated_at - ) - SELECT ?, 1, ?, NULL, ?, ?, ? FROM dataflows - WHERE dataflow_id = ? AND status NOT IN (?, ?, ?, ?) - ON CONFLICT(dataflow_id) DO UPDATE SET - generation = dataflow_activations.generation + 1, - desired_active = excluded.desired_active, - owner_epoch = NULL, - launch_args = %s, - requested_at = excluded.requested_at, - updated_at = excluded.updated_at - WHERE EXISTS ( - SELECT 1 FROM dataflows - WHERE dataflow_id = excluded.dataflow_id AND status NOT IN (?, ?, ?, ?) - ) - ]]):format(update_args), { - dataflow_id, true, encoded_args or sql.as.null(), now_value, now_value, dataflow_id, - TERMINAL_VALUES[1], TERMINAL_VALUES[2], TERMINAL_VALUES[3], TERMINAL_VALUES[4], - TERMINAL_VALUES[1], TERMINAL_VALUES[2], TERMINAL_VALUES[3], TERMINAL_VALUES[4], - }) - if write_err then return nil, "failed to advance activation: " .. tostring(write_err) end - if not result or (result.rows_affected or 0) == 0 then - return nil, "activation request made no change" - end - - local row, row_err = get_tx(tx, dataflow_id) - if row_err then return nil, row_err end - if not row then return nil, "activation row missing after advance" end - row.changed = true - row.terminal = false - return row, nil -end - -function activation_repo.request_activation_tx(tx, dataflow_id, launch_args, now_value) - if not tx then return nil, "transaction is required" end - local valid, id_err = validate_id(dataflow_id) - 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 = 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 - return advance_activation_tx(tx, dataflow_id, launch_args, now_value, false) -end - -function activation_repo.activate_for_signal_tx(tx, dataflow_id, wake_key, wake_at, now_value) - if not tx then return nil, "transaction is required" end - local valid, validation_err = validate_id(dataflow_id) - if not valid then return nil, validation_err end - if type(wake_key) ~= "string" or not wake_key:match("^signal:.+") then - return nil, "signal wake_key is required" - end - valid, validation_err = validate_timestamp(wake_at, "wake_at") - if not valid then return nil, validation_err end - valid, validation_err = validate_timestamp(now_value, "requested_at") - if not valid then return nil, validation_err end - - 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 - terminal.wake_inserted = false - return terminal, nil - end - - local insert_result, insert_err = tx_execute(tx, [[ - INSERT INTO dataflow_wakes(dataflow_id, wake_key, wake_at, activation_generation) - SELECT ?, ?, ?, NULL FROM dataflows - WHERE dataflow_id = ? AND status NOT IN (?, ?, ?, ?) - ON CONFLICT(dataflow_id, wake_key) DO NOTHING - ]], { - dataflow_id, wake_key, wake_at, dataflow_id, - TERMINAL_VALUES[1], TERMINAL_VALUES[2], TERMINAL_VALUES[3], TERMINAL_VALUES[4], - }) - if insert_err then return nil, "failed to insert signal wake: " .. tostring(insert_err) end - - if not insert_result or (insert_result.rows_affected or 0) == 0 then - local rows, row_err = tx_query(tx, [[ - SELECT activation_generation FROM dataflow_wakes - WHERE dataflow_id = ? AND wake_key = ? LIMIT 1 - ]], { dataflow_id, wake_key }) - if row_err then return nil, row_err end - return { - changed = false, - terminal = false, - wake_inserted = false, - generation = rows and rows[1] and tonumber(rows[1].activation_generation) or nil, - }, nil - end - - local activation, activation_err = advance_activation_tx(tx, dataflow_id, nil, now_value, true) - if activation_err then return nil, activation_err end - if activation.terminal then return nil, "signal wake inserted for terminal dataflow" end - - local stamp_result, stamp_err = tx_execute(tx, [[ - UPDATE dataflow_wakes SET activation_generation = ? - WHERE dataflow_id = ? AND wake_key = ? AND activation_generation IS NULL - ]], { activation.generation, dataflow_id, wake_key }) - if stamp_err then return nil, "failed to fence signal wake: " .. tostring(stamp_err) end - if not stamp_result or (stamp_result.rows_affected or 0) ~= 1 then - return nil, "signal wake generation fence was not written" - end - - activation.wake_inserted = true - return activation, nil -end - -function activation_repo.activate_due_tx(tx, dataflow_id, wake_key, now_value) - if not tx then return nil, "transaction is required" end - local valid, validation_err = validate_id(dataflow_id) - 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 - valid, validation_err = validate_timestamp(now_value, "now") - if not valid then return nil, validation_err end - - 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 - local cleaned, cleanup_err = cleanup_terminal_tx(tx, dataflow_id, status, now_value) - if cleanup_err then return nil, cleanup_err end - cleaned.promoted = false - return cleaned, nil - end - - -- This conditional no-op update is the row lock/CAS. On PostgreSQL a - -- concurrent scanner waits and then rechecks activation_generation; on - -- SQLite it acquires the database writer lock before generation advances. - local lock_result, lock_err = tx_execute(tx, [[ - UPDATE dataflow_wakes SET wake_at = wake_at - WHERE dataflow_id = ? AND wake_key = ? AND wake_at <= ? - AND activation_generation IS NULL - AND EXISTS ( - SELECT 1 FROM dataflows - WHERE dataflow_id = ? AND status NOT IN (?, ?, ?, ?) - ) - ]], { - dataflow_id, wake_key, now_value, dataflow_id, - TERMINAL_VALUES[1], TERMINAL_VALUES[2], TERMINAL_VALUES[3], TERMINAL_VALUES[4], - }) - if lock_err then return nil, "failed to lock due wake: " .. tostring(lock_err) end - - if lock_result and (lock_result.rows_affected or 0) > 0 then - local activation, activation_err = advance_activation_tx(tx, dataflow_id, nil, now_value, true) - if activation_err then return nil, activation_err end - if activation.terminal then return nil, "due wake promoted for terminal dataflow" end - local stamp_result, stamp_err = tx_execute(tx, [[ - UPDATE dataflow_wakes SET activation_generation = ? - WHERE dataflow_id = ? AND wake_key = ? AND activation_generation IS NULL - ]], { activation.generation, dataflow_id, wake_key }) - if stamp_err then return nil, "failed to fence due wake: " .. tostring(stamp_err) end - if not stamp_result or (stamp_result.rows_affected or 0) ~= 1 then - return nil, "due wake generation fence was not written" - end - activation.promoted = true - return activation, nil - end - - local rows, row_err = tx_query(tx, [[ - SELECT wake_at, activation_generation FROM dataflow_wakes - WHERE dataflow_id = ? AND wake_key = ? LIMIT 1 - ]], { dataflow_id, wake_key }) - if row_err then return nil, row_err end - local row = rows and rows[1] or nil - if not row then - return { changed = false, terminal = false, promoted = false, missing = true }, nil - end - if row.activation_generation ~= nil then - return { - changed = false, - terminal = false, - promoted = false, - already_promoted = true, - generation = tonumber(row.activation_generation), - }, nil - end - return { changed = false, terminal = false, promoted = false, due = false }, nil -end - -function activation_repo.release_if_generation_tx(tx, dataflow_id, generation, now_value) - if not tx then return nil, "transaction is required" end - local valid, validation_err = validate_id(dataflow_id) - if not valid then return nil, validation_err end - generation = tonumber(generation) - if not generation or generation < 1 or generation % 1 ~= 0 then - return nil, "generation must be a positive integer" - end - valid, validation_err = validate_timestamp(now_value, "updated_at") - if not valid then return nil, validation_err end - - 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 - terminal.released = false - return terminal, nil - end - - local result, update_err = tx_execute(tx, [[ - UPDATE dataflow_activations - SET desired_active = ?, launch_args = NULL, updated_at = ? - WHERE dataflow_id = ? AND generation = ? AND desired_active = ? - AND EXISTS ( - SELECT 1 FROM dataflows - WHERE dataflow_id = ? AND status NOT IN (?, ?, ?, ?) - ) - ]], { - false, now_value, dataflow_id, generation, true, dataflow_id, - TERMINAL_VALUES[1], TERMINAL_VALUES[2], TERMINAL_VALUES[3], TERMINAL_VALUES[4], - }) - if update_err then return nil, "failed to release activation: " .. tostring(update_err) end - if result and (result.rows_affected or 0) > 0 then - return { changed = true, released = true, generation = generation, terminal = false }, nil - end - - local current, current_err = get_tx(tx, dataflow_id) - if current_err then return nil, current_err end - return { - changed = false, - released = false, - terminal = false, - generation = current and current.generation or nil, - }, nil -end - --- Fence process ownership before spawn. A generation can be claimed only from --- the exact epoch observed by the overseer. The write happens before process --- creation, so an overseer crash between claim and spawn is classified as a --- same-runtime loss rather than retried into a process flood. -function activation_repo.claim_epoch_tx( - tx, dataflow_id, generation, observed_epoch, runtime_epoch, now_value) - if not tx then return nil, "transaction is required" end - local valid, validation_err = validate_id(dataflow_id) - if not valid then return nil, validation_err end - generation = tonumber(generation) - if not generation or generation < 1 or generation % 1 ~= 0 then - return nil, "generation must be a positive integer" - end - if type(runtime_epoch) ~= "string" or runtime_epoch == "" then - return nil, "runtime_epoch is required" - end - valid, validation_err = validate_timestamp(now_value, "updated_at") - if not valid then return nil, validation_err end - - 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 - terminal.claimed = false - return terminal, nil - end - - local epoch_predicate = "owner_epoch IS NULL" - local params = { runtime_epoch, now_value, dataflow_id, generation, true } - if observed_epoch ~= nil then - if type(observed_epoch) ~= "string" or observed_epoch == "" then - return nil, "observed_epoch must be nil or a non-empty string" - end - epoch_predicate = "owner_epoch = ?" - table.insert(params, observed_epoch) - end - local result, update_err = tx_execute(tx, [[ - UPDATE dataflow_activations - SET owner_epoch = ?, updated_at = ? - WHERE dataflow_id = ? AND generation = ? AND desired_active = ? - AND ]] .. epoch_predicate, params) - if update_err then return nil, "failed to claim activation epoch: " .. tostring(update_err) end - - local current, current_err = get_tx(tx, dataflow_id) - if current_err then return nil, current_err end - if not current then return nil, "activation row missing after epoch claim" end - current.claimed = result ~= nil and (result.rows_affected or 0) == 1 - current.terminal = false - return current, nil -end - -function activation_repo.consume_wake_tx(tx, dataflow_id, wake_key, generation) - if not tx then return nil, "transaction is required" end - local valid, validation_err = validate_id(dataflow_id) - 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 = 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 - terminal.consumed = false - return terminal, nil - end - - local query = "DELETE FROM dataflow_wakes WHERE dataflow_id = ? AND wake_key = ?" - local params = { dataflow_id, wake_key } - if generation ~= nil then - generation = tonumber(generation) - if not generation or generation < 1 or generation % 1 ~= 0 then - return nil, "generation must be a positive integer" - end - query = query .. " AND activation_generation = ?" - table.insert(params, generation) - end - local result, delete_err = tx_execute(tx, query, params) - if delete_err then return nil, "failed to consume wake: " .. tostring(delete_err) end - return { changed = result and (result.rows_affected or 0) > 0, consumed = result and (result.rows_affected or 0) > 0 }, nil -end - --- Register or re-arm a durable yield deadline. Reusing the same logical yield --- is a new wait episode, so any activation fence left by the previous episode --- must be cleared atomically with the new deadline. -function activation_repo.register_yield_wake_tx(tx, dataflow_id, yield_id, wake_at) - if not tx then return nil, "transaction is required" end - local valid, validation_err = validate_id(dataflow_id) - if not valid then return nil, validation_err end - if type(yield_id) ~= "string" or yield_id == "" then return nil, "yield_id is required" end - valid, validation_err = validate_timestamp(wake_at, "wake_at") - if not valid then return nil, validation_err end - - local result, write_err = tx_execute(tx, [[ - INSERT INTO dataflow_wakes(dataflow_id, wake_key, wake_at, activation_generation) - VALUES (?, ?, ?, NULL) - ON CONFLICT(dataflow_id, wake_key) DO UPDATE SET - wake_at = excluded.wake_at, - activation_generation = NULL - ]], { dataflow_id, "yield:" .. yield_id, wake_at }) - if write_err then return nil, "failed to register yield wake: " .. tostring(write_err) end - return { - changed = result ~= nil and (result.rows_affected or 0) > 0, - }, nil -end - function activation_repo.disable_terminal_tx(tx, dataflow_id, now_value) - if not tx then return nil, "transaction is required" end + if not tx then return nil, typed(errors.INVALID, "transaction is required") end local valid, validation_err = validate_id(dataflow_id) 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 = 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 + if not TERMINAL_STATUS[status] then return nil, typed(errors.CONFLICT, "dataflow is not terminal") end return cleanup_terminal_tx(tx, dataflow_id, status, now_value) end @@ -618,160 +284,6 @@ function activation_repo.get(dataflow_id) return normalize_row(rows and rows[1] or nil) end -local function admission_key_valid(key) - return type(key) == "string" and key ~= "" -end - -local function evidence_from_row(row, status, key) - if not row then - return { state = TERMINAL_STATUS[status] and "terminal" or "created", - admission_key = key, ever_activated = false, - terminal_status = TERMINAL_STATUS[status] and status or nil }, nil - end - local outcome = row.terminal_outcome_json - if type(outcome) == "string" and outcome ~= "" then - local decoded, err = json.decode(outcome) - if err then return nil, "invalid terminal outcome: " .. tostring(err) end - outcome = decoded - end - local state = "activated" - if TERMINAL_STATUS[status] or row.terminal_status then - state = row.terminal_status and "terminal" or "unknown" - elseif status == consts.STATUS.RUNNING then state = "running" end - return { - state = state, admission_key = key, generation = row.generation, - ever_activated = row.ever_activated, terminal_status = row.terminal_status, - terminal_outcome = outcome, terminal_generation = row.terminal_generation, - terminal_ack_at = row.terminal_ack_at, - }, nil -end - -local function transaction(fn) - local db, db_err = sql.get(consts.APP_DB) - if db_err then return nil, db_err end - local tx, begin_err = db:begin() - if begin_err then db:release(); return nil, begin_err end - local value, operation_err = fn(tx) - if operation_err then tx:rollback(); db:release(); return nil, operation_err end - local committed, commit_err = tx:commit() - if not committed or commit_err then - tx:rollback(); db:release() - return nil, commit_err or "transaction did not commit" - end - db:release() - return value, nil -end - -function activation_repo.ensure_activation(dataflow_id, admission_key, now_value) - local valid, id_err = validate_id(dataflow_id) - if not valid then return nil, id_err end - if not admission_key_valid(admission_key) then return nil, "admission_key is required" end - valid, id_err = validate_timestamp(now_value, "requested_at") - if not valid then return nil, id_err end - return transaction(function(tx) - local status, lock_err = activation_repo.lock_workflow_tx(tx, dataflow_id) - if lock_err == "dataflow not found" then - return { state = "absent", admission_key = admission_key, - ever_activated = false }, nil - end - if lock_err then return nil, lock_err end - local row, row_err = get_tx(tx, dataflow_id) - if row_err then return nil, row_err end - if row and row.admission_key and row.admission_key ~= admission_key then - return nil, "CONFLICT: dataflow has another admission key" - end - if not row and TERMINAL_STATUS[status] then - return evidence_from_row(nil, status, admission_key) - end - if not row then - local result, insert_err = tx_execute(tx, [[ - INSERT INTO dataflow_activations(dataflow_id,generation,desired_active, - owner_epoch,launch_args,requested_at,updated_at,admission_key,ever_activated) - VALUES (?,1,?,NULL,NULL,?,?,?,?) - ]], { dataflow_id, true, now_value, now_value, admission_key, true }) - if insert_err then return nil, insert_err end - if not result or (result.rows_affected or 0) ~= 1 then - return nil, "activation insert made no change" - end - elseif not row.admission_key then - local _, update_err = tx_execute(tx, [[ - UPDATE dataflow_activations - SET admission_key = ?, ever_activated = ?, updated_at = ? - WHERE dataflow_id = ? AND admission_key IS NULL - ]], { admission_key, true, now_value, dataflow_id }) - if update_err then return nil, update_err end - end - if TERMINAL_STATUS[status] then - local _, cleanup_err = cleanup_terminal_tx(tx, dataflow_id, status, now_value) - if cleanup_err then return nil, cleanup_err end - end - row, row_err = get_tx(tx, dataflow_id) - if row_err then return nil, row_err end - return evidence_from_row(row, status, admission_key) - end) -end - -function activation_repo.get_activation_evidence(dataflow_id, admission_key) - local valid, id_err = validate_id(dataflow_id) - if not valid then return nil, id_err end - if not admission_key_valid(admission_key) then return nil, "admission_key is required" end - local db, db_err = sql.get(consts.APP_DB) - if db_err then return nil, db_err end - local rows, query_err = db_query(db, [[ - SELECT d.status, a.dataflow_id, a.generation, a.desired_active, - a.owner_epoch, a.launch_args, a.requested_at, a.updated_at, - a.admission_key, a.ever_activated, a.terminal_status, - a.terminal_outcome_json, a.terminal_generation, a.terminal_ack_at - FROM dataflows d LEFT JOIN dataflow_activations a ON a.dataflow_id = d.dataflow_id - WHERE d.dataflow_id = ? LIMIT 1 - ]], { dataflow_id }) - db:release() - if query_err then return nil, query_err end - local joined = rows and rows[1] or nil - if not joined then return { state = "absent", admission_key = admission_key, - ever_activated = false }, nil end - if joined.admission_key and tostring(joined.admission_key) ~= admission_key then - return nil, "CONFLICT: dataflow has another admission key" - end - local row, row_err = normalize_row(joined.dataflow_id and joined or nil) - if row_err then return nil, row_err end - return evidence_from_row(row, tostring(joined.status), admission_key) -end - -function activation_repo.ack_terminal(dataflow_id, admission_key, generation, now_value) - local valid, id_err = validate_id(dataflow_id) - if not valid then return nil, id_err end - if not admission_key_valid(admission_key) then return nil, "admission_key is required" end - generation = tonumber(generation) - if not generation or generation < 1 or generation % 1 ~= 0 then - return nil, "generation must be a positive integer" - end - valid, id_err = validate_timestamp(now_value, "terminal_ack_at") - if not valid then return nil, id_err end - return transaction(function(tx) - local _, lock_err = activation_repo.lock_workflow_tx(tx, dataflow_id) - if lock_err then return nil, lock_err end - local row, row_err = get_tx(tx, dataflow_id) - if row_err then return nil, row_err end - if not row or row.admission_key ~= admission_key or - row.terminal_generation ~= generation then - return nil, "CONFLICT: terminal admission or generation differs" - end - if not row.terminal_status then return nil, "terminal evidence is unavailable" end - if not row.terminal_ack_at then - local _, update_err = tx_execute(tx, [[ - UPDATE dataflow_activations SET terminal_ack_at = ? - WHERE dataflow_id = ? AND admission_key = ? - AND terminal_generation = ? AND terminal_ack_at IS NULL - ]], { now_value, dataflow_id, admission_key, generation }) - if update_err then return nil, update_err end - row, row_err = get_tx(tx, dataflow_id) - if row_err then return nil, row_err end - end - return { acknowledged = true, terminal_ack_at = row.terminal_ack_at }, nil - end) -end - function activation_repo.list_active() local db, db_err = sql.get(consts.APP_DB) if db_err then return nil, db_err end @@ -797,4 +309,17 @@ function activation_repo.list_active() return result, nil end -return activation_repo +local shared = { + sql = sql, json = json, consts = consts, + TERMINAL_STATUS = TERMINAL_STATUS, TERMINAL_VALUES = TERMINAL_VALUES, + tx_query = tx_query, tx_execute = tx_execute, db_query = db_query, + validate_id = validate_id, validate_timestamp = validate_timestamp, + normalize_row = normalize_row, get_tx = get_tx, + cleanup_terminal_tx = cleanup_terminal_tx, encode_launch_args = encode_launch_args, + terminal_result_from_status = terminal_result_from_status, + rebind = rebind, typed = typed, +} +require("activation_operations")(activation_repo, shared) +require("activation_evidence")(activation_repo, shared) + +return activation_repo :: any diff --git a/src/persist/ops.lua b/src/persist/ops.lua index f91f0c6..ef6e9fe 100644 --- a/src/persist/ops.lua +++ b/src/persist/ops.lua @@ -1226,17 +1226,39 @@ end handlers[constants.COMMAND_TYPES.DELETE_WORKFLOW] = function(tx, dataflow_id, op_id, command) if not dataflow_id or dataflow_id == "" then - return nil, "Workflow ID is required" + return nil, errors.new({ message = "Workflow ID is required", kind = errors.INVALID }) end local payload = command.payload or {} local wf_id_to_delete = payload.dataflow_id or dataflow_id + if type(wf_id_to_delete) ~= "string" or wf_id_to_delete == "" then + return nil, errors.new({ message = "Workflow ID is required", kind = errors.INVALID }) + end + + -- Lock the parent before inspecting evidence or deleting any child rows. + -- A workflow consumer may still need the admission or terminal result. + local _, lock_err = activation_repo.lock_workflow_tx(tx, wf_id_to_delete) + if lock_err then return nil, lock_err end + local evidence_rows, evidence_err = sql.builder.select("admission_key", "terminal_ack_at") + :from("dataflow_activations") + :where("dataflow_id = ?", wf_id_to_delete) + :limit(1) + :run_with(tx) + :query() + if evidence_err then return nil, evidence_err end + local evidence = evidence_rows and evidence_rows[1] + if evidence and evidence.admission_key ~= nil and evidence.terminal_ack_at == nil then + return nil, errors.new({ + message = "Workflow admission evidence has not been acknowledged", + kind = errors.CONFLICT, + }) + end local wake_result, wake_err = sql.builder.delete("dataflow_wakes") :where("dataflow_id = ?", wf_id_to_delete) :run_with(tx) :exec() - if wake_err then return nil, "Failed to clear deleted dataflow wake: " .. tostring(wake_err) end + if wake_err then return nil, wake_err end local wake_index_changed = (wake_result.rows_affected or 0) > 0 local delete_query = sql.builder.delete("dataflows") @@ -1246,11 +1268,11 @@ handlers[constants.COMMAND_TYPES.DELETE_WORKFLOW] = function(tx, dataflow_id, op local result_exec, err_exec = executor:exec() if err_exec then - return nil, "Failed to delete dataflow: " .. err_exec + return nil, err_exec end if result_exec.rows_affected == 0 then - return nil, "Workflow not found" + return nil, errors.new({ message = "Workflow not found", kind = errors.NOT_FOUND }) end return { @@ -1316,6 +1338,7 @@ function ops.execute(tx, dataflow_id, op_id, commands) local result, err_handler = handler(tx, dataflow_id, op_id, command) if err_handler then + if type(err_handler) == "userdata" then return nil, err_handler end return nil, "Error executing command at index " .. i .. ": " .. err_handler end From 493fc944b24767a083012a7c847dd6d32667d9de Mon Sep 17 00:00:00 2001 From: Wolfy-J Date: Fri, 25 Sep 2026 14:40:55 -0400 Subject: [PATCH 3/3] Assert typed dataflow errors in native tests --- src/persist/activation_repo_test.lua | 3 ++- src/persist/ops_test.lua | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/src/persist/activation_repo_test.lua b/src/persist/activation_repo_test.lua index 2115a77..25a015e 100644 --- a/src/persist/activation_repo_test.lua +++ b/src/persist/activation_repo_test.lua @@ -123,7 +123,8 @@ local function define_tests() return activation_repo.request_activation_tx(tx, id, { "not", "an", "object" }, now(2)) end) test.is_nil(invalid) - test.contains(invalid_err, "plain object") + test.is_true(errors.is(invalid_err, errors.INVALID)) + test.eq(invalid_err:message(), "launch_args must be a plain object") stored = test.not_nil(select(1, activation_repo.get(id))) :: any test.eq(stored.generation, 2) end) diff --git a/src/persist/ops_test.lua b/src/persist/ops_test.lua index f0fcb50..c106a4d 100644 --- a/src/persist/ops_test.lua +++ b/src/persist/ops_test.lua @@ -1379,7 +1379,8 @@ local function define_tests() local result, err = ops.execute(tx, fake_dataflow_id, nil, delete_command) test.is_nil(result) - test.contains(err, "Workflow not found") + test.is_true(errors.is(err, errors.NOT_FOUND)) + test.eq(err:message(), "dataflow not found") end) it("should delete a specific dataflow when provided in command", function()