From e69198b8b1f721e2cba46803932b4ac8f8a89f2d Mon Sep 17 00:00:00 2001 From: crprashant <5108573+crprashant@users.noreply.github.com> Date: Mon, 22 Jun 2026 18:01:21 -0700 Subject: [PATCH] Harden instance/node ID collisions with retry + composite PK (#129) Maintainer-directed minimal change for #129 / PR #238: df.start() now retries instance-ID generation on collision (the ID stays VARCHAR(8) HEX, unchanged), and df.nodes uses a composite PRIMARY KEY (instance_id, id) so node IDs only need to be unique per instance rather than globally. The legacy nodes_instance_node_key UNIQUE constraint is promoted to the primary key. Collision handling is atomic: both the instance reserve and every node insert claim their ID via INSERT ... ON CONFLICT DO NOTHING RETURNING id and re-roll when zero rows return, so there is no check-then-insert TOCTOU window. The shared pick_id_with_retry helper treats the claim as the loop tail and returns an error on exhaustion, never an unverified ID. The instance row is reserved up front with its root_node bound to a pre-generated ID that is then forced onto the root node, satisfying the deferred same-instance FK at commit without an extra UPDATE (ordinary df roles cannot UPDATE df.instances.root_node). update_node_status now requires instance_id and always scopes its UPDATE by (instance_id, id), removing the global-ID fallback and asserting exactly one row is affected. Adding instance_id to the activity input is a duroxide replay-breaking change for orchestrations in flight across the 0.2.3 -> 0.2.4 binary upgrade, so operators must drain or recreate in-flight instances before upgrading. This is documented in docs/upgrade-testing.md alongside the instance-retry vs node-composite-PK asymmetry rationale and ADD PRIMARY KEY lock guidance. Add e2e tests 50 (composite-PK schema contract + instance_id-scoped node-status regression) and 51 (cross-instance node-ID collision). Update CHANGELOG and the E2E test inventory. --- CHANGELOG.md | 2 + docs/E2E_TESTING.md | 8 +- docs/upgrade-testing.md | 13 +- sql/pg_durable--0.2.3--0.2.4.sql | 55 +++ src/activities/update_node_status.rs | 81 ++-- src/dsl.rs | 408 +++++++++++++----- src/lib.rs | 12 +- src/orchestrations/execute_function_graph.rs | 2 + src/types.rs | 18 +- tests/e2e/sql/51_node_composite_pk.sql | 126 ++++++ .../52_node_id_collision_across_instances.sql | 87 ++++ 11 files changed, 670 insertions(+), 142 deletions(-) create mode 100644 tests/e2e/sql/51_node_composite_pk.sql create mode 100644 tests/e2e/sql/52_node_id_collision_across_instances.sql diff --git a/CHANGELOG.md b/CHANGELOG.md index 74a6086b..445c1a2e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,8 @@ Pre-1.0 note: while `pg_durable` is in major version `0`, minor releases may inc > ⚠️ **Replay-breaking for in-flight `wait_for_schedule` instances.** This change adds a recorded `utc_now()` decision before the WAIT_SCHEDULE timer, altering the orchestration's history sequence. Any durable function that was started under a `<= 0.2.3` binary and is **mid-`wait_for_schedule`** (parked on its timer) when this `.so` is loaded will fail with a duroxide nondeterminism error on replay, because its recorded history no longer matches the new code. Drain or allow such in-flight `wait_for_schedule` instances to complete before upgrading. Instances that are not currently inside a `wait_for_schedule` node are unaffected. We accepted this break (rather than introducing orchestration versioning) given the early pre-1.0 stage of the project. +- **Instance/node ID collision hardening (#129):** `df.start()` now reserves IDs with `INSERT ... ON CONFLICT DO NOTHING RETURNING id` and re-rolls the random 8-hex value on collision — instances arbitrate on the `df.instances` primary key (`id`), nodes on the new composite `PRIMARY KEY (instance_id, id)` — replacing the previous `SELECT EXISTS` pre-check. Doing the conflict check at the index level (rather than a pre-check `SELECT`) closes a TOCTOU window and, for instances, an RLS blind spot where the pre-check could not see another role's rows. `df.nodes` now uses the composite `PRIMARY KEY (instance_id, id)` instead of a global single-column key, so the random 8-hex node ID is no longer the sole cross-instance collision guard. The `update-node-status` activity now scopes its `df.nodes` update by `instance_id` (a required activity-input field) and asserts it affects exactly one row. IDs stay `VARCHAR(8)` HEX; the `0.2.3 → 0.2.4` upgrade restructures the `df.nodes` keys in place (#238). + - **Breaking for in-flight work:** the new activity-input shape changes the string duroxide records in orchestration history, and duroxide validates activity inputs by exact equality on replay, so any instance left **in flight across the 0.2.3 → 0.2.4 binary upgrade** cannot complete. Drain or cancel in-flight instances before deploying 0.2.4. The in-place `df.nodes` key restructure also takes an `ACCESS EXCLUSIVE` lock whose duration scales with table size — run the upgrade in a maintenance window. See the #129 section of `docs/upgrade-testing.md` for the full drain-before-upgrade contract. - **`df.grant_usage()` / `df.revoke_usage()`:** dropped the explicit per-function `EXECUTE` allowlist. Schema `USAGE` on `df` is the real access gate for ordinary `df.*` functions, so the helpers now grant/revoke schema `USAGE`, the table privileges, and `EXECUTE` only on the sensitive functions (`df.http`, `df.grant_usage`, `df.revoke_usage`). Function signatures are unchanged and existing privileges are unaffected (#242). ### Removed diff --git a/docs/E2E_TESTING.md b/docs/E2E_TESTING.md index 28230ed6..e8323880 100644 --- a/docs/E2E_TESTING.md +++ b/docs/E2E_TESTING.md @@ -58,6 +58,7 @@ The test suite is organized into 23 files. Files `01`–`09` open with `SET SESS | `07_signals.sql` | `df.signal()` — send signals to a running workflow from within the polling loop | | `08_scenarios.sql` | End-to-end workflow scenarios using `playground.*` tables (ETL, parallel counts, conditional load, order processing, three-step) | | `09_graph_and_validation.sql` | `df.explain()` graph reuse, invalid `node_type` rejection | +| `51_node_composite_pk.sql` | `df.nodes` composite PRIMARY KEY `(instance_id, id)` — schema contract (legacy `nodes_instance_node_key` UNIQUE promoted to the PK) and multi-node workflow regression under `instance_id`-scoped node-status updates and `df.result()` (issue #129) | ### Superuser Tests (runs as `postgres`) @@ -70,6 +71,7 @@ The test suite is organized into 23 files. Files `01`–`09` open with `SET SESS | `14_database.sql` | Wrong-database `CREATE EXTENSION` rejection; `df.start(query, label, database)` multi-database routing | | `15_rls.sql` | RLS on `df.instances` / `df.nodes` / `df.vars` — per-user visibility, cross-user cancel/signal denied, column-level UPDATE, superuser bypass, per-user variable isolation | | `16_heartbeat.sql` | Worker heartbeat liveness — `df._worker_epoch.last_seen_at` advances over time | +| `52_node_id_collision_across_instances.sql` | Cross-instance node-ID collision — two instances own the same 8-hex node id; asserts composite-PK coexistence, that `(instance_id, id)` addresses exactly one row, `df.result()` is instance-scoped, and a scoped `update_node_status`-style UPDATE affects exactly one row (issue #129) | ### Build-Phase Specific @@ -99,7 +101,11 @@ pg_durable/ ├── 45_connection_limit_timeout.sql ├── 46_connection_limit_startup_validation.sql ├── 47_http_dsl_disabled.sql - └── 48_http_allow_all.sql + ├── 48_http_allow_all.sql + ├── 49_quoted_role_names.sql + ├── 50_metrics_grants.sql + ├── 51_node_composite_pk.sql + └── 52_node_id_collision_across_instances.sql ``` ## Writing New Tests diff --git a/docs/upgrade-testing.md b/docs/upgrade-testing.md index 586e359c..3008e633 100644 --- a/docs/upgrade-testing.md +++ b/docs/upgrade-testing.md @@ -73,7 +73,7 @@ We test against all previous versions in the same provider compatibility line. T | DSL construction | `df.sql()`, `df.seq()`, `df.if()`, `df.loop()`, `df.sleep()`, `df.http()` | | Execution | Starting and completing orchestrations | | Monitoring | `df.status()`, `df.result()`, `df.list_instances()`, `df.instance_info()` | -| In-flight work | Orchestrations started before `.so` swap complete after swap | +| In-flight work | Orchestrations started before `.so` swap complete after swap (except across an activity-input change — see #129) | **What it catches:** - SQL queries in Rust code referencing columns/constraints that don't exist in the old schema @@ -100,7 +100,7 @@ This is a **chain test** (like Scenario A) — upgrade scripts are applied seque |------|---------------| | Variables | Pre-existing vars accessible via `df.getvar()` after upgrade | | Pre-existing instances | `df.result()`, `df.instance_info()`, and `df.list_instances()` work for instances created before upgrade | -| In-flight work | Work started before `ALTER EXTENSION UPDATE` can still complete afterward | +| In-flight work | Work started before `ALTER EXTENSION UPDATE` can still complete afterward (except across an activity-input change — see #129) | | New operations | `df.start()` works with new schema | **Priority:** High — validates the upgrade doesn't corrupt or lose existing data. @@ -229,6 +229,15 @@ what the upgrade script handles, and any backward compatibility considerations. - **Scenario B2 considerations:** No data migration. Existing instances, nodes, and vars are untouched. After `ALTER EXTENSION UPDATE`, `df.debug_connection()` no longer exists; the simplified `df.grant_usage()` never references it. - **Dependent-object note:** The upgrade runs `DROP FUNCTION IF EXISTS df.debug_connection()` with PostgreSQL's default `RESTRICT` behavior. If a customer created their own object that depends on the function (e.g. a view or SQL function that calls it), `ALTER EXTENSION UPDATE` aborts with a dependency error and the customer must drop or repoint that object first. This is intentional for a removed debug helper — the script deliberately does not `CASCADE`, to avoid silently dropping customer-owned objects. The fresh-install (`tests/e2e/sql/18_delegated_grants.sql`) and upgrade (`scripts/test-upgrade.sh` B2 grant test) suites assert the function is absent and that `df.grant_usage()` still works after the drop. +#### #129 Promote df.nodes to a composite primary key (instance_id, id) +- **DDL change (df schema):** `df.nodes` previously had a single-column `PRIMARY KEY (id)` plus a separate composite `UNIQUE (instance_id, id)` (`nodes_instance_node_key`). The single-column key forced the random 8-hex node ID to be globally unique, so it was the sole cross-instance collision guard. Node IDs only need to be unique per instance, so the composite key is promoted to be the primary key and the global single-column key is dropped. Fresh installs (`src/lib.rs`) declare `id`/`instance_id` as `NOT NULL` and create `nodes_pkey PRIMARY KEY (instance_id, id)` directly; the upgrade script (`sql/pg_durable--0.2.3--0.2.4.sql`) restructures the existing keys in place. The three same-instance foreign keys (`nodes_left_node_same_instance_fkey`, `nodes_right_node_same_instance_fkey`, `instances_root_node_same_instance_fkey`) reference the composite key, so the upgrade drops them first, swaps the keys, then recreates them with their original `DEFERRABLE INITIALLY DEFERRED NOT VALID` definition. `nodes_instance_identity_fkey` references `df.instances`, not `df.nodes`, and is left untouched. IDs remain `VARCHAR(8)` HEX. +- **Companion runtime change (#129):** `df.start()` now reserves the instance ID by attempting the insert itself — `INSERT INTO df.instances ... ON CONFLICT (id) DO NOTHING RETURNING id` — and re-rolling the random 8-hex ID when zero rows come back (a collision); there is no separate `SELECT EXISTS` pre-check. Because `ON CONFLICT` arbitration runs against the global `id` index *below* row-level security, this also re-rolls on collisions with another role's instance that the caller cannot `SELECT`. Node inserts use the same pattern against the composite key — `INSERT INTO df.nodes ... ON CONFLICT (instance_id, id) DO NOTHING RETURNING id` — re-rolling on a per-instance collision. `df.start()` pre-generates the root node's ID and reserves the instance with `root_node` set to that value; `insert_nodes` then inserts the root node with the same forced ID. The same-instance FK on `root_node` is `DEFERRABLE INITIALLY DEFERRED`, so it is checked only at commit, by which point the referenced root node row exists — no post-insert `UPDATE` is needed (and `df.grant_usage()` deliberately grants `UPDATE (status, updated_at)` but not `UPDATE (root_node)` on `df.instances`, so an update path would fail for ordinary df roles). The `update-node-status` activity and `df.result()` now scope their `df.nodes` lookups by `instance_id` in addition to `id`, and the activity asserts the scoped `UPDATE` affects exactly one row. `instance_id` is a **required** field of the activity input — node IDs are unique only per instance, so updating by node ID alone could silently write to a *different* instance's node. There is deliberately no node-ID-only fallback. +- **Design note — collision handling for both ID spaces (#129):** Both IDs stay 8-hex `VARCHAR(8)` (the requested minimal change) and re-roll on conflict via `INSERT ... ON CONFLICT DO NOTHING RETURNING id`; the mechanism is symmetric and only the conflict target differs. `df.instances.id` is a *global* identifier with no natural scoping column, so its reserve arbitrates on the single-column primary key (`id`). `df.nodes.id` is always used together with its owning `instance_id`, so promoting the pre-existing `(instance_id, id)` UNIQUE to the primary key lets node inserts arbitrate per instance — the random node ID never has to be globally unique. Using `ON CONFLICT DO NOTHING` rather than a `SELECT EXISTS` pre-check closes a TOCTOU window and, for instances, an RLS blind spot: the pre-check only saw the caller's own rows, whereas `ON CONFLICT` detects a clash with any role's row at the index level. The retry bound (`MAX_ID_ATTEMPTS`) surfaces a hard error on exhaustion rather than returning an unverified ID. +- **In-flight orchestration compatibility (#129 — breaking for in-flight work):** Adding `instance_id` to the `update-node-status` activity input changes the input string that duroxide records in orchestration history. duroxide validates activity inputs by exact equality during replay, so any orchestration that was **in flight across the binary upgrade** (it recorded the old `{node_id, status}` input under 0.2.3) fails deterministic replay under the new `.so` and cannot complete. This is an intentional break of the general "in-flight work completes after the swap" expectation (the Scenario B1 and B2 "In-flight work" rows above) **for this release**, and follows the same drain-or-recreate precedent as the v0.1.0 → v0.1.1 execution-model change (Scenario B2, below): **operators must drain in-flight instances to a terminal state before deploying 0.2.4**, or cancel and recreate any that cannot drain. Instances that completed before the upgrade are terminal and unaffected; instances started after the upgrade carry `instance_id` from their first node update and replay normally. +- **Scenario A considerations:** Fresh-install and upgraded schemas must both end with exactly one identity constraint on `df.nodes`: `nodes_pkey PRIMARY KEY (instance_id, id)` (constraint key order `instance_id, id`), its matching unique index `nodes_pkey ON df.nodes USING btree (instance_id, id)`, and no surviving `nodes_instance_node_key` constraint or index. The recreated foreign keys keep identical names and referencing columns, so the constraint/index snapshot diff is empty. +- **Scenario B1 considerations:** The schema change is to table constraints only; the new `.so` issues the same column lists against `df.nodes`/`df.instances`, now with `ON CONFLICT ... DO NOTHING RETURNING id`. The instance reserve arbitrates on `id` (the primary key in both old and new schemas) and the node insert arbitrates on `(instance_id, id)` — an index that exists in both the pre-0.2.4 schema (the `nodes_instance_node_key` composite UNIQUE) and the new schema (the composite primary key) — so both statements stay valid against a schema that has not run `ALTER EXTENSION UPDATE`. The pre-generated-`root_id` reserve is also old-schema-safe: `instances_root_node_same_instance_fkey` is `DEFERRABLE INITIALLY DEFERRED` in every shipped schema, so `root_node` is not checked until commit, by which point the forced-ID root node row has been inserted within the same transaction. No `UPDATE df.instances` is issued, so the change relies only on the `INSERT (..., root_node, ...)` privilege every shipped `df.grant_usage()` already grants, not on any `UPDATE (root_node)` grant. One benign residual exists against the *old* schema only: a node ID that is globally duplicated but per-instance-unique would clash with the surviving single-column `nodes_pkey (id)`, which `ON CONFLICT (instance_id, id)` does not arbitrate, so it raises just as it did before this change — astronomically rare, strictly no worse than prior behavior, and eliminated once `ALTER EXTENSION UPDATE` swaps in the composite primary key. This covers **schema** compatibility only — the SQL stays valid against the old table shape. The separate in-flight *replay* break introduced by the changed activity-input shape is documented under "In-flight orchestration compatibility" above and requires draining before upgrade. +- **Scenario B2 considerations:** `ADD PRIMARY KEY (instance_id, id)` sets `NOT NULL` on both columns and builds a unique index over existing rows. `id` was already the old primary key (implicitly `NOT NULL`). `instance_id` carries a `nodes_instance_id_present_chk CHECK (instance_id IS NOT NULL)` constraint, but it was added `NOT VALID`, so it only guarantees rows written on 0.2.2+; in the unlikely event a database still holds pre-0.2.2 node rows with a NULL `instance_id`, the `ADD PRIMARY KEY` (and the explicit `ALTER COLUMN instance_id SET NOT NULL` that precedes it) will abort and the operator must backfill or remove those rows before retrying the upgrade. On an empty database the restructure is metadata-only; on a populated one PostgreSQL rebuilds the `df.nodes` primary-key index in place. Because `ADD PRIMARY KEY` / `ALTER COLUMN ... SET NOT NULL` take an `ACCESS EXCLUSIVE` lock on `df.nodes` and rebuild the index, on a large `df.nodes` the upgrade blocks concurrent access for a period that scales with the table's size; run `ALTER EXTENSION UPDATE` inside a maintenance window and consider `SET lock_timeout` for the session so the migration fails fast instead of queuing behind (or stalling in front of) long-running transactions. Combined with the in-flight replay break noted above, the recommended upgrade sequence is: stop new `df.start()` calls, drain or cancel in-flight instances, then run the upgrade. + ### v0.2.2 → v0.2.3 #### Rename duroxide provider schema to `_duroxide` for fresh installs diff --git a/sql/pg_durable--0.2.3--0.2.4.sql b/sql/pg_durable--0.2.3--0.2.4.sql index 5d86e52f..93380f42 100644 --- a/sql/pg_durable--0.2.3--0.2.4.sql +++ b/sql/pg_durable--0.2.3--0.2.4.sql @@ -181,3 +181,58 @@ CREATE FUNCTION df."await_instance"( STRICT LANGUAGE c AS 'MODULE_PATHNAME', 'await_instance_wrapper'; + +-- ============================================================================ +-- Promote df.nodes to a composite primary key (instance_id, id) (issue #129). +-- +-- The single-column PRIMARY KEY (id) forced node IDs to be globally unique, so +-- the random 8-hex node ID was the sole collision guard across every instance. +-- Node IDs only need to be unique per instance, so the existing composite +-- UNIQUE (instance_id, id) — already referenced by the same-instance foreign +-- keys — is promoted to be the primary key and the global single-column key is +-- dropped. This matches the fresh-install schema in src/lib.rs so a fresh +-- install and an upgraded database end with identical df.nodes constraints. +-- +-- The three same-instance foreign keys reference the composite key, so +-- PostgreSQL will not allow dropping it (nor the old single-column PRIMARY KEY) +-- while those foreign keys exist. Drop them first, restructure the keys, then +-- recreate the foreign keys against the new primary key. The recreated foreign +-- keys keep their original DEFERRABLE INITIALLY DEFERRED NOT VALID definition. +-- +-- nodes_instance_identity_fkey references df.instances, not df.nodes, so it is +-- left untouched. ADD PRIMARY KEY (instance_id, id) sets NOT NULL on both +-- columns: id was already the old primary key (implicitly NOT NULL), and +-- instance_id carries nodes_instance_id_present_chk CHECK (instance_id IS NOT +-- NULL). That check is NOT VALID, so it only guarantees rows written on 0.2.2+; +-- in the unlikely event a database still holds pre-0.2.2 rows with a NULL +-- instance_id, the ALTER COLUMN ... SET NOT NULL below will abort and the +-- operator must backfill or remove those rows before retrying the upgrade. +-- ============================================================================ +ALTER TABLE df.nodes DROP CONSTRAINT nodes_left_node_same_instance_fkey; +ALTER TABLE df.nodes DROP CONSTRAINT nodes_right_node_same_instance_fkey; +ALTER TABLE df.instances DROP CONSTRAINT instances_root_node_same_instance_fkey; + +ALTER TABLE df.nodes DROP CONSTRAINT nodes_instance_node_key; +ALTER TABLE df.nodes DROP CONSTRAINT nodes_pkey; + +ALTER TABLE df.nodes + ALTER COLUMN id SET NOT NULL, + ALTER COLUMN instance_id SET NOT NULL, + ADD CONSTRAINT nodes_pkey + PRIMARY KEY (instance_id, id); + +ALTER TABLE df.nodes + ADD CONSTRAINT nodes_left_node_same_instance_fkey + FOREIGN KEY (instance_id, left_node) + REFERENCES df.nodes (instance_id, id) + DEFERRABLE INITIALLY DEFERRED NOT VALID, + ADD CONSTRAINT nodes_right_node_same_instance_fkey + FOREIGN KEY (instance_id, right_node) + REFERENCES df.nodes (instance_id, id) + DEFERRABLE INITIALLY DEFERRED NOT VALID; + +ALTER TABLE df.instances + ADD CONSTRAINT instances_root_node_same_instance_fkey + FOREIGN KEY (id, root_node) + REFERENCES df.nodes (instance_id, id) + DEFERRABLE INITIALLY DEFERRED NOT VALID; diff --git a/src/activities/update_node_status.rs b/src/activities/update_node_status.rs index 8ef7f28f..ca751429 100644 --- a/src/activities/update_node_status.rs +++ b/src/activities/update_node_status.rs @@ -22,44 +22,73 @@ pub async fn execute( let node_id = input["node_id"].as_str().ok_or("Missing node_id")?; let status = input["status"].as_str().ok_or("Missing status")?; let result = input.get("result").and_then(|r| r.as_str()); + // instance_id scopes the UPDATE to the owning instance and is REQUIRED. + // Node IDs are only unique per instance (issue #129 / composite PK + // (instance_id, id) on df.nodes), so updating by node_id alone could touch a + // different instance's node -- a fail-open cross-instance corruption path. + // The scope must travel through the activity input: duroxide's + // ctx.instance_id() returns the *orchestration* id (an auto-generated token + // for parallel/loop subtrees), not the df instance id, so it cannot be used + // here. The serialized graph carried in the input preserves the df id. + // + // Upgrade note: duroxide compares activity inputs by exact equality during + // replay, so adding instance_id changes the recorded input shape. Instances + // in flight across the binary upgrade recorded the old shape and cannot be + // replayed -- they must be drained/recreated before upgrading (see + // docs/upgrade-testing.md, issue #129 section). Every post-upgrade instance + // carries instance_id from the start, so requiring it here is safe and there + // is deliberately no node_id-only fallback (it would be dead code that only + // re-opened the corruption path above). + let instance_id = input["instance_id"].as_str().ok_or("Missing instance_id")?; - let query = if let Some(res) = result { - // The result column is JSONB, so normalize invalid JSON payloads into - // a JSON string before binding. - let json_result = serde_json::from_str::(res) - .unwrap_or_else(|_| serde_json::Value::String(res.to_string())); - - sqlx::query( - "UPDATE df.nodes + // The UPDATE is always scoped by (id, instance_id), which the composite + // primary key makes unique, so it can affect at most one row. Positional + // placeholders match the bind order below. + let sql: &str = if result.is_some() { + "UPDATE df.nodes SET status = $1, result = $2::jsonb, updated_at = now() - WHERE id = $3", - ) - .bind(status) - .bind(json_result) - .bind(node_id) + WHERE id = $3 AND instance_id = $4" } else if status == "running" { // When marking as running, clear any stale result from a previous // loop iteration to satisfy the constraint: // (result IS NULL OR status IN ('completed', 'failed')) - sqlx::query( - "UPDATE df.nodes + "UPDATE df.nodes SET status = $1, result = NULL, updated_at = now() - WHERE id = $2", - ) - .bind(status) - .bind(node_id) + WHERE id = $2 AND instance_id = $3" } else { - sqlx::query( - "UPDATE df.nodes + "UPDATE df.nodes SET status = $1, updated_at = now() - WHERE id = $2", - ) - .bind(status) - .bind(node_id) + WHERE id = $2 AND instance_id = $3" }; + let mut query = sqlx::query(sql).bind(status); + if let Some(res) = result { + // The result column is JSONB, so normalize invalid JSON payloads into + // a JSON string before binding. + let json_result = serde_json::from_str::(res) + .unwrap_or_else(|_| serde_json::Value::String(res.to_string())); + query = query.bind(json_result); + } + query = query.bind(node_id).bind(instance_id); + match query.execute(pool.as_ref()).await { - Ok(_) => Ok("Node status updated".to_string()), + Ok(done) => { + let rows = done.rows_affected(); + if rows == 1 { + Ok("Node status updated".to_string()) + } else { + // Exactly one row must match (instance_id, id). Anything else + // (typically zero rows: a missing node or a mismatched + // instance_id) is a correctness violation we must surface rather + // than silently swallow. + let err_msg = format!( + "update_node_status affected {rows} row(s) for node {node_id} \ + in instance {instance_id} (expected exactly 1)" + ); + ctx.trace_info(&err_msg); + Err(err_msg) + } + } Err(e) => { let err_msg = format!("Failed to update node status: {e}"); ctx.trace_info(&err_msg); diff --git a/src/dsl.rs b/src/dsl.rs index 58f7d2be..9a3ef226 100644 --- a/src/dsl.rs +++ b/src/dsl.rs @@ -634,6 +634,39 @@ pub fn signal(instance_id: &str, signal_name: &str, signal_data: default!(&str, // Orchestration Control Functions // ============================================================================ +/// Maximum number of attempts to generate a collision-free random ID before +/// giving up. The 8-hex ID space (`short_id`) makes collisions rare, so a small +/// bound is plenty; exhausting it signals either an astronomically unlucky run +/// or a genuinely saturated ID space, both of which should surface as an error +/// rather than an unverified ID. +const MAX_ID_ATTEMPTS: usize = 10; + +/// Generate a random ID and claim it, retrying on collision (issue #129). +/// +/// `generate` produces a fresh candidate ID; `try_claim` attempts to durably +/// reserve it, returning `Ok(true)` when the candidate was claimed, `Ok(false)` +/// when it collided with an existing ID (re-roll), or `Err` when the claim +/// failed for any other reason (propagated immediately). +/// +/// The claim — not the generation — is the loop tail, so this only ever returns +/// an ID that `try_claim` confirmed was inserted. On exhaustion it returns an +/// `Err` rather than a last, unverified candidate (review finding C1). +fn pick_id_with_retry( + mut generate: impl FnMut() -> String, + mut try_claim: impl FnMut(&str) -> Result, + max_attempts: usize, +) -> Result { + for _ in 0..max_attempts { + let candidate = generate(); + if try_claim(&candidate)? { + return Ok(candidate); + } + } + Err(format!( + "exhausted {max_attempts} attempts to generate a collision-free ID" + )) +} + /// Starts a durable SQL function. /// The fut argument can be either Durofut JSON or plain SQL string (auto-wrapped). /// Variables from df.vars are captured and passed to the orchestration. @@ -653,7 +686,10 @@ pub fn start( if let Err(e) = durofut.validate_recursive() { pgrx::error!("Invalid durable function graph: {}", e); } - let instance_id = short_id(); + // The instance ID is reserved later (after identity validation), using an + // INSERT ... ON CONFLICT (id) DO NOTHING retry so collisions — including + // against other roles' instances invisible under RLS — re-roll instead of + // surfacing a primary-key error (issue #129). // Validate that the target database exists (if specified) if let Some(db) = database { @@ -729,6 +765,7 @@ pub fn start( fn insert_nodes( node: &Durofut, instance_id: &str, + force_id: Option<&str>, current_user_oid: pgrx::pg_sys::Oid, database: Option<&str>, legacy_login_role: bool, @@ -742,13 +779,12 @@ pub fn start( crate::types::MAX_GRAPH_NODES ); } - let node_id = short_id(); - // Recursively insert children FIRST to get their IDs let left_id = node.left_node.as_ref().map(|n| { insert_nodes( n, instance_id, + None, current_user_oid, database, legacy_login_role, @@ -759,6 +795,7 @@ pub fn start( insert_nodes( n, instance_id, + None, current_user_oid, database, legacy_login_role, @@ -771,6 +808,7 @@ pub fn start( Ok(insert_nodes( child, instance_id, + None, current_user_oid, database, legacy_login_role, @@ -781,125 +819,215 @@ pub fn start( Err(e) => pgrx::error!("Invalid config in {} node: {}", node.node_type, e), }; - // Build parameterized args for the INSERT - let query_arg: DatumWithOid = match &query_val { - Some(q) => q.as_str().into(), - None => DatumWithOid::null::(), - }; - let result_name_arg: DatumWithOid = match &node.result_name { - Some(n) => n.as_str().into(), - None => DatumWithOid::null::(), - }; - let left_node_arg: DatumWithOid = match &left_id { - Some(id) => id.as_str().into(), - None => DatumWithOid::null::(), - }; - let right_node_arg: DatumWithOid = match &right_id { - Some(id) => id.as_str().into(), - None => DatumWithOid::null::(), - }; - let database_arg: DatumWithOid = match database { - Some(db) => db.into(), - None => DatumWithOid::null::(), - }; - - // Insert this node with parameterized query - // B1 backward compat: v0.1.x schema has login_role NOT NULL on - // df.nodes; include it (= submitted_by) so the INSERT succeeds. - let (node_sql, node_args): (&str, Vec) = if legacy_login_role { - ( - "INSERT INTO df.nodes (id, instance_id, node_type, query, result_name, left_node, right_node, submitted_by, login_role, database) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8::oid::regrole, $9::oid::regrole, $10)", - vec![ - node_id.as_str().into(), - instance_id.into(), - node.node_type.as_str().into(), - query_arg, - result_name_arg, - left_node_arg, - right_node_arg, - current_user_oid.into(), - current_user_oid.into(), // login_role = submitted_by - database_arg, - ], - ) + // Claim a per-instance-unique node ID. Node IDs only need to be unique + // within their owning instance — the df.nodes composite PRIMARY KEY + // (instance_id, id) is the guard — so we INSERT ... ON CONFLICT + // (instance_id, id) DO NOTHING RETURNING id and re-roll on collision + // instead of surfacing a raw key violation (issue #129). Near the + // MAX_GRAPH_NODES ceiling the per-instance birthday-collision odds are + // small but non-trivial, so the retry keeps large graphs from flaking. + // B1 backward compat: the v0.1.x schema has login_role NOT NULL on + // df.nodes, so the legacy branch still sets it (= submitted_by). + // Caveat: the ON CONFLICT (instance_id, id) clause below needs the + // composite key that 0.1.1->0.2.0 adds, so a true pre-0.2.0 runtime + // cannot run df.start() until it upgrades. That is fine: the supported + // B1 floor is 0.2.2 (docs/upgrade-testing.md), so this legacy branch is + // effectively dead code for every supported install. + // The root node's ID is forced (force_id) to the value the instance row + // already points at via root_node, so the deferred same-instance FK is + // satisfied at commit without an UPDATE the caller isn't privileged to + // run. The instance is freshly reserved, so the only way the forced + // insert can conflict is a child node in this same graph randomly + // claiming the identical 8-hex value (~1 in 2^32): give it a single + // attempt and let the error abort the txn (the caller retries) rather + // than re-rolling away from the reserved ID. Non-root nodes generate a + // random ID and re-roll on collision. + let mut forced_id = force_id.map(str::to_string); + let max_attempts = if force_id.is_some() { + 1 } else { - ( - "INSERT INTO df.nodes (id, instance_id, node_type, query, result_name, left_node, right_node, submitted_by, database) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8::oid::regrole, $9)", - vec![ - node_id.as_str().into(), - instance_id.into(), - node.node_type.as_str().into(), - query_arg, - result_name_arg, - left_node_arg, - right_node_arg, - current_user_oid.into(), - database_arg, - ], - ) + MAX_ID_ATTEMPTS + }; + let node_id = match pick_id_with_retry( + move || forced_id.take().unwrap_or_else(short_id), + |candidate| { + let query_arg: DatumWithOid = match &query_val { + Some(q) => q.as_str().into(), + None => DatumWithOid::null::(), + }; + let result_name_arg: DatumWithOid = match &node.result_name { + Some(n) => n.as_str().into(), + None => DatumWithOid::null::(), + }; + let left_node_arg: DatumWithOid = match &left_id { + Some(id) => id.as_str().into(), + None => DatumWithOid::null::(), + }; + let right_node_arg: DatumWithOid = match &right_id { + Some(id) => id.as_str().into(), + None => DatumWithOid::null::(), + }; + let database_arg: DatumWithOid = match database { + Some(db) => db.into(), + None => DatumWithOid::null::(), + }; + let (node_sql, node_args): (&str, Vec) = if legacy_login_role { + ( + "INSERT INTO df.nodes (id, instance_id, node_type, query, result_name, left_node, right_node, submitted_by, login_role, database) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8::oid::regrole, $9::oid::regrole, $10) + ON CONFLICT (instance_id, id) DO NOTHING + RETURNING id", + vec![ + candidate.into(), + instance_id.into(), + node.node_type.as_str().into(), + query_arg, + result_name_arg, + left_node_arg, + right_node_arg, + current_user_oid.into(), + current_user_oid.into(), // login_role = submitted_by + database_arg, + ], + ) + } else { + ( + "INSERT INTO df.nodes (id, instance_id, node_type, query, result_name, left_node, right_node, submitted_by, database) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8::oid::regrole, $9) + ON CONFLICT (instance_id, id) DO NOTHING + RETURNING id", + vec![ + candidate.into(), + instance_id.into(), + node.node_type.as_str().into(), + query_arg, + result_name_arg, + left_node_arg, + right_node_arg, + current_user_oid.into(), + database_arg, + ], + ) + }; + Spi::connect_mut( + |client| match client.update(node_sql, Some(1), &node_args) { + Ok(table) => Ok(!table.is_empty()), + Err(e) => Err(format!("{e:?}")), + }, + ) + }, + max_attempts, + ) { + Ok(id) => id, + Err(e) => match force_id { + // The forced root id collided with an already-inserted child + // node id in this same graph (~1 in 2^32). The whole df.start() + // txn aborts cleanly (no orphan, no corruption) so the caller + // can simply retry df.start(). + Some(forced) => pgrx::error!( + "root node id '{}' collided with a child node id in the same graph \ + (~1 in 2^32); df.start() aborted safely, retry it ({})", + forced, + e + ), + None => pgrx::error!("Failed to insert node: {}", e), + }, }; - if let Err(e) = Spi::run_with_args(node_sql, &node_args) { - pgrx::error!("Failed to insert node {}: {:?}", node_id, e); - } // Return the generated ID for parent to reference node_id } let legacy_login_role = legacy_login_role_schema(); + + // Pre-generate the root node's ID so the instance row can point at it up + // front. The same-instance FK on root_node is DEFERRABLE INITIALLY + // DEFERRED, so the referenced node row need not exist yet — insert_nodes + // below inserts the root node with this exact ID before commit. + let root_id = short_id(); + + // Reserve the instance ID before inserting nodes so node rows can reference + // it. Collisions on the 8-hex ID space are rare, but we reserve via + // INSERT ... ON CONFLICT (id) DO NOTHING RETURNING id and re-roll on + // collision so a raw primary-key error never reaches the caller (issue + // #129). ON CONFLICT arbitration runs against the global id index *below* + // RLS, so this also re-rolls on collisions with another role's instance + // that the caller cannot SELECT — closing the gap left by the old + // RLS-limited pre-check. root_node is set to the pre-generated root_id; the + // same-instance FK on root_node is DEFERRABLE INITIALLY DEFERRED, so the + // referenced root node row is inserted (with that ID) before commit — no + // post-insert UPDATE is needed (df callers aren't granted UPDATE on + // root_node). + let instance_id = match pick_id_with_retry( + short_id, + |candidate| { + let label_arg: DatumWithOid = match label { + Some(l) => l.into(), + None => DatumWithOid::null::(), + }; + let database_arg: DatumWithOid = match database { + Some(db) => db.into(), + None => DatumWithOid::null::(), + }; + let (inst_sql, inst_args): (&str, Vec) = if legacy_login_role { + ( + "INSERT INTO df.instances (id, label, root_node, submitted_by, login_role, database) + VALUES ($1, $2, $3, $4::oid::regrole, $5::oid::regrole, $6) + ON CONFLICT (id) DO NOTHING + RETURNING id", + vec![ + candidate.into(), + label_arg, + root_id.as_str().into(), + current_user_oid.into(), + current_user_oid.into(), // login_role = submitted_by + database_arg, + ], + ) + } else { + ( + "INSERT INTO df.instances (id, label, root_node, submitted_by, database) + VALUES ($1, $2, $3, $4::oid::regrole, $5) + ON CONFLICT (id) DO NOTHING + RETURNING id", + vec![ + candidate.into(), + label_arg, + root_id.as_str().into(), + current_user_oid.into(), + database_arg, + ], + ) + }; + Spi::connect_mut( + |client| match client.update(inst_sql, Some(1), &inst_args) { + Ok(table) => Ok(!table.is_empty()), + Err(e) => Err(format!("{e:?}")), + }, + ) + }, + MAX_ID_ATTEMPTS, + ) { + Ok(id) => id, + Err(e) => pgrx::error!("Failed to create instance: {}", e), + }; + + // Insert the graph. The top-level (root) node's ID is forced to root_id — + // the value the instance row already points at via root_node — so the + // deferred same-instance FK is satisfied at commit with no UPDATE (df + // callers aren't granted UPDATE on root_node). Child node IDs are random + // with collision re-roll. let mut node_count: usize = 0; - let root_node_id = insert_nodes( + insert_nodes( &durofut, &instance_id, + Some(&root_id), current_user_oid, database, legacy_login_role, &mut node_count, ); - // Build parameterized args for the instance INSERT - let label_arg: DatumWithOid = match label { - Some(l) => l.into(), - None => DatumWithOid::null::(), - }; - let database_arg: DatumWithOid = match database { - Some(db) => db.into(), - None => DatumWithOid::null::(), - }; - - // Create instance record with root node ID - // B1 backward compat: v0.1.x schema has login_role NOT NULL on - // df.instances; include it (= submitted_by) so the INSERT succeeds. - let (inst_sql, inst_args): (&str, Vec) = if legacy_login_role { - ( - "INSERT INTO df.instances (id, label, root_node, submitted_by, login_role, database) VALUES ($1, $2, $3, $4::oid::regrole, $5::oid::regrole, $6)", - vec![ - instance_id.as_str().into(), - label_arg, - root_node_id.as_str().into(), - current_user_oid.into(), - current_user_oid.into(), // login_role = submitted_by - database_arg, - ], - ) - } else { - ( - "INSERT INTO df.instances (id, label, root_node, submitted_by, database) VALUES ($1, $2, $3, $4::oid::regrole, $5)", - vec![ - instance_id.as_str().into(), - label_arg, - root_node_id.as_str().into(), - current_user_oid.into(), - database_arg, - ], - ) - }; - if let Err(e) = Spi::run_with_args(inst_sql, &inst_args) { - pgrx::error!("Failed to create instance: {:?}", e); - } - // Capture vars from df.vars using the installed extension version as the // compatibility boundary: pre-0.2.0 uses legacy global vars, 0.2.0+ uses // owner-scoped vars. @@ -1010,8 +1138,9 @@ pub fn run(instance_id: default!(Option<&str>, "NULL")) -> String { pub fn result(instance_id: &str) -> Option { Spi::get_one_with_args::( r#"SELECT result::text FROM df.nodes - WHERE id = (SELECT root_node FROM df.instances WHERE id = $1) - AND status = 'completed'"#, + WHERE instance_id = $1 + AND id = (SELECT root_node FROM df.instances WHERE id = $1) + AND status = 'completed'"#, &[instance_id.into()], ) .ok() @@ -1126,7 +1255,7 @@ pub fn wait_for_completion( #[cfg(test)] mod tests { - use super::parse_semver; + use super::{parse_semver, pick_id_with_retry}; #[test] fn test_parse_semver_basic() { @@ -1158,4 +1287,67 @@ mod tests { assert!(parse_semver("0.3.0").unwrap() >= (0, 2, 0)); assert!(parse_semver("1.0.0").unwrap() >= (0, 2, 0)); } + + #[test] + fn test_pick_id_with_retry_succeeds_first_try() { + let mut gen_calls = 0; + let id = pick_id_with_retry( + || { + gen_calls += 1; + "aaaa0000".to_string() + }, + |_candidate| Ok(true), + 10, + ) + .expect("first candidate should be claimed"); + assert_eq!(id, "aaaa0000"); + assert_eq!(gen_calls, 1); + } + + #[test] + fn test_pick_id_with_retry_rerolls_on_collision() { + // generate yields two colliding candidates then a free one; try_claim + // reports the known duplicate as a collision and accepts anything else. + let mut candidates = vec!["dup00000", "dup00000", "uniq0000"].into_iter(); + let mut claim_attempts = 0; + let id = pick_id_with_retry( + || candidates.next().unwrap().to_string(), + |candidate| { + claim_attempts += 1; + Ok(candidate != "dup00000") + }, + 10, + ) + .expect("should re-roll past collisions to a free ID"); + assert_eq!(id, "uniq0000"); + assert_eq!(claim_attempts, 3); + } + + #[test] + fn test_pick_id_with_retry_exhausts_without_returning_unverified_id() { + // Every claim collides, so the helper must error (review finding C1) + // rather than hand back an unverified candidate. + let mut claim_attempts = 0; + let result = pick_id_with_retry( + || "same0000".to_string(), + |_candidate| { + claim_attempts += 1; + Ok(false) + }, + 3, + ); + assert_eq!(claim_attempts, 3); + let err = result.unwrap_err(); + assert!(err.contains("exhausted"), "unexpected error: {err}"); + } + + #[test] + fn test_pick_id_with_retry_propagates_claim_error() { + let result = pick_id_with_retry( + || "x".to_string(), + |_candidate| Err("claim blew up".to_string()), + 10, + ); + assert_eq!(result.unwrap_err(), "claim blew up"); + } } diff --git a/src/lib.rs b/src/lib.rs index 21c2c5e9..60fce559 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -182,8 +182,8 @@ extension_sql!( r#" -- Table to store function nodes (SQL steps, THEN chains, etc.) CREATE TABLE df.nodes ( - id VARCHAR(8) PRIMARY KEY, - instance_id VARCHAR(8), + id VARCHAR(8) NOT NULL, + instance_id VARCHAR(8) NOT NULL, node_type TEXT NOT NULL, query TEXT, result_name TEXT, @@ -297,8 +297,12 @@ ALTER TABLE df.nodes ELSE FALSE END ) NOT VALID, - ADD CONSTRAINT nodes_instance_node_key - UNIQUE (instance_id, id); + -- Composite primary key: node IDs only need to be unique per instance, so + -- the random 8-hex node ID is never the sole uniqueness guarantee (issue + -- #129). The same-instance foreign keys below reference (instance_id, id), + -- which this primary key satisfies. + ADD CONSTRAINT nodes_pkey + PRIMARY KEY (instance_id, id); ALTER TABLE df.nodes ADD CONSTRAINT nodes_instance_identity_fkey diff --git a/src/orchestrations/execute_function_graph.rs b/src/orchestrations/execute_function_graph.rs index d4286e09..301d61cf 100644 --- a/src/orchestrations/execute_function_graph.rs +++ b/src/orchestrations/execute_function_graph.rs @@ -330,6 +330,7 @@ async fn execute_function_node_with_vars( // Mark node as running let running_input = serde_json::json!({ "node_id": node_id, + "instance_id": graph.instance_id, "status": "running" }); let _ = ctx @@ -353,6 +354,7 @@ async fn execute_function_node_with_vars( }; let status_input = serde_json::json!({ "node_id": node_id, + "instance_id": graph.instance_id, "status": status, "result": status_result, }); diff --git a/src/types.rs b/src/types.rs index 4fdab4f9..b0ba2648 100644 --- a/src/types.rs +++ b/src/types.rs @@ -99,7 +99,23 @@ pub const MAX_GRAPH_DEPTH: usize = 256; /// unbounded INSERTs and memory exhaustion from extremely large graphs. pub const MAX_GRAPH_NODES: usize = 10_000; -/// Generate a short 8-character instance ID from a UUID +/// Generate a short 8-character ID from a UUID. +/// +/// This serves two distinct uniqueness contracts (#129). Both keep the value +/// `VARCHAR(8)` HEX (the maintainer-requested minimal change) and manage +/// collision risk by retrying on conflict rather than by widening the value: +/// - **Instance IDs** (`df.instances.id`) are global with no scoping column. +/// `df.start()` reserves the ID with `INSERT ... ON CONFLICT (id) DO NOTHING +/// RETURNING id` and re-rolls on collision; the primary key on `df.instances` +/// is the hard guarantee. +/// - **Node IDs** (`df.nodes.id`) only need to be unique per instance. Node +/// inserts use `INSERT ... ON CONFLICT (instance_id, id) DO NOTHING RETURNING +/// id` and re-roll on collision; the composite primary key `(instance_id, id)` +/// is the hard guarantee. +/// +/// The mechanism is symmetric (re-roll on conflict); only the conflict target +/// differs — the global `id` index for instances vs. the per-instance +/// `(instance_id, id)` index for nodes. pub fn short_id() -> String { let uuid = Uuid::new_v4(); uuid.to_string() diff --git a/tests/e2e/sql/51_node_composite_pk.sql b/tests/e2e/sql/51_node_composite_pk.sql new file mode 100644 index 00000000..71bad8a2 --- /dev/null +++ b/tests/e2e/sql/51_node_composite_pk.sql @@ -0,0 +1,126 @@ +-- Copyright (c) Microsoft Corporation. +-- Licensed under the PostgreSQL License. + +-- Test: df.nodes composite primary key (instance_id, id) — issue #129 +-- Verifies two things: +-- 1. Schema contract: df.nodes uses a composite PRIMARY KEY (instance_id, id) +-- instead of a global single-column key, and the legacy +-- nodes_instance_node_key UNIQUE constraint is gone (promoted to the PK). +-- 2. Regression: a multi-node workflow still completes end-to-end, every node +-- row is updated to 'completed' under instance_id scoping, and df.result() +-- returns the root result. This exercises the instance_id-scoped node-status +-- updates and df.result() lookup that accompany the composite key. + +SET SESSION AUTHORIZATION df_e2e_user; + +-- === Part 1: schema contract — composite primary key === +DO $$ +DECLARE + pk_def TEXT; + legacy_unique BOOLEAN; +BEGIN + SELECT pg_get_constraintdef(c.oid) INTO pk_def + FROM pg_constraint c + JOIN pg_class t ON t.oid = c.conrelid + JOIN pg_namespace ns ON ns.oid = t.relnamespace + WHERE ns.nspname = 'df' AND t.relname = 'nodes' AND c.contype = 'p'; + + IF pk_def IS NULL THEN + RAISE EXCEPTION 'TEST FAILED: df.nodes has no primary key'; + END IF; + + -- pg_get_constraintdef reports key columns in constraint order. + IF pk_def NOT LIKE 'PRIMARY KEY (instance_id, id)%' THEN + RAISE EXCEPTION 'TEST FAILED: expected composite PK (instance_id, id), got: %', pk_def; + END IF; + + -- The composite UNIQUE was promoted to the PK, so it must no longer exist. + SELECT EXISTS( + SELECT 1 + FROM pg_constraint c + JOIN pg_class t ON t.oid = c.conrelid + JOIN pg_namespace ns ON ns.oid = t.relnamespace + WHERE ns.nspname = 'df' + AND t.relname = 'nodes' + AND c.conname = 'nodes_instance_node_key' + ) INTO legacy_unique; + + IF legacy_unique THEN + RAISE EXCEPTION 'TEST FAILED: legacy nodes_instance_node_key UNIQUE still present'; + END IF; + + RAISE NOTICE 'PASSED: df.nodes composite primary key (instance_id, id)'; +END $$; + +-- === Part 2: regression — multi-node workflow under instance_id scoping === +CREATE TEMP TABLE _test_state (instance_id TEXT); + +INSERT INTO _test_state SELECT df.start( + 'SELECT 21 AS num' |=> 'a' + ~> 'SELECT ($a::int * 2) AS doubled', + 'test-composite-pk-regression' +); + +DO $$ +DECLARE + inst_id TEXT; + status TEXT; + node_total INT; + node_completed INT; + bad_instance INT; + result_text TEXT; +BEGIN + SELECT instance_id INTO inst_id FROM _test_state; + RAISE NOTICE 'Testing composite-PK regression: %', inst_id; + + SELECT df.await_instance(inst_id) INTO status; + + IF status != 'completed' THEN + RAISE EXCEPTION 'TEST FAILED: status = %', status; + END IF; + + -- Every node of this instance must have been updated to 'completed'. If the + -- instance_id-scoped UPDATE in update_node_status were wrong, some node would + -- linger in 'pending'/'running'. + SELECT count(*), + count(*) FILTER (WHERE n.status = 'completed'), + count(*) FILTER (WHERE n.instance_id IS DISTINCT FROM inst_id) + INTO node_total, node_completed, bad_instance + FROM df.nodes n + WHERE n.instance_id = inst_id; + + IF node_total < 2 THEN + RAISE EXCEPTION 'TEST FAILED: expected a multi-node graph, got % node(s)', node_total; + END IF; + + IF node_completed <> node_total THEN + RAISE EXCEPTION 'TEST FAILED: % of % nodes completed', node_completed, node_total; + END IF; + + IF bad_instance <> 0 THEN + RAISE EXCEPTION 'TEST FAILED: % node(s) have a mismatched instance_id', bad_instance; + END IF; + + -- df.result() is scoped by instance_id; it must return the root node's + -- result. The root SQL ('SELECT ($a::int * 2) AS doubled') returns a row + -- set, so df.result wraps it as {"rows": [{"doubled": 42}], "row_count": 1}. + -- Assert the exact nested field/value rather than a loose substring so a + -- regression that surfaced a different node's result (or lost scoping) + -- cannot pass on an incidental '42' appearing somewhere in the payload. + SELECT df.result(inst_id) INTO result_text; + IF result_text IS NULL THEN + RAISE EXCEPTION 'TEST FAILED: df.result() returned NULL'; + END IF; + IF (result_text::jsonb #>> '{rows,0,doubled}') IS NULL + OR (result_text::jsonb #>> '{rows,0,doubled}')::int <> 42 + OR (result_text::jsonb ->> 'row_count')::int <> 1 THEN + RAISE EXCEPTION 'TEST FAILED: expected root result rows[0].doubled = 42 (row_count 1), got %', result_text; + END IF; + + RAISE NOTICE 'PASSED: composite-PK regression (% nodes completed)', node_total; +END $$; + +DROP TABLE _test_state; + +RESET SESSION AUTHORIZATION; +SELECT 'TEST PASSED' AS result; diff --git a/tests/e2e/sql/52_node_id_collision_across_instances.sql b/tests/e2e/sql/52_node_id_collision_across_instances.sql new file mode 100644 index 00000000..d6953fb4 --- /dev/null +++ b/tests/e2e/sql/52_node_id_collision_across_instances.sql @@ -0,0 +1,87 @@ +-- Copyright (c) Microsoft Corporation. +-- Licensed under the PostgreSQL License. + +-- Test: cross-instance node-ID collision — issue #129 +-- Node IDs are random 8-hex and only unique *per instance* (df.nodes uses the +-- composite PK (instance_id, id)). This test forces the collision that the +-- composite key and the instance_id-scoped node-status updates exist to handle: +-- two different instances each own a df.nodes row carrying the SAME node id. +-- It asserts that +-- 1. both same-id rows coexist (composite PK, not a global single-column key); +-- 2. (instance_id, id) addresses exactly one row — the contract that +-- update_node_status now depends on (instance_id is required there); +-- 3. df.result() is instance-scoped and returns only the querying instance's +-- own node result, never the colliding sibling's; +-- 4. an instance-scoped UPDATE (mirroring update_node_status) affects exactly +-- one row. +-- +-- Runs as the privileged harness role (no SET SESSION AUTHORIZATION): it writes +-- runtime-owned columns (status, result, root_node) and a deterministic shared +-- node id directly, which an ordinary RLS user cannot do. submitted_by is set to +-- current_user so the rows are owned by (and visible to) this role regardless of +-- RLS. This is a schema/scoping regression test, not an RLS test (see 15_rls). + +DO $$ +DECLARE + role_a regrole := current_user::regrole; + shared_rows INT; + direct_a TEXT; + res_a TEXT; + res_b TEXT; + updated_rows INT; +BEGIN + -- Two instances whose root node is the SAME node id 'cccc0051'. + INSERT INTO df.instances (id, root_node, status, submitted_by) + VALUES ('aaaa0051', 'cccc0051', 'completed', role_a), + ('bbbb0051', 'cccc0051', 'completed', role_a); + + -- Same node id under two different instances, with distinct results. + INSERT INTO df.nodes (id, instance_id, node_type, query, status, result, submitted_by) + VALUES ('cccc0051', 'aaaa0051', 'SQL', 'SELECT 1', 'completed', '{"v": 111}'::jsonb, role_a), + ('cccc0051', 'bbbb0051', 'SQL', 'SELECT 1', 'completed', '{"v": 222}'::jsonb, role_a); + + -- 1. Composite PK lets both same-id rows coexist (a global single-column key + -- would have rejected the second insert). + SELECT count(*) INTO shared_rows FROM df.nodes WHERE id = 'cccc0051'; + IF shared_rows <> 2 THEN + RAISE EXCEPTION 'TEST FAILED: expected 2 df.nodes rows sharing id cccc0051, got %', shared_rows; + END IF; + + -- 2. (instance_id, id) addresses exactly one row — the invariant + -- update_node_status relies on. Deterministic regardless of row order. + SELECT result::text INTO direct_a + FROM df.nodes WHERE instance_id = 'aaaa0051' AND id = 'cccc0051'; + IF (direct_a::jsonb ->> 'v')::int <> 111 THEN + RAISE EXCEPTION 'TEST FAILED: (aaaa0051, cccc0051) addressed wrong row: %', direct_a; + END IF; + + -- 3. df.result() is instance-scoped: each instance sees ONLY its own node's + -- result, never the colliding sibling's. If the instance_id scoping in + -- df.result were lost, the shared node id would match both rows. + SELECT df.result('aaaa0051') INTO res_a; + SELECT df.result('bbbb0051') INTO res_b; + IF res_a IS NULL OR (res_a::jsonb ->> 'v')::int <> 111 THEN + RAISE EXCEPTION 'TEST FAILED: df.result(aaaa0051) = % (expected v=111)', res_a; + END IF; + IF res_b IS NULL OR (res_b::jsonb ->> 'v')::int <> 222 THEN + RAISE EXCEPTION 'TEST FAILED: df.result(bbbb0051) = % (expected v=222)', res_b; + END IF; + + -- 4. An instance-scoped UPDATE (the shape update_node_status issues) must + -- touch exactly one of the two colliding rows. + UPDATE df.nodes + SET status = 'completed', updated_at = now() + WHERE id = 'cccc0051' AND instance_id = 'aaaa0051'; + GET DIAGNOSTICS updated_rows = ROW_COUNT; + IF updated_rows <> 1 THEN + RAISE EXCEPTION 'TEST FAILED: instance-scoped UPDATE affected % row(s), expected 1', updated_rows; + END IF; + + -- Cleanup (delete nodes before instances; same-instance FKs are deferred). + DELETE FROM df.nodes WHERE id = 'cccc0051'; + DELETE FROM df.instances WHERE id IN ('aaaa0051', 'bbbb0051'); + + RAISE NOTICE 'PASSED: cross-instance node-ID collision resolved by instance_id scoping'; +END $$; + +SELECT 'TEST PASSED' AS result;