From d8ac99fd7b74281eb2ecc32353ce5a13bcb02bde Mon Sep 17 00:00:00 2001 From: Wolfy-J Date: Sat, 8 Aug 2026 20:46:48 -0400 Subject: [PATCH] fix(persist): the storage boundary covers every write and respects binary MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every textual write — content, metadata, config, commit payloads — passes the encoding boundary; declared octet-stream content passes byte-identical. A terminal table result that cannot encode persists as its failure text. The unconsumed execution-identity contract leaves with its stubs. --- src/_index.yaml | 27 --------- src/persist/commit_repo.lua | 1 + src/persist/encoding.lua | 11 ++++ src/persist/ops.lua | 25 ++++---- src/runner/workflow_state.lua | 8 +++ src/runner/workflow_state_test.lua | 13 ++-- test/_index.yaml | 36 ----------- test/execution_identity.lua | 95 ------------------------------ 8 files changed, 38 insertions(+), 178 deletions(-) delete mode 100644 test/execution_identity.lua diff --git a/src/_index.yaml b/src/_index.yaml index e0903db..049ae3c 100644 --- a/src/_index.yaml +++ b/src/_index.yaml @@ -113,33 +113,6 @@ entries: agent_ref: userspace.dataflow:agent_ref method: run_tests - - name: execution_identity - kind: contract.definition - meta: - title: Dataflow Execution Identity - comment: Optional app-provided bridge for durable workflow identity capture and orchestrator revival. - methods: - - name: capture - description: Capture the current execution identity as an opaque actor_id plus actor_context row. Input { reason? }. - input_schemas: - - definition: | - { "type": "object", "properties": { "reason": { "type": "string" } } } - format: application/schema+json - output_schemas: - - definition: | - { "type": "object", "properties": { "success": { "type": "boolean" }, "actor_id": { "type": "string" }, "actor_context": {}, "error": { "type": "string" } }, "required": ["success"] } - format: application/schema+json - - name: spawn_orchestrator - description: Spawn the dataflow orchestrator for a persisted dataflow under that dataflow's captured identity. Input { dataflow_id, process_id, host_id, args? }. - input_schemas: - - definition: | - { "type": "object", "properties": { "dataflow_id": { "type": "string" }, "process_id": { "type": "string" }, "host_id": { "type": "string" }, "args": { "type": "object" } }, "required": ["dataflow_id", "process_id", "host_id"] } - format: application/schema+json - output_schemas: - - definition: | - { "type": "object", "properties": { "success": { "type": "boolean" }, "pid": { "type": "string" }, "error": { "type": "string" } }, "required": ["success"] } - format: application/schema+json - - name: execution_frame kind: library.lua meta: diff --git a/src/persist/commit_repo.lua b/src/persist/commit_repo.lua index b4f6f5b..9cbc0f4 100644 --- a/src/persist/commit_repo.lua +++ b/src/persist/commit_repo.lua @@ -92,6 +92,7 @@ function commit_repo.create(commit_id, dataflow_id, payload, metadata) metadata_json = metadata end end + metadata_json = encoding.ensure_utf8(metadata_json) -- Create timestamp local created_at = time.now():format(time.RFC3339) diff --git a/src/persist/encoding.lua b/src/persist/encoding.lua index 353e1f6..ec49eb5 100644 --- a/src/persist/encoding.lua +++ b/src/persist/encoding.lua @@ -7,6 +7,17 @@ local M = {} local REPLACEMENT = "\239\191\189" -- U+FFFD +local BINARY_CONTENT_TYPE = "application/octet-stream" + +-- ensure_storable(value, content_type) -> value. The boundary is content-type +-- aware: declared binary passes through byte-identical — it lives in binary +-- columns and sanitizing it would be corruption. Everything else is text and +-- must be valid UTF-8. +function M.ensure_storable(value: any, content_type: any): any + if content_type == BINARY_CONTENT_TYPE then return value end + return M.ensure_utf8(value) +end + function M.ensure_utf8(s: any): any if type(s) ~= "string" then return s end local n = #s diff --git a/src/persist/ops.lua b/src/persist/ops.lua index c22328b..39407b6 100644 --- a/src/persist/ops.lua +++ b/src/persist/ops.lua @@ -142,9 +142,9 @@ local function replace_iteration_terminal(tx, dataflow_id, op_id, payload, data_ :set("type", payload.data_type) :set("discriminator", payload.discriminator) :set("key", payload.key) - :set("content", content_value) + :set("content", encoding.ensure_storable(content_value, content_type)) :set("content_type", content_type) - :set("metadata", metadata) + :set("metadata", encoding.ensure_utf8(metadata)) :run_with(tx) :exec() if err then return nil, "Failed to replace iteration terminal data: " .. err end @@ -317,7 +317,7 @@ handlers[constants.COMMAND_TYPES.CREATE_NODE] = function(tx, dataflow_id, op_id, parent_node_id = parent_node_id, type = payload.node_type, status = status, - config = config, + config = encoding.ensure_utf8(config), metadata = metadata, created_at = now_ts, updated_at = now_ts @@ -379,7 +379,7 @@ handlers[constants.COMMAND_TYPES.UPDATE_NODE] = function(tx, dataflow_id, op_id, end config = encoded end - update_query = update_query:set("config", config) + update_query = update_query:set("config", encoding.ensure_utf8(config)) has_update = true end @@ -464,7 +464,7 @@ handlers[constants.COMMAND_TYPES.UPDATE_NODE] = function(tx, dataflow_id, op_id, end end - update_query = update_query:set("metadata", meta_val_for_db) + update_query = update_query:set("metadata", encoding.ensure_utf8(meta_val_for_db)) has_update = true end @@ -554,11 +554,10 @@ handlers[constants.COMMAND_TYPES.CREATE_DATA] = function(tx, dataflow_id, op_id, end content_value = encoded end - -- Storage boundary: external content can carry arbitrary bytes; what is - -- written is always valid UTF-8. - content_value = encoding.ensure_utf8(content_value) - local content_type = payload.content_type or "application/json" + -- Storage boundary: external content can carry arbitrary bytes; every + -- textual write is valid UTF-8, declared binary passes byte-identical. + content_value = encoding.ensure_storable(content_value, content_type) local node_id = payload.node_id or sql.as.null() local metadata = payload.metadata or "{}" @@ -569,7 +568,7 @@ handlers[constants.COMMAND_TYPES.CREATE_DATA] = function(tx, dataflow_id, op_id, end metadata = encoded end - + metadata = encoding.ensure_utf8(metadata) if is_iteration_terminal_unique_slot(payload) then local existing_row, existing_err = find_existing_unique_data_row(tx, dataflow_id, payload) @@ -732,7 +731,7 @@ handlers[constants.COMMAND_TYPES.UPDATE_DATA] = function(tx, dataflow_id, op_id, content_value = encoded end - update_query = update_query:set("content", content_value) + update_query = update_query:set("content", encoding.ensure_storable(content_value, payload.content_type)) has_update = true end @@ -751,7 +750,7 @@ handlers[constants.COMMAND_TYPES.UPDATE_DATA] = function(tx, dataflow_id, op_id, metadata = encoded end - update_query = update_query:set("metadata", metadata) + update_query = update_query:set("metadata", encoding.ensure_utf8(metadata)) has_update = true end @@ -1084,7 +1083,7 @@ handlers[constants.COMMAND_TYPES.UPDATE_WORKFLOW] = function(tx, dataflow_id, op end end - update_query_builder = update_query_builder:set("metadata", meta_val_for_db) + update_query_builder = update_query_builder:set("metadata", encoding.ensure_utf8(meta_val_for_db)) has_real_update_field = true end diff --git a/src/runner/workflow_state.lua b/src/runner/workflow_state.lua index 15201eb..0ccedb7 100644 --- a/src/runner/workflow_state.lua +++ b/src/runner/workflow_state.lua @@ -1182,6 +1182,14 @@ function methods:handle_process_exit(pid, success, result) and content_kind ~= "boolean" and content_kind ~= "table" then result_content = result_content ~= nil and tostring(result_content) or nil end + if content_kind == "table" then + -- A table must actually encode (cycles do not); an unserializable + -- result persists as its failure text instead of poisoning the batch. + local ok, _, encode_err = pcall(function() return json.encode(result_content) end) + if not ok or encode_err ~= nil then + result_content = "unserializable node result: " .. tostring(result) + end + end if result_content == nil then result_content = success and "Completed" or "Failed" end diff --git a/src/runner/workflow_state_test.lua b/src/runner/workflow_state_test.lua index 4bfcd19..52177cf 100644 --- a/src/runner/workflow_state_test.lua +++ b/src/runner/workflow_state_test.lua @@ -860,13 +860,14 @@ local function define_tests() test.eq(type(content), "string") end) - it("persists a table result whose values cannot encode as its string form", function() + it("persists a cyclic table result as its failure text", function() local ws = workflow_state.new(test_ctx.dataflow_id) :: any ws.nodes["node-1"] = { status = consts.STATUS.RUNNING, type = "test_node" } ws:track_process("node-1", "pid-123") - local poison = { message = "boom", raw = coroutine.create(function() end) } - local exit_info = ws:handle_process_exit("pid-123", false, poison) :: any + local cyclic: any = { message = "boom" } + cyclic.self = cyclic + local exit_info = ws:handle_process_exit("pid-123", false, cyclic) :: any test.not_nil(exit_info) local content = nil @@ -874,10 +875,8 @@ local function define_tests() local p = (cmd :: any).payload or {} if p.data_type == consts.DATA_TYPE.NODE_RESULT then content = p.content end end - test.not_nil(content) - local encoded, encode_err = json.encode(content) - test.is_nil(encode_err) - test.not_nil(encoded) + test.eq(type(content), "string") + test.is_true((content :: string):find("unserializable", 1, true) ~= nil) end) end) diff --git a/test/_index.yaml b/test/_index.yaml index 41d2f7c..6df5748 100644 --- a/test/_index.yaml +++ b/test/_index.yaml @@ -156,42 +156,6 @@ entries: test_only: true comment: Named policy group for test-only process services. - - name: execution_identity.capture - kind: function.lua - meta: - environment: test - test_only: true - comment: Captures the explicit test actor and policy used by the test runner. - source: file://execution_identity.lua - method: capture - modules: &execution_identity_modules [json, process, security] - imports: &execution_identity_imports - dataflow_repo: userspace.dataflow.persist:dataflow_repo - - - name: execution_identity.spawn_orchestrator - kind: function.lua - meta: - environment: test - test_only: true - comment: Reconstructs the persisted test identity for durable wake delivery. - source: file://execution_identity.lua - method: spawn_orchestrator - modules: *execution_identity_modules - imports: *execution_identity_imports - - - name: execution_identity.binding - kind: contract.binding - meta: - environment: test - test_only: true - comment: Test-only implementation of Dataflow's application-owned identity bridge. - contracts: - - contract: userspace.dataflow:execution_identity - default: true - methods: - capture: app:execution_identity.capture - spawn_orchestrator: app:execution_identity.spawn_orchestrator - - name: gateway kind: http.service addr: :19097 diff --git a/test/execution_identity.lua b/test/execution_identity.lua deleted file mode 100644 index 95cc4b5..0000000 --- a/test/execution_identity.lua +++ /dev/null @@ -1,95 +0,0 @@ -local json = require("json") -local process = require("process") -local security = require("security") -local dataflow_repo = require("dataflow_repo") - -local M = {} - -local function failure(message: any): any - return { success = false, error = tostring(message or "unknown error") } -end - -local function policy_ids(scope: any): (string[]?, string?) - if not scope or type(scope.policies) ~= "function" then - return nil, "current scope is unavailable" - end - local ids = {} :: string[] - for _, policy in ipairs(scope:policies()) do - local id = policy and tostring(policy:id()) or "" - if id == "" then return nil, "scope contains an invalid policy" end - ids[#ids + 1] = id - end - table.sort(ids) - return ids, nil -end - -local function reconstruct_scope(ids: any): (any?, string?) - if type(ids) ~= "table" then return nil, "persisted policies are invalid" end - local policies = {} :: security.Policy[] - for _, id in ipairs(ids) do - local policy, policy_err = security.policy(tostring(id)) - if policy_err or not policy then - return nil, tostring(policy_err or ("persisted policy is unavailable: " .. tostring(id))) - end - policies[#policies + 1] = policy - end - local scope, scope_err = security.new_scope(policies) - if scope_err or not scope then - return nil, tostring(scope_err or "test scope is unavailable") - end - return scope, nil -end - -function M.capture(): any - local actor = security.actor() - local actor_id = actor and tostring(actor:id()) or "" - if actor_id == "" then return failure("current actor is unavailable") end - local ids, ids_err = policy_ids(security.scope()) - if ids_err or not ids then return failure(ids_err) end - - local actor_context, context_err = json.encode({ policies = ids }) - if context_err or not actor_context then - return failure(context_err or "identity encoding failed") - end - return { - success = true, - actor_id = actor_id, - actor_context = actor_context, - } -end - -function M.spawn_orchestrator(args: any): any - if type(args) ~= "table" then return failure("args must be a table") end - local dataflow_id = tostring(args.dataflow_id or "") - local process_id = tostring(args.process_id or "") - local host_id = tostring(args.host_id or "") - if dataflow_id == "" then return failure("dataflow_id is required") end - if process_id == "" then return failure("process_id is required") end - if host_id == "" then return failure("host_id is required") end - - local workflow, workflow_err = dataflow_repo.get(dataflow_id) - if workflow_err or type(workflow) ~= "table" then - return failure(workflow_err or "dataflow not found") - end - - local actor_id = tostring(workflow.actor_id or "") - if actor_id == "" then return failure("persisted actor is unavailable") end - local actor_context, context_err = json.decode(tostring(workflow.actor_context or "")) - if context_err or type(actor_context) ~= "table" or type(actor_context.policies) ~= "table" then - return failure(context_err or "persisted test identity is invalid") - end - - local actor, actor_err = security.new_actor(actor_id, {}) - if actor_err or not actor then return failure(actor_err or "actor reconstruction failed") end - local scope, scope_err = reconstruct_scope(actor_context.policies) - if scope_err or not scope then return failure(scope_err) end - - local pid, spawn_err = process.with_context({}) - :with_actor(actor) - :with_scope(scope) - :spawn(process_id, host_id, args.args or { dataflow_id = dataflow_id }) - if spawn_err or not pid then return failure(spawn_err or "spawn failed") end - return { success = true, pid = tostring(pid) } -end - -return M