starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-06 16:02:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

backend: migrate job tables to v2 schema (v2 phase 1) (#5084)

Browse Source
This commit is contained in:
Lucas Abel
2025-02-06 12:37:11 +01:00
committed by GitHub
parent 20815ecb1c
commit cecc19dafc
82 changed files with 2839 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-036c84bb9ce72748956bc9c18fbe276444fab025a281dc4784596b0e31c1cb9d.json
Generated
+12
View File
@@ -0,0 +1,12 @@
{
"db_name": "PostgreSQL",
"query": "create index concurrently if not exists ix_job_workspace_id_created_at_new_9 ON v2_job (workspace_id, created_at DESC) where kind in ('dependencies', 'flowdependencies', 'appdependencies') AND parent_job IS NULL",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "036c84bb9ce72748956bc9c18fbe276444fab025a281dc4784596b0e31c1cb9d"
}
backend/.sqlx/query-0df84fc35f2780ceb7c473b0165ebab93a4bc1bcab166aae68244ab1f3d4df9f.json
Generated
+104
View File
@@ -0,0 +1,104 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO completed_job AS cj\n ( workspace_id\n , id\n , parent_job\n , created_by\n , created_at\n , started_at\n , duration_ms\n , success\n , script_hash\n , script_path\n , args\n , result\n , raw_code\n , raw_lock\n , canceled\n , canceled_by\n , canceled_reason\n , job_kind\n , schedule_path\n , permissioned_as\n , flow_status\n , raw_flow\n , is_flow_step\n , is_skipped\n , language\n , email\n , visible_to_owner\n , mem_peak\n , tag\n , priority\n )\n VALUES ($1, $2, $3, $4, $5, COALESCE($6, now()), COALESCE($30::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE($6, now()))))*1000), $7, $8, $9,$10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29)\n ON CONFLICT (id) DO UPDATE SET success = $7, result = $11 RETURNING duration_ms AS \"duration_ms!\"",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "duration_ms!",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Varchar",
"Uuid",
"Uuid",
"Varchar",
"Timestamptz",
"Timestamptz",
"Bool",
"Int8",
"Varchar",
"Jsonb",
"Jsonb",
"Text",
"Text",
"Bool",
"Varchar",
"Text",
{
"Custom": {
"name": "job_kind",
"kind": {
"Enum": [
"script",
"preview",
"flow",
"dependencies",
"flowpreview",
"script_hub",
"identity",
"flowdependencies",
"http",
"graphql",
"postgresql",
"noop",
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"flowscript",
"flownode",
"appscript"
]
}
}
},
"Varchar",
"Varchar",
"Jsonb",
"Jsonb",
"Bool",
"Bool",
{
"Custom": {
"name": "script_lang",
"kind": {
"Enum": [
"python3",
"deno",
"go",
"bash",
"postgresql",
"nativets",
"bun",
"mysql",
"bigquery",
"snowflake",
"graphql",
"powershell",
"mssql",
"php",
"bunnative",
"rust",
"ansible",
"csharp",
"oracledb"
]
}
}
},
"Varchar",
"Bool",
"Int4",
"Varchar",
"Int2",
"Int8"
]
},
"nullable": [
true
]
},
"hash": "0df84fc35f2780ceb7c473b0165ebab93a4bc1bcab166aae68244ab1f3d4df9f"
}
backend/.sqlx/query-119469ebfe8572c78ed3ee5ab5b1a6a1cb1b0f31e357b5370f9bb7eab1e20a7b.json
Generated
+27
View File
@@ -0,0 +1,27 @@
{
"db_name": "PostgreSQL",
"query": "WITH uuid_table as (\n select gen_random_uuid() as uuid from generate_series(1, $6)\n )\n INSERT INTO job\n (id, workspace_id, raw_code, raw_lock, raw_flow, tag)\n (SELECT uuid, $1, $2, $3, $4, $5 FROM uuid_table)\n RETURNING id AS \"id!\"",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text",
"Jsonb",
"Varchar",
"Int4"
]
},
"nullable": [
true
]
},
"hash": "119469ebfe8572c78ed3ee5ab5b1a6a1cb1b0f31e357b5370f9bb7eab1e20a7b"
}
backend/.sqlx/query-14540eef4594d9282cee3df4f92a7ed2e67243e5c1522850045b2da42fa914bc.json
Generated
+91
View File
@@ -0,0 +1,91 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n queue.job_kind AS \"job_kind!: JobKind\",\n queue.script_hash AS \"script_hash: ScriptHash\",\n queue.raw_flow AS \"raw_flow: sqlx::types::Json<Box<RawValue>>\",\n completed_job.parent_job AS \"parent_job: Uuid\",\n completed_job.created_at AS \"created_at!: chrono::NaiveDateTime\",\n completed_job.created_by AS \"created_by!\",\n queue.script_path,\n queue.args AS \"args: sqlx::types::Json<Box<RawValue>>\"\n FROM queue\n JOIN completed_job ON completed_job.parent_job = queue.id\n WHERE completed_job.id = $1 AND completed_job.workspace_id = $2\n LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "job_kind!: JobKind",
"type_info": {
"Custom": {
"name": "job_kind",
"kind": {
"Enum": [
"script",
"preview",
"flow",
"dependencies",
"flowpreview",
"script_hub",
"identity",
"flowdependencies",
"http",
"graphql",
"postgresql",
"noop",
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"flowscript",
"flownode",
"appscript"
]
}
}
}
},
{
"ordinal": 1,
"name": "script_hash: ScriptHash",
"type_info": "Int8"
},
{
"ordinal": 2,
"name": "raw_flow: sqlx::types::Json<Box<RawValue>>",
"type_info": "Jsonb"
},
{
"ordinal": 3,
"name": "parent_job: Uuid",
"type_info": "Uuid"
},
{
"ordinal": 4,
"name": "created_at!: chrono::NaiveDateTime",
"type_info": "Timestamptz"
},
{
"ordinal": 5,
"name": "created_by!",
"type_info": "Varchar"
},
{
"ordinal": 6,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 7,
"name": "args: sqlx::types::Json<Box<RawValue>>",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
true,
true,
true,
true,
true,
true,
true,
true
]
},
"hash": "14540eef4594d9282cee3df4f92a7ed2e67243e5c1522850045b2da42fa914bc"
}
backend/.sqlx/query-15557c0acea71cee03f42516553fb4f5709e0e1a02a0187e88fa5d9e94ffb91a.json
Generated
+104
View File
@@ -0,0 +1,104 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO queue\n (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, \n script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner, root_job, tag, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, cache_ttl, priority, last_ping)\n VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, now()), $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, NULL) RETURNING id AS \"id!\"",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Varchar",
"Uuid",
"Bool",
"Uuid",
"Varchar",
"Varchar",
"Timestamptz",
"Int8",
"Varchar",
"Text",
"Text",
"Jsonb",
{
"Custom": {
"name": "job_kind",
"kind": {
"Enum": [
"script",
"preview",
"flow",
"dependencies",
"flowpreview",
"script_hub",
"identity",
"flowdependencies",
"http",
"graphql",
"postgresql",
"noop",
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"flowscript",
"flownode",
"appscript"
]
}
}
},
"Varchar",
"Jsonb",
"Jsonb",
"Bool",
{
"Custom": {
"name": "script_lang",
"kind": {
"Enum": [
"python3",
"deno",
"go",
"bash",
"postgresql",
"nativets",
"bun",
"mysql",
"bigquery",
"snowflake",
"graphql",
"powershell",
"mssql",
"php",
"bunnative",
"rust",
"ansible",
"csharp",
"oracledb"
]
}
}
},
"Bool",
"Text",
"Varchar",
"Bool",
"Uuid",
"Varchar",
"Int4",
"Int4",
"Int4",
"Varchar",
"Int4",
"Int2"
]
},
"nullable": [
true
]
},
"hash": "15557c0acea71cee03f42516553fb4f5709e0e1a02a0187e88fa5d9e94ffb91a"
}
backend/.sqlx/query-17f04341c5c52173a776b71672f9e4d932d2f072b6c08ce419e4b95a2fd83e96.json
Generated
+34
View File
@@ -0,0 +1,34 @@
{
"db_name": "PostgreSQL",
"query": "WITH zombie_jobs AS (\n UPDATE queue SET running = false, started_at = null\n WHERE last_ping < now() - ($1 || ' seconds')::interval\n AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow') AND same_worker = false\n RETURNING id, workspace_id, last_ping\n ),\n update_concurrency AS (\n UPDATE concurrency_counter cc\n SET job_uuids = job_uuids - zj.id::text\n FROM zombie_jobs zj\n INNER JOIN concurrency_key ck ON ck.job_id = zj.id\n WHERE cc.concurrency_id = ck.key\n )\n SELECT id AS \"id!\", workspace_id AS \"workspace_id!\", last_ping FROM zombie_jobs",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "workspace_id!",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "last_ping",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
true,
true,
true
]
},
"hash": "17f04341c5c52173a776b71672f9e4d932d2f072b6c08ce419e4b95a2fd83e96"
}
backend/.sqlx/query-2cef109784efc04999e4537e0d1d3fb3221e04f3a7c1abe91dd763f366d06618.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT success AS \"success!\" FROM completed_job WHERE id = ANY($1)",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "success!",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"UuidArray"
]
},
"nullable": [
true
]
},
"hash": "2cef109784efc04999e4537e0d1d3fb3221e04f3a7c1abe91dd763f366d06618"
}
backend/.sqlx/query-36b26b3a6458d8a0b4f770d52c1bb09370b905d610b9ceb3cfac11365586320d.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT running AS \"running!\" FROM queue WHERE id = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "running!",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
true
]
},
"hash": "36b26b3a6458d8a0b4f770d52c1bb09370b905d610b9ceb3cfac11365586320d"
}
backend/.sqlx/query-3a5edf3dd884b5a8862bb112f6520967ed4a218782192c6c6fc1498f45d753a6.json
Generated
+12
View File
@@ -0,0 +1,12 @@
{
"db_name": "PostgreSQL",
"query": "create index concurrently if not exists ix_job_created_at ON v2_job (created_at DESC)",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "3a5edf3dd884b5a8862bb112f6520967ed4a218782192c6c6fc1498f45d753a6"
}
backend/.sqlx/query-3bacf9cd9aa63f4bec5f983f4a0c3030216b5a4ed669f77962509d1c2c6cb780.json
Generated
+12
View File
@@ -0,0 +1,12 @@
{
"db_name": "PostgreSQL",
"query": "create index concurrently if not exists root_job_index_by_path_2 ON v2_job (workspace_id, runnable_path, created_at desc) WHERE parent_job IS NULL",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "3bacf9cd9aa63f4bec5f983f4a0c3030216b5a4ed669f77962509d1c2c6cb780"
}
backend/.sqlx/query-3bc1919515120116705d7c250a34f2b9bf7c4bcaedb87c28f974e46c9c42200c.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id AS \"id!\" FROM queue WHERE id = ANY($1) AND schedule_path IS NULL AND ($2::text[] IS NULL OR tag = ANY($2))",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"UuidArray",
"TextArray"
]
},
"nullable": [
true
]
},
"hash": "3bc1919515120116705d7c250a34f2b9bf7c4bcaedb87c28f974e46c9c42200c"
}
backend/.sqlx/query-3de870d80a3617e7b31209d922164bf957b8bac3b5c04626ec750b18f705cbfd.json
Generated
+12
View File
@@ -0,0 +1,12 @@
{
"db_name": "PostgreSQL",
"query": "CREATE INDEX CONCURRENTLY ix_v2_job_labels ON v2_job\n USING GIN (labels)\n WHERE labels IS NOT NULL",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "3de870d80a3617e7b31209d922164bf957b8bac3b5c04626ec750b18f705cbfd"
}
backend/.sqlx/query-402fd5bff6e8420c9b3477f05df625cae355fda673b9e85284e0fcd7d9232eb7.json
Generated
+73
View File
@@ -0,0 +1,73 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n script_path, script_hash AS \"script_hash: ScriptHash\",\n job_kind AS \"job_kind!: JobKind\",\n flow_status AS \"flow_status: Json<Box<RawValue>>\",\n raw_flow AS \"raw_flow: Json<Box<RawValue>>\"\n FROM completed_job WHERE id = $1 and workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "script_hash: ScriptHash",
"type_info": "Int8"
},
{
"ordinal": 2,
"name": "job_kind!: JobKind",
"type_info": {
"Custom": {
"name": "job_kind",
"kind": {
"Enum": [
"script",
"preview",
"flow",
"dependencies",
"flowpreview",
"script_hub",
"identity",
"flowdependencies",
"http",
"graphql",
"postgresql",
"noop",
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"flowscript",
"flownode",
"appscript"
]
}
}
}
},
{
"ordinal": 3,
"name": "flow_status: Json<Box<RawValue>>",
"type_info": "Jsonb"
},
{
"ordinal": 4,
"name": "raw_flow: Json<Box<RawValue>>",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
true,
true,
true,
true,
true
]
},
"hash": "402fd5bff6e8420c9b3477f05df625cae355fda673b9e85284e0fcd7d9232eb7"
}
backend/.sqlx/query-44a317f7647e2b515f90dc9c04f7ac75c2c87c7c3036acd96ba72fb2a21700db.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM completed_job WHERE created_at <= now() - ($1::bigint::text || ' s')::interval AND started_at + ((duration_ms/1000 + $1::bigint) || ' s')::interval <= now() RETURNING id AS \"id!\"",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Int8"
]
},
"nullable": [
true
]
},
"hash": "44a317f7647e2b515f90dc9c04f7ac75c2c87c7c3036acd96ba72fb2a21700db"
}
backend/.sqlx/query-4a1cb9f3ad7f2a692dadb9f75cdc99135af62ccaae5ca122d88356fe7da6eedc.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT script_entrypoint_override FROM v2_job WHERE id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "script_entrypoint_override",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
true
]
},
"hash": "4a1cb9f3ad7f2a692dadb9f75cdc99135af62ccaae5ca122d88356fe7da6eedc"
}
backend/.sqlx/query-4d3ef32120623584bf5c13d86ea6ad7b3aa41d9b581738d16fbfff4cc5b72a7a.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT canceled AS \"canceled!\" FROM queue WHERE id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "canceled!",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
true
]
},
"hash": "4d3ef32120623584bf5c13d86ea6ad7b3aa41d9b581738d16fbfff4cc5b72a7a"
}
backend/.sqlx/query-5401c521b5e63b7d9e7bc51c19d116599f6bcedbe70f3bf346b482fe79501958.json
Generated
+12
View File
@@ -0,0 +1,12 @@
{
"db_name": "PostgreSQL",
"query": "create index concurrently if not exists ix_job_workspace_id_created_at_new_8 ON v2_job (workspace_id, created_at DESC) where kind in ('deploymentcallback') AND parent_job IS NULL",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "5401c521b5e63b7d9e7bc51c19d116599f6bcedbe70f3bf346b482fe79501958"
}
backend/.sqlx/query-5bce731932a35dbecc38c7b9665ef1117a15acf7d0d41b93de165e788b55d93f.json
Generated
+42
View File
@@ -0,0 +1,42 @@
{
"db_name": "PostgreSQL",
"query": "SELECT created_by AS \"created_by!\", CONCAT(coalesce(completed_job.logs, ''), coalesce(job_logs.logs, '')) as logs, job_logs.log_offset, job_logs.log_file_index\n FROM completed_job \n LEFT JOIN job_logs ON job_logs.job_id = completed_job.id \n WHERE completed_job.id = $1 AND completed_job.workspace_id = $2 AND ($3::text[] IS NULL OR completed_job.tag = ANY($3))",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "created_by!",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "logs",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "log_offset",
"type_info": "Int4"
},
{
"ordinal": 3,
"name": "log_file_index",
"type_info": "TextArray"
}
],
"parameters": {
"Left": [
"Uuid",
"Text",
"TextArray"
]
},
"nullable": [
true,
null,
false,
true
]
},
"hash": "5bce731932a35dbecc38c7b9665ef1117a15acf7d0d41b93de165e788b55d93f"
}
backend/.sqlx/query-6b0115e40d4361b3ca72dbd071b0a8c0319c5ae0b92f289ec1e74d2478c9e740.json
Generated
+42
View File
@@ -0,0 +1,42 @@
{
"db_name": "PostgreSQL",
"query": "SELECT created_by AS \"created_by!\", CONCAT(coalesce(queue.logs, ''), coalesce(job_logs.logs, '')) as logs, coalesce(job_logs.log_offset, 0) as log_offset, job_logs.log_file_index\n FROM queue \n LEFT JOIN job_logs ON job_logs.job_id = queue.id \n WHERE queue.id = $1 AND queue.workspace_id = $2 AND ($3::text[] IS NULL OR queue.tag = ANY($3))",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "created_by!",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "logs",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "log_offset",
"type_info": "Int4"
},
{
"ordinal": 3,
"name": "log_file_index",
"type_info": "TextArray"
}
],
"parameters": {
"Left": [
"Uuid",
"Text",
"TextArray"
]
},
"nullable": [
true,
null,
null,
true
]
},
"hash": "6b0115e40d4361b3ca72dbd071b0a8c0319c5ae0b92f289ec1e74d2478c9e740"
}
backend/.sqlx/query-74a2a90d12ca0179c8a80f9bf574066db4e8735c0f717d91391a28bf832c0e71.json
Generated
+25
View File
@@ -0,0 +1,25 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE queue SET canceled = true, canceled_by = $1, canceled_reason = $2, scheduled_for = now(), suspend = 0 WHERE id = $3 AND workspace_id = $4 AND (canceled = false OR canceled_reason != $2) RETURNING id AS \"id!\"",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Varchar",
"Text",
"Uuid",
"Text"
]
},
"nullable": [
true
]
},
"hash": "74a2a90d12ca0179c8a80f9bf574066db4e8735c0f717d91391a28bf832c0e71"
}
backend/.sqlx/query-7dc7bc4e22942792938d273655962a95486f6da82cdc08f79dd6cef508256474.json
Generated
+25
View File
@@ -0,0 +1,25 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO completed_job AS cj\n ( workspace_id\n , id\n , parent_job\n , created_by\n , created_at\n , started_at\n , duration_ms\n , success\n , script_hash\n , script_path\n , args\n , result\n , raw_code\n , raw_lock\n , canceled\n , canceled_by\n , canceled_reason\n , job_kind\n , schedule_path\n , permissioned_as\n , flow_status\n , raw_flow\n , is_flow_step\n , is_skipped\n , language\n , email\n , visible_to_owner\n , mem_peak\n , tag\n , priority\n )\n SELECT workspace_id\n , id\n , parent_job\n , created_by\n , created_at\n , now()\n , 0\n , false\n , script_hash\n , script_path\n , args\n , $4\n , raw_code\n , raw_lock\n , true\n , $1\n , canceled_reason\n , job_kind\n , schedule_path\n , permissioned_as\n , flow_status\n , raw_flow\n , is_flow_step\n , false\n , language\n , email\n , visible_to_owner\n , mem_peak\n , tag\n , priority FROM queue \n WHERE id = any($2) AND running = false AND parent_job IS NULL AND workspace_id = $3 AND schedule_path IS NULL FOR UPDATE SKIP LOCKED\n ON CONFLICT (id) DO NOTHING RETURNING id AS \"id!\"",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Varchar",
"UuidArray",
"Text",
"Jsonb"
]
},
"nullable": [
true
]
},
"hash": "7dc7bc4e22942792938d273655962a95486f6da82cdc08f79dd6cef508256474"
}
backend/.sqlx/query-8be277b89102a26dda506202a3ef7eb05342cfb3aa9b4f5d80c70fbc50d437ba.json
Generated
+12
View File
@@ -0,0 +1,12 @@
{
"db_name": "PostgreSQL",
"query": "create index concurrently if not exists ix_job_workspace_id_created_at_new_6 ON v2_job (workspace_id, created_at DESC) where kind in ('script', 'flow') AND parent_job IS NULL",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "8be277b89102a26dda506202a3ef7eb05342cfb3aa9b4f5d80c70fbc50d437ba"
}
backend/.sqlx/query-931703a98d2ee5fb58d3380896baaee032e731db1e6bd49d991a54f49ab8fa46.json
Generated
+12
View File
@@ -0,0 +1,12 @@
{
"db_name": "PostgreSQL",
"query": "create index concurrently if not exists ix_completed_job_workspace_id_started_at_new_2 ON v2_job_completed (workspace_id, started_at DESC)",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "931703a98d2ee5fb58d3380896baaee032e731db1e6bd49d991a54f49ab8fa46"
}
backend/.sqlx/query-96a9357888af26e5ec1e314bb565af05de561a8f9899e4ddca958982fdb67803.json
Generated
+24
View File
@@ -0,0 +1,24 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id AS \"id!\" FROM queue WHERE schedule_path = $1 AND workspace_id = $2 AND id != $3 AND running = true",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Uuid"
]
},
"nullable": [
true
]
},
"hash": "96a9357888af26e5ec1e314bb565af05de561a8f9899e4ddca958982fdb67803"
}
backend/.sqlx/query-9d518842a9ad90ff9c28dc39690deb0ee6b62cf1d8ae1a02b28c23255d377b3d.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT labels FROM v2_job WHERE id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "labels",
"type_info": "TextArray"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
true
]
},
"hash": "9d518842a9ad90ff9c28dc39690deb0ee6b62cf1d8ae1a02b28c23255d377b3d"
}
backend/.sqlx/query-a33e282d02c53e5d6142dc7e6882a6d8f3d068c55cddf209eb6b6431ca26c910.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id AS \"id!\" FROM queue WHERE workspace_id = $1 and root_job = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Text",
"Uuid"
]
},
"nullable": [
true
]
},
"hash": "a33e282d02c53e5d6142dc7e6882a6d8f3d068c55cddf209eb6b6431ca26c910"
}
backend/.sqlx/query-a56eef5f5ecbe1a8d309ff65d9a8c456a3c165f7f2a107cf7fa6a4cdd30d55c0.json
Generated
+28
View File
@@ -0,0 +1,28 @@
{
"db_name": "PostgreSQL",
"query": "SELECT preprocessed, script_entrypoint_override FROM v2_job WHERE id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "preprocessed",
"type_info": "Bool"
},
{
"ordinal": 1,
"name": "script_entrypoint_override",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
true,
true
]
},
"hash": "a56eef5f5ecbe1a8d309ff65d9a8c456a3c165f7f2a107cf7fa6a4cdd30d55c0"
}
backend/.sqlx/query-ae8dfecd46425d5f86003eea9a578e9831fc0e700cc76ab9627afe9040a4efe0.json
Generated
+12
View File
@@ -0,0 +1,12 @@
{
"db_name": "PostgreSQL",
"query": "create index concurrently if not exists ix_job_workspace_id_created_at_new_7 ON v2_job (workspace_id, created_at DESC) where kind in ('script', 'flow') AND parent_job IS NULL",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "ae8dfecd46425d5f86003eea9a578e9831fc0e700cc76ab9627afe9040a4efe0"
}
backend/.sqlx/query-b06915e02398511033717ea13b710c86a24fe666884cfd49996dee961751ce51.json
Generated
+89
View File
@@ -0,0 +1,89 @@
{
"db_name": "PostgreSQL",
"query": "WITH uuid_table as (\n select unnest($11::uuid[]) as uuid\n )\n INSERT INTO queue \n (id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id, concurrent_limit, concurrency_time_window_s, timeout, flow_status)\n (SELECT uuid, $1, $2, $3, $4, ('{ \"uuid\": \"' || uuid || '\" }')::jsonb, $5, $6, $7, $8, $9, $10, $12, $13, $14, $15 FROM uuid_table) \n RETURNING id AS \"id!\"",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Int8",
"Varchar",
{
"Custom": {
"name": "job_kind",
"kind": {
"Enum": [
"script",
"preview",
"flow",
"dependencies",
"flowpreview",
"script_hub",
"identity",
"flowdependencies",
"http",
"graphql",
"postgresql",
"noop",
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"flowscript",
"flownode",
"appscript"
]
}
}
},
{
"Custom": {
"name": "script_lang",
"kind": {
"Enum": [
"python3",
"deno",
"go",
"bash",
"postgresql",
"nativets",
"bun",
"mysql",
"bigquery",
"snowflake",
"graphql",
"powershell",
"mssql",
"php",
"bunnative",
"rust",
"ansible",
"csharp",
"oracledb"
]
}
}
},
"Varchar",
"Varchar",
"Varchar",
"Varchar",
"Timestamptz",
"Varchar",
"UuidArray",
"Int4",
"Int4",
"Int4",
"Jsonb"
]
},
"nullable": [
true
]
},
"hash": "b06915e02398511033717ea13b710c86a24fe666884cfd49996dee961751ce51"
}
backend/.sqlx/query-bb93ba18709648b47cfbd04d91afd3b38546b1a718d0abff6b2795d7c2a29c97.json
Generated
+40
View File
@@ -0,0 +1,40 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT id AS \"id!\", flow_status, suspend AS \"suspend!\", script_path\n FROM queue\n WHERE id = $1\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "flow_status",
"type_info": "Jsonb"
},
{
"ordinal": 2,
"name": "suspend!",
"type_info": "Int4"
},
{
"ordinal": 3,
"name": "script_path",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
true,
true,
true,
true
]
},
"hash": "bb93ba18709648b47cfbd04d91afd3b38546b1a718d0abff6b2795d7c2a29c97"
}
backend/.sqlx/query-bd5a0c06e2f2361c9fc670eb0b975b58d65ca93d68b29124d04bd526239b9df2.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE v2_job SET labels = $2 WHERE id = $1 AND $2::TEXT[] IS NOT NULL",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"TextArray"
]
},
"nullable": []
},
"hash": "bd5a0c06e2f2361c9fc670eb0b975b58d65ca93d68b29124d04bd526239b9df2"
}
backend/.sqlx/query-bf91cb319e5b83c2235292a9e3ce8aa1c097c94b01aad0d9f7bce76a2a272bcc.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT success AS \"success!\"\n FROM completed_job WHERE id = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "success!",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
true
]
},
"hash": "bf91cb319e5b83c2235292a9e3ce8aa1c097c94b01aad0d9f7bce76a2a272bcc"
}
backend/.sqlx/query-c0b96d2f421afc43e256a8475825623bcb3dd4cbc37d570fc4273127bbf77c24.json
Generated
+67
View File
@@ -0,0 +1,67 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n job_kind AS \"job_kind!: JobKind\",\n script_hash AS \"script_hash: ScriptHash\",\n flow_status AS \"flow_status!: Json<Box<RawValue>>\",\n raw_flow AS \"raw_flow: Json<Box<RawValue>>\"\n FROM queue WHERE id = $1 AND workspace_id = $2 LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "job_kind!: JobKind",
"type_info": {
"Custom": {
"name": "job_kind",
"kind": {
"Enum": [
"script",
"preview",
"flow",
"dependencies",
"flowpreview",
"script_hub",
"identity",
"flowdependencies",
"http",
"graphql",
"postgresql",
"noop",
"appdependencies",
"deploymentcallback",
"singlescriptflow",
"flowscript",
"flownode",
"appscript"
]
}
}
}
},
{
"ordinal": 1,
"name": "script_hash: ScriptHash",
"type_info": "Int8"
},
{
"ordinal": 2,
"name": "flow_status!: Json<Box<RawValue>>",
"type_info": "Jsonb"
},
{
"ordinal": 3,
"name": "raw_flow: Json<Box<RawValue>>",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
true,
true,
true,
true
]
},
"hash": "c0b96d2f421afc43e256a8475825623bcb3dd4cbc37d570fc4273127bbf77c24"
}
backend/.sqlx/query-c4bdbf1c6bc7d93db4cf2633105b088b781354cb7c02628d9f8ff7f9ea0e7ed9.json
Generated
+12
View File
@@ -0,0 +1,12 @@
{
"db_name": "PostgreSQL",
"query": "CREATE INDEX CONCURRENTLY labeled_jobs_on_jobs ON v2_job_completed USING GIN ((result -> 'wm_labels')) WHERE result ? 'wm_labels'",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "c4bdbf1c6bc7d93db4cf2633105b088b781354cb7c02628d9f8ff7f9ea0e7ed9"
}
backend/.sqlx/query-c7be5fa2eaf66147c1213046e615f5e9fd168ef1e3aba8af64b15341055d6007.json
Generated
+12
View File
@@ -0,0 +1,12 @@
{
"db_name": "PostgreSQL",
"query": "create index concurrently if not exists ix_job_workspace_id_created_at_new_5 ON v2_job (workspace_id, created_at DESC) where kind in ('preview', 'flowpreview') AND parent_job IS NULL",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "c7be5fa2eaf66147c1213046e615f5e9fd168ef1e3aba8af64b15341055d6007"
}
backend/.sqlx/query-cd5f02cf10cbf92dd1df53a54f2110efa11a7731ad0f0e5509f55efabdf535cd.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT preprocessed FROM v2_job WHERE id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "preprocessed",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
true
]
},
"hash": "cd5f02cf10cbf92dd1df53a54f2110efa11a7731ad0f0e5509f55efabdf535cd"
}
backend/.sqlx/query-d7f1e2920aec0f4eab9238d01370465945acdfa779f16b99cdc1a6b7ef84943e.json
Generated
+40
View File
@@ -0,0 +1,40 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT id AS \"id!\", flow_status, suspend AS \"suspend!\", script_path\n FROM queue\n WHERE id = ( SELECT parent_job FROM queue WHERE id = $1 UNION ALL SELECT parent_job FROM completed_job WHERE id = $1)\n FOR UPDATE\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "flow_status",
"type_info": "Jsonb"
},
{
"ordinal": 2,
"name": "suspend!",
"type_info": "Int4"
},
{
"ordinal": 3,
"name": "script_path",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
true,
true,
true,
true
]
},
"hash": "d7f1e2920aec0f4eab9238d01370465945acdfa779f16b99cdc1a6b7ef84943e"
}
backend/.sqlx/query-f64ae18811e211dbf0cb98b43d3b018b0dcc0abc7e4a1f0b45885cfe18efd9b2.json
Generated
+12
View File
@@ -0,0 +1,12 @@
{
"db_name": "PostgreSQL",
"query": "create index concurrently if not exists ix_job_workspace_id_created_at_new_3 ON v2_job (workspace_id, created_at DESC)",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "f64ae18811e211dbf0cb98b43d3b018b0dcc0abc7e4a1f0b45885cfe18efd9b2"
}
backend/.sqlx/query-f8f25948ae14fcb71c666cdc5e51d888e1f22fb2300a78bbeafebf64e82658db.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id AS \"id!\" FROM queue WHERE parent_job = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
true
]
},
"hash": "f8f25948ae14fcb71c666cdc5e51d888e1f22fb2300a78bbeafebf64e82658db"
}
backend/migrations/20250201124344_v2_job_runtime.down.sql
+4
View File
@@ -0,0 +1,4 @@
-- Add down migration script here
-- Lock `queue` in access exclusive to prevent deadlocks when dropping the foreign key to `queue`.
LOCK TABLE queue IN ACCESS EXCLUSIVE MODE;
DROP TABLE v2_job_runtime CASCADE;
backend/migrations/20250201124344_v2_job_runtime.up.sql
+14
View File
@@ -0,0 +1,14 @@
-- Add up migration script here
CREATE TABLE IF NOT EXISTS v2_job_runtime (
id UUID REFERENCES queue (id) ON DELETE CASCADE PRIMARY KEY NOT NULL,
-- Metrics fields:
ping TIMESTAMP WITH TIME ZONE DEFAULT now(),
memory_peak INTEGER
);
CREATE POLICY admin_policy ON v2_job_runtime
AS PERMISSIVE
FOR ALL
TO windmill_admin;
GRANT ALL ON v2_job_runtime TO windmill_user, windmill_admin;
backend/migrations/20250201124345_v2_job_status.down.sql
+4
View File
@@ -0,0 +1,4 @@
-- Add down migration script here
-- Lock `queue` in access exclusive to prevent deadlocks when dropping the foreign key to `queue`.
LOCK TABLE queue IN ACCESS EXCLUSIVE MODE;
DROP TABLE v2_job_status CASCADE;
backend/migrations/20250201124345_v2_job_status.up.sql
+16
View File
@@ -0,0 +1,16 @@
-- Add up migration script here
CREATE TABLE IF NOT EXISTS v2_job_status (
id UUID REFERENCES queue (id) ON DELETE CASCADE PRIMARY KEY NOT NULL,
-- Flow status fields:
flow_status JSONB,
flow_leaf_jobs JSONB,
-- Workflow as code fields:
workflow_as_code_status JSONB
);
CREATE POLICY admin_policy ON v2_job_status
AS PERMISSIVE
FOR ALL
TO windmill_admin;
GRANT ALL ON v2_job_status TO windmill_user, windmill_admin;
backend/migrations/20250201124409_v2_job_queue_compatiblity_view.down.sql
+3
View File
@@ -0,0 +1,3 @@
-- Add down migration script here
DROP VIEW queue;
ALTER TABLE v2_job_queue RENAME TO queue;
backend/migrations/20250201124409_v2_job_queue_compatiblity_view.up.sql
+48
View File
@@ -0,0 +1,48 @@
-- Add down migration script here
ALTER TABLE queue RENAME TO v2_job_queue;
CREATE OR REPLACE VIEW queue AS (
SELECT
id,
workspace_id,
parent_job AS parent_job,
created_by AS created_by,
created_at,
started_at,
scheduled_for,
running,
script_hash AS script_hash,
script_path AS script_path,
args AS args,
logs AS logs,
raw_code AS raw_code,
canceled AS canceled,
canceled_by,
canceled_reason,
last_ping AS last_ping,
job_kind AS job_kind,
env_id AS env_id,
schedule_path AS schedule_path,
permissioned_as AS permissioned_as,
flow_status AS flow_status,
raw_flow AS raw_flow,
is_flow_step AS is_flow_step,
language AS language,
suspend,
suspend_until,
same_worker AS same_worker,
raw_lock AS raw_lock,
pre_run_error AS pre_run_error,
email AS email,
visible_to_owner AS visible_to_owner,
mem_peak AS mem_peak,
root_job AS root_job,
leaf_jobs AS leaf_jobs,
tag,
concurrent_limit AS concurrent_limit,
concurrency_time_window_s AS concurrency_time_window_s,
timeout AS timeout,
flow_step_id AS flow_step_id,
cache_ttl AS cache_ttl,
priority
FROM v2_job_queue
);
backend/migrations/20250201124410_v2_job_queue.down.sql
+34
View File
@@ -0,0 +1,34 @@
-- Add down migration script here
ALTER TABLE v2_job_queue
DROP COLUMN worker CASCADE,
DROP COLUMN extras CASCADE;
ALTER TABLE v2_job_queue RENAME COLUMN __parent_job TO parent_job;
ALTER TABLE v2_job_queue RENAME COLUMN __created_by TO created_by;
ALTER TABLE v2_job_queue RENAME COLUMN __script_hash TO script_hash;
ALTER TABLE v2_job_queue RENAME COLUMN __script_path TO script_path;
ALTER TABLE v2_job_queue RENAME COLUMN __args TO args;
ALTER TABLE v2_job_queue RENAME COLUMN __logs TO logs;
ALTER TABLE v2_job_queue RENAME COLUMN __raw_code TO raw_code;
ALTER TABLE v2_job_queue RENAME COLUMN __canceled TO canceled;
ALTER TABLE v2_job_queue RENAME COLUMN __last_ping TO last_ping;
ALTER TABLE v2_job_queue RENAME COLUMN __job_kind TO job_kind;
ALTER TABLE v2_job_queue RENAME COLUMN __env_id TO env_id;
ALTER TABLE v2_job_queue RENAME COLUMN __schedule_path TO schedule_path;
ALTER TABLE v2_job_queue RENAME COLUMN __permissioned_as TO permissioned_as;
ALTER TABLE v2_job_queue RENAME COLUMN __flow_status TO flow_status;
ALTER TABLE v2_job_queue RENAME COLUMN __raw_flow TO raw_flow;
ALTER TABLE v2_job_queue RENAME COLUMN __is_flow_step TO is_flow_step;
ALTER TABLE v2_job_queue RENAME COLUMN __language TO language;
ALTER TABLE v2_job_queue RENAME COLUMN __same_worker TO same_worker;
ALTER TABLE v2_job_queue RENAME COLUMN __raw_lock TO raw_lock;
ALTER TABLE v2_job_queue RENAME COLUMN __pre_run_error TO pre_run_error;
ALTER TABLE v2_job_queue RENAME COLUMN __email TO email;
ALTER TABLE v2_job_queue RENAME COLUMN __visible_to_owner TO visible_to_owner;
ALTER TABLE v2_job_queue RENAME COLUMN __mem_peak TO mem_peak;
ALTER TABLE v2_job_queue RENAME COLUMN __root_job TO root_job;
ALTER TABLE v2_job_queue RENAME COLUMN __leaf_jobs TO leaf_jobs;
ALTER TABLE v2_job_queue RENAME COLUMN __concurrent_limit TO concurrent_limit;
ALTER TABLE v2_job_queue RENAME COLUMN __concurrency_time_window_s TO concurrency_time_window_s;
ALTER TABLE v2_job_queue RENAME COLUMN __timeout TO timeout;
ALTER TABLE v2_job_queue RENAME COLUMN __flow_step_id TO flow_step_id;
ALTER TABLE v2_job_queue RENAME COLUMN __cache_ttl TO cache_ttl;
backend/migrations/20250201124410_v2_job_queue.up.sql
+34
View File
@@ -0,0 +1,34 @@
-- Add up migration script here
ALTER TABLE v2_job_queue
ADD COLUMN IF NOT EXISTS worker VARCHAR(255),
ADD COLUMN IF NOT EXISTS extras JSONB;
ALTER TABLE v2_job_queue RENAME COLUMN parent_job TO __parent_job;
ALTER TABLE v2_job_queue RENAME COLUMN created_by TO __created_by;
ALTER TABLE v2_job_queue RENAME COLUMN script_hash TO __script_hash;
ALTER TABLE v2_job_queue RENAME COLUMN script_path TO __script_path;
ALTER TABLE v2_job_queue RENAME COLUMN args TO __args;
ALTER TABLE v2_job_queue RENAME COLUMN logs TO __logs;
ALTER TABLE v2_job_queue RENAME COLUMN raw_code TO __raw_code;
ALTER TABLE v2_job_queue RENAME COLUMN canceled TO __canceled;
ALTER TABLE v2_job_queue RENAME COLUMN last_ping TO __last_ping;
ALTER TABLE v2_job_queue RENAME COLUMN job_kind TO __job_kind;
ALTER TABLE v2_job_queue RENAME COLUMN env_id TO __env_id;
ALTER TABLE v2_job_queue RENAME COLUMN schedule_path TO __schedule_path;
ALTER TABLE v2_job_queue RENAME COLUMN permissioned_as TO __permissioned_as;
ALTER TABLE v2_job_queue RENAME COLUMN flow_status TO __flow_status;
ALTER TABLE v2_job_queue RENAME COLUMN raw_flow TO __raw_flow;
ALTER TABLE v2_job_queue RENAME COLUMN is_flow_step TO __is_flow_step;
ALTER TABLE v2_job_queue RENAME COLUMN language TO __language;
ALTER TABLE v2_job_queue RENAME COLUMN same_worker TO __same_worker;
ALTER TABLE v2_job_queue RENAME COLUMN raw_lock TO __raw_lock;
ALTER TABLE v2_job_queue RENAME COLUMN pre_run_error TO __pre_run_error;
ALTER TABLE v2_job_queue RENAME COLUMN email TO __email;
ALTER TABLE v2_job_queue RENAME COLUMN visible_to_owner TO __visible_to_owner;
ALTER TABLE v2_job_queue RENAME COLUMN mem_peak TO __mem_peak;
ALTER TABLE v2_job_queue RENAME COLUMN root_job TO __root_job;
ALTER TABLE v2_job_queue RENAME COLUMN leaf_jobs TO __leaf_jobs;
ALTER TABLE v2_job_queue RENAME COLUMN concurrent_limit TO __concurrent_limit;
ALTER TABLE v2_job_queue RENAME COLUMN concurrency_time_window_s TO __concurrency_time_window_s;
ALTER TABLE v2_job_queue RENAME COLUMN timeout TO __timeout;
ALTER TABLE v2_job_queue RENAME COLUMN flow_step_id TO __flow_step_id;
ALTER TABLE v2_job_queue RENAME COLUMN cache_ttl TO __cache_ttl;
backend/migrations/20250201124422_v2_job_completed_compatibility_view.down.sql
+3
View File
@@ -0,0 +1,3 @@
-- Add down migration script here
DROP VIEW IF EXISTS completed_job;
ALTER TABLE IF EXISTS v2_job_completed RENAME TO completed_job;
backend/migrations/20250201124422_v2_job_completed_compatibility_view.up.sql
+39
View File
@@ -0,0 +1,39 @@
-- Add down migration script here
ALTER TABLE completed_job RENAME TO v2_job_completed;
CREATE OR REPLACE VIEW completed_job AS (
SELECT
id,
workspace_id,
parent_job AS parent_job,
created_by AS created_by,
created_at AS created_at,
duration_ms,
success AS success,
script_hash AS script_hash,
script_path AS script_path,
args AS args,
result,
logs AS logs,
deleted,
raw_code AS raw_code,
canceled AS canceled,
canceled_by,
canceled_reason,
job_kind AS job_kind,
env_id AS env_id,
schedule_path AS schedule_path,
permissioned_as AS permissioned_as,
flow_status,
raw_flow AS raw_flow,
is_flow_step AS is_flow_step,
language AS language,
started_at,
is_skipped AS is_skipped,
raw_lock AS raw_lock,
email AS email,
visible_to_owner AS visible_to_owner,
mem_peak AS mem_peak,
tag AS tag,
priority AS priority
FROM v2_job_completed
);
backend/migrations/20250201124423_v2_job_completed.down.sql
+34
View File
@@ -0,0 +1,34 @@
-- Add down migration script here
ALTER TABLE v2_job_completed
DROP COLUMN status CASCADE,
DROP COLUMN completed_at CASCADE,
DROP COLUMN worker CASCADE,
DROP COLUMN workflow_as_code_status CASCADE,
DROP COLUMN result_columns CASCADE,
DROP COLUMN retries CASCADE,
DROP COLUMN extras CASCADE;
ALTER TABLE v2_job_completed RENAME COLUMN memory_peak TO mem_peak;
ALTER TABLE v2_job_completed RENAME COLUMN __parent_job TO parent_job;
ALTER TABLE v2_job_completed RENAME COLUMN __created_by TO created_by;
ALTER TABLE v2_job_completed RENAME COLUMN __created_at TO created_at;
ALTER TABLE v2_job_completed RENAME COLUMN __success TO success;
ALTER TABLE v2_job_completed RENAME COLUMN __script_hash TO script_hash;
ALTER TABLE v2_job_completed RENAME COLUMN __script_path TO script_path;
ALTER TABLE v2_job_completed RENAME COLUMN __args TO args;
ALTER TABLE v2_job_completed RENAME COLUMN __logs TO logs;
ALTER TABLE v2_job_completed RENAME COLUMN __raw_code TO raw_code;
ALTER TABLE v2_job_completed RENAME COLUMN __canceled TO canceled;
ALTER TABLE v2_job_completed RENAME COLUMN __job_kind TO job_kind;
ALTER TABLE v2_job_completed RENAME COLUMN __env_id TO env_id;
ALTER TABLE v2_job_completed RENAME COLUMN __schedule_path TO schedule_path;
ALTER TABLE v2_job_completed RENAME COLUMN __permissioned_as TO permissioned_as;
ALTER TABLE v2_job_completed RENAME COLUMN __raw_flow TO raw_flow;
ALTER TABLE v2_job_completed RENAME COLUMN __is_flow_step TO is_flow_step;
ALTER TABLE v2_job_completed RENAME COLUMN __language TO language;
ALTER TABLE v2_job_completed RENAME COLUMN __is_skipped TO is_skipped;
ALTER TABLE v2_job_completed RENAME COLUMN __raw_lock TO raw_lock;
ALTER TABLE v2_job_completed RENAME COLUMN __email TO email;
ALTER TABLE v2_job_completed RENAME COLUMN __visible_to_owner TO visible_to_owner;
ALTER TABLE v2_job_completed RENAME COLUMN __tag TO tag;
ALTER TABLE v2_job_completed RENAME COLUMN __priority TO priority;
DROP TYPE IF EXISTS job_status CASCADE;
backend/migrations/20250201124423_v2_job_completed.up.sql
+34
View File
@@ -0,0 +1,34 @@
-- Add up migration script here
CREATE TYPE job_status AS ENUM ('success', 'failure', 'canceled', 'skipped');
ALTER TABLE v2_job_completed
ADD COLUMN IF NOT EXISTS status job_status,
ADD COLUMN IF NOT EXISTS completed_at TIMESTAMP WITH TIME ZONE,
ADD COLUMN IF NOT EXISTS worker VARCHAR(255),
ADD COLUMN IF NOT EXISTS workflow_as_code_status JSONB,
ADD COLUMN IF NOT EXISTS result_columns TEXT[],
ADD COLUMN IF NOT EXISTS retries UUID[],
ADD COLUMN IF NOT EXISTS extras JSONB;
ALTER TABLE v2_job_completed RENAME COLUMN mem_peak TO memory_peak;
ALTER TABLE v2_job_completed RENAME COLUMN parent_job TO __parent_job;
ALTER TABLE v2_job_completed RENAME COLUMN created_by TO __created_by;
ALTER TABLE v2_job_completed RENAME COLUMN created_at TO __created_at;
ALTER TABLE v2_job_completed RENAME COLUMN success TO __success;
ALTER TABLE v2_job_completed RENAME COLUMN script_hash TO __script_hash;
ALTER TABLE v2_job_completed RENAME COLUMN script_path TO __script_path;
ALTER TABLE v2_job_completed RENAME COLUMN args TO __args;
ALTER TABLE v2_job_completed RENAME COLUMN logs TO __logs;
ALTER TABLE v2_job_completed RENAME COLUMN raw_code TO __raw_code;
ALTER TABLE v2_job_completed RENAME COLUMN canceled TO __canceled;
ALTER TABLE v2_job_completed RENAME COLUMN job_kind TO __job_kind;
ALTER TABLE v2_job_completed RENAME COLUMN env_id TO __env_id;
ALTER TABLE v2_job_completed RENAME COLUMN schedule_path TO __schedule_path;
ALTER TABLE v2_job_completed RENAME COLUMN permissioned_as TO __permissioned_as;
ALTER TABLE v2_job_completed RENAME COLUMN raw_flow TO __raw_flow;
ALTER TABLE v2_job_completed RENAME COLUMN is_flow_step TO __is_flow_step;
ALTER TABLE v2_job_completed RENAME COLUMN language TO __language;
ALTER TABLE v2_job_completed RENAME COLUMN is_skipped TO __is_skipped;
ALTER TABLE v2_job_completed RENAME COLUMN raw_lock TO __raw_lock;
ALTER TABLE v2_job_completed RENAME COLUMN email TO __email;
ALTER TABLE v2_job_completed RENAME COLUMN visible_to_owner TO __visible_to_owner;
ALTER TABLE v2_job_completed RENAME COLUMN tag TO __tag;
ALTER TABLE v2_job_completed RENAME COLUMN priority TO __priority;
backend/migrations/20250201124430_v2_job_compatibility_view.down.sql
+3
View File
@@ -0,0 +1,3 @@
-- Add down migration script here
DROP VIEW IF EXISTS job;
ALTER TABLE IF EXISTS v2_job RENAME TO job;
backend/migrations/20250201124430_v2_job_compatibility_view.up.sql
+6
View File
@@ -0,0 +1,6 @@
-- Add down migration script here
ALTER TABLE job RENAME TO v2_job;
CREATE OR REPLACE VIEW job AS (
SELECT id, raw_code, raw_lock, raw_flow, tag, workspace_id
FROM v2_job
);
backend/migrations/20250201124431_v2_job.down.sql
+36
View File
@@ -0,0 +1,36 @@
-- Add down migration script here
DROP POLICY see_folder_extra_perms_user ON v2_job;
DROP POLICY see_own_path ON v2_job;
DROP POLICY see_member_path ON v2_job;
DROP POLICY see_own ON v2_job;
DROP POLICY see_member ON v2_job;
DROP POLICY admin_policy ON v2_job;
ALTER TABLE v2_job
DROP COLUMN created_at CASCADE,
DROP COLUMN created_by CASCADE,
DROP COLUMN permissioned_as CASCADE,
DROP COLUMN permissioned_as_email CASCADE,
DROP COLUMN kind CASCADE,
DROP COLUMN runnable_id CASCADE,
DROP COLUMN runnable_path CASCADE,
DROP COLUMN parent_job CASCADE,
DROP COLUMN root_job CASCADE,
DROP COLUMN script_lang CASCADE,
DROP COLUMN script_entrypoint_override CASCADE,
DROP COLUMN flow_step CASCADE,
DROP COLUMN flow_step_id CASCADE,
DROP COLUMN flow_innermost_root_job CASCADE,
DROP COLUMN trigger CASCADE,
DROP COLUMN trigger_kind CASCADE,
DROP COLUMN same_worker CASCADE,
DROP COLUMN visible_to_owner CASCADE,
DROP COLUMN concurrent_limit CASCADE,
DROP COLUMN concurrency_time_window_s CASCADE,
DROP COLUMN cache_ttl CASCADE,
DROP COLUMN timeout CASCADE,
DROP COLUMN priority CASCADE,
DROP COLUMN preprocessed CASCADE,
DROP COLUMN args CASCADE,
DROP COLUMN labels CASCADE,
DROP COLUMN pre_run_error CASCADE;
DROP TYPE job_trigger_kind;
backend/migrations/20250201124431_v2_job.up.sql
+78
View File
@@ -0,0 +1,78 @@
-- Add up migration script here
CREATE TYPE job_trigger_kind AS ENUM (
'webhook', 'http', 'websocket', 'kafka', 'email', 'nats', 'schedule', 'app', 'ui', 'postgres'
);
ALTER TABLE v2_job
ADD COLUMN IF NOT EXISTS created_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,
ADD COLUMN IF NOT EXISTS created_by VARCHAR(255) DEFAULT 'missing' NOT NULL,
ADD COLUMN IF NOT EXISTS permissioned_as VARCHAR(55) DEFAULT 'g/all' NOT NULL,
ADD COLUMN IF NOT EXISTS permissioned_as_email VARCHAR(255) DEFAULT 'missing@email.xyz' NOT NULL,
ADD COLUMN IF NOT EXISTS kind job_kind DEFAULT 'script'::job_kind NOT NULL,
ADD COLUMN IF NOT EXISTS runnable_id BIGINT,
ADD COLUMN IF NOT EXISTS runnable_path VARCHAR(255),
ADD COLUMN IF NOT EXISTS parent_job UUID,
ADD COLUMN IF NOT EXISTS root_job UUID,
ADD COLUMN IF NOT EXISTS script_lang script_lang DEFAULT 'python3'::script_lang,
ADD COLUMN IF NOT EXISTS script_entrypoint_override VARCHAR(255),
ADD COLUMN IF NOT EXISTS flow_step INTEGER,
ADD COLUMN IF NOT EXISTS flow_step_id VARCHAR(255),
ADD COLUMN IF NOT EXISTS flow_innermost_root_job UUID,
ADD COLUMN IF NOT EXISTS trigger VARCHAR(255),
ADD COLUMN IF NOT EXISTS trigger_kind job_trigger_kind,
ADD COLUMN IF NOT EXISTS same_worker BOOLEAN DEFAULT FALSE NOT NULL,
ADD COLUMN IF NOT EXISTS visible_to_owner BOOLEAN DEFAULT TRUE NOT NULL,
ADD COLUMN IF NOT EXISTS concurrent_limit INTEGER,
ADD COLUMN IF NOT EXISTS concurrency_time_window_s INTEGER,
ADD COLUMN IF NOT EXISTS cache_ttl INTEGER,
ADD COLUMN IF NOT EXISTS timeout INTEGER,
ADD COLUMN IF NOT EXISTS priority SMALLINT,
ADD COLUMN IF NOT EXISTS preprocessed BOOLEAN,
ADD COLUMN IF NOT EXISTS args JSONB,
ADD COLUMN IF NOT EXISTS labels TEXT[],
ADD COLUMN IF NOT EXISTS pre_run_error TEXT;
CREATE POLICY see_folder_extra_perms_user ON v2_job
AS PERMISSIVE
FOR ALL
TO windmill_user
USING ((visible_to_owner IS TRUE) AND (SPLIT_PART((runnable_path)::TEXT, '/'::TEXT, 1) = 'f'::TEXT) AND
(SPLIT_PART((runnable_path)::TEXT, '/'::TEXT, 2) = ANY (
REGEXP_SPLIT_TO_ARRAY(CURRENT_SETTING('session.folders_read'::TEXT), ','::TEXT))));
CREATE POLICY see_own_path ON v2_job
AS PERMISSIVE
FOR ALL
TO windmill_user
USING ((visible_to_owner IS TRUE) AND (SPLIT_PART((runnable_path)::TEXT, '/'::TEXT, 1) = 'u'::TEXT) AND
(SPLIT_PART((runnable_path)::TEXT, '/'::TEXT, 2) = CURRENT_SETTING('session.user'::TEXT)));
CREATE POLICY see_member_path ON v2_job
AS PERMISSIVE
FOR ALL
TO windmill_user
USING ((visible_to_owner IS TRUE) AND (SPLIT_PART((runnable_path)::TEXT, '/'::TEXT, 1) = 'g'::TEXT) AND
(SPLIT_PART((runnable_path)::TEXT, '/'::TEXT, 2) = ANY
(REGEXP_SPLIT_TO_ARRAY(CURRENT_SETTING('session.groups'::TEXT), ','::TEXT))));
CREATE POLICY see_own ON v2_job
AS PERMISSIVE
FOR ALL
TO windmill_user
USING ((SPLIT_PART((permissioned_as)::TEXT, '/'::TEXT, 1) = 'f'::TEXT) AND
(SPLIT_PART((permissioned_as)::TEXT, '/'::TEXT, 2) = CURRENT_SETTING('session.user'::TEXT)));
CREATE POLICY see_member ON v2_job
AS PERMISSIVE
FOR ALL
TO windmill_user
USING ((SPLIT_PART((permissioned_as)::TEXT, '/'::TEXT, 1) = 'g'::TEXT) AND
(SPLIT_PART((permissioned_as)::TEXT, '/'::TEXT, 2) = ANY
(REGEXP_SPLIT_TO_ARRAY(CURRENT_SETTING('session.groups'::TEXT), ','::TEXT))));
CREATE POLICY admin_policy ON v2_job
AS PERMISSIVE
FOR ALL
TO windmill_admin;
GRANT ALL ON v2_job TO windmill_user, windmill_admin;
backend/migrations/20250201124743_v2_job_queue_sync.down.sql
+4
View File
@@ -0,0 +1,4 @@
-- Add down migration script here
DROP FUNCTION v2_job_queue_before_insert() CASCADE;
DROP FUNCTION v2_job_queue_after_insert() CASCADE;
DROP FUNCTION v2_job_queue_before_update() CASCADE;
backend/migrations/20250201124743_v2_job_queue_sync.up.sql
+220
View File
@@ -0,0 +1,220 @@
-- Add up migration script here
-- This trigger will be removed once all server(s)/worker(s) are updated to use `v2_*` tables
CREATE OR REPLACE FUNCTION v2_job_queue_before_insert() RETURNS TRIGGER AS $$
DECLARE job v2_job;
BEGIN
IF NEW.__created_by IS NOT NULL THEN
-- v1 -> v2
IF NEW.__logs IS NOT NULL THEN
INSERT INTO job_logs (job_id, workspace_id, logs)
VALUES (NEW.id, NEW.workspace_id, NEW.__logs)
ON CONFLICT (job_id) DO UPDATE SET
logs = CONCAT(job_logs.logs, EXCLUDED.logs)
;
-- Need to be "before" to remove logs:
NEW.__logs := NULL;
END IF;
RETURN NEW;
END IF;
-- v2 -> v1
-- When inserting to `v2_job_queue` from `v2` code, set `v1` columns:
SELECT * INTO job FROM v2_job WHERE id = NEW.id;
NEW.__parent_job := job.parent_job;
NEW.__created_by := job.created_by;
NEW.__script_hash := job.runnable_id;
NEW.__script_path := job.runnable_path;
NEW.__args := job.args;
-- __logs
NEW.__raw_code := job.raw_code;
NEW.__canceled := NEW.canceled_by IS NOT NULL;
-- __last_ping
NEW.__job_kind := job.kind;
NEW.__env_id := 123456789; -- Magic used bellow.
NEW.__schedule_path := CASE WHEN job.trigger_kind = 'schedule'::job_trigger_kind THEN job.trigger END;
NEW.__permissioned_as := job.permissioned_as;
-- __flow_status
NEW.__raw_flow := job.raw_flow;
NEW.__is_flow_step := job.flow_step_id IS NOT NULL;
NEW.__language := job.script_lang;
NEW.__same_worker := job.same_worker;
NEW.__raw_lock := job.raw_lock;
NEW.__pre_run_error := job.pre_run_error;
NEW.__email := job.permissioned_as_email;
NEW.__visible_to_owner := job.visible_to_owner;
-- __mem_peak
NEW.__root_job := job.flow_innermost_root_job;
-- __leaf_jobs
NEW.__concurrent_limit := job.concurrent_limit;
NEW.__concurrency_time_window_s := job.concurrency_time_window_s;
NEW.__timeout := job.timeout;
NEW.__flow_step_id := job.flow_step_id;
NEW.__cache_ttl := job.cache_ttl;
IF job.script_entrypoint_override IS NOT NULL THEN
NEW.__args = jsonb_set(
coalesce(NEW.__args, '{}'::JSONB),
'{_ENTRYPOINT_OVERRIDE}',
to_jsonb(job.script_entrypoint_override)
);
END IF;
RETURN NEW;
END $$ LANGUAGE plpgsql;
CREATE OR REPLACE TRIGGER v2_job_queue_before_insert_trigger
BEFORE INSERT ON v2_job_queue
FOR EACH ROW
WHEN (pg_trigger_depth() < 1) -- Prevent infinite loop v1 <-> v2
EXECUTE FUNCTION v2_job_queue_before_insert();
-- v1 -> v2
-- On every insert to `v2_job_queue`, insert to `v2_job`, `v2_job_runtime` and `v2_job_status` as well
-- This trigger will be removed once all server(s)/worker(s) are updated to use `v2_*` tables
CREATE OR REPLACE FUNCTION v2_job_queue_after_insert() RETURNS TRIGGER AS $$ BEGIN
INSERT INTO v2_job (
id, workspace_id, created_at, created_by, permissioned_as, permissioned_as_email,
kind, runnable_id, runnable_path, parent_job,
script_lang, script_entrypoint_override,
flow_step, flow_step_id, flow_innermost_root_job,
trigger, trigger_kind,
tag, same_worker, visible_to_owner, concurrent_limit, concurrency_time_window_s, cache_ttl, timeout, priority,
preprocessed, args, pre_run_error,
raw_code, raw_lock, raw_flow
) VALUES (
NEW.id, NEW.workspace_id, NEW.created_at, NEW.__created_by, NEW.__permissioned_as, NEW.__email,
NEW.__job_kind, NEW.__script_hash, NEW.__script_path, NEW.__parent_job,
NEW.__language, NULLIF(NEW.__args->>'_ENTRYPOINT_OVERRIDE', '__WM_PREPROCESSOR'),
NULL, NEW.__flow_step_id, NEW.__root_job,
NEW.__schedule_path, CASE WHEN NEW.__schedule_path IS NOT NULL THEN 'schedule'::job_trigger_kind END,
NEW.tag, NEW.__same_worker, NEW.__visible_to_owner, NEW.__concurrent_limit, NEW.__concurrency_time_window_s,
NEW.__cache_ttl, NEW.__timeout, NEW.priority,
CASE WHEN (
NEW.__args->>'_ENTRYPOINT_OVERRIDE' IN ('__WM_PREPROCESSOR', 'preprocessor')
OR NEW.__flow_status->'preprocessor_module' IS NOT NULL
) THEN FALSE END, NEW.__args, NEW.__pre_run_error,
NEW.__raw_code, NEW.__raw_lock, NEW.__raw_flow
) ON CONFLICT (id) DO UPDATE SET
workspace_id = EXCLUDED.workspace_id,
created_at = EXCLUDED.created_at,
created_by = EXCLUDED.created_by,
permissioned_as = EXCLUDED.permissioned_as,
permissioned_as_email = EXCLUDED.permissioned_as_email,
kind = EXCLUDED.kind,
runnable_id = EXCLUDED.runnable_id,
runnable_path = EXCLUDED.runnable_path,
parent_job = EXCLUDED.parent_job,
script_lang = EXCLUDED.script_lang,
script_entrypoint_override = EXCLUDED.script_entrypoint_override,
flow_step = EXCLUDED.flow_step,
flow_step_id = EXCLUDED.flow_step_id,
flow_innermost_root_job = EXCLUDED.flow_innermost_root_job,
trigger = EXCLUDED.trigger,
trigger_kind = EXCLUDED.trigger_kind,
tag = EXCLUDED.tag,
same_worker = EXCLUDED.same_worker,
visible_to_owner = EXCLUDED.visible_to_owner,
concurrent_limit = EXCLUDED.concurrent_limit,
concurrency_time_window_s = EXCLUDED.concurrency_time_window_s,
cache_ttl = EXCLUDED.cache_ttl,
timeout = EXCLUDED.timeout,
priority = EXCLUDED.priority,
preprocessed = EXCLUDED.preprocessed,
args = EXCLUDED.args,
pre_run_error = EXCLUDED.pre_run_error,
raw_code = COALESCE(v2_job.raw_code, EXCLUDED.raw_code),
raw_lock = COALESCE(v2_job.raw_lock, EXCLUDED.raw_lock),
raw_flow = COALESCE(v2_job.raw_flow, EXCLUDED.raw_flow)
;
INSERT INTO v2_job_runtime (id, ping, memory_peak)
VALUES (NEW.id, NEW.__last_ping, NEW.__mem_peak)
ON CONFLICT (id) DO UPDATE SET
ping = COALESCE(v2_job_runtime.ping, EXCLUDED.ping),
memory_peak = COALESCE(v2_job_runtime.memory_peak, EXCLUDED.memory_peak)
;
IF NEW.__flow_status IS NOT NULL OR NEW.__leaf_jobs IS NOT NULL THEN
INSERT INTO v2_job_status (id, flow_status, flow_leaf_jobs)
VALUES (NEW.id, NEW.__flow_status, NEW.__leaf_jobs)
ON CONFLICT (id) DO UPDATE SET
flow_status = COALESCE(v2_job_status.flow_status, EXCLUDED.flow_status),
flow_leaf_jobs = COALESCE(v2_job_status.flow_leaf_jobs, EXCLUDED.flow_leaf_jobs)
;
END IF;
RETURN NEW;
END $$ LANGUAGE plpgsql;
CREATE OR REPLACE TRIGGER v2_job_queue_after_insert_trigger
AFTER INSERT ON v2_job_queue
FOR EACH ROW
-- Prevent infinite loop v1 <-> v2
WHEN (pg_trigger_depth() < 1 AND NEW.__created_by IS NOT NULL AND NEW.__env_id IS DISTINCT FROM 123456789)
EXECUTE FUNCTION v2_job_queue_after_insert();
-- On every update to `v2_job_queue`, update `v2_job`, `v2_job_runtime` and `v2_job_status` as well
-- This trigger will be removed once all server(s)/worker(s) are updated to use `v2_*` tables
CREATE OR REPLACE FUNCTION v2_job_queue_before_update() RETURNS TRIGGER AS $$ BEGIN
IF NEW.canceled_by IS NOT NULL THEN
NEW.__canceled := TRUE;
END IF;
-- `v2_job`: Only `args` are updated
IF NEW.__args::TEXT IS DISTINCT FROM OLD.__args::TEXT THEN
UPDATE v2_job SET
args = NEW.__args,
preprocessed = CASE WHEN preprocessed = FALSE THEN TRUE ELSE preprocessed END
WHERE id = NEW.id;
END IF;
-- `v2_job_runtime`:
IF NEW.__last_ping IS DISTINCT FROM OLD.__last_ping OR NEW.__mem_peak IS DISTINCT FROM OLD.__mem_peak THEN
INSERT INTO v2_job_runtime (id, ping, memory_peak)
VALUES (NEW.id, NEW.__last_ping, NEW.__mem_peak)
ON CONFLICT (id) DO UPDATE SET
ping = EXCLUDED.ping,
memory_peak = EXCLUDED.memory_peak
;
END IF;
-- `v2_job_status`:
IF NEW.__flow_status::TEXT IS DISTINCT FROM OLD.__flow_status::TEXT OR
NEW.__leaf_jobs::TEXT IS DISTINCT FROM OLD.__leaf_jobs::TEXT THEN
IF NEW.__is_flow_step = false AND NEW.__parent_job IS NOT NULL THEN
-- `workflow_as_code`:
INSERT INTO v2_job_status (id, workflow_as_code_status)
VALUES (NEW.id, NEW.__flow_status)
ON CONFLICT (id) DO UPDATE SET
workflow_as_code_status = EXCLUDED.workflow_as_code_status
;
ELSE
INSERT INTO v2_job_status (id, flow_status, flow_leaf_jobs)
VALUES (NEW.id, NEW.__flow_status, NEW.__leaf_jobs)
ON CONFLICT (id) DO UPDATE SET
flow_status = EXCLUDED.flow_status,
flow_leaf_jobs = EXCLUDED.flow_leaf_jobs
;
END IF;
END IF;
-- `job_logs`:
IF NEW.__logs IS DISTINCT FROM OLD.__logs THEN
INSERT INTO job_logs (job_id, workspace_id, logs)
VALUES (NEW.id, NEW.workspace_id, NEW.__logs)
ON CONFLICT (job_id) DO UPDATE SET
logs = CONCAT(job_logs.logs, EXCLUDED.logs)
;
NEW.__logs := NULL;
END IF;
RETURN NEW;
END $$ LANGUAGE plpgsql;
CREATE OR REPLACE TRIGGER v2_job_queue_before_update_trigger
BEFORE UPDATE ON v2_job_queue
FOR EACH ROW
WHEN (pg_trigger_depth() < 1) -- Prevent infinite loop v1 <-> v2
EXECUTE FUNCTION v2_job_queue_before_update();
-- v2 -> v1: update preprocessed args
CREATE OR REPLACE FUNCTION v2_job_after_update() RETURNS TRIGGER AS $$ BEGIN
UPDATE v2_job_queue SET __args = NEW.args WHERE id = NEW.id;
UPDATE v2_job_completed SET __args = NEW.args WHERE id = NEW.id;
RETURN NEW;
END $$ LANGUAGE PLPGSQL;
CREATE OR REPLACE TRIGGER v2_job_after_update_trigger
AFTER UPDATE ON v2_job
FOR EACH ROW
WHEN (pg_trigger_depth() < 1 AND NEW.args::TEXT IS DISTINCT FROM OLD.args::TEXT) -- Prevent infinite loop v1 <-> v2
EXECUTE FUNCTION v2_job_after_update();
backend/migrations/20250201124744_v2_job_completed_sync.down.sql
+3
View File
@@ -0,0 +1,3 @@
-- Add down migration script here
DROP FUNCTION v2_job_completed_before_insert() CASCADE;
DROP FUNCTION v2_job_completed_before_update() CASCADE;
backend/migrations/20250201124744_v2_job_completed_sync.up.sql
+153
View File
@@ -0,0 +1,153 @@
-- Add up migration script here
-- v1 -> v2
-- On every insert to `v2_job_completed`, insert to `v2_job` as well
-- This trigger will be removed once all server(s)/worker(s) are updated to use `v2_*` tables
CREATE OR REPLACE FUNCTION v2_job_completed_before_insert() RETURNS TRIGGER AS $$
DECLARE job v2_job;
DECLARE final_labels TEXT[];
BEGIN
-- New columns synchronization:
-- 1. `result_columns` <-> `flow_status._metadata.column_order`
-- 2. `v2_job.labels` <-> `result.wm_labels`
-- 3. v2 <-> flow_status._metadata.preprocessed_args`
IF NEW.__created_by IS NULL THEN
-- v2 -> v1
-- When inserting to `v2_job_completed` from `v2` code, set `v1` columns:
SELECT * INTO job FROM v2_job WHERE id = NEW.id;
NEW.__parent_job := job.parent_job;
NEW.__created_by := job.created_by;
NEW.__created_at := job.created_at;
NEW.__success := NEW.status = 'success'::job_status;
NEW.__script_hash := job.runnable_id;
NEW.__script_path := job.runnable_path;
NEW.__args := job.args;
-- __logs
NEW.__raw_code := job.raw_code;
NEW.__canceled := NEW.status = 'canceled'::job_status;
NEW.__job_kind := job.kind;
-- __env_id
NEW.__schedule_path := CASE WHEN job.trigger_kind = 'schedule'::job_trigger_kind THEN job.trigger END;
NEW.__permissioned_as := job.permissioned_as;
NEW.__raw_flow := job.raw_flow;
NEW.__is_flow_step := job.flow_step_id IS NOT NULL;
NEW.__language := job.script_lang;
NEW.__is_skipped := NEW.status = 'skipped'::job_status;
NEW.__raw_lock := job.raw_lock;
NEW.__email := job.permissioned_as_email;
NEW.__visible_to_owner := job.visible_to_owner;
NEW.__tag := job.tag;
NEW.__priority := job.priority;
-- 1. `result_columns` -> `flow_status._metadata.column_order`
IF NEW.result_columns IS NOT NULL AND (NEW.flow_status IS NULL OR jsonb_typeof(NEW.flow_status) = 'object') THEN
NEW.flow_status := jsonb_set(
coalesce(NEW.flow_status, '{}'::JSONB),
'{_metadata}',
jsonb_set(
coalesce(NEW.flow_status->'_metadata', '{}'::JSONB),
'{column_order}',
to_jsonb(NEW.result_columns)
)
);
END IF;
-- 2. `v2_job.labels` -> `result.wm_labels`
IF job.labels IS NOT NULL AND (NEW.result IS NULL OR jsonb_typeof(NEW.result) = 'object') THEN
IF jsonb_typeof(NEW.result->'wm_labels') = 'array' AND (
SELECT bool_and(jsonb_typeof(elem) = 'string')
FROM jsonb_array_elements(NEW.result->'wm_labels') AS elem
) THEN
SELECT array_agg(DISTINCT all_labels) INTO final_labels
FROM unnest(
coalesce(job.labels, ARRAY[]::TEXT[]) || translate(NEW.result->>'wm_labels', '[]', '{}')::TEXT[]
) all_labels;
ELSE
final_labels := job.labels;
END IF;
-- Update `v2_job.labels` if needed
IF job.labels IS DISTINCT FROM final_labels THEN
UPDATE v2_job SET labels = final_labels WHERE id = NEW.id;
END IF;
NEW.result := jsonb_set(
coalesce(NEW.result, '{}'::JSONB),
'{wm_labels}',
to_jsonb(final_labels)
);
END IF;
-- 3. v2 -> flow_status._metadata.preprocessed_args`
IF job.kind = 'script' AND job.preprocessed = TRUE
AND (NEW.flow_status IS NULL OR jsonb_typeof(NEW.flow_status) = 'object')
THEN
NEW.flow_status := jsonb_set(
coalesce(NEW.flow_status, '{}'::JSONB),
'{_metadata}',
jsonb_set(
coalesce(NEW.flow_status->'_metadata', '{}'::JSONB),
'{preprocessed_args}',
'true'::JSONB
)
);
END IF;
ELSE
-- v1 -> v2
NEW.completed_at := now();
NEW.status := CASE
WHEN NEW.__is_skipped THEN 'skipped'::job_status
WHEN NEW.__canceled THEN 'canceled'::job_status
WHEN NEW.__success THEN 'success'::job_status
ELSE 'failure'::job_status
END;
-- 1. `result_columns` <- `flow_status._metadata.column_order`
IF jsonb_typeof(NEW.flow_status->'_metadata'->'column_order') = 'array' AND (
SELECT bool_and(jsonb_typeof(elem) = 'string')
FROM jsonb_array_elements(NEW.flow_status->'_metadata'->'column_order') AS elem
) THEN
NEW.result_columns := translate(NEW.flow_status->'_metadata'->>'column_order', '[]', '{}')::TEXT[];
END IF;
-- 2. `v2_job.labels` <- `result.wm_labels`
IF jsonb_typeof(NEW.result->'wm_labels') = 'array' AND (
SELECT bool_and(jsonb_typeof(elem) = 'string')
FROM jsonb_array_elements(NEW.result->'wm_labels') AS elem
) THEN
UPDATE v2_job SET
labels = (
SELECT array_agg(DISTINCT all_labels)
FROM unnest(
coalesce(labels, ARRAY[]::TEXT[])
|| translate(NEW.result->>'wm_labels', '[]', '{}')::TEXT[]
) all_labels
)
WHERE id = NEW.id;
END IF;
-- 3. v2 <- flow_status._metadata.preprocessed_args`
IF NEW.flow_status->'_metadata'->'preprocessed_args' = 'true'::JSONB THEN
UPDATE v2_job SET
args = NEW.__args,
preprocessed = TRUE
WHERE id = NEW.id AND preprocessed = FALSE;
END IF;
END IF;
RETURN NEW;
END $$ LANGUAGE plpgsql;
CREATE OR REPLACE TRIGGER v2_job_completed_before_insert_trigger
BEFORE INSERT ON v2_job_completed
FOR EACH ROW
WHEN (pg_trigger_depth() < 1) -- Prevent infinite loop v1 <-> v2
EXECUTE FUNCTION v2_job_completed_before_insert();
CREATE OR REPLACE FUNCTION v2_job_completed_before_update() RETURNS TRIGGER AS $$ BEGIN
-- `v2_job`: Only `args` are updated
IF NEW.__args::TEXT IS DISTINCT FROM OLD.__args::TEXT THEN
UPDATE v2_job SET
args = NEW.__args,
preprocessed = CASE WHEN preprocessed = FALSE THEN TRUE ELSE preprocessed END
WHERE id = NEW.id;
END IF;
RETURN NEW;
END $$ LANGUAGE plpgsql;
CREATE OR REPLACE TRIGGER v2_job_completed_before_update_trigger
BEFORE UPDATE ON v2_job_completed
FOR EACH ROW
WHEN (pg_trigger_depth() < 1) -- Prevent infinite loop v1 <-> v2
EXECUTE FUNCTION v2_job_completed_before_update();
backend/migrations/20250201124748_v2_migrate_from_v1.down.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add down migration script here
-- Nothing to do here
backend/migrations/20250201124748_v2_migrate_from_v1.up.sql
+194
View File
@@ -0,0 +1,194 @@
-- Add up migration script here
-- Set new columns in `v2_job_completed`:
UPDATE v2_job_completed
SET completed_at = started_at + (interval '1 millisecond' * duration_ms),
status = CASE
WHEN __is_skipped THEN 'skipped'::job_status
WHEN __canceled THEN 'canceled'::job_status
WHEN __success THEN 'success'::job_status
ELSE 'failure'::job_status
END,
result_columns = CASE
WHEN jsonb_typeof(flow_status->'_metadata'->'column_order') = 'array' AND (
SELECT bool_and(jsonb_typeof(elem) = 'string')
FROM jsonb_array_elements(flow_status->'_metadata'->'column_order') AS elem
) THEN translate(flow_status->'_metadata'->>'column_order', '[]', '{}')::TEXT[]
END
WHERE status IS NULL;
-- Insert missing `v2_job` rows from `v2_job_queue`:
INSERT INTO v2_job (
id, workspace_id, created_at, created_by, permissioned_as, permissioned_as_email,
kind, runnable_id, runnable_path, parent_job,
script_lang, script_entrypoint_override,
flow_step_id, flow_innermost_root_job,
trigger, trigger_kind,
tag, same_worker, visible_to_owner, concurrent_limit, concurrency_time_window_s, cache_ttl, timeout, priority,
preprocessed, args, pre_run_error,
raw_code, raw_lock, raw_flow
) SELECT
id, workspace_id, created_at, __created_by, __permissioned_as, __email,
__job_kind, __script_hash, __script_path, __parent_job,
__language, NULLIF(__args->>'_ENTRYPOINT_OVERRIDE', '__WM_PREPROCESSOR'),
__flow_step_id, __root_job,
__schedule_path, CASE WHEN __schedule_path IS NOT NULL THEN 'schedule'::job_trigger_kind END,
tag, __same_worker, __visible_to_owner, __concurrent_limit, __concurrency_time_window_s,
__cache_ttl, __timeout, priority,
CASE
WHEN __args->>'_ENTRYPOINT_OVERRIDE' IN ('__WM_PREPROCESSOR', 'preprocessor') THEN FALSE
WHEN __flow_status->'preprocessor_module' IS NOT NULL THEN __args->'wm_trigger' IS NULL
END, __args, __pre_run_error,
__raw_code, __raw_lock, __raw_flow
FROM v2_job_queue
WHERE NOT EXISTS (SELECT 1 FROM v2_job WHERE v2_job.id = v2_job_queue.id);
-- Insert missing `v2_job` rows from `v2_job_completed`:
INSERT INTO v2_job (
id, workspace_id, created_at, created_by, permissioned_as, permissioned_as_email,
kind, runnable_id, runnable_path, parent_job,
script_lang, script_entrypoint_override,
trigger, trigger_kind,
tag, visible_to_owner, priority,
preprocessed, args,
raw_code, raw_lock, raw_flow,
labels
) SELECT
id, workspace_id, __created_at, __created_by, __permissioned_as, __email,
__job_kind, __script_hash, __script_path, __parent_job,
__language, NULLIF(__args->>'_ENTRYPOINT_OVERRIDE', '__WM_PREPROCESSOR'),
__schedule_path, CASE WHEN __schedule_path IS NOT NULL THEN 'schedule'::job_trigger_kind END,
__tag, __visible_to_owner, __priority,
CASE WHEN v2_job_completed.flow_status->'_metadata'->'preprocessed_args' = 'true'::JSONB THEN TRUE END, __args,
__raw_code, __raw_lock, __raw_flow,
CASE
WHEN jsonb_typeof(result->'wm_labels') = 'array' AND (
SELECT bool_and(jsonb_typeof(elem) = 'string')
FROM jsonb_array_elements(result->'wm_labels') AS elem
) THEN (
SELECT array_agg(DISTINCT all_labels)
FROM unnest(
translate(result->>'wm_labels', '[]', '{}')::TEXT[]
) all_labels
)
END
FROM v2_job_completed
WHERE NOT EXISTS (SELECT 1 FROM v2_job WHERE v2_job.id = v2_job_completed.id);
-- Set existing `v2_job` rows from `v2_job_queue`:
UPDATE v2_job SET
created_at = v2_job_queue.created_at,
created_by = v2_job_queue.__created_by,
permissioned_as = v2_job_queue.__permissioned_as,
permissioned_as_email = v2_job_queue.__email,
kind = v2_job_queue.__job_kind,
runnable_id = v2_job_queue.__script_hash,
runnable_path = v2_job_queue.__script_path,
parent_job = v2_job_queue.__parent_job,
script_lang = v2_job_queue.__language,
script_entrypoint_override = NULLIF(v2_job_queue.__args->>'_ENTRYPOINT_OVERRIDE', '__WM_PREPROCESSOR'),
flow_step_id = v2_job_queue.__flow_step_id,
flow_innermost_root_job = v2_job_queue.__root_job,
trigger = v2_job_queue.__schedule_path,
trigger_kind = CASE WHEN v2_job_queue.__schedule_path IS NOT NULL THEN 'schedule'::job_trigger_kind END,
tag = v2_job_queue.tag,
same_worker = v2_job_queue.__same_worker,
visible_to_owner = v2_job_queue.__visible_to_owner,
concurrent_limit = v2_job_queue.__concurrent_limit,
concurrency_time_window_s = v2_job_queue.__concurrency_time_window_s,
cache_ttl = v2_job_queue.__cache_ttl,
timeout = v2_job_queue.__timeout,
priority = v2_job_queue.priority,
preprocessed = CASE
WHEN __args->>'_ENTRYPOINT_OVERRIDE' IN ('__WM_PREPROCESSOR', 'preprocessor') THEN FALSE
WHEN __flow_status->'preprocessor_module' IS NOT NULL THEN __args->'wm_trigger' IS NULL
END,
args = v2_job_queue.__args,
pre_run_error = v2_job_queue.__pre_run_error,
raw_code = COALESCE(v2_job.raw_code, v2_job_queue.__raw_code),
raw_lock = COALESCE(v2_job.raw_lock, v2_job_queue.__raw_lock),
raw_flow = COALESCE(v2_job.raw_flow, v2_job_queue.__raw_flow)
FROM v2_job_queue
WHERE v2_job.id = v2_job_queue.id AND v2_job.created_by = 'missing';
-- Set existing `v2_job` rows from `v2_job_completed`:
UPDATE v2_job SET
created_at = v2_job_completed.__created_at,
created_by = v2_job_completed.__created_by,
permissioned_as = v2_job_completed.__permissioned_as,
permissioned_as_email = v2_job_completed.__email,
kind = v2_job_completed.__job_kind,
runnable_id = v2_job_completed.__script_hash,
runnable_path = v2_job_completed.__script_path,
parent_job = v2_job_completed.__parent_job,
script_lang = v2_job_completed.__language,
script_entrypoint_override = NULLIF(v2_job_completed.__args->>'_ENTRYPOINT_OVERRIDE', '__WM_PREPROCESSOR'),
trigger = v2_job_completed.__schedule_path,
trigger_kind = CASE WHEN v2_job_completed.__schedule_path IS NOT NULL THEN 'schedule'::job_trigger_kind END,
tag = v2_job_completed.__tag,
visible_to_owner = v2_job_completed.__visible_to_owner,
priority = v2_job_completed.__priority,
preprocessed = CASE WHEN v2_job_completed.flow_status->'_metadata'->'preprocessed_args' = 'true'::JSONB THEN TRUE END,
args = v2_job_completed.__args,
raw_code = COALESCE(v2_job.raw_code, v2_job_completed.__raw_code),
raw_lock = COALESCE(v2_job.raw_lock, v2_job_completed.__raw_lock),
raw_flow = COALESCE(v2_job.raw_flow, v2_job_completed.__raw_flow),
labels = CASE
WHEN jsonb_typeof(v2_job_completed.result->'wm_labels') = 'array' AND (
SELECT bool_and(jsonb_typeof(elem) = 'string')
FROM jsonb_array_elements(v2_job_completed.result->'wm_labels') AS elem
) THEN (
SELECT array_agg(DISTINCT all_labels)
FROM unnest(
coalesce(v2_job.labels, ARRAY[]::TEXT[])
|| translate(v2_job_completed.result->>'wm_labels', '[]', '{}')::TEXT[]
) all_labels
)
END
FROM v2_job_completed
WHERE v2_job.id = v2_job_completed.id AND v2_job.created_by = 'missing';
-- Migrate `v2_job_queue` moved columns to `v2_job_runtime`:
INSERT INTO v2_job_runtime (id, ping, memory_peak)
SELECT id, __last_ping, __mem_peak
FROM v2_job_queue
WHERE __last_ping IS NOT NULL OR __mem_peak IS NOT NULL
-- Locked ones will sync within triggers
FOR UPDATE SKIP LOCKED
ON CONFLICT (id) DO NOTHING;
-- Migrate `v2_job_queue` moved columns to `v2_job_status` for `workflow_as_code`:
INSERT INTO v2_job_status (id, workflow_as_code_status)
SELECT id, __flow_status
FROM v2_job_queue
WHERE __flow_status IS NOT NULL AND __is_flow_step = false AND __parent_job IS NOT NULL
-- Locked ones will sync within triggers
FOR UPDATE SKIP LOCKED
ON CONFLICT (id) DO NOTHING;
-- Migrate `v2_job_queue` moved columns to `v2_job_status`:
INSERT INTO v2_job_status (id, flow_status, flow_leaf_jobs)
SELECT id, __flow_status, __leaf_jobs
FROM v2_job_queue
WHERE __flow_status IS NOT NULL OR __leaf_jobs IS NOT NULL
-- Locked ones will sync within triggers
FOR UPDATE SKIP LOCKED
ON CONFLICT (id) DO NOTHING;
-- Migrate old `v2_job_queue.__logs` to `job_logs`
INSERT INTO job_logs (job_id, workspace_id, logs)
SELECT id, workspace_id, __logs
FROM v2_job_queue
WHERE __logs IS NOT NULL
ON CONFLICT (job_id) DO UPDATE SET
logs = CONCAT(job_logs.logs, EXCLUDED.logs)
;
-- Migrate old `v2_job_completed.__logs` to `job_logs`
INSERT INTO job_logs (job_id, workspace_id, logs)
SELECT id, workspace_id, __logs
FROM v2_job_completed
WHERE __logs IS NOT NULL AND __logs IS DISTINCT FROM '##DELETED##'
ON CONFLICT (job_id) DO UPDATE SET
logs = CONCAT(job_logs.logs, EXCLUDED.logs)
;
backend/migrations/20250201124804_v2_job_runtime_sync.down.sql
+3
View File
@@ -0,0 +1,3 @@
-- Add down migration script here
DROP FUNCTION v2_job_runtime_before_insert() CASCADE;
DROP FUNCTION v2_job_runtime_before_update() CASCADE;
backend/migrations/20250201124804_v2_job_runtime_sync.up.sql
+31
View File
@@ -0,0 +1,31 @@
-- Add up migration script here
-- On every insert/update to `v2_job_runtime`, reflect to `v2_job_queue` as well
-- This triggers will be removed once all server(s)/worker(s) are updated to use `v2_*` tables
CREATE OR REPLACE FUNCTION v2_job_runtime_before_insert() RETURNS TRIGGER AS $$ BEGIN
UPDATE v2_job_queue
SET __last_ping = NEW.ping, __mem_peak = NEW.memory_peak
WHERE id = NEW.id;
RETURN NEW;
END $$ LANGUAGE plpgsql;
CREATE OR REPLACE TRIGGER v2_job_runtime_before_insert_trigger
BEFORE INSERT ON v2_job_runtime
FOR EACH ROW
WHEN (pg_trigger_depth() < 1) -- Prevent infinite loop v2 <-> v1
EXECUTE FUNCTION v2_job_runtime_before_insert();
CREATE OR REPLACE FUNCTION v2_job_runtime_before_update() RETURNS TRIGGER AS $$ BEGIN
IF NEW.ping IS DISTINCT FROM OLD.ping OR NEW.memory_peak IS DISTINCT FROM OLD.memory_peak THEN
UPDATE v2_job_queue
SET __last_ping = NEW.ping, __mem_peak = NEW.memory_peak
WHERE id = NEW.id;
END IF;
RETURN NEW;
END $$ LANGUAGE plpgsql;
CREATE OR REPLACE TRIGGER v2_job_runtime_before_update_trigger
BEFORE UPDATE ON v2_job_runtime
FOR EACH ROW
WHEN (pg_trigger_depth() < 1) -- Prevent infinite loop v2 <-> v1
EXECUTE FUNCTION v2_job_runtime_before_update();
backend/migrations/20250201124805_v2_job_status_sync.down.sql
+3
View File
@@ -0,0 +1,3 @@
-- Add down migration script here
DROP FUNCTION v2_job_status_before_insert() CASCADE;
DROP FUNCTION v2_job_status_before_update() CASCADE;
backend/migrations/20250201124805_v2_job_status_sync.up.sql
+37
View File
@@ -0,0 +1,37 @@
-- Add up migration script here
-- On every insert/update to `v2_job_status`, reflect to `v2_job_queue` as well
-- This triggers will be removed once all server(s)/worker(s) are updated to use `v2_*` tables
CREATE OR REPLACE FUNCTION v2_job_status_before_insert() RETURNS TRIGGER AS $$ BEGIN
UPDATE v2_job_queue
SET __flow_status = NEW.flow_status, __leaf_jobs = NEW.flow_leaf_jobs
WHERE id = NEW.id;
RETURN NEW;
END $$ LANGUAGE plpgsql;
CREATE OR REPLACE TRIGGER v2_job_status_before_insert_trigger
BEFORE INSERT ON v2_job_status
FOR EACH ROW
WHEN (pg_trigger_depth() < 1) -- Prevent infinite loop v2 <-> v1
EXECUTE FUNCTION v2_job_status_before_insert();
CREATE OR REPLACE FUNCTION v2_job_status_before_update() RETURNS TRIGGER AS $$ BEGIN
IF NEW.flow_status::TEXT IS DISTINCT FROM OLD.flow_status::TEXT OR
NEW.flow_leaf_jobs::TEXT IS DISTINCT FROM OLD.flow_leaf_jobs::TEXT THEN
UPDATE v2_job_queue
SET __flow_status = NEW.flow_status, __leaf_jobs = NEW.flow_leaf_jobs
WHERE id = NEW.id;
END IF;
IF NEW.workflow_as_code_status::TEXT IS DISTINCT FROM OLD.workflow_as_code_status::TEXT THEN
UPDATE v2_job_queue
SET __flow_status = NEW.workflow_as_code_status
WHERE id = NEW.id;
END IF;
RETURN NEW;
END $$ LANGUAGE plpgsql;
CREATE OR REPLACE TRIGGER v2_job_status_before_update_trigger
BEFORE UPDATE ON v2_job_status
FOR EACH ROW
WHEN (pg_trigger_depth() < 1) -- Prevent infinite loop v2 <-> v1
EXECUTE FUNCTION v2_job_status_before_update();
backend/migrations/20250201124919_v2_job_constraints.down.sql
+4
View File
@@ -0,0 +1,4 @@
-- Add down migration script here
ALTER TABLE v2_job ALTER COLUMN workspace_id DROP NOT NULL;
ALTER TABLE v2_job ALTER COLUMN tag DROP DEFAULT;
ALTER TABLE v2_job ALTER COLUMN tag DROP NOT NULL;
backend/migrations/20250201124919_v2_job_constraints.up.sql
+4
View File
@@ -0,0 +1,4 @@
-- Add up migration script here
ALTER TABLE v2_job ALTER COLUMN workspace_id SET NOT NULL;
ALTER TABLE v2_job ALTER COLUMN tag SET DEFAULT 'other';
ALTER TABLE v2_job ALTER COLUMN tag SET NOT NULL;
backend/migrations/20250201145628_v2_job_queue_constraints.down.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add up migration script here
ALTER TABLE v2_job_queue ALTER COLUMN __created_by SET NOT NULL;
backend/migrations/20250201145628_v2_job_queue_constraints.up.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add up migration script here
ALTER TABLE v2_job_queue ALTER COLUMN __created_by DROP NOT NULL;
backend/migrations/20250201145629_v2_job_completed_constraints.down.sql
+10
View File
@@ -0,0 +1,10 @@
-- Add down migration script here
LOCK TABLE v2_job_completed IN ACCESS EXCLUSIVE MODE;
UPDATE v2_job_completed SET started_at = completed_at WHERE started_at IS NULL;
ALTER TABLE v2_job_completed ALTER COLUMN status DROP NOT NULL;
ALTER TABLE v2_job_completed ALTER COLUMN completed_at DROP DEFAULT;
ALTER TABLE v2_job_completed ALTER COLUMN completed_at DROP NOT NULL;
ALTER TABLE v2_job_completed ALTER COLUMN started_at SET NOT NULL;
ALTER TABLE v2_job_completed ALTER COLUMN __created_at SET NOT NULL;
ALTER TABLE v2_job_completed ALTER COLUMN __created_by SET NOT NULL;
ALTER TABLE v2_job_completed ALTER COLUMN __success SET NOT NULL;
backend/migrations/20250201145629_v2_job_completed_constraints.up.sql
+8
View File
@@ -0,0 +1,8 @@
-- Add up migration script here
ALTER TABLE v2_job_completed ALTER COLUMN status SET NOT NULL;
ALTER TABLE v2_job_completed ALTER COLUMN completed_at SET DEFAULT now();
ALTER TABLE v2_job_completed ALTER COLUMN completed_at SET NOT NULL;
ALTER TABLE v2_job_completed ALTER COLUMN started_at DROP NOT NULL;
ALTER TABLE v2_job_completed ALTER COLUMN __created_at DROP NOT NULL;
ALTER TABLE v2_job_completed ALTER COLUMN __created_by DROP NOT NULL;
ALTER TABLE v2_job_completed ALTER COLUMN __success DROP NOT NULL;
backend/src/monitor.rs
+24 -24
View File
@@ -735,11 +735,11 @@ pub async fn delete_expired_items(db: &DB) -> () {
match db.begin().await {
Ok(mut tx) => {
let deleted_jobs = sqlx::query_scalar!(
"DELETE FROM completed_job WHERE created_at <= now() - ($1::bigint::text || ' s')::interval AND started_at + ((duration_ms/1000 + $1::bigint) || ' s')::interval <= now() RETURNING id",
job_retention_secs
)
.fetch_all(&mut *tx)
.await;
"DELETE FROM completed_job WHERE created_at <= now() - ($1::bigint::text || ' s')::interval AND started_at + ((duration_ms/1000 + $1::bigint) || ' s')::interval <= now() RETURNING id AS \"id!\"",
job_retention_secs
)
.fetch_all(&mut *tx)
.await;
match deleted_jobs {
Ok(deleted_jobs) => {
@@ -1507,26 +1507,26 @@ pub async fn reload_base_url_setting(db: &DB) -> error::Result<()> {
async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str, worker_name: &str) {
if *RESTART_ZOMBIE_JOBS {
let restarted = sqlx::query!(
"WITH zombie_jobs AS (
UPDATE queue SET running = false, started_at = null
WHERE last_ping < now() - ($1 || ' seconds')::interval
AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow') AND same_worker = false
RETURNING id, workspace_id, last_ping
),
update_concurrency AS (
UPDATE concurrency_counter cc
SET job_uuids = job_uuids - zj.id::text
FROM zombie_jobs zj
INNER JOIN concurrency_key ck ON ck.job_id = zj.id
WHERE cc.concurrency_id = ck.key
)
SELECT id, workspace_id, last_ping FROM zombie_jobs",
*ZOMBIE_JOB_TIMEOUT,
"WITH zombie_jobs AS (
UPDATE queue SET running = false, started_at = null
WHERE last_ping < now() - ($1 || ' seconds')::interval
AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow') AND same_worker = false
RETURNING id, workspace_id, last_ping
),
update_concurrency AS (
UPDATE concurrency_counter cc
SET job_uuids = job_uuids - zj.id::text
FROM zombie_jobs zj
INNER JOIN concurrency_key ck ON ck.job_id = zj.id
WHERE cc.concurrency_id = ck.key
)
.fetch_all(db)
.await
.ok()
.unwrap_or_else(|| vec![]);
SELECT id AS \"id!\", workspace_id AS \"workspace_id!\", last_ping FROM zombie_jobs",
*ZOMBIE_JOB_TIMEOUT,
)
.fetch_all(db)
.await
.ok()
.unwrap_or_else(|| vec![]);
#[cfg(feature = "prometheus")]
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
backend/tests/fixtures/base.sql
Vendored
+184 -4
View File
@@ -29,7 +29,7 @@ END;
$$ LANGUAGE PLPGSQL;
CREATE TRIGGER "notify_insert_on_completed_job"
AFTER INSERT ON "completed_job"
AFTER INSERT ON "v2_job_completed"
FOR EACH ROW
EXECUTE FUNCTION "notify_insert_on_completed_job" ();
@@ -43,12 +43,192 @@ END;
$$ LANGUAGE PLPGSQL;
CREATE TRIGGER "notify_queue_after_insert"
AFTER INSERT ON "queue"
AFTER INSERT ON "v2_job_queue"
FOR EACH ROW
EXECUTE FUNCTION "notify_queue" ();
CREATE TRIGGER "notify_queue_after_flow_status_update"
AFTER UPDATE ON "queue"
AFTER UPDATE ON "v2_job_status"
FOR EACH ROW
WHEN (NEW.flow_status IS DISTINCT FROM OLD.flow_status)
EXECUTE FUNCTION "notify_queue" ();
EXECUTE FUNCTION "notify_queue" ();
-- TODO(uael): remove before phase 4
CREATE OR REPLACE FUNCTION zzz_v2_job_queue_integrity_check() RETURNS TRIGGER AS $$
DECLARE job v2_job;
DECLARE job_runtime v2_job_runtime;
DECLARE job_status v2_job_status;
BEGIN
IF (OLD.canceled_by IS NOT NULL) IS DISTINCT FROM OLD.__canceled THEN
RAISE EXCEPTION 'canceled mismatch';
END IF;
-- v2_job:
SELECT * INTO job FROM v2_job WHERE id = OLD.id;
IF job.tag IS DISTINCT FROM OLD.tag THEN
RAISE EXCEPTION 'tag mismatch';
END IF;
IF job.workspace_id IS DISTINCT FROM OLD.workspace_id THEN
RAISE EXCEPTION 'workspace_id mismatch';
END IF;
IF job.created_at IS DISTINCT FROM OLD.created_at THEN
RAISE EXCEPTION 'created_at mismatch';
END IF;
IF job.created_by IS DISTINCT FROM OLD.__created_by THEN
RAISE EXCEPTION 'created_by mismatch';
END IF;
IF job.permissioned_as IS DISTINCT FROM OLD.__permissioned_as THEN
RAISE EXCEPTION 'permissioned_as mismatch';
END IF;
IF job.permissioned_as_email IS DISTINCT FROM OLD.__email THEN
RAISE EXCEPTION 'permissioned_as_email mismatch';
END IF;
IF job.kind IS DISTINCT FROM OLD.__job_kind THEN
RAISE EXCEPTION 'kind mismatch';
END IF;
IF job.runnable_id IS DISTINCT FROM OLD.__script_hash THEN
RAISE EXCEPTION 'runnable_id mismatch';
END IF;
IF job.runnable_path IS DISTINCT FROM OLD.__script_path THEN
RAISE EXCEPTION 'runnable_path mismatch';
END IF;
IF job.parent_job IS DISTINCT FROM OLD.__parent_job THEN
RAISE EXCEPTION 'parent_job mismatch';
END IF;
IF job.script_lang IS DISTINCT FROM OLD.__language THEN
RAISE EXCEPTION 'script_lang mismatch';
END IF;
IF job.script_entrypoint_override IS DISTINCT FROM NULLIF(OLD.__args->>'_ENTRYPOINT_OVERRIDE', '__WM_PREPROCESSOR')
AND OLD.__args->>'reason' IS DISTINCT FROM 'PREPROCESSOR_ARGS_ARE_DISCARDED'
THEN
RAISE EXCEPTION 'script_entrypoint_override mismatch';
END IF;
IF job.flow_step_id IS DISTINCT FROM OLD.__flow_step_id THEN
RAISE EXCEPTION 'flow_step_id mismatch';
END IF;
IF (job.flow_step_id IS NOT NULL) IS DISTINCT FROM OLD.__is_flow_step THEN
RAISE EXCEPTION 'is_flow_step mismatch';
END IF;
IF job.flow_innermost_root_job IS DISTINCT FROM OLD.__root_job THEN
RAISE EXCEPTION 'flow_innermost_root_job mismatch';
END IF;
IF job.trigger IS DISTINCT FROM OLD.__schedule_path THEN
RAISE EXCEPTION 'trigger mismatch';
END IF;
IF job.same_worker IS DISTINCT FROM OLD.__same_worker THEN
RAISE EXCEPTION 'same_worker mismatch';
END IF;
IF job.visible_to_owner IS DISTINCT FROM OLD.__visible_to_owner THEN
RAISE EXCEPTION 'visible_to_owner mismatch';
END IF;
IF job.concurrent_limit IS DISTINCT FROM OLD.__concurrent_limit THEN
RAISE EXCEPTION 'concurrent_limit mismatch';
END IF;
IF job.concurrency_time_window_s IS DISTINCT FROM OLD.__concurrency_time_window_s THEN
RAISE EXCEPTION 'concurrency_time_window_s mismatch';
END IF;
IF job.cache_ttl IS DISTINCT FROM OLD.__cache_ttl THEN
RAISE EXCEPTION 'cache_ttl mismatch';
END IF;
IF job.timeout IS DISTINCT FROM OLD.__timeout THEN
RAISE EXCEPTION 'timeout mismatch';
END IF;
IF job.priority IS DISTINCT FROM OLD.priority THEN
RAISE EXCEPTION 'priority mismatch';
END IF;
IF job.args::TEXT IS DISTINCT FROM OLD.__args::TEXT AND OLD.__args->>'_ENTRYPOINT_OVERRIDE' IS DISTINCT FROM '__WM_PREPROCESSOR' THEN
RAISE EXCEPTION 'args mismatch';
END IF;
IF job.pre_run_error IS DISTINCT FROM OLD.__pre_run_error THEN
RAISE EXCEPTION 'pre_run_error mismatch';
END IF;
-- v2_job_runtime:
SELECT * INTO job_runtime FROM v2_job_runtime WHERE id = OLD.id;
IF job_runtime.ping IS DISTINCT FROM OLD.__last_ping THEN
RAISE EXCEPTION 'ping mismatch';
END IF;
IF job_runtime.memory_peak IS DISTINCT FROM OLD.__mem_peak THEN
RAISE EXCEPTION 'memory_peak mismatch';
END IF;
-- v2_job_status:
IF EXISTS(SELECT 1 FROM v2_job_status WHERE id = OLD.id) THEN
SELECT * INTO job_status FROM v2_job_status WHERE id = OLD.id;
IF job_status.flow_status::TEXT IS DISTINCT FROM OLD.__flow_status::TEXT THEN
RAISE EXCEPTION 'flow_status mismatch';
END IF;
IF job_status.flow_leaf_jobs::TEXT IS DISTINCT FROM OLD.__leaf_jobs::TEXT THEN
RAISE EXCEPTION 'leaf_jobs mismatch';
END IF;
END IF;
RETURN OLD;
END $$ LANGUAGE PLPGSQL;
CREATE OR REPLACE TRIGGER zzz_v2_job_queue_integrity_check_before_delete
BEFORE DELETE ON v2_job_queue
FOR EACH ROW
EXECUTE FUNCTION zzz_v2_job_queue_integrity_check();
-- TODO(uael): remove before phase 4
CREATE OR REPLACE FUNCTION zzz_v2_job_completed_integrity_check() RETURNS TRIGGER AS $$
DECLARE job v2_job;
BEGIN
IF (NEW.canceled_by IS NOT NULL) IS DISTINCT FROM NEW.__canceled THEN
RAISE EXCEPTION 'canceled mismatch';
END IF;
SELECT * INTO job FROM v2_job WHERE id = NEW.id;
IF job.tag IS DISTINCT FROM NEW.__tag THEN
RAISE EXCEPTION 'tag mismatch % %', job.tag, NEW.__tag;
END IF;
IF job.workspace_id IS DISTINCT FROM NEW.workspace_id THEN
RAISE EXCEPTION 'workspace_id mismatch';
END IF;
IF job.created_at IS DISTINCT FROM NEW.__created_at THEN
RAISE EXCEPTION 'created_at mismatch';
END IF;
IF job.created_by IS DISTINCT FROM NEW.__created_by THEN
RAISE EXCEPTION 'created_by mismatch';
END IF;
IF job.permissioned_as IS DISTINCT FROM NEW.__permissioned_as THEN
RAISE EXCEPTION 'permissioned_as mismatch';
END IF;
IF job.permissioned_as_email IS DISTINCT FROM NEW.__email THEN
RAISE EXCEPTION 'permissioned_as_email mismatch';
END IF;
IF job.kind IS DISTINCT FROM NEW.__job_kind THEN
RAISE EXCEPTION 'kind mismatch';
END IF;
IF job.runnable_id IS DISTINCT FROM NEW.__script_hash THEN
RAISE EXCEPTION 'runnable_id mismatch';
END IF;
IF job.runnable_path IS DISTINCT FROM NEW.__script_path THEN
RAISE EXCEPTION 'runnable_path mismatch';
END IF;
IF job.parent_job IS DISTINCT FROM NEW.__parent_job THEN
RAISE EXCEPTION 'parent_job mismatch';
END IF;
IF job.script_lang IS DISTINCT FROM NEW.__language THEN
RAISE EXCEPTION 'script_lang mismatch';
END IF;
IF job.script_entrypoint_override IS DISTINCT FROM NULLIF(NEW.__args->>'_ENTRYPOINT_OVERRIDE', '__WM_PREPROCESSOR')
AND NEW.__args->>'reason' IS DISTINCT FROM 'PREPROCESSOR_ARGS_ARE_DISCARDED'
THEN
RAISE EXCEPTION 'script_entrypoint_override mismatch';
END IF;
IF (job.flow_step_id IS NOT NULL) IS DISTINCT FROM NEW.__is_flow_step THEN
RAISE EXCEPTION 'is_flow_step mismatch';
END IF;
IF job.trigger IS DISTINCT FROM NEW.__schedule_path THEN
RAISE EXCEPTION 'trigger mismatch';
END IF;
IF job.visible_to_owner IS DISTINCT FROM NEW.__visible_to_owner THEN
RAISE EXCEPTION 'visible_to_owner mismatch';
END IF;
IF job.args::TEXT IS DISTINCT FROM NEW.__args::TEXT AND NEW.__args->>'_ENTRYPOINT_OVERRIDE' IS DISTINCT FROM '__WM_PREPROCESSOR' THEN
RAISE EXCEPTION 'args mismatch';
END IF;
RETURN NEW;
END $$ LANGUAGE PLPGSQL;
CREATE OR REPLACE TRIGGER zzz_v2_job_completed_integrity_check_after_insert
AFTER INSERT ON v2_job_completed
FOR EACH ROW
EXECUTE FUNCTION zzz_v2_job_completed_integrity_check();
backend/tests/fixtures/hello.sql
Vendored
+39 -2
View File
@@ -13,6 +13,23 @@ export function main(world: string) {
'',
'f/system/hello', 123412, 'deno', '');
INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES (
'test-workspace',
'system',
'
export async function preprocessor() {
return { foo: ''bar'', bar: ''baz'' };
}
export async function main(foo: string, bar: string) {
return ''Hello '' + foo + '' '' + bar;
}
',
'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"foo":{"default":null,"description":"","originalType":"string","type":"string"},"bar":{"default":null,"description":"","originalType":"string","type":"string"}},"required":["foo","bar"],"type":"object"}',
'',
'',
'f/system/hello_with_preprocessor', 123413, 'deno', '');
INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES (
'test-workspace',
'',
@@ -40,7 +57,7 @@ INSERT INTO public.flow(workspace_id, summary, description, path, versions, sche
'f/system/hello_with_nodes_flow',
'{1443253234253454}',
'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"world":{"default":"world","description":"","type":"string"}},"required":[],"type":"object"}',
E'{"modules":[{"id":"a","value":{"type":"forloopflow","modules":[{"id":"b","value":{"type":"rawscript","content":"export function main(world: string) {\\n const greet = `Hello ${world}!`;\\n console.log(greet)\\n return greet\\n}\\n","language":"deno","input_transforms":{"world":{"type":"javascript","expr":"flow_input.iter.value"}},"is_trigger":false}},{"id":"c","value":{"type":"rawscript","content":"export function main(hello: string) {\\n const dareyou = `Did you just say \\"${hello}\\"??!`;\\n console.log(dareyou)\\n return dareyou\\n}","language":"deno","input_transforms":{"hello":{"type":"javascript","value":"${results.b}","expr":"`${results.b}`"}},"is_trigger":false}}],"iterator":{"type":"javascript","expr":"[\'foo\', \'bar\', \'baz\']"},"skip_failures":true,"parallel":false}}],"same_worker":false}',
E'{"modules":[{"id":"a","value":{"type":"forloopflow","modules":[{"id":"b","value":{"type":"rawscript","content":"export function main(world: string) {\\n const greet = `Hello ${world}!`;\\n console.log(greet)\\n return greet\\n}\\n","language":"deno","input_transforms":{"world":{"type":"javascript","expr":"flow_input.iter.value"}},"is_trigger":false}},{"id":"c","value":{"type":"rawscript","content":"export function main(hello: string) {\\n const dareyou = `Did you just say \\"${hello}\\"??!`;\\n console.log(dareyou)\\n return dareyou\\n}","language":"deno","input_transforms":{"hello":{"type":"javascript","value":"${results.b}","expr":"`${results.b}`"}},"is_trigger":false}}],"iterator":{"type":"javascript","expr":"[''foo'', ''bar'', ''baz'']"},"skip_failures":true,"parallel":false}}],"same_worker":false}',
'system'
);
@@ -49,6 +66,26 @@ INSERT INTO public.flow_version(id, workspace_id, path, schema, value, created_b
'test-workspace',
'f/system/hello_with_nodes_flow',
'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"world":{"default":"world","description":"","type":"string"}},"required":[],"type":"object"}',
E'{"modules":[{"id":"a","value":{"type":"forloopflow","modules":[{"id":"b","value":{"type":"rawscript","content":"export function main(world: string) {\\n const greet = `Hello ${world}!`;\\n console.log(greet)\\n return greet\\n}\\n","language":"deno","input_transforms":{"world":{"type":"javascript","expr":"flow_input.iter.value"}},"is_trigger":false}},{"id":"c","value":{"type":"rawscript","content":"export function main(hello: string) {\\n const dareyou = `Did you just say \\"${hello}\\"??!`;\\n console.log(dareyou)\\n return dareyou\\n}","language":"deno","input_transforms":{"hello":{"type":"javascript","value":"${results.b}","expr":"`${results.b}`"}},"is_trigger":false}}],"iterator":{"type":"javascript","expr":"[\'foo\', \'bar\', \'baz\']"},"skip_failures":true,"parallel":false}}],"same_worker":false}',
E'{"modules":[{"id":"a","value":{"type":"forloopflow","modules":[{"id":"b","value":{"type":"rawscript","content":"export function main(world: string) {\\n const greet = `Hello ${world}!`;\\n console.log(greet)\\n return greet\\n}\\n","language":"deno","input_transforms":{"world":{"type":"javascript","expr":"flow_input.iter.value"}},"is_trigger":false}},{"id":"c","value":{"type":"rawscript","content":"export function main(hello: string) {\\n const dareyou = `Did you just say \\"${hello}\\"??!`;\\n console.log(dareyou)\\n return dareyou\\n}","language":"deno","input_transforms":{"hello":{"type":"javascript","value":"${results.b}","expr":"`${results.b}`"}},"is_trigger":false}}],"iterator":{"type":"javascript","expr":"[''foo'', ''bar'', ''baz'']"},"skip_failures":true,"parallel":false}}],"same_worker":false}',
'system'
);
INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES (
'test-workspace',
'',
'',
'f/system/hello_with_preprocessor',
'{1443253234253456}',
'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"foo":{"type":"string","description":"","default":""},"bar":{"type":"string","description":"","default":""}},"required":[],"type":"object","order":["foo","bar"]}',
E'{"modules":[{"id":"a","value":{"type":"rawscript","content":"export async function main(foo: string, bar: string) {\\n return ''Hello '' + foo + ''-'' + bar\\n}\\n","language":"deno","is_trigger":false,"input_transforms":{"bar":{"expr":"flow_input.bar","type":"javascript"},"foo":{"expr":"flow_input.foo","type":"javascript"}}}}],"preprocessor_module":{"id":"preprocessor","value":{"type":"rawscript","content":"export async function preprocessor(\\n\\twm_trigger: {\\n\\t\\tkind: ''http'' | ''email'' | ''webhook'' | ''websocket'' | ''kafka'' | ''nats'',\\n\\t\\thttp?: {\\n\\t\\t\\troute: string\\n\\t\\t\\tpath: string\\n\\t\\t\\tmethod: string\\n\\t\\t\\tparams: Record<string, string>\\n\\t\\t\\tquery: Record<string, string>\\n\\t\\t\\theaders: Record<string, string>\\n\\t\\t},\\n\\t\\twebsocket?: {\\n\\t\\t\\turl: string\\n\\t\\t},\\n\\t\\tkafka?: {\\n\\t\\t\\tbrokers: string[]\\n\\t\\t\\ttopic: string\\n\\t\\t\\tgroup_id: string\\n\\t\\t},\\n\\t\\tnats?: {\\n\\t\\t\\tservers: string[]\\n\\t\\t\\tsubject: string\\n\\t\\t\\theaders?: Record<string, string[]>\\n\\t\\t\\tstatus?: number\\n\\t\\t\\tdescription?: string\\n\\t\\t\\tlength: number\\n\\t\\t}\\n\\t},\\n) {\\n\\treturn {\\n\\t\\tfoo: ''bar'',\\n\\t\\tbar: ''baz''\\n\\t}\\n}","language":"deno","is_trigger":false,"input_transforms":{"wm_trigger":{"type":"static"}}}}}',
'system'
);
INSERT INTO public.flow_version(id, workspace_id, path, schema, value, created_by) VALUES (
1443253234253456,
'test-workspace',
'f/system/hello_with_preprocessor',
'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"foo":{"type":"string","description":"","default":""},"bar":{"type":"string","description":"","default":""}},"required":[],"type":"object","order":["foo","bar"]}',
E'{"modules":[{"id":"a","value":{"type":"rawscript","content":"export async function main(foo: string, bar: string) {\\n return ''Hello '' + foo + ''-'' + bar\\n}\\n","language":"deno","is_trigger":false,"input_transforms":{"bar":{"expr":"flow_input.bar","type":"javascript"},"foo":{"expr":"flow_input.foo","type":"javascript"}}}}],"preprocessor_module":{"id":"preprocessor","value":{"type":"rawscript","content":"export async function preprocessor(\\n\\twm_trigger: {\\n\\t\\tkind: ''http'' | ''email'' | ''webhook'' | ''websocket'' | ''kafka'' | ''nats'',\\n\\t\\thttp?: {\\n\\t\\t\\troute: string\\n\\t\\t\\tpath: string\\n\\t\\t\\tmethod: string\\n\\t\\t\\tparams: Record<string, string>\\n\\t\\t\\tquery: Record<string, string>\\n\\t\\t\\theaders: Record<string, string>\\n\\t\\t},\\n\\t\\twebsocket?: {\\n\\t\\t\\turl: string\\n\\t\\t},\\n\\t\\tkafka?: {\\n\\t\\t\\tbrokers: string[]\\n\\t\\t\\ttopic: string\\n\\t\\t\\tgroup_id: string\\n\\t\\t},\\n\\t\\tnats?: {\\n\\t\\t\\tservers: string[]\\n\\t\\t\\tsubject: string\\n\\t\\t\\theaders?: Record<string, string[]>\\n\\t\\t\\tstatus?: number\\n\\t\\t\\tdescription?: string\\n\\t\\t\\tlength: number\\n\\t\\t}\\n\\t},\\n) {\\n\\treturn {\\n\\t\\tfoo: ''bar'',\\n\\t\\tbar: ''baz''\\n\\t}\\n}","language":"deno","is_trigger":false,"input_transforms":{"wm_trigger":{"type":"static"}}}}}',
'system'
);
backend/tests/fixtures/result_format.sql
Vendored
+6
View File
@@ -1,3 +1,9 @@
INSERT INTO public.v2_job (
id, workspace_id, created_by, created_at, kind, script_lang
) VALUES (
'1eecb96a-c8b0-4a3d-b1b6-087878c55e41', 'test-workspace', 'test-user', '2023-01-01 00:00:00', 'script', 'postgresql'
);
INSERT INTO public.completed_job (
id, workspace_id, created_by, created_at, duration_ms, success, flow_status, result, job_kind, language
) VALUES (
backend/tests/worker.rs
+197 -1
View File
@@ -951,6 +951,21 @@ impl RunJob {
let r = completed_job(uuid, db).await;
r
}
/// push the job, spawn a worker, wait until the job is in completed_job
async fn run_until_complete_with<F: Future<Output = ()>>(
self,
db: &Pool<Postgres>,
port: u16,
test: impl Fn(Uuid) -> F,
) -> CompletedJob {
let uuid = self.push(db).await;
let listener = listen_for_completed_jobs(db).await;
test(uuid).await;
in_test_worker(db, listener.find(&uuid), port).await;
let r = completed_job(uuid, db).await;
r
}
}
async fn run_job_in_new_worker_until_complete(
@@ -3743,6 +3758,73 @@ async fn test_result_format(db: Pool<Postgres>) {
assert_eq!(result.get(), correct_result);
}
#[sqlx::test(fixtures("base"))]
async fn test_job_labels(db: Pool<Postgres>) {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await;
let port = server.addr.port();
let db = &db;
let test = |original_labels: &'static [&'static str]| async move {
let job = RunJob::from(JobPayload::RawFlow {
value: serde_json::from_value(json!({
"modules": [{
"id": "a",
"value": {
"type": "rawscript",
"content": r#"export function main(world: string) {
const greet = `Hello ${world}!`;
console.log(greet)
return { greet, wm_labels: ["yolo", "greet", "greet", world] };
}"#,
"language": "deno",
"input_transforms": {
"world": { "type": "javascript", "expr": "flow_input.world" }
}
}
}],
"schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"properties": { "world": { "type": "string" } },
"type": "object",
"order": [ "world" ]
}
}))
.unwrap(),
path: None,
restarted_from: None,
})
.arg("world", json!("you"))
.run_until_complete_with(&db, port, |id| async move {
sqlx::query!(
"UPDATE v2_job SET labels = $2 WHERE id = $1 AND $2::TEXT[] IS NOT NULL",
id,
original_labels as &[&str],
)
.execute(db)
.await
.unwrap();
})
.await;
let result = job.json_result().unwrap();
assert_eq!(result.get("greet"), Some(&json!("Hello you!")));
let labels = sqlx::query_scalar!("SELECT labels FROM v2_job WHERE id = $1", job.id)
.fetch_one(db)
.await
.unwrap();
let mut expected_labels = original_labels
.iter()
.chain(&["yolo", "greet", "you"])
.map(ToString::to_string)
.collect::<Vec<_>>();
expected_labels.sort();
assert_eq!(labels, Some(expected_labels));
};
test(&[]).await;
test(&["z", "a", "x"]).await;
}
async fn test_for_versions<F: Future<Output = ()>>(
version_flags: impl Iterator<Item = Arc<RwLock<bool>>>,
test: impl Fn() -> F,
@@ -3799,6 +3881,53 @@ mod job_payload {
test_for_versions(VERSION_FLAGS.iter().cloned(), test).await;
}
#[sqlx::test(fixtures("base", "hello"))]
async fn test_script_hash_payload_with_preprocessor(db: Pool<Postgres>) {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await;
let port = server.addr.port();
let test = || async {
let db = &db;
let job = RunJob::from(JobPayload::ScriptHash {
hash: ScriptHash(123413),
path: "f/system/hello_with_preprocessor".to_string(),
custom_concurrency_key: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
dedicated_worker: None,
language: ScriptLang::Deno,
priority: None,
apply_preprocessor: true,
})
.run_until_complete_with(db, port, |id| async move {
let job = sqlx::query!("SELECT preprocessed FROM v2_job WHERE id = $1", id)
.fetch_one(db)
.await
.unwrap();
assert_eq!(job.preprocessed, Some(false));
})
.await;
let args = job.args.as_ref().unwrap();
assert_eq!(args.get("foo"), Some(&json!("bar")));
assert_eq!(args.get("bar"), Some(&json!("baz")));
// TODO: remove this check on v2 phase 4
assert_eq!(
job.flow_status.as_ref().unwrap().get("_metadata"),
Some(&json!({"preprocessed_args": true}))
);
assert_eq!(job.json_result().unwrap(), json!("Hello bar baz"));
let job = sqlx::query!("SELECT preprocessed FROM v2_job WHERE id = $1", job.id)
.fetch_one(db)
.await
.unwrap();
assert_eq!(job.preprocessed, Some(true));
};
test_for_versions(VERSION_FLAGS.iter().cloned(), test).await;
}
#[sqlx::test(fixtures("base", "hello"))]
async fn test_flow_script_payload(db: Pool<Postgres>) {
initialize_tracing().await;
@@ -4081,7 +4210,7 @@ mod job_payload {
let result = RunJob::from(JobPayload::Flow {
path: "f/system/hello_with_nodes_flow".to_string(),
dedicated_worker: None,
apply_preprocessor: true,
apply_preprocessor: false,
})
.run_until_complete(&db, port)
.await
@@ -4113,6 +4242,73 @@ mod job_payload {
test_for_versions(VERSION_FLAGS.iter().cloned(), test).await;
}
#[sqlx::test(fixtures("base", "hello"))]
async fn test_flow_payload_with_preprocessor(db: Pool<Postgres>) {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await;
let port = server.addr.port();
let db = &db;
let test = || async {
let job = RunJob::from(JobPayload::Flow {
path: "f/system/hello_with_preprocessor".to_string(),
dedicated_worker: None,
apply_preprocessor: true,
})
.run_until_complete_with(db, port, |id| async move {
let job = sqlx::query!("SELECT preprocessed FROM v2_job WHERE id = $1", id)
.fetch_one(db)
.await
.unwrap();
assert_eq!(job.preprocessed, Some(false));
})
.await;
let args = job.args.as_ref().unwrap();
let flow_status = job.flow_status.as_ref().unwrap();
assert_eq!(args.get("foo"), Some(&json!("bar")));
assert_eq!(args.get("bar"), Some(&json!("baz")));
assert_eq!(job.json_result().unwrap(), json!("Hello bar-baz"));
let job = sqlx::query!("SELECT preprocessed FROM v2_job WHERE id = $1", job.id)
.fetch_one(db)
.await
.unwrap();
assert_eq!(job.preprocessed, Some(true));
let flow_status = serde_json::from_value::<FlowStatus>(flow_status.clone()).unwrap();
let FlowStatusModule::Success { job, .. } = flow_status.preprocessor_module.unwrap()
else {
panic!("Expected a success preprocessor module");
};
let pp_id = job;
let job = sqlx::query!(
"SELECT preprocessed, script_entrypoint_override FROM v2_job WHERE id = $1",
pp_id
)
.fetch_one(db)
.await
.unwrap();
assert_eq!(job.preprocessed, Some(true));
assert_eq!(
job.script_entrypoint_override.as_deref(),
Some("preprocessor")
);
};
// Test the not "lite" flow.
test_for_versions(VERSION_FLAGS.iter().cloned(), test).await;
// Deploy the flow to produce the "lite" version.
let _ = RunJob::from(JobPayload::FlowDependencies {
path: "f/system/hello_with_preprocessor".to_string(),
dedicated_worker: None,
version: 1443253234253456,
})
.run_until_complete(db, port)
.await
.json_result()
.unwrap();
// Test the "lite" flow.
test_for_versions(VERSION_FLAGS.iter().cloned(), test).await;
}
#[sqlx::test(fixtures("base", "hello"))]
async fn test_restarted_flow_payload(db: Pool<Postgres>) {
initialize_tracing().await;
backend/windmill-api/src/db.rs
+22 -11
View File
@@ -405,60 +405,60 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
.await?;
});
run_windmill_migration!("fix_job_completed_index_4", &db, {
run_windmill_migration!("fix_job_index_1", &db, {
let migration_job_name = "fix_job_completed_index_4";
let mut i = 1;
tracing::info!("step {i} of {migration_job_name} migration");
sqlx::query("create index concurrently if not exists ix_completed_job_workspace_id_created_at_new_3 ON completed_job (workspace_id, created_at DESC)")
sqlx::query!("create index concurrently if not exists ix_job_workspace_id_created_at_new_3 ON v2_job (workspace_id, created_at DESC)")
.execute(db)
.await?;
i += 1;
tracing::info!("step {i} of {migration_job_name} migration");
sqlx::query("create index concurrently if not exists ix_completed_job_workspace_id_created_at_new_8 ON completed_job (workspace_id, created_at DESC) where job_kind in ('deploymentcallback') AND parent_job IS NULL")
sqlx::query!("create index concurrently if not exists ix_job_workspace_id_created_at_new_8 ON v2_job (workspace_id, created_at DESC) where kind in ('deploymentcallback') AND parent_job IS NULL")
.execute(db)
.await?;
i += 1;
tracing::info!("step {i} of {migration_job_name} migration");
sqlx::query("create index concurrently if not exists ix_completed_job_workspace_id_created_at_new_9 ON completed_job (workspace_id, created_at DESC) where job_kind in ('dependencies', 'flowdependencies', 'appdependencies') AND parent_job IS NULL")
sqlx::query!("create index concurrently if not exists ix_job_workspace_id_created_at_new_9 ON v2_job (workspace_id, created_at DESC) where kind in ('dependencies', 'flowdependencies', 'appdependencies') AND parent_job IS NULL")
.execute(db)
.await?;
i += 1;
tracing::info!("step {i} of {migration_job_name} migration");
sqlx::query("create index concurrently if not exists ix_completed_job_workspace_id_created_at_new_5 ON completed_job (workspace_id, created_at DESC) where job_kind in ('preview', 'flowpreview') AND parent_job IS NULL")
sqlx::query!("create index concurrently if not exists ix_job_workspace_id_created_at_new_5 ON v2_job (workspace_id, created_at DESC) where kind in ('preview', 'flowpreview') AND parent_job IS NULL")
.execute(db)
.await?;
i += 1;
tracing::info!("step {i} of {migration_job_name} migration");
sqlx::query("create index concurrently if not exists ix_completed_job_workspace_id_created_at_new_6 ON completed_job (workspace_id, created_at DESC) where job_kind in ('script', 'flow') AND parent_job IS NULL")
sqlx::query!("create index concurrently if not exists ix_job_workspace_id_created_at_new_6 ON v2_job (workspace_id, created_at DESC) where kind in ('script', 'flow') AND parent_job IS NULL")
.execute(db)
.await?;
i += 1;
tracing::info!("step {i} of {migration_job_name} migration");
sqlx::query("create index concurrently if not exists ix_completed_job_workspace_id_created_at_new_7 ON completed_job (workspace_id, success, created_at DESC) where job_kind in ('script', 'flow') AND parent_job IS NULL")
sqlx::query!("create index concurrently if not exists ix_job_workspace_id_created_at_new_7 ON v2_job (workspace_id, created_at DESC) where kind in ('script', 'flow') AND parent_job IS NULL")
.execute(db)
.await?;
i += 1;
tracing::info!("step {i} of {migration_job_name} migration");
sqlx::query("create index concurrently if not exists ix_completed_job_workspace_id_started_at_new_2 ON completed_job (workspace_id, started_at DESC)")
sqlx::query!("create index concurrently if not exists ix_completed_job_workspace_id_started_at_new_2 ON v2_job_completed (workspace_id, started_at DESC)")
.execute(db)
.await?;
i += 1;
tracing::info!("step {i} of {migration_job_name} migration");
sqlx::query("create index concurrently if not exists root_job_index_by_path_2 ON completed_job (workspace_id, script_path, created_at desc) WHERE parent_job IS NULL")
sqlx::query!("create index concurrently if not exists root_job_index_by_path_2 ON v2_job (workspace_id, runnable_path, created_at desc) WHERE parent_job IS NULL")
.execute(db)
.await?;
i += 1;
tracing::info!("step {i} of {migration_job_name} migration");
sqlx::query("create index concurrently if not exists ix_completed_job_created_at ON completed_job (created_at DESC)")
sqlx::query!("create index concurrently if not exists ix_job_created_at ON v2_job (created_at DESC)")
.execute(db)
.await?;
@@ -492,10 +492,21 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
.execute(db)
.await?;
sqlx::query!(
"CREATE INDEX CONCURRENTLY labeled_jobs_on_jobs ON completed_job USING GIN ((result -> 'wm_labels')) WHERE result ? 'wm_labels'"
"CREATE INDEX CONCURRENTLY labeled_jobs_on_jobs ON v2_job_completed USING GIN ((result -> 'wm_labels')) WHERE result ? 'wm_labels'"
).execute(db).await?;
});
run_windmill_migration!("v2_labeled_jobs_index", &db, {
tracing::info!("Special migration to add index concurrently on job labels");
sqlx::query!(
"CREATE INDEX CONCURRENTLY ix_v2_job_labels ON v2_job
USING GIN (labels)
WHERE labels IS NOT NULL"
)
.execute(db)
.await?;
});
Ok(())
}
backend/windmill-api/src/jobs.rs
+14 -13
View File
@@ -602,7 +602,7 @@ async fn get_flow_job_debug_info(
let mut job_ids = vec![];
let jobs_with_root = sqlx::query_scalar!(
"SELECT id FROM queue WHERE workspace_id = $1 and root_job = $2",
"SELECT id AS \"id!\" FROM queue WHERE workspace_id = $1 and root_job = $2",
&w_id,
&id,
)
@@ -1009,7 +1009,7 @@ async fn get_job_logs(
.flatten();
let record = sqlx::query!(
"SELECT created_by, CONCAT(coalesce(completed_job.logs, ''), coalesce(job_logs.logs, '')) as logs, job_logs.log_offset, job_logs.log_file_index
"SELECT created_by AS \"created_by!\", CONCAT(coalesce(completed_job.logs, ''), coalesce(job_logs.logs, '')) as logs, job_logs.log_offset, job_logs.log_file_index
FROM completed_job
LEFT JOIN job_logs ON job_logs.job_id = completed_job.id
WHERE completed_job.id = $1 AND completed_job.workspace_id = $2 AND ($3::text[] IS NULL OR completed_job.tag = ANY($3))",
@@ -1046,7 +1046,7 @@ async fn get_job_logs(
Ok(content_plain(Body::from(logs)))
} else {
let text = sqlx::query!(
"SELECT created_by, CONCAT(coalesce(queue.logs, ''), coalesce(job_logs.logs, '')) as logs, coalesce(job_logs.log_offset, 0) as log_offset, job_logs.log_file_index
"SELECT created_by AS \"created_by!\", CONCAT(coalesce(queue.logs, ''), coalesce(job_logs.logs, '')) as logs, coalesce(job_logs.log_offset, 0) as log_offset, job_logs.log_file_index
FROM queue
LEFT JOIN job_logs ON job_logs.job_id = queue.id
WHERE queue.id = $1 AND queue.workspace_id = $2 AND ($3::text[] IS NULL OR queue.tag = ANY($3))",
@@ -1551,7 +1551,7 @@ async fn cancel_jobs(
, tag
, priority FROM queue
WHERE id = any($2) AND running = false AND parent_job IS NULL AND workspace_id = $3 AND schedule_path IS NULL FOR UPDATE SKIP LOCKED
ON CONFLICT (id) DO NOTHING RETURNING id", username, &jobs, w_id, serde_json::json!({"error": { "message": format!("Job canceled: cancel all by {username}"), "name": "Canceled", "reason": "cancel all", "canceler": username}}))
ON CONFLICT (id) DO NOTHING RETURNING id AS \"id!\"", username, &jobs, w_id, serde_json::json!({"error": { "message": format!("Job canceled: cancel all by {username}"), "name": "Canceled", "reason": "cancel all", "canceler": username}}))
.fetch_all(&mut *tx)
.await?.into_iter().map(|x| x.id).collect::<Vec<Uuid>>();
@@ -1624,7 +1624,7 @@ async fn cancel_selection(
let mut tx = user_db.begin(&authed).await?;
let tags = get_scope_tags(&authed).map(|v| v.iter().map(|s| s.to_string()).collect_vec());
let jobs_to_cancel = sqlx::query_scalar!(
"SELECT id FROM queue WHERE id = ANY($1) AND schedule_path IS NULL AND ($2::text[] IS NULL OR tag = ANY($2))",
"SELECT id AS \"id!\" FROM queue WHERE id = ANY($1) AND schedule_path IS NULL AND ($2::text[] IS NULL OR tag = ANY($2))",
&jobs,
tags.as_ref().map(|v| v.as_slice())
)
@@ -2096,7 +2096,7 @@ async fn get_suspended_parent_flow_info(job_id: Uuid, db: &DB) -> error::Result<
let flow = sqlx::query_as!(
FlowInfo,
r#"
SELECT id, flow_status, suspend, script_path
SELECT id AS "id!", flow_status, suspend AS "suspend!", script_path
FROM queue
WHERE id = ( SELECT parent_job FROM queue WHERE id = $1 UNION ALL SELECT parent_job FROM completed_job WHERE id = $1)
FOR UPDATE
@@ -2116,7 +2116,7 @@ async fn get_suspended_flow_info<'c>(
let flow = sqlx::query_as!(
FlowInfo,
r#"
SELECT id, flow_status, suspend, script_path
SELECT id AS "id!", flow_status, suspend AS "suspend!", script_path
FROM queue
WHERE id = $1
"#,
@@ -4860,16 +4860,17 @@ async fn add_batch_jobs(
let uuids = sqlx::query_scalar!(
r#"WITH uuid_table as (
select gen_random_uuid() as uuid from generate_series(1, $5)
select gen_random_uuid() as uuid from generate_series(1, $6)
)
INSERT INTO job
(id, workspace_id, raw_code, raw_lock, raw_flow)
(SELECT uuid, $1, $2, $3, $4 FROM uuid_table)
RETURNING id"#,
(id, workspace_id, raw_code, raw_lock, raw_flow, tag)
(SELECT uuid, $1, $2, $3, $4, $5 FROM uuid_table)
RETURNING id AS "id!""#,
w_id,
raw_code,
raw_lock,
raw_flow.map(sqlx::types::Json) as Option<sqlx::types::Json<FlowValue>>,
tag,
n
)
.fetch_all(&mut *tx)
@@ -4882,7 +4883,7 @@ async fn add_batch_jobs(
INSERT INTO queue
(id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id, concurrent_limit, concurrency_time_window_s, timeout, flow_status)
(SELECT uuid, $1, $2, $3, $4, ('{ "uuid": "' || uuid || '" }')::jsonb, $5, $6, $7, $8, $9, $10, $12, $13, $14, $15 FROM uuid_table)
RETURNING id"#,
RETURNING id AS "id!""#,
hash.map(|h| h.0),
path,
job_kind.clone() as JobKind,
@@ -5715,7 +5716,7 @@ async fn get_completed_job_result_maybe(
.into_response())
} else if get_started.is_some_and(|x| x) {
let started = sqlx::query_scalar!(
"SELECT running FROM queue WHERE id = $1 AND workspace_id = $2",
"SELECT running AS \"running!\" FROM queue WHERE id = $1 AND workspace_id = $2",
id,
w_id
)
backend/windmill-api/src/slack_approvals.rs
+2 -2
View File
@@ -977,11 +977,11 @@ async fn get_modal_blocks(
let (job_kind, script_hash, raw_flow, parent_job_id, created_at, created_by, script_path, args) = sqlx::query!(
"SELECT
queue.job_kind AS \"job_kind: JobKind\",
queue.job_kind AS \"job_kind!: JobKind\",
queue.script_hash AS \"script_hash: ScriptHash\",
queue.raw_flow AS \"raw_flow: sqlx::types::Json<Box<RawValue>>\",
completed_job.parent_job AS \"parent_job: Uuid\",
completed_job.created_at AS \"created_at: chrono::NaiveDateTime\",
completed_job.created_at AS \"created_at!: chrono::NaiveDateTime\",
completed_job.created_by AS \"created_by!\",
queue.script_path,
queue.args AS \"args: sqlx::types::Json<Box<RawValue>>\"
backend/windmill-queue/src/jobs.rs
+7 -6
View File
@@ -173,7 +173,7 @@ pub async fn cancel_single_job<'c>(
});
} else {
let id: Option<Uuid> = sqlx::query_scalar!(
"UPDATE queue SET canceled = true, canceled_by = $1, canceled_reason = $2, scheduled_for = now(), suspend = 0 WHERE id = $3 AND workspace_id = $4 AND (canceled = false OR canceled_reason != $2) RETURNING id",
"UPDATE queue SET canceled = true, canceled_by = $1, canceled_reason = $2, scheduled_for = now(), suspend = 0 WHERE id = $3 AND workspace_id = $4 AND (canceled = false OR canceled_reason != $2) RETURNING id AS \"id!\"",
username,
reason,
job_running.id,
@@ -249,7 +249,7 @@ pub async fn cancel_job<'c>(
while !jobs.is_empty() {
let p_job = jobs.pop();
let new_jobs = sqlx::query_scalar!(
"SELECT id FROM queue WHERE parent_job = $1 AND workspace_id = $2",
"SELECT id AS \"id!\" FROM queue WHERE parent_job = $1 AND workspace_id = $2",
p_job,
w_id
)
@@ -609,7 +609,7 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
)
VALUES ($1, $2, $3, $4, $5, COALESCE($6, now()), COALESCE($30::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE($6, now()))))*1000), $7, $8, $9,\
$10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29)
ON CONFLICT (id) DO UPDATE SET success = $7, result = $11 RETURNING duration_ms",
ON CONFLICT (id) DO UPDATE SET success = $7, result = $11 RETURNING duration_ms AS \"duration_ms!\"",
queued_job.workspace_id,
queued_job.id,
queued_job.parent_job,
@@ -2309,7 +2309,8 @@ pub async fn get_result_and_success_by_id_from_flow(
let success = match &job_result {
JobResult::SingleJob(job_id) => {
sqlx::query_scalar!(
"SELECT success FROM completed_job WHERE id = $1 AND workspace_id = $2",
"SELECT success AS \"success!\"
FROM completed_job WHERE id = $1 AND workspace_id = $2",
job_id,
w_id
)
@@ -3731,7 +3732,7 @@ pub async fn push<'c, 'd>(
visible_to_owner, root_job, tag, concurrent_limit, concurrency_time_window_s, timeout, \
flow_step_id, cache_ttl, priority, last_ping)
VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, now()), $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, NULL) \
RETURNING id",
RETURNING id AS \"id!\"",
workspace_id,
job_id,
is_running,
@@ -3916,7 +3917,7 @@ async fn restarted_flows_resolution(
let row = sqlx::query!(
"SELECT
script_path, script_hash AS \"script_hash: ScriptHash\",
job_kind AS \"job_kind: JobKind\",
job_kind AS \"job_kind!: JobKind\",
flow_status AS \"flow_status: Json<Box<RawValue>>\",
raw_flow AS \"raw_flow: Json<Box<RawValue>>\"
FROM completed_job WHERE id = $1 and workspace_id = $2",
backend/windmill-worker/src/worker_flow.rs
+3 -3
View File
@@ -222,7 +222,7 @@ pub async fn update_flow_status_after_job_completion_internal(
let (job_kind, script_hash, old_status, raw_flow) = sqlx::query!(
"SELECT
job_kind AS \"job_kind: JobKind\",
job_kind AS \"job_kind!: JobKind\",
script_hash AS \"script_hash: ScriptHash\",
flow_status AS \"flow_status!: Json<Box<RawValue>>\",
raw_flow AS \"raw_flow: Json<Box<RawValue>>\"
@@ -552,7 +552,7 @@ pub async fn update_flow_status_after_job_completion_internal(
let new_status = if skip_loop_failures
|| sqlx::query_scalar!(
"SELECT success FROM completed_job WHERE id = ANY($1)",
"SELECT success AS \"success!\" FROM completed_job WHERE id = ANY($1)",
jobs.as_slice()
)
.fetch_all(&mut *tx)
@@ -1687,7 +1687,7 @@ async fn push_next_flow_job(
.await?;
if no_flow_overlap {
let overlapping = sqlx::query_scalar!(
"SELECT id FROM queue WHERE schedule_path = $1 AND workspace_id = $2 AND id != $3 AND running = true",
"SELECT id AS \"id!\" FROM queue WHERE schedule_path = $1 AND workspace_id = $2 AND id != $3 AND running = true",
flow_job.schedule_path.as_ref().unwrap(),
flow_job.workspace_id.as_str(),
flow_job.id
backend/windmill-worker/src/worker_lockfiles.rs
+22 -18
View File
@@ -633,15 +633,17 @@ pub async fn handle_flow_dependency_job(
let new_flow_value = Json(serde_json::value::to_raw_value(&flow).map_err(to_anyhow)?);
// Re-check cancellation to ensure we don't accidentally override a flow.
if sqlx::query_scalar!("SELECT canceled FROM queue WHERE id = $1", job.id)
.fetch_optional(db)
.await
.map(|v| Some(true) == v)
.unwrap_or_else(|err| {
tracing::error!(%job.id, %err, "error checking cancellation for job {0}: {err}", job.id);
false
})
{
if sqlx::query_scalar!(
"SELECT canceled AS \"canceled!\" FROM queue WHERE id = $1",
job.id
)
.fetch_optional(db)
.await
.map(|v| Some(true) == v)
.unwrap_or_else(|err| {
tracing::error!(%job.id, %err, "error checking cancellation for job {0}: {err}", job.id);
false
}) {
return Ok(to_raw_value_owned(json!({
"status": "Flow lock generation was canceled",
})));
@@ -1525,15 +1527,17 @@ pub async fn handle_app_dependency_job(
.await?;
// Re-check cancelation to ensure we don't accidentially override an app.
if sqlx::query_scalar!("SELECT canceled FROM queue WHERE id = $1", job.id)
.fetch_optional(db)
.await
.map(|v| Some(true) == v)
.unwrap_or_else(|err| {
tracing::error!(%job.id, %err, "error checking cancelation for job {0}: {err}", job.id);
false
})
{
if sqlx::query_scalar!(
"SELECT canceled AS \"canceled!\" FROM queue WHERE id = $1",
job.id
)
.fetch_optional(db)
.await
.map(|v| Some(true) == v)
.unwrap_or_else(|err| {
tracing::error!(%job.id, %err, "error checking cancelation for job {0}: {err}", job.id);
false
}) {
return Ok(());
}