From 667167a022ed31cb67a6c189cc5bd54b09473f4f Mon Sep 17 00:00:00 2001 From: Lucas Abel <22837557+uael@users.noreply.github.com> Date: Thu, 5 Dec 2024 17:42:33 +0100 Subject: [PATCH] fix: fix `flow_node` uniqueness (#4850) `jsonb` comparison wasn't working as expected, and duplicated entries were inserted within `flow_node`. To resolve this add a second hash column, `hash_v2` with a unique default for uniqueness, and use this new column to ensure unique entries. The previous hash column is left for backward compatibility. Duplicated entries already insterted will remain as is without breaking, and only new ones will preserve uniqueness. --- ...ca0e75f30faa8c39ad4754bd568ae6f210806.json | 27 ++++++++++++ ...05135747_fix_flow_node_uniqueness.down.sql | 2 + ...1205135747_fix_flow_node_uniqueness.up.sql | 4 ++ .../windmill-worker/src/worker_lockfiles.rs | 43 +++++++------------ 4 files changed, 48 insertions(+), 28 deletions(-) create mode 100644 backend/.sqlx/query-83cc9e432aea1450f79e9fce04eca0e75f30faa8c39ad4754bd568ae6f210806.json create mode 100644 backend/migrations/20241205135747_fix_flow_node_uniqueness.down.sql create mode 100644 backend/migrations/20241205135747_fix_flow_node_uniqueness.up.sql diff --git a/backend/.sqlx/query-83cc9e432aea1450f79e9fce04eca0e75f30faa8c39ad4754bd568ae6f210806.json b/backend/.sqlx/query-83cc9e432aea1450f79e9fce04eca0e75f30faa8c39ad4754bd568ae6f210806.json new file mode 100644 index 0000000000..4e1242428a --- /dev/null +++ b/backend/.sqlx/query-83cc9e432aea1450f79e9fce04eca0e75f30faa8c39ad4754bd568ae6f210806.json @@ -0,0 +1,27 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO flow_node (path, workspace_id, hash_v2, lock, code, flow)\n VALUES ($1, $2, $3, $4, $5, $6)\n ON CONFLICT (hash_v2) DO UPDATE SET path = EXCLUDED.path -- trivial update to return the id\n RETURNING id\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Bpchar", + "Text", + "Text", + "Jsonb" + ] + }, + "nullable": [ + false + ] + }, + "hash": "83cc9e432aea1450f79e9fce04eca0e75f30faa8c39ad4754bd568ae6f210806" +} diff --git a/backend/migrations/20241205135747_fix_flow_node_uniqueness.down.sql b/backend/migrations/20241205135747_fix_flow_node_uniqueness.down.sql new file mode 100644 index 0000000000..7a5d08f3f8 --- /dev/null +++ b/backend/migrations/20241205135747_fix_flow_node_uniqueness.down.sql @@ -0,0 +1,2 @@ +-- Add down migration script here +ALTER TABLE flow_node DROP COLUMN hash_v2; diff --git a/backend/migrations/20241205135747_fix_flow_node_uniqueness.up.sql b/backend/migrations/20241205135747_fix_flow_node_uniqueness.up.sql new file mode 100644 index 0000000000..799f33d341 --- /dev/null +++ b/backend/migrations/20241205135747_fix_flow_node_uniqueness.up.sql @@ -0,0 +1,4 @@ +-- Add up migration script here +CREATE SEQUENCE IF NOT EXISTS flow_node_hash_seq; +ALTER TABLE flow_node ALTER COLUMN hash DROP NOT NULL; +ALTER TABLE flow_node ADD COLUMN hash_v2 CHAR(64) NOT NULL UNIQUE DEFAULT to_hex(nextval('flow_node_hash_seq')); diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index 71834a9f04..0ea7456ec3 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -4,6 +4,7 @@ use std::path::{Component, Path, PathBuf}; use async_recursion::async_recursion; use serde_json::value::RawValue; use serde_json::{json, Value}; +use sha2::Digest; use sqlx::types::Json; use uuid::Uuid; use windmill_common::error::Error; @@ -1025,46 +1026,32 @@ async fn insert_flow_node<'c>( flow: Option<&Json>>, ) -> Result<(sqlx::Transaction<'c, sqlx::Postgres>, FlowNodeId)> { let hash = { - use std::hash::{DefaultHasher, Hash, Hasher}; - - let mut hasher = DefaultHasher::new(); - code.hash(&mut hasher); - lock.hash(&mut hasher); - flow.inspect(|flow| flow.get().hash(&mut hasher)); - hasher.finish() as i64 + let mut hasher = sha2::Sha256::new(); + hasher.update(path); + hasher.update(workspace_id); + hasher.update(code.unwrap_or(&Default::default())); + hasher.update(lock.unwrap_or(&Default::default())); + hasher.update(flow.unwrap_or(&Default::default()).get()); + format!("{:x}", hasher.finalize()) }; // Insert the flow node if it doesn't exist. let id = sqlx::query_scalar!( r#" - WITH existing AS ( - SELECT id FROM flow_node - WHERE hash = $1 AND path = $2 AND workspace_id = $3 - AND (code IS NOT DISTINCT FROM $4) - AND (lock IS NOT DISTINCT FROM $5) - AND (flow IS NOT DISTINCT FROM $6) - LIMIT 1 - ), - inserted AS ( - INSERT INTO flow_node (hash, path, workspace_id, code, lock, flow) - VALUES ($1, $2, $3, $4, $5, $6) - ON CONFLICT DO NOTHING - RETURNING id - ) - SELECT id FROM existing - UNION ALL - SELECT id FROM inserted + INSERT INTO flow_node (path, workspace_id, hash_v2, lock, code, flow) + VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT (hash_v2) DO UPDATE SET path = EXCLUDED.path -- trivial update to return the id + RETURNING id "#, - hash, path, workspace_id, - code, + hash, lock, + code, flow as Option<&Json>> ) .fetch_one(&mut *tx) - .await? - .ok_or(error::Error::InternalErr("Failed to cache".to_string()))?; + .await?; Ok((tx, FlowNodeId(id))) }