diff --git a/.agents/skills/local-review-codex/SKILL.md b/.agents/skills/local-review-codex/SKILL.md index ec1277fe3a..3483453f66 100644 --- a/.agents/skills/local-review-codex/SKILL.md +++ b/.agents/skills/local-review-codex/SKILL.md @@ -1,6 +1,6 @@ --- name: local-review-codex -description: Run the CI Codex PR review locally against this branch's unpushed work (committed + uncommitted) before pushing. Same policy and reasoning effort as the codex-pr-review GitHub action, on a newer model. +description: Run the CI Codex PR review locally against this branch's unpushed work (committed + uncommitted) before pushing. Same policy, model and reasoning effort as the codex-pr-review GitHub action. --- # Local Codex Review (pre-push) @@ -11,18 +11,18 @@ before the PR exists. Use this before `git push` on a non-trivial change. **Correspondence with CI** — identical: - Policy: `REVIEW.md` (severity triage, public-surface checklist, AGENTS.md compliance, test coverage). +- Model: `gpt-6.1-sol`. - Reasoning effort: `model_reasoning_effort="xhigh"`. - Output: markdown starting with `## Codex Review`, findings tagged P0 / P1 / P2 with file:line. **Differences from CI** — local-only: -- Model is `gpt-6-astra`; CI stays on `gpt-5.6-sol`. Not an oversight to reconcile: `gpt-6-astra` is confirmed on the ChatGPT auth `codex login` uses locally, while CI authenticates with `OPENAI_API_KEY` (`codex-pr-review.yml` prefers it over `CODEX_AUTH_JSON`) and that tier is unverified for the model. Move CI once API access is confirmed, or once CI switches to `CODEX_AUTH_JSON`. - Scope is the current branch vs `main` at the merge-base, **including uncommitted changes** (CI reviews a pushed PR diff). - Sandbox is `read-only` (CI uses `danger-full-access` on an ephemeral runner). Codex reads the diff and files but cannot modify your working tree. - Fresh context is inherent: `codex exec` is a separate cold process, so it does not anchor on the current chat session — the same reason `local-review` insists on a subagent. ## Prerequisites -- `codex` CLI **>= 0.153.4** installed and authed via `codex login` (an `OPENAI_API_KEY` in the environment takes priority and may not reach `gpt-6-astra` — see the model note above). Older CLIs reject the model with "requires a newer version of Codex"; `run.sh` checks the version up front. Upgrade with `npm install --global @openai/codex@0.153.4` (may need `sudo` for a global install). This matches the pin in `.github/workflows/codex-pr-review.yml` — the CLI version is the same on both sides, only the model differs. +- `codex` CLI **>= 0.159.3** installed and authed via `codex login` or an `OPENAI_API_KEY` in the environment (which takes priority). Older CLIs reject the model with "requires a newer version of Codex"; `run.sh` checks the version up front. Upgrade with `npm install --global @openai/codex@0.159.3` (may need `sudo` for a global install). This matches the pin in `.github/workflows/codex-pr-review.yml` — the CLI version is the same on both sides. - `git fetch` the base ref if it's stale, so the merge-base is accurate. ## Run diff --git a/.agents/skills/local-review-codex/run.sh b/.agents/skills/local-review-codex/run.sh index 948d3820eb..e1fef4ce8e 100755 --- a/.agents/skills/local-review-codex/run.sh +++ b/.agents/skills/local-review-codex/run.sh @@ -1,17 +1,14 @@ #!/usr/bin/env bash # Local Codex review — mirrors the .github/workflows/codex-pr-review.yml CI job, # but scoped to this branch's unpushed work (committed + uncommitted) so you can -# review before pushing. Same policy (REVIEW.md) and reasoning effort (xhigh) as CI. -# -# The model deliberately differs from CI: gpt-6-astra is confirmed available on the -# ChatGPT auth `codex login` uses here, but CI authenticates with OPENAI_API_KEY and -# that tier is unverified for it, so codex-pr-review.yml stays on gpt-5.6-sol. +# review before pushing. Same policy (REVIEW.md), model and reasoning effort (xhigh) +# as CI. # # Usage: run.sh [BASE_REF] (BASE_REF defaults to "main") set -euo pipefail -MODEL="gpt-6-astra" -CODEX_MIN="0.153.4" +MODEL="gpt-6.1-sol" +CODEX_MIN="0.159.3" BASE_REF="${1:-main}" REPO_ROOT="$(git rev-parse --show-toplevel)" diff --git a/.github/workflows/ai-evals-test.yml b/.github/workflows/ai-evals-test.yml index 573bc77df9..84b488ae46 100644 --- a/.github/workflows/ai-evals-test.yml +++ b/.github/workflows/ai-evals-test.yml @@ -147,7 +147,7 @@ jobs: mkdir -p results # One cheap model per provider (anthropic/openai/googleai/deepseek). fail=0 - for m in haiku 4o gemini-3-flash-preview deepseek-v4-flash; do + for m in haiku gpt-6-luna gemini-3.8-flash deepseek-flash; do echo "::group::global-test1-script-create ($m)" if ! bun run cli -- run global global-test1-script-create \ --model "$m" --execution-only --output "$PWD/results/ci-$m.json"; then diff --git a/.github/workflows/claude-plan.yml b/.github/workflows/claude-plan.yml index bc81801417..b3798c76f5 100644 --- a/.github/workflows/claude-plan.yml +++ b/.github/workflows/claude-plan.yml @@ -49,7 +49,7 @@ jobs: allowed_bots: 'windmill-internal-app[bot]' trigger_phrase: '/plan' claude_args: | - --model claude-opus-5 + --model claude-opus-5-5 --system-prompt "# Claude Planning Mode You are operating in PLANNING MODE ONLY. Your role is to create detailed, structured plans without making any code changes. diff --git a/.github/workflows/claude.yml b/.github/workflows/claude.yml index 5e9aab00d3..5250c5a04b 100644 --- a/.github/workflows/claude.yml +++ b/.github/workflows/claude.yml @@ -93,4 +93,4 @@ jobs: } claude_args: | --allowedTools "Bash,WebFetch,WebSearch" - --model claude-opus-5 + --model claude-opus-5-5 diff --git a/.github/workflows/codex-pr-review.yml b/.github/workflows/codex-pr-review.yml index 55036776c6..2d70ad7f7a 100644 --- a/.github/workflows/codex-pr-review.yml +++ b/.github/workflows/codex-pr-review.yml @@ -222,7 +222,7 @@ jobs: - name: Install Codex CLI if: steps.codex_config.outputs.enabled == 'true' && steps.pr.outputs.skip != 'true' - run: npm install --global @openai/codex@0.153.4 + run: npm install --global @openai/codex@0.159.3 - name: Configure Codex auth if: steps.codex_config.outputs.enabled == 'true' && steps.pr.outputs.skip != 'true' @@ -359,7 +359,7 @@ jobs: # e.g. a GitHub Action's index.js, which then runs with our credentials. codex exec \ -C "$GITHUB_WORKSPACE" \ - -m gpt-5.6-sol \ + -m gpt-6.1-sol \ -c 'model_reasoning_effort="xhigh"' \ -s "$SANDBOX_MODE" \ -o "$RUNNER_TEMP/codex-final-message.md" \ diff --git a/.github/workflows/pr-ready-review.yml b/.github/workflows/pr-ready-review.yml index 5e5c6a8c61..90714cba53 100644 --- a/.github/workflows/pr-ready-review.yml +++ b/.github/workflows/pr-ready-review.yml @@ -208,4 +208,4 @@ jobs: ${{ env.REVIEW_PROMPT }} claude_args: | --allowedTools "mcp__github_inline_comment__create_inline_comment,Bash(gh pr comment:*),Bash(gh pr diff:*),Bash(gh pr view:*)" - --model claude-opus-5 + --model claude-opus-5-5 diff --git a/.github/workflows/publish_extra.yml b/.github/workflows/publish_extra.yml index 863c850266..c33d806183 100644 --- a/.github/workflows/publish_extra.yml +++ b/.github/workflows/publish_extra.yml @@ -57,7 +57,7 @@ jobs: uses: docker/setup-buildx-action@v3 - name: Build test image - uses: docker/build-push-action@v5 + uses: docker/build-push-action@v7 with: context: . file: ./docker/DockerfileExtra diff --git a/AGENTS.md b/AGENTS.md index 54b5953748..8451a0a600 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -49,7 +49,7 @@ Open-source platform for internal tools, workflows, API integrations, background - **Backend patterns**: use the `rust-backend` skill when writing Rust code - **Frontend patterns**: use the `svelte-frontend` skill when writing Svelte code. Do NOT edit svelte files unless you have read that skill. - **Frontend UUIDs**: do not call `crypto.randomUUID()` in frontend code. Import `randomUUID` from `$lib/utils/uuid` instead. -- **Code review**: review the current PR or branch against the shared review policy in `REVIEW.md` (severity triage, public-surface checklist, AGENTS.md compliance, test-coverage assessment). The skill at `.agents/skills/local-review/SKILL.md` orchestrates it. All three CLIs auto-discover the same SKILL — Claude reads `.claude/skills/` (symlinked to the canonical `.agents/skills/` file), Codex and Pi read `.agents/skills/` directly. Invoke with `/local-review` in Claude Code, `$local-review` (or `/skills` selector) in Codex, or `pi --skill local-review` / `/skill:local-review` in Pi. For a Codex-driven pass that mirrors the `codex-pr-review` GitHub action against your unpushed work (committed + uncommitted) before you push, use `/local-review-codex` (`.agents/skills/local-review-codex/`) — same `REVIEW.md` policy and `xhigh` reasoning, on `gpt-6-astra` rather than the action's `gpt-5.6-sol`; requires the `codex` CLI >= 0.153.4. +- **Code review**: review the current PR or branch against the shared review policy in `REVIEW.md` (severity triage, public-surface checklist, AGENTS.md compliance, test-coverage assessment). The skill at `.agents/skills/local-review/SKILL.md` orchestrates it. All three CLIs auto-discover the same SKILL — Claude reads `.claude/skills/` (symlinked to the canonical `.agents/skills/` file), Codex and Pi read `.agents/skills/` directly. Invoke with `/local-review` in Claude Code, `$local-review` (or `/skills` selector) in Codex, or `pi --skill local-review` / `/skill:local-review` in Pi. For a Codex-driven pass that mirrors the `codex-pr-review` GitHub action against your unpushed work (committed + uncommitted) before you push, use `/local-review-codex` (`.agents/skills/local-review-codex/`) — same `REVIEW.md` policy, `gpt-6.1-sol` model and `xhigh` reasoning as the action; requires the `codex` CLI >= 0.159.3. - **Domain guides**: `.claude/skills/native-trigger/` - **Brand/UI guidelines**: `frontend/brand-guidelines.md` - **Domain vocabulary**: `CONTEXT.md` — the words this codebase uses for its own concepts (step, step setting, trigger step, …). Name things the way it does. diff --git a/ai_evals/core/models.test.ts b/ai_evals/core/models.test.ts index ba53c24592..d51d61d224 100644 --- a/ai_evals/core/models.test.ts +++ b/ai_evals/core/models.test.ts @@ -34,6 +34,10 @@ describe("resolveEvalModel", () => { it("supports DeepSeek aliases for frontend evals", () => { expect(resolveEvalModel("flow", "deepseek").frontend).toEqual({ + provider: "deepseek", + model: "deepseek-flash", + }); + expect(resolveEvalModel("flow", "deepseek-v4-flash").frontend).toEqual({ provider: "deepseek", model: "deepseek-v4-flash", }); diff --git a/ai_evals/core/models.ts b/ai_evals/core/models.ts index 1580cf5b0b..06689c36e1 100644 --- a/ai_evals/core/models.ts +++ b/ai_evals/core/models.ts @@ -192,12 +192,21 @@ export const EVAL_MODELS: EvalModelSpec[] = [ { id: "deepseek-v4-flash", label: "DeepSeek V4 Flash", - aliases: ["deepseek", "deepseek-v4", "deepseek-v4-flash"], + aliases: ["deepseek-v4", "deepseek-v4-flash"], frontend: { provider: "deepseek", model: "deepseek-v4-flash", }, }, + { + id: "deepseek-flash", + label: "DeepSeek V4.1 Flash", + aliases: ["deepseek", "deepseek-flash", "deepseek-v4.1-flash"], + frontend: { + provider: "deepseek", + model: "deepseek-flash", + }, + }, { id: "deepseek-v4-pro", label: "DeepSeek V4 Pro", diff --git a/backend/.sqlx/query-11b5d006acf040b2461b9539863b1e511c23e1a5f9be97da066a6808a49f2092.json b/backend/.sqlx/query-0a18205b64e68529b7f90bc1c5db2cbea9ee5eaa962970b65a92597fd5e62707.json similarity index 52% rename from backend/.sqlx/query-11b5d006acf040b2461b9539863b1e511c23e1a5f9be97da066a6808a49f2092.json rename to backend/.sqlx/query-0a18205b64e68529b7f90bc1c5db2cbea9ee5eaa962970b65a92597fd5e62707.json index bc821f2b7b..a3707d28c5 100644 --- a/backend/.sqlx/query-11b5d006acf040b2461b9539863b1e511c23e1a5f9be97da066a6808a49f2092.json +++ b/backend/.sqlx/query-0a18205b64e68529b7f90bc1c5db2cbea9ee5eaa962970b65a92597fd5e62707.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n WITH job_info AS (\n SELECT id, kind::text AS kind, parent_job\n FROM v2_job\n WHERE id = $1 AND workspace_id = $2\n )\n SELECT\n q.id AS \"id!\",\n s.flow_status,\n q.suspend AS \"suspend!\",\n j.runnable_path AS script_path,\n j.permissioned_as_email AS email,\n (ji.kind IN ('flow', 'flowpreview', 'singlestepflow')) AS \"is_flow_level!\",\n (ji.kind NOT IN ('flow', 'flowpreview', 'singlestepflow') AND q.id = ji.id) AS \"is_wac!\"\n FROM job_info ji\n JOIN v2_job_queue q ON q.id = CASE\n WHEN ji.kind IN ('flow', 'flowpreview', 'singlestepflow') THEN ji.id\n ELSE COALESCE(ji.parent_job, ji.id)\n END\n JOIN v2_job j ON j.id = q.id\n LEFT JOIN v2_job_status s ON s.id = q.id\n FOR UPDATE OF q\n ", + "query": "\n WITH job_info AS (\n SELECT id, kind::text AS kind, parent_job\n FROM v2_job\n WHERE id = $1 AND workspace_id = $2\n )\n SELECT\n q.id AS \"id!\",\n q.suspend AS \"suspend!\",\n j.runnable_path AS script_path,\n j.permissioned_as_email AS email,\n (ji.kind IN ('flow', 'flowpreview', 'singlestepflow')) AS \"is_flow_level!\",\n (ji.kind NOT IN ('flow', 'flowpreview', 'singlestepflow') AND q.id = ji.id) AS \"is_wac!\"\n FROM job_info ji\n JOIN v2_job_queue q ON q.id = CASE\n WHEN ji.kind IN ('flow', 'flowpreview', 'singlestepflow') THEN ji.id\n ELSE COALESCE(ji.parent_job, ji.id)\n END\n JOIN v2_job j ON j.id = q.id\n FOR UPDATE OF q\n ", "describe": { "columns": [ { @@ -10,31 +10,26 @@ }, { "ordinal": 1, - "name": "flow_status", - "type_info": "Jsonb" - }, - { - "ordinal": 2, "name": "suspend!", "type_info": "Int4" }, { - "ordinal": 3, + "ordinal": 2, "name": "script_path", "type_info": "Varchar" }, { - "ordinal": 4, + "ordinal": 3, "name": "email", "type_info": "Varchar" }, { - "ordinal": 5, + "ordinal": 4, "name": "is_flow_level!", "type_info": "Bool" }, { - "ordinal": 6, + "ordinal": 5, "name": "is_wac!", "type_info": "Bool" } @@ -47,7 +42,6 @@ }, "nullable": [ false, - true, false, true, false, @@ -55,5 +49,5 @@ null ] }, - "hash": "11b5d006acf040b2461b9539863b1e511c23e1a5f9be97da066a6808a49f2092" + "hash": "0a18205b64e68529b7f90bc1c5db2cbea9ee5eaa962970b65a92597fd5e62707" } diff --git a/backend/.sqlx/query-141725e44ccd3f9cc0b5eb070ea3de94b89a4940901e11c48c5f2014ff28086a.json b/backend/.sqlx/query-141725e44ccd3f9cc0b5eb070ea3de94b89a4940901e11c48c5f2014ff28086a.json new file mode 100644 index 0000000000..7aab03b025 --- /dev/null +++ b/backend/.sqlx/query-141725e44ccd3f9cc0b5eb070ea3de94b89a4940901e11c48c5f2014ff28086a.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "WITH RECURSIVE up AS (\n SELECT id, parent_job, kind, runnable_id, runnable_path, 0 AS depth\n FROM v2_job WHERE id = $1 AND workspace_id = $2\n UNION ALL\n SELECT p.id, p.parent_job, p.kind, p.runnable_id, p.runnable_path, u.depth + 1\n FROM v2_job p JOIN up u ON p.id = u.parent_job\n WHERE p.workspace_id = $2 AND u.kind <> 'flow' AND u.depth < 100\n )\n SELECT EXISTS (SELECT 1 FROM up WHERE kind = 'flow'\n AND runnable_id = (SELECT f.versions[array_upper(f.versions, 1)]\n FROM flow f WHERE f.path = up.runnable_path AND f.workspace_id = $2))\n AND NOT EXISTS (SELECT 1 FROM up WHERE kind <> 'flow' AND (\n EXISTS (SELECT 1 FROM v2_job_completed c WHERE c.id = up.id\n AND jsonb_typeof(c.flow_status->'restarted_from') = 'object')\n OR EXISTS (SELECT 1 FROM v2_job_status st WHERE st.id = up.id\n AND jsonb_typeof(st.flow_status->'restarted_from') = 'object')))\n AS \"current!\"", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "current!", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "141725e44ccd3f9cc0b5eb070ea3de94b89a4940901e11c48c5f2014ff28086a" +} diff --git a/backend/.sqlx/query-819c4d78f10979a83529155c3f101f42f6958742bc68ab384944fa3c5cba094c.json b/backend/.sqlx/query-2d73fe76e45becbcae7a9d4384bbb2f9739cab232b8f3e357e27dc32bb35c3a8.json similarity index 62% rename from backend/.sqlx/query-819c4d78f10979a83529155c3f101f42f6958742bc68ab384944fa3c5cba094c.json rename to backend/.sqlx/query-2d73fe76e45becbcae7a9d4384bbb2f9739cab232b8f3e357e27dc32bb35c3a8.json index b25b301a04..23b00521d2 100644 --- a/backend/.sqlx/query-819c4d78f10979a83529155c3f101f42f6958742bc68ab384944fa3c5cba094c.json +++ b/backend/.sqlx/query-2d73fe76e45becbcae7a9d4384bbb2f9739cab232b8f3e357e27dc32bb35c3a8.json @@ -1,12 +1,17 @@ { "db_name": "PostgreSQL", - "query": "DELETE FROM script WHERE workspace_id = $1 AND path = ANY($2) RETURNING path", + "query": "DELETE FROM script WHERE workspace_id = $1 AND path = ANY($2) RETURNING path, hash", "describe": { "columns": [ { "ordinal": 0, "name": "path", "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "hash", + "type_info": "Int8" } ], "parameters": { @@ -16,8 +21,9 @@ ] }, "nullable": [ + false, false ] }, - "hash": "819c4d78f10979a83529155c3f101f42f6958742bc68ab384944fa3c5cba094c" + "hash": "2d73fe76e45becbcae7a9d4384bbb2f9739cab232b8f3e357e27dc32bb35c3a8" } diff --git a/backend/.sqlx/query-5176c354a8c14b8acfbcc7e32da2fcd5013543d600a07a36f3451ddb00cad6aa.json b/backend/.sqlx/query-5176c354a8c14b8acfbcc7e32da2fcd5013543d600a07a36f3451ddb00cad6aa.json deleted file mode 100644 index dda9bc9869..0000000000 --- a/backend/.sqlx/query-5176c354a8c14b8acfbcc7e32da2fcd5013543d600a07a36f3451ddb00cad6aa.json +++ /dev/null @@ -1,101 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "WITH RECURSIVE lineage AS (\n SELECT id, parent_job, kind, runnable_path, runnable_id, trigger_kind, trigger,\n permissioned_as, args, 0 AS depth\n FROM v2_job WHERE id = $1 AND workspace_id = $2\n UNION ALL\n SELECT p.id, p.parent_job, p.kind, p.runnable_path, p.runnable_id, p.trigger_kind,\n p.trigger, p.permissioned_as, p.args, l.depth + 1\n FROM v2_job p JOIN lineage l ON p.id = l.parent_job\n WHERE p.workspace_id = $2 AND l.depth < 100\n )\n SELECT id AS \"id!\", parent_job, kind AS \"kind!: JobKind\", runnable_path,\n trigger_kind::text AS trigger_kind, permissioned_as AS \"permissioned_as!\",\n -- A restart takes its flow version from the request, so a trusted flow path\n -- can carry another flow's code: the version must belong to that path.\n CASE kind\n WHEN 'flow' THEN EXISTS (SELECT 1 FROM flow_version fv\n WHERE fv.id = runnable_id AND fv.path = runnable_path AND fv.workspace_id = $2)\n WHEN 'script' THEN EXISTS (SELECT 1 FROM script s\n WHERE s.hash = runnable_id AND s.path = runnable_path AND s.workspace_id = $2)\n ELSE true\n END AS \"origin_verified!\",\n -- Only deployed-app runs are stamped with their app; an app editor preview\n -- runs app code at an app path it does not have to own.\n COALESCE(trigger_kind = 'app' AND starts_with(runnable_path, trigger || '/'), false)\n AS \"app_stamped!\",\n -- A preview's modules come from its args, which its parent may have taken from\n -- the caller.\n COALESCE(jsonb_typeof(args->'_MODULES') = 'object', false) AS \"args_modules!\"\n FROM lineage ORDER BY depth", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "id!", - "type_info": "Uuid" - }, - { - "ordinal": 1, - "name": "parent_job", - "type_info": "Uuid" - }, - { - "ordinal": 2, - "name": "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", - "singlestepflow", - "flowscript", - "flownode", - "appscript", - "aiagent", - "unassigned_script", - "unassigned_flow", - "unassigned_singlestepflow" - ] - } - } - } - }, - { - "ordinal": 3, - "name": "runnable_path", - "type_info": "Varchar" - }, - { - "ordinal": 4, - "name": "trigger_kind", - "type_info": "Text" - }, - { - "ordinal": 5, - "name": "permissioned_as!", - "type_info": "Varchar" - }, - { - "ordinal": 6, - "name": "origin_verified!", - "type_info": "Bool" - }, - { - "ordinal": 7, - "name": "app_stamped!", - "type_info": "Bool" - }, - { - "ordinal": 8, - "name": "args_modules!", - "type_info": "Bool" - } - ], - "parameters": { - "Left": [ - "Uuid", - "Text" - ] - }, - "nullable": [ - null, - null, - null, - null, - null, - null, - null, - null, - null - ] - }, - "hash": "5176c354a8c14b8acfbcc7e32da2fcd5013543d600a07a36f3451ddb00cad6aa" -} diff --git a/backend/.sqlx/query-b44afcf1b9c047ac525638f0952c2cb01d65b1b46693331ac157dfca0dab6824.json b/backend/.sqlx/query-65e33a933ed8568c82f9b4e121b9ba4280123fe22a959d4d3d978ca767aeed29.json similarity index 88% rename from backend/.sqlx/query-b44afcf1b9c047ac525638f0952c2cb01d65b1b46693331ac157dfca0dab6824.json rename to backend/.sqlx/query-65e33a933ed8568c82f9b4e121b9ba4280123fe22a959d4d3d978ca767aeed29.json index 9187cc29b7..8e059f92ed 100644 --- a/backend/.sqlx/query-b44afcf1b9c047ac525638f0952c2cb01d65b1b46693331ac157dfca0dab6824.json +++ b/backend/.sqlx/query-65e33a933ed8568c82f9b4e121b9ba4280123fe22a959d4d3d978ca767aeed29.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT content AS \"content!: String\",\n lock AS \"lock: String\", language AS \"language: Option\", envs AS \"envs: Vec\", schema AS \"schema: String\", schema_validation AS \"schema_validation: bool\", codebase LIKE '%.tar' as use_tar, codebase LIKE '%.esm%' as is_esm, modules AS \"modules: serde_json::Value\" FROM script WHERE hash = $1 LIMIT 1", + "query": "SELECT content AS \"content!: String\",\n lock AS \"lock: String\", language AS \"language: Option\", envs AS \"envs: Vec\", schema AS \"schema: String\", schema_validation AS \"schema_validation: bool\", codebase LIKE '%.tar' as use_tar, codebase LIKE '%.esm%' as is_esm, modules AS \"modules: serde_json::Value\", deleted FROM script WHERE hash = $1 ORDER BY deleted LIMIT 1", "describe": { "columns": [ { @@ -80,6 +80,11 @@ "ordinal": 8, "name": "modules: serde_json::Value", "type_info": "Jsonb" + }, + { + "ordinal": 9, + "name": "deleted", + "type_info": "Bool" } ], "parameters": { @@ -96,8 +101,9 @@ false, null, null, - true + true, + false ] }, - "hash": "b44afcf1b9c047ac525638f0952c2cb01d65b1b46693331ac157dfca0dab6824" + "hash": "65e33a933ed8568c82f9b4e121b9ba4280123fe22a959d4d3d978ca767aeed29" } diff --git a/backend/.sqlx/query-84ad6f24aa9ccdc6f6e705b7de1bb707637a9bb1475711f9f9f3f2a6e28cb169.json b/backend/.sqlx/query-84ad6f24aa9ccdc6f6e705b7de1bb707637a9bb1475711f9f9f3f2a6e28cb169.json deleted file mode 100644 index 510e47e50c..0000000000 --- a/backend/.sqlx/query-84ad6f24aa9ccdc6f6e705b7de1bb707637a9bb1475711f9f9f3f2a6e28cb169.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE v2_job_queue SET suspend = $1 WHERE id = $2", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Int4", - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "84ad6f24aa9ccdc6f6e705b7de1bb707637a9bb1475711f9f9f3f2a6e28cb169" -} diff --git a/backend/.sqlx/query-a35164456ade8e79cb8f5418c8fe82c45be45409f881292f0d4a3316362ba1f4.json b/backend/.sqlx/query-a35164456ade8e79cb8f5418c8fe82c45be45409f881292f0d4a3316362ba1f4.json deleted file mode 100644 index aedbbf424e..0000000000 --- a/backend/.sqlx/query-a35164456ade8e79cb8f5418c8fe82c45be45409f881292f0d4a3316362ba1f4.json +++ /dev/null @@ -1,14 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE v2_job_queue SET suspend = GREATEST(suspend - 1, 0) WHERE id = $1", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "a35164456ade8e79cb8f5418c8fe82c45be45409f881292f0d4a3316362ba1f4" -} diff --git a/backend/.sqlx/query-ee769c78ec0a782de18911cdcb655aad93cf23321cd0f42f7ffa5348bfbea26e.json b/backend/.sqlx/query-ee769c78ec0a782de18911cdcb655aad93cf23321cd0f42f7ffa5348bfbea26e.json new file mode 100644 index 0000000000..bb7e4fa36b --- /dev/null +++ b/backend/.sqlx/query-ee769c78ec0a782de18911cdcb655aad93cf23321cd0f42f7ffa5348bfbea26e.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT EXISTS (SELECT 1 FROM v2_job_queue WHERE id = $1 AND workspace_id = $2 AND running)", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "exists", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "ee769c78ec0a782de18911cdcb655aad93cf23321cd0f42f7ffa5348bfbea26e" +} diff --git a/backend/.sqlx/query-ee92100fcd681515ba81f23970e64af7f33464879bef4d1eff682126dffeef64.json b/backend/.sqlx/query-ee92100fcd681515ba81f23970e64af7f33464879bef4d1eff682126dffeef64.json new file mode 100644 index 0000000000..3c52f8ba96 --- /dev/null +++ b/backend/.sqlx/query-ee92100fcd681515ba81f23970e64af7f33464879bef4d1eff682126dffeef64.json @@ -0,0 +1,125 @@ +{ + "db_name": "PostgreSQL", + "query": "WITH RECURSIVE lineage AS (\n SELECT id, parent_job, kind, runnable_path, runnable_id, trigger_kind, trigger,\n permissioned_as, args, 0 AS depth\n FROM v2_job WHERE id = $1 AND workspace_id = $2\n UNION ALL\n SELECT p.id, p.parent_job, p.kind, p.runnable_path, p.runnable_id, p.trigger_kind,\n p.trigger, p.permissioned_as, p.args, l.depth + 1\n FROM v2_job p JOIN lineage l ON p.id = l.parent_job\n WHERE p.workspace_id = $2 AND l.depth < 100\n )\n SELECT id AS \"id!\", parent_job, kind AS \"kind!: JobKind\", runnable_path, runnable_id,\n trigger_kind::text AS trigger_kind, permissioned_as AS \"permissioned_as!\",\n -- A restart takes its flow version from the request, so a trusted flow path\n -- can carry another flow's code: the version must belong to that path.\n CASE kind\n WHEN 'flow' THEN EXISTS (SELECT 1 FROM flow_version fv\n WHERE fv.id = runnable_id AND fv.path = runnable_path AND fv.workspace_id = $2)\n WHEN 'script' THEN EXISTS (SELECT 1 FROM script s\n WHERE s.hash = runnable_id AND s.path = runnable_path AND s.workspace_id = $2\n AND NOT s.deleted)\n ELSE true\n END AS \"origin_verified!\",\n -- What `run/f` and `run/p` resolve the path to now; for scripts the predicate of\n -- `get_latest_deployed_script_hash`.\n COALESCE(CASE kind\n WHEN 'flow' THEN runnable_id = (SELECT f.versions[array_upper(f.versions, 1)]\n FROM flow f WHERE f.path = runnable_path AND f.workspace_id = $2)\n WHEN 'script' THEN runnable_id = (SELECT s.hash FROM script s\n WHERE s.path = runnable_path AND s.workspace_id = $2 AND NOT s.deleted\n AND s.lock IS NOT NULL AND s.lock_error_logs IS NULL\n ORDER BY s.created_at DESC LIMIT 1)\n END, false) AS \"current_version!\",\n EXISTS (SELECT 1 FROM v2_job_status st WHERE st.id = lineage.id\n AND jsonb_typeof(st.flow_status->'restarted_from') = 'object')\n OR EXISTS (SELECT 1 FROM v2_job_completed c WHERE c.id = lineage.id\n AND jsonb_typeof(c.flow_status->'restarted_from') = 'object')\n AS \"restarted!\",\n COALESCE(\n (SELECT st.flow_status->'restarted_from'->>'flow_job_id' FROM v2_job_status st\n WHERE st.id = lineage.id),\n (SELECT c.flow_status->'restarted_from'->>'flow_job_id' FROM v2_job_completed c\n WHERE c.id = lineage.id)\n ) AS restarted_from,\n -- Only deployed-app runs are stamped with their app; an app editor preview\n -- runs app code at an app path it does not have to own.\n COALESCE(trigger_kind = 'app' AND starts_with(runnable_path, trigger || '/'), false)\n AS \"app_stamped!\",\n -- A preview's modules come from its args, which its parent may have taken from\n -- the caller.\n COALESCE(jsonb_typeof(args->'_MODULES') = 'object', false) AS \"args_modules!\"\n FROM lineage ORDER BY depth", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id!", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "parent_job", + "type_info": "Uuid" + }, + { + "ordinal": 2, + "name": "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", + "singlestepflow", + "flowscript", + "flownode", + "appscript", + "aiagent", + "unassigned_script", + "unassigned_flow", + "unassigned_singlestepflow" + ] + } + } + } + }, + { + "ordinal": 3, + "name": "runnable_path", + "type_info": "Varchar" + }, + { + "ordinal": 4, + "name": "runnable_id", + "type_info": "Int8" + }, + { + "ordinal": 5, + "name": "trigger_kind", + "type_info": "Text" + }, + { + "ordinal": 6, + "name": "permissioned_as!", + "type_info": "Varchar" + }, + { + "ordinal": 7, + "name": "origin_verified!", + "type_info": "Bool" + }, + { + "ordinal": 8, + "name": "current_version!", + "type_info": "Bool" + }, + { + "ordinal": 9, + "name": "restarted!", + "type_info": "Bool" + }, + { + "ordinal": 10, + "name": "restarted_from", + "type_info": "Text" + }, + { + "ordinal": 11, + "name": "app_stamped!", + "type_info": "Bool" + }, + { + "ordinal": 12, + "name": "args_modules!", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + null, + null, + null, + null, + null, + null, + null, + null, + null, + null, + null, + null, + null + ] + }, + "hash": "ee92100fcd681515ba81f23970e64af7f33464879bef4d1eff682126dffeef64" +} diff --git a/backend/.sqlx/query-0e7d95f4913e5775651971d741a3b5c1ef5dfe079be5325abe2866d39a7fe5fb.json b/backend/.sqlx/query-fa85a22c77844bb2c538b802b0984d4d69aecdd8a65db02e9495b0d6011c5c1f.json similarity index 65% rename from backend/.sqlx/query-0e7d95f4913e5775651971d741a3b5c1ef5dfe079be5325abe2866d39a7fe5fb.json rename to backend/.sqlx/query-fa85a22c77844bb2c538b802b0984d4d69aecdd8a65db02e9495b0d6011c5c1f.json index 9548fa6997..6c54ede027 100644 --- a/backend/.sqlx/query-0e7d95f4913e5775651971d741a3b5c1ef5dfe079be5325abe2866d39a7fe5fb.json +++ b/backend/.sqlx/query-fa85a22c77844bb2c538b802b0984d4d69aecdd8a65db02e9495b0d6011c5c1f.json @@ -1,12 +1,12 @@ { "db_name": "PostgreSQL", - "query": "DELETE FROM script WHERE path = $1 AND workspace_id = $2 RETURNING path", + "query": "DELETE FROM script WHERE path = $1 AND workspace_id = $2 RETURNING hash", "describe": { "columns": [ { "ordinal": 0, - "name": "path", - "type_info": "Varchar" + "name": "hash", + "type_info": "Int8" } ], "parameters": { @@ -19,5 +19,5 @@ false ] }, - "hash": "0e7d95f4913e5775651971d741a3b5c1ef5dfe079be5325abe2866d39a7fe5fb" + "hash": "fa85a22c77844bb2c538b802b0984d4d69aecdd8a65db02e9495b0d6011c5c1f" } diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 076b313eb6..c2c253e8f0 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -40ac1c5f8cbce3843b582d9b392d3f3cc7eca3e6 +ff4d04f17721d84fb9a1f655af50ff7e4e7dcbc7 diff --git a/backend/oauth_connect.json b/backend/oauth_connect.json index bc2498a943..0509e7f30a 100644 --- a/backend/oauth_connect.json +++ b/backend/oauth_connect.json @@ -18,7 +18,32 @@ "slack": { "auth_url": "https://slack.com/oauth/v2/authorize", "token_url": "https://slack.com/api/oauth.v2.access", - "scopes": ["chat:write", "chat:write.public", "channels:join", "files:write"] + "scopes": [ + "chat:write", + "chat:write.public", + "channels:join", + "files:write", + "users:read", + "users:read.email", + "channels:read", + "channels:history" + ], + "user_scopes": [ + "channels:history", + "groups:history", + "im:history", + "mpim:history", + "search:read", + "users:read", + "chat:write", + "users:read.email", + "channels:read", + "groups:read", + "im:read", + "mpim:read", + "files:read" + ], + "token_response_path": "authed_user" }, "supabase_wizard": { "auth_url": "https://api.supabase.com/v1/oauth/authorize", @@ -30,8 +55,8 @@ "token_url": "https://oauth2.googleapis.com/token", "scopes": ["https://www.googleapis.com/auth/spreadsheets"], "scope_options": [ - "https://www.googleapis.com/auth/spreadsheets", - "https://www.googleapis.com/auth/spreadsheets.readonly" + { "value": "https://www.googleapis.com/auth/spreadsheets", "label": "Read, edit, create and delete spreadsheets" }, + { "value": "https://www.googleapis.com/auth/spreadsheets.readonly", "label": "Read spreadsheets" } ], "extra_params": { "access_type": "offline", @@ -43,9 +68,9 @@ "token_url": "https://oauth2.googleapis.com/token", "scopes": ["https://www.googleapis.com/auth/drive"], "scope_options": [ - "https://www.googleapis.com/auth/drive.file", - "https://www.googleapis.com/auth/drive.readonly", - "https://www.googleapis.com/auth/drive" + { "value": "https://www.googleapis.com/auth/drive.file", "label": "Only files created or opened with this connection" }, + { "value": "https://www.googleapis.com/auth/drive.readonly", "label": "Read all Drive files" }, + { "value": "https://www.googleapis.com/auth/drive", "label": "Read, edit, create and delete all Drive files" } ], "extra_params": { "access_type": "offline", @@ -57,11 +82,11 @@ "token_url": "https://oauth2.googleapis.com/token", "scopes": ["https://www.googleapis.com/auth/gmail.send"], "scope_options": [ - "https://www.googleapis.com/auth/gmail.send", - "https://www.googleapis.com/auth/gmail.readonly", - "https://www.googleapis.com/auth/gmail.compose", - "https://www.googleapis.com/auth/gmail.modify", - "https://www.googleapis.com/auth/gmail.labels" + { "value": "https://www.googleapis.com/auth/gmail.send", "label": "Send mail only" }, + { "value": "https://www.googleapis.com/auth/gmail.readonly", "label": "Read and search mail" }, + { "value": "https://www.googleapis.com/auth/gmail.compose", "label": "Draft and send mail" }, + { "value": "https://www.googleapis.com/auth/gmail.modify", "label": "Read, search, label, draft and send mail" }, + { "value": "https://www.googleapis.com/auth/gmail.labels", "label": "Create, edit and delete labels" } ], "extra_params": { "access_type": "offline", @@ -73,10 +98,10 @@ "token_url": "https://oauth2.googleapis.com/token", "scopes": ["https://www.googleapis.com/auth/calendar.events"], "scope_options": [ - "https://www.googleapis.com/auth/calendar.events", - "https://www.googleapis.com/auth/calendar.events.readonly", - "https://www.googleapis.com/auth/calendar.readonly", - "https://www.googleapis.com/auth/calendar" + { "value": "https://www.googleapis.com/auth/calendar.events", "label": "Read and edit events on all calendars" }, + { "value": "https://www.googleapis.com/auth/calendar.events.readonly", "label": "Read events on all calendars" }, + { "value": "https://www.googleapis.com/auth/calendar.readonly", "label": "Read all calendars and events" }, + { "value": "https://www.googleapis.com/auth/calendar", "label": "Read, edit, share and delete all calendars" } ], "extra_params": { "access_type": "offline", @@ -88,10 +113,43 @@ "token_url": "https://oauth2.googleapis.com/token", "scopes": ["https://www.googleapis.com/auth/forms"], "scope_options": [ - "https://www.googleapis.com/auth/forms", - "https://www.googleapis.com/auth/forms.body", - "https://www.googleapis.com/auth/forms.body.readonly", - "https://www.googleapis.com/auth/forms.responses.readonly" + { "value": "https://www.googleapis.com/auth/forms", "label": "Read and manage forms in Drive" }, + { "value": "https://www.googleapis.com/auth/forms.body", "label": "Read, edit, create and delete forms" }, + { "value": "https://www.googleapis.com/auth/forms.body.readonly", "label": "Read forms" }, + { "value": "https://www.googleapis.com/auth/forms.responses.readonly", "label": "Read form responses" } + ], + "extra_params": { + "access_type": "offline", + "prompt": "consent" + } + }, + "gdocs": { + "auth_url": "https://accounts.google.com/o/oauth2/v2/auth", + "token_url": "https://oauth2.googleapis.com/token", + "scopes": ["https://www.googleapis.com/auth/documents"], + "scope_options": [ + { "value": "https://www.googleapis.com/auth/documents", "label": "Read, edit, create and delete documents" }, + { "value": "https://www.googleapis.com/auth/documents.readonly", "label": "Read documents" }, + { "value": "https://www.googleapis.com/auth/drive.file", "label": "Only files created or opened with this connection" } + ], + "extra_params": { + "access_type": "offline", + "prompt": "consent" + } + }, + "gchat": { + "auth_url": "https://accounts.google.com/o/oauth2/v2/auth", + "token_url": "https://oauth2.googleapis.com/token", + "scopes": [ + "https://www.googleapis.com/auth/chat.spaces.readonly", + "https://www.googleapis.com/auth/chat.messages.create" + ], + "scope_options": [ + { "value": "https://www.googleapis.com/auth/chat.spaces.readonly", "label": "Read spaces" }, + { "value": "https://www.googleapis.com/auth/chat.messages", "label": "Read, send, edit and delete messages" }, + { "value": "https://www.googleapis.com/auth/chat.messages.readonly", "label": "Read messages" }, + { "value": "https://www.googleapis.com/auth/chat.messages.create", "label": "Send messages only" }, + { "value": "https://www.googleapis.com/auth/chat.memberships.readonly", "label": "Read space members" } ], "extra_params": { "access_type": "offline", @@ -103,8 +161,8 @@ "token_url": "https://oauth2.googleapis.com/token", "scopes": ["https://www.googleapis.com/auth/cloud-platform"], "scope_options": [ - "https://www.googleapis.com/auth/cloud-platform", - "https://www.googleapis.com/auth/cloud-platform.read-only" + { "value": "https://www.googleapis.com/auth/cloud-platform", "label": "Read and manage all Google Cloud resources" }, + { "value": "https://www.googleapis.com/auth/cloud-platform.read-only", "label": "Read Google Cloud resources" } ], "extra_params": { "access_type": "offline", @@ -121,13 +179,13 @@ "https://www.googleapis.com/auth/admin.directory.orgunit" ], "scope_options": [ - "https://www.googleapis.com/auth/admin.directory.user", - "https://www.googleapis.com/auth/admin.directory.user.readonly", - "https://www.googleapis.com/auth/admin.directory.group", - "https://www.googleapis.com/auth/admin.directory.group.readonly", - "https://www.googleapis.com/auth/admin.directory.orgunit", - "https://www.googleapis.com/auth/admin.directory.orgunit.readonly", - "https://www.googleapis.com/auth/admin.directory.user.security" + { "value": "https://www.googleapis.com/auth/admin.directory.user", "label": "Read and manage users" }, + { "value": "https://www.googleapis.com/auth/admin.directory.user.readonly", "label": "Read users" }, + { "value": "https://www.googleapis.com/auth/admin.directory.group", "label": "Read and manage groups" }, + { "value": "https://www.googleapis.com/auth/admin.directory.group.readonly", "label": "Read groups" }, + { "value": "https://www.googleapis.com/auth/admin.directory.orgunit", "label": "Read and manage organizational units" }, + { "value": "https://www.googleapis.com/auth/admin.directory.orgunit.readonly", "label": "Read organizational units" }, + { "value": "https://www.googleapis.com/auth/admin.directory.user.security", "label": "Manage user security settings" } ], "extra_params": { "access_type": "offline", diff --git a/backend/src/main.rs b/backend/src/main.rs index 257ad77d1d..23978d7e62 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -1232,6 +1232,11 @@ Windmill Community Edition {GIT_VERSION} default_base_internal_url.clone() }; + // Deletions are delivered by `notify_event` to running processes only, so one that + // was down when a version was deleted would keep its code on disk: start from an + // empty script cache, refilled from the database. + windmill_common::cache::script::clear(); + initial_load( &conn, killpill_tx.clone(), @@ -1884,6 +1889,12 @@ async fn process_notify_event( ); windmill_common::workspaces::PUBLIC_APP_RATE_LIMIT_CACHE.remove(payload); } + windmill_common::SCRIPT_VERSION_DELETED_CHANNEL => { + match serde_json::from_str(payload) { + Ok(deleted) => windmill_api_scripts::scripts::evict_deleted_script_versions(deleted), + Err(e) => tracing::error!("Invalid script version deletion payload {payload}: {e}"), + } + } "notify_runnable_version_change" => { tracing::info!("Runnable version change detected: {}", payload); match payload.split(':').collect::>().as_slice() { diff --git a/backend/tests/deleted_script_version.rs b/backend/tests/deleted_script_version.rs new file mode 100644 index 0000000000..fdc49043b9 --- /dev/null +++ b/backend/tests/deleted_script_version.rs @@ -0,0 +1,56 @@ +//! Script data is cached by hash, and the hash leaves out the workspace, so forks and clones +//! share it: a deleted copy in one workspace must not stop the live copy in another. + +use sqlx::{Pool, Postgres}; +use windmill_common::{cache, scripts::ScriptHash, worker::Connection}; + +#[sqlx::test(fixtures("base"))] +async fn a_deleted_copy_does_not_shadow_a_live_one(db: Pool) { + const SHARED: i64 = 0x5de1_e7ed_0002; + sqlx::query("INSERT INTO workspace (id, name, owner) VALUES ('fork', 'fork', 'test-user')") + .execute(&db) + .await + .unwrap(); + sqlx::query( + "INSERT INTO script (workspace_id, hash, path, summary, description, content, + created_by, language, lock, archived, deleted) + VALUES ('fork', $1, 'f/infra/tool', '', '', '', 'test-user', 'bash', '', true, true), + ('test-workspace', $1, 'f/infra/tool', '', '', 'echo shared', 'test-user', + 'bash', '', false, false)", + ) + .bind(SHARED) + .execute(&db) + .await + .unwrap(); + + let conn = Connection::Sql(db.clone()); + let (data, _) = cache::script::fetch(&conn, ScriptHash(SHARED)) + .await + .unwrap(); + assert_eq!(data.code, "echo shared"); +} + +#[sqlx::test(fixtures("base"))] +async fn a_large_deletion_is_split_across_events(db: Pool) { + let deleted = windmill_common::DeletedScriptVersions::new( + "test-workspace", + (0..501).map(|h| (format!("f/infra/s{}", h % 2), h)), + ); + let mut conn = db.acquire().await.unwrap(); + deleted.notify(&mut conn).await.unwrap(); + + let events: Vec = sqlx::query_scalar( + "SELECT payload FROM notify_event WHERE channel = $1 ORDER BY id", + ) + .bind(windmill_common::SCRIPT_VERSION_DELETED_CHANNEL) + .fetch_all(&db) + .await + .unwrap(); + let events: Vec = + events.iter().map(|e| serde_json::from_str(e).unwrap()).collect(); + assert_eq!(events.len(), 2); + let hashes: Vec = events.iter().flat_map(|e| e.hashes.clone()).collect(); + let paths: Vec = events.iter().flat_map(|e| e.paths.clone()).collect(); + assert_eq!(hashes, deleted.hashes); + assert_eq!(paths, deleted.paths); +} diff --git a/backend/tests/fixtures/job_provenance_versions.sql b/backend/tests/fixtures/job_provenance_versions.sql new file mode 100644 index 0000000000..5f0f204305 --- /dev/null +++ b/backend/tests/fixtures/job_provenance_versions.sql @@ -0,0 +1,92 @@ +-- Add-on fixture for job_provenance.rs (combine with `base.sql` and `job_provenance.sql`): +-- past versions of f/t/tool and f/t/agent, a deleted version of f/t/tool, and flow steps +-- of f/t/agent under a regular run and under restarts. + +INSERT INTO script (workspace_id, hash, path, content, language, kind, created_by, schema, summary, description, lock, created_at, archived) +VALUES ('test-workspace', 1111111110, 'f/t/tool', 'echo old tool', 'bash', 'script', 'test-user', '{}', '', '', '', NOW() - interval '1 day', true); + +INSERT INTO script (workspace_id, hash, path, content, language, kind, created_by, schema, summary, description, lock, created_at, archived, deleted) +VALUES ('test-workspace', 1111111109, 'f/t/tool', '', 'bash', 'script', 'test-user', '{}', '', '', '', NOW() - interval '2 days', true, true); + +INSERT INTO flow_version (id, workspace_id, path, value, schema, created_by, created_at) +VALUES (2222222221, 'test-workspace', 'f/t/agent', '{"modules":[]}', '{}', 'test-user', NOW() - interval '1 day'); + +UPDATE flow SET versions = ARRAY[2222222221::bigint, 2222222222::bigint] +WHERE workspace_id = 'test-workspace' AND path = 'f/t/agent'; + +INSERT INTO v2_job (id, workspace_id, kind, runnable_path, runnable_id, parent_job, created_by, permissioned_as, permissioned_as_email) +VALUES + -- the past tool version, run under the current flow version + ('3bb0c0de-0000-4000-8000-000000000101', 'test-workspace', 'script', 'f/t/tool', 1111111110, '3bb0c0de-0000-4000-8000-000000000001', 'test-user', 'u/test-user', 'test@windmill.dev'), + -- the current tool version, run under the past flow version + ('3bb0c0de-0000-4000-8000-000000000102', 'test-workspace', 'flow', 'f/t/agent', 2222222221, NULL, 'test-user', 'u/test-user', 'test@windmill.dev'), + ('3bb0c0de-0000-4000-8000-000000000103', 'test-workspace', 'script', 'f/t/tool', 1111111111, '3bb0c0de-0000-4000-8000-000000000102', 'test-user', 'u/test-user', 'test@windmill.dev'), + -- the deleted tool version + ('3bb0c0de-0000-4000-8000-000000000104', 'test-workspace', 'script', 'f/t/tool', 1111111109, NULL, 'test-user', 'u/test-user', 'test@windmill.dev'), + -- under the current flow version: an inline step, a loop body and a step of that body + ('3bb0c0de-0000-4000-8000-000000000105', 'test-workspace', 'flowscript', 'f/t/agent/a', 4444444442, '3bb0c0de-0000-4000-8000-000000000001', 'test-user', 'u/test-user', 'test@windmill.dev'), + ('3bb0c0de-0000-4000-8000-000000000106', 'test-workspace', 'flownode', 'f/t/agent/loop-0', 4444444444, '3bb0c0de-0000-4000-8000-000000000001', 'test-user', 'u/test-user', 'test@windmill.dev'), + ('3bb0c0de-0000-4000-8000-000000000107', 'test-workspace', 'flowscript', 'f/t/agent/loop-0/b', 4444444446, '3bb0c0de-0000-4000-8000-000000000106', 'test-user', 'u/test-user', 'test@windmill.dev'), + -- an inline step under the past flow version + ('3bb0c0de-0000-4000-8000-000000000108', 'test-workspace', 'flowscript', 'f/t/agent/a', 4444444441, '3bb0c0de-0000-4000-8000-000000000102', 'test-user', 'u/test-user', 'test@windmill.dev'), + -- a loop body restarted on its own, and a step of it + ('3bb0c0de-0000-4000-8000-000000000109', 'test-workspace', 'flownode', 'f/t/agent/loop-0', 4444444443, NULL, 'test-user', 'u/test-user', 'test@windmill.dev'), + ('3bb0c0de-0000-4000-8000-000000000110', 'test-workspace', 'flowscript', 'f/t/agent/loop-0/b', 4444444449, '3bb0c0de-0000-4000-8000-000000000109', 'test-user', 'u/test-user', 'test@windmill.dev'), + -- a body kept by a nested restart under the current flow version, and a step of it + ('3bb0c0de-0000-4000-8000-000000000111', 'test-workspace', 'flownode', 'f/t/agent/branchone-1', 4444444445, '3bb0c0de-0000-4000-8000-000000000001', 'test-user', 'u/test-user', 'test@windmill.dev'), + ('3bb0c0de-0000-4000-8000-000000000112', 'test-workspace', 'flowscript', 'f/t/agent/branchone-1/c', 4444444447, '3bb0c0de-0000-4000-8000-000000000111', 'test-user', 'u/test-user', 'test@windmill.dev'); + +-- An inline body (a flow deployed before flow nodes runs it as a preview) kept by a nested +-- restart under the current flow version, and a step of it. +INSERT INTO v2_job (id, workspace_id, kind, runnable_path, parent_job, created_by, permissioned_as, permissioned_as_email) +VALUES + ('3bb0c0de-0000-4000-8000-000000000113', 'test-workspace', 'flowpreview', 'f/t/agent/branchone-0', '3bb0c0de-0000-4000-8000-000000000001', 'test-user', 'u/test-user', 'test@windmill.dev'), + ('3bb0c0de-0000-4000-8000-000000000114', 'test-workspace', 'preview', 'f/t/agent/branchone-0/c', '3bb0c0de-0000-4000-8000-000000000113', 'test-user', 'u/test-user', 'test@windmill.dev'); + +-- The restarted bodies: two still running, one completed. +INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, running) +VALUES + ('3bb0c0de-0000-4000-8000-000000000111', 'test-workspace', NOW(), true), + ('3bb0c0de-0000-4000-8000-000000000113', 'test-workspace', NOW(), true); + +INSERT INTO v2_job_status (id, flow_status) +VALUES + ('3bb0c0de-0000-4000-8000-000000000111', + '{"restarted_from": {"flow_job_id": "3bb0c0de-0000-4000-8000-0000000000ff", "step_id": "br"}}'), + ('3bb0c0de-0000-4000-8000-000000000113', + '{"restarted_from": {"flow_job_id": "3bb0c0de-0000-4000-8000-0000000000fd", "step_id": "br"}}'); + +INSERT INTO v2_job_completed (id, workspace_id, duration_ms, status, flow_status) +VALUES ('3bb0c0de-0000-4000-8000-000000000109', 'test-workspace', 0, 'success', + '{"restarted_from": {"flow_job_id": "3bb0c0de-0000-4000-8000-0000000000fe", "step_id": "l"}}'); + +-- Loop bodies restarted on their own from completed runs: one of the current flow version, +-- one of the past version, and one that was itself a restart; with a step of the first. +INSERT INTO v2_job (id, workspace_id, kind, runnable_path, runnable_id, parent_job, created_by, permissioned_as, permissioned_as_email) +VALUES + ('3bb0c0de-0000-4000-8000-000000000115', 'test-workspace', 'flownode', 'f/t/agent/loop-0', 4444444444, '3bb0c0de-0000-4000-8000-000000000001', 'test-user', 'u/test-user', 'test@windmill.dev'), + ('3bb0c0de-0000-4000-8000-000000000116', 'test-workspace', 'flownode', 'f/t/agent/loop-0', 4444444444, NULL, 'test-user', 'u/test-user', 'test@windmill.dev'), + ('3bb0c0de-0000-4000-8000-000000000117', 'test-workspace', 'flowscript', 'f/t/agent/loop-0/b', 4444444446, '3bb0c0de-0000-4000-8000-000000000116', 'test-user', 'u/test-user', 'test@windmill.dev'), + ('3bb0c0de-0000-4000-8000-000000000118', 'test-workspace', 'flownode', 'f/t/agent/loop-0', 4444444443, '3bb0c0de-0000-4000-8000-000000000102', 'test-user', 'u/test-user', 'test@windmill.dev'), + ('3bb0c0de-0000-4000-8000-000000000119', 'test-workspace', 'flownode', 'f/t/agent/loop-0', 4444444443, NULL, 'test-user', 'u/test-user', 'test@windmill.dev'), + ('3bb0c0de-0000-4000-8000-000000000120', 'test-workspace', 'flownode', 'f/t/agent/loop-0', 4444444443, NULL, 'test-user', 'u/test-user', 'test@windmill.dev'); + +INSERT INTO v2_job_completed (id, workspace_id, duration_ms, status) +VALUES + ('3bb0c0de-0000-4000-8000-000000000115', 'test-workspace', 0, 'success'), + ('3bb0c0de-0000-4000-8000-000000000118', 'test-workspace', 0, 'success'); + +INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, running) +VALUES + ('3bb0c0de-0000-4000-8000-000000000116', 'test-workspace', NOW(), true), + ('3bb0c0de-0000-4000-8000-000000000119', 'test-workspace', NOW(), true), + ('3bb0c0de-0000-4000-8000-000000000120', 'test-workspace', NOW(), true); + +INSERT INTO v2_job_status (id, flow_status) +VALUES + ('3bb0c0de-0000-4000-8000-000000000116', + '{"restarted_from": {"flow_job_id": "3bb0c0de-0000-4000-8000-000000000115", "step_id": "b"}}'), + ('3bb0c0de-0000-4000-8000-000000000119', + '{"restarted_from": {"flow_job_id": "3bb0c0de-0000-4000-8000-000000000118", "step_id": "b"}}'), + ('3bb0c0de-0000-4000-8000-000000000120', + '{"restarted_from": {"flow_job_id": "3bb0c0de-0000-4000-8000-000000000109", "step_id": "b"}}'); diff --git a/backend/tests/job_provenance.rs b/backend/tests/job_provenance.rs index af9bc19493..42ccf829e6 100644 --- a/backend/tests/job_provenance.rs +++ b/backend/tests/job_provenance.rs @@ -3,8 +3,9 @@ use sqlx::{Pool, Postgres}; use uuid::Uuid; -use windmill_common::job_provenance::{ - job_provenance, ClaimedItem, ClaimedPath, JobProvenance, +use windmill_common::{ + job_provenance::{job_provenance, ClaimedItem, ClaimedPath, JobProvenance}, + scripts::ScriptHash, }; async fn provenance(db: &Pool, job: &str) -> JobProvenance { @@ -101,3 +102,51 @@ async fn claimed_paths_are_the_request_supplied_origins(db: Pool) { [claim("f/prod/deploy", ClaimedItem::Script)] ); } + +/// Any past version can be run by hash or flow version id: it stays deployed, but only +/// the versions the paths currently resolve to are latest. +#[sqlx::test(fixtures("base", "job_provenance", "job_provenance_versions"))] +async fn only_current_versions_are_latest(db: Pool) { + let current = provenance(&db, "3bb0c0de-0000-4000-8000-000000000002").await; + assert!(current.deployed && current.latest); + let current_hash = ScriptHash(1111111111).to_string(); + assert_eq!(current.version.as_deref(), Some(current_hash.as_str())); + assert_eq!(current.root_version.as_deref(), Some("2222222222")); + + let past_tool = provenance(&db, "3bb0c0de-0000-4000-8000-000000000101").await; + assert!(past_tool.deployed && !past_tool.latest); + + let under_past_flow = provenance(&db, "3bb0c0de-0000-4000-8000-000000000103").await; + assert!(under_past_flow.deployed && !under_past_flow.latest); + assert_eq!(under_past_flow.root_version.as_deref(), Some("2222222221")); + + assert!(!deployed(&db, "3bb0c0de-0000-4000-8000-000000000104").await); +} + +/// Flow steps run what the version of the flow above them defines, except under a +/// restart, which rebuilds a body from a past run, under the current flow version or with +/// no flow above it at all: that body is current only if the run it came from was. +#[sqlx::test(fixtures("base", "job_provenance", "job_provenance_versions"))] +async fn flow_steps_are_latest_only_under_a_current_unrestarted_flow(db: Pool) { + for (job, latest) in [ + ("105", true), + ("106", true), + ("107", true), + ("108", false), + ("109", false), + ("110", false), + ("111", false), + ("112", false), + ("00a", true), + ("113", false), + ("114", false), + ("116", true), + ("117", true), + ("119", false), + ("120", false), + ] { + let p = provenance(&db, &format!("3bb0c0de-0000-4000-8000-000000000{job}")).await; + assert!(p.deployed, "job {job}"); + assert_eq!(p.latest, latest, "job {job}"); + } +} diff --git a/backend/tests/suspend_resume.rs b/backend/tests/suspend_resume.rs index d112ca0b96..6fb5d68004 100644 --- a/backend/tests/suspend_resume.rs +++ b/backend/tests/suspend_resume.rs @@ -961,4 +961,118 @@ mod suspend_resume { assert_eq!(result, json!(42)); Ok(()) } + + /// Approvals that arrive concurrently for a step waiting on several events must each + /// decrement the suspend counter; a lost decrement parks the flow until its timeout. + #[cfg(feature = "deno_core")] + #[sqlx::test(fixtures("base"))] + async fn concurrent_approvals_all_decrement_suspend(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + + const APPROVALS: u32 = 8; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + let flow: FlowValue = serde_json::from_value(json!({ + "modules": [ + { + "id": "a", + "suspend": { "required_events": APPROVALS, "timeout": 86400 }, + "value": { + "type": "rawscript", + "language": "deno", + "content": "export async function main() { return 1 }", + "input_transforms": {}, + }, + }, + { + "id": "b", + "value": { + "type": "rawscript", + "language": "deno", + "content": "export function main(resumes) { return resumes.length }", + "input_transforms": { + "resumes": { "type": "javascript", "expr": "resumes" }, + }, + }, + }, + ], + })) + .unwrap(); + + let flow = + RunJob::from(JobPayload::RawFlow { value: flow, path: None, restarted_from: None }) + .push(&db) + .await; + + let completed = listen_for_completed_jobs(&db).await; + let queue = listen_for_queue(&db).await; + let db_ = db.clone(); + + in_test_worker( + &db, + async move { + let db = db_; + wait_until_flow_suspends(flow, queue, &db).await; + + let step: Uuid = sqlx::query_scalar( + "SELECT (flow_status->'modules'->0->>'job')::uuid \ + FROM v2_job_status WHERE id = $1", + ) + .bind(flow) + .fetch_one(&db) + .await + .unwrap(); + + let token = windmill_common::auth::create_token_for_owner( + &db, "test-workspace", "u/test-user", "", 100, "", &Uuid::nil(), None, None, + ) + .await + .unwrap(); + let client = reqwest::Client::new(); + let mut urls = vec![]; + for resume_id in 0..APPROVALS { + let secret = client + .get(format!( + "http://localhost:{port}/api/w/test-workspace/jobs/job_signature/{step}/{resume_id}?token={token}" + )) + .send() + .await + .unwrap() + .error_for_status() + .unwrap() + .text() + .await + .unwrap(); + urls.push(format!( + "http://localhost:{port}/api/w/test-workspace/jobs_u/resume/{step}/{resume_id}/{secret}" + )); + } + + let responses = futures::future::join_all( + urls.iter().map(|url| client.post(url).json(&json!(null)).send()), + ) + .await; + for r in responses { + r.unwrap().error_for_status().unwrap(); + } + + tokio::time::timeout( + std::time::Duration::from_secs(30), + completed.find(&flow), + ) + .await + .expect("flow stayed suspended after every approval was recorded") + .unwrap(); + }, + port, + ) + .await; + + server.close().await.unwrap(); + + let result = completed_job(flow, &db).await.json_result().unwrap(); + assert_eq!(result, json!(APPROVALS)); + Ok(()) + } } diff --git a/backend/windmill-api-integration-tests/tests/scripts.rs b/backend/windmill-api-integration-tests/tests/scripts.rs index 3a146add27..d6afc9f302 100644 --- a/backend/windmill-api-integration-tests/tests/scripts.rs +++ b/backend/windmill-api-integration-tests/tests/scripts.rs @@ -15,6 +15,17 @@ fn authed(builder: reqwest::RequestBuilder) -> reqwest::RequestBuilder { builder.header("Authorization", "Bearer SECRET_TOKEN") } +async fn last_script_deletion(db: &Pool) -> windmill_common::DeletedScriptVersions { + let payload: String = sqlx::query_scalar( + "SELECT payload FROM notify_event WHERE channel = $1 ORDER BY id DESC LIMIT 1", + ) + .bind(windmill_common::SCRIPT_VERSION_DELETED_CHANNEL) + .fetch_one(db) + .await + .unwrap(); + serde_json::from_str(&payload).unwrap() +} + async fn authed_get(port: u16, endpoint: &str, path: &str) -> reqwest::Response { authed(client().get(script_url(port, endpoint, path))) .send() @@ -366,6 +377,12 @@ async fn test_script_endpoints(db: Pool) -> anyhow::Result<()> { .await .unwrap(); assert_eq!(resp.status(), 200); + // Every process evicts deleted versions from its script caches on this event. + let deleted = last_script_deletion(&db).await; + let another: windmill_common::scripts::ScriptHash = + serde_json::from_value(json!(another_hash))?; + assert_eq!(deleted.paths, ["u/test-user/another_script"]); + assert_eq!(deleted.hashes, [another.0]); // --- delete_bulk --- let resp = authed(client().delete(format!("{base}/delete_bulk"))) @@ -374,6 +391,9 @@ async fn test_script_endpoints(db: Pool) -> anyhow::Result<()> { .await .unwrap(); assert_eq!(resp.status(), 200, "delete_bulk: {}", resp.text().await?); + let deleted = last_script_deletion(&db).await; + assert_eq!(deleted.paths, ["u/test-user/test_script"]); + assert!(!deleted.hashes.is_empty(), "{deleted:?}"); let resp = authed_get(port, "exists/p", "u/test-user/test_script").await; assert_eq!(resp.json::().await?, false); diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index 80a175dd99..bce67755bb 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -17,7 +17,7 @@ use windmill_common::{ utils::{BulkDeleteRequest, WithStarredInfoQuery, HTTP_CLIENT}, webhook::{WebhookMessage, WebhookShared}, workspaces::{check_deploy_rules, RuleCheckResult}, - DB, + DeletedScriptVersions, DB, }; use windmill_queue::schedule::clear_schedule; @@ -818,6 +818,16 @@ fn invalidate_script_path_caches(w_id: &str, script_path: &str) { RAW_SCRIPT_LATEST_HASH_CACHE.remove(&format!("{w_id}:{script_path}")); } +/// [`DeletedScriptVersions::evict`], plus this crate's own path -> hash cache. Run by the +/// deleting process after its commit and by every process on the deletion event. +pub fn evict_deleted_script_versions(deleted: DeletedScriptVersions) { + deleted.evict_with(|deleted| { + for path in &deleted.paths { + RAW_SCRIPT_LATEST_HASH_CACHE.remove(&format!("{}:{path}", deleted.workspace_id)); + } + }); +} + /// What a script deploy still has to do once its transaction has committed. enum PostCommitDeploy { /// Everything, for a deploy with no dependency job to hand it to. @@ -4161,6 +4171,8 @@ async fn delete_script_by_hash( // the script was never a pipeline member. clear_script_triggers(&mut *tx, &w_id, &script.path, AssetUsageKind::Script).await?; clear_macro_registry(&mut *tx, &w_id, &script.path).await?; + let deleted = DeletedScriptVersions::new(&w_id, [(script.path.clone(), hash.0)]); + deleted.notify(&mut *tx).await?; audit_log( &mut *tx, @@ -4173,6 +4185,7 @@ async fn delete_script_by_hash( ) .await?; tx.commit().await?; + evict_deleted_script_versions(deleted); webhook.send_message( w_id.clone(), @@ -4240,14 +4253,24 @@ async fn delete_script_by_path( .fetch_all(&mut *tx) .await?; - let script = sqlx::query_scalar!( - "DELETE FROM script WHERE path = $1 AND workspace_id = $2 RETURNING path", + let deleted_hashes = sqlx::query_scalar!( + "DELETE FROM script WHERE path = $1 AND workspace_id = $2 RETURNING hash", path, w_id ) - .fetch_one(&mut *tx) + .fetch_all(&mut *tx) .await .map_err(|e| Error::internal_err(format!("deleting script by path {w_id}: {e:#}")))?; + if deleted_hashes.is_empty() { + return Err(Error::NotFound(format!( + "script {path} not found in {w_id}" + ))); + } + let script = path.to_string(); + let deleted = DeletedScriptVersions::new( + &w_id, + deleted_hashes.into_iter().map(|h| (script.clone(), h)), + ); // After the DELETE, never before: every dbt writer locks the `script` row // first, so taking a sidecar ahead of it deadlocks one of the pair. The @@ -4287,6 +4310,7 @@ async fn delete_script_by_path( // the script was never a pipeline member. clear_script_triggers(&mut *tx, &w_id, path, AssetUsageKind::Script).await?; clear_macro_registry(&mut *tx, &w_id, path).await?; + deleted.notify(&mut *tx).await?; if !query.keep_captures.unwrap_or(false) { sqlx::query!( @@ -4317,6 +4341,7 @@ async fn delete_script_by_path( ) .await?; tx.commit().await?; + evict_deleted_script_versions(deleted); handle_deployment_metadata( &authed.email, @@ -4412,14 +4437,21 @@ async fn delete_scripts_bulk( } } - let mut deleted_paths = sqlx::query_scalar!( - "DELETE FROM script WHERE workspace_id = $1 AND path = ANY($2) RETURNING path", - w_id, - &request.paths - ) - .fetch_all(&mut *tx) - .await - .map_err(|e| Error::internal_err(format!("deleting scripts in bulk {w_id}: {e:#}")))?; + let deleted = DeletedScriptVersions::new( + &w_id, + sqlx::query!( + "DELETE FROM script WHERE workspace_id = $1 AND path = ANY($2) RETURNING path, hash", + w_id, + &request.paths + ) + .fetch_all(&mut *tx) + .await + .map_err(|e| Error::internal_err(format!("deleting scripts in bulk {w_id}: {e:#}")))? + .into_iter() + .map(|r| (r.path, r.hash)), + ); + deleted.notify(&mut *tx).await?; + let deleted_paths = deleted.paths.clone(); // Same reason as the single-path delete, over every requested path rather // than the deleted ones: a path that had no script left can still hold state. @@ -4428,10 +4460,6 @@ async fn delete_scripts_bulk( windmill_common::dbt_manifest::clear_dbt_editor_graphs(&mut tx, &w_id, p).await?; } - // remove duplicates from deleted_paths - deleted_paths.sort(); - deleted_paths.dedup(); - sqlx::query!( "DELETE FROM draft WHERE workspace_id = $1 AND path = ANY($2) AND typ = 'script'", w_id, @@ -4472,6 +4500,7 @@ async fn delete_scripts_bulk( .await?; tx.commit().await?; + evict_deleted_script_versions(deleted); try_join_all(deleted_paths.iter().map(|path| { handle_deployment_metadata( diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index 8c891bf0cf..81fead981b 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -14367,18 +14367,27 @@ async fn prune_versions( let pruned = match req.resource_type.as_str() { "scripts" => { - let result = sqlx::query( - "DELETE FROM script - WHERE workspace_id = $1 AND hash NOT IN ( - SELECT DISTINCT ON (path) hash FROM script - WHERE workspace_id = $1 AND deleted = false - ORDER BY path, created_at DESC - )", - ) - .bind(&w_id) - .execute(&db) - .await?; - result.rows_affected() + let mut tx = db.begin().await?; + let deleted = windmill_common::DeletedScriptVersions::new( + &w_id, + sqlx::query_as::<_, (String, i64)>( + "DELETE FROM script + WHERE workspace_id = $1 AND hash NOT IN ( + SELECT DISTINCT ON (path) hash FROM script + WHERE workspace_id = $1 AND deleted = false + ORDER BY path, created_at DESC + ) + RETURNING path, hash", + ) + .bind(&w_id) + .fetch_all(&mut *tx) + .await?, + ); + deleted.notify(&mut *tx).await?; + tx.commit().await?; + let pruned = deleted.hashes.len() as u64; + deleted.evict(); + pruned } "flows" => { let deleted = sqlx::query( diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 250247c0ef..785c6ff935 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -4788,15 +4788,7 @@ pub async fn resume_suspended_flow_as_owner( .await?; if is_wac { - // WAC: directly decrement suspend counter - if flow.suspend > 0 { - sqlx::query!( - "UPDATE v2_job_queue SET suspend = GREATEST(suspend - 1, 0) WHERE id = $1", - flow.id, - ) - .execute(&mut *tx) - .await?; - } + decrement_suspend(flow.id, &mut tx).await?; } else { resume_immediately_if_relevant(flow, job_id, &mut tx).await?; } @@ -4947,12 +4939,7 @@ async fn resume_suspended( .execute(&mut *tx) .await?; } else if is_wac { - if flow.suspend > 0 { - sqlx::query("UPDATE v2_job_queue SET suspend = GREATEST(suspend - 1, 0) WHERE id = $1") - .bind(&flow.id) - .execute(&mut *tx) - .await?; - } + decrement_suspend(flow.id, &mut tx).await?; } else { resume_immediately_if_relevant(flow, resume_job_id, &mut tx).await?; } @@ -5430,8 +5417,13 @@ async fn resume_suspended_job_internal( let value = value.unwrap_or(serde_json::Value::Null); verify_suspended_secret(&w_id, &db, job_id, resume_id, &approver, secret).await?; - // Get flow info - works for step-level, flow-level, and WAC approval - let (flow_info, is_flow_level, is_wac) = get_flow_info_for_resume(job_id, &w_id, &db).await?; + let mut tx: Transaction<'_, Postgres> = db.begin().await?; + + // Get flow info - works for step-level, flow-level, and WAC approval. This locks the + // queue row until commit, so resumes serialize with each other and with the worker + // entering the suspend. + let (flow_info, is_flow_level, is_wac) = + get_flow_info_for_resume(job_id, &w_id, &mut tx).await?; // HMAC secret = full capability. Skip approval_conditions checks: possession of the full // resume URL is the authorization (it is only disclosed to intended approvers, e.g. when a @@ -5444,7 +5436,7 @@ async fn resume_suspended_job_internal( "#, Uuid::from_u128(job_id.as_u128() ^ resume_id as u128), ) - .fetch_one(&db) + .fetch_one(&mut *tx) .await? .unwrap_or(false); @@ -5462,7 +5454,6 @@ async fn resume_suspended_job_internal( } else { authed.as_ref().map(|x| x.username.clone()) }; - let mut tx: Transaction<'_, Postgres> = db.begin().await?; // Inside the transaction that inserts the row and moves the suspend counter: // validating earlier would let the workflow resolve this step and suspend on the @@ -5491,20 +5482,11 @@ async fn resume_suspended_job_internal( .await?; } else if is_wac { // WAC approval: decrement suspend counter directly on the WAC parent job. - // `flow_info.suspend` was read before this transaction took the queue-row - // lock, so gating on it would skip the decrement for a workflow that - // suspended in between and leave the approval parked until timeout. - sqlx::query!( - "UPDATE v2_job_queue SET suspend = GREATEST(suspend - 1, 0) \ - WHERE id = $1 AND suspend > 0", - flow_info.id, - ) - .execute(&mut *tx) - .await?; + decrement_suspend(flow_info.id, &mut tx).await?; } else if is_flow_level { - // For flow-level resumes, decrement the suspend counter if the flow is currently suspended - // The approval will be matched when the worker checks for resumes (both step-level and flow-level) - resume_immediately_for_flow_level(&flow_info, &mut tx).await?; + // Flow-level resumes don't match a step: the approval is picked up when the worker + // checks for resumes (both step-level and flow-level). + decrement_suspend(flow_info.id, &mut tx).await?; } else { // For step-level resumes, try to resume immediately if the step is waiting resume_immediately_if_relevant(flow_info, job_id, &mut tx).await?; @@ -5582,42 +5564,27 @@ async fn resume_immediately_if_relevant<'c>( job_id: Uuid, tx: &mut Transaction<'c, Postgres>, ) -> error::Result<()> { - Ok( - if let Some(suspend) = (0 < flow.suspend).then(|| flow.suspend - 1) { - let status = - serde_json::from_value::(flow.flow_status.context("no flow status")?) - .context("deserialize flow status")?; - if matches!(status.current_step(), Some(FlowStatusModule::WaitingForEvents { job, .. }) if job == &job_id) - { - sqlx::query!( - "UPDATE v2_job_queue SET suspend = $1 WHERE id = $2", - suspend, - flow.id, - ) - .execute(&mut **tx) - .await?; - } - }, - ) + Ok(if 0 < flow.suspend { + let status = + serde_json::from_value::(flow.flow_status.context("no flow status")?) + .context("deserialize flow status")?; + if matches!(status.current_step(), Some(FlowStatusModule::WaitingForEvents { job, .. }) if job == &job_id) + { + decrement_suspend(flow.id, tx).await?; + } + }) } -/// For flow-level resumes, decrement the suspend counter if the flow is currently suspended. -/// Unlike step-level resumes, we don't check if the job_id matches - we just need the flow -/// to be in a suspended state. -async fn resume_immediately_for_flow_level<'c>( - flow: &FlowInfo, - tx: &mut Transaction<'c, Postgres>, -) -> error::Result<()> { - if flow.suspend > 0 { - let new_suspend = flow.suspend - 1; - sqlx::query!( - "UPDATE v2_job_queue SET suspend = $1 WHERE id = $2", - new_suspend, - flow.id, - ) - .execute(&mut **tx) - .await?; - } +/// Relative rather than `suspend = - 1`, so concurrent approvals each remove one +/// event from the count instead of overwriting each other's decrement. +async fn decrement_suspend(flow_id: Uuid, tx: &mut Transaction<'_, Postgres>) -> error::Result<()> { + sqlx::query!( + "UPDATE v2_job_queue SET suspend = GREATEST(suspend - 1, 0) \ + WHERE id = $1 AND suspend > 0", + flow_id, + ) + .execute(&mut **tx) + .await?; Ok(()) } @@ -5666,10 +5633,10 @@ struct FlowInfo { async fn get_flow_info_for_resume( job_id: Uuid, w_id: &str, - db: &DB, + tx: &mut Transaction<'_, Postgres>, ) -> error::Result<(FlowInfo, bool, bool)> { // Single query that determines if job_id is a flow, step, or WAC job, - // and fetches the appropriate suspended job info. + // and locks the appropriate suspended job's queue row. // For WAC jobs (no parent, not a flow), the job itself is the suspended target. let result = sqlx::query!( r#" @@ -5680,7 +5647,6 @@ async fn get_flow_info_for_resume( ) SELECT q.id AS "id!", - s.flow_status, q.suspend AS "suspend!", j.runnable_path AS script_path, j.permissioned_as_email AS email, @@ -5692,13 +5658,12 @@ async fn get_flow_info_for_resume( ELSE COALESCE(ji.parent_job, ji.id) END JOIN v2_job j ON j.id = q.id - LEFT JOIN v2_job_status s ON s.id = q.id FOR UPDATE OF q "#, job_id, w_id, ) - .fetch_optional(db) + .fetch_optional(&mut **tx) .await? .ok_or_else(|| { Error::NotFound(format!( @@ -5706,9 +5671,18 @@ async fn get_flow_info_for_resume( )) })?; + // Read in its own statement: a statement that waited on the queue-row lock above re-reads + // only the locked row, so a status joined into it could predate the worker's commit. + let flow_status: Option = + sqlx::query_scalar("SELECT flow_status FROM v2_job_status WHERE id = $1") + .bind(result.id) + .fetch_optional(&mut **tx) + .await? + .flatten(); + let flow_info = FlowInfo { id: result.id, - flow_status: result.flow_status, + flow_status, suspend: result.suspend, script_path: result.script_path, email: Some(result.email), @@ -5721,6 +5695,13 @@ async fn get_suspended_flow_info<'c>( job_id: Uuid, tx: &mut Transaction<'c, Postgres>, ) -> error::Result<(FlowInfo, Uuid, bool)> { + // Lock before reading: a single statement that waited on this lock would re-read only + // the queue row, pairing a fresh suspend counter with a stale flow status. + sqlx::query("SELECT 1 FROM v2_job_queue WHERE id = $1 FOR UPDATE") + .bind(job_id) + .execute(&mut **tx) + .await?; + let flow = sqlx::query_as!( FlowInfo, r#" diff --git a/backend/windmill-common/src/cache.rs b/backend/windmill-common/src/cache.rs index 3c385da79f..1518e8e8e3 100644 --- a/backend/windmill-common/src/cache.rs +++ b/backend/windmill-common/src/cache.rs @@ -687,8 +687,9 @@ pub mod script { schema_validation AS \"schema_validation: bool\", \ codebase LIKE '%.tar' as use_tar, \ codebase LIKE '%.esm%' as is_esm, \ - modules AS \"modules: serde_json::Value\" \ - FROM script WHERE hash = $1 LIMIT 1", + modules AS \"modules: serde_json::Value\", \ + deleted \ + FROM script WHERE hash = $1 ORDER BY deleted LIMIT 1", hash.0 ) .fetch_optional(db) @@ -696,6 +697,14 @@ pub mod script { .map_err(Into::into) .and_then(unwrap_or_error(&loc, "Script", hash)) .and_then(|r| { + // The hash leaves out the workspace, so forks and clones share it, and a deleted + // copy keeps its row with the content wiped. Any live copy holds the same code; + // only when every copy is deleted is there nothing left to run. + if r.deleted { + return Err(error::Error::NotFound(format!( + "Script version {hash} was deleted" + ))); + } Ok(RawScript { content: r.content, lock: r.lock, diff --git a/backend/windmill-common/src/job_provenance.rs b/backend/windmill-common/src/job_provenance.rs index e3dfdbc205..63075b0c37 100644 --- a/backend/windmill-common/src/job_provenance.rs +++ b/backend/windmill-common/src/job_provenance.rs @@ -1,12 +1,19 @@ use uuid::Uuid; -use crate::{db::DB, error::Result, jobs::JobKind}; +use crate::{db::DB, error::Result, jobs::JobKind, scripts::ScriptHash}; /// Where a job's code comes from, as far as its chain of parents can prove it. pub struct JobProvenance { /// True only when the job and every ancestor run code at a path taken from a /// deployed version, never from the request that created the job. pub deployed: bool, + /// Deployed, and every script and flow in the chain runs the version its path + /// currently resolves to. Any past version still counts as deployed, and a run started + /// before a redeploy stops being latest once the redeploy lands. App runs never count: + /// app scripts are not tied to an app version. + pub latest: bool, + /// The job's own script hash (hex) or flow version id; `None` for other kinds. + pub version: Option, /// The chain reached a parent it cannot read (other workspace, depth cap), so /// nothing above that point is known. pub unproven_ancestry: bool, @@ -21,6 +28,7 @@ pub struct JobProvenance { pub root_path: Option, pub root_kind: JobKind, pub root_trigger_kind: Option, + pub root_version: Option, } #[derive(Debug, PartialEq, Eq)] @@ -48,9 +56,14 @@ struct LineageJob { parent_job: Option, kind: JobKind, runnable_path: Option, + runnable_id: Option, trigger_kind: Option, permissioned_as: String, origin_verified: bool, + current_version: bool, + restarted: bool, + /// The job a restart rebuilt this one from. + restarted_from: Option, app_stamped: bool, args_modules: bool, } @@ -69,7 +82,7 @@ pub async fn job_provenance(db: &DB, job_id: &Uuid, w_id: &str) -> Result