From 7e802bd6119618804da09d98da48316cc1f93b03 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 25 Mar 2025 01:51:25 +0100 Subject: [PATCH] perf: cache workspace premium check --- ...f84fef7557f3e3046e7188ee06f6c3a2b2add.json | 15 -- ...52a6591e7593b1a7c3d7a6848587679a48187.json | 15 -- ...a6ed16c25aea4584103c1b9a3651eb31bfe53.json | 16 --- ...b6e80888e511fb4787cb1a7976878abed1d30.json | 22 --- ...707bd1c48c58aaa5b2ae7560cbed3c8d46329.json | 24 ---- ...359d6d9fdfdbffb40a2464a45056660a3a2d4.json | 17 --- ...3242b54d77b46b5f785ae66b8104f89f43219.json | 24 ---- ...cc62a0fc9d4ba56014f3e554cc2b50ce1da06.json | 16 --- ...4ee12e8cffe925bd7cc4e9d02dc90ffe200f3.json | 15 -- ...ba8354d4cd733335db3d76155b663f668b9d7.json | 16 --- ...18cd141e68cf83a7dcae7d3c1b8a7ba94425b.json | 15 -- ...77eb9b71337a9aa2a64cd2db031e64e929251.json | 15 -- ...4f0bf040da3af9e2eac3229264ec95b4755d8.json | 16 --- ...dc6b39dabab77ae7810f553f1773995db51f1.json | 16 --- ...706f62e700695810f23108e975dfcd1fee7a3.json | 17 --- ...a4420f232867a23cacb9a62b386f129a86a4b.json | 16 --- backend/Cargo.toml | 2 +- ...003851_workspace_premium_listener.down.sql | 3 + ...25003851_workspace_premium_listener.up.sql | 13 ++ backend/src/main.rs | 25 ++-- backend/windmill-api/Cargo.toml | 1 + backend/windmill-api/src/workspaces.rs | 20 ++- backend/windmill-common/Cargo.toml | 1 + backend/windmill-common/src/workspaces.rs | 21 ++- backend/windmill-queue/src/jobs.rs | 18 +-- backend/windmill-worker/src/common.rs | 10 +- backend/windmill-worker/src/worker.rs | 135 +++++++++--------- 27 files changed, 138 insertions(+), 386 deletions(-) delete mode 100644 backend/.sqlx/query-1ea8c2ce29da8d1be19dfd3ecc7f84fef7557f3e3046e7188ee06f6c3a2b2add.json delete mode 100644 backend/.sqlx/query-2bb2cf6accb18d3e37a63388cca52a6591e7593b1a7c3d7a6848587679a48187.json delete mode 100644 backend/.sqlx/query-35795d27c4ca69d2f145b4dba08a6ed16c25aea4584103c1b9a3651eb31bfe53.json delete mode 100644 backend/.sqlx/query-4c970f10d345bcdcf956dcbfa22b6e80888e511fb4787cb1a7976878abed1d30.json delete mode 100644 backend/.sqlx/query-56a2fe44b73728c9aba10f3d1c7707bd1c48c58aaa5b2ae7560cbed3c8d46329.json delete mode 100644 backend/.sqlx/query-894e5080e83b098839a050bb472359d6d9fdfdbffb40a2464a45056660a3a2d4.json delete mode 100644 backend/.sqlx/query-9aeee333b1dbe58ba819ba3b2713242b54d77b46b5f785ae66b8104f89f43219.json delete mode 100644 backend/.sqlx/query-9c956541e068154193243795d96cc62a0fc9d4ba56014f3e554cc2b50ce1da06.json delete mode 100644 backend/.sqlx/query-a28f8697da75d7ac2c6e13dda394ee12e8cffe925bd7cc4e9d02dc90ffe200f3.json delete mode 100644 backend/.sqlx/query-ab9783a48e0f6cacd5dd86cfa21ba8354d4cd733335db3d76155b663f668b9d7.json delete mode 100644 backend/.sqlx/query-acbea8740b28c26942c50edcf5618cd141e68cf83a7dcae7d3c1b8a7ba94425b.json delete mode 100644 backend/.sqlx/query-af2f780da994befd09959d3b1b077eb9b71337a9aa2a64cd2db031e64e929251.json delete mode 100644 backend/.sqlx/query-b656927cd70b6667f3c72186ec04f0bf040da3af9e2eac3229264ec95b4755d8.json delete mode 100644 backend/.sqlx/query-baf63525ca210c22d3ad8c0b197dc6b39dabab77ae7810f553f1773995db51f1.json delete mode 100644 backend/.sqlx/query-c35f44f91b08fa57e29a2b4a685706f62e700695810f23108e975dfcd1fee7a3.json delete mode 100644 backend/.sqlx/query-dc58e5b4715601a93b3c01a2564a4420f232867a23cacb9a62b386f129a86a4b.json create mode 100644 backend/migrations/20250325003851_workspace_premium_listener.down.sql create mode 100644 backend/migrations/20250325003851_workspace_premium_listener.up.sql diff --git a/backend/.sqlx/query-1ea8c2ce29da8d1be19dfd3ecc7f84fef7557f3e3046e7188ee06f6c3a2b2add.json b/backend/.sqlx/query-1ea8c2ce29da8d1be19dfd3ecc7f84fef7557f3e3046e7188ee06f6c3a2b2add.json deleted file mode 100644 index 17c406c101..0000000000 --- a/backend/.sqlx/query-1ea8c2ce29da8d1be19dfd3ecc7f84fef7557f3e3046e7188ee06f6c3a2b2add.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "DELETE FROM workspace_runnable_usage WHERE resource_path = $1 AND workspace_id = $2 and resource_kind = 'flow'", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Text" - ] - }, - "nullable": [] - }, - "hash": "1ea8c2ce29da8d1be19dfd3ecc7f84fef7557f3e3046e7188ee06f6c3a2b2add" -} diff --git a/backend/.sqlx/query-2bb2cf6accb18d3e37a63388cca52a6591e7593b1a7c3d7a6848587679a48187.json b/backend/.sqlx/query-2bb2cf6accb18d3e37a63388cca52a6591e7593b1a7c3d7a6848587679a48187.json deleted file mode 100644 index 11e7df65bf..0000000000 --- a/backend/.sqlx/query-2bb2cf6accb18d3e37a63388cca52a6591e7593b1a7c3d7a6848587679a48187.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE flow_workspace_runnables SET workspace_id = $1 WHERE workspace_id = $2", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Text" - ] - }, - "nullable": [] - }, - "hash": "2bb2cf6accb18d3e37a63388cca52a6591e7593b1a7c3d7a6848587679a48187" -} diff --git a/backend/.sqlx/query-35795d27c4ca69d2f145b4dba08a6ed16c25aea4584103c1b9a3651eb31bfe53.json b/backend/.sqlx/query-35795d27c4ca69d2f145b4dba08a6ed16c25aea4584103c1b9a3651eb31bfe53.json deleted file mode 100644 index 06730d3e26..0000000000 --- a/backend/.sqlx/query-35795d27c4ca69d2f145b4dba08a6ed16c25aea4584103c1b9a3651eb31bfe53.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO flow_workspace_runnables (flow_path, runnable_path, runnable_is_flow, workspace_id) VALUES ($1, $2, TRUE, $3) ON CONFLICT DO NOTHING", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Varchar" - ] - }, - "nullable": [] - }, - "hash": "35795d27c4ca69d2f145b4dba08a6ed16c25aea4584103c1b9a3651eb31bfe53" -} diff --git a/backend/.sqlx/query-4c970f10d345bcdcf956dcbfa22b6e80888e511fb4787cb1a7976878abed1d30.json b/backend/.sqlx/query-4c970f10d345bcdcf956dcbfa22b6e80888e511fb4787cb1a7976878abed1d30.json deleted file mode 100644 index 26894a04ef..0000000000 --- a/backend/.sqlx/query-4c970f10d345bcdcf956dcbfa22b6e80888e511fb4787cb1a7976878abed1d30.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT premium FROM workspace WHERE workspace.id = $1", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "premium", - "type_info": "Bool" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - false - ] - }, - "hash": "4c970f10d345bcdcf956dcbfa22b6e80888e511fb4787cb1a7976878abed1d30" -} diff --git a/backend/.sqlx/query-56a2fe44b73728c9aba10f3d1c7707bd1c48c58aaa5b2ae7560cbed3c8d46329.json b/backend/.sqlx/query-56a2fe44b73728c9aba10f3d1c7707bd1c48c58aaa5b2ae7560cbed3c8d46329.json deleted file mode 100644 index 22870b1d2d..0000000000 --- a/backend/.sqlx/query-56a2fe44b73728c9aba10f3d1c7707bd1c48c58aaa5b2ae7560cbed3c8d46329.json +++ /dev/null @@ -1,24 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT f.path\n FROM workspace_runnable_usage wru \n JOIN flow f\n ON wru.resource_path = f.path AND wru.workspace_id = f.workspace_id\n WHERE wru.runnable_path = $1 AND wru.runnable_is_flow = $2 AND wru.workspace_id = $3", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "path", - "type_info": "Varchar" - } - ], - "parameters": { - "Left": [ - "Text", - "Bool", - "Text" - ] - }, - "nullable": [ - false - ] - }, - "hash": "56a2fe44b73728c9aba10f3d1c7707bd1c48c58aaa5b2ae7560cbed3c8d46329" -} diff --git a/backend/.sqlx/query-894e5080e83b098839a050bb472359d6d9fdfdbffb40a2464a45056660a3a2d4.json b/backend/.sqlx/query-894e5080e83b098839a050bb472359d6d9fdfdbffb40a2464a45056660a3a2d4.json deleted file mode 100644 index 7ee434f66e..0000000000 --- a/backend/.sqlx/query-894e5080e83b098839a050bb472359d6d9fdfdbffb40a2464a45056660a3a2d4.json +++ /dev/null @@ -1,17 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO workspace_runnable_usage (resource_path, runnable_path, script_hash, runnable_is_flow, workspace_id, resource_kind) VALUES ($1, $2, $3, FALSE, $4, 'flow') ON CONFLICT DO NOTHING", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Int8", - "Varchar" - ] - }, - "nullable": [] - }, - "hash": "894e5080e83b098839a050bb472359d6d9fdfdbffb40a2464a45056660a3a2d4" -} diff --git a/backend/.sqlx/query-9aeee333b1dbe58ba819ba3b2713242b54d77b46b5f785ae66b8104f89f43219.json b/backend/.sqlx/query-9aeee333b1dbe58ba819ba3b2713242b54d77b46b5f785ae66b8104f89f43219.json deleted file mode 100644 index 271a508a83..0000000000 --- a/backend/.sqlx/query-9aeee333b1dbe58ba819ba3b2713242b54d77b46b5f785ae66b8104f89f43219.json +++ /dev/null @@ -1,24 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT f.path\n FROM flow_workspace_runnables fwr \n JOIN flow f \n ON fwr.flow_path = f.path AND fwr.workspace_id = f.workspace_id\n WHERE fwr.runnable_path = $1 AND fwr.runnable_is_flow = $2 AND fwr.workspace_id = $3", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "path", - "type_info": "Varchar" - } - ], - "parameters": { - "Left": [ - "Text", - "Bool", - "Text" - ] - }, - "nullable": [ - false - ] - }, - "hash": "9aeee333b1dbe58ba819ba3b2713242b54d77b46b5f785ae66b8104f89f43219" -} diff --git a/backend/.sqlx/query-9c956541e068154193243795d96cc62a0fc9d4ba56014f3e554cc2b50ce1da06.json b/backend/.sqlx/query-9c956541e068154193243795d96cc62a0fc9d4ba56014f3e554cc2b50ce1da06.json deleted file mode 100644 index 690f2dedcc..0000000000 --- a/backend/.sqlx/query-9c956541e068154193243795d96cc62a0fc9d4ba56014f3e554cc2b50ce1da06.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE workspace_runnable_usage SET resource_path = REGEXP_REPLACE(resource_path,'u/' || $2 || '/(.*)','u/' || $1 || '/\\1') WHERE resource_path LIKE ('u/' || $2 || '/%') AND workspace_id = $3", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Text", - "Text" - ] - }, - "nullable": [] - }, - "hash": "9c956541e068154193243795d96cc62a0fc9d4ba56014f3e554cc2b50ce1da06" -} diff --git a/backend/.sqlx/query-a28f8697da75d7ac2c6e13dda394ee12e8cffe925bd7cc4e9d02dc90ffe200f3.json b/backend/.sqlx/query-a28f8697da75d7ac2c6e13dda394ee12e8cffe925bd7cc4e9d02dc90ffe200f3.json deleted file mode 100644 index 62c2a10550..0000000000 --- a/backend/.sqlx/query-a28f8697da75d7ac2c6e13dda394ee12e8cffe925bd7cc4e9d02dc90ffe200f3.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "DELETE FROM workspace_runnable_usage WHERE resource_path = $1 AND workspace_id = $2 and resource_kind = 'app'", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Text" - ] - }, - "nullable": [] - }, - "hash": "a28f8697da75d7ac2c6e13dda394ee12e8cffe925bd7cc4e9d02dc90ffe200f3" -} diff --git a/backend/.sqlx/query-ab9783a48e0f6cacd5dd86cfa21ba8354d4cd733335db3d76155b663f668b9d7.json b/backend/.sqlx/query-ab9783a48e0f6cacd5dd86cfa21ba8354d4cd733335db3d76155b663f668b9d7.json deleted file mode 100644 index 7dd52db3fa..0000000000 --- a/backend/.sqlx/query-ab9783a48e0f6cacd5dd86cfa21ba8354d4cd733335db3d76155b663f668b9d7.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO workspace_runnable_usage (resource_path, runnable_path, runnable_is_flow, workspace_id, resource_kind) VALUES ($1, $2, TRUE, $3, 'flow') ON CONFLICT DO NOTHING", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Varchar" - ] - }, - "nullable": [] - }, - "hash": "ab9783a48e0f6cacd5dd86cfa21ba8354d4cd733335db3d76155b663f668b9d7" -} diff --git a/backend/.sqlx/query-acbea8740b28c26942c50edcf5618cd141e68cf83a7dcae7d3c1b8a7ba94425b.json b/backend/.sqlx/query-acbea8740b28c26942c50edcf5618cd141e68cf83a7dcae7d3c1b8a7ba94425b.json deleted file mode 100644 index 410aed9cc5..0000000000 --- a/backend/.sqlx/query-acbea8740b28c26942c50edcf5618cd141e68cf83a7dcae7d3c1b8a7ba94425b.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "DELETE FROM flow_workspace_runnables WHERE flow_path = $1 AND workspace_id = $2", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Text" - ] - }, - "nullable": [] - }, - "hash": "acbea8740b28c26942c50edcf5618cd141e68cf83a7dcae7d3c1b8a7ba94425b" -} diff --git a/backend/.sqlx/query-af2f780da994befd09959d3b1b077eb9b71337a9aa2a64cd2db031e64e929251.json b/backend/.sqlx/query-af2f780da994befd09959d3b1b077eb9b71337a9aa2a64cd2db031e64e929251.json deleted file mode 100644 index 8ccc824ace..0000000000 --- a/backend/.sqlx/query-af2f780da994befd09959d3b1b077eb9b71337a9aa2a64cd2db031e64e929251.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE workspace_runnable_usage SET workspace_id = $1 WHERE workspace_id = $2", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Text" - ] - }, - "nullable": [] - }, - "hash": "af2f780da994befd09959d3b1b077eb9b71337a9aa2a64cd2db031e64e929251" -} diff --git a/backend/.sqlx/query-b656927cd70b6667f3c72186ec04f0bf040da3af9e2eac3229264ec95b4755d8.json b/backend/.sqlx/query-b656927cd70b6667f3c72186ec04f0bf040da3af9e2eac3229264ec95b4755d8.json deleted file mode 100644 index c05edddc9f..0000000000 --- a/backend/.sqlx/query-b656927cd70b6667f3c72186ec04f0bf040da3af9e2eac3229264ec95b4755d8.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE flow_workspace_runnables SET flow_path = REGEXP_REPLACE(flow_path,'u/' || $2 || '/(.*)','u/' || $1 || '/\\1') WHERE flow_path LIKE ('u/' || $2 || '/%') AND workspace_id = $3", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Text", - "Text" - ] - }, - "nullable": [] - }, - "hash": "b656927cd70b6667f3c72186ec04f0bf040da3af9e2eac3229264ec95b4755d8" -} diff --git a/backend/.sqlx/query-baf63525ca210c22d3ad8c0b197dc6b39dabab77ae7810f553f1773995db51f1.json b/backend/.sqlx/query-baf63525ca210c22d3ad8c0b197dc6b39dabab77ae7810f553f1773995db51f1.json deleted file mode 100644 index dfc8414434..0000000000 --- a/backend/.sqlx/query-baf63525ca210c22d3ad8c0b197dc6b39dabab77ae7810f553f1773995db51f1.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE workspace_runnable_usage SET runnable_path = REGEXP_REPLACE(runnable_path,'u/' || $2 || '/(.*)','u/' || $1 || '/\\1') WHERE runnable_path LIKE ('u/' || $2 || '/%') AND workspace_id = $3", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Text", - "Text" - ] - }, - "nullable": [] - }, - "hash": "baf63525ca210c22d3ad8c0b197dc6b39dabab77ae7810f553f1773995db51f1" -} diff --git a/backend/.sqlx/query-c35f44f91b08fa57e29a2b4a685706f62e700695810f23108e975dfcd1fee7a3.json b/backend/.sqlx/query-c35f44f91b08fa57e29a2b4a685706f62e700695810f23108e975dfcd1fee7a3.json deleted file mode 100644 index 77e89c52a8..0000000000 --- a/backend/.sqlx/query-c35f44f91b08fa57e29a2b4a685706f62e700695810f23108e975dfcd1fee7a3.json +++ /dev/null @@ -1,17 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO flow_workspace_runnables (flow_path, runnable_path, script_hash, runnable_is_flow, workspace_id) VALUES ($1, $2, $3, FALSE, $4) ON CONFLICT DO NOTHING", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Int8", - "Varchar" - ] - }, - "nullable": [] - }, - "hash": "c35f44f91b08fa57e29a2b4a685706f62e700695810f23108e975dfcd1fee7a3" -} diff --git a/backend/.sqlx/query-dc58e5b4715601a93b3c01a2564a4420f232867a23cacb9a62b386f129a86a4b.json b/backend/.sqlx/query-dc58e5b4715601a93b3c01a2564a4420f232867a23cacb9a62b386f129a86a4b.json deleted file mode 100644 index d3b8f33ff0..0000000000 --- a/backend/.sqlx/query-dc58e5b4715601a93b3c01a2564a4420f232867a23cacb9a62b386f129a86a4b.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE flow_workspace_runnables SET runnable_path = REGEXP_REPLACE(runnable_path,'u/' || $2 || '/(.*)','u/' || $1 || '/\\1') WHERE runnable_path LIKE ('u/' || $2 || '/%') AND workspace_id = $3", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Text", - "Text" - ] - }, - "nullable": [] - }, - "hash": "dc58e5b4715601a93b3c01a2564a4420f232867a23cacb9a62b386f129a86a4b" -} diff --git a/backend/Cargo.toml b/backend/Cargo.toml index f07ef205fb..c001db30c2 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -58,7 +58,7 @@ parquet = ["windmill-api/parquet", "windmill-common/parquet", "windmill-worker/p prometheus = ["windmill-common/prometheus", "windmill-api/prometheus", "windmill-worker/prometheus", "windmill-queue/prometheus", "dep:prometheus"] flow_testing = ["windmill-worker/flow_testing"] openidconnect = ["windmill-api/openidconnect"] -cloud = ["windmill-queue/cloud", "windmill-worker/cloud"] +cloud = ["windmill-queue/cloud", "windmill-worker/cloud", "windmill-common/cloud", "windmill-api/cloud"] jemalloc = ["windmill-common/jemalloc", "dep:tikv-jemallocator", "dep:tikv-jemalloc-sys", "dep:tikv-jemalloc-ctl"] tantivy = ["dep:windmill-indexer", "windmill-api/tantivy", "windmill-indexer/enterprise", "windmill-indexer/parquet", "windmill-common/tantivy", "enterprise", "parquet"] sqlx = ["windmill-worker/sqlx"] diff --git a/backend/migrations/20250325003851_workspace_premium_listener.down.sql b/backend/migrations/20250325003851_workspace_premium_listener.down.sql new file mode 100644 index 0000000000..5515b4226d --- /dev/null +++ b/backend/migrations/20250325003851_workspace_premium_listener.down.sql @@ -0,0 +1,3 @@ +-- Add down migration script here +DROP TRIGGER workspace_premium_change_trigger ON workspace; +DROP FUNCTION notify_workspace_premium_change(); diff --git a/backend/migrations/20250325003851_workspace_premium_listener.up.sql b/backend/migrations/20250325003851_workspace_premium_listener.up.sql new file mode 100644 index 0000000000..e841dfbe16 --- /dev/null +++ b/backend/migrations/20250325003851_workspace_premium_listener.up.sql @@ -0,0 +1,13 @@ +-- Add up migration script here +CREATE OR REPLACE FUNCTION notify_workspace_premium_change() +RETURNS TRIGGER AS $$ +BEGIN + PERFORM pg_notify('notify_workspace_premium_change', NEW.id); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +CREATE TRIGGER workspace_premium_change_trigger +AFTER UPDATE OF premium ON workspace +FOR EACH ROW +EXECUTE FUNCTION notify_workspace_premium_change(); diff --git a/backend/src/main.rs b/backend/src/main.rs index c349a87249..c608b214b0 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -707,6 +707,11 @@ Windmill Community Edition {GIT_VERSION} tracing::info!("Workspace envs change detected, invalidating workspace envs cache: {}", workspace_id); windmill_common::variables::CUSTOM_ENVS_CACHE.remove(workspace_id); }, + "notify_workspace_premium_change" => { + let workspace_id = n.payload(); + tracing::info!("Workspace premium change detected, invalidating workspace premium cache: {}", workspace_id); + windmill_common::workspaces::IS_PREMIUM_CACHE.remove(workspace_id); + }, "notify_global_setting_change" => { tracing::info!("Global setting change detected: {}", n.payload()); match n.payload() { @@ -980,15 +985,17 @@ async fn listen_pg(url: &str) -> Option { } }; - if let Err(e) = listener - .listen_all(vec![ - "notify_config_change", - "notify_global_setting_change", - "notify_webhook_change", - "notify_workspace_envs_change", - ]) - .await - { + #[allow(unused_mut)] + let mut channels = vec![ + "notify_config_change", + "notify_global_setting_change", + "notify_webhook_change", + "notify_workspace_envs_change", + ]; + #[cfg(feature = "cloud")] + channels.push("notify_workspace_premium_change"); + + if let Err(e) = listener.listen_all(channels).await { tracing::error!(error = %e, "Could not listen to database"); return None; } diff --git a/backend/windmill-api/Cargo.toml b/backend/windmill-api/Cargo.toml index 20fe893517..97928baa19 100644 --- a/backend/windmill-api/Cargo.toml +++ b/backend/windmill-api/Cargo.toml @@ -31,6 +31,7 @@ static_frontend = ["dep:rust-embed"] postgres_trigger = ["dep:rust-postgres", "dep:pg_escape", "dep:byteorder", "dep:thiserror", "dep:rust_decimal", "dep:rust-postgres-native-tls"] mqtt_trigger = ["dep:thiserror", "dep:rumqttc"] sqs_trigger = ["dep:aws-sdk-sqs", "dep:thiserror", "dep:aws-config"] +cloud = ["windmill-common/cloud"] [dependencies] windmill-queue.workspace = true diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index c83370bfb5..03ba214bba 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -393,19 +393,15 @@ async fn list_pending_invites( async fn is_premium( authed: ApiAuthed, - Extension(db): Extension, - Path(w_id): Path, + Extension(_db): Extension, + Path(_w_id): Path, ) -> JsonResult { require_admin(authed.is_admin, &authed.username)?; - let mut tx = db.begin().await?; - let row = sqlx::query_scalar!( - "SELECT premium FROM workspace WHERE workspace.id = $1", - &w_id - ) - .fetch_one(&mut *tx) - .await?; - tx.commit().await?; - Ok(Json(row)) + #[cfg(feature = "cloud")] + let premium = windmill_common::workspaces::is_premium_workspace(&_db, &_w_id).await; + #[cfg(not(feature = "cloud"))] + let premium = false; + Ok(Json(premium)) } async fn exists_workspace( @@ -1362,7 +1358,7 @@ struct UsedTriggers { pub nats_used: bool, pub postgres_used: bool, pub mqtt_used: bool, - pub sqs_used: bool + pub sqs_used: bool, } async fn get_used_triggers( diff --git a/backend/windmill-common/Cargo.toml b/backend/windmill-common/Cargo.toml index f19ef1b06c..7ff360b42b 100644 --- a/backend/windmill-common/Cargo.toml +++ b/backend/windmill-common/Cargo.toml @@ -17,6 +17,7 @@ otel = ["dep:opentelemetry-semantic-conventions", "dep:opentelemetry-otlp", "dep "dep:opentelemetry", "dep:tracing-opentelemetry", "dep:opentelemetry-appender-tracing", "dep:tonic"] smtp = ["dep:mail-send"] scoped_cache = [] +cloud = [] [lib] name = "windmill_common" diff --git a/backend/windmill-common/src/workspaces.rs b/backend/windmill-common/src/workspaces.rs index 8fd5471a00..1238d5b243 100644 --- a/backend/windmill-common/src/workspaces.rs +++ b/backend/windmill-common/src/workspaces.rs @@ -1,3 +1,4 @@ +use quick_cache::sync::Cache; use serde::{Deserialize, Serialize}; #[derive(Serialize, Deserialize, Debug, Default)] @@ -27,7 +28,7 @@ pub enum ObjectType { ResourceType, User, Group, - Trigger + Trigger, } #[derive(Serialize, Deserialize, Debug)] @@ -38,3 +39,21 @@ pub struct GitRepositorySettings { pub group_by_folder: Option, pub exclude_types_override: Option>, } + +lazy_static::lazy_static! { + pub static ref IS_PREMIUM_CACHE: Cache = Cache::new(5000); +} + +#[cfg(feature = "cloud")] +pub async fn is_premium_workspace(_db: &crate::DB, _w_id: &str) -> bool { + let cached = IS_PREMIUM_CACHE.get(_w_id); + if let Some(cached) = cached { + return cached; + } + let premium = sqlx::query_scalar!("SELECT premium FROM workspace WHERE id = $1", _w_id) + .fetch_one(_db) + .await + .unwrap_or(false); + IS_PREMIUM_CACHE.insert(_w_id.to_string(), premium); + premium +} diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 865b7d47bc..f62984942d 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -928,13 +928,7 @@ pub async fn add_completed_job( if *CLOUD_HOSTED && !queued_job.is_flow() && _duration > 1000 { let additional_usage = _duration / 1000; let w_id = &queued_job.workspace_id; - let premium_workspace = - sqlx::query_scalar!("SELECT premium FROM workspace WHERE id = $1", w_id) - .fetch_one(db) - .await - .map_err(|e| { - Error::internal_err(format!("fetching if {w_id} is premium: {e:#}")) - })?; + let premium_workspace = windmill_common::workspaces::is_premium_workspace(db, w_id).await; let _ = sqlx::query!( "INSERT INTO usage (id, is_workspace, month_, usage) VALUES ($1, TRUE, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), $2) @@ -2996,15 +2990,7 @@ pub async fn push<'c, 'd>( #[cfg(feature = "cloud")] if *CLOUD_HOSTED { let premium_workspace = - sqlx::query_scalar!("SELECT premium FROM workspace WHERE id = $1", workspace_id) - .fetch_one(_db) - .await - .map_err(|e| { - Error::internal_err(format!( - "fetching if {workspace_id} is premium and overquota: {e:#}" - )) - })?; - + windmill_common::workspaces::is_premium_workspace(_db, workspace_id).await; // we track only non flow steps let (workspace_usage, user_usage) = if !matches!( job_payload, diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 67df5e2f23..cd4f26b776 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -564,14 +564,8 @@ pub async fn resolve_job_timeout( ) -> (Duration, Option, bool) { let mut warn_msg: Option = None; #[cfg(feature = "cloud")] - let cloud_premium_workspace = *CLOUD_HOSTED - && sqlx::query_scalar!("SELECT premium FROM workspace WHERE id = $1", _w_id) - .fetch_one(_db) - .await - .map_err(|e| { - tracing::error!(%e, "error getting premium workspace for job {_job_id}: {e:#}"); - }) - .unwrap_or(false); + let cloud_premium_workspace = + *CLOUD_HOSTED && windmill_common::workspaces::is_premium_workspace(_db, _w_id).await; #[cfg(not(feature = "cloud"))] let cloud_premium_workspace = false; diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index c696dbe6c6..a1f1754ab5 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -1283,70 +1283,9 @@ pub async fn run_worker( "received {} from same worker channel", same_worker_job.job_id ); - let r = sqlx::query_as::<_, PulledJob>( - "WITH ping AS ( - UPDATE v2_job_runtime SET ping = NOW() WHERE id = $1 - ), - started_at AS ( - UPDATE v2_job_queue SET started_at = NOW() WHERE id = $1 - ) - SELECT - v2_job_queue.workspace_id, - v2_job_queue.id, - v2_job.args, - v2_job.parent_job, - v2_job.created_by, - v2_job_queue.started_at, - scheduled_for, - v2_job.runnable_path, - v2_job.kind, - v2_job.runnable_id, - v2_job_queue.canceled_reason, - v2_job_queue.canceled_by, - v2_job.permissioned_as, - v2_job.permissioned_as_email, - v2_job_status.flow_status, - v2_job.tag, - v2_job.script_lang, - v2_job.same_worker, - v2_job.pre_run_error, - v2_job.concurrent_limit, - v2_job.concurrency_time_window_s, - v2_job.flow_innermost_root_job, - v2_job.timeout, - v2_job.flow_step_id, - v2_job.cache_ttl, - v2_job_queue.priority, - v2_job.preprocessed, - v2_job.script_entrypoint_override, - v2_job.trigger, - v2_job.trigger_kind, - v2_job.visible_to_owner, - v2_job.raw_code, - v2_job.raw_lock, - v2_job.raw_flow, - pj.runnable_path as parent_runnable_path, - p.email as permissioned_as_email, p.username as permissioned_as_username, p.is_admin as permissioned_as_is_admin, - p.is_operator as permissioned_as_is_operator, p.groups as permissioned_as_groups, p.folders as permissioned_as_folders - FROM v2_job_queue - INNER JOIN v2_job ON v2_job.id = v2_job_queue.id - LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id - LEFT JOIN job_perms p ON p.job_id = v2_job.id - LEFT JOIN v2_job pj ON v2_job.parent_job = pj.id - WHERE v2_job_queue.id = $1 -", - ) - .bind(same_worker_job.job_id) - .fetch_optional(db) - .await - .map_err(|e| { - Error::internal_err(format!( - "Impossible to fetch same_worker job {}: {}", - same_worker_job.job_id, e - )) - }); + let job = get_same_worker_job(db, &same_worker_job).await; // tracing::error!("r: {:?}", r); - if r.is_err() && !same_worker_job.recoverable { + if job.is_err() && !same_worker_job.recoverable { tracing::error!( worker = %worker_name, hostname = %hostname, "failed to fetch same_worker job on a non recoverable job, exiting" @@ -1358,7 +1297,7 @@ pub async fn run_worker( .expect("send kill to job completed tx"); break; } else { - r + job } } else if let Ok(_) = killpill_rx.try_recv() { if !killed_but_draining_same_worker_jobs { @@ -1862,6 +1801,74 @@ pub async fn run_worker( tracing::info!(worker = %worker_name, hostname = %hostname, "number of jobs executed: {}", jobs_executed); } +async fn get_same_worker_job( + db: &Pool, + same_worker_job: &SameWorkerPayload, +) -> windmill_common::error::Result> { + sqlx::query_as::<_, PulledJob>( + "WITH ping AS ( + UPDATE v2_job_runtime SET ping = NOW() WHERE id = $1 + ), + started_at AS ( + UPDATE v2_job_queue SET started_at = NOW() WHERE id = $1 + ) + SELECT + v2_job_queue.workspace_id, + v2_job_queue.id, + v2_job.args, + v2_job.parent_job, + v2_job.created_by, + v2_job_queue.started_at, + scheduled_for, + v2_job.runnable_path, + v2_job.kind, + v2_job.runnable_id, + v2_job_queue.canceled_reason, + v2_job_queue.canceled_by, + v2_job.permissioned_as, + v2_job.permissioned_as_email, + v2_job_status.flow_status, + v2_job.tag, + v2_job.script_lang, + v2_job.same_worker, + v2_job.pre_run_error, + v2_job.concurrent_limit, + v2_job.concurrency_time_window_s, + v2_job.flow_innermost_root_job, + v2_job.timeout, + v2_job.flow_step_id, + v2_job.cache_ttl, + v2_job_queue.priority, + v2_job.preprocessed, + v2_job.script_entrypoint_override, + v2_job.trigger, + v2_job.trigger_kind, + v2_job.visible_to_owner, + v2_job.raw_code, + v2_job.raw_lock, + v2_job.raw_flow, + pj.runnable_path as parent_runnable_path, + p.email as permissioned_as_email, p.username as permissioned_as_username, p.is_admin as permissioned_as_is_admin, + p.is_operator as permissioned_as_is_operator, p.groups as permissioned_as_groups, p.folders as permissioned_as_folders + FROM v2_job_queue + INNER JOIN v2_job ON v2_job.id = v2_job_queue.id + LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id + LEFT JOIN job_perms p ON p.job_id = v2_job.id + LEFT JOIN v2_job pj ON v2_job.parent_job = pj.id + WHERE v2_job_queue.id = $1 +", + ) + .bind(same_worker_job.job_id) + .fetch_optional(db) + .await + .map_err(|e| { + Error::internal_err(format!( + "Impossible to fetch same_worker job {}: {}", + same_worker_job.job_id, e + )) + }) +} + async fn queue_init_bash_maybe<'c>( db: &Pool, same_worker_tx: SameWorkerSender,