From 778f8f13e7827e1a5da476bf22d73ba605100ef4 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Mon, 9 Feb 2026 23:48:29 +0100 Subject: [PATCH] fix: reduce DB pool contention by eliminating dual-connection patterns (#7861) --- ...1f7f387f5055c47f493271d26731336257384.json | 10 ++-- ...bc7e9ee85289ff5c52af43166928337c257e7.json | 3 +- ...7b23d6dcf2e5364852a5e85b3c41c417bace2.json | 3 +- ...9e8c0bb132b70668301b535872d1632753e30.json | 3 +- ...66f38ffe2f3a0958adea53e67757c6ac463ec.json | 3 +- ...243ad2e0aef485a780daea5d76b0be2bb2ea2.json | 3 +- ...a81d68f3d56d4782e99e05b4f1cb362683514.json | 15 ------ ...adbb2860ce52d37a56b2861de1215260ecff8.json | 6 ++- ...fbe7db4887c4ac5295052c796bd78a7edc50b.json | 3 +- ...f518b42f4cdb33e034441015494bd26c172d2.json | 3 +- ...cad6d8995242dc20a4e14c1a8fae4fc6a9fd2.json | 3 +- ...2f7175cc6f2e02d9c565bb5159cae3e0b6835.json | 3 +- ...08532a1020cabaf5e22325a1e05f0491d785c.json | 3 +- ...e35adf640d44d607e62a1fb296e38ac3fdd36.json | 6 ++- ...4e9f7a91240a00f10d24d3f28d5b722f68768.json | 15 ++++++ ...c6c70f7f3958564a79bb0c6b3863b1ebb0cbf.json | 3 +- ...1a803e1bced2f8e1a12f8933b7598ee85a147.json | 3 +- ...4f412c57af931f80621aee5012e9cb3535626.json | 3 +- ...8d92a983470adcd3cc850734960c905e61e83.json | 3 +- ...ca846dd5e7fe7ceb1aa8b3fcd26a2bd30eb2c.json | 3 +- ...a22e4276c1da653475628889ce17dc904fbaa.json | 3 +- ...6eb9016207975af1e37042d81df8cb6ae2c53.json | 3 +- ...506bf08c28efe82b68b3d44bafbd3d0e91c29.json | 3 +- ...d2393e297043e4e58555d872b5c4dd89c196a.json | 3 +- ...3a960b89e02ec40b292f3b5198baf2a1d3dbf.json | 3 +- ...fbafe5dc3cdecc17a3d5a7476b160c1b6e7e1.json | 3 +- ...2ede46b4471e21b11f607573a844013911abe.json | 3 +- ...4b3f18dbb388bd8b9bc88b99aebedae9c3fec.json | 20 ++++++++ ...e33244fe8ab9a803248fc23f334034d24aad4.json | 3 +- ...bb6a8f961c119516e4c33dc928cd3b4f4edbc.json | 3 +- ...9e637e9e156cd85b69126c87dfff89a47711d.json | 3 +- ...5693dedf34366e51044ddf30695995d194090.json | 14 ------ ...9746147aa9b8d0a81642a7b7cb4738a0cad66.json | 3 +- ...d434f860922fbc021185beffb9827647feb8e.json | 3 +- ...42b8ad59b9129f032c6b918c27426ab304f2b.json | 20 ++++++++ ...13f1f882b695f896f94e5cf91d205633355a1.json | 3 +- ...e1ea2fe96a1b94f5a92953fb1c1642d15c016.json | 3 +- ...1a8a9cc35f362197c0766a5787436af141106.json | 3 +- ...9352d4d1e3cb3a8a08f1c9c03055a1cab1235.json | 3 +- ...f03f72ef81db302d727e18e63a307ac902022.json | 3 +- ...c35c7342ac9a814c73f6290e1d6ebd3a55423.json | 3 +- ...50d6a6935420fbedc83aa3ad1e9be7465c8dd.json | 3 +- ...d8ba80993a28b2c2e5df70968acf6d7cebe4.json} | 4 +- backend/src/monitor.rs | 48 ++++++++++++++++++- backend/tests/common/mod.rs | 17 +++++-- backend/tests/worker.rs | 2 +- backend/windmill-common/src/bench.rs | 43 ++++++++++++++++- backend/windmill-queue/src/jobs.rs | 34 ++++++------- backend/windmill-worker/src/worker.rs | 17 +++++++ backend/windmill-worker/src/worker_flow.rs | 10 ++-- 50 files changed, 278 insertions(+), 105 deletions(-) delete mode 100644 backend/.sqlx/query-1bceaf6e9f25745b7f70128054ca81d68f3d56d4782e99e05b4f1cb362683514.json create mode 100644 backend/.sqlx/query-5a9cf9cc229f7b7ddbee2485cda4e9f7a91240a00f10d24d3f28d5b722f68768.json create mode 100644 backend/.sqlx/query-a3e75f0309be42aca0fd74834f34b3f18dbb388bd8b9bc88b99aebedae9c3fec.json delete mode 100644 backend/.sqlx/query-bb46e5dcf5490ef3511faa131ad5693dedf34366e51044ddf30695995d194090.json create mode 100644 backend/.sqlx/query-c825fa5c6e287068aeaad994c0b42b8ad59b9129f032c6b918c27426ab304f2b.json rename backend/.sqlx/{query-f8ac5379ecfbff7b8ae75c821680737b249a64c8d9e8f7dbcc46fce98e874571.json => query-f0070b36f7c4fc84dc9c23bb6c73d8ba80993a28b2c2e5df70968acf6d7cebe4.json} (57%) diff --git a/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json b/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json index e7ed0aee65..d29a18c691 100644 --- a/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json +++ b/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json @@ -46,11 +46,11 @@ ] }, "nullable": [ - false, - false, - false, - false, - false, + true, + true, + true, + true, + true, true, true ] diff --git a/backend/.sqlx/query-089d7bc7acdbb97cf477159e111bc7e9ee85289ff5c52af43166928337c257e7.json b/backend/.sqlx/query-089d7bc7acdbb97cf477159e111bc7e9ee85289ff5c52af43166928337c257e7.json index a032a87239..b925141065 100644 --- a/backend/.sqlx/query-089d7bc7acdbb97cf477159e111bc7e9ee85289ff5c52af43166928337c257e7.json +++ b/backend/.sqlx/query-089d7bc7acdbb97cf477159e111bc7e9ee85289ff5c52af43166928337c257e7.json @@ -30,7 +30,8 @@ "sqs", "gcp", "mqtt", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-0f7e01b613a94b29784aae6d7b17b23d6dcf2e5364852a5e85b3c41c417bace2.json b/backend/.sqlx/query-0f7e01b613a94b29784aae6d7b17b23d6dcf2e5364852a5e85b3c41c417bace2.json index 1648882c1a..de4dd4a7e3 100644 --- a/backend/.sqlx/query-0f7e01b613a94b29784aae6d7b17b23d6dcf2e5364852a5e85b3c41c417bace2.json +++ b/backend/.sqlx/query-0f7e01b613a94b29784aae6d7b17b23d6dcf2e5364852a5e85b3c41c417bace2.json @@ -24,7 +24,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-14276a040cb4db88d71fccdc3579e8c0bb132b70668301b535872d1632753e30.json b/backend/.sqlx/query-14276a040cb4db88d71fccdc3579e8c0bb132b70668301b535872d1632753e30.json index d4f7afa966..bed99ef1b7 100644 --- a/backend/.sqlx/query-14276a040cb4db88d71fccdc3579e8c0bb132b70668301b535872d1632753e30.json +++ b/backend/.sqlx/query-14276a040cb4db88d71fccdc3579e8c0bb132b70668301b535872d1632753e30.json @@ -122,7 +122,8 @@ "sqs", "gcp", "mqtt", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-16d438374b03a9c515f4c2d638366f38ffe2f3a0958adea53e67757c6ac463ec.json b/backend/.sqlx/query-16d438374b03a9c515f4c2d638366f38ffe2f3a0958adea53e67757c6ac463ec.json index c091114374..1fdc78c472 100644 --- a/backend/.sqlx/query-16d438374b03a9c515f4c2d638366f38ffe2f3a0958adea53e67757c6ac463ec.json +++ b/backend/.sqlx/query-16d438374b03a9c515f4c2d638366f38ffe2f3a0958adea53e67757c6ac463ec.json @@ -24,7 +24,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-19b59c478744d029c6006b01f04243ad2e0aef485a780daea5d76b0be2bb2ea2.json b/backend/.sqlx/query-19b59c478744d029c6006b01f04243ad2e0aef485a780daea5d76b0be2bb2ea2.json index 43c68f8c5a..30e071f8ce 100644 --- a/backend/.sqlx/query-19b59c478744d029c6006b01f04243ad2e0aef485a780daea5d76b0be2bb2ea2.json +++ b/backend/.sqlx/query-19b59c478744d029c6006b01f04243ad2e0aef485a780daea5d76b0be2bb2ea2.json @@ -40,7 +40,8 @@ "sqs", "gcp", "mqtt", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-1bceaf6e9f25745b7f70128054ca81d68f3d56d4782e99e05b4f1cb362683514.json b/backend/.sqlx/query-1bceaf6e9f25745b7f70128054ca81d68f3d56d4782e99e05b4f1cb362683514.json deleted file mode 100644 index 1be74d47f2..0000000000 --- a/backend/.sqlx/query-1bceaf6e9f25745b7f70128054ca81d68f3d56d4782e99e05b4f1cb362683514.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "WITH inserted_concurrency_counter AS (\n INSERT INTO concurrency_counter (concurrency_id, job_uuids) \n VALUES ($1, '{}'::jsonb)\n ON CONFLICT DO NOTHING\n )\n INSERT INTO concurrency_key(key, job_id) VALUES ($1, $2)", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "1bceaf6e9f25745b7f70128054ca81d68f3d56d4782e99e05b4f1cb362683514" -} diff --git a/backend/.sqlx/query-212553c83e4dcdc6d045eb2fe2dadbb2860ce52d37a56b2861de1215260ecff8.json b/backend/.sqlx/query-212553c83e4dcdc6d045eb2fe2dadbb2860ce52d37a56b2861de1215260ecff8.json index 7d8b2eba1e..b0354db034 100644 --- a/backend/.sqlx/query-212553c83e4dcdc6d045eb2fe2dadbb2860ce52d37a56b2861de1215260ecff8.json +++ b/backend/.sqlx/query-212553c83e4dcdc6d045eb2fe2dadbb2860ce52d37a56b2861de1215260ecff8.json @@ -34,7 +34,8 @@ "sqs", "gcp", "mqtt", - "nextcloud" + "nextcloud", + "google" ] } } @@ -67,7 +68,8 @@ "sqs", "gcp", "mqtt", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-22e0e8a1aa48f8b21763452bd36fbe7db4887c4ac5295052c796bd78a7edc50b.json b/backend/.sqlx/query-22e0e8a1aa48f8b21763452bd36fbe7db4887c4ac5295052c796bd78a7edc50b.json index 66c9e6e96a..235f255dd1 100644 --- a/backend/.sqlx/query-22e0e8a1aa48f8b21763452bd36fbe7db4887c4ac5295052c796bd78a7edc50b.json +++ b/backend/.sqlx/query-22e0e8a1aa48f8b21763452bd36fbe7db4887c4ac5295052c796bd78a7edc50b.json @@ -24,7 +24,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-23419adcd74c326d716527293eff518b42f4cdb33e034441015494bd26c172d2.json b/backend/.sqlx/query-23419adcd74c326d716527293eff518b42f4cdb33e034441015494bd26c172d2.json index 242b358ff5..7bd7367d8d 100644 --- a/backend/.sqlx/query-23419adcd74c326d716527293eff518b42f4cdb33e034441015494bd26c172d2.json +++ b/backend/.sqlx/query-23419adcd74c326d716527293eff518b42f4cdb33e034441015494bd26c172d2.json @@ -40,7 +40,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-42b4b73e9d60348e2d90fcade9dcad6d8995242dc20a4e14c1a8fae4fc6a9fd2.json b/backend/.sqlx/query-42b4b73e9d60348e2d90fcade9dcad6d8995242dc20a4e14c1a8fae4fc6a9fd2.json index e6d71f386f..537ecbeacf 100644 --- a/backend/.sqlx/query-42b4b73e9d60348e2d90fcade9dcad6d8995242dc20a4e14c1a8fae4fc6a9fd2.json +++ b/backend/.sqlx/query-42b4b73e9d60348e2d90fcade9dcad6d8995242dc20a4e14c1a8fae4fc6a9fd2.json @@ -24,7 +24,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-45024b932383199974616bba1fc2f7175cc6f2e02d9c565bb5159cae3e0b6835.json b/backend/.sqlx/query-45024b932383199974616bba1fc2f7175cc6f2e02d9c565bb5159cae3e0b6835.json index f215cf6147..30c5ff7a49 100644 --- a/backend/.sqlx/query-45024b932383199974616bba1fc2f7175cc6f2e02d9c565bb5159cae3e0b6835.json +++ b/backend/.sqlx/query-45024b932383199974616bba1fc2f7175cc6f2e02d9c565bb5159cae3e0b6835.json @@ -30,7 +30,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-4d272cf4a77aab7007a5b35589e08532a1020cabaf5e22325a1e05f0491d785c.json b/backend/.sqlx/query-4d272cf4a77aab7007a5b35589e08532a1020cabaf5e22325a1e05f0491d785c.json index 896b21dc7f..cfb975b838 100644 --- a/backend/.sqlx/query-4d272cf4a77aab7007a5b35589e08532a1020cabaf5e22325a1e05f0491d785c.json +++ b/backend/.sqlx/query-4d272cf4a77aab7007a5b35589e08532a1020cabaf5e22325a1e05f0491d785c.json @@ -37,7 +37,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-4f547c0fd54f3bc57212ce87810e35adf640d44d607e62a1fb296e38ac3fdd36.json b/backend/.sqlx/query-4f547c0fd54f3bc57212ce87810e35adf640d44d607e62a1fb296e38ac3fdd36.json index 789c0334d7..4e8ec56d12 100644 --- a/backend/.sqlx/query-4f547c0fd54f3bc57212ce87810e35adf640d44d607e62a1fb296e38ac3fdd36.json +++ b/backend/.sqlx/query-4f547c0fd54f3bc57212ce87810e35adf640d44d607e62a1fb296e38ac3fdd36.json @@ -32,7 +32,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } @@ -70,7 +71,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-5a9cf9cc229f7b7ddbee2485cda4e9f7a91240a00f10d24d3f28d5b722f68768.json b/backend/.sqlx/query-5a9cf9cc229f7b7ddbee2485cda4e9f7a91240a00f10d24d3f28d5b722f68768.json new file mode 100644 index 0000000000..a971ecc058 --- /dev/null +++ b/backend/.sqlx/query-5a9cf9cc229f7b7ddbee2485cda4e9f7a91240a00f10d24d3f28d5b722f68768.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "WITH inserted_concurrency_counter AS (\n INSERT INTO concurrency_counter (concurrency_id, job_uuids)\n VALUES ($1, '{}'::jsonb)\n ON CONFLICT DO NOTHING\n )\n INSERT INTO concurrency_key(key, job_id) VALUES ($1, $2)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "5a9cf9cc229f7b7ddbee2485cda4e9f7a91240a00f10d24d3f28d5b722f68768" +} diff --git a/backend/.sqlx/query-5bf200f2c8db25ddf231b564503c6c70f7f3958564a79bb0c6b3863b1ebb0cbf.json b/backend/.sqlx/query-5bf200f2c8db25ddf231b564503c6c70f7f3958564a79bb0c6b3863b1ebb0cbf.json index 6d3941ad69..9422e3e8d0 100644 --- a/backend/.sqlx/query-5bf200f2c8db25ddf231b564503c6c70f7f3958564a79bb0c6b3863b1ebb0cbf.json +++ b/backend/.sqlx/query-5bf200f2c8db25ddf231b564503c6c70f7f3958564a79bb0c6b3863b1ebb0cbf.json @@ -245,7 +245,8 @@ "sqs", "gcp", "mqtt", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-66a0e51cf149ba532463e29dd361a803e1bced2f8e1a12f8933b7598ee85a147.json b/backend/.sqlx/query-66a0e51cf149ba532463e29dd361a803e1bced2f8e1a12f8933b7598ee85a147.json index 0d12be2448..5f9be1cba9 100644 --- a/backend/.sqlx/query-66a0e51cf149ba532463e29dd361a803e1bced2f8e1a12f8933b7598ee85a147.json +++ b/backend/.sqlx/query-66a0e51cf149ba532463e29dd361a803e1bced2f8e1a12f8933b7598ee85a147.json @@ -35,7 +35,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-7065f23d04e26831664048f2cfc4f412c57af931f80621aee5012e9cb3535626.json b/backend/.sqlx/query-7065f23d04e26831664048f2cfc4f412c57af931f80621aee5012e9cb3535626.json index 02823e8291..48ea42f8a2 100644 --- a/backend/.sqlx/query-7065f23d04e26831664048f2cfc4f412c57af931f80621aee5012e9cb3535626.json +++ b/backend/.sqlx/query-7065f23d04e26831664048f2cfc4f412c57af931f80621aee5012e9cb3535626.json @@ -29,7 +29,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-73fdd01bad58b8be1a52f89faef8d92a983470adcd3cc850734960c905e61e83.json b/backend/.sqlx/query-73fdd01bad58b8be1a52f89faef8d92a983470adcd3cc850734960c905e61e83.json index cba7ffdfef..8aacd8a805 100644 --- a/backend/.sqlx/query-73fdd01bad58b8be1a52f89faef8d92a983470adcd3cc850734960c905e61e83.json +++ b/backend/.sqlx/query-73fdd01bad58b8be1a52f89faef8d92a983470adcd3cc850734960c905e61e83.json @@ -24,7 +24,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-757ef6215d3d385cb3a69e26ee4ca846dd5e7fe7ceb1aa8b3fcd26a2bd30eb2c.json b/backend/.sqlx/query-757ef6215d3d385cb3a69e26ee4ca846dd5e7fe7ceb1aa8b3fcd26a2bd30eb2c.json index cd795e6fec..59b56ceda7 100644 --- a/backend/.sqlx/query-757ef6215d3d385cb3a69e26ee4ca846dd5e7fe7ceb1aa8b3fcd26a2bd30eb2c.json +++ b/backend/.sqlx/query-757ef6215d3d385cb3a69e26ee4ca846dd5e7fe7ceb1aa8b3fcd26a2bd30eb2c.json @@ -40,7 +40,8 @@ "sqs", "gcp", "mqtt", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-7fbf72d9059fcd77e4c1112fa4fa22e4276c1da653475628889ce17dc904fbaa.json b/backend/.sqlx/query-7fbf72d9059fcd77e4c1112fa4fa22e4276c1da653475628889ce17dc904fbaa.json index 393a920b7c..8c380df861 100644 --- a/backend/.sqlx/query-7fbf72d9059fcd77e4c1112fa4fa22e4276c1da653475628889ce17dc904fbaa.json +++ b/backend/.sqlx/query-7fbf72d9059fcd77e4c1112fa4fa22e4276c1da653475628889ce17dc904fbaa.json @@ -27,7 +27,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-836bac47d89113d90bd03a471446eb9016207975af1e37042d81df8cb6ae2c53.json b/backend/.sqlx/query-836bac47d89113d90bd03a471446eb9016207975af1e37042d81df8cb6ae2c53.json index 437d644eb2..3012e9ef77 100644 --- a/backend/.sqlx/query-836bac47d89113d90bd03a471446eb9016207975af1e37042d81df8cb6ae2c53.json +++ b/backend/.sqlx/query-836bac47d89113d90bd03a471446eb9016207975af1e37042d81df8cb6ae2c53.json @@ -24,7 +24,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-87564a196a1662f524407d853db506bf08c28efe82b68b3d44bafbd3d0e91c29.json b/backend/.sqlx/query-87564a196a1662f524407d853db506bf08c28efe82b68b3d44bafbd3d0e91c29.json index 9350442134..5a61f82be3 100644 --- a/backend/.sqlx/query-87564a196a1662f524407d853db506bf08c28efe82b68b3d44bafbd3d0e91c29.json +++ b/backend/.sqlx/query-87564a196a1662f524407d853db506bf08c28efe82b68b3d44bafbd3d0e91c29.json @@ -35,7 +35,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-940b6d78bab940a37a42492f030d2393e297043e4e58555d872b5c4dd89c196a.json b/backend/.sqlx/query-940b6d78bab940a37a42492f030d2393e297043e4e58555d872b5c4dd89c196a.json index 47554ad43e..09aadbda1e 100644 --- a/backend/.sqlx/query-940b6d78bab940a37a42492f030d2393e297043e4e58555d872b5c4dd89c196a.json +++ b/backend/.sqlx/query-940b6d78bab940a37a42492f030d2393e297043e4e58555d872b5c4dd89c196a.json @@ -24,7 +24,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-9c50e3a136a8ee3ec56e083f26d3a960b89e02ec40b292f3b5198baf2a1d3dbf.json b/backend/.sqlx/query-9c50e3a136a8ee3ec56e083f26d3a960b89e02ec40b292f3b5198baf2a1d3dbf.json index 9759bad4d4..66a487695f 100644 --- a/backend/.sqlx/query-9c50e3a136a8ee3ec56e083f26d3a960b89e02ec40b292f3b5198baf2a1d3dbf.json +++ b/backend/.sqlx/query-9c50e3a136a8ee3ec56e083f26d3a960b89e02ec40b292f3b5198baf2a1d3dbf.json @@ -32,7 +32,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-9ecb404e46a4eac55f977f05a3afbafe5dc3cdecc17a3d5a7476b160c1b6e7e1.json b/backend/.sqlx/query-9ecb404e46a4eac55f977f05a3afbafe5dc3cdecc17a3d5a7476b160c1b6e7e1.json index 79c8f0b45d..b6fe1b7fa0 100644 --- a/backend/.sqlx/query-9ecb404e46a4eac55f977f05a3afbafe5dc3cdecc17a3d5a7476b160c1b6e7e1.json +++ b/backend/.sqlx/query-9ecb404e46a4eac55f977f05a3afbafe5dc3cdecc17a3d5a7476b160c1b6e7e1.json @@ -30,7 +30,8 @@ "sqs", "gcp", "mqtt", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-a1745a4f525b251d2f5a602ab2b2ede46b4471e21b11f607573a844013911abe.json b/backend/.sqlx/query-a1745a4f525b251d2f5a602ab2b2ede46b4471e21b11f607573a844013911abe.json index a25f845f91..33636da608 100644 --- a/backend/.sqlx/query-a1745a4f525b251d2f5a602ab2b2ede46b4471e21b11f607573a844013911abe.json +++ b/backend/.sqlx/query-a1745a4f525b251d2f5a602ab2b2ede46b4471e21b11f607573a844013911abe.json @@ -155,7 +155,8 @@ "sqs", "gcp", "mqtt", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-a3e75f0309be42aca0fd74834f34b3f18dbb388bd8b9bc88b99aebedae9c3fec.json b/backend/.sqlx/query-a3e75f0309be42aca0fd74834f34b3f18dbb388bd8b9bc88b99aebedae9c3fec.json new file mode 100644 index 0000000000..66af36c0fd --- /dev/null +++ b/backend/.sqlx/query-a3e75f0309be42aca0fd74834f34b3f18dbb388bd8b9bc88b99aebedae9c3fec.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM job_result_stream_v2 WHERE job_id NOT IN (SELECT id FROM v2_job_queue) RETURNING job_id", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "job_id", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + false + ] + }, + "hash": "a3e75f0309be42aca0fd74834f34b3f18dbb388bd8b9bc88b99aebedae9c3fec" +} diff --git a/backend/.sqlx/query-a4b6371d33206010b2f3ffd2b09e33244fe8ab9a803248fc23f334034d24aad4.json b/backend/.sqlx/query-a4b6371d33206010b2f3ffd2b09e33244fe8ab9a803248fc23f334034d24aad4.json index da6d213748..03ee58ca3f 100644 --- a/backend/.sqlx/query-a4b6371d33206010b2f3ffd2b09e33244fe8ab9a803248fc23f334034d24aad4.json +++ b/backend/.sqlx/query-a4b6371d33206010b2f3ffd2b09e33244fe8ab9a803248fc23f334034d24aad4.json @@ -185,7 +185,8 @@ "sqs", "gcp", "mqtt", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-b3771b690c5966272b1f42c9965bb6a8f961c119516e4c33dc928cd3b4f4edbc.json b/backend/.sqlx/query-b3771b690c5966272b1f42c9965bb6a8f961c119516e4c33dc928cd3b4f4edbc.json index d08af6ffdd..4fa7da00e0 100644 --- a/backend/.sqlx/query-b3771b690c5966272b1f42c9965bb6a8f961c119516e4c33dc928cd3b4f4edbc.json +++ b/backend/.sqlx/query-b3771b690c5966272b1f42c9965bb6a8f961c119516e4c33dc928cd3b4f4edbc.json @@ -160,7 +160,8 @@ "sqs", "gcp", "mqtt", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-b3f0595cacba194e08b9a3e244d9e637e9e156cd85b69126c87dfff89a47711d.json b/backend/.sqlx/query-b3f0595cacba194e08b9a3e244d9e637e9e156cd85b69126c87dfff89a47711d.json index 54a4e3cd93..186822a000 100644 --- a/backend/.sqlx/query-b3f0595cacba194e08b9a3e244d9e637e9e156cd85b69126c87dfff89a47711d.json +++ b/backend/.sqlx/query-b3f0595cacba194e08b9a3e244d9e637e9e156cd85b69126c87dfff89a47711d.json @@ -24,7 +24,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-bb46e5dcf5490ef3511faa131ad5693dedf34366e51044ddf30695995d194090.json b/backend/.sqlx/query-bb46e5dcf5490ef3511faa131ad5693dedf34366e51044ddf30695995d194090.json deleted file mode 100644 index d9ca7cfa52..0000000000 --- a/backend/.sqlx/query-bb46e5dcf5490ef3511faa131ad5693dedf34366e51044ddf30695995d194090.json +++ /dev/null @@ -1,14 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "DELETE FROM job_result_stream_v2 WHERE job_id = $1", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "bb46e5dcf5490ef3511faa131ad5693dedf34366e51044ddf30695995d194090" -} diff --git a/backend/.sqlx/query-be6d2c92a62b7b284651c45af809746147aa9b8d0a81642a7b7cb4738a0cad66.json b/backend/.sqlx/query-be6d2c92a62b7b284651c45af809746147aa9b8d0a81642a7b7cb4738a0cad66.json index 0688afebd3..4118760af2 100644 --- a/backend/.sqlx/query-be6d2c92a62b7b284651c45af809746147aa9b8d0a81642a7b7cb4738a0cad66.json +++ b/backend/.sqlx/query-be6d2c92a62b7b284651c45af809746147aa9b8d0a81642a7b7cb4738a0cad66.json @@ -105,7 +105,8 @@ "sqs", "gcp", "mqtt", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-c3b1152b554812d65eb27f95b1fd434f860922fbc021185beffb9827647feb8e.json b/backend/.sqlx/query-c3b1152b554812d65eb27f95b1fd434f860922fbc021185beffb9827647feb8e.json index 0098e51ab2..07243717a0 100644 --- a/backend/.sqlx/query-c3b1152b554812d65eb27f95b1fd434f860922fbc021185beffb9827647feb8e.json +++ b/backend/.sqlx/query-c3b1152b554812d65eb27f95b1fd434f860922fbc021185beffb9827647feb8e.json @@ -31,7 +31,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-c825fa5c6e287068aeaad994c0b42b8ad59b9129f032c6b918c27426ab304f2b.json b/backend/.sqlx/query-c825fa5c6e287068aeaad994c0b42b8ad59b9129f032c6b918c27426ab304f2b.json new file mode 100644 index 0000000000..a68387f905 --- /dev/null +++ b/backend/.sqlx/query-c825fa5c6e287068aeaad994c0b42b8ad59b9129f032c6b918c27426ab304f2b.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM job_perms\nWHERE job_id NOT IN (SELECT id FROM v2_job_queue)\nRETURNING job_id", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "job_id", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + false + ] + }, + "hash": "c825fa5c6e287068aeaad994c0b42b8ad59b9129f032c6b918c27426ab304f2b" +} diff --git a/backend/.sqlx/query-ccef7a1bde5cac6c362c5fedb6c13f1f882b695f896f94e5cf91d205633355a1.json b/backend/.sqlx/query-ccef7a1bde5cac6c362c5fedb6c13f1f882b695f896f94e5cf91d205633355a1.json index 5091b2fc69..2b39145512 100644 --- a/backend/.sqlx/query-ccef7a1bde5cac6c362c5fedb6c13f1f882b695f896f94e5cf91d205633355a1.json +++ b/backend/.sqlx/query-ccef7a1bde5cac6c362c5fedb6c13f1f882b695f896f94e5cf91d205633355a1.json @@ -24,7 +24,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-d41ea93fd58381b89e151c965eae1ea2fe96a1b94f5a92953fb1c1642d15c016.json b/backend/.sqlx/query-d41ea93fd58381b89e151c965eae1ea2fe96a1b94f5a92953fb1c1642d15c016.json index 11271e94d8..9d2f8d9a9f 100644 --- a/backend/.sqlx/query-d41ea93fd58381b89e151c965eae1ea2fe96a1b94f5a92953fb1c1642d15c016.json +++ b/backend/.sqlx/query-d41ea93fd58381b89e151c965eae1ea2fe96a1b94f5a92953fb1c1642d15c016.json @@ -105,7 +105,8 @@ "sqs", "gcp", "mqtt", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-d495c94b580fd34d5ae90615ef21a8a9cc35f362197c0766a5787436af141106.json b/backend/.sqlx/query-d495c94b580fd34d5ae90615ef21a8a9cc35f362197c0766a5787436af141106.json index 9cef6500d3..67a38d1904 100644 --- a/backend/.sqlx/query-d495c94b580fd34d5ae90615ef21a8a9cc35f362197c0766a5787436af141106.json +++ b/backend/.sqlx/query-d495c94b580fd34d5ae90615ef21a8a9cc35f362197c0766a5787436af141106.json @@ -25,7 +25,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-e4d71278fb80126a7a9da73f1889352d4d1e3cb3a8a08f1c9c03055a1cab1235.json b/backend/.sqlx/query-e4d71278fb80126a7a9da73f1889352d4d1e3cb3a8a08f1c9c03055a1cab1235.json index 5b07bcd9c9..7309b03a02 100644 --- a/backend/.sqlx/query-e4d71278fb80126a7a9da73f1889352d4d1e3cb3a8a08f1c9c03055a1cab1235.json +++ b/backend/.sqlx/query-e4d71278fb80126a7a9da73f1889352d4d1e3cb3a8a08f1c9c03055a1cab1235.json @@ -185,7 +185,8 @@ "sqs", "gcp", "mqtt", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-e80177f3ffd4c1f52cdb4757483f03f72ef81db302d727e18e63a307ac902022.json b/backend/.sqlx/query-e80177f3ffd4c1f52cdb4757483f03f72ef81db302d727e18e63a307ac902022.json index 1e38e57fb5..dd65a586d1 100644 --- a/backend/.sqlx/query-e80177f3ffd4c1f52cdb4757483f03f72ef81db302d727e18e63a307ac902022.json +++ b/backend/.sqlx/query-e80177f3ffd4c1f52cdb4757483f03f72ef81db302d727e18e63a307ac902022.json @@ -31,7 +31,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-eac595e19e5c8e70f1514ef29dec35c7342ac9a814c73f6290e1d6ebd3a55423.json b/backend/.sqlx/query-eac595e19e5c8e70f1514ef29dec35c7342ac9a814c73f6290e1d6ebd3a55423.json index 57607ae052..7c45c44b13 100644 --- a/backend/.sqlx/query-eac595e19e5c8e70f1514ef29dec35c7342ac9a814c73f6290e1d6ebd3a55423.json +++ b/backend/.sqlx/query-eac595e19e5c8e70f1514ef29dec35c7342ac9a814c73f6290e1d6ebd3a55423.json @@ -24,7 +24,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-ed8facbf29ebb670d05fe8aa34b50d6a6935420fbedc83aa3ad1e9be7465c8dd.json b/backend/.sqlx/query-ed8facbf29ebb670d05fe8aa34b50d6a6935420fbedc83aa3ad1e9be7465c8dd.json index 43fd90abe5..1876db927e 100644 --- a/backend/.sqlx/query-ed8facbf29ebb670d05fe8aa34b50d6a6935420fbedc83aa3ad1e9be7465c8dd.json +++ b/backend/.sqlx/query-ed8facbf29ebb670d05fe8aa34b50d6a6935420fbedc83aa3ad1e9be7465c8dd.json @@ -24,7 +24,8 @@ "mqtt", "gcp", "default_email", - "nextcloud" + "nextcloud", + "google" ] } } diff --git a/backend/.sqlx/query-f8ac5379ecfbff7b8ae75c821680737b249a64c8d9e8f7dbcc46fce98e874571.json b/backend/.sqlx/query-f0070b36f7c4fc84dc9c23bb6c73d8ba80993a28b2c2e5df70968acf6d7cebe4.json similarity index 57% rename from backend/.sqlx/query-f8ac5379ecfbff7b8ae75c821680737b249a64c8d9e8f7dbcc46fce98e874571.json rename to backend/.sqlx/query-f0070b36f7c4fc84dc9c23bb6c73d8ba80993a28b2c2e5df70968acf6d7cebe4.json index 70fbb752eb..b31e532fc2 100644 --- a/backend/.sqlx/query-f8ac5379ecfbff7b8ae75c821680737b249a64c8d9e8f7dbcc46fce98e874571.json +++ b/backend/.sqlx/query-f0070b36f7c4fc84dc9c23bb6c73d8ba80993a28b2c2e5df70968acf6d7cebe4.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "VACUUM v2_job, v2_job_completed, job_result_stream_v2, job_stats, job_logs, concurrency_key, log_file, metrics", + "query": "VACUUM v2_job, v2_job_completed, job_result_stream_v2, job_stats, job_logs, job_perms, concurrency_key, log_file, metrics", "describe": { "columns": [], "parameters": { @@ -8,5 +8,5 @@ }, "nullable": [] }, - "hash": "f8ac5379ecfbff7b8ae75c821680737b249a64c8d9e8f7dbcc46fce98e874571" + "hash": "f0070b36f7c4fc84dc9c23bb6c73d8ba80993a28b2c2e5df70968acf6d7cebe4" } diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 909202b8e3..bfbfaa2d87 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -1103,7 +1103,6 @@ async fn delete_expired_jobs_batch( job_retention_secs ); - // Delete related records for this batch if let Err(e) = sqlx::query!( "DELETE FROM job_stats WHERE job_id = ANY($1)", &deleted_jobs @@ -1783,6 +1782,19 @@ pub async fn monitor_db( } }; + let cleanup_job_live_rows_f = async { + if server_mode && !initial_load { + if let Some(db) = conn.as_sql() { + if let Err(e) = cleanup_job_perms_orphaned(&db).await { + tracing::error!("Error cleaning up orphaned job_perms: {:?}", e); + } + if let Err(e) = cleanup_job_result_stream_orphaned_jobs(&db).await { + tracing::error!("Error cleaning up orphaned job_result_stream_v2: {:?}", e); + } + } + } + }; + // run every hour (60 minutes / 30 seconds = 120) let cleanup_worker_group_stats_f = async { if server_mode && iteration.is_some() && iteration.as_ref().unwrap().should_run(120) { @@ -1956,6 +1968,7 @@ pub async fn monitor_db( cleanup_debounce_keys_f, cleanup_debounce_keys_completed_f, cleanup_flow_iterator_data_f, + cleanup_job_live_rows_f, cleanup_worker_group_stats_f, native_triggers_sync_f, cleanup_notify_events_f, @@ -1963,7 +1976,7 @@ pub async fn monitor_db( } async fn vacuuming_tables(db: &Pool) -> error::Result<()> { - sqlx::query!("VACUUM v2_job, v2_job_completed, job_result_stream_v2, job_stats, job_logs, concurrency_key, log_file, metrics") + sqlx::query!("VACUUM v2_job, v2_job_completed, job_result_stream_v2, job_stats, job_logs, job_perms, concurrency_key, log_file, metrics") .execute(db) .await?; Ok(()) @@ -3062,6 +3075,37 @@ RETURNING key,job_id Ok(()) } +async fn cleanup_job_perms_orphaned(db: &DB) -> error::Result<()> { + let result = sqlx::query_scalar!( + "DELETE FROM job_perms +WHERE job_id NOT IN (SELECT id FROM v2_job_queue) +RETURNING job_id" + ) + .fetch_all(db) + .await?; + + if !result.is_empty() { + tracing::info!("Cleaned up {} orphaned job_perms rows", result.len()); + } + Ok(()) +} + +async fn cleanup_job_result_stream_orphaned_jobs(db: &DB) -> error::Result<()> { + let result = sqlx::query!( + "DELETE FROM job_result_stream_v2 WHERE job_id NOT IN (SELECT id FROM v2_job_queue) RETURNING job_id", + ) + .fetch_all(db) + .await?; + + if result.len() > 0 { + tracing::info!( + "Cleaned up {} orphaned job_result_stream_v2 rows", + result.len() + ); + } + Ok(()) +} + async fn cleanup_flow_iterator_data_orphaned_jobs(db: &DB) -> error::Result<()> { let result = sqlx::query!( " diff --git a/backend/tests/common/mod.rs b/backend/tests/common/mod.rs index 7f98d0937c..d19fdac786 100644 --- a/backend/tests/common/mod.rs +++ b/backend/tests/common/mod.rs @@ -149,11 +149,17 @@ pub struct RunJob { pub payload: JobPayload, pub args: serde_json::Map, pub scheduled_for_o: Option>, + pub email: String, } impl From for RunJob { fn from(payload: JobPayload) -> Self { - Self { payload, args: Default::default(), scheduled_for_o: None } + Self { + payload, + args: Default::default(), + scheduled_for_o: None, + email: "test@windmill.dev".to_string(), + } } } @@ -171,8 +177,13 @@ impl RunJob { self } + pub fn email(mut self, email: impl Into) -> Self { + self.email = email.into(); + self + } + pub async fn push(self, db: &Pool) -> Uuid { - let RunJob { payload, args, scheduled_for_o } = self; + let RunJob { payload, args, scheduled_for_o, email } = self; let mut hm_args = std::collections::HashMap::new(); for (k, v) in args { hm_args.insert(k, windmill_common::worker::to_raw_value(&v)); @@ -186,7 +197,7 @@ impl RunJob { payload, windmill_queue::PushArgs::from(&hm_args), /* user */ "test-user", - /* email */ "test@windmill.dev", + /* email */ &email, /* permissioned_as */ "u/test-user".to_string(), /* token_prefix */ None, scheduled_for_o, diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index da936c4f18..f98e1e3ad7 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -3343,7 +3343,6 @@ async fn test_duckdb_ffi(db: Pool) -> anyhow::Result<()> { /// Test that flow substeps with tags that are not available for the workspace fail. /// This validates that `check_tag_available_for_workspace_internal` is properly called /// when pushing jobs from worker_flow. -#[cfg(feature = "deno_core")] #[sqlx::test(fixtures("base"))] async fn test_flow_substep_tag_availability_check(db: Pool) -> anyhow::Result<()> { use windmill_common::worker::{ @@ -3385,6 +3384,7 @@ async fn test_flow_substep_tag_availability_check(db: Pool) -> anyhow: let result = RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) + .email("test2@windmill.dev") .run_until_complete(&db, false, server.addr.port()) .await; diff --git a/backend/windmill-common/src/bench.rs b/backend/windmill-common/src/bench.rs index c851198b2a..cd475edaac 100644 --- a/backend/windmill-common/src/bench.rs +++ b/backend/windmill-common/src/bench.rs @@ -5,6 +5,31 @@ use crate::{ use serde::Serialize; use tokio::time::Instant; +#[derive(Serialize)] +pub struct PoolStats { + pub peak_active_conns: u32, + pub pool_saturation_histogram: Vec, +} + +impl PoolStats { + pub fn new(pool_size: u32) -> Self { + PoolStats { + peak_active_conns: 0, + pool_saturation_histogram: vec![0; pool_size as usize + 1], + } + } + + pub fn sample(&mut self, pool_size: u32, num_idle: u32) { + let active = pool_size.saturating_sub(num_idle); + if active > self.peak_active_conns { + self.peak_active_conns = active; + } + if let Some(bucket) = self.pool_saturation_histogram.get_mut(active as usize) { + *bucket += 1; + } + } +} + #[derive(Serialize)] pub struct BenchmarkInfo { #[serde(skip)] @@ -14,6 +39,7 @@ pub struct BenchmarkInfo { timings: Vec, pub iter_durations: Vec, pub total_duration: Option, + pub pool_stats: Option, } impl BenchmarkInfo { @@ -24,6 +50,17 @@ impl BenchmarkInfo { start: Instant::now(), iter_durations: vec![], total_duration: None, + pool_stats: None, + } + } + + pub fn init_pool_stats(&mut self, pool_size: u32) { + self.pool_stats = Some(PoolStats::new(pool_size)); + } + + pub fn sample_pool(&mut self, pool_size: u32, num_idle: u32) { + if let Some(stats) = self.pool_stats.as_mut() { + stats.sample(pool_size, num_idle); } } @@ -40,8 +77,11 @@ impl BenchmarkInfo { let total_duration = self.start.elapsed().as_millis() as u64; self.total_duration = Some(total_duration as u64); + let pool_info = self.pool_stats.as_ref().map_or(String::new(), |ps| { + format!(", peak active conns: {}", ps.peak_active_conns) + }); println!( - "Writing benchmark {path}, duration of benchmark: {total_duration}ms and RPS: {}", + "Writing benchmark {path}, duration of benchmark: {total_duration}ms and RPS: {}{pool_info}", self.iters as f64 / total_duration as f64 * 1000.0 ); write_file(TMP_DIR, path, &serde_json::to_string(&self).unwrap()).expect("write profiling"); @@ -220,6 +260,7 @@ pub async fn benchmark_init(benchmark_jobs: i32, db: &DB) { .await .unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs (4)")); } + "none" => {} _ => { let uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9 FROM generate_series(1, $10)) RETURNING id", None::, diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index e414b45003..a6148922f0 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -1184,7 +1184,19 @@ async fn commit_completed_job( .concurrent_limit .is_some() { - let concurrency_key = concurrency_key(db, &completed_job.id).await?; + let concurrency_key = sqlx::query_scalar!( + "SELECT key FROM concurrency_key WHERE job_id = $1", + &completed_job.id + ) + .fetch_optional(&mut *tx) + .warn_after_seconds(10) + .await + .map_err(|e| { + Error::internal_err(format!( + "Could not get concurrency key for job {}: {e:#}", + completed_job.id + )) + })?; if *DISABLE_CONCURRENCY_LIMIT || concurrency_key.is_none() { tracing::warn!("Concurrency limit is disabled, skipping"); } else { @@ -1221,18 +1233,6 @@ async fn commit_completed_job( tracing::debug!("decremented concurrency counter"); } - sqlx::query!("DELETE FROM job_perms WHERE job_id = $1", job_id) - .execute(&mut *tx) - .warn_after_seconds(10) - .await?; - - if !success || has_stream { - sqlx::query!("DELETE FROM job_result_stream_v2 WHERE job_id = $1", job_id) - .execute(&mut *tx) - .warn_after_seconds(10) - .await?; - } - tx.commit().warn_after_seconds(10).await?; tracing::info!( @@ -5386,7 +5386,7 @@ async fn push_inner<'c, 'd>( &runnable_path, job_kind, concurrency_settings.concurrency_key.clone(), - &mut tx, + &mut *tx, job_id, ) .await?; @@ -5709,7 +5709,7 @@ pub async fn insert_concurrency_key<'d, 'c>( script_path: &Option, job_kind: JobKind, custom_concurrency_key: Option, - tx: &mut Transaction<'c, Postgres>, + db: impl PgExecutor<'c>, job_id: Uuid, ) -> Result<(), Error> { let concurrency_key = custom_concurrency_key @@ -5737,7 +5737,7 @@ pub async fn insert_concurrency_key<'d, 'c>( )); sqlx::query!( "WITH inserted_concurrency_counter AS ( - INSERT INTO concurrency_counter (concurrency_id, job_uuids) + INSERT INTO concurrency_counter (concurrency_id, job_uuids) VALUES ($1, '{}'::jsonb) ON CONFLICT DO NOTHING ) @@ -5745,7 +5745,7 @@ pub async fn insert_concurrency_key<'d, 'c>( concurrency_key, job_id, ) - .execute(&mut **tx) + .execute(db) .warn_after_seconds(3) .await .map_err(|e| Error::internal_err(format!("Could not insert concurrency_key={concurrency_key} for job_id={job_id} script_path={script_path:?} workspace_id={workspace_id}: {e:#}")))?; diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 608ae1c517..be8b44993e 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -1618,6 +1618,11 @@ pub async fn run_worker( #[cfg(feature = "benchmark")] let mut infos = BenchmarkInfo::new(); + #[cfg(feature = "benchmark")] + if let Some(db) = conn.as_sql() { + infos.init_pool_stats(db.size()); + } + let vacuum_shift = rand::rng().random_range(0..VACUUM_PERIOD); IS_READY.store(true, Ordering::Relaxed); @@ -2067,6 +2072,9 @@ pub async fn run_worker( #[cfg(feature = "benchmark")] { add_time!(bench, "sent to dedicated worker"); + if let Some(db) = conn.as_sql() { + infos.sample_pool(db.size(), db.num_idle() as u32); + } infos.add_iter(bench, true); } @@ -2137,6 +2145,9 @@ pub async fn run_worker( #[cfg(feature = "benchmark")] { add_time!(bench, "sent to flow runner"); + if let Some(db) = conn.as_sql() { + infos.sample_pool(db.size(), db.num_idle() as u32); + } infos.add_iter(bench, true); } @@ -2437,6 +2448,9 @@ pub async fn run_worker( { if started { add_time!(bench, "job processed"); + if let Some(db) = conn.as_sql() { + infos.sample_pool(db.size(), db.num_idle() as u32); + } infos.add_iter(bench, true); } } @@ -2465,6 +2479,9 @@ pub async fn run_worker( #[cfg(feature = "benchmark")] { add_time!(bench, "sleep because empty job queue"); + if let Some(db) = conn.as_sql() { + infos.sample_pool(db.size(), db.num_idle() as u32); + } infos.add_iter(bench, false); } #[cfg(feature = "prometheus")] diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index bf32ccc9da..b1ef84f342 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -1365,35 +1365,31 @@ pub async fn update_flow_status_after_job_completion_internal( let args_hm = args.unwrap_or_default().0; let args = PushArgs::from(&args_hm); if let Some(ck) = concurrency_key { - let mut tx = db.begin().await?; insert_concurrency_key( &flow_job.workspace_id, &args, &flow_job.runnable_path, JobKind::Flow, Some(ck), - &mut tx, + db, flow, ) .await?; - tx.commit().await?; } if let Some(t) = tag { tag = Some(interpolate_args(t, &args, &flow_job.workspace_id)); } } else if concurrent_limit.is_some() { - let mut tx = db.begin().await?; insert_concurrency_key( &flow_job.workspace_id, &PushArgs::from(&HashMap::new()), &flow_job.runnable_path, JobKind::Flow, concurrency_key, - &mut tx, + db, flow, ) .await?; - tx.commit().await?; } // let tag = tag_and_concurrency_key.and_then(|tc| tc.tag.map(|t| interpolate_args(t.clone(), &args, &workspace_id))); @@ -3257,7 +3253,7 @@ async fn push_next_flow_job( "UPDATE v2_job_runtime SET ping = now() WHERE id = $1 AND ping < now()", flow_job.id, ) - .execute(db) + .execute(&mut *tx) .warn_after_seconds(3) .await?; }