chore(main): release 1.628.2 (#7828)

* chore(main): release 1.628.2

* Apply automatic changes

---------

Co-authored-by: rubenfiszel <275584+rubenfiszel@users.noreply.github.com>

.github/workflows/backend-check.yml

@@ -33,10 +33,10 @@ jobs:
         with:
           fetch-depth: 0
 
-      - name: install xmlsec1
+      - name: install xmlsec1 and gssapi
         run: |
           sudo apt-get update
-          sudo apt-get install -y libxml2-dev libxmlsec1-dev
+          sudo apt-get install -y libxml2-dev libxmlsec1-dev libkrb5-dev libsasl2-dev
 
       - uses: actions-rust-lang/setup-rust-toolchain@v1
         with:
@@ -100,10 +100,10 @@ jobs:
           token: ${{ secrets.WINDMILL_EE_PRIVATE_ACCESS }}
           fetch-depth: 0
 
-      - name: install xmlsec1
+      - name: install xmlsec1 and gssapi
         run: |
           sudo apt-get update
-          sudo apt-get install -y libxml2-dev libxmlsec1-dev
+          sudo apt-get install -y libxml2-dev libxmlsec1-dev libkrb5-dev libsasl2-dev
 
       - name: Substitute EE code (EE logic is behind feature flag)
         run: |

.github/workflows/claude.yml

@@ -75,10 +75,10 @@ jobs:
           npm install
           npm run generate-backend-client
 
-      - name: install xmlsec1
+      - name: install xmlsec1 and gssapi
         run: |
           sudo apt-get update
-          sudo apt-get install -y libxml2-dev libxmlsec1-dev
+          sudo apt-get install -y libxml2-dev libxmlsec1-dev libkrb5-dev libsasl2-dev
 
       - uses: actions-rust-lang/setup-rust-toolchain@v1
         with:

.github/workflows/docker-image.yml

@@ -24,11 +24,6 @@ on:
         description: "Tag the image"
         required: true
         default: "test"
-      nsjail:
-        description: "Build nsjail image (true, false)"
-        required: false
-        default: false
-        type: boolean
       slim:
         description: "Build slim image (true, false)"
         required: false
@@ -106,7 +101,7 @@ jobs:
 
   build_ee:
     runs-on: ubicloud
-    if: (github.event_name != 'workflow_dispatch') || (github.event.inputs.ee || github.event.inputs.nsjail)
+    if: (github.event_name != 'workflow_dispatch') || github.event.inputs.ee
     steps:
       - uses: actions/checkout@v4
         with:
@@ -370,67 +365,10 @@ jobs:
   #         ignore-unchanged: true
   #         only-fixed: true
 
-  build_ee_nsjail:
-    needs: [build_ee]
-    runs-on: ubicloud
-    if: (github.event_name != 'pull_request') && ((github.event_name != 'workflow_dispatch') || (github.event.inputs.ee || github.event.inputs.nsjail))
-
-    steps:
-      - uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
-          ref: ${{ github.ref }}
-
-      # - name: Set up Docker Buildx
-      #   uses: docker/setup-buildx-action@v2
-
-      - uses: depot/setup-action@v1
-
-      - name: Docker meta
-        id: meta-ee-public
-        uses: docker/metadata-action@v5
-        with:
-          images: |
-            ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}-ee-nsjail
-          flavor: |
-            latest=false
-          tags: |
-            type=semver,pattern={{version}}
-            type=semver,pattern={{major}}.{{minor}}
-            type=sha,enable=true,priority=100,prefix=,suffix=,format=short
-            type=ref,event=branch
-            type=ref,event=pr
-
-      - name: Login to registry
-        uses: docker/login-action@v3
-        with:
-          registry: ${{ env.REGISTRY }}
-          username: ${{ github.actor }}
-          password: ${{ secrets.GITHUB_TOKEN }}
-
-      - name: Update Dockerfile image reference
-        run: |
-          sed -i 's|FROM ghcr.io/windmill-labs/windmill-ee:dev|FROM ghcr.io/${{ env.IMAGE_NAME }}-ee:${{ env.DEV_SHA }}|' ./docker/DockerfileNsjail
-          cat ./docker/DockerfileNsjail | grep "FROM"
-
-      - name: Build and push publicly ee
-        uses: depot/build-push-action@v1
-        with:
-          context: .
-          platforms: linux/amd64,linux/arm64
-          push: true
-          file: "./docker/DockerfileNsjail"
-          tags: |
-            ${{ steps.meta-ee-public.outputs.tags }}
-          labels: |
-            ${{ steps.meta-ee-public.outputs.labels }}
-            org.opencontainers.image.licenses=Windmill-Enterprise-License
-
   publish_ecr_s3:
-    needs: [build_ee_nsjail]
+    needs: [build_ee_full]
     runs-on: ubicloud-standard-2-arm
-    if: (github.event_name != 'pull_request') && (github.event_name !=
-      'workflow_dispatch')
+    if: ${{ startsWith(github.ref, 'refs/tags/v') }}
     env:
       AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
       AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
@@ -449,23 +387,18 @@ jobs:
           username: ${{ github.actor }}
           password: ${{ secrets.GITHUB_TOKEN }}
 
-      - name: get git hash
-        if: github.event_name != 'pull_request'
-        id: git_hash
-        run: |
-          git_hash=$(git rev-parse --short "$GITHUB_SHA")
-          echo "GIT_HASH=${git_hash:0:7}" >> "$GITHUB_OUTPUT"
+      - name: Get version from tag
+        id: version
+        run: echo "VERSION=${GITHUB_REF_NAME#v}" >> "$GITHUB_OUTPUT"
 
       - uses: shrink/actions-docker-extract@v3
-        if: github.event_name != 'pull_request'
         id: extract
         with:
           image: |-
-            ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}-ee-nsjail:${{ steps.git_hash.outputs.GIT_HASH }}
+            ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}-ee-full:${{ steps.version.outputs.VERSION }}
           path: "/static_frontend/."
 
       - uses: reggionick/s3-deploy@v4
-        if: github.event_name != 'pull_request'
         with:
           folder: ${{ steps.extract.outputs.destination }}
           bucket: windmill-frontend

.github/workflows/git-commands.yaml

@@ -68,11 +68,11 @@ jobs:
         with:
           workspaces: "./backend -> target"
 
-      - name: Install xmlsec build-time deps
+      - name: Install xmlsec and gssapi build-time deps
         run: |
           sudo apt-get update
           sudo apt-get install -y --no-install-recommends \
-            pkg-config libxml2-dev libssl-dev \
+            pkg-config libxml2-dev libssl-dev libkrb5-dev \
             xmlsec1 libxmlsec1-dev libxmlsec1-openssl
 
       - name: Run update-sqlx script

.github/workflows/spawn-ephemeral-backend.yml

@@ -3,6 +3,8 @@ name: Spawn Ephemeral Backend
 on:
   issue_comment:
     types: [created]
+  pull_request_review_comment:
+    types: [created]
   workflow_dispatch:
     inputs:
       pr_number:
@@ -11,11 +13,42 @@ on:
         type: number
 
 jobs:
+  determine-commenter:
+    if: |
+      (github.event_name == 'issue_comment' && contains(github.event.comment.body, '/spawnbackend')) ||
+      (github.event_name == 'pull_request_review_comment' && contains(github.event.comment.body, '/spawnbackend'))
+    runs-on: ubicloud-standard-2
+    outputs:
+      commenter: ${{ steps.determine-commenter.outputs.commenter }}
+    steps:
+      - name: Determine commenter
+        id: determine-commenter
+        run: |
+          # Work out who wrote the comment / review
+          if [[ "${{ github.event_name }}" == "issue_comment" || \
+                "${{ github.event_name }}" == "pull_request_review_comment" ]]; then
+            COMMENTER="${{ github.event.comment.user.login }}"
+          elif [[ "${{ github.event_name }}" == "pull_request_review" ]]; then
+            COMMENTER="${{ github.event.review.user.login }}"
+          else
+            COMMENTER="${{ github.event.issue.user.login }}"
+          fi
+          echo "commenter=$COMMENTER" >> $GITHUB_OUTPUT
+
+  check-membership:
+    needs: determine-commenter
+    uses: ./.github/workflows/check-org-membership.yml
+    with:
+      commenter: ${{ needs.determine-commenter.outputs.commenter }}
+    secrets:
+      access_token: ${{ secrets.ORG_ACCESS_TOKEN }}
+
   spawn-backend:
+    needs: [determine-commenter, check-membership]
     # Only run on PR comments that contain /spawn-backend, or manual dispatch
     if: |
       github.event_name == 'workflow_dispatch' ||
-      (github.event.issue.pull_request && contains(github.event.comment.body, '/spawn-backend'))
+      (github.event.issue.pull_request && needs.check-membership.outputs.is_member == 'true')
     runs-on: ubuntu-latest
     permissions:
       pull-requests: write
@@ -36,40 +69,80 @@ jobs:
               repo: context.repo.repo,
               pull_number: prNumber
             });
+
+            // Get branch name and format it for Cloudflare Pages
+            // Replace '/' with '-' for the URL
+            const branchName = pr.data.head.ref;
+            const formattedBranch = branchName.replace(/\//g, '-');
+            const cfFrontendUrl = `https://${formattedBranch}.windmill.pages.dev`;
+
             core.setOutput('commit_hash', pr.data.head.sha);
-            core.setOutput('pr_number', context.issue.number);
+            core.setOutput('pr_number', prNumber);
+            core.setOutput('branch_name', branchName);
+            core.setOutput('cf_frontend_url', cfFrontendUrl);
 
-      - name: Trigger Windmill flow
-        id: trigger-flow
+      - name: Check manager URL
+        id: check-manager-url
         run: |
-          RESPONSE=$(curl -s -X POST "https://app.windmill.dev/api/w/windmill-labs/jobs/run/f/f/all/run_ephemeral_backend" \
-            -H "Authorization: Bearer ${{ secrets.WINDMILL_TOKEN }}" \
-            -H "Content-Type: application/json" \
-            -d '{
-              "manager_url": "${{ secrets.EPHEMERAL_BACKEND_QUEUE_URL }}",
-              "commit_hash": "${{ steps.pr-details.outputs.commit_hash }}",
-              "pr_number": ${{ steps.pr-details.outputs.pr_number }}
-            }')
-
-          JOB_UUID=$(echo "$RESPONSE" | jq -r '.id // empty')
-
-          if [ -z "$JOB_UUID" ]; then
-            echo "Failed to get job UUID from response: $RESPONSE"
-            exit 1
+          if [ -z "${{ secrets.EPHEMERAL_BACKEND_QUEUE_URL }}" ]; then
+            echo "manager_url_set=false" >> $GITHUB_OUTPUT
+          else
+            echo "manager_url_set=true" >> $GITHUB_OUTPUT
           fi
 
-          echo "job_uuid=$JOB_UUID" >> $GITHUB_OUTPUT
-
-      - name: Post comment with job link
+      - name: Post error comment if manager not running
+        if: steps.check-manager-url.outputs.manager_url_set == 'false'
         uses: actions/github-script@v7
         with:
           script: |
-            const jobUuid = '${{ steps.trigger-flow.outputs.job_uuid }}';
-            const jobUrl = `https://app.windmill.dev/run/${jobUuid}?workspace=windmill-labs`;
+            const prNumber = context.eventName === 'workflow_dispatch'
+              ? Number(context.payload.inputs.pr_number)
+              : context.issue.number;
 
             await github.rest.issues.createComment({
               owner: context.repo.owner,
               repo: context.repo.repo,
-              issue_number: context.issue.number,
-              body: `🚀 Ephemeral backend spawning started!\n\nView job progress: ${jobUrl}`
+              issue_number: prNumber,
+              body: `❌ Manager URL not set (did you start the ephemeral backend manager?)\n\nThe ephemeral backend manager needs to be running to spawn backends. Please start the manager first.`
+            });
+
+      - name: Fail if manager not running
+        if: steps.check-manager-url.outputs.manager_url_set == 'false'
+        run: |
+          echo "Error: EPHEMERAL_BACKEND_QUEUE_URL secret is not set"
+          exit 1
+
+      - name: Trigger Windmill flow
+        if: steps.check-manager-url.outputs.manager_url_set == 'true'
+        id: trigger-flow
+        run: |
+          JOB_UUID=$(curl -s -X POST "https://app.windmill.dev/api/w/windmill-labs/jobs/run/f/f/all/run_ephemeral_backend" \
+            -H "Authorization: Bearer ${{ secrets.WINDMILL_RUN_FLOW_TOKEN }}" \
+            -H "Content-Type: application/json" \
+            -d '{
+              "manager_url": "${{ secrets.EPHEMERAL_BACKEND_QUEUE_URL }}",
+              "commit_hash": "${{ steps.pr-details.outputs.commit_hash }}",
+              "pr_number": ${{ steps.pr-details.outputs.pr_number }},
+              "cf_frontend_url": "${{ steps.pr-details.outputs.cf_frontend_url }}"
+            }' | tr -d '"')
+
+          echo "Job UUID: $JOB_UUID"
+          echo "job_uuid=$JOB_UUID" >> $GITHUB_OUTPUT
+
+      - name: Post comment with job link
+        if: steps.check-manager-url.outputs.manager_url_set == 'true'
+        uses: actions/github-script@v7
+        with:
+          script: |
+            const jobUuid = '${{ steps.trigger-flow.outputs.job_uuid }}';
+            const appUrl = `https://app.windmill.dev/public/windmill-labs/a106bad0256c1dfa7a4f9279c42b1a4b#${jobUuid}`;
+            const prNumber = context.eventName === 'workflow_dispatch'
+              ? Number(context.payload.inputs.pr_number)
+              : context.issue.number;
+
+            await github.rest.issues.createComment({
+              owner: context.repo.owner,
+              repo: context.repo.repo,
+              issue_number: prNumber,
+              body: `🚀 Spawning new ephemeral backend!\n\n${appUrl}`
             });

CHANGELOG.md

@@ -1,5 +1,85 @@
 # Changelog
 
+## [1.628.2](https://github.com/windmill-labs/windmill/compare/v1.628.1...v1.628.2) (2026-02-06)
+
+
+### Bug Fixes
+
+* execute CONCURRENTLY statements individually in migrations ([7a7b118](https://github.com/windmill-labs/windmill/commit/7a7b118bf36e8086b2df5a940e99e1c031f57c81))
+
+## [1.628.1](https://github.com/windmill-labs/windmill/compare/v1.628.0...v1.628.1) (2026-02-06)
+
+
+### Bug Fixes
+
+* prevent deadlock in consolidate live index migration ([f39b28a](https://github.com/windmill-labs/windmill/commit/f39b28ac416cfdc2420a58b549ee07479f316493))
+* use concurrent index ops to prevent deadlock on upgrade ([9967f83](https://github.com/windmill-labs/windmill/commit/9967f835ab0cba04bdad4f72b7df786bd1b02fa0))
+
+## [1.628.0](https://github.com/windmill-labs/windmill/compare/v1.627.0...v1.628.0) (2026-02-06)
+
+
+### Features
+
+* kafka trigger kerberos/gssapi support ([#7815](https://github.com/windmill-labs/windmill/issues/7815)) ([795e2be](https://github.com/windmill-labs/windmill/commit/795e2bebe65db9c6f721e7cd24af1446aeb896ab))
+
+
+### Bug Fixes
+
+* make notify_event trigger functions SECURITY DEFINER ([#7826](https://github.com/windmill-labs/windmill/issues/7826)) ([33fb08c](https://github.com/windmill-labs/windmill/commit/33fb08cf3d08c4a6b86f32b3ae8bf2df8c1adcaa))
+* prevent schedule pool connection exhaustion ([#7821](https://github.com/windmill-labs/windmill/issues/7821)) ([e655a06](https://github.com/windmill-labs/windmill/commit/e655a065637b288080118661650bc14641dd0c6f))
+
+## [1.627.0](https://github.com/windmill-labs/windmill/compare/v1.626.0...v1.627.0) (2026-02-05)
+
+
+### Features
+
+* mssql integrated auth (gssapi) ([#7760](https://github.com/windmill-labs/windmill/issues/7760)) ([afa6e7a](https://github.com/windmill-labs/windmill/commit/afa6e7ab5bb26972acbfe19af41dd3e6ac5df363))
+* restriction rulesets for workspaces ([#7791](https://github.com/windmill-labs/windmill/issues/7791)) ([a1cd02d](https://github.com/windmill-labs/windmill/commit/a1cd02d7f80c97eb07eda04113f3aae815fada69))
+
+
+### Bug Fixes
+
+* allow unauthed private pwsh repo ([#7817](https://github.com/windmill-labs/windmill/issues/7817)) ([476e6fd](https://github.com/windmill-labs/windmill/commit/476e6fd4bd2cdb062fa15a8c4048371ef4b31845))
+* fix asset grant ([e28c5b1](https://github.com/windmill-labs/windmill/commit/e28c5b18af25710b0ed3a3ffbceb18f3da76cd75))
+
+## [1.626.0](https://github.com/windmill-labs/windmill/compare/v1.625.0...v1.626.0) (2026-02-05)
+
+
+### Features
+
+* **local-dev:** create Claude skills when doing `wmill init` ([#7699](https://github.com/windmill-labs/windmill/issues/7699)) ([a7ce548](https://github.com/windmill-labs/windmill/commit/a7ce5484b8ec386af59f501c36e5ffc147e1d34a))
+
+
+### Bug Fixes
+
+* fix DB Manager not working with db resources with 4+ path segments ([#7809](https://github.com/windmill-labs/windmill/issues/7809)) ([3476ef4](https://github.com/windmill-labs/windmill/commit/3476ef4b9c795fb8511a83f2297154a4f55aa829))
+* fix indexer select performances busiying the db ([c3815c8](https://github.com/windmill-labs/windmill/commit/c3815c8c99d5b7d6b2dfc0e3b59d1ba51022ee39))
+* **frontend:** dedicated worker broken runnable select ([#7808](https://github.com/windmill-labs/windmill/issues/7808)) ([6f6ff9d](https://github.com/windmill-labs/windmill/commit/6f6ff9d4217e99901562b01eb258c7ccdcb0e3f4))
+* python client oidc pass session token ([#7799](https://github.com/windmill-labs/windmill/issues/7799)) ([b468603](https://github.com/windmill-labs/windmill/commit/b468603f6bc52961057fbd88539eb379a19efd9d))
+
+## [1.625.0](https://github.com/windmill-labs/windmill/compare/v1.624.0...v1.625.0) (2026-02-04)
+
+
+### Features
+
+* add filters to Kafka triggers ([#7750](https://github.com/windmill-labs/windmill/issues/7750)) ([3c8daa9](https://github.com/windmill-labs/windmill/commit/3c8daa9a58b5e4a2e8c85a9805a5b194ed75d055))
+* Assets page exploration UI ([#7784](https://github.com/windmill-labs/windmill/issues/7784)) ([0508425](https://github.com/windmill-labs/windmill/commit/05084254a34da81d227813a5190e3ce3dc0f816e))
+* cache lockfile results for scripts with same raw_workspace_dependencies ([#7787](https://github.com/windmill-labs/windmill/issues/7787)) ([4098679](https://github.com/windmill-labs/windmill/commit/4098679fd7eca059dfa128a6f8b8e1698a65b632))
+* column-level asset tracking for ducklake and datatables ([#7774](https://github.com/windmill-labs/windmill/issues/7774)) ([0caa533](https://github.com/windmill-labs/windmill/commit/0caa533fbd70fffec27d86d62e16bb92cf7a612a))
+* favorite datatable and ducklake tables + asset page nits ([#7795](https://github.com/windmill-labs/windmill/issues/7795)) ([a3d75ba](https://github.com/windmill-labs/windmill/commit/a3d75ba10ae85e5ecb55351555879be7fe0bfcca))
+* make nsjail available in all standard images (CE) ([#7793](https://github.com/windmill-labs/windmill/issues/7793)) ([149da9b](https://github.com/windmill-labs/windmill/commit/149da9b763e4f5dd93d2905be89b5df81bb61934))
+* public app rate limiting + fork hub raw apps + raw apps publish to hub button ([#7789](https://github.com/windmill-labs/windmill/issues/7789)) ([63f9d85](https://github.com/windmill-labs/windmill/commit/63f9d85bf6a5dd25977995978a8b0a4d32fee995))
+* replace LISTEN/NOTIFY with polling-based event system ([#7778](https://github.com/windmill-labs/windmill/issues/7778)) ([e860847](https://github.com/windmill-labs/windmill/commit/e860847073b56be469ba37af5e3a8cb7d30ef7bc))
+* upgrade bun to v1.3.8 with regression tests ([#7761](https://github.com/windmill-labs/windmill/issues/7761)) ([ef89a51](https://github.com/windmill-labs/windmill/commit/ef89a51f3a1cc1ae562d97b413c78393c0ea92cf))
+
+
+### Bug Fixes
+
+* fix forking raw apps and summary setting in deploy drawer ([#7792](https://github.com/windmill-labs/windmill/issues/7792)) ([db56518](https://github.com/windmill-labs/windmill/commit/db56518e4fc53931e3498db06bbefd511c343d23))
+* handle Date serialization in quickjs flow eval via toJSON ([f151fdc](https://github.com/windmill-labs/windmill/commit/f151fdcf7f91a7b0ac75a133d5193538f4a9b4d8))
+* make private registries settings password in the instance settings ([727bd21](https://github.com/windmill-labs/windmill/commit/727bd2164059e4d44f2e2f6f70a567e7fac3a921))
+* persist ws_error_handler_muted for flows in create/update ([#7797](https://github.com/windmill-labs/windmill/issues/7797)) ([d113546](https://github.com/windmill-labs/windmill/commit/d113546169a790997d4842b7cfeb43ec2c90c6ea))
+
 ## [1.624.0](https://github.com/windmill-labs/windmill/compare/v1.623.1...v1.624.0) (2026-02-03)
 
 

Dockerfile

@@ -1,6 +1,26 @@
 ARG DEBIAN_IMAGE=debian:bookworm-slim
 ARG RUST_IMAGE=rust:1.90-slim-bookworm
 
+FROM debian:bookworm-slim AS nsjail
+
+WORKDIR /nsjail
+
+RUN apt-get -y update \
+    && apt-get install -y \
+    bison=2:3.8.* \
+    flex=2.6.* \
+    g++=4:12.2.* \
+    gcc=4:12.2.* \
+    git=1:2.39.* \
+    libprotobuf-dev=3.21.* \
+    libnl-route-3-dev=3.7.* \
+    make=4.3-4.1 \
+    pkg-config=1.8.* \
+    protobuf-compiler=3.21.*
+
+RUN git clone -b master --single-branch https://github.com/google/nsjail.git . && git checkout dccf911fd2659e7b08ce9507c25b2b38ec2c5800
+RUN make
+
 FROM ${RUST_IMAGE} AS rust_base
 
 RUN apt-get update && apt-get install -y git libssl-dev pkg-config npm
@@ -77,7 +97,7 @@ ARG features=""
 
 COPY --from=planner /windmill/recipe.json recipe.json
 
-RUN apt-get update && apt-get install -y libxml2-dev=2.9.* libxmlsec1-dev=1.2.* clang=1:14.0-55.* libclang-dev=1:14.0-55.* cmake=3.25.* && \
+RUN apt-get update && apt-get install -y libxml2-dev=2.9.* libxmlsec1-dev=1.2.* libkrb5-dev libsasl2-dev clang=1:14.0-55.* libclang-dev=1:14.0-55.* cmake=3.25.* && \
     apt-get clean && \
     rm -rf /var/lib/apt/lists/*
 
@@ -129,7 +149,7 @@ ENV PATH /usr/local/bin:/root/.local/bin:/tmp/.local/bin:$PATH
 
 
 RUN apt-get update \
-    && apt-get install -y --no-install-recommends netbase tzdata ca-certificates wget curl jq unzip build-essential unixodbc xmlsec1 software-properties-common tini \
+    && apt-get install -y --no-install-recommends netbase tzdata ca-certificates wget curl jq unzip build-essential unixodbc xmlsec1 software-properties-common tini libsasl2-modules-gssapi-mit \
     && apt-get clean \
     && rm -rf /var/lib/apt/lists/*
 
@@ -246,6 +266,11 @@ ENV RUSTUP_HOME="/usr/local/rustup"
 ENV CARGO_HOME="/usr/local/cargo"
 ENV LD_LIBRARY_PATH="."
 
+# nsjail runtime deps and binary
+RUN apt-get update && apt-get install -y libprotobuf-dev libnl-route-3-dev \
+    && apt-get clean && rm -rf /var/lib/apt/lists/*
+COPY --from=nsjail /nsjail/nsjail /bin/nsjail
+
 WORKDIR ${APP}
 
 RUN ln -s ${APP}/windmill /usr/local/bin/windmill

backend/.sqlx/query-08f288d2781d823e109a9e5b8848234ca7d1efeee9661f3901f298da375e73f7.json

@@ -152,6 +152,11 @@
         "ordinal": 29,
         "name": "success_handler",
         "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 30,
+        "name": "public_app_execution_limit_per_minute",
+        "type_info": "Int4"
       }
     ],
     "parameters": {
@@ -189,6 +194,7 @@
       true,
       true,
       true,
+      true,
       true
     ]
   },

backend/.sqlx/query-1c5caaaa86e3488549cad179e992172315be5d53dcc266d713da01fd27f310b6.json

@@ -13,7 +13,8 @@
             "kind": {
               "Enum": [
                 "script",
-                "flow"
+                "flow",
+                "job"
               ]
             }
           }

backend/.sqlx/query-22e14fc3bb5d8cf3006f0002e8522b8cc0b2fece43f03c0f025e7acefa0d4f32.json

@@ -16,7 +16,8 @@
                 "app",
                 "script",
                 "flow",
-                "raw_app"
+                "raw_app",
+                "asset"
               ]
             }
           }

backend/.sqlx/query-31eada57708ac3347d1e0a4fb5fd412a4a0c6046dbe4cf91849def2eea2e4c62.json

@@ -1,53 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind, asset_detection_kind, job_id)\n                VALUES ($1, $2, $3, $4, $5, $6, 'static', NULL) ON CONFLICT DO NOTHING",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Varchar",
-        "Varchar",
-        {
-          "Custom": {
-            "name": "asset_kind",
-            "kind": {
-              "Enum": [
-                "s3object",
-                "resource",
-                "variable",
-                "ducklake",
-                "datatable"
-              ]
-            }
-          }
-        },
-        {
-          "Custom": {
-            "name": "asset_access_type",
-            "kind": {
-              "Enum": [
-                "r",
-                "w",
-                "rw"
-              ]
-            }
-          }
-        },
-        "Varchar",
-        {
-          "Custom": {
-            "name": "asset_usage_kind",
-            "kind": {
-              "Enum": [
-                "script",
-                "flow"
-              ]
-            }
-          }
-        }
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "31eada57708ac3347d1e0a4fb5fd412a4a0c6046dbe4cf91849def2eea2e4c62"
-}

backend/.sqlx/query-3b6bd7b41f130ce6df62fdecb351a3e01be0726d02d3f863e0ea5c476a8e785e.json

@@ -1,26 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "\n            UPDATE kafka_trigger \n            SET \n                kafka_resource_path = $1,\n                group_id = $2,\n                topics = $3,\n                script_path = $4,\n                path = $5,\n                is_flow = $6,\n                edited_by = $7,\n                email = $8,\n                edited_at = now(),\n                server_id = NULL,\n                error = NULL,\n                error_handler_path = $11,\n                error_handler_args = $12,\n                retry = $13\n            WHERE \n                workspace_id = $9 AND path = $10\n            ",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Varchar",
-        "Varchar",
-        "VarcharArray",
-        "Varchar",
-        "Varchar",
-        "Bool",
-        "Varchar",
-        "Varchar",
-        "Text",
-        "Text",
-        "Varchar",
-        "Jsonb",
-        "Jsonb"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "3b6bd7b41f130ce6df62fdecb351a3e01be0726d02d3f863e0ea5c476a8e785e"
-}

backend/.sqlx/query-3c6bed058591ad5438428b8c154ba6ee737e59aa033d833843b54e9f58eef3f2.json

@@ -1,42 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind, asset_detection_kind, job_id)\n                VALUES ($1, $2, $3, NULL, $4, $5, 'runtime', $6) ON CONFLICT DO NOTHING",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Varchar",
-        "Varchar",
-        {
-          "Custom": {
-            "name": "asset_kind",
-            "kind": {
-              "Enum": [
-                "s3object",
-                "resource",
-                "variable",
-                "ducklake",
-                "datatable"
-              ]
-            }
-          }
-        },
-        "Varchar",
-        {
-          "Custom": {
-            "name": "asset_usage_kind",
-            "kind": {
-              "Enum": [
-                "script",
-                "flow"
-              ]
-            }
-          }
-        },
-        "Uuid"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "3c6bed058591ad5438428b8c154ba6ee737e59aa033d833843b54e9f58eef3f2"
-}

backend/.sqlx/query-3fe6f5d77332cce5ad249b8d6e1ea34aa57650c6effc3a9a2f4f720ea934669b.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE workspace_settings SET public_app_execution_limit_per_minute = $1 WHERE workspace_id = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Int4",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "3fe6f5d77332cce5ad249b8d6e1ea34aa57650c6effc3a9a2f4f720ea934669b"
+}

backend/.sqlx/query-46395d77f3393a058e25ff4efb0b61c743c58418ba2ae1104bcfc99fae7925e4.json

@@ -1,15 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "DELETE FROM asset WHERE workspace_id = $1 AND usage_kind = 'script' AND usage_path = (SELECT path FROM script WHERE hash = $2 AND workspace_id = $1) AND asset_detection_kind = 'static'",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Int8"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "46395d77f3393a058e25ff4efb0b61c743c58418ba2ae1104bcfc99fae7925e4"
-}

backend/.sqlx/query-4f666058177fed05c25036852f772d3cc5e2a5f947f307597b5a4a50f571c89b.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT\n            jsonb_strip_nulls(jsonb_build_object(\n                'path', favorite.path\n            )) as \"favorite_asset!: _\"\n        FROM favorite\n        WHERE favorite.workspace_id = $1\n          AND favorite.usr = $2\n          AND favorite_kind = 'asset'\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "favorite_asset!: _",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "4f666058177fed05c25036852f772d3cc5e2a5f947f307597b5a4a50f571c89b"
+}

backend/.sqlx/query-551fee7919fdeb911e3f9cc5852e158ea47e3db4895c2b2b1d3cb6b16fceeda9.json

@@ -1,23 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "SELECT fv.id\n                                                                                FROM flow f\n                                                                                INNER JOIN flow_version fv ON fv.id = f.versions[array_upper(f.versions, 1)]\n                                                                                WHERE fv.value->'preprocessor_module'->'value'->>'path' = $1 AND f.workspace_id = $2",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "id",
-        "type_info": "Int8"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Text"
-      ]
-    },
-    "nullable": [
-      false
-    ]
-  },
-  "hash": "551fee7919fdeb911e3f9cc5852e158ea47e3db4895c2b2b1d3cb6b16fceeda9"
-}

backend/.sqlx/query-653685b39d93008762818d0518b953632040122a9af98332d3fd1d12244b1b80.json

@@ -16,7 +16,8 @@
                 "app",
                 "script",
                 "flow",
-                "raw_app"
+                "raw_app",
+                "asset"
               ]
             }
           }

backend/.sqlx/query-6bde827da007b470b9d0acccfc3e00ce6aac650b9138a236f34c614eed753849.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "INSERT INTO flow (\n        workspace_id, path, summary, description,\n        dependency_job, lock_error_logs, draft_only, tag,\n        dedicated_worker, visible_to_runner_only, on_behalf_of_email,\n        value, schema, edited_by, edited_at\n    ) VALUES (\n        $1, $2, $3, $4,\n        NULL, '', $5, $6,\n        $7, $8, $9,\n        $10, $11::text::json, $12, now()\n    )",
+  "query": "INSERT INTO flow (\n        workspace_id, path, summary, description,\n        dependency_job, lock_error_logs, draft_only, tag,\n        dedicated_worker, visible_to_runner_only, on_behalf_of_email,\n        ws_error_handler_muted,\n        value, schema, edited_by, edited_at\n    ) VALUES (\n        $1, $2, $3, $4,\n        NULL, '', $5, $6,\n        $7, $8, $9,\n        $10,\n        $11, $12::text::json, $13, now()\n    )",
   "describe": {
     "columns": [],
     "parameters": {
@@ -14,6 +14,7 @@
         "Bool",
         "Bool",
         "Text",
+        "Bool",
         "Jsonb",
         "Text",
         "Varchar"
@@ -21,5 +22,5 @@
     },
     "nullable": []
   },
-  "hash": "081dc94a7d0fdaade77cfb593a025d8c48d7eab3dbb30ca0b43fb1ef45d8d8bd"
+  "hash": "6bde827da007b470b9d0acccfc3e00ce6aac650b9138a236f34c614eed753849"
 }

backend/.sqlx/query-6fdab4c131f3126d5020b780ba45927dc778ac6fefdc2f91982f600b7cb9954f.json

@@ -1,15 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "DELETE FROM asset WHERE workspace_id = $1 AND usage_kind = 'script' AND usage_path = $2",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Text"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "6fdab4c131f3126d5020b780ba45927dc778ac6fefdc2f91982f600b7cb9954f"
-}

backend/.sqlx/query-752fc50d3374ee5ccdb49f10c11e50bbffeec1088364a64f0849fb9b25e11ef4.json

@@ -1,15 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "DELETE FROM asset WHERE workspace_id = $1 AND usage_kind = 'script' AND usage_path = $2 AND asset_detection_kind = 'static'",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Text"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "752fc50d3374ee5ccdb49f10c11e50bbffeec1088364a64f0849fb9b25e11ef4"
-}

backend/.sqlx/query-76033e76f15cee2aa0394d4ec2ff62130e7e48cb40d3b1534b0d791760b33ec7.json

@@ -1,34 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "SELECT\n                jsonb_build_object(\n                    'path', path,\n                    'kind', kind,\n                    'access_type', usage_access_type\n                ) as \"list!: _\"\n            FROM asset\n            WHERE workspace_id = $1 AND usage_path = $2 AND usage_kind = $3\n            ORDER BY path, kind",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "list!: _",
-        "type_info": "Jsonb"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Text",
-        {
-          "Custom": {
-            "name": "asset_usage_kind",
-            "kind": {
-              "Enum": [
-                "script",
-                "flow"
-              ]
-            }
-          }
-        }
-      ]
-    },
-    "nullable": [
-      null
-    ]
-  },
-  "hash": "76033e76f15cee2aa0394d4ec2ff62130e7e48cb40d3b1534b0d791760b33ec7"
-}

backend/.sqlx/query-77ac7257be02fb04c4b3213e2221e6f60621b4b2909d770de744ef5671e12ed9.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n        UPDATE\n            flow\n        SET\n            path = $1,\n            summary = $2,\n            description = $3,\n            dependency_job = NULL,\n            lock_error_logs = '',\n            draft_only = NULL,\n            tag = $4,\n            dedicated_worker = $5,\n            visible_to_runner_only = $6,\n            on_behalf_of_email = $7,\n            value = $8,\n            schema = $9::text::json,\n            edited_by = $10,\n            edited_at = now()\n        WHERE\n            path = $11 AND workspace_id = $12",
+  "query": "\n        UPDATE\n            flow\n        SET\n            path = $1,\n            summary = $2,\n            description = $3,\n            dependency_job = NULL,\n            lock_error_logs = '',\n            draft_only = NULL,\n            tag = $4,\n            dedicated_worker = $5,\n            visible_to_runner_only = $6,\n            on_behalf_of_email = $7,\n            ws_error_handler_muted = $8,\n            value = $9,\n            schema = $10::text::json,\n            edited_by = $11,\n            edited_at = now()\n        WHERE\n            path = $12 AND workspace_id = $13",
   "describe": {
     "columns": [],
     "parameters": {
@@ -12,6 +12,7 @@
         "Bool",
         "Bool",
         "Text",
+        "Bool",
         "Jsonb",
         "Text",
         "Varchar",
@@ -21,5 +22,5 @@
     },
     "nullable": []
   },
-  "hash": "207a0721b6f0b8b6ddd4120343eba524a2bc1e9047bdde5f568af4d993dbb74c"
+  "hash": "77ac7257be02fb04c4b3213e2221e6f60621b4b2909d770de744ef5671e12ed9"
 }

backend/.sqlx/query-7d73a51d5a83b11c1c4088678dd22ef369e9e26fafb75b351abfdef2b8a7a3e4.json

@@ -1,34 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "SELECT\n                jsonb_build_object(\n                    'path', path,\n                    'kind', kind,\n                    'access_type', usage_access_type\n                ) as \"list!: _\"\n            FROM asset\n            WHERE workspace_id = $1 AND usage_path = $2 AND usage_kind = $3\n              AND asset_detection_kind = 'static'\n            ORDER BY path, kind",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "list!: _",
-        "type_info": "Jsonb"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Text",
-        {
-          "Custom": {
-            "name": "asset_usage_kind",
-            "kind": {
-              "Enum": [
-                "script",
-                "flow"
-              ]
-            }
-          }
-        }
-      ]
-    },
-    "nullable": [
-      null
-    ]
-  },
-  "hash": "7d73a51d5a83b11c1c4088678dd22ef369e9e26fafb75b351abfdef2b8a7a3e4"
-}

backend/.sqlx/query-8ff8b53147d857db7829889ccbb59d2ca0492704a0b4a3cc8d1f4f733e0c794b.json

@@ -1,26 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "DELETE FROM asset WHERE workspace_id = $1 AND usage_path = $2 AND usage_kind = $3 AND asset_detection_kind = 'static'",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Text",
-        {
-          "Custom": {
-            "name": "asset_usage_kind",
-            "kind": {
-              "Enum": [
-                "script",
-                "flow"
-              ]
-            }
-          }
-        }
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "8ff8b53147d857db7829889ccbb59d2ca0492704a0b4a3cc8d1f4f733e0c794b"
-}

backend/.sqlx/query-97101fae34227628efbcd72457a83a683745a6e2fb120329deba5117cc523bf2.json

@@ -1,15 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "DELETE FROM asset WHERE workspace_id = $1 AND usage_kind = 'flow' AND usage_path = $2 AND asset_detection_kind = 'static'",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Text"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "97101fae34227628efbcd72457a83a683745a6e2fb120329deba5117cc523bf2"
-}

backend/.sqlx/query-a14d43c1baebcdf7e8cebcb7c752f78a2212fe55d4247c2321b37d182d2a4386.json

@@ -1,37 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "\n        DELETE FROM asset\n        WHERE id IN (\n            SELECT id FROM (\n                SELECT a.id, ROW_NUMBER() OVER (\n                    PARTITION BY a.workspace_id, a.path, a.kind\n                    ORDER BY a.created_at DESC\n                ) as rn,\n                limits.max_n\n                FROM asset a\n                INNER JOIN (\n                    SELECT * FROM UNNEST(\n                        $1::varchar[], \n                        $2::varchar[], \n                        $3::asset_kind[],\n                        $4::int[]\n                    ) AS t(workspace_id, path, kind, max_n)\n                ) limits\n                ON a.workspace_id = limits.workspace_id \n                AND a.path = limits.path \n                AND a.kind = limits.kind\n                WHERE a.asset_detection_kind = 'runtime'\n            ) ranked\n            WHERE rn > max_n\n        )",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "VarcharArray",
-        "VarcharArray",
-        {
-          "Custom": {
-            "name": "asset_kind[]",
-            "kind": {
-              "Array": {
-                "Custom": {
-                  "name": "asset_kind",
-                  "kind": {
-                    "Enum": [
-                      "s3object",
-                      "resource",
-                      "variable",
-                      "ducklake",
-                      "datatable"
-                    ]
-                  }
-                }
-              }
-            }
-          }
-        },
-        "Int4Array"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "a14d43c1baebcdf7e8cebcb7c752f78a2212fe55d4247c2321b37d182d2a4386"
-}

backend/.sqlx/query-a21337dac33ef7a1aff1c6797a3ca47fed231f542d5bd8725d6c86b8dd1a2535.json

@@ -1,22 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "SELECT\n            jsonb_strip_nulls(jsonb_build_object(\n                'path', asset.path,\n                'kind', asset.kind,\n                'usages', ARRAY_AGG(DISTINCT jsonb_build_object(\n                    'path', asset.usage_path,\n                    'kind', asset.usage_kind,\n                    'access_type', asset.usage_access_type,\n                    'detection_kinds', (\n                      SELECT ARRAY_AGG(DISTINCT a2.asset_detection_kind)\n                      FROM asset a2\n                      WHERE a2.workspace_id = asset.workspace_id\n                        AND a2.path = asset.path\n                        AND a2.kind = asset.kind\n                        AND a2.usage_path = asset.usage_path\n                        AND a2.usage_kind = asset.usage_kind\n                    )\n                )),\n                'metadata', (CASE\n                  WHEN asset.kind = 'resource' THEN\n                    jsonb_build_object('resource_type', resource.resource_type)\n                  ELSE\n                    NULL\n                  END\n                )\n            )) as \"list!: _\"\n        FROM asset\n        LEFT JOIN resource ON asset.kind = 'resource'\n          AND array_to_string((string_to_array(asset.path, '/'))[1:3], '/') = resource.path -- With specific table, asset path can be e.g u/diego/pg_db/table_name\n          AND resource.workspace_id = $1\n        WHERE asset.workspace_id = $1\n          AND (asset.kind <> 'resource' OR resource.path IS NOT NULL)\n          AND (asset.usage_kind <> 'flow' OR asset.usage_path = ANY(SELECT path FROM flow WHERE workspace_id = $1))\n          AND (asset.usage_kind <> 'script' OR asset.usage_path = ANY(SELECT path FROM script WHERE workspace_id = $1))\n          GROUP BY asset.path, asset.kind, resource.resource_type\n        ORDER BY asset.path, asset.kind",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "list!: _",
-        "type_info": "Jsonb"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Text"
-      ]
-    },
-    "nullable": [
-      null
-    ]
-  },
-  "hash": "a21337dac33ef7a1aff1c6797a3ca47fed231f542d5bd8725d6c86b8dd1a2535"
-}

backend/.sqlx/query-a479cd371fb5d1f52e7c727730cf48ab229e63b8dfe377975d48dcd223251e7c.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n        SELECT\n            workspace_id,\n            slack_team_id,\n            teams_team_id,\n            teams_team_name,\n            teams_team_guid,\n            slack_name,\n            slack_command_script,\n            teams_command_script,\n            slack_email,\n            slack_oauth_client_id,\n            slack_oauth_client_secret,\n            customer_id,\n            plan,\n            webhook,\n            deploy_to,\n            ai_config,\n            large_file_storage,\n            datatable,\n            ducklake,\n            git_sync,\n            deploy_ui,\n            default_app,\n            default_scripts,\n            mute_critical_alerts,\n            color,\n            operator_settings,\n            git_app_installations,\n            auto_invite,\n            error_handler,\n            success_handler\n        FROM\n            workspace_settings\n        WHERE\n            workspace_id = $1\n        ",
+  "query": "\n        SELECT\n            workspace_id,\n            slack_team_id,\n            teams_team_id,\n            teams_team_name,\n            teams_team_guid,\n            slack_name,\n            slack_command_script,\n            teams_command_script,\n            slack_email,\n            slack_oauth_client_id,\n            slack_oauth_client_secret,\n            customer_id,\n            plan,\n            webhook,\n            deploy_to,\n            ai_config,\n            large_file_storage,\n            datatable,\n            ducklake,\n            git_sync,\n            deploy_ui,\n            default_app,\n            default_scripts,\n            mute_critical_alerts,\n            color,\n            operator_settings,\n            git_app_installations,\n            auto_invite,\n            error_handler,\n            success_handler,\n            public_app_execution_limit_per_minute\n        FROM\n            workspace_settings\n        WHERE\n            workspace_id = $1\n        ",
   "describe": {
     "columns": [
       {
@@ -152,6 +152,11 @@
         "ordinal": 29,
         "name": "success_handler",
         "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 30,
+        "name": "public_app_execution_limit_per_minute",
+        "type_info": "Int4"
       }
     ],
     "parameters": {
@@ -189,8 +194,9 @@
       false,
       true,
       true,
+      true,
       true
     ]
   },
-  "hash": "289919809e16aee33c81951b05a4795de710421bcd3e4c06588e56092677bd05"
+  "hash": "a479cd371fb5d1f52e7c727730cf48ab229e63b8dfe377975d48dcd223251e7c"
 }

backend/.sqlx/query-a750630d79166f5b6d5be0e049c36135212666ce74dbf290d701b6402b800f13.json

@@ -0,0 +1,35 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT\n                jsonb_strip_nulls(jsonb_build_object(\n                    'path', path,\n                    'kind', kind,\n                    'access_type', usage_access_type,\n                    'columns', columns\n                )) as \"list!: _\"\n            FROM asset\n            WHERE workspace_id = $1 AND usage_path = $2 AND usage_kind = $3\n            ORDER BY path, kind",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "list!: _",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        {
+          "Custom": {
+            "name": "asset_usage_kind",
+            "kind": {
+              "Enum": [
+                "script",
+                "flow",
+                "job"
+              ]
+            }
+          }
+        }
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "a750630d79166f5b6d5be0e049c36135212666ce74dbf290d701b6402b800f13"
+}

backend/.sqlx/query-a9e29764b5b9d94269e2b8aa755c71b61774c8ff8ae218d7a8d6ed0ac0169366.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind)\n                VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT DO NOTHING",
+  "query": "INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind, columns)\n                VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT DO NOTHING",
   "describe": {
     "columns": [],
     "parameters": {
@@ -40,14 +40,16 @@
             "kind": {
               "Enum": [
                 "script",
-                "flow"
+                "flow",
+                "job"
               ]
             }
           }
-        }
+        },
+        "Jsonb"
       ]
     },
     "nullable": []
   },
-  "hash": "74a2871aba7e35527dcefb2538b8cf41c35618f7c15ea047c5082f8feb7a8464"
+  "hash": "a9e29764b5b9d94269e2b8aa755c71b61774c8ff8ae218d7a8d6ed0ac0169366"
 }

backend/.sqlx/query-aed5439aa6dad950e505f9f8f6914fa5ca21319c501b2822c9bc751ddfc9a0a4.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n            INSERT INTO kafka_trigger (\n                workspace_id,\n                path,\n                kafka_resource_path,\n                group_id,\n                topics,\n                script_path,\n                is_flow,\n                mode,\n                edited_by,\n                email,\n                edited_at,\n                error_handler_path,\n                error_handler_args,\n                retry\n            ) VALUES (\n                $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, now(), $11, $12, $13\n            )\n            ",
+  "query": "\n            INSERT INTO kafka_trigger (\n                workspace_id,\n                path,\n                kafka_resource_path,\n                group_id,\n                topics,\n                filters,\n                script_path,\n                is_flow,\n                mode,\n                edited_by,\n                email,\n                edited_at,\n                error_handler_path,\n                error_handler_args,\n                retry\n            ) VALUES (\n                $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, now(), $12, $13, $14\n            )\n            ",
   "describe": {
     "columns": [],
     "parameters": {
@@ -10,6 +10,7 @@
         "Varchar",
         "Varchar",
         "VarcharArray",
+        "JsonbArray",
         "Varchar",
         "Bool",
         {
@@ -33,5 +34,5 @@
     },
     "nullable": []
   },
-  "hash": "1de3e078c108a8a0136fccdf9187cc3500260bac39c7f1dfa05a93628569465b"
+  "hash": "aed5439aa6dad950e505f9f8f6914fa5ca21319c501b2822c9bc751ddfc9a0a4"
 }

backend/.sqlx/query-c94cd50ff1233025b170efee6489e6637676a8b0435a4612a7efbfff6ea2543d.json

@@ -1,22 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "SELECT\n            jsonb_strip_nulls(jsonb_build_object(\n                'path', asset.path,\n                'kind', asset.kind,\n                'usages', ARRAY_AGG(jsonb_build_object(\n                    'path', asset.usage_path,\n                    'kind', asset.usage_kind,\n                    'access_type', asset.usage_access_type\n                )),\n                'metadata', (CASE\n                  WHEN asset.kind = 'resource' THEN\n                    jsonb_build_object('resource_type', resource.resource_type)\n                  ELSE\n                    NULL\n                  END\n                )\n            )) as \"list!: _\"\n        FROM asset\n        LEFT JOIN resource ON asset.kind = 'resource'\n          AND array_to_string((string_to_array(asset.path, '/'))[1:3], '/') = resource.path -- With specific table, asset path can be e.g u/diego/pg_db/table_name\n          AND resource.workspace_id = $1\n        WHERE asset.workspace_id = $1\n          AND (asset.kind <> 'resource' OR resource.path IS NOT NULL)\n          AND (asset.usage_kind <> 'flow' OR asset.usage_path = ANY(SELECT path FROM flow WHERE workspace_id = $1))\n          AND (asset.usage_kind <> 'script' OR asset.usage_path = ANY(SELECT path FROM script WHERE workspace_id = $1))\n          GROUP BY asset.path, asset.kind, resource.resource_type\n        ORDER BY asset.path, asset.kind",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "list!: _",
-        "type_info": "Jsonb"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Text"
-      ]
-    },
-    "nullable": [
-      null
-    ]
-  },
-  "hash": "c94cd50ff1233025b170efee6489e6637676a8b0435a4612a7efbfff6ea2543d"
-}

backend/.sqlx/query-df290155c071e823a1d60573a3cba932d070ed730a37450cd30ba9b90205ed3b.json

@@ -1,37 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "SELECT COUNT(DISTINCT asset.job_id)::bigint as \"count!\"\n        FROM asset\n        WHERE asset.workspace_id = $1\n          AND asset.path = $2\n          AND asset.kind = $3\n          AND asset.asset_detection_kind = 'runtime'\n          AND asset.job_id IS NOT NULL",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "count!",
-        "type_info": "Int8"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Text",
-        {
-          "Custom": {
-            "name": "asset_kind",
-            "kind": {
-              "Enum": [
-                "s3object",
-                "resource",
-                "variable",
-                "ducklake",
-                "datatable"
-              ]
-            }
-          }
-        }
-      ]
-    },
-    "nullable": [
-      null
-    ]
-  },
-  "hash": "df290155c071e823a1d60573a3cba932d070ed730a37450cd30ba9b90205ed3b"
-}

backend/.sqlx/query-e2921e44c70cf6c76c55177f2b56985e84c59ecb3e1a13fcf27d5f7ae5f8d84c.json

@@ -0,0 +1,27 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n            UPDATE kafka_trigger\n            SET\n                kafka_resource_path = $1,\n                group_id = $2,\n                topics = $3,\n                filters = $4,\n                script_path = $5,\n                path = $6,\n                is_flow = $7,\n                edited_by = $8,\n                email = $9,\n                edited_at = now(),\n                server_id = NULL,\n                error = NULL,\n                error_handler_path = $12,\n                error_handler_args = $13,\n                retry = $14\n            WHERE\n                workspace_id = $10 AND path = $11\n            ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "VarcharArray",
+        "JsonbArray",
+        "Varchar",
+        "Varchar",
+        "Bool",
+        "Varchar",
+        "Varchar",
+        "Text",
+        "Text",
+        "Varchar",
+        "Jsonb",
+        "Jsonb"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "e2921e44c70cf6c76c55177f2b56985e84c59ecb3e1a13fcf27d5f7ae5f8d84c"
+}

backend/.sqlx/query-fb6656bc8fbd2e1b2c59f9ecbab55e5cdccdd7c99af0b61946694c6e03e7f636.json

@@ -1,63 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "SELECT DISTINCT\n            v2_job.id,\n            v2_job.created_at,\n            v2_job.created_by,\n            v2_job.runnable_path,\n            CASE\n              WHEN v2_job_completed.id IS NOT NULL THEN v2_job_completed.status::text\n              ELSE NULL\n            END as status\n        FROM asset\n        INNER JOIN v2_job ON asset.job_id = v2_job.id\n        LEFT JOIN v2_job_completed ON v2_job.id = v2_job_completed.id\n        WHERE asset.workspace_id = $1\n          AND asset.path = $2\n          AND asset.kind = $3\n          AND asset.asset_detection_kind = 'runtime'\n          AND asset.job_id IS NOT NULL\n        ORDER BY v2_job.created_at DESC\n        LIMIT $4 OFFSET $5",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "id",
-        "type_info": "Uuid"
-      },
-      {
-        "ordinal": 1,
-        "name": "created_at",
-        "type_info": "Timestamptz"
-      },
-      {
-        "ordinal": 2,
-        "name": "created_by",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 3,
-        "name": "runnable_path",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 4,
-        "name": "status",
-        "type_info": "Text"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Text",
-        {
-          "Custom": {
-            "name": "asset_kind",
-            "kind": {
-              "Enum": [
-                "s3object",
-                "resource",
-                "variable",
-                "ducklake",
-                "datatable"
-              ]
-            }
-          }
-        },
-        "Int8",
-        "Int8"
-      ]
-    },
-    "nullable": [
-      false,
-      false,
-      false,
-      true,
-      null
-    ]
-  },
-  "hash": "fb6656bc8fbd2e1b2c59f9ecbab55e5cdccdd7c99af0b61946694c6e03e7f636"
-}

backend/.sqlx/query-ff64f77d46fe8e1a07f26634df040ec5ced23cbb4390cd851c3bd7013dfd5758.json

@@ -1,15 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "DELETE FROM asset WHERE workspace_id = $1 AND usage_kind = 'flow' AND usage_path = $2",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Text"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "ff64f77d46fe8e1a07f26634df040ec5ced23cbb4390cd851c3bd7013dfd5758"
-}

backend/COMPILATION_OPTIMIZATION.md

@@ -1,97 +0,0 @@
-# Backend Compilation Optimization
-
-## Summary
-
-Feature-gated heavy dependencies that were compiled by default but only used behind enterprise/EE feature flags. This reduces the default build from **761 crates to 609 crates** (20% reduction).
-
-## Changes
-
-### Root `Cargo.toml`
-- Removed 12 unused direct dependencies: `kube`, `k8s-openapi`, `aws-sigv4`, `aws-sdk-config`, `opentelemetry-proto`, `systemstat`, `globset`, `libloading`, `bitflags`, `memchr`, `quote`, `pep440_rs`
-
-### `windmill-worker/Cargo.toml`
-- Made optional (only needed for EE OTEL tracing proxy): `hudsucker`, `hyper-http-proxy`, `hyper-tls`, `hyper-util`, `rcgen`, `opentelemetry-proto`, `prost`
-- Made optional (only needed for EE features): `aws-config`, `aws-credential-types`, `aws-smithy-types`
-- Created `otel_proxy` feature to group the OTEL proxy deps
-- Updated `private` feature to include `otel_proxy`
-
-### `windmill-common/Cargo.toml`
-- Made optional: `aws-config`, `aws-credential-types`, `aws-smithy-types`, `systemstat`, `globset`
-- Added AWS deps to `private`, `parquet`, `aws_auth`, `bedrock` features
-- Added `systemstat` to `private` feature
-- Added `globset` to `parquet` feature
-
-### `windmill-api/Cargo.toml`
-- Made optional: `aws-sigv4`, `aws-sdk-config`, `aws-credential-types`, `aws-smithy-types`, `windmill-parser-py-imports`, `windmill-autoscaling`
-- Added AWS deps to `parquet` and `bedrock` features
-- Added `windmill-parser-py-imports` to `python` and `agent_worker_server` features
-- Added `windmill-autoscaling` to `enterprise` feature
-
-### `windmill-autoscaling/Cargo.toml`
-- Made optional: `kube`, `k8s-openapi` (only used in EE code)
-- Added to `private` feature
-
-### `parsers/windmill-parser-py-imports/Cargo.toml`
-- Removed unused direct dependencies: `malachite`, `malachite-bigint` (still available transitively via `rustpython-parser`)
-
-## Benchmarks
-
-### Default build (no features)
-
-| Metric | Before | After |
-|---|---|---|
-| Crates compiled | 761 | 609 |
-| Notable deps eliminated | - | aws-sdk-config (9.5s), k8s-openapi (7.3s), zstd-sys (7.7s), kube-client (1.9s) |
-
-### Incremental compilation (stable-state, warm cache)
-
-| Scenario | Before | After |
-|---|---|---|
-| Touch `windmill-api/src/users.rs` | ~5.6s | ~5.4s |
-| Touch `windmill-worker/src/worker.rs` | ~6.7s | ~6.2s |
-| Touch `windmill-common/src/worker.rs` (cascade) | ~8.5s | ~8.5s |
-
-Incremental compilation improvement from feature-gating alone is modest because the bottleneck is the compilation of the windmill crates themselves (especially windmill-api at 90k LOC), not the dependencies.
-
-## Developer-Local Speed Tips
-
-These settings are **not committed** because they are developer-local preferences that depend on toolchain availability. Combined, they yield ~16% faster incremental compilation.
-
-### mold linker (~6% improvement)
-
-Install `mold` and add to `.cargo/config.toml`:
-
-```toml
-[target.x86_64-unknown-linux-gnu]
-linker = "clang"
-rustflags = ["-C", "link-arg=-fuse-ld=mold"]
-```
-
-### Reduced debug info (~10% improvement)
-
-Add to `[profile.dev]` in `Cargo.toml`:
-
-```toml
-split-debuginfo = "unpacked"
-debug = "line-tables-only"
-```
-
-### SQLX offline mode (~4% improvement)
-
-If you're not modifying SQL queries:
-
-```bash
-export SQLX_OFFLINE=true
-```
-
-### Combined effect
-
-| Scenario | Baseline | With all tips |
-|---|---|---|
-| Touch `windmill-api` file | 5.6s | **4.7s** |
-| Touch `windmill-worker` file | 6.7s | **6.0s** |
-| Touch `windmill-common` file (cascade) | 8.5s | **7.6s** |
-
-## What would help more (future work)
-
-The single biggest improvement would be **splitting `windmill-api`** (90k LOC) into smaller crates. Currently, any file change in the crate triggers re-analysis of all 90k lines. However, this requires significant refactoring due to tight coupling between the triggers subsystem, jobs, users, and the axum router initialization.

backend/Cargo.lock

@@ -219,9 +219,9 @@ dependencies = [
 
 [[package]]
 name = "anyhow"
-version = "1.0.100"
+version = "1.0.101"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a23eb6b1614318a8071c9b2521f36b424b2c83db5eb3a0fead4a6c0809af6e61"
+checksum = "5f0e0fee31ef5ed1ba1316088939cea399010ed7731dba877ed44aeb407a75ea"
 
 [[package]]
 name = "ar_archive_writer"
@@ -829,7 +829,7 @@ dependencies = [
  "aws-sdk-sts",
  "aws-smithy-async",
  "aws-smithy-http 0.62.6",
- "aws-smithy-json",
+ "aws-smithy-json 0.61.9",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
  "aws-smithy-types",
@@ -882,23 +882,23 @@ dependencies = [
 
 [[package]]
 name = "aws-runtime"
-version = "1.5.18"
+version = "1.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "959dab27ce613e6c9658eb3621064d0e2027e5f2acb65bc526a43577facea557"
+checksum = "c635c2dc792cb4a11ce1a4f392a925340d1bdf499289b5ec1ec6810954eb43f5"
 dependencies = [
  "aws-credential-types",
  "aws-sigv4",
  "aws-smithy-async",
  "aws-smithy-eventstream",
- "aws-smithy-http 0.62.6",
+ "aws-smithy-http 0.63.3",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
  "aws-smithy-types",
  "aws-types",
  "bytes",
  "fastrand",
- "http 0.2.12",
- "http-body 0.4.6",
+ "http 1.4.0",
+ "http-body 1.0.1",
  "percent-encoding",
  "pin-project-lite",
  "tracing",
@@ -907,15 +907,15 @@ dependencies = [
 
 [[package]]
 name = "aws-sdk-bedrock"
-version = "1.129.0"
+version = "1.130.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2d009c7cbb8332c805be2be32c0def27cb7bd4804d9333f5cf204abfd96b6d25"
+checksum = "883f0f0b2014a0bb59b6bfce8265a6ca6dc7e849d09314124f94a347b9d9e9f5"
 dependencies = [
  "aws-credential-types",
  "aws-runtime",
  "aws-smithy-async",
- "aws-smithy-http 0.62.6",
- "aws-smithy-json",
+ "aws-smithy-http 0.63.3",
+ "aws-smithy-json 0.62.3",
  "aws-smithy-observability",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
@@ -924,6 +924,7 @@ dependencies = [
  "bytes",
  "fastrand",
  "http 0.2.12",
+ "http 1.4.0",
  "regex-lite",
  "tracing",
 ]
@@ -940,7 +941,7 @@ dependencies = [
  "aws-smithy-async",
  "aws-smithy-eventstream",
  "aws-smithy-http 0.62.6",
- "aws-smithy-json",
+ "aws-smithy-json 0.61.9",
  "aws-smithy-observability",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
@@ -964,7 +965,7 @@ dependencies = [
  "aws-runtime",
  "aws-smithy-async",
  "aws-smithy-http 0.62.6",
- "aws-smithy-json",
+ "aws-smithy-json 0.61.9",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
  "aws-smithy-types",
@@ -979,16 +980,16 @@ dependencies = [
 
 [[package]]
 name = "aws-sdk-rds"
-version = "1.123.0"
+version = "1.124.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "43adf02ed73dcbaa18979ab849162331e4b772e276988e5750fd277687d92897"
+checksum = "24627b8374f8f242f351aa8a69adae8a416826a89819a7bfd010fc48b81f51e8"
 dependencies = [
  "aws-credential-types",
  "aws-runtime",
  "aws-sigv4",
  "aws-smithy-async",
- "aws-smithy-http 0.62.6",
- "aws-smithy-json",
+ "aws-smithy-http 0.63.3",
+ "aws-smithy-json 0.62.3",
  "aws-smithy-observability",
  "aws-smithy-query",
  "aws-smithy-runtime",
@@ -998,6 +999,7 @@ dependencies = [
  "aws-types",
  "fastrand",
  "http 0.2.12",
+ "http 1.4.0",
  "regex-lite",
  "tracing",
  "url",
@@ -1013,7 +1015,7 @@ dependencies = [
  "aws-runtime",
  "aws-smithy-async",
  "aws-smithy-http 0.62.6",
- "aws-smithy-json",
+ "aws-smithy-json 0.61.9",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
  "aws-smithy-types",
@@ -1035,7 +1037,7 @@ dependencies = [
  "aws-runtime",
  "aws-smithy-async",
  "aws-smithy-http 0.62.6",
- "aws-smithy-json",
+ "aws-smithy-json 0.61.9",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
  "aws-smithy-types",
@@ -1057,7 +1059,7 @@ dependencies = [
  "aws-runtime",
  "aws-smithy-async",
  "aws-smithy-http 0.62.6",
- "aws-smithy-json",
+ "aws-smithy-json 0.61.9",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
  "aws-smithy-types",
@@ -1079,7 +1081,7 @@ dependencies = [
  "aws-runtime",
  "aws-smithy-async",
  "aws-smithy-http 0.62.6",
- "aws-smithy-json",
+ "aws-smithy-json 0.61.9",
  "aws-smithy-query",
  "aws-smithy-runtime",
  "aws-smithy-runtime-api",
@@ -1094,13 +1096,13 @@ dependencies = [
 
 [[package]]
 name = "aws-sigv4"
-version = "1.3.7"
+version = "1.3.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "69e523e1c4e8e7e8ff219d732988e22bfeae8a1cafdbe6d9eca1546fa080be7c"
+checksum = "efa49f3c607b92daae0c078d48a4571f599f966dce3caee5f1ea55c4d9073f99"
 dependencies = [
  "aws-credential-types",
  "aws-smithy-eventstream",
- "aws-smithy-http 0.62.6",
+ "aws-smithy-http 0.63.3",
  "aws-smithy-runtime-api",
  "aws-smithy-types",
  "bytes",
@@ -1219,6 +1221,15 @@ dependencies = [
  "aws-smithy-types",
 ]
 
+[[package]]
+name = "aws-smithy-json"
+version = "0.62.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3cb96aa208d62ee94104645f7b2ecaf77bf27edf161590b6224bfbac2832f979"
+dependencies = [
+ "aws-smithy-types",
+]
+
 [[package]]
 name = "aws-smithy-observability"
 version = "0.2.4"
@@ -1621,6 +1632,26 @@ dependencies = [
  "syn 2.0.114",
 ]
 
+[[package]]
+name = "bindgen"
+version = "0.71.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5f58bf3d7db68cfbac37cfc485a8d711e87e064c3d0fe0435b92f7a407f9d6b3"
+dependencies = [
+ "bitflags 2.9.4",
+ "cexpr",
+ "clang-sys",
+ "itertools 0.13.0",
+ "log",
+ "prettyplease",
+ "proc-macro2",
+ "quote",
+ "regex",
+ "rustc-hash 2.1.1",
+ "shlex",
+ "syn 2.0.114",
+]
+
 [[package]]
 name = "bindgen"
 version = "0.72.1"
@@ -5114,6 +5145,18 @@ dependencies = [
  "zeroize",
 ]
 
+[[package]]
+name = "duct"
+version = "0.13.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e4ab5718d1224b63252cd0c6f74f6480f9ffeb117438a2e0f5cf6d9a4798929c"
+dependencies = [
+ "libc",
+ "once_cell",
+ "os_pipe",
+ "shared_child",
+]
+
 [[package]]
 name = "dunce"
 version = "1.0.5"
@@ -6093,6 +6136,17 @@ dependencies = [
  "unicode-width 0.2.2",
 ]
 
+[[package]]
+name = "getrandom"
+version = "0.1.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8fc3cb4d91f53b50155bdcfd23f6a4c39ae1969c2ae85982b135750cccaf5fce"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "wasi 0.9.0+wasi-snapshot-preview1",
+]
+
 [[package]]
 name = "getrandom"
 version = "0.2.17"
@@ -6102,7 +6156,7 @@ dependencies = [
  "cfg-if",
  "js-sys",
  "libc",
- "wasi",
+ "wasi 0.11.1+wasi-snapshot-preview1",
  "wasm-bindgen",
 ]
 
@@ -6959,7 +7013,7 @@ dependencies = [
  "tokio",
  "tokio-rustls 0.26.4",
  "tower-service",
- "webpki-roots 1.0.5",
+ "webpki-roots 1.0.6",
 ]
 
 [[package]]
@@ -7905,6 +7959,28 @@ dependencies = [
  "cc",
 ]
 
+[[package]]
+name = "libgssapi"
+version = "0.8.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9e668df13f2e97f3eed52d9301f6b1c4c1ccfccc30eab9e6628e4a8c1fc3546"
+dependencies = [
+ "bitflags 2.9.4",
+ "bytes",
+ "lazy_static",
+ "libgssapi-sys",
+]
+
+[[package]]
+name = "libgssapi-sys"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7518e6902e94f92e7c7271232684b60988b4bd813529b4ef9d97aead96956ae8"
+dependencies = [
+ "bindgen 0.71.1",
+ "pkg-config",
+]
+
 [[package]]
 name = "libloading"
 version = "0.7.4"
@@ -8320,6 +8396,12 @@ dependencies = [
  "digest 0.10.7",
 ]
 
+[[package]]
+name = "md5"
+version = "0.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7e6bcd6433cff03a4bfc3d9834d504467db1f1cf6d0ea765d37d330249ed629d"
+
 [[package]]
 name = "measure_time"
 version = "0.9.0"
@@ -8471,7 +8553,7 @@ checksum = "a4a650543ca06a924e8b371db273b2756685faae30f8487da1b56505a8f78b0c"
 dependencies = [
  "libc",
  "log",
- "wasi",
+ "wasi 0.11.1+wasi-snapshot-preview1",
  "windows-sys 0.48.0",
 ]
 
@@ -8482,7 +8564,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a69bcab0ad47271a0234d9422b131806bf3968021e5dc9328caf2d4cd58557fc"
 dependencies = [
  "libc",
- "wasi",
+ "wasi 0.11.1+wasi-snapshot-preview1",
  "windows-sys 0.61.2",
 ]
 
@@ -9830,9 +9912,9 @@ checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
 
 [[package]]
 name = "pest"
-version = "2.8.5"
+version = "2.8.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2c9eb05c21a464ea704b53158d358a31e6425db2f63a1a7312268b05fe2b75f7"
+checksum = "e0848c601009d37dfa3430c4666e147e49cdcf1b92ecd3e63657d8a5f19da662"
 dependencies = [
  "memchr",
  "ucd-trie",
@@ -9840,9 +9922,9 @@ dependencies = [
 
 [[package]]
 name = "pest_derive"
-version = "2.8.5"
+version = "2.8.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "68f9dbced329c441fa79d80472764b1a2c7e57123553b8519b36663a2fb234ed"
+checksum = "11f486f1ea21e6c10ed15d5a7c77165d0ee443402f0780849d1768e7d9d6fe77"
 dependencies = [
  "pest",
  "pest_generator",
@@ -9850,9 +9932,9 @@ dependencies = [
 
 [[package]]
 name = "pest_generator"
-version = "2.8.5"
+version = "2.8.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3bb96d5051a78f44f43c8f712d8e810adb0ebf923fc9ed2655a7f66f63ba8ee5"
+checksum = "8040c4647b13b210a963c1ed407c1ff4fdfa01c31d6d2a098218702e6664f94f"
 dependencies = [
  "pest",
  "pest_meta",
@@ -9863,9 +9945,9 @@ dependencies = [
 
 [[package]]
 name = "pest_meta"
-version = "2.8.5"
+version = "2.8.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "602113b5b5e8621770cfd490cfd90b9f84ab29bd2b0e49ad83eb6d186cef2365"
+checksum = "89815c69d36021a140146f26659a81d6c2afa33d216d736dd4be5381a7362220"
 dependencies = [
  "pest",
  "sha2 0.10.9",
@@ -10621,6 +10703,19 @@ dependencies = [
  "nibble_vec",
 ]
 
+[[package]]
+name = "rand"
+version = "0.7.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6a6b1679d49b24bbfe0c803429aa1874472f50d9b363131f0e89fc356b544d03"
+dependencies = [
+ "getrandom 0.1.16",
+ "libc",
+ "rand_chacha 0.2.2",
+ "rand_core 0.5.1",
+ "rand_hc",
+]
+
 [[package]]
 name = "rand"
 version = "0.8.5"
@@ -10643,6 +10738,16 @@ dependencies = [
  "zerocopy",
 ]
 
+[[package]]
+name = "rand_chacha"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f4c8ed856279c9737206bf725bf36935d8666ead7aa69b52be55af369d193402"
+dependencies = [
+ "ppv-lite86",
+ "rand_core 0.5.1",
+]
+
 [[package]]
 name = "rand_chacha"
 version = "0.3.1"
@@ -10663,6 +10768,15 @@ dependencies = [
  "rand_core 0.9.5",
 ]
 
+[[package]]
+name = "rand_core"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "90bde5296fc891b0cef12a6d03ddccc162ce7b2aff54160af9338f8d40df6d19"
+dependencies = [
+ "getrandom 0.1.16",
+]
+
 [[package]]
 name = "rand_core"
 version = "0.6.4"
@@ -10701,6 +10815,15 @@ dependencies = [
  "rand 0.9.0",
 ]
 
+[[package]]
+name = "rand_hc"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ca3129af7b92a17112d59ad498c6f81eaf463253766b90396d39ea7a39d6613c"
+dependencies = [
+ "rand_core 0.5.1",
+]
+
 [[package]]
 name = "range-alloc"
 version = "0.1.4"
@@ -10797,6 +10920,7 @@ dependencies = [
  "num_enum",
  "openssl-sys",
  "pkg-config",
+ "sasl2-sys",
 ]
 
 [[package]]
@@ -10986,7 +11110,7 @@ dependencies = [
  "wasm-bindgen-futures",
  "wasm-streams",
  "web-sys",
- "webpki-roots 1.0.5",
+ "webpki-roots 1.0.6",
 ]
 
 [[package]]
@@ -11036,9 +11160,9 @@ dependencies = [
 
 [[package]]
 name = "reqwest-middleware"
-version = "0.5.0"
+version = "0.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f42e2f48018a33d679ee7f477a446b697663a14e91ab0b3a0206792a22dd3aa8"
+checksum = "199dda04a536b532d0cc04d7979e39b1c763ea749bf91507017069c00b96056f"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -11051,9 +11175,9 @@ dependencies = [
 
 [[package]]
 name = "reqwest-retry"
-version = "0.9.0"
+version = "0.9.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "180e3930c1c07a59122f304ff269a634a4d5013afecbe646ccad4ff98e0803c2"
+checksum = "fe2412db2af7d2268e7a5406be0431f37d9eb67ff390f35b395716f5f06c2eaa"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -11580,7 +11704,7 @@ dependencies = [
  "rustls-webpki 0.103.9",
  "security-framework 3.5.1",
  "security-framework-sys",
- "webpki-root-certs 1.0.5",
+ "webpki-root-certs 1.0.6",
  "windows-sys 0.61.2",
 ]
 
@@ -11798,6 +11922,18 @@ dependencies = [
  "winapi-util",
 ]
 
+[[package]]
+name = "sasl2-sys"
+version = "0.1.22+2.1.28"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "05f2a7f7efd9fc98b3a9033272df10709f5ee3fa0eabbd61a527a3a1ed6bd3c6"
+dependencies = [
+ "cc",
+ "duct",
+ "libc",
+ "pkg-config",
+]
+
 [[package]]
 name = "saturating"
 version = "0.1.0"
@@ -12286,6 +12422,16 @@ dependencies = [
  "lazy_static",
 ]
 
+[[package]]
+name = "shared_child"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7e297bd52991bbe0686c086957bee142f13df85d1e79b0b21630a99d374ae9dc"
+dependencies = [
+ "libc",
+ "windows-sys 0.59.0",
+]
+
 [[package]]
 name = "shellexpand"
 version = "2.1.2"
@@ -13775,8 +13921,7 @@ dependencies = [
 [[package]]
 name = "tiberius"
 version = "0.12.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a1446cb4198848d1562301a3340424b4f425ef79f35ef9ee034769a9dd92c10d"
+source = "git+https://github.com/prisma/tiberius?rev=59db57960a14b422fb3a1309aa4aa47880896ff8#59db57960a14b422fb3a1309aa4aa47880896ff8"
 dependencies = [
  "async-trait",
  "asynchronous-codec",
@@ -13787,6 +13932,7 @@ dependencies = [
  "encoding_rs",
  "enumflags2",
  "futures-util",
+ "libgssapi",
  "num-traits",
  "once_cell",
  "pin-project-lite",
@@ -13799,6 +13945,7 @@ dependencies = [
  "tokio-util",
  "tracing",
  "uuid",
+ "winauth",
 ]
 
 [[package]]
@@ -15120,6 +15267,12 @@ dependencies = [
  "try-lock",
 ]
 
+[[package]]
+name = "wasi"
+version = "0.9.0+wasi-snapshot-preview1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cccddf32554fecc6acb585f82a32a72e28b48f8c4c1883ddfeeeaa96f7d8e519"
+
 [[package]]
 name = "wasi"
 version = "0.11.1+wasi-snapshot-preview1"
@@ -15300,14 +15453,14 @@ version = "0.26.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "75c7f0ef91146ebfb530314f5f1d24528d7f0767efbfd31dce919275413e393e"
 dependencies = [
- "webpki-root-certs 1.0.5",
+ "webpki-root-certs 1.0.6",
 ]
 
 [[package]]
 name = "webpki-root-certs"
-version = "1.0.5"
+version = "1.0.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "36a29fc0408b113f68cf32637857ab740edfafdf460c326cd2afaa2d84cc05dc"
+checksum = "804f18a4ac2676ffb4e8b5b5fa9ae38af06df08162314f96a68d2a363e21a8ca"
 dependencies = [
  "rustls-pki-types",
 ]
@@ -15318,14 +15471,14 @@ version = "0.26.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9"
 dependencies = [
- "webpki-roots 1.0.5",
+ "webpki-roots 1.0.6",
 ]
 
 [[package]]
 name = "webpki-roots"
-version = "1.0.5"
+version = "1.0.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "12bed680863276c63889429bfd6cab3b99943659923822de1c8a39c49e4d722c"
+checksum = "22cfaf3c063993ff62e73cb4311efde4db1efb31ab78a3e5c457939ad5cc0bed"
 dependencies = [
  "rustls-pki-types",
 ]
@@ -15485,13 +15638,29 @@ version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
 
+[[package]]
+name = "winauth"
+version = "0.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f820cd208ce9c6b050812dc2d724ba98c6c1e9db5ce9b3f58d925ae5723a5e6"
+dependencies = [
+ "bitflags 1.3.2",
+ "byteorder",
+ "md5",
+ "rand 0.7.3",
+ "winapi",
+]
+
 [[package]]
 name = "windmill"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
+ "aws-sdk-config",
+ "aws-sigv4",
  "axum 0.7.9",
  "base64 0.22.1",
+ "bitflags 2.9.4",
  "chrono",
  "constant_time_eq 0.3.1",
  "deno_core",
@@ -15499,10 +15668,18 @@ dependencies = [
  "futures",
  "gethostname",
  "git-version",
+ "globset",
+ "k8s-openapi",
+ "kube",
  "lazy_static",
+ "libloading 0.8.9",
+ "memchr",
  "object_store",
  "once_cell",
+ "opentelemetry-proto 0.29.0",
+ "pep440_rs",
  "prometheus",
+ "quote",
  "rand 0.9.0",
  "reqwest 0.13.1",
  "rustls 0.23.35",
@@ -15515,6 +15692,7 @@ dependencies = [
  "sql-builder",
  "sqlx",
  "strum 0.27.2",
+ "systemstat",
  "tempfile",
  "tikv-jemalloc-ctl",
  "tikv-jemalloc-sys",
@@ -15539,7 +15717,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-api"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "argon2",
@@ -15575,6 +15753,7 @@ dependencies = [
  "constant_time_eq 0.3.1",
  "cookie 0.17.0",
  "cron",
+ "dashmap 6.1.0",
  "datafusion",
  "deno_core",
  "deno_error",
@@ -15670,7 +15849,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-api-client"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "reqwest 0.12.28",
  "serde",
@@ -15680,7 +15859,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-audit"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "chrono",
  "lazy_static",
@@ -15694,7 +15873,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-autoscaling"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "axum 0.7.9",
@@ -15713,7 +15892,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-common"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "async-recursion",
@@ -15809,7 +15988,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-git-sync"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "regex",
  "serde",
@@ -15824,7 +16003,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-indexer"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "astral-tokio-tar",
@@ -15848,7 +16027,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-macros"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "itertools 0.14.0",
  "lazy_static",
@@ -15864,7 +16043,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-mcp"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -15884,7 +16063,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-oauth"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "async-oauth2",
@@ -15908,7 +16087,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "convert_case 0.6.0",
  "serde",
@@ -15917,7 +16096,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-bash"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "lazy_static",
@@ -15929,7 +16108,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-csharp"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "serde_json",
@@ -15941,7 +16120,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-go"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "gosyn",
@@ -15953,7 +16132,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-graphql"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "lazy_static",
@@ -15965,7 +16144,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-java"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "serde_json",
@@ -15977,7 +16156,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-nu"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "nu-parser",
@@ -15988,7 +16167,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-php"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "itertools 0.14.0",
@@ -15999,7 +16178,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-py"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "itertools 0.14.0",
@@ -16012,12 +16191,14 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-py-imports"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "async-recursion",
  "itertools 0.14.0",
  "lazy_static",
+ "malachite",
+ "malachite-bigint",
  "pep440_rs",
  "phf 0.11.3",
  "regex",
@@ -16034,7 +16215,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-ruby"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "lazy_static",
@@ -16048,7 +16229,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-rust"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "convert_case 0.6.0",
@@ -16065,7 +16246,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-sql"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "lazy_static",
@@ -16079,7 +16260,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-ts"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "lazy_static",
@@ -16098,7 +16279,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-parser-yaml"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "serde",
@@ -16109,7 +16290,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-queue"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "async-recursion",
@@ -16146,7 +16327,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-sql-datatype-parser-wasm"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "wasm-bindgen",
  "wasm-bindgen-test",
@@ -16156,7 +16337,7 @@ dependencies = [
 
 [[package]]
 name = "windmill-worker"
-version = "1.624.0"
+version = "1.628.2"
 dependencies = [
  "anyhow",
  "async-once-cell",
@@ -17057,18 +17238,18 @@ dependencies = [
 
 [[package]]
 name = "zerocopy"
-version = "0.8.38"
+version = "0.8.39"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "57cf3aa6855b23711ee9852dfc97dfaa51c45feaba5b645d0c777414d494a961"
+checksum = "db6d35d663eadb6c932438e763b262fe1a70987f9ae936e60158176d710cae4a"
 dependencies = [
  "zerocopy-derive",
 ]
 
 [[package]]
 name = "zerocopy-derive"
-version = "0.8.38"
+version = "0.8.39"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8a616990af1a287837c4fe6596ad77ef57948f787e46ce28e166facc0cc1cb75"
+checksum = "4122cd3169e94605190e77839c9a40d40ed048d305bfdc146e7df40ab0f3e517"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -17151,9 +17332,9 @@ dependencies = [
 
 [[package]]
 name = "zip"
-version = "7.2.0"
+version = "7.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c42e33efc22a0650c311c2ef19115ce232583abbe80850bc8b66509ebef02de0"
+checksum = "cc12baa6db2b15a140161ce53d72209dacea594230798c24774139b54ecaa980"
 dependencies = [
  "crc32fast",
  "indexmap 2.11.1",

backend/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "windmill"
-version = "1.624.0"
+version = "1.628.2"
 authors.workspace = true
 edition.workspace = true
 
@@ -35,7 +35,7 @@ members = [
 exclude = ["./windmill-duckdb-ffi-internal"]
 
 [workspace.package]
-version = "1.624.0"
+version = "1.628.2"
 authors = ["Ruben Fiszel <ruben@windmill.dev>"]
 edition = "2021"
 
@@ -71,6 +71,7 @@ deno_core = ["windmill-worker/deno_core", "windmill-api/deno_core", "dep:deno_co
 quickjs = ["windmill-worker/quickjs"]
 deno_core_mac = ["deno_core", "windmill-worker/libffi_mac"]
 kafka = ["windmill-api/kafka"]
+kafka-gssapi = ["windmill-api/kafka-gssapi"]
 nats = ["windmill-api/nats"]
 otel = ["windmill-common/otel", "windmill-worker/otel"]
 dind = ["windmill-worker/dind"]
@@ -98,22 +99,26 @@ mysql = ["windmill-worker/mysql"]
 oracledb = ["windmill-worker/oracledb"]
 duckdb = ["windmill-worker/duckdb"]
 mssql = ["windmill-worker/mssql"]
+mssql-kerberos = ["windmill-worker/mssql-kerberos"]  # Linux/Unix integrated auth
+mssql-winauth = ["windmill-worker/mssql-winauth"]  # Windows integrated auth
 bigquery = ["windmill-worker/bigquery"]
 php = ["windmill-worker/php"]
 csharp = ["windmill-worker/csharp"]
 nu = ["windmill-worker/nu"]
 java = ["windmill-worker/java"]
 ruby = ["windmill-worker/ruby"]
-all_languages = ["python", "deno_core", "rust", "mysql", "oracledb", "duckdb", "mssql", "bigquery", "csharp", "nu", "php", "java", "ruby"]
+all_languages = ["python", "deno_core", "rust", "mysql", "oracledb", "duckdb", "mssql-kerberos", "bigquery", "csharp", "nu", "php", "java", "ruby"]
 # For windows we have another set of languages enabled
-all_languages_windows = ["python", "deno_core", "rust", "mysql", "oracledb", "duckdb", "mssql", "bigquery", "csharp", "nu", "php", "java"]
+all_languages_windows = ["python", "deno_core", "rust", "mysql", "oracledb", "duckdb", "mssql-winauth", "bigquery", "csharp", "nu", "php", "java"]
 all_sqlx_features = ["all_languages", "enterprise", "enterprise_saml", "embedding", "parquet", "prometheus", "flow_testing",
- "openidconnect", "cloud", "jemalloc", "tantivy", "sqlx", "kafka", "nats", "otel", "dind", "websocket", "http_trigger",
+ "openidconnect", "cloud", "jemalloc", "tantivy", "sqlx", "kafka", "kafka-gssapi", "nats", "otel", "dind", "websocket", "http_trigger",
   "postgres_trigger", "mcp", "mqtt_trigger", "sqs_trigger", "gcp_trigger", "smtp", "stripe",
    "license", "oauth2", "zip", "static_frontend", "scoped_cache", "agent_worker_server", "bedrock", "native_trigger"]
 
 [patch.crates-io]
 object_store = { git = "https://github.com/apache/arrow-rs-object-store", rev = "36752c975d4f29e20b57c91f81a10872dcd48ae7" }
+# Use tiberius main branch for libgssapi 0.8.1 fix (https://github.com/prisma/tiberius/issues/343)
+tiberius = { git = "https://github.com/prisma/tiberius", rev = "59db57960a14b422fb3a1309aa4aa47880896ff8" }
 
 [dependencies]
 anyhow.workspace = true
@@ -150,13 +155,21 @@ deno_core = { workspace = true, optional = true }
 object_store = { workspace = true, optional = true }
 sha1 = { workspace = true, optional = true }
 constant_time_eq = { workspace = true, optional = true }
-
-
+quote.workspace = true
+memchr.workspace = true
 v8 = { workspace = true, optional = true }
 rustls.workspace = true
+pep440_rs.workspace = true
 strum.workspace = true
-
-
+aws-sigv4.workspace = true
+aws-sdk-config.workspace = true
+kube.workspace = true
+k8s-openapi.workspace = true
+libloading.workspace = true
+bitflags.workspace = true
+globset.workspace = true
+opentelemetry-proto.workspace = true
+systemstat.workspace = true
 
 [target.'cfg(windows)'.dependencies]
 windows-service = "0.7"

backend/ee-repo-ref.txt

@@ -1 +1 @@
-138a4f5f868f3bded5bb7cb77b222b532c07e4af
+1f4c304ea02a2bd19f67fc4c4202175d2103fa7a

backend/migrations/20260130150931_kafka_trigger_filters.down.sql

@@ -0,0 +1 @@
+ALTER TABLE kafka_trigger DROP COLUMN filters;

backend/migrations/20260130150931_kafka_trigger_filters.up.sql

@@ -0,0 +1 @@
+ALTER TABLE kafka_trigger ADD COLUMN filters JSONB[] NOT NULL DEFAULT '{}';

backend/migrations/20260203122047_asset_columns.down.sql

@@ -0,0 +1,2 @@
+-- Remove columns field from asset table
+ALTER TABLE asset DROP COLUMN columns;

backend/migrations/20260203122047_asset_columns.up.sql

@@ -0,0 +1,3 @@
+-- Add columns field to asset table to store column-level access information
+-- This is a JSONB map of column name to access type (r, w, or rw)
+ALTER TABLE asset ADD COLUMN columns JSONB;

backend/migrations/20260204120403_add_public_app_rate_limit.down.sql

@@ -0,0 +1,5 @@
+DROP TRIGGER IF EXISTS workspace_rate_limit_change_trigger ON workspace_settings;
+DROP FUNCTION IF EXISTS notify_workspace_rate_limit_change();
+
+ALTER TABLE workspace_settings
+DROP COLUMN IF EXISTS public_app_execution_limit_per_minute;

backend/migrations/20260204120403_add_public_app_rate_limit.up.sql

@@ -0,0 +1,19 @@
+ALTER TABLE workspace_settings
+ADD COLUMN IF NOT EXISTS public_app_execution_limit_per_minute INTEGER DEFAULT NULL;
+
+-- Add trigger function for rate limit changes
+CREATE OR REPLACE FUNCTION notify_workspace_rate_limit_change()
+RETURNS TRIGGER AS $$
+BEGIN
+    INSERT INTO notify_event (channel, payload)
+    VALUES ('notify_workspace_rate_limit_change', NEW.workspace_id);
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+-- Create trigger on workspace_settings (drop first if exists)
+DROP TRIGGER IF EXISTS workspace_rate_limit_change_trigger ON workspace_settings;
+CREATE TRIGGER workspace_rate_limit_change_trigger
+AFTER UPDATE OF public_app_execution_limit_per_minute ON workspace_settings
+FOR EACH ROW
+EXECUTE FUNCTION notify_workspace_rate_limit_change();

backend/migrations/20260204152754_add_asset_favorite_kind.down.sql

@@ -0,0 +1 @@
+-- Add down migration script here

backend/migrations/20260204152754_add_asset_favorite_kind.up.sql

@@ -0,0 +1,10 @@
+-- Add up migration script here
+
+DO
+$do$
+BEGIN
+  ALTER TYPE FAVORITE_KIND ADD VALUE 'asset';
+EXCEPTION WHEN OTHERS THEN
+  RAISE NOTICE 'Couldn''t create FAVORITE_KIND::asset: %', SQLERRM;
+END
+$do$;

backend/migrations/20260205193914_grant_asset_id_seq.down.sql

@@ -0,0 +1,2 @@
+REVOKE ALL ON SEQUENCE asset_id_seq FROM windmill_user;
+REVOKE ALL ON SEQUENCE asset_id_seq FROM windmill_admin;

backend/migrations/20260205193914_grant_asset_id_seq.up.sql

@@ -0,0 +1,2 @@
+GRANT ALL ON SEQUENCE asset_id_seq TO windmill_user;
+GRANT ALL ON SEQUENCE asset_id_seq TO windmill_admin;

backend/migrations/20260206000000_consolidate_live_index_migrations.down.sql

@@ -0,0 +1,2 @@
+-- No-op: indexes and windmill_migrations entries are safe to leave in place.
+-- Rolling back this migration does not require removing the indexes.

backend/migrations/20260206000000_consolidate_live_index_migrations.up.sql

@@ -0,0 +1,89 @@
+-- Consolidate live index migrations into a regular SQL migration.
+-- All statements are idempotent (IF EXISTS / IF NOT EXISTS).
+
+-- === DROP obsolete indexes ===
+DROP INDEX IF EXISTS ix_completed_job_workspace_id_created_at;
+DROP INDEX IF EXISTS ix_completed_job_workspace_id_created_at_new;
+DROP INDEX IF EXISTS index_completed_job_on_schedule_path;
+DROP INDEX IF EXISTS concurrency_limit_stats_queue;
+DROP INDEX IF EXISTS root_job_index;
+DROP INDEX IF EXISTS index_completed_on_created;
+DROP INDEX IF EXISTS root_job_index_by_path_2;
+DROP INDEX IF EXISTS ix_completed_job_workspace_id_created_at_new_2;
+DROP INDEX IF EXISTS ix_completed_job_workspace_id_started_at_new;
+DROP INDEX IF EXISTS root_job_index_by_path;
+DROP INDEX IF EXISTS labeled_jobs_on_jobs;
+DROP INDEX IF EXISTS ix_job_workspace_id_created_at_new_6;
+DROP INDEX IF EXISTS ix_job_workspace_id_created_at_new_7;
+DROP INDEX IF EXISTS queue_sort;
+DROP INDEX IF EXISTS queue_sort_2;
+DROP INDEX IF EXISTS log_file_hostname_log_ts_idx;
+DROP INDEX IF EXISTS ix_completed_job_workspace_id_started_at_new_2;
+DROP INDEX IF EXISTS ix_job_created_at;
+DROP INDEX IF EXISTS ix_v2_job_root_by_path;
+
+-- === CREATE indexes ===
+CREATE INDEX IF NOT EXISTS ix_job_workspace_id_created_at_new_3
+    ON v2_job (workspace_id, created_at DESC);
+
+CREATE INDEX 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;
+
+CREATE INDEX 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;
+
+CREATE INDEX 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;
+
+CREATE INDEX IF NOT EXISTS labeled_jobs_on_jobs
+    ON v2_job_completed USING GIN ((result -> 'wm_labels'))
+    WHERE result ? 'wm_labels';
+
+CREATE INDEX IF NOT EXISTS ix_v2_job_labels
+    ON v2_job USING GIN (labels)
+    WHERE labels IS NOT NULL;
+
+ALTER TABLE v2_job ENABLE ROW LEVEL SECURITY;
+
+CREATE INDEX IF NOT EXISTS ix_v2_job_workspace_id_created_at
+    ON v2_job (workspace_id, created_at DESC)
+    WHERE kind IN ('script', 'flow', 'singlestepflow') AND parent_job IS NULL;
+
+CREATE INDEX IF NOT EXISTS queue_sort_v2
+    ON v2_job_queue (priority DESC NULLS LAST, scheduled_for, tag)
+    WHERE running = false;
+
+CREATE INDEX IF NOT EXISTS ix_audit_timestamps
+    ON audit (timestamp DESC);
+
+CREATE INDEX IF NOT EXISTS ix_job_completed_completed_at
+    ON v2_job_completed (completed_at DESC);
+
+CREATE INDEX IF NOT EXISTS alerts_by_workspace
+    ON alerts (workspace_id);
+
+CREATE INDEX IF NOT EXISTS v2_job_queue_suspend
+    ON v2_job_queue (workspace_id, suspend)
+    WHERE suspend > 0;
+
+CREATE INDEX IF NOT EXISTS idx_audit_recent_login_activities
+    ON audit (timestamp, username)
+    WHERE operation IN ('users.login', 'oauth.login', 'users.token.refresh');
+
+CREATE INDEX IF NOT EXISTS script_not_archived
+    ON script (workspace_id, path, created_at DESC)
+    WHERE archived = false;
+
+CREATE INDEX IF NOT EXISTS ix_job_workspace_id_completed_at_all
+    ON v2_job_completed (workspace_id, completed_at DESC);
+
+CREATE INDEX IF NOT EXISTS idx_job_v2_job_root_by_path_2
+    ON v2_job (workspace_id, runnable_path)
+    WHERE parent_job IS NULL;
+
+CREATE INDEX IF NOT EXISTS ix_job_root_job_index_by_path_2
+    ON v2_job (workspace_id, runnable_path, created_at DESC)
+    WHERE parent_job IS NULL;

backend/migrations/20260206060555_notify_triggers_security_definer.down.sql

@@ -0,0 +1,12 @@
+ALTER FUNCTION notify_config_change() SECURITY INVOKER;
+ALTER FUNCTION notify_global_setting_change() SECURITY INVOKER;
+ALTER FUNCTION notify_global_setting_delete() SECURITY INVOKER;
+ALTER FUNCTION notify_webhook_change() SECURITY INVOKER;
+ALTER FUNCTION notify_workspace_envs_change() SECURITY INVOKER;
+ALTER FUNCTION notify_workspace_premium_change() SECURITY INVOKER;
+ALTER FUNCTION notify_team_plan_status_change() SECURITY INVOKER;
+ALTER FUNCTION notify_runnable_version_change() SECURITY INVOKER;
+ALTER FUNCTION notify_http_trigger_change() SECURITY INVOKER;
+ALTER FUNCTION notify_token_invalidation() SECURITY INVOKER;
+ALTER FUNCTION notify_workspace_key_change() SECURITY INVOKER;
+ALTER FUNCTION notify_workspace_rate_limit_change() SECURITY INVOKER;

backend/migrations/20260206060555_notify_triggers_security_definer.up.sql

@@ -0,0 +1,18 @@
+-- Make all notify_event trigger functions SECURITY DEFINER so that
+-- INSERT INTO notify_event runs as the function owner (typically the
+-- superuser that created the function) rather than the invoking role.
+-- This prevents "permission denied for table notify_event" errors when
+-- windmill_user or windmill_admin fire these triggers.
+
+ALTER FUNCTION notify_config_change() SECURITY DEFINER;
+ALTER FUNCTION notify_global_setting_change() SECURITY DEFINER;
+ALTER FUNCTION notify_global_setting_delete() SECURITY DEFINER;
+ALTER FUNCTION notify_webhook_change() SECURITY DEFINER;
+ALTER FUNCTION notify_workspace_envs_change() SECURITY DEFINER;
+ALTER FUNCTION notify_workspace_premium_change() SECURITY DEFINER;
+ALTER FUNCTION notify_team_plan_status_change() SECURITY DEFINER;
+ALTER FUNCTION notify_runnable_version_change() SECURITY DEFINER;
+ALTER FUNCTION notify_http_trigger_change() SECURITY DEFINER;
+ALTER FUNCTION notify_token_invalidation() SECURITY DEFINER;
+ALTER FUNCTION notify_workspace_key_change() SECURITY DEFINER;
+ALTER FUNCTION notify_workspace_rate_limit_change() SECURITY DEFINER;

backend/migrations/20260207000001_concurrent_indexes_v2_job.down.sql

@@ -0,0 +1 @@
+-- no-op: indexes are safe to leave in place

backend/migrations/20260207000001_concurrent_indexes_v2_job.up.sql

@@ -0,0 +1,41 @@
+-- v2_job: drop obsolete indexes and create new ones CONCURRENTLY
+DROP INDEX CONCURRENTLY IF EXISTS root_job_index;
+DROP INDEX CONCURRENTLY IF EXISTS root_job_index_by_path_2;
+DROP INDEX CONCURRENTLY IF EXISTS root_job_index_by_path;
+DROP INDEX CONCURRENTLY IF EXISTS ix_job_workspace_id_created_at_new_6;
+DROP INDEX CONCURRENTLY IF EXISTS ix_job_workspace_id_created_at_new_7;
+DROP INDEX CONCURRENTLY IF EXISTS ix_job_created_at;
+DROP INDEX CONCURRENTLY IF EXISTS ix_v2_job_root_by_path;
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_job_workspace_id_created_at_new_3
+    ON v2_job (workspace_id, created_at DESC);
+
+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;
+
+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;
+
+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;
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_v2_job_labels
+    ON v2_job USING GIN (labels)
+    WHERE labels IS NOT NULL;
+
+ALTER TABLE v2_job ENABLE ROW LEVEL SECURITY;
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_v2_job_workspace_id_created_at
+    ON v2_job (workspace_id, created_at DESC)
+    WHERE kind IN ('script', 'flow', 'singlestepflow') AND parent_job IS NULL;
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_job_v2_job_root_by_path_2
+    ON v2_job (workspace_id, runnable_path)
+    WHERE parent_job IS NULL;
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_job_root_job_index_by_path_2
+    ON v2_job (workspace_id, runnable_path, created_at DESC)
+    WHERE parent_job IS NULL;

backend/migrations/20260207000002_concurrent_indexes_v2_job_completed.down.sql

@@ -0,0 +1 @@
+-- no-op: indexes are safe to leave in place

backend/migrations/20260207000002_concurrent_indexes_v2_job_completed.up.sql

@@ -0,0 +1,19 @@
+-- v2_job_completed: drop obsolete indexes and create new ones CONCURRENTLY
+DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at;
+DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at_new;
+DROP INDEX CONCURRENTLY IF EXISTS index_completed_job_on_schedule_path;
+DROP INDEX CONCURRENTLY IF EXISTS index_completed_on_created;
+DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at_new_2;
+DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_started_at_new;
+DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_started_at_new_2;
+DROP INDEX CONCURRENTLY IF EXISTS labeled_jobs_on_jobs;
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS labeled_jobs_on_jobs
+    ON v2_job_completed USING GIN ((result -> 'wm_labels'))
+    WHERE result ? 'wm_labels';
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_job_completed_completed_at
+    ON v2_job_completed (completed_at DESC);
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_job_workspace_id_completed_at_all
+    ON v2_job_completed (workspace_id, completed_at DESC);

backend/migrations/20260207000003_concurrent_indexes_v2_job_queue.down.sql

@@ -0,0 +1 @@
+-- no-op: indexes are safe to leave in place

backend/migrations/20260207000003_concurrent_indexes_v2_job_queue.up.sql

@@ -0,0 +1,12 @@
+-- v2_job_queue: drop obsolete indexes and create new ones CONCURRENTLY
+DROP INDEX CONCURRENTLY IF EXISTS concurrency_limit_stats_queue;
+DROP INDEX CONCURRENTLY IF EXISTS queue_sort;
+DROP INDEX CONCURRENTLY IF EXISTS queue_sort_2;
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS queue_sort_v2
+    ON v2_job_queue (priority DESC NULLS LAST, scheduled_for, tag)
+    WHERE running = false;
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS v2_job_queue_suspend
+    ON v2_job_queue (workspace_id, suspend)
+    WHERE suspend > 0;

backend/migrations/20260207000004_concurrent_indexes_other.down.sql

@@ -0,0 +1 @@
+-- no-op: indexes are safe to leave in place

backend/migrations/20260207000004_concurrent_indexes_other.up.sql

@@ -0,0 +1,16 @@
+-- audit, alerts, script, log_file: drop obsolete indexes and create new ones CONCURRENTLY
+DROP INDEX CONCURRENTLY IF EXISTS log_file_hostname_log_ts_idx;
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_audit_timestamps
+    ON audit (timestamp DESC);
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS alerts_by_workspace
+    ON alerts (workspace_id);
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_audit_recent_login_activities
+    ON audit (timestamp, username)
+    WHERE operation IN ('users.login', 'oauth.login', 'users.token.refresh');
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS script_not_archived
+    ON script (workspace_id, path, created_at DESC)
+    WHERE archived = false;

backend/parsers/windmill-parser-py-imports/Cargo.toml

@@ -18,6 +18,8 @@ regex.workspace = true
 windmill-parser.workspace = true
 windmill-common.workspace = true
 rustpython-parser.workspace = true
+malachite.workspace = true
+malachite-bigint.workspace = true
 phf.workspace = true
 itertools.workspace = true
 serde_json.workspace = true

backend/parsers/windmill-parser-py/src/asset_parser.rs

@@ -19,9 +19,12 @@ pub fn parse_assets(input: &str) -> anyhow::Result<ParseAssetsOutput> {
         // if a db = wmill.datatable() was never used (e.g db.query(...)),
         // we still want to register the asset as unknown access type
         if asset_was_used(&assets_finder.assets, (kind, &path)) == false {
-            assets_finder
-                .assets
-                .push(ParseAssetsResult { kind, access_type: None, path });
+            assets_finder.assets.push(ParseAssetsResult {
+                kind,
+                path,
+                access_type: None,
+                columns: None,
+            });
         }
     }
 
@@ -48,8 +51,12 @@ impl Visitor for AssetsFinder {
             match removed {
                 Some((kind, path, _)) => {
                     if !asset_was_used(&self.assets, (kind, &path)) {
-                        self.assets
-                            .push(ParseAssetsResult { kind, access_type: None, path });
+                        self.assets.push(ParseAssetsResult {
+                            kind,
+                            path,
+                            access_type: None,
+                            columns: None,
+                        });
                     }
                 }
                 None => {}
@@ -76,6 +83,7 @@ impl Visitor for AssetsFinder {
                         kind,
                         path: path.to_string(),
                         access_type: None,
+                        columns: None,
                     });
                 }
             }
@@ -97,6 +105,7 @@ impl Visitor for AssetsFinder {
                                 kind,
                                 path: path.to_string(),
                                 access_type: None,
+                                columns: None,
                             });
                         }
                     }
@@ -252,8 +261,12 @@ impl AssetsFinder {
                 let path = parse_asset_syntax(&value, false)
                     .map(|(_, p)| p)
                     .unwrap_or(&value);
-                self.assets
-                    .push(ParseAssetsResult { kind, path: path.to_string(), access_type });
+                self.assets.push(ParseAssetsResult {
+                    kind,
+                    path: path.to_string(),
+                    access_type,
+                    columns: None,
+                });
             }
             _ => return Err(()),
         };
@@ -266,6 +279,8 @@ struct Arg(usize, &'static str);
 
 #[cfg(test)]
 mod tests {
+    use std::collections::BTreeMap;
+
     use super::*;
 
     #[test]
@@ -281,7 +296,8 @@ def main():
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::S3Object,
                 path: "/test.csv".to_string(),
-                access_type: Some(R)
+                access_type: Some(R),
+                columns: None,
             },])
         );
     }
@@ -299,7 +315,8 @@ def main():
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::DataTable,
                 path: "main".to_string(),
-                access_type: None
+                access_type: None,
+                columns: None,
             },])
         );
     }
@@ -318,7 +335,8 @@ def main(x: int):
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::DataTable,
                 path: "dt/friends".to_string(),
-                access_type: Some(R)
+                access_type: Some(R),
+                columns: None,
             },])
         );
     }
@@ -340,12 +358,14 @@ def main(x: int):
                 ParseAssetsResult {
                     kind: AssetKind::DataTable,
                     path: "dt/analytics".to_string(),
-                    access_type: Some(R)
+                    access_type: Some(R),
+                    columns: None,
                 },
                 ParseAssetsResult {
                     kind: AssetKind::DataTable,
                     path: "dt/friends".to_string(),
-                    access_type: Some(RW)
+                    access_type: Some(RW),
+                    columns: Some(BTreeMap::from([("x".to_string(), AssetUsageAccessType::W)])),
                 },
             ])
         );
@@ -372,17 +392,20 @@ def g():
                 ParseAssetsResult {
                     kind: AssetKind::DataTable,
                     path: "another1/customers".to_string(),
-                    access_type: Some(W)
+                    access_type: Some(W),
+                    columns: None,
                 },
                 ParseAssetsResult {
                     kind: AssetKind::Ducklake,
                     path: "another2".to_string(),
-                    access_type: None
+                    access_type: None,
+                    columns: None,
                 },
                 ParseAssetsResult {
                     kind: AssetKind::DataTable,
                     path: "main/friends".to_string(),
-                    access_type: Some(R)
+                    access_type: Some(R),
+                    columns: None,
                 },
             ])
         );
@@ -404,12 +427,14 @@ def g():
                 ParseAssetsResult {
                     kind: AssetKind::DataTable,
                     path: "another1".to_string(),
-                    access_type: None
+                    access_type: None,
+                    columns: None,
                 },
                 ParseAssetsResult {
                     kind: AssetKind::Ducklake,
                     path: "main".to_string(),
-                    access_type: None
+                    access_type: None,
+                    columns: None,
                 },
             ])
         );
@@ -429,7 +454,8 @@ def main(x: int):
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::DataTable,
                 path: "dt/public.friends".to_string(),
-                access_type: Some(R)
+                access_type: Some(R),
+                columns: None,
             },])
         );
     }
@@ -448,7 +474,8 @@ def main():
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::Ducklake,
                 path: "lake1/analytics.metrics".to_string(),
-                access_type: Some(R)
+                access_type: Some(R),
+                columns: None,
             },])
         );
     }
@@ -468,7 +495,8 @@ def main(x: int):
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::DataTable,
                 path: "dt/public.users".to_string(),
-                access_type: Some(RW)
+                access_type: Some(RW),
+                columns: None,
             },])
         );
     }
@@ -486,7 +514,8 @@ def main():
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::DataTable,
                 path: "dt".to_string(),
-                access_type: None
+                access_type: None,
+                columns: None,
             },])
         );
     }

backend/parsers/windmill-parser-sql/src/asset_parser.rs

@@ -1,9 +1,9 @@
-use std::collections::HashMap;
+use std::collections::BTreeMap;
 
 use sqlparser::{
     ast::{
-        CopyTarget, Expr, ObjectName, TableFactor, TableObject, Value, ValueWithSpan, Visit,
-        Visitor,
+        CopyTarget, Expr, ObjectName, ObjectNamePart, SelectItem, TableFactor, TableObject, Value,
+        ValueWithSpan, Visit, Visitor,
     },
     dialect::DuckDbDialect,
     parser::Parser,
@@ -24,9 +24,12 @@ pub fn parse_assets(input: &str) -> anyhow::Result<ParseAssetsOutput> {
 
     for (_, (kind, path)) in collector.var_identifiers {
         if !asset_was_used(&collector.assets, (kind, &path)) {
-            collector
-                .assets
-                .push(ParseAssetsResult { kind, access_type: None, path: path });
+            collector.assets.push(ParseAssetsResult {
+                kind,
+                access_type: None,
+                path: path,
+                columns: None,
+            });
         }
     }
 
@@ -39,7 +42,7 @@ struct AssetCollector {
     // e.g set to Read when we are inside a SELECT ... FROM ... statement
     current_access_type_stack: Vec<AssetUsageAccessType>,
     // e.g ATTACH 'ducklake://a' AS dl; => { "dl": (Ducklake, "a") }
-    var_identifiers: HashMap<String, (AssetKind, String)>,
+    var_identifiers: BTreeMap<String, (AssetKind, String)>,
     // e.g USE dl;
     currently_used_asset: Option<(AssetKind, String)>,
 }
@@ -49,15 +52,19 @@ impl AssetCollector {
         Self {
             assets: Vec::new(),
             current_access_type_stack: Vec::with_capacity(8),
-            var_identifiers: HashMap::new(),
+            var_identifiers: BTreeMap::new(),
             currently_used_asset: None,
         }
     }
 
     // Detect when we do 'a.b' and 'a' is associated with an asset in var_identifiers
     // Or when we access 'b' and we did USE a;
-    fn get_associated_asset_from_obj_name(&self, name: &ObjectName) -> Option<ParseAssetsResult> {
-        let access_type = self.current_access_type_stack.last().copied();
+    fn get_associated_asset_from_obj_name(
+        &self,
+        name: &ObjectName,
+        access_type: Option<AssetUsageAccessType>,
+    ) -> Option<ParseAssetsResult> {
+        let access_type = access_type.or_else(|| self.current_access_type_stack.last().copied());
         if let Some((kind, path)) = &self.currently_used_asset {
             // We don't want to infer that any simple identifier refers to an asset if
             // we are not in a known R/W context
@@ -80,8 +87,15 @@ impl AssetCollector {
                     .map(|id| id.as_ident().map(|id| id.value.clone()))
                     .collect::<Option<Vec<String>>>()?
                     .join(".");
-                let path = format!("{}/{}", path, specific_table);
-                return Some(ParseAssetsResult { kind: *kind, access_type, path });
+
+                // For Resource assets, use ?table= query parameter syntax
+                // For Ducklake and DataTable, maintain /table syntax
+                let path = if *kind == AssetKind::Resource {
+                    format!("{}?table={}", path, specific_table)
+                } else {
+                    format!("{}/{}", path, specific_table)
+                };
+                return Some(ParseAssetsResult { kind: *kind, access_type, path, columns: None });
             }
         }
 
@@ -97,11 +111,18 @@ impl AssetCollector {
                 .map(|id| id.as_ident().map(|id| id.value.clone()))
                 .collect::<Option<Vec<String>>>()?
                 .join(".");
-            format!("{}/{}", path, specific_table)
+
+            // For Resource assets, use ?table= query parameter syntax
+            // For Ducklake and DataTable, maintain /table syntax
+            if *kind == AssetKind::Resource {
+                format!("{}?table={}", path, specific_table)
+            } else {
+                format!("{}/{}", path, specific_table)
+            }
         } else {
             path.clone()
         };
-        Some(ParseAssetsResult { kind: *kind, access_type, path })
+        Some(ParseAssetsResult { kind: *kind, access_type, path, columns: None })
     }
 
     fn handle_string_literal(&mut self, s: &str) {
@@ -112,6 +133,7 @@ impl AssetCollector {
                     kind,
                     path: path.to_string(),
                     access_type: self.current_access_type_stack.last().copied(),
+                    columns: None,
                 });
             }
         }
@@ -126,13 +148,6 @@ impl AssetCollector {
         if let Some(str_lit) = get_str_lit_from_obj_name(name) {
             self.handle_string_literal(str_lit);
         }
-
-        // Writes to tables should be handled directly when visiting the statement
-        if self.current_access_type_stack.last() == Some(&R) {
-            if let Some(asset) = self.get_associated_asset_from_obj_name(name) {
-                self.assets.push(asset);
-            }
-        }
     }
 
     fn handle_obj_name_post(&mut self, name: &ObjectName) {
@@ -146,20 +161,144 @@ impl AssetCollector {
         }
     }
 
-    fn handle_table_with_joins(&mut self, table_with_joins: &sqlparser::ast::TableWithJoins) {
-        if let TableFactor::Table { name, .. } = &table_with_joins.relation {
-            if let Some(asset) = self.get_associated_asset_from_obj_name(name) {
+    fn handle_table_with_joins(
+        &mut self,
+        table_with_joins: &sqlparser::ast::TableWithJoins,
+        access_type: Option<AssetUsageAccessType>,
+    ) {
+        if let TableFactor::Table { name, args, .. } = &table_with_joins.relation {
+            if args.is_some() && args.as_ref().map_or(0, |a| a.args.len()) > 0 {
+                return;
+            }
+            if let Some(asset) = self.get_associated_asset_from_obj_name(name, access_type) {
                 self.assets.push(asset);
             }
         }
         for join in &table_with_joins.joins {
             if let TableFactor::Table { name, .. } = &join.relation {
-                if let Some(asset) = self.get_associated_asset_from_obj_name(name) {
+                if let Some(asset) = self.get_associated_asset_from_obj_name(name, access_type) {
                     self.assets.push(asset);
                 }
             }
         }
     }
+
+    // Extract columns from SELECT items and create individual asset results for each column
+    // Only processes columns that reference known assets to avoid false positives
+    fn extract_column_assets(
+        &mut self,
+        projection: &[SelectItem],
+        from_tables: &[sqlparser::ast::TableWithJoins],
+    ) {
+        // Check if this is a single-table SELECT (to avoid ambiguity)
+        let single_table = if from_tables.len() == 1 {
+            if let TableFactor::Table { name, args, .. } = &from_tables[0].relation {
+                if args.is_some() && args.as_ref().map_or(0, |a| a.args.len()) > 0 {
+                    return; // Skip table functions
+                }
+                self.get_associated_asset_from_obj_name(name, Some(R))
+            } else {
+                None
+            }
+        } else {
+            None
+        };
+
+        // Build a map of table aliases/names to assets for multi-table queries
+        let mut table_to_asset: BTreeMap<String, ParseAssetsResult> = BTreeMap::new();
+        for table_with_joins in from_tables {
+            if let TableFactor::Table { name, alias, args, .. } = &table_with_joins.relation {
+                if args.is_some() && args.as_ref().map_or(0, |a| a.args.len()) > 0 {
+                    continue; // Skip table functions
+                }
+                if let Some(asset) = self.get_associated_asset_from_obj_name(name, Some(R)) {
+                    // Use alias if present, otherwise use the table name
+                    let table_key = if let Some(alias) = alias {
+                        alias.name.value.clone()
+                    } else {
+                        // For qualified names like "dl.table1", use just the last part
+                        name.0
+                            .last()
+                            .and_then(|id| id.as_ident())
+                            .map(|id| id.value.clone())
+                            .unwrap_or_default()
+                    };
+                    table_to_asset.insert(table_key, asset);
+                }
+            }
+        }
+
+        // Process each SELECT item
+        for item in projection {
+            match item {
+                SelectItem::UnnamedExpr(Expr::Identifier(ident))
+                | SelectItem::ExprWithAlias { expr: Expr::Identifier(ident), .. } => {
+                    // Simple column: SELECT a
+                    // Only add if we have a single table (unambiguous)
+                    if let Some(asset) = &single_table {
+                        let mut columns = BTreeMap::new();
+                        columns.insert(ident.value.clone(), R);
+                        self.assets.push(ParseAssetsResult {
+                            kind: asset.kind,
+                            path: asset.path.clone(),
+                            access_type: Some(R),
+                            columns: Some(columns),
+                        });
+                    }
+                }
+                SelectItem::UnnamedExpr(Expr::CompoundIdentifier(parts))
+                | SelectItem::ExprWithAlias { expr: Expr::CompoundIdentifier(parts), .. } => {
+                    // Qualified column: SELECT table1.a or SELECT x.table1.a
+                    if parts.len() >= 2 {
+                        let column_name = parts.last().map(|id| id.value.clone());
+
+                        if let Some(column_name) = column_name {
+                            // Check if the prefix matches a known table
+                            let table_prefix = parts.first().map(|id| id.value.clone());
+
+                            if let Some(table_prefix) = table_prefix {
+                                if let Some(asset) = table_to_asset.get(&table_prefix) {
+                                    // Found a matching table, add column asset
+                                    let mut columns = BTreeMap::new();
+                                    columns.insert(column_name.clone(), R);
+                                    self.assets.push(ParseAssetsResult {
+                                        kind: asset.kind,
+                                        path: asset.path.clone(),
+                                        access_type: Some(R),
+                                        columns: Some(columns),
+                                    });
+                                } else if parts.len() >= 3 {
+                                    // Could be x.table1.column format or db.schema.table.column
+                                    // Convert Idents to ObjectNameParts
+                                    let obj_parts: Vec<ObjectNamePart> = parts[..parts.len() - 1]
+                                        .iter()
+                                        .cloned()
+                                        .map(|ident| ObjectNamePart::Identifier(ident))
+                                        .collect();
+                                    let obj_name = ObjectName(obj_parts);
+                                    if let Some(asset) =
+                                        self.get_associated_asset_from_obj_name(&obj_name, Some(R))
+                                    {
+                                        let mut columns = BTreeMap::new();
+                                        columns.insert(column_name.clone(), R);
+                                        self.assets.push(ParseAssetsResult {
+                                            kind: asset.kind,
+                                            path: asset.path.clone(),
+                                            access_type: Some(R),
+                                            columns: Some(columns),
+                                        });
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+                _ => {
+                    // Ignore wildcards, expressions, etc.
+                }
+            }
+        }
+    }
 }
 
 impl Visitor for AssetCollector {
@@ -218,51 +357,154 @@ impl Visitor for AssetCollector {
         statement: &sqlparser::ast::Statement,
     ) -> std::ops::ControlFlow<Self::Break> {
         match statement {
-            sqlparser::ast::Statement::Query(_) => {
-                // don't forget pop() in post_visit_statement
-                self.current_access_type_stack.push(R);
+            sqlparser::ast::Statement::Query(q) => {
+                if let Some(select) = q.body.as_select() {
+                    // First, handle table references (adds table-level assets)
+                    for t in &select.from {
+                        self.handle_table_with_joins(t, Some(R));
+                    }
+                    // Then, extract column-level assets
+                    self.extract_column_assets(&select.projection, &select.from);
+                }
             }
 
             sqlparser::ast::Statement::Insert(insert) => {
                 let access_type = if insert.returning.is_some() { RW } else { W };
-                self.current_access_type_stack.push(access_type);
                 match insert.table {
                     TableObject::TableName(ref name) => {
-                        if let Some(asset) = self.get_associated_asset_from_obj_name(name) {
-                            self.assets.push(asset);
+                        if let Some(asset) =
+                            self.get_associated_asset_from_obj_name(name, Some(access_type))
+                        {
+                            // Add table-level asset
+                            self.assets.push(ParseAssetsResult {
+                                kind: asset.kind,
+                                path: asset.path.clone(),
+                                access_type: asset.access_type,
+                                columns: None,
+                            });
+
+                            // Extract column information for INSERT with explicit columns (Write access)
+                            if !insert.columns.is_empty() {
+                                for col in &insert.columns {
+                                    let columns = BTreeMap::from([(col.value.clone(), W)]);
+                                    self.assets.push(ParseAssetsResult {
+                                        kind: asset.kind,
+                                        path: asset.path.clone(),
+                                        access_type: Some(W),
+                                        columns: Some(columns),
+                                    });
+                                }
+                            }
+
+                            // Extract column information from RETURNING clause (Read access)
+                            if let Some(returning) = &insert.returning {
+                                for item in returning {
+                                    match item {
+                                        SelectItem::UnnamedExpr(Expr::Identifier(ident))
+                                        | SelectItem::ExprWithAlias {
+                                            expr: Expr::Identifier(ident),
+                                            ..
+                                        } => {
+                                            let mut col_map = BTreeMap::new();
+                                            col_map.insert(ident.value.clone(), R);
+                                            self.assets.push(ParseAssetsResult {
+                                                kind: asset.kind,
+                                                path: asset.path.clone(),
+                                                access_type: Some(R),
+                                                columns: Some(col_map),
+                                            });
+                                        }
+                                        _ => {
+                                            // Ignore wildcards and complex expressions
+                                        }
+                                    }
+                                }
+                            }
                         }
                     }
                     _ => {}
                 }
-                self.current_access_type_stack.pop();
             }
 
-            sqlparser::ast::Statement::Update { returning, table, from, .. } => {
+            sqlparser::ast::Statement::Update { returning, table, from, assignments, .. } => {
                 if let Some(from_tables) = from {
                     let from_tables = match from_tables {
                         sqlparser::ast::UpdateTableFromKind::AfterSet(tables) => tables,
                         sqlparser::ast::UpdateTableFromKind::BeforeSet(tables) => tables,
                     };
-                    self.current_access_type_stack.push(R);
                     for table_with_joins in from_tables {
-                        self.handle_table_with_joins(table_with_joins);
+                        self.handle_table_with_joins(table_with_joins, Some(R));
                     }
-                    self.current_access_type_stack.pop();
                 }
 
                 let access_type = if returning.is_some() { RW } else { W };
-                self.current_access_type_stack.push(access_type);
+                self.handle_table_with_joins(table, Some(access_type));
 
-                self.handle_table_with_joins(table);
+                // Extract column information from UPDATE SET clauses (Write access)
+                // Only process if it's a single table update
+                if let TableFactor::Table { name, .. } = &table.relation {
+                    if let Some(asset) =
+                        self.get_associated_asset_from_obj_name(name, Some(access_type))
+                    {
+                        // Process each assignment to extract column names
+                        for assignment in assignments {
+                            // assignment.target is an AssignmentTarget enum
+                            // We only handle simple column names (ColumnName variant)
+                            if let sqlparser::ast::AssignmentTarget::ColumnName(col_name) =
+                                &assignment.target
+                            {
+                                // For simple column updates, this is typically a single ident
+                                if col_name.0.len() == 1 {
+                                    if let Some(col_ident) =
+                                        col_name.0.first().and_then(|p| p.as_ident())
+                                    {
+                                        let mut col_map = BTreeMap::new();
+                                        col_map.insert(col_ident.value.clone(), W);
+                                        self.assets.push(ParseAssetsResult {
+                                            kind: asset.kind,
+                                            path: asset.path.clone(),
+                                            access_type: Some(W),
+                                            columns: Some(col_map),
+                                        });
+                                    }
+                                }
+                            }
+                        }
 
-                self.current_access_type_stack.pop();
+                        // Extract column information from RETURNING clause (Read access)
+                        if let Some(returning_items) = returning {
+                            for item in returning_items {
+                                match item {
+                                    SelectItem::UnnamedExpr(Expr::Identifier(ident))
+                                    | SelectItem::ExprWithAlias {
+                                        expr: Expr::Identifier(ident),
+                                        ..
+                                    } => {
+                                        let mut col_map = BTreeMap::new();
+                                        col_map.insert(ident.value.clone(), R);
+                                        self.assets.push(ParseAssetsResult {
+                                            kind: asset.kind,
+                                            path: asset.path.clone(),
+                                            access_type: Some(R),
+                                            columns: Some(col_map),
+                                        });
+                                    }
+                                    _ => {
+                                        // Ignore wildcards and complex expressions
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
             }
 
             sqlparser::ast::Statement::Delete(delete) => {
                 let access_type = if delete.returning.is_some() { RW } else { W };
-                self.current_access_type_stack.push(access_type);
                 for name in &delete.tables {
-                    if let Some(asset) = self.get_associated_asset_from_obj_name(name) {
+                    if let Some(asset) =
+                        self.get_associated_asset_from_obj_name(name, Some(access_type))
+                    {
                         self.assets.push(asset);
                     }
                 }
@@ -271,25 +513,22 @@ impl Visitor for AssetCollector {
                     sqlparser::ast::FromTable::WithoutKeyword(tables) => tables,
                 };
                 for table_with_joins in tables {
-                    self.handle_table_with_joins(table_with_joins);
+                    self.handle_table_with_joins(table_with_joins, Some(access_type));
                 }
-                self.current_access_type_stack.pop();
             }
 
             sqlparser::ast::Statement::CreateTable(create_table) => {
-                self.current_access_type_stack.push(W);
-                if let Some(asset) = self.get_associated_asset_from_obj_name(&create_table.name) {
+                if let Some(asset) =
+                    self.get_associated_asset_from_obj_name(&create_table.name, Some(W))
+                {
                     self.assets.push(asset);
                 }
-                self.current_access_type_stack.pop();
             }
 
             sqlparser::ast::Statement::CreateView { name, .. } => {
-                self.current_access_type_stack.push(W);
-                if let Some(asset) = self.get_associated_asset_from_obj_name(name) {
+                if let Some(asset) = self.get_associated_asset_from_obj_name(name, Some(W)) {
                     self.assets.push(asset);
                 }
-                self.current_access_type_stack.pop();
             }
 
             sqlparser::ast::Statement::Copy { target: CopyTarget::File { filename }, .. } => {
@@ -339,14 +578,8 @@ impl Visitor for AssetCollector {
 
     fn post_visit_statement(
         &mut self,
-        statement: &sqlparser::ast::Statement,
+        _statement: &sqlparser::ast::Statement,
     ) -> std::ops::ControlFlow<Self::Break> {
-        match statement {
-            sqlparser::ast::Statement::Query(_) => {
-                self.current_access_type_stack.pop();
-            }
-            _ => {}
-        }
         std::ops::ControlFlow::Continue(())
     }
 
@@ -409,17 +642,20 @@ mod tests {
                 ParseAssetsResult {
                     kind: AssetKind::S3Object,
                     path: "/a.parquet".to_string(),
-                    access_type: Some(R)
+                    access_type: Some(R),
+                    columns: None
                 },
                 ParseAssetsResult {
                     kind: AssetKind::S3Object,
                     path: "/c.parquet".to_string(),
-                    access_type: Some(W)
+                    access_type: Some(W),
+                    columns: None
                 },
                 ParseAssetsResult {
                     kind: AssetKind::S3Object,
                     path: "snd/b.parquet".to_string(),
-                    access_type: Some(R)
+                    access_type: Some(R),
+                    columns: None
                 },
             ])
         );
@@ -438,7 +674,8 @@ mod tests {
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::Ducklake,
                 path: "my_dl".to_string(),
-                access_type: None
+                access_type: None,
+                columns: None
             },])
         );
     }
@@ -455,7 +692,8 @@ mod tests {
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::Ducklake,
                 path: "my_dl/table1".to_string(),
-                access_type: Some(R)
+                access_type: Some(R),
+                columns: None
             },])
         );
     }
@@ -473,7 +711,8 @@ mod tests {
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::DataTable,
                 path: "my_dt/table1".to_string(),
-                access_type: Some(W)
+                access_type: Some(W),
+                columns: None
             },])
         );
     }
@@ -504,7 +743,8 @@ mod tests {
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::Ducklake,
                 path: "my_dl/table1".to_string(),
-                access_type: Some(W)
+                access_type: Some(W),
+                columns: None
             },])
         );
     }
@@ -521,7 +761,8 @@ mod tests {
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::DataTable,
                 path: "main/table1".to_string(),
-                access_type: Some(W)
+                access_type: Some(W),
+                columns: None
             },])
         );
     }
@@ -543,7 +784,8 @@ mod tests {
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::Ducklake,
                 path: "main/friends".to_string(),
-                access_type: Some(RW)
+                access_type: Some(RW),
+                columns: None
             },])
         );
     }
@@ -561,7 +803,8 @@ mod tests {
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::Ducklake,
                 path: "main".to_string(),
-                access_type: None
+                access_type: None,
+                columns: None
             },])
         );
     }
@@ -579,7 +822,8 @@ mod tests {
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::Ducklake,
                 path: "main/table1".to_string(),
-                access_type: Some(W)
+                access_type: Some(W),
+                columns: None
             },])
         );
     }
@@ -597,7 +841,8 @@ mod tests {
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::Ducklake,
                 path: "main/table1".to_string(),
-                access_type: Some(W)
+                access_type: Some(W),
+                columns: Some(BTreeMap::from([("id".to_string(), W)])),
             },])
         );
     }
@@ -614,8 +859,9 @@ mod tests {
             s.map_err(|e| e.to_string()),
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::Resource,
-                path: "u/user/pg_resource/table1".to_string(),
-                access_type: Some(R)
+                path: "u/user/pg_resource?table=table1".to_string(),
+                access_type: Some(R),
+                columns: None
             },])
         );
     }
@@ -632,7 +878,77 @@ mod tests {
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::Ducklake,
                 path: "main/table1".to_string(),
-                access_type: Some(W)
+                access_type: Some(W),
+                columns: Some(BTreeMap::from([("id".to_string(), W)])),
+            },])
+        );
+    }
+
+    #[test]
+    fn test_sql_asset_parser_resource_vs_ducklake_syntax() {
+        // Test that Resource uses ?table= while Ducklake uses /table
+        let input_resource = r#"
+            ATTACH 'res://u/user/pg_resource' AS db (TYPE postgres);
+            SELECT * FROM db.users;
+        "#;
+        let s = parse_assets(input_resource).map(|s| s.assets);
+        assert_eq!(
+            s.map_err(|e| e.to_string()),
+            Ok(vec![ParseAssetsResult {
+                kind: AssetKind::Resource,
+                path: "u/user/pg_resource?table=users".to_string(),
+                access_type: Some(R),
+                columns: None
+            },])
+        );
+
+        let input_ducklake = r#"
+            ATTACH 'ducklake://my_lake' AS dl;
+            SELECT * FROM dl.users;
+        "#;
+        let s = parse_assets(input_ducklake).map(|s| s.assets);
+        assert_eq!(
+            s.map_err(|e| e.to_string()),
+            Ok(vec![ParseAssetsResult {
+                kind: AssetKind::Ducklake,
+                path: "my_lake/users".to_string(),
+                access_type: Some(R),
+                columns: None
+            },])
+        );
+
+        let input_datatable = r#"
+            ATTACH 'datatable://dt1' AS dt;
+            SELECT * FROM dt.users;
+        "#;
+        let s = parse_assets(input_datatable).map(|s| s.assets);
+        assert_eq!(
+            s.map_err(|e| e.to_string()),
+            Ok(vec![ParseAssetsResult {
+                kind: AssetKind::DataTable,
+                path: "dt1/users".to_string(),
+                access_type: Some(R),
+                columns: None
+            },])
+        );
+    }
+
+    #[test]
+    fn test_sql_asset_parser_resource_with_long_path() {
+        // Test that Resource works with paths longer than 3 components
+        let input = r#"
+            ATTACH 'res://u/diego/a/b/c/my_postgres_resource' AS db (TYPE postgres);
+            USE db;
+            SELECT * FROM my_table;
+        "#;
+        let s = parse_assets(input).map(|s| s.assets);
+        assert_eq!(
+            s.map_err(|e| e.to_string()),
+            Ok(vec![ParseAssetsResult {
+                kind: AssetKind::Resource,
+                path: "u/diego/a/b/c/my_postgres_resource?table=my_table".to_string(),
+                access_type: Some(R),
+                columns: None
             },])
         );
     }
@@ -650,7 +966,8 @@ mod tests {
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::Ducklake,
                 path: "main/sch.table1".to_string(),
-                access_type: Some(RW)
+                access_type: Some(RW),
+                columns: Some(BTreeMap::from([("id".to_string(), W)])),
             },])
         );
     }
@@ -669,8 +986,289 @@ mod tests {
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::Ducklake,
                 path: "main/sch.table1".to_string(),
-                access_type: Some(RW)
+                access_type: Some(RW),
+                columns: Some(BTreeMap::from([("id".to_string(), W)])),
             },])
         );
     }
+
+    #[test]
+    fn test_sql_asset_parser_single_table_column_detection() {
+        let input = r#"
+            ATTACH 'ducklake://my_dl' AS dl;
+            SELECT a, b FROM dl.table1;
+        "#;
+        let s = parse_assets(input).map(|s| s.assets);
+
+        let result = s.unwrap();
+
+        // Should have one asset with merged columns
+        assert_eq!(result.len(), 1);
+        assert_eq!(result[0].path, "my_dl/table1");
+        assert_eq!(result[0].access_type, Some(R));
+
+        // Check that both columns are present in the merged asset
+        let columns = result[0].columns.as_ref().expect("Should have columns");
+        assert_eq!(columns.len(), 2);
+        assert_eq!(columns.get("a"), Some(&R));
+        assert_eq!(columns.get("b"), Some(&R));
+    }
+
+    #[test]
+    fn test_sql_asset_parser_explicit_table_prefix_columns() {
+        let input = r#"
+            ATTACH 'ducklake://my_dl' AS dl;
+            SELECT dl.table1.a, dl.table1.b FROM dl.table1;
+        "#;
+        let s = parse_assets(input).map(|s| s.assets);
+
+        // Should detect columns with explicit table prefix
+        let result = s.unwrap();
+        // Check we have the table asset
+
+        // Check we have column assets
+        assert!(result.iter().any(|a| {
+            a.path == "my_dl/table1"
+                && a.columns
+                    .as_ref()
+                    .map_or(false, |cols| cols.contains_key("a"))
+        }));
+        assert!(result.iter().any(|a| {
+            a.path == "my_dl/table1"
+                && a.columns
+                    .as_ref()
+                    .map_or(false, |cols| cols.contains_key("b"))
+        }));
+    }
+
+    #[test]
+    fn test_sql_asset_parser_multi_table_no_simple_columns() {
+        let input = r#"
+            ATTACH 'ducklake://my_dl' AS dl;
+            SELECT a, b FROM dl.table1, dl.table2;
+        "#;
+        let s = parse_assets(input).map(|s| s.assets);
+
+        // Simple columns (a, b) should NOT be detected with multiple tables
+        // Only table-level assets should be present
+        let result = s.unwrap();
+
+        // Should have 2 table assets
+        assert_eq!(result.iter().filter(|a| a.columns.is_none()).count(), 2);
+
+        // Should have NO column assets (ambiguous which table they belong to)
+        assert_eq!(result.iter().filter(|a| a.columns.is_some()).count(), 0);
+    }
+
+    #[test]
+    fn test_sql_asset_parser_multi_table_with_qualified_columns() {
+        let input = r#"
+            ATTACH 'ducklake://my_dl1' AS dl1;
+            ATTACH 'ducklake://my_dl2' AS dl2;
+            SELECT table1.a, table2.b FROM dl1.table1, dl2.table2;
+        "#;
+        let s = parse_assets(input).map(|s| s.assets);
+
+        // Qualified columns should be detected even with multiple tables
+        let result = s.unwrap();
+
+        // Check we have column assets for both tables
+        assert!(result.iter().any(|a| {
+            a.path == "my_dl1/table1"
+                && a.columns
+                    .as_ref()
+                    .map_or(false, |cols| cols.contains_key("a"))
+        }));
+        assert!(result.iter().any(|a| {
+            a.path == "my_dl2/table2"
+                && a.columns
+                    .as_ref()
+                    .map_or(false, |cols| cols.contains_key("b"))
+        }));
+    }
+
+    #[test]
+    fn test_sql_asset_parser_use_with_simple_columns() {
+        let input = r#"
+            ATTACH 'ducklake://my_dl' AS dl;
+            USE dl;
+            SELECT a, b, c FROM table1;
+        "#;
+        let s = parse_assets(input).map(|s| s.assets);
+
+        let result = s.unwrap();
+
+        // Should detect columns since it's a single table
+        assert!(result.iter().any(|a| {
+            a.path == "my_dl/table1"
+                && a.columns
+                    .as_ref()
+                    .map_or(false, |cols| cols.contains_key("a"))
+        }));
+        assert!(result.iter().any(|a| {
+            a.path == "my_dl/table1"
+                && a.columns
+                    .as_ref()
+                    .map_or(false, |cols| cols.contains_key("b"))
+        }));
+        assert!(result.iter().any(|a| {
+            a.path == "my_dl/table1"
+                && a.columns
+                    .as_ref()
+                    .map_or(false, |cols| cols.contains_key("c"))
+        }));
+    }
+
+    #[test]
+    fn test_sql_asset_parser_wildcard_no_columns() {
+        let input = r#"
+            ATTACH 'ducklake://my_dl' AS dl;
+            SELECT * FROM dl.table1;
+        "#;
+        let s = parse_assets(input).map(|s| s.assets);
+
+        let result = s.unwrap();
+
+        // Wildcard should NOT create column assets, only table asset
+        assert_eq!(result.len(), 1);
+        assert!(result[0].columns.is_none());
+    }
+
+    #[test]
+    fn test_sql_asset_parser_columns_with_alias() {
+        let input = r#"
+            ATTACH 'ducklake://my_dl' AS dl;
+            SELECT a AS column_a, b AS column_b FROM dl.table1;
+        "#;
+        let s = parse_assets(input).map(|s| s.assets);
+
+        let result = s.unwrap();
+
+        // Should detect columns even when aliased
+        assert!(result.iter().any(|a| {
+            a.path == "my_dl/table1"
+                && a.columns
+                    .as_ref()
+                    .map_or(false, |cols| cols.contains_key("a"))
+        }));
+        assert!(result.iter().any(|a| {
+            a.path == "my_dl/table1"
+                && a.columns
+                    .as_ref()
+                    .map_or(false, |cols| cols.contains_key("b"))
+        }));
+    }
+
+    #[test]
+    fn test_sql_asset_parser_columns_with_table_alias() {
+        let input = r#"
+            ATTACH 'ducklake://my_dl' AS dl;
+            SELECT t.a, t.b FROM dl.table1 AS t;
+        "#;
+        let s = parse_assets(input).map(|s| s.assets);
+
+        let result = s.unwrap();
+
+        // Should detect columns using the table alias
+        assert!(result.iter().any(|a| {
+            a.path == "my_dl/table1"
+                && a.columns
+                    .as_ref()
+                    .map_or(false, |cols| cols.contains_key("a"))
+        }));
+        assert!(result.iter().any(|a| {
+            a.path == "my_dl/table1"
+                && a.columns
+                    .as_ref()
+                    .map_or(false, |cols| cols.contains_key("b"))
+        }));
+    }
+
+    #[test]
+    fn test_sql_asset_parser_insert_with_columns() {
+        let input = r#"
+            ATTACH 'ducklake://my_dl' AS dl;
+            INSERT INTO dl.table1 (name, age, email) VALUES ('John', 30, 'john@example.com');
+        "#;
+        let s = parse_assets(input).map(|s| s.assets);
+
+        let result = s.unwrap();
+
+        // Should have one asset with merged columns
+        assert_eq!(result.len(), 1);
+        assert_eq!(result[0].path, "my_dl/table1");
+        assert_eq!(result[0].access_type, Some(W));
+
+        // Check that all columns are present in the merged asset
+        let columns = result[0].columns.as_ref().expect("Should have columns");
+        assert_eq!(columns.len(), 3);
+        assert_eq!(columns.get("name"), Some(&W));
+        assert_eq!(columns.get("age"), Some(&W));
+        assert_eq!(columns.get("email"), Some(&W));
+    }
+
+    #[test]
+    fn test_sql_asset_parser_insert_without_columns() {
+        let input = r#"
+            ATTACH 'ducklake://my_dl' AS dl;
+            INSERT INTO dl.table1 VALUES ('John', 30);
+        "#;
+        let s = parse_assets(input).map(|s| s.assets);
+
+        let result = s.unwrap();
+
+        // Should have one asset without column information
+        assert_eq!(result.len(), 1);
+        assert_eq!(result[0].path, "my_dl/table1");
+        assert_eq!(result[0].access_type, Some(W));
+        assert!(result[0].columns.is_none());
+    }
+
+    #[test]
+    fn test_sql_asset_parser_update_multiple_columns() {
+        let input = r#"
+            ATTACH 'ducklake://my_dl' AS dl;
+            UPDATE dl.table1 SET name = 'Jane', age = 25, active = true;
+        "#;
+        let s = parse_assets(input).map(|s| s.assets);
+
+        let result = s.unwrap();
+
+        // Should have one asset with merged columns
+        assert_eq!(result.len(), 1);
+        assert_eq!(result[0].path, "my_dl/table1");
+        assert_eq!(result[0].access_type, Some(W));
+
+        // Check that all columns are present
+        let columns = result[0].columns.as_ref().expect("Should have columns");
+        assert_eq!(columns.len(), 3);
+        assert_eq!(columns.get("name"), Some(&W));
+        assert_eq!(columns.get("age"), Some(&W));
+        assert_eq!(columns.get("active"), Some(&W));
+    }
+
+    #[test]
+    fn test_sql_asset_parser_update_returning() {
+        let input = r#"
+            ATTACH 'ducklake://my_dl' AS dl;
+            UPDATE dl.table1 SET name = 'Jane', age = 26 RETURNING id, name;
+        "#;
+        let s = parse_assets(input).map(|s| s.assets);
+
+        let result = s.unwrap();
+
+        // Should have RW access type when RETURNING is used
+        assert_eq!(result.len(), 1);
+        assert_eq!(result[0].path, "my_dl/table1");
+        assert_eq!(result[0].access_type, Some(RW));
+
+        // Check that columns are present with correct access types
+        // name and age are written (W), id and name are read (R)
+        // name should be RW (both written and read)
+        let columns = result[0].columns.as_ref().expect("Should have columns");
+        assert_eq!(columns.len(), 3);
+        assert_eq!(columns.get("name"), Some(&RW)); // Written in SET, read in RETURNING
+        assert_eq!(columns.get("age"), Some(&W)); // Only written
+        assert_eq!(columns.get("id"), Some(&R)); // Only read
+    }
 }

backend/parsers/windmill-parser-ts/src/asset_parser.rs

@@ -117,6 +117,7 @@ impl Visit for AssetsFinder {
                         kind,
                         path: path.to_string(),
                         access_type: None,
+                        columns: None,
                     });
                 }
             }
@@ -177,8 +178,12 @@ impl Visit for AssetsFinder {
             if asset_was_used(&self.assets, (kind, path)) {
                 continue;
             }
-            self.assets
-                .push(ParseAssetsResult { kind, access_type: None, path: path.clone() });
+            self.assets.push(ParseAssetsResult {
+                kind,
+                access_type: None,
+                path: path.clone(),
+                columns: None,
+            });
         }
 
         // Restore state - identifiers declared in this block go out of scope
@@ -294,8 +299,12 @@ impl AssetsFinder {
                 let path = parse_asset_syntax(&value, false)
                     .map(|(_, p)| p)
                     .unwrap_or(&value);
-                self.assets
-                    .push(ParseAssetsResult { kind, path: path.to_string(), access_type });
+                self.assets.push(ParseAssetsResult {
+                    kind,
+                    path: path.to_string(),
+                    access_type,
+                    columns: None,
+                });
             }
             _ => return Err(()),
         }
@@ -305,6 +314,8 @@ impl AssetsFinder {
 
 #[cfg(test)]
 mod tests {
+    use std::collections::BTreeMap;
+
     use super::*;
 
     #[test]
@@ -321,7 +332,8 @@ mod tests {
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::S3Object,
                 path: "/test.csv".to_string(),
-                access_type: Some(R)
+                access_type: Some(R),
+                columns: None,
             },])
         );
     }
@@ -340,7 +352,8 @@ mod tests {
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::DataTable,
                 path: "dt".to_string(),
-                access_type: None
+                access_type: None,
+                columns: None,
             },])
         );
     }
@@ -360,7 +373,8 @@ mod tests {
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::DataTable,
                 path: "dt/friends".to_string(),
-                access_type: Some(R)
+                access_type: Some(R),
+                columns: None,
             },])
         );
     }
@@ -383,12 +397,17 @@ mod tests {
                 ParseAssetsResult {
                     kind: AssetKind::DataTable,
                     path: "dt/analytics".to_string(),
-                    access_type: Some(R)
+                    access_type: Some(R),
+                    columns: None,
                 },
                 ParseAssetsResult {
                     kind: AssetKind::DataTable,
                     path: "dt/friends".to_string(),
-                    access_type: Some(RW)
+                    access_type: Some(RW),
+                    columns: Some(BTreeMap::from([(
+                        "name".to_string(),
+                        AssetUsageAccessType::W
+                    )])),
                 },
             ])
         );
@@ -418,17 +437,20 @@ mod tests {
                 ParseAssetsResult {
                     kind: AssetKind::DataTable,
                     path: "another1/customers".to_string(),
-                    access_type: Some(W)
+                    access_type: Some(W),
+                    columns: None,
                 },
                 ParseAssetsResult {
                     kind: AssetKind::Ducklake,
                     path: "another2".to_string(),
-                    access_type: None
+                    access_type: None,
+                    columns: None,
                 },
                 ParseAssetsResult {
                     kind: AssetKind::DataTable,
                     path: "main/friends".to_string(),
-                    access_type: Some(R)
+                    access_type: Some(R),
+                    columns: None,
                 },
             ])
         );
@@ -452,12 +474,14 @@ mod tests {
                 ParseAssetsResult {
                     kind: AssetKind::DataTable,
                     path: "another1".to_string(),
-                    access_type: None
+                    access_type: None,
+                    columns: None,
                 },
                 ParseAssetsResult {
                     kind: AssetKind::Ducklake,
                     path: "main".to_string(),
-                    access_type: None
+                    access_type: None,
+                    columns: None,
                 },
             ])
         );
@@ -478,7 +502,8 @@ mod tests {
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::DataTable,
                 path: "main/myschema.friends".to_string(),
-                access_type: Some(R)
+                access_type: Some(R),
+                columns: None,
             },])
         );
     }
@@ -499,7 +524,8 @@ mod tests {
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::DataTable,
                 path: "dt/public.users".to_string(),
-                access_type: Some(RW)
+                access_type: Some(RW),
+                columns: None,
             },])
         );
     }
@@ -518,7 +544,8 @@ mod tests {
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::DataTable,
                 path: "dt".to_string(),
-                access_type: None
+                access_type: None,
+                columns: None,
             },])
         );
     }
@@ -539,7 +566,8 @@ mod tests {
             Ok(vec![ParseAssetsResult {
                 kind: AssetKind::DataTable,
                 path: "dt/users".to_string(),
-                access_type: Some(R)
+                access_type: Some(R),
+                columns: None,
             },])
         );
     }
@@ -562,12 +590,14 @@ mod tests {
                 ParseAssetsResult {
                     kind: AssetKind::DataTable,
                     path: "dt/private.users".to_string(),
-                    access_type: Some(R)
+                    access_type: Some(R),
+                    columns: None,
                 },
                 ParseAssetsResult {
                     kind: AssetKind::DataTable,
                     path: "dt/test".to_string(),
-                    access_type: Some(W)
+                    access_type: Some(W),
+                    columns: None,
                 },
             ])
         );

backend/parsers/windmill-parser-yaml/src/asset_parser.rs

@@ -12,6 +12,7 @@ pub fn parse_assets(input: &str) -> anyhow::Result<ParseAssetsOutput> {
                 kind: AssetKind::Resource,
                 path: delegate_to_git_repo_details.resource,
                 access_type: Some(AssetUsageAccessType::R),
+                columns: None,
             })
         }
 
@@ -21,6 +22,7 @@ pub fn parse_assets(input: &str) -> anyhow::Result<ParseAssetsOutput> {
                     kind: AssetKind::Resource,
                     path: pinned_res,
                     access_type: Some(AssetUsageAccessType::R),
+                    columns: None,
                 })
             }
         }
@@ -31,6 +33,7 @@ pub fn parse_assets(input: &str) -> anyhow::Result<ParseAssetsOutput> {
                     kind: AssetKind::Resource,
                     path: resource,
                     access_type: Some(AssetUsageAccessType::R),
+                    columns: None,
                 })
             }
         }

backend/parsers/windmill-parser/src/asset_parser.rs

@@ -1,4 +1,5 @@
 use serde::Serialize;
+use std::collections::BTreeMap;
 
 #[derive(Serialize, PartialEq, Clone, Copy, Debug)]
 #[serde(rename_all(serialize = "lowercase"))]
@@ -19,12 +20,14 @@ pub enum AssetKind {
     DataTable,
 }
 
-#[derive(Serialize, Debug, PartialEq)]
+#[derive(Serialize, Debug, PartialEq, Clone)]
 pub struct ParseAssetsResult {
     pub kind: AssetKind,
     pub path: String,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub access_type: Option<AssetUsageAccessType>, // None in case of ambiguity
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub columns: Option<BTreeMap<String, AssetUsageAccessType>>, // Map column name to access type, "*" represents wildcard
 }
 
 #[derive(Serialize, Debug, PartialEq)]
@@ -66,6 +69,8 @@ pub fn merge_assets(assets: Vec<ParseAssetsResult>) -> Vec<ParseAssetsResult> {
                 (Some(R), Some(R)) => Some(R),
                 (Some(W), Some(W)) => Some(W),
             };
+            // merge columns: union the column sets and merge access types per column
+            existing.columns = merge_column_maps(existing.columns.take(), asset.columns);
         } else {
             arr.push(asset);
         }
@@ -74,18 +79,49 @@ pub fn merge_assets(assets: Vec<ParseAssetsResult>) -> Vec<ParseAssetsResult> {
     arr
 }
 
+fn merge_column_maps(
+    existing: Option<BTreeMap<String, AssetUsageAccessType>>,
+    new: Option<BTreeMap<String, AssetUsageAccessType>>,
+) -> Option<BTreeMap<String, AssetUsageAccessType>> {
+    match (existing, new) {
+        (None, None) => None,
+        (Some(map), None) | (None, Some(map)) => Some(map),
+        (Some(mut existing_map), Some(new_map)) => {
+            for (col_name, new_access) in new_map {
+                existing_map
+                    .entry(col_name)
+                    .and_modify(|existing_access| {
+                        *existing_access = merge_access_types(*existing_access, new_access);
+                    })
+                    .or_insert(new_access);
+            }
+            Some(existing_map)
+        }
+    }
+}
+
+fn merge_access_types(a: AssetUsageAccessType, b: AssetUsageAccessType) -> AssetUsageAccessType {
+    match (a, b) {
+        (R, W) | (W, R) => RW,
+        (RW, _) | (_, RW) => RW,
+        (R, R) => R,
+        (W, W) => W,
+    }
+}
+
 // Will return false if the user assigned an asset to a variable like:
 //   let sql = wmill.datatable('main')
 // But never used it. In that case we don't know which table is being used,
 // but we still want to add the main datatable as an asset with unknown access type.
 //
-// This function takes care of the fact that assets can be suffixed (e.g. "main/users")
+// This function takes care of the fact that assets can be suffixed (e.g. "main/users" or "u/user/resource?table=table1")
 pub fn asset_was_used(assets: &Vec<ParseAssetsResult>, (kind, path): (AssetKind, &String)) -> bool {
     assets.iter().any(|a| {
         let a_path = a.path.as_str();
+        // Check for /table suffix (Ducklake, DataTable) or ?table= suffix (Resource)
         let has_same_path_base = a_path
             .strip_prefix(path)
-            .map(|p| p.starts_with('/'))
+            .map(|p| p.starts_with('/') || p.starts_with('?'))
             .unwrap_or(false);
         (has_same_path_base || a_path == path) && a.kind == kind
     })

backend/src/main.rs

@@ -1145,16 +1145,22 @@ Windmill Community Edition {GIT_VERSION}
                     let db = db.clone();
                     let h = tokio::spawn(async move {
                         // Initialize last_event_id to current max to avoid processing old events on startup
-                        let mut last_event_id: i64 = match windmill_common::notify_events::get_latest_event_id(&db).await {
-                            Ok(id) => {
-                                tracing::info!("Initialized notify event polling with last_event_id: {}", id);
-                                id
-                            }
-                            Err(e) => {
-                                tracing::warn!("Could not get latest event id, starting from 0: {e:#}");
-                                0
-                            }
-                        };
+                        let mut last_event_id: i64 =
+                            match windmill_common::notify_events::get_latest_event_id(&db).await {
+                                Ok(id) => {
+                                    tracing::info!(
+                                        "Initialized notify event polling with last_event_id: {}",
+                                        id
+                                    );
+                                    id
+                                }
+                                Err(e) => {
+                                    tracing::warn!(
+                                        "Could not get latest event id, starting from 0: {e:#}"
+                                    );
+                                    0
+                                }
+                            };
                         let mut last_settings_reload = Instant::now();
                         let mut monitor_iteration: u64 = 0;
                         let rd_shift: u8 = rand::rng().random_range(0..200);
@@ -1382,36 +1388,57 @@ async fn process_notify_event(
     tx: &KillpillSender,
     server_mode: bool,
     worker_mode: bool,
-    #[cfg(feature = "parquet")]
-    disable_s3_store: bool,
+    #[cfg(feature = "parquet")] disable_s3_store: bool,
 ) {
     match channel {
         "notify_config_change" => {
             if payload == "server" && server_mode {
-                tracing::error!("Server config change detected but server config is obsolete: {}", payload);
+                tracing::error!(
+                    "Server config change detected but server config is obsolete: {}",
+                    payload
+                );
             } else if worker_mode && payload == format!("worker__{}", *WORKER_GROUP) {
                 tracing::info!("Worker config change detected: {}", payload);
                 reload_worker_config(db, tx.clone(), true).await;
             } else {
                 tracing::debug!("config changed but did not target this server/worker");
             }
-        },
+        }
         "notify_webhook_change" => {
-            tracing::info!("Webhook change detected, invalidating webhook cache: {}", payload);
+            tracing::info!(
+                "Webhook change detected, invalidating webhook cache: {}",
+                payload
+            );
             windmill_api::webhook_util::WEBHOOK_CACHE.remove(payload);
-        },
+        }
         "notify_workspace_envs_change" => {
-            tracing::info!("Workspace envs change detected, invalidating workspace envs cache: {}", payload);
+            tracing::info!(
+                "Workspace envs change detected, invalidating workspace envs cache: {}",
+                payload
+            );
             windmill_common::variables::CUSTOM_ENVS_CACHE.remove(payload);
-        },
+        }
         "notify_workspace_key_change" => {
-            tracing::info!("Workspace key change detected, invalidating workspace key cache: {}", payload);
+            tracing::info!(
+                "Workspace key change detected, invalidating workspace key cache: {}",
+                payload
+            );
             windmill_common::variables::WORKSPACE_CRYPT_CACHE.remove(payload);
-        },
+        }
         "notify_workspace_premium_change" => {
-            tracing::info!("Workspace premium change detected, invalidating workspace premium cache: {}", payload);
+            tracing::info!(
+                "Workspace premium change detected, invalidating workspace premium cache: {}",
+                payload
+            );
             windmill_common::workspaces::TEAM_PLAN_CACHE.remove(payload);
-        },
+        }
+        "notify_workspace_rate_limit_change" => {
+            tracing::info!(
+                "Workspace rate limit change detected, invalidating rate limit cache: {}",
+                payload
+            );
+            windmill_common::workspaces::PUBLIC_APP_RATE_LIMIT_CACHE.remove(payload);
+        }
         "notify_runnable_version_change" => {
             tracing::info!("Runnable version change detected: {}", payload);
             match payload.split(':').collect::<Vec<&str>>().as_slice() {
@@ -1446,39 +1473,48 @@ async fn process_notify_event(
                             }
                         }
                         "flow" => {
-                            let dynamic_input_key = windmill_common::jobs::generate_dynamic_input_key(workspace_id, path);
+                            let dynamic_input_key =
+                                windmill_common::jobs::generate_dynamic_input_key(
+                                    workspace_id,
+                                    path,
+                                );
                             windmill_common::DYNAMIC_INPUT_CACHE.remove(&dynamic_input_key);
                             windmill_common::FLOW_VERSION_CACHE.remove(&key);
-                        },
+                        }
                         _ => {
                             tracing::warn!("Unknown runnable version change payload: {}", payload);
                         }
                     }
-                },
+                }
                 _ => {
                     tracing::warn!("Unknown runnable version change payload: {}", payload);
                 }
             }
-        },
+        }
         #[cfg(feature = "http_trigger")]
         "notify_http_trigger_change" => {
             tracing::info!("HTTP trigger change detected: {}", payload);
             match windmill_api::triggers::http::refresh_routers(db).await {
                 Ok((true, _)) => {
                     tracing::info!("Refreshed HTTP routers (trigger change)");
-                },
+                }
                 Ok((false, _)) => {
-                    tracing::warn!("Should have refreshed HTTP routers (trigger change) but did not");
-                },
+                    tracing::warn!(
+                        "Should have refreshed HTTP routers (trigger change) but did not"
+                    );
+                }
                 Err(err) => {
                     tracing::error!("Error refreshing HTTP routers (trigger change): {err:#}");
                 }
             };
-        },
+        }
         "notify_token_invalidation" => {
-            tracing::info!("Token invalidation detected for token: {}...", payload.get(..8).unwrap_or(payload));
+            tracing::info!(
+                "Token invalidation detected for token: {}...",
+                payload.get(..8).unwrap_or(payload)
+            );
             windmill_api::auth::invalidate_token_from_cache(payload);
-        },
+        }
         "notify_global_setting_change" => {
             tracing::info!("Global setting change detected: {}", payload);
             match payload {
@@ -1486,176 +1522,150 @@ async fn process_notify_event(
                     if let Err(e) = reload_base_url_setting(conn).await {
                         tracing::error!(error = %e, "Could not reload base url setting");
                     }
-                },
+                }
                 OAUTH_SETTING => {
                     if let Err(e) = reload_base_url_setting(conn).await {
                         tracing::error!(error = %e, "Could not reload oauth setting");
                     }
-                },
+                }
                 CUSTOM_TAGS_SETTING => {
                     if let Err(e) = reload_custom_tags_setting(db).await {
                         tracing::error!(error = %e, "Could not reload custom tags setting");
                     }
-                },
+                }
                 LICENSE_KEY_SETTING => {
                     if let Err(e) = reload_license_key(&db.into()).await {
                         tracing::error!("Failed to reload license key: {e:#}");
                     }
-                },
+                }
                 DEFAULT_TAGS_PER_WORKSPACE_SETTING => {
                     if let Err(e) = load_tag_per_workspace_enabled(db).await {
                         tracing::error!("Error loading default tag per workspace: {e:#}");
                     }
-                },
+                }
                 DEFAULT_TAGS_WORKSPACES_SETTING => {
                     if let Err(e) = load_tag_per_workspace_workspaces(db).await {
-                        tracing::error!("Error loading default tag per workspace workspaces: {e:#}");
+                        tracing::error!(
+                            "Error loading default tag per workspace workspaces: {e:#}"
+                        );
                     }
-                },
+                }
                 SMTP_SETTING => {
                     reload_smtp_config(db).await;
-                },
+                }
                 TEAMS_SETTING => {
                     tracing::info!("Teams setting changed.");
-                },
+                }
                 INDEXER_SETTING => {
                     reload_indexer_config(db).await;
-                },
-                TIMEOUT_WAIT_RESULT_SETTING => {
-                    reload_timeout_wait_result_setting(conn).await
-                },
-                RETENTION_PERIOD_SECS_SETTING => {
-                    reload_retention_period_setting(conn).await
-                },
+                }
+                TIMEOUT_WAIT_RESULT_SETTING => reload_timeout_wait_result_setting(conn).await,
+                RETENTION_PERIOD_SECS_SETTING => reload_retention_period_setting(conn).await,
                 MONITOR_LOGS_ON_OBJECT_STORE_SETTING => {
                     reload_delete_logs_periodically_setting(conn).await
-                },
-                JOB_DEFAULT_TIMEOUT_SECS_SETTING => {
-                    reload_job_default_timeout_setting(conn).await
-                },
+                }
+                JOB_DEFAULT_TIMEOUT_SECS_SETTING => reload_job_default_timeout_setting(conn).await,
                 #[cfg(feature = "parquet")]
                 OBJECT_STORE_CONFIG_SETTING => {
                     if !disable_s3_store {
                         reload_object_store_setting(db).await;
                     }
-                },
-                SCIM_TOKEN_SETTING => {
-                    reload_scim_token_setting(conn).await
-                },
-                EXTRA_PIP_INDEX_URL_SETTING => {
-                    reload_extra_pip_index_url_setting(conn).await
-                },
-                PIP_INDEX_URL_SETTING => {
-                    reload_pip_index_url_setting(conn).await
-                },
+                }
+                SCIM_TOKEN_SETTING => reload_scim_token_setting(conn).await,
+                EXTRA_PIP_INDEX_URL_SETTING => reload_extra_pip_index_url_setting(conn).await,
+                PIP_INDEX_URL_SETTING => reload_pip_index_url_setting(conn).await,
                 INSTANCE_PYTHON_VERSION_SETTING => {
                     reload_instance_python_version_setting(conn).await
-                },
-                NPM_CONFIG_REGISTRY_SETTING => {
-                    reload_npm_config_registry_setting(conn).await
-                },
-                BUNFIG_INSTALL_SCOPES_SETTING => {
-                    reload_bunfig_install_scopes_setting(conn).await
-                },
-                NUGET_CONFIG_SETTING => {
-                    reload_nuget_config_setting(conn).await
-                },
-                POWERSHELL_REPO_URL_SETTING => {
-                    reload_powershell_repo_url_setting(conn).await
-                },
-                POWERSHELL_REPO_PAT_SETTING => {
-                    reload_powershell_repo_pat_setting(conn).await
-                },
-                MAVEN_REPOS_SETTING => {
-                    reload_maven_repos_setting(conn).await
-                },
-                NO_DEFAULT_MAVEN_SETTING => {
-                    reload_no_default_maven_setting(conn).await
-                },
-                RUBY_REPOS_SETTING => {
-                    reload_ruby_repos_setting(conn).await
-                },
-                HUB_API_SECRET_SETTING => {
-                    reload_hub_api_secret_setting(conn).await
-                },
+                }
+                NPM_CONFIG_REGISTRY_SETTING => reload_npm_config_registry_setting(conn).await,
+                BUNFIG_INSTALL_SCOPES_SETTING => reload_bunfig_install_scopes_setting(conn).await,
+                NUGET_CONFIG_SETTING => reload_nuget_config_setting(conn).await,
+                POWERSHELL_REPO_URL_SETTING => reload_powershell_repo_url_setting(conn).await,
+                POWERSHELL_REPO_PAT_SETTING => reload_powershell_repo_pat_setting(conn).await,
+                MAVEN_REPOS_SETTING => reload_maven_repos_setting(conn).await,
+                NO_DEFAULT_MAVEN_SETTING => reload_no_default_maven_setting(conn).await,
+                RUBY_REPOS_SETTING => reload_ruby_repos_setting(conn).await,
+                HUB_API_SECRET_SETTING => reload_hub_api_secret_setting(conn).await,
                 KEEP_JOB_DIR_SETTING => {
                     load_keep_job_dir(conn).await;
-                },
+                }
                 OTEL_TRACING_PROXY_SETTING => {
                     reload_otel_tracing_proxy_setting(conn).await;
                     if worker_mode {
                         tracing::info!("OTEL tracing proxy setting changed, restarting worker");
                         send_delayed_killpill(tx, 4, "OTEL tracing proxy setting change").await;
                     }
-                },
+                }
                 REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING => {
                     load_require_preexisting_user(db).await;
-                },
+                }
                 EXPOSE_METRICS_SETTING => {
                     tracing::info!("Metrics setting changed, restarting");
                     send_delayed_killpill(tx, 40, "metrics setting change").await;
-                },
+                }
                 EMAIL_DOMAIN_SETTING => {
                     tracing::info!("Email domain setting changed");
                     if server_mode {
                         send_delayed_killpill(tx, 4, "email domain setting change").await;
                     }
-                },
+                }
                 EXPOSE_DEBUG_METRICS_SETTING => {
                     if let Err(e) = load_metrics_debug_enabled(conn).await {
                         tracing::error!(error = %e, "Could not reload debug metrics setting");
                     }
-                },
+                }
                 APP_WORKSPACED_ROUTE_SETTING => {
                     if let Err(e) = reload_app_workspaced_route_setting(db).await {
                         tracing::error!(error = %e, "Could not reload app workspaced route setting");
                     }
-                },
+                }
                 OTEL_SETTING => {
                     tracing::info!("OTEL setting changed, restarting");
                     send_delayed_killpill(tx, 4, "OTEL setting change").await;
-                },
+                }
                 REQUEST_SIZE_LIMIT_SETTING => {
                     if server_mode {
                         tracing::info!("Request limit size change detected, killing server expecting to be restarted");
                         send_delayed_killpill(tx, 4, "request size limit change").await;
                     }
-                },
+                }
                 SAML_METADATA_SETTING => {
-                    tracing::info!("SAML metadata change detected, killing server expecting to be restarted");
+                    tracing::info!(
+                        "SAML metadata change detected, killing server expecting to be restarted"
+                    );
                     send_delayed_killpill(tx, 0, "SAML metadata change").await;
-                },
+                }
                 HUB_BASE_URL_SETTING => {
                     if let Err(e) = reload_hub_base_url_setting(conn, server_mode).await {
                         tracing::error!(error = %e, "Could not reload hub base url setting");
                     }
-                },
+                }
                 CRITICAL_ERROR_CHANNELS_SETTING => {
                     if let Err(e) = reload_critical_error_channels_setting(db).await {
                         tracing::error!(error = %e, "Could not reload critical error emails setting");
                     }
-                },
+                }
                 CRITICAL_ALERTS_ON_DB_OVERSIZE_SETTING => {
                     if let Err(e) = reload_critical_alerts_on_db_oversize(db).await {
                         tracing::error!(error = %e, "Could not reload critical alerts on db oversize setting");
                     }
-                },
+                }
                 JWT_SECRET_SETTING => {
                     if let Err(e) = reload_jwt_secret_setting(db).await {
                         tracing::error!(error = %e, "Could not reload jwt secret setting");
                     }
-                },
+                }
                 CRITICAL_ALERT_MUTE_UI_SETTING => {
                     tracing::info!("Critical alert UI setting changed");
                     if let Err(e) = reload_critical_alert_mute_ui_setting(conn).await {
                         tracing::error!(error = %e, "Could not reload critical alert UI setting");
                     }
-                },
+                }
                 _ => {
                     tracing::info!("Unrecognized Global Setting Change Payload: {:?}", payload);
                 }
             }
-        },
+        }
         _ => {
             tracing::warn!("Unknown notification channel: {}", channel);
         }

backend/tests/fixtures/schedule_push.sql

@@ -0,0 +1,84 @@
+-- Fixture for schedule push tests
+-- Sets up scripts, flows, users, and schedules needed to test push_scheduled_job
+
+-- Password entries for auth resolution
+INSERT INTO password (email, password_hash, login_type, super_admin, verified, name)
+VALUES
+    ('test@windmill.dev', 'dummy_hash', 'password', false, true, 'Test User'),
+    ('obo@windmill.dev', 'dummy_hash', 'password', false, true, 'OBO User')
+ON CONFLICT (email) DO NOTHING;
+
+-- OBO user in workspace
+INSERT INTO usr (workspace_id, email, username, is_admin, role)
+VALUES ('test-workspace', 'obo@windmill.dev', 'obo-user', false, 'Developer')
+ON CONFLICT (workspace_id, username) DO NOTHING;
+
+-- A simple script
+INSERT INTO script (workspace_id, created_by, content, schema, summary, description, path, hash, language, lock, kind)
+VALUES (
+    'test-workspace', 'test-user',
+    'export async function main() { return "ok"; }',
+    '{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}',
+    'Test script', '', 'f/system/test_script', 100001, 'deno', '', 'script'
+);
+
+-- A script with on_behalf_of_email
+INSERT INTO script (workspace_id, created_by, content, schema, summary, description, path, hash, language, lock, kind, on_behalf_of_email)
+VALUES (
+    'test-workspace', 'test-user',
+    'export async function main() { return "obo"; }',
+    '{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}',
+    'OBO script', '', 'f/system/obo_script', 100002, 'deno', '', 'script', 'obo@windmill.dev'
+);
+
+-- A script with a tag
+INSERT INTO script (workspace_id, created_by, content, schema, summary, description, path, hash, language, lock, kind, tag)
+VALUES (
+    'test-workspace', 'test-user',
+    'export async function main() { return "tagged"; }',
+    '{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}',
+    'Tagged script', '', 'f/system/tagged_script', 100003, 'deno', '', 'script', 'custom-tag'
+);
+
+-- A script with timeout
+INSERT INTO script (workspace_id, created_by, content, schema, summary, description, path, hash, language, lock, kind, timeout)
+VALUES (
+    'test-workspace', 'test-user',
+    'export async function main() { return "timeout"; }',
+    '{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}',
+    'Timeout script', '', 'f/system/timeout_script', 100004, 'deno', '', 'script', 300
+);
+
+-- A flow
+INSERT INTO flow (workspace_id, summary, description, path, versions, schema, value, edited_by)
+VALUES (
+    'test-workspace', 'Test flow', '', 'f/system/test_flow', '{200001}',
+    '{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}',
+    '{"modules": [{"id": "a", "value": {"path": "f/system/test_script", "type": "script", "input_transforms": {}}}]}',
+    'test-user'
+);
+
+INSERT INTO flow_version (id, workspace_id, path, schema, value, created_by)
+VALUES (
+    200001, 'test-workspace', 'f/system/test_flow',
+    '{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}',
+    '{"modules": [{"id": "a", "value": {"path": "f/system/test_script", "type": "script", "input_transforms": {}}}]}',
+    'test-user'
+);
+
+-- A flow with on_behalf_of_email
+INSERT INTO flow (workspace_id, summary, description, path, versions, schema, value, edited_by, on_behalf_of_email)
+VALUES (
+    'test-workspace', 'OBO flow', '', 'f/system/obo_flow', '{200002}',
+    '{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}',
+    '{"modules": [{"id": "a", "value": {"path": "f/system/test_script", "type": "script", "input_transforms": {}}}]}',
+    'test-user', 'obo@windmill.dev'
+);
+
+INSERT INTO flow_version (id, workspace_id, path, schema, value, created_by)
+VALUES (
+    200002, 'test-workspace', 'f/system/obo_flow',
+    '{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}',
+    '{"modules": [{"id": "a", "value": {"path": "f/system/test_script", "type": "script", "input_transforms": {}}}]}',
+    'test-user'
+);

backend/tests/schedule_push.rs

@@ -0,0 +1,591 @@
+mod common;
+
+mod schedule_push {
+    use chrono::Utc;
+    use sqlx::{Pool, Postgres};
+    use windmill_common::db::Authed;
+    use windmill_common::jobs::{JobKind, JobTriggerKind};
+    use windmill_common::schedule::Schedule;
+    use windmill_common::scripts::ScriptHash;
+    use windmill_common::users::username_to_permissioned_as;
+    use windmill_queue::jobs::{handle_maybe_scheduled_job, MiniCompletedJob};
+    use windmill_queue::schedule::push_scheduled_job;
+
+    fn make_schedule(overrides: impl FnOnce(&mut Schedule)) -> Schedule {
+        let mut s = Schedule {
+            workspace_id: "test-workspace".to_string(),
+            path: "f/system/test_schedule".to_string(),
+            edited_by: "test-user".to_string(),
+            edited_at: Utc::now(),
+            schedule: "0 0 */5 * * *".to_string(),
+            timezone: "UTC".to_string(),
+            enabled: true,
+            script_path: "f/system/test_script".to_string(),
+            is_flow: false,
+            args: None,
+            extra_perms: serde_json::json!({}),
+            email: "test@windmill.dev".to_string(),
+            error: None,
+            on_failure: None,
+            on_failure_times: None,
+            on_failure_exact: None,
+            on_failure_extra_args: None,
+            on_recovery: None,
+            on_recovery_times: None,
+            on_recovery_extra_args: None,
+            on_success: None,
+            on_success_extra_args: None,
+            ws_error_handler_muted: false,
+            retry: None,
+            no_flow_overlap: false,
+            summary: None,
+            description: None,
+            tag: None,
+            paused_until: None,
+            cron_version: None,
+            dynamic_skip: None,
+        };
+        overrides(&mut s);
+        s
+    }
+
+    fn make_authed() -> Authed {
+        Authed {
+            email: "test@windmill.dev".to_string(),
+            username: "test-user".to_string(),
+            is_admin: true,
+            is_operator: false,
+            groups: vec![],
+            folders: vec![],
+            scopes: None,
+            token_prefix: None,
+        }
+    }
+
+    fn make_completed_job(schedule: &Schedule) -> MiniCompletedJob {
+        MiniCompletedJob {
+            id: uuid::Uuid::new_v4(),
+            workspace_id: schedule.workspace_id.clone(),
+            runnable_id: Some(ScriptHash(100001)),
+            scheduled_for: Utc::now() - chrono::Duration::minutes(5),
+            parent_job: None,
+            flow_innermost_root_job: None,
+            runnable_path: Some(schedule.script_path.clone()),
+            kind: JobKind::Script,
+            started_at: Some(Utc::now() - chrono::Duration::minutes(4)),
+            permissioned_as: username_to_permissioned_as(&schedule.edited_by),
+            created_by: schedule.edited_by.clone(),
+            script_lang: None,
+            permissioned_as_email: schedule.email.clone(),
+            flow_step_id: None,
+            trigger_kind: Some(JobTriggerKind::Schedule),
+            trigger: Some(schedule.path.clone()),
+            priority: None,
+            concurrent_limit: None,
+            tag: "deno".to_string(),
+            cache_ttl: None,
+            cache_ignore_s3_path: None,
+            runnable_settings_handle: None,
+        }
+    }
+
+    async fn count_queued_jobs(db: &Pool<Postgres>) -> i64 {
+        sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM v2_job_queue")
+            .fetch_one(db)
+            .await
+            .unwrap()
+    }
+
+    async fn get_queued_job(
+        db: &Pool<Postgres>,
+    ) -> Option<(
+        String,         // workspace_id
+        Option<String>, // runnable_path
+        Option<String>, // trigger
+        Option<String>, // trigger_kind as text
+    )> {
+        sqlx::query_as::<_, (String, Option<String>, Option<String>, Option<String>)>(
+            "SELECT j.workspace_id, j.runnable_path, j.trigger, j.trigger_kind::text
+             FROM v2_job j JOIN v2_job_queue q ON j.id = q.id
+             LIMIT 1",
+        )
+        .fetch_optional(db)
+        .await
+        .unwrap()
+    }
+
+    // -----------------------------------------------------------------------
+    // push_scheduled_job: basic script schedule
+    // -----------------------------------------------------------------------
+
+    #[sqlx::test(fixtures("base", "schedule_push"))]
+    async fn test_push_script_schedule(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let schedule = make_schedule(|_| {});
+        let authed = make_authed();
+
+        let tx = db.begin().await?;
+        let tx = push_scheduled_job(&db, tx, &schedule, Some(&authed), None).await?;
+        tx.commit().await?;
+
+        assert_eq!(count_queued_jobs(&db).await, 1);
+        let (ws, path, trigger, trigger_kind) = get_queued_job(&db).await.unwrap();
+        assert_eq!(ws, "test-workspace");
+        assert_eq!(path.as_deref(), Some("f/system/test_script"));
+        assert_eq!(trigger.as_deref(), Some("f/system/test_schedule"));
+        assert_eq!(trigger_kind.as_deref(), Some("schedule"));
+        Ok(())
+    }
+
+    // -----------------------------------------------------------------------
+    // push_scheduled_job: flow schedule
+    // -----------------------------------------------------------------------
+
+    #[sqlx::test(fixtures("base", "schedule_push"))]
+    async fn test_push_flow_schedule(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let schedule = make_schedule(|s| {
+            s.is_flow = true;
+            s.script_path = "f/system/test_flow".to_string();
+            s.path = "f/system/flow_schedule".to_string();
+        });
+        let authed = make_authed();
+
+        let tx = db.begin().await?;
+        let tx = push_scheduled_job(&db, tx, &schedule, Some(&authed), None).await?;
+        tx.commit().await?;
+
+        assert_eq!(count_queued_jobs(&db).await, 1);
+        let (_, path, trigger, _) = get_queued_job(&db).await.unwrap();
+        assert_eq!(path.as_deref(), Some("f/system/test_flow"));
+        assert_eq!(trigger.as_deref(), Some("f/system/flow_schedule"));
+        Ok(())
+    }
+
+    // -----------------------------------------------------------------------
+    // push_scheduled_job: on_behalf_of_email (script)
+    // -----------------------------------------------------------------------
+
+    #[sqlx::test(fixtures("base", "schedule_push"))]
+    async fn test_push_script_on_behalf_of_email(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let schedule = make_schedule(|s| {
+            s.script_path = "f/system/obo_script".to_string();
+            s.path = "f/system/obo_schedule".to_string();
+        });
+
+        // No pre-computed authed: forces the obo path inside push_scheduled_job
+        let tx = db.begin().await?;
+        let tx = push_scheduled_job(&db, tx, &schedule, None, None).await?;
+        tx.commit().await?;
+
+        assert_eq!(count_queued_jobs(&db).await, 1);
+
+        let email = sqlx::query_scalar::<_, String>(
+            "SELECT permissioned_as_email FROM v2_job j JOIN v2_job_queue q ON j.id = q.id LIMIT 1",
+        )
+        .fetch_one(&db)
+        .await?;
+        assert_eq!(email, "obo@windmill.dev");
+        Ok(())
+    }
+
+    // -----------------------------------------------------------------------
+    // push_scheduled_job: on_behalf_of_email (flow)
+    // -----------------------------------------------------------------------
+
+    #[sqlx::test(fixtures("base", "schedule_push"))]
+    async fn test_push_flow_on_behalf_of_email(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let schedule = make_schedule(|s| {
+            s.is_flow = true;
+            s.script_path = "f/system/obo_flow".to_string();
+            s.path = "f/system/obo_flow_schedule".to_string();
+        });
+
+        let tx = db.begin().await?;
+        let tx = push_scheduled_job(&db, tx, &schedule, None, None).await?;
+        tx.commit().await?;
+
+        assert_eq!(count_queued_jobs(&db).await, 1);
+
+        let email = sqlx::query_scalar::<_, String>(
+            "SELECT permissioned_as_email FROM v2_job j JOIN v2_job_queue q ON j.id = q.id LIMIT 1",
+        )
+        .fetch_one(&db)
+        .await?;
+        assert_eq!(email, "obo@windmill.dev");
+        Ok(())
+    }
+
+    // -----------------------------------------------------------------------
+    // push_scheduled_job: with retry wraps in SingleStepFlow
+    // -----------------------------------------------------------------------
+
+    #[sqlx::test(fixtures("base", "schedule_push"))]
+    async fn test_push_script_with_retry(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let schedule = make_schedule(|s| {
+            s.retry = Some(serde_json::json!({
+                "constant": { "attempts": 3, "seconds": 10 }
+            }));
+        });
+        let authed = make_authed();
+
+        let tx = db.begin().await?;
+        let tx = push_scheduled_job(&db, tx, &schedule, Some(&authed), None).await?;
+        tx.commit().await?;
+
+        assert_eq!(count_queued_jobs(&db).await, 1);
+
+        // When retry is set, the job kind is singlescriptflow (SingleStepFlow wraps it)
+        let kind = sqlx::query_scalar::<_, String>(
+            "SELECT kind::text FROM v2_job j JOIN v2_job_queue q ON j.id = q.id LIMIT 1",
+        )
+        .fetch_one(&db)
+        .await?;
+        assert_eq!(kind, "singlestepflow");
+        Ok(())
+    }
+
+    // -----------------------------------------------------------------------
+    // push_scheduled_job: duplicate detection (same schedule + time = skip)
+    // -----------------------------------------------------------------------
+
+    #[sqlx::test(fixtures("base", "schedule_push"))]
+    async fn test_push_duplicate_skipped(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let schedule = make_schedule(|_| {});
+        let authed = make_authed();
+
+        // First push
+        let tx = db.begin().await?;
+        let tx = push_scheduled_job(&db, tx, &schedule, Some(&authed), None).await?;
+        tx.commit().await?;
+        assert_eq!(count_queued_jobs(&db).await, 1);
+
+        // Second push with same schedule — should be idempotent
+        let tx = db.begin().await?;
+        let tx = push_scheduled_job(&db, tx, &schedule, Some(&authed), None).await?;
+        tx.commit().await?;
+        assert_eq!(count_queued_jobs(&db).await, 1); // Still 1
+        Ok(())
+    }
+
+    // -----------------------------------------------------------------------
+    // push_scheduled_job: invalid timezone
+    // -----------------------------------------------------------------------
+
+    #[sqlx::test(fixtures("base", "schedule_push"))]
+    async fn test_push_invalid_timezone(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let schedule = make_schedule(|s| {
+            s.timezone = "Invalid/Timezone".to_string();
+        });
+        let authed = make_authed();
+
+        let tx = db.begin().await?;
+        let result = push_scheduled_job(&db, tx, &schedule, Some(&authed), None).await;
+        assert!(result.is_err());
+        assert_eq!(count_queued_jobs(&db).await, 0);
+        Ok(())
+    }
+
+    // -----------------------------------------------------------------------
+    // push_scheduled_job: invalid cron expression
+    // -----------------------------------------------------------------------
+
+    #[sqlx::test(fixtures("base", "schedule_push"))]
+    async fn test_push_invalid_cron(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let schedule = make_schedule(|s| {
+            s.schedule = "not a cron".to_string();
+        });
+        let authed = make_authed();
+
+        let tx = db.begin().await?;
+        let result = push_scheduled_job(&db, tx, &schedule, Some(&authed), None).await;
+        assert!(result.is_err());
+        assert_eq!(count_queued_jobs(&db).await, 0);
+        Ok(())
+    }
+
+    // -----------------------------------------------------------------------
+    // push_scheduled_job: invalid args (not a dict)
+    // -----------------------------------------------------------------------
+
+    #[sqlx::test(fixtures("base", "schedule_push"))]
+    async fn test_push_invalid_args(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let schedule = make_schedule(|s| {
+            let raw = serde_json::value::RawValue::from_string("[1,2,3]".to_string()).unwrap();
+            s.args = Some(sqlx::types::Json(raw));
+        });
+        let authed = make_authed();
+
+        let tx = db.begin().await?;
+        let result = push_scheduled_job(&db, tx, &schedule, Some(&authed), None).await;
+        assert!(result.is_err());
+        assert_eq!(count_queued_jobs(&db).await, 0);
+        Ok(())
+    }
+
+    // -----------------------------------------------------------------------
+    // push_scheduled_job: with schedule args passed to job
+    // -----------------------------------------------------------------------
+
+    #[sqlx::test(fixtures("base", "schedule_push"))]
+    async fn test_push_with_args(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let schedule = make_schedule(|s| {
+            let raw =
+                serde_json::value::RawValue::from_string(r#"{"key":"value"}"#.to_string()).unwrap();
+            s.args = Some(sqlx::types::Json(raw));
+        });
+        let authed = make_authed();
+
+        let tx = db.begin().await?;
+        let tx = push_scheduled_job(&db, tx, &schedule, Some(&authed), None).await?;
+        tx.commit().await?;
+
+        assert_eq!(count_queued_jobs(&db).await, 1);
+
+        let args = sqlx::query_scalar::<_, serde_json::Value>(
+            "SELECT args FROM v2_job j JOIN v2_job_queue q ON j.id = q.id LIMIT 1",
+        )
+        .fetch_one(&db)
+        .await?;
+        assert_eq!(args, serde_json::json!({"key": "value"}));
+        Ok(())
+    }
+
+    // -----------------------------------------------------------------------
+    // push_scheduled_job: script not found
+    // -----------------------------------------------------------------------
+
+    #[sqlx::test(fixtures("base", "schedule_push"))]
+    async fn test_push_script_not_found(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let schedule = make_schedule(|s| {
+            s.script_path = "f/system/nonexistent".to_string();
+        });
+        let authed = make_authed();
+
+        let tx = db.begin().await?;
+        let result = push_scheduled_job(&db, tx, &schedule, Some(&authed), None).await;
+        assert!(result.is_err());
+        assert_eq!(count_queued_jobs(&db).await, 0);
+        Ok(())
+    }
+
+    // -----------------------------------------------------------------------
+    // push_scheduled_job: flow not found
+    // -----------------------------------------------------------------------
+
+    #[sqlx::test(fixtures("base", "schedule_push"))]
+    async fn test_push_flow_not_found(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let schedule = make_schedule(|s| {
+            s.is_flow = true;
+            s.script_path = "f/system/nonexistent_flow".to_string();
+        });
+        let authed = make_authed();
+
+        let tx = db.begin().await?;
+        let result = push_scheduled_job(&db, tx, &schedule, Some(&authed), None).await;
+        assert!(result.is_err());
+        assert_eq!(count_queued_jobs(&db).await, 0);
+        Ok(())
+    }
+
+    // -----------------------------------------------------------------------
+    // push_scheduled_job: paused schedule (paused_until in future)
+    // -----------------------------------------------------------------------
+
+    #[sqlx::test(fixtures("base", "schedule_push"))]
+    async fn test_push_paused_schedule(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let schedule = make_schedule(|s| {
+            s.paused_until = Some(Utc::now() + chrono::Duration::hours(1));
+        });
+        let authed = make_authed();
+
+        let tx = db.begin().await?;
+        let tx = push_scheduled_job(&db, tx, &schedule, Some(&authed), None).await?;
+        tx.commit().await?;
+
+        // Job is still pushed, but scheduled_for will be after paused_until
+        assert_eq!(count_queued_jobs(&db).await, 1);
+
+        let scheduled_for = sqlx::query_scalar::<_, chrono::DateTime<Utc>>(
+            "SELECT scheduled_for FROM v2_job_queue LIMIT 1",
+        )
+        .fetch_one(&db)
+        .await?;
+        assert!(scheduled_for > Utc::now());
+        Ok(())
+    }
+
+    // -----------------------------------------------------------------------
+    // push_scheduled_job: clock shift detection (now_cutoff >= now)
+    // -----------------------------------------------------------------------
+
+    #[sqlx::test(fixtures("base", "schedule_push"))]
+    async fn test_push_clock_shift(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let schedule = make_schedule(|_| {});
+        let authed = make_authed();
+
+        // Pass a now_cutoff far in the future — simulates clock shift
+        let future_cutoff = Utc::now() + chrono::Duration::hours(24);
+        let tx = db.begin().await?;
+        let tx =
+            push_scheduled_job(&db, tx, &schedule, Some(&authed), Some(future_cutoff)).await?;
+        tx.commit().await?;
+
+        assert_eq!(count_queued_jobs(&db).await, 1);
+
+        // The scheduled_for should be after the cutoff
+        let scheduled_for = sqlx::query_scalar::<_, chrono::DateTime<Utc>>(
+            "SELECT scheduled_for FROM v2_job_queue LIMIT 1",
+        )
+        .fetch_one(&db)
+        .await?;
+        assert!(scheduled_for > future_cutoff);
+        Ok(())
+    }
+
+    // -----------------------------------------------------------------------
+    // handle_maybe_scheduled_job: disabled schedule does not push
+    // -----------------------------------------------------------------------
+
+    #[sqlx::test(fixtures("base", "schedule_push"))]
+    async fn test_handle_disabled_schedule(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let schedule = make_schedule(|s| {
+            s.enabled = false;
+        });
+        let job = make_completed_job(&schedule);
+
+        let result = handle_maybe_scheduled_job(
+            &db,
+            &job,
+            &schedule,
+            &schedule.script_path,
+            "test-workspace",
+        )
+        .await;
+        assert!(result.is_ok());
+        assert_eq!(count_queued_jobs(&db).await, 0);
+        Ok(())
+    }
+
+    // -----------------------------------------------------------------------
+    // handle_maybe_scheduled_job: script path mismatch does not push
+    // -----------------------------------------------------------------------
+
+    #[sqlx::test(fixtures("base", "schedule_push"))]
+    async fn test_handle_path_mismatch(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let schedule = make_schedule(|_| {});
+        let job = make_completed_job(&schedule);
+
+        let result = handle_maybe_scheduled_job(
+            &db,
+            &job,
+            &schedule,
+            "f/system/different_script",
+            "test-workspace",
+        )
+        .await;
+        assert!(result.is_ok());
+        assert_eq!(count_queued_jobs(&db).await, 0);
+        Ok(())
+    }
+
+    // -----------------------------------------------------------------------
+    // handle_maybe_scheduled_job: enabled + matching path pushes next job
+    // -----------------------------------------------------------------------
+
+    #[sqlx::test(fixtures("base", "schedule_push"))]
+    async fn test_handle_enabled_pushes_next_job(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let schedule = make_schedule(|_| {});
+        let job = make_completed_job(&schedule);
+
+        let result = handle_maybe_scheduled_job(
+            &db,
+            &job,
+            &schedule,
+            &schedule.script_path,
+            "test-workspace",
+        )
+        .await;
+        assert!(result.is_ok());
+        assert_eq!(count_queued_jobs(&db).await, 1);
+
+        let (_, path, trigger, trigger_kind) = get_queued_job(&db).await.unwrap();
+        assert_eq!(path.as_deref(), Some("f/system/test_script"));
+        assert_eq!(trigger.as_deref(), Some("f/system/test_schedule"));
+        assert_eq!(trigger_kind.as_deref(), Some("schedule"));
+        Ok(())
+    }
+
+    // -----------------------------------------------------------------------
+    // handle_maybe_scheduled_job: on_behalf_of_email via handle path
+    // -----------------------------------------------------------------------
+
+    #[sqlx::test(fixtures("base", "schedule_push"))]
+    async fn test_handle_on_behalf_of_email(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let schedule = make_schedule(|s| {
+            s.script_path = "f/system/obo_script".to_string();
+            s.path = "f/system/obo_schedule".to_string();
+        });
+        let job = make_completed_job(&schedule);
+
+        let result = handle_maybe_scheduled_job(
+            &db,
+            &job,
+            &schedule,
+            &schedule.script_path,
+            "test-workspace",
+        )
+        .await;
+        assert!(result.is_ok());
+        assert_eq!(count_queued_jobs(&db).await, 1);
+
+        let email = sqlx::query_scalar::<_, String>(
+            "SELECT permissioned_as_email FROM v2_job j JOIN v2_job_queue q ON j.id = q.id LIMIT 1",
+        )
+        .fetch_one(&db)
+        .await?;
+        assert_eq!(email, "obo@windmill.dev");
+        Ok(())
+    }
+
+    // -----------------------------------------------------------------------
+    // handle_maybe_scheduled_job: push failure disables schedule
+    // -----------------------------------------------------------------------
+
+    #[sqlx::test(fixtures("base", "schedule_push"))]
+    async fn test_handle_push_failure_disables_schedule(db: Pool<Postgres>) -> anyhow::Result<()> {
+        // Insert a schedule row so handle_maybe_scheduled_job can disable it
+        sqlx::query(
+            "INSERT INTO schedule (workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, email, extra_perms, ws_error_handler_muted, no_flow_overlap)
+             VALUES ('test-workspace', 'f/system/bad_schedule', 'test-user', now(), '0 0 */5 * * *', 'UTC', true, 'f/system/nonexistent', false, 'test@windmill.dev', '{}', false, false)"
+        )
+        .execute(&db)
+        .await?;
+
+        let schedule = make_schedule(|s| {
+            s.path = "f/system/bad_schedule".to_string();
+            s.script_path = "f/system/nonexistent".to_string();
+        });
+        let job = make_completed_job(&schedule);
+
+        let result = handle_maybe_scheduled_job(
+            &db,
+            &job,
+            &schedule,
+            &schedule.script_path,
+            "test-workspace",
+        )
+        .await;
+        // Should succeed (error is handled internally by disabling schedule)
+        assert!(result.is_ok());
+        assert_eq!(count_queued_jobs(&db).await, 0);
+
+        // Schedule should be disabled with an error
+        let (enabled, error): (bool, Option<String>) = sqlx::query_as(
+            "SELECT enabled, error FROM schedule WHERE workspace_id = 'test-workspace' AND path = 'f/system/bad_schedule'",
+        )
+        .fetch_one(&db)
+        .await?;
+        assert!(!enabled);
+        assert!(error.is_some());
+        Ok(())
+    }
+}

backend/windmill-api/Cargo.toml

@@ -11,17 +11,18 @@ path = "src/lib.rs"
 [features]
 default = []
 private = ["windmill-audit/private", "windmill-common/private"]
-enterprise = ["windmill-queue/enterprise", "windmill-audit/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "windmill-worker/enterprise", "dep:windmill-autoscaling"]
+enterprise = ["windmill-queue/enterprise", "windmill-audit/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "windmill-worker/enterprise"]
 stripe = []
-agent_worker_server = ["dep:windmill-parser-py-imports"]
+agent_worker_server = []
 enterprise_saml = ["dep:samael", "dep:libxml"]
 benchmark = []
 embedding = ["dep:tinyvector", "dep:hf-hub", "dep:tokenizers", "dep:candle-core", "dep:candle-transformers", "dep:candle-nn"]
-parquet = ["dep:datafusion", "dep:object_store", "windmill-common/parquet", "windmill-worker/parquet", "dep:aws-sigv4", "dep:aws-sdk-config"]
+parquet = ["dep:datafusion", "dep:object_store", "windmill-common/parquet", "windmill-worker/parquet"]
 prometheus = ["windmill-common/prometheus", "windmill-queue/prometheus", "dep:prometheus", "windmill-worker/prometheus"]
 openidconnect = ["dep:openidconnect", "windmill-common/openidconnect"]
 tantivy = ["dep:windmill-indexer"]
 kafka = ["dep:rdkafka", "dep:rdkafka-sys"]
+kafka-gssapi = ["kafka", "rdkafka/gssapi"]
 nats = ["dep:async-nats", "dep:nkeys"]
 websocket = ["dep:tokio-tungstenite"]
 smtp = ["dep:mail-parser", "dep:openssl", "windmill-common/smtp"]
@@ -38,8 +39,8 @@ deno_core = ["dep:deno_core", "dep:deno_error"]
 gcp_trigger = ["dep:thiserror", "dep:google-cloud-pubsub", "dep:google-cloud-googleapis", "dep:tonic"]
 cloud = ["windmill-common/cloud"]
 mcp = ["dep:windmill-mcp", "windmill-mcp/server", "windmill-mcp/auth"]
-bedrock = ["dep:aws-sdk-bedrock", "dep:aws-sdk-bedrockruntime", "windmill-common/bedrock", "dep:aws-config", "dep:aws-sdk-config", "dep:aws-credential-types", "dep:aws-smithy-types"]
-python = ["dep:windmill-parser-py-imports"]
+bedrock = ["dep:aws-sdk-bedrock", "dep:aws-sdk-bedrockruntime", "windmill-common/bedrock", "dep:aws-config"]
+python = []
 
 [dependencies]
 windmill-mcp = { workspace = true, optional = true }
@@ -50,10 +51,10 @@ windmill-parser.workspace = true
 windmill-parser-sql.workspace = true
 windmill-parser-ts.workspace = true
 windmill-parser-py.workspace = true
-windmill-parser-py-imports = { workspace = true, optional = true }
+windmill-parser-py-imports.workspace = true
 windmill-git-sync.workspace = true
 windmill-indexer = { workspace = true, optional = true }
-windmill-autoscaling = { workspace = true, optional = true }
+windmill-autoscaling.workspace = true
 windmill-worker.workspace = true
 tokio.workspace = true
 tokio-stream.workspace = true
@@ -152,13 +153,13 @@ aws-sdk-ssooidc = { workspace = true, optional = true }
 aws-sdk-sts = { workspace = true, optional = true }
 rustls = { workspace = true }
 
-aws-sigv4 = { workspace = true, optional = true }
-aws-sdk-config = { workspace = true, optional = true }
+aws-sigv4.workspace = true
+aws-sdk-config.workspace = true
 aws-config = { workspace = true, optional = true }
-aws-credential-types = { workspace = true, optional = true }
+aws-credential-types.workspace = true
 aws-sdk-bedrock = { workspace = true, optional = true }
 aws-sdk-bedrockruntime = { workspace = true, optional = true }
-aws-smithy-types = { workspace = true, optional = true }
+aws-smithy-types.workspace = true
 async-trait.workspace = true
 google-cloud-pubsub = { workspace = true, optional = true }
 google-cloud-googleapis = { workspace = true , optional = true }
@@ -169,6 +170,7 @@ tar.workspace = true
 flate2.workspace = true
 backon = {workspace = true, optional = true}
 strum = { workspace = true, optional = true }
+dashmap.workspace = true
 
 [build-dependencies]
 deno_core = { workspace = true, optional = true }

backend/windmill-api/src/ai.rs

@@ -4,19 +4,21 @@ use crate::db::{ApiAuthed, DB};
 
 #[cfg(feature = "bedrock")]
 use axum::routing::get;
-use axum::{
-    body::Bytes, extract::Path, response::IntoResponse, routing::post, Extension, Router,
-};
 #[cfg(feature = "bedrock")]
 use axum::Json;
+use axum::{body::Bytes, extract::Path, response::IntoResponse, routing::post, Extension, Router};
+use futures::StreamExt;
 use http::{HeaderMap, Method};
 use quick_cache::sync::Cache;
 use reqwest::{Client, RequestBuilder};
 use serde::{Deserialize, Serialize};
 use serde_json::{json, value::RawValue};
 use std::collections::HashMap;
+use std::time::Duration;
 use windmill_audit::{audit_oss::audit_log, ActionKind};
-use windmill_common::ai_providers::{empty_string_as_none, AIProvider, ProviderConfig, ProviderModel};
+use windmill_common::ai_providers::{
+    empty_string_as_none, AIProvider, ProviderConfig, ProviderModel,
+};
 use windmill_common::error::{to_anyhow, Error, Result};
 use windmill_common::utils::configure_client;
 use windmill_common::variables::get_variable_or_self;
@@ -27,6 +29,7 @@ const AI_TIMEOUT_MAX_SECS: u64 = 86400; // 24 hours
 const AI_TIMEOUT_DEFAULT_SECS: u64 = 3600; // 1 hour
 const HTTP_POOL_MAX_IDLE_PER_HOST: usize = 10;
 const HTTP_POOL_IDLE_TIMEOUT_SECS: u64 = 90;
+const KEEPALIVE_INTERVAL_SECS: u64 = 15;
 
 lazy_static::lazy_static! {
     /// AI request timeout in seconds.
@@ -151,9 +154,17 @@ struct AIStandardResource {
     organization_id: Option<String>,
     #[serde(default, deserialize_with = "empty_string_as_none")]
     region: Option<String>,
-    #[serde(alias = "awsAccessKeyId", default, deserialize_with = "empty_string_as_none")]
+    #[serde(
+        alias = "awsAccessKeyId",
+        default,
+        deserialize_with = "empty_string_as_none"
+    )]
     aws_access_key_id: Option<String>,
-    #[serde(alias = "awsSecretAccessKey", default, deserialize_with = "empty_string_as_none")]
+    #[serde(
+        alias = "awsSecretAccessKey",
+        default,
+        deserialize_with = "empty_string_as_none"
+    )]
     aws_secret_access_key: Option<String>,
     /// Platform for Anthropic API (standard or google_vertex_ai)
     #[serde(default)]
@@ -213,9 +224,7 @@ impl AIRequestConfig {
                 let base_url = if matches!(provider, AIProvider::AWSBedrock) {
                     String::new()
                 } else {
-                    provider
-                        .get_base_url(resource.base_url, db)
-                        .await?
+                    provider.get_base_url(resource.base_url, db).await?
                 };
                 let api_key = if let Some(api_key) = resource.api_key {
                     Some(get_variable_or_self(api_key, db, w_id).await?)
@@ -338,7 +347,8 @@ impl AIRequestConfig {
 
         let is_azure = provider.is_azure_openai(base_url);
         let is_anthropic = matches!(provider, AIProvider::Anthropic);
-        let is_anthropic_vertex = is_anthropic && self.platform == AnthropicPlatform::GoogleVertexAi;
+        let is_anthropic_vertex =
+            is_anthropic && self.platform == AnthropicPlatform::GoogleVertexAi;
         let is_anthropic_sdk = headers.get("X-Anthropic-SDK").is_some();
         let is_google_ai = matches!(provider, AIProvider::GoogleAI);
 
@@ -477,7 +487,9 @@ fn transform_anthropic_for_vertex(body: &Bytes) -> Result<(String, Bytes)> {
     let model = json_body
         .remove("model")
         .and_then(|v| v.as_str().map(|s| s.to_string()))
-        .ok_or_else(|| Error::BadRequest("Missing 'model' field in Anthropic request".to_string()))?;
+        .ok_or_else(|| {
+            Error::BadRequest("Missing 'model' field in Anthropic request".to_string())
+        })?;
 
     // Add anthropic_version to body (required for Vertex AI)
     json_body.insert(
@@ -561,6 +573,40 @@ async fn check_bedrock_credentials(
     Ok(Json(response))
 }
 
+fn is_sse_response(headers: &HeaderMap) -> bool {
+    headers
+        .get(http::header::CONTENT_TYPE)
+        .and_then(|v| v.to_str().ok())
+        .map(|ct| ct.contains("text/event-stream"))
+        .unwrap_or(false)
+}
+
+fn inject_keepalives<S>(
+    upstream: S,
+    interval: Duration,
+) -> impl futures::Stream<Item = std::result::Result<Bytes, reqwest::Error>>
+where
+    S: futures::Stream<Item = std::result::Result<Bytes, reqwest::Error>> + Unpin,
+{
+    async_stream::stream! {
+        tokio::pin!(upstream);
+        loop {
+            tokio::select! {
+                biased;
+                chunk = upstream.next() => {
+                    match chunk {
+                        Some(item) => yield item,
+                        None => break,
+                    }
+                }
+                _ = tokio::time::sleep(interval) => {
+                    yield Ok(Bytes::from(": keepalive\n\n"));
+                }
+            }
+        }
+    }
+}
+
 async fn global_proxy(
     authed: ApiAuthed,
     Extension(db): Extension<DB>,
@@ -625,7 +671,15 @@ async fn global_proxy(
     let status_code = response.status();
     let headers = response.headers().clone();
     let stream = response.bytes_stream();
-    Ok((status_code, headers, axum::body::Body::from_stream(stream)))
+    let body = if is_sse_response(&headers) {
+        axum::body::Body::from_stream(inject_keepalives(
+            stream,
+            Duration::from_secs(KEEPALIVE_INTERVAL_SECS),
+        ))
+    } else {
+        axum::body::Body::from_stream(stream)
+    };
+    Ok((status_code, headers, body))
 }
 
 async fn proxy(
@@ -732,20 +786,21 @@ async fn proxy(
     #[cfg(feature = "bedrock")]
     {
         // Extract model and streaming flag for Bedrock transformation (only for POST requests)
-        let (model, is_streaming) =
-            if matches!(provider, AIProvider::AWSBedrock) && method == Method::POST {
-                #[derive(Deserialize, Debug)]
-                struct BedrockRequest {
-                    model: String,
-                    #[serde(default)]
-                    stream: bool,
-                }
-                let parsed: BedrockRequest = serde_json::from_slice(&body)
-                    .map_err(|e| Error::internal_err(format!("Failed to parse request body: {}", e)))?;
-                (Some(parsed.model), parsed.stream)
-            } else {
-                (None, false)
-            };
+        let (model, is_streaming) = if matches!(provider, AIProvider::AWSBedrock)
+            && method == Method::POST
+        {
+            #[derive(Deserialize, Debug)]
+            struct BedrockRequest {
+                model: String,
+                #[serde(default)]
+                stream: bool,
+            }
+            let parsed: BedrockRequest = serde_json::from_slice(&body)
+                .map_err(|e| Error::internal_err(format!("Failed to parse request body: {}", e)))?;
+            (Some(parsed.model), parsed.stream)
+        } else {
+            (None, false)
+        };
 
         // For Bedrock requests, use the SDK-based approach
         if matches!(provider, AIProvider::AWSBedrock) {
@@ -850,5 +905,13 @@ async fn proxy(
     let status_code = response.status();
     let headers = response.headers().clone();
     let stream = response.bytes_stream();
-    Ok((status_code, headers, axum::body::Body::from_stream(stream)))
+    let body = if is_sse_response(&headers) {
+        axum::body::Body::from_stream(inject_keepalives(
+            stream,
+            Duration::from_secs(KEEPALIVE_INTERVAL_SECS),
+        ))
+    } else {
+        axum::body::Body::from_stream(stream)
+    };
+    Ok((status_code, headers, body))
 }

backend/windmill-api/src/apps.rs

@@ -126,6 +126,7 @@ pub fn global_service() -> Router {
     Router::new()
         .route("/hub/list", get(list_hub_apps))
         .route("/hub/get/:id", get(get_hub_app_by_id))
+        .route("/hub/get_raw/:id", get(get_hub_raw_app_by_id))
 }
 
 #[derive(FromRow, Deserialize, Serialize)]
@@ -1312,6 +1313,24 @@ pub async fn get_hub_app_by_id(
     Ok(Json(value))
 }
 
+pub async fn get_hub_raw_app_by_id(
+    Path(id): Path<i32>,
+    Extension(db): Extension<DB>,
+) -> JsonResult<Box<serde_json::value::RawValue>> {
+    let value = http_get_from_hub(
+        &HTTP_CLIENT,
+        &format!("{}/raw_apps/{}/json", *HUB_BASE_URL.read().await, id),
+        false,
+        None,
+        Some(&db),
+    )
+    .await?
+    .json()
+    .await
+    .map_err(to_anyhow)?;
+    Ok(Json(value))
+}
+
 async fn delete_app(
     authed: ApiAuthed,
     Extension(db): Extension<DB>,
@@ -1910,6 +1929,15 @@ async fn execute_component(
         }
     };
 
+    // Check rate limit for anonymous (public) executions
+    if matches!(policy.execution_mode, ExecutionMode::Anonymous) && opt_authed.is_none() {
+        if let Some(limit) = crate::workspaces::get_public_app_rate_limit(&db, &w_id).await? {
+            if limit > 0 {
+                crate::public_app_rate_limit::check_and_increment(&w_id, limit)?;
+            }
+        }
+    }
+
     // Execution is publisher and an user is authenticated: check if the user is authorized to
     // execute the app.
     if let (ExecutionMode::Publisher, Some(authed)) = (policy.execution_mode, opt_authed.as_ref()) {

backend/windmill-api/src/assets.rs

@@ -18,6 +18,7 @@ pub fn workspaced_service() -> Router {
     Router::new()
         .route("/list", get(list_assets))
         .route("/list_by_usages", post(list_assets_by_usages))
+        .route("/list_favorites", get(list_favorites))
 }
 
 #[derive(Deserialize)]
@@ -158,6 +159,7 @@ async fn list_assets(
                         'path', asset.usage_path,
                         'kind', asset.usage_kind,
                         'access_type', asset.usage_access_type,
+                        'columns', asset.columns,
                         'created_at', asset.created_at,
                         'metadata', (CASE
                             WHEN asset.usage_kind = 'job' THEN
@@ -182,7 +184,13 @@ async fn list_assets(
         FROM asset
         INNER JOIN asset_summary ON asset.path = asset_summary.path AND asset.kind = asset_summary.kind
         LEFT JOIN resource ON asset.kind = 'resource'
-          AND array_to_string((string_to_array(asset.path, '/'))[1:3], '/') = resource.path
+          AND (
+            -- Extract base path before '?' for ?table= syntax
+            CASE
+              WHEN asset.path LIKE '%?%' THEN split_part(asset.path, '?', 1)
+              ELSE asset.path
+            END
+          ) = resource.path
           AND resource.workspace_id = $1
         LEFT JOIN v2_job job ON asset.usage_kind = 'job'
           AND asset.usage_path = job.id::text
@@ -266,11 +274,12 @@ async fn list_assets_by_usages(
     for usage in body.usages {
         let assets = sqlx::query_scalar!(
             r#"SELECT
-                jsonb_build_object(
+                jsonb_strip_nulls(jsonb_build_object(
                     'path', path,
                     'kind', kind,
-                    'access_type', usage_access_type
-                ) as "list!: _"
+                    'access_type', usage_access_type,
+                    'columns', columns
+                )) as "list!: _"
             FROM asset
             WHERE workspace_id = $1 AND usage_path = $2 AND usage_kind = $3
             ORDER BY path, kind"#,
@@ -284,3 +293,29 @@ async fn list_assets_by_usages(
     }
     Ok(Json(assets_vec))
 }
+
+async fn list_favorites(
+    authed: ApiAuthed,
+    Path(w_id): Path<String>,
+    Extension(user_db): Extension<UserDB>,
+) -> JsonResult<Vec<Value>> {
+    let mut tx = user_db.begin(&authed).await?;
+
+    let favorites = sqlx::query_scalar!(
+        r#"SELECT
+            jsonb_strip_nulls(jsonb_build_object(
+                'path', favorite.path
+            )) as "favorite_asset!: _"
+        FROM favorite
+        WHERE favorite.workspace_id = $1
+          AND favorite.usr = $2
+          AND favorite_kind = 'asset'
+        "#,
+        &w_id,
+        &authed.username
+    )
+    .fetch_all(&mut *tx)
+    .await?;
+
+    Ok(Json(favorites))
+}

backend/windmill-api/src/db.rs

@@ -64,6 +64,19 @@ lazy_static::lazy_static! {
                     (20260126235947, include_str!(
                         "../../custom_migrations/lowercase_emails_safe.sql"
                     ).to_string()),
+                    (20260206000000, "".to_string()),
+                    (20260207000001, include_str!(
+                        "../../migrations/20260207000001_concurrent_indexes_v2_job.up.sql"
+                    ).to_string()),
+                    (20260207000002, include_str!(
+                        "../../migrations/20260207000002_concurrent_indexes_v2_job_completed.up.sql"
+                    ).to_string()),
+                    (20260207000003, include_str!(
+                        "../../migrations/20260207000003_concurrent_indexes_v2_job_queue.up.sql"
+                    ).to_string()),
+                    (20260207000004, include_str!(
+                        "../../migrations/20260207000004_concurrent_indexes_other.up.sql"
+                    ).to_string()),
                     ].into_iter().collect();
 }
 
@@ -177,11 +190,25 @@ impl Migrate for CustomMigrator {
 
             if let Some(migration_sql) = OVERRIDDEN_MIGRATIONS.get(&migration.version) {
                 tracing::info!("Using custom migration for version {}", migration.version);
-                // tracing::info!("Migration SQL: {}", migration_sql);
 
-                self.inner
-                    .execute(&**migration_sql)
-                    .await?;
+                if migration_sql.contains("CONCURRENTLY") {
+                    // CONCURRENTLY operations cannot run inside a transaction block
+                    // or a multi-statement query (PostgreSQL requires top-level execution).
+                    // Split into individual statements and execute each separately.
+                    for stmt in migration_sql.split(';') {
+                        let stmt = stmt.trim();
+                        if !stmt.is_empty()
+                            && stmt.lines().any(|l| {
+                                let t = l.trim();
+                                !t.is_empty() && !t.starts_with("--")
+                            })
+                        {
+                            self.inner.execute(stmt).await?;
+                        }
+                    }
+                } else if !migration_sql.is_empty() {
+                    self.inner.execute(&**migration_sql).await?;
+                }
                 let _ = sqlx::query(
                     r#"
                 INSERT INTO _sqlx_migrations ( version, description, success, checksum, execution_time )

backend/windmill-api/src/favorite.rs

@@ -31,6 +31,7 @@ pub enum FavoriteKind {
     App,
     #[allow(non_camel_case_types)]
     Raw_App,
+    Asset,
 }
 #[derive(Deserialize)]
 pub struct Favorite {

backend/windmill-api/src/flows.rs

@@ -473,12 +473,14 @@ async fn create_flow(
         workspace_id, path, summary, description,
         dependency_job, lock_error_logs, draft_only, tag,
         dedicated_worker, visible_to_runner_only, on_behalf_of_email,
+        ws_error_handler_muted,
         value, schema, edited_by, edited_at
     ) VALUES (
         $1, $2, $3, $4,
         NULL, '', $5, $6,
         $7, $8, $9,
-        $10, $11::text::json, $12, now()
+        $10,
+        $11, $12::text::json, $13, now()
     )"#,
         w_id,
         nf.path,
@@ -489,6 +491,7 @@ async fn create_flow(
         nf.dedicated_worker,
         nf.visible_to_runner_only.unwrap_or(false),
         nf.on_behalf_of_email.and(Some(&authed.email)),
+        nf.ws_error_handler_muted.unwrap_or(false),
         sqlx::types::Json(&nf.value) as _,
         schema_str,
         &authed.username,
@@ -897,12 +900,13 @@ async fn update_flow(
             dedicated_worker = $5,
             visible_to_runner_only = $6,
             on_behalf_of_email = $7,
-            value = $8,
-            schema = $9::text::json,
-            edited_by = $10,
+            ws_error_handler_muted = $8,
+            value = $9,
+            schema = $10::text::json,
+            edited_by = $11,
             edited_at = now()
         WHERE
-            path = $11 AND workspace_id = $12",
+            path = $12 AND workspace_id = $13",
         if is_new_path { flow_path } else { &nf.path },
         nf.summary,
         nf.description.as_deref().unwrap_or(""),
@@ -910,6 +914,7 @@ async fn update_flow(
         nf.dedicated_worker,
         nf.visible_to_runner_only.unwrap_or(false),
         nf.on_behalf_of_email.and(Some(&authed.email)),
+        nf.ws_error_handler_muted.unwrap_or(false),
         sqlx::types::Json(&nf.value) as _,
         schema_str,
         authed.username,

backend/windmill-api/src/jobs.rs

@@ -6378,10 +6378,18 @@ fn register_potential_assets_on_inline_execution(
     match assets {
         Some(Ok(assets)) => {
             for asset in assets {
+                let columns = asset.columns.as_ref().map(|cols| {
+                    cols.iter()
+                        .map(|(col_name, col_access_type)| {
+                            (col_name.clone(), (*col_access_type).into())
+                        })
+                        .collect()
+                });
                 register_runtime_asset(InsertRuntimeAssetParams {
                     access_type: asset.access_type.map(|a| a.into()),
                     asset_kind: asset.kind.into(),
                     asset_path: asset.path,
+                    columns,
                     job_id,
                     workspace_id: w_id.to_string(),
                     created_at: None,

backend/windmill-api/src/lib.rs

@@ -167,6 +167,7 @@ mod teams_approvals_oss;
 #[cfg(feature = "native_trigger")]
 pub mod native_triggers;
 mod public_app_layer;
+mod public_app_rate_limit;
 mod static_assets;
 #[cfg(all(feature = "stripe", feature = "enterprise", feature = "private"))]
 pub mod stripe_ee;

backend/windmill-api/src/live_migrations.rs

@@ -18,13 +18,6 @@ pub async fn custom_migrations(migrator: &mut CustomMigrator, db: &DB) -> Result
         tracing::error!("Could not apply flow versioning fix migration: {err:#}");
     }
 
-    let db2 = db.clone();
-    let _ = tokio::task::spawn(async move {
-        if let Err(err) = fix_job_completed_index(&db2).await {
-            tracing::error!("Could not apply job completed index fix migration: {err:#}");
-        }
-    });
-
     Ok(())
 }
 
@@ -75,376 +68,3 @@ async fn fix_flow_versioning_migration(
     Ok(())
 }
 
-async fn has_done_migration(db: &DB, migration_job_name: &str) -> bool {
-    sqlx::query_scalar!(
-        "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = $1)",
-        migration_job_name
-    )
-    .fetch_one(db)
-    .await
-    .ok()
-    .flatten()
-    .unwrap_or(false)
-}
-
-use sqlx::Pool;
-
-macro_rules! run_windmill_migration {
-    ($migration_job_name:expr, $db:expr, |$tx:ident| $code:block) => {
-        {
-            let migration_job_name = $migration_job_name;
-            let db: &Pool<Postgres> = $db;
-
-            let has_done = has_done_migration(db, migration_job_name).await;
-            if !has_done {
-                tracing::info!("Applying {migration_job_name} migration");
-                let mut $tx = db.begin().await?;
-                let mut r = false;
-                while !r {
-                    r = sqlx::query_scalar!("SELECT pg_try_advisory_lock(4242)")
-                        .fetch_one(&mut *$tx)
-                        .await
-                        .map_err(|e| {
-                            tracing::error!("Error acquiring {migration_job_name} lock: {e:#}");
-                            sqlx::migrate::MigrateError::Execute(e)
-                        })?
-                        .unwrap_or(false);
-
-                    if !r {
-                        tracing::info!("PG {migration_job_name} lock already acquired by another server or worker, retrying in 5s. (look for the advisory lock in pg_lock with granted = true)");
-                        drop($tx);
-                        tokio::time::sleep(std::time::Duration::from_secs(5)).await;
-                        $tx = db.begin().await?;
-                    }
-                }
-                tracing::info!("acquired lock for {migration_job_name}");
-
-                let has_done = has_done_migration(db, migration_job_name).await;
-
-                if !has_done {
-
-                    $code
-
-                    sqlx::query!(
-                        "INSERT INTO windmill_migrations (name) VALUES ($1) ON CONFLICT DO NOTHING",
-                        migration_job_name
-                    )
-                    .execute(&mut *$tx)
-                    .await?;
-                    tracing::info!("Finished applying {migration_job_name} migration");
-                } else {
-                    tracing::debug!("migration {migration_job_name} already done");
-                }
-
-                let _ = sqlx::query("SELECT pg_advisory_unlock(4242)")
-                    .execute(&mut *$tx)
-                    .await?;
-                $tx.commit().await?;
-                tracing::info!("released lock for {migration_job_name}");
-            } else {
-                tracing::debug!("migration {migration_job_name} already done");
-
-            }
-        }
-    };
-}
-
-async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
-    // let has_done_migration = sqlx::query_scalar!(
-    //     "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_job_completed_index')"
-    // )
-    // .fetch_one(db)
-    // .await?
-    // .unwrap_or(false);
-    // if !has_done_migration {
-    //     tracing::info!("Applying fix_job_completed_index migration");
-    //     let mut tx = db.begin().await?;
-    //     let mut r = false;
-    //     while !r {
-    //         r = sqlx::query_scalar!("SELECT pg_try_advisory_lock(4242)")
-    //             .fetch_one(&mut *tx)
-    //             .await
-    //             .map_err(|e| {
-    //                 tracing::error!("Error acquiring fix_job_completed_index lock: {e:#}");
-    //                 sqlx::migrate::MigrateError::Execute(e)
-    //             })?
-    //             .unwrap_or(false);
-    //         if !r {
-    //             tracing::info!("PG fix_job_completed_index_migration lock already acquired by another server or worker, retrying in 5s. (look for the advisory lock in pg_lock with granted = true)");
-    //             tokio::time::sleep(std::time::Duration::from_secs(5)).await;
-    //         }
-    //     }
-    //     // sqlx::query(
-    //     //     "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_created_at_new ON completed_job (workspace_id, job_kind, is_skipped, is_flow_step, created_at DESC, started_at DESC)"
-    //     // ).execute(db).await?;
-
-    //     sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at")
-    //         .execute(db)
-    //         .await?;
-
-    //     sqlx::query!("INSERT INTO windmill_migrations (name) VALUES ('fix_job_completed_index') ON CONFLICT DO NOTHING")
-    //         .execute(&mut *tx)
-    //         .await?;
-    //     let _ = sqlx::query("SELECT pg_advisory_unlock(4242)")
-    //         .execute(&mut *tx)
-    //         .await?;
-    //     tx.commit().await?;
-    // }
-
-    run_windmill_migration!("fix_job_completed_index_2", &db, |tx| {
-        //     sqlx::query(
-        //     "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_created_at_new_2 ON completed_job (workspace_id, job_kind, success, is_skipped, is_flow_step, created_at DESC)"
-        // ).execute(db).await?;
-
-        //     sqlx::query(
-        //     "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_started_at_new ON completed_job (workspace_id, job_kind, success, is_skipped, is_flow_step, started_at DESC)"
-        // ).execute(db).await?;
-
-        sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at")
-            .execute(db)
-            .await?;
-
-        sqlx::query(
-            "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at_new",
-        )
-        .execute(db)
-        .await?;
-    });
-
-    run_windmill_migration!("fix_job_completed_index_3", &db, |tx| {
-        sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS index_completed_job_on_schedule_path")
-            .execute(db)
-            .await?;
-
-        sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS concurrency_limit_stats_queue")
-            .execute(db)
-            .await?;
-
-        sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS root_job_index")
-            .execute(db)
-            .await?;
-
-        sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS index_completed_on_created")
-            .execute(db)
-            .await?;
-    });
-
-    run_windmill_migration!("fix_job_index_1_II", &db, |tx| {
-        let migration_job_name = "fix_job_index_1_II";
-        let mut i = 1;
-        tracing::info!("step {i} of {migration_job_name} migration");
-        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_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_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_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!("DROP INDEX CONCURRENTLY IF EXISTS root_job_index_by_path_2")
-            .execute(db)
-            .await?;
-
-        i += 1;
-        tracing::info!("step {i} of {migration_job_name} migration");
-
-        sqlx::query(
-            "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at_new_2",
-        )
-        .execute(db)
-        .await?;
-        i += 1;
-        tracing::info!("step {i} of {migration_job_name} migration");
-
-        sqlx::query(
-            "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_started_at_new",
-        )
-        .execute(db)
-        .await?;
-        i += 1;
-        tracing::info!("step {i} of {migration_job_name} migration");
-
-        sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS root_job_index_by_path")
-            .execute(db)
-            .await?;
-    });
-
-    run_windmill_migration!("fix_labeled_jobs_index", &db, |tx| {
-        tracing::info!("Special migration to add index concurrently on job labels 2");
-        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS labeled_jobs_on_jobs")
-            .execute(db)
-            .await?;
-        sqlx::query!(
-        "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, |tx| {
-        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?;
-    });
-
-    run_windmill_migration!("v2_jobs_rls", &db, |tx| {
-        sqlx::query!("ALTER TABLE v2_job ENABLE ROW LEVEL SECURITY")
-            .execute(db)
-            .await?;
-    });
-
-    run_windmill_migration!("v2_improve_v2_job_indices_ii", &db, |tx| {
-        sqlx::query!("create index concurrently if not exists ix_v2_job_workspace_id_created_at ON v2_job  (workspace_id, created_at DESC) where kind in ('script', 'flow', 'singlestepflow') AND parent_job IS NULL")
-        .execute(db)
-        .await?;
-
-        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS ix_job_workspace_id_created_at_new_6")
-            .execute(db)
-            .await?;
-
-        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS ix_job_workspace_id_created_at_new_7")
-            .execute(db)
-            .await?;
-    });
-
-    run_windmill_migration!("v2_improve_v2_queued_jobs_indices", &db, |tx| {
-        sqlx::query!("CREATE INDEX CONCURRENTLY IF NOT EXISTS queue_sort_v2 ON v2_job_queue (priority DESC NULLS LAST, scheduled_for, tag) WHERE running = false")
-            .execute(db)
-            .await?;
-
-        // sqlx::query!("CREATE INDEX CONCURRENTLY queue_sort_2_v2 ON v2_job_queue (tag, priority DESC NULLS LAST, scheduled_for) WHERE running = false")
-        //     .execute(db)
-        //     .await?;
-
-        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS queue_sort")
-            .execute(db)
-            .await?;
-
-        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS queue_sort_2")
-            .execute(db)
-            .await?;
-    });
-
-    run_windmill_migration!("audit_timestamps", db, |tx| {
-        sqlx::query!(
-            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_audit_timestamps ON audit (timestamp DESC)"
-        )
-        .execute(db)
-        .await?;
-    });
-
-    run_windmill_migration!("job_completed_completed_at", db, |tx| {
-        sqlx::query!(
-            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_job_completed_completed_at ON v2_job_completed (completed_at DESC)"
-        )
-        .execute(db)
-        .await?;
-    });
-
-    run_windmill_migration!("alerts_by_workspace", db, |tx| {
-        sqlx::query!(
-            "CREATE INDEX CONCURRENTLY IF NOT EXISTS alerts_by_workspace ON alerts (workspace_id);"
-        )
-        .execute(db)
-        .await?;
-    });
-
-    run_windmill_migration!("remove_redundant_log_file_index", db, |tx| {
-        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS log_file_hostname_log_ts_idx")
-            .execute(db)
-            .await?;
-    });
-
-    run_windmill_migration!("v2_job_queue_suspend", db, |tx| {
-        sqlx::query!(
-            "CREATE INDEX CONCURRENTLY IF NOT EXISTS v2_job_queue_suspend ON v2_job_queue (workspace_id, suspend) WHERE suspend > 0;"
-        )
-        .execute(db)
-        .await?;
-    });
-
-    run_windmill_migration!("audit_recent_login_activities", db, |tx| {
-        sqlx::query!(
-            "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_audit_recent_login_activities 
-ON audit (timestamp, username) 
-WHERE operation IN ('users.login', 'oauth.login', 'users.token.refresh');"
-        )
-        .execute(db)
-        .await?;
-    });
-
-    run_windmill_migration!("v2_script_lock_index", db, |tx| {
-        sqlx::query!(
-            "CREATE INDEX CONCURRENTLY IF NOT EXISTS script_not_archived ON script (workspace_id, path, created_at DESC) where archived = false;"
-        )
-        .execute(db)
-        .await?;
-    });
-
-    run_windmill_migration!("v2_job_completed_completed_at_9", db, |tx| {
-        let migration_job_name = "v2_job_completed_completed_at";
-        let mut i = 1;
-        tracing::info!("step {i} of {migration_job_name} migration");
-        sqlx::query!("create index concurrently  if not exists ix_job_workspace_id_completed_at_all ON v2_job_completed  (workspace_id,  completed_at DESC)")
-                .execute(db)
-                .await?;
-        i += 1;
-
-        sqlx::query!("CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_job_v2_job_root_by_path_2 ON v2_job (workspace_id, runnable_path)  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_job_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!(
-            "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_started_at_new_2"
-        )
-        .execute(db)
-        .await?;
-
-        i += 1;
-        tracing::info!("step {i} of {migration_job_name} migration");
-
-        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS ix_job_created_at")
-            .execute(db)
-            .await?;
-
-        i += 1;
-
-        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS ix_v2_job_root_by_path")
-            .execute(db)
-            .await?;
-
-        i += 1;
-        tracing::info!("step {i} of {migration_job_name} migration");
-    });
-
-    Ok(())
-}

backend/windmill-api/src/public_app_rate_limit.rs

@@ -0,0 +1,48 @@
+/*
+ * Author: Windmill Labs, Inc
+ * Copyright: Windmill Labs, Inc 2024
+ * This file and its contents are licensed under the AGPLv3 License.
+ * Please see the included NOTICE for copyright information and
+ * LICENSE-AGPL for a copy of the license.
+ */
+
+use chrono::Utc;
+use dashmap::DashMap;
+use hyper::StatusCode;
+use std::sync::LazyLock;
+use windmill_common::error::{Error, Result};
+
+struct RateLimitEntry {
+    count: i32,
+    minute_bucket: i64,
+}
+
+static RATE_LIMIT_COUNTER: LazyLock<DashMap<String, RateLimitEntry>> =
+    LazyLock::new(DashMap::new);
+
+pub fn check_and_increment(workspace_id: &str, limit: i32) -> Result<()> {
+    let current_minute = Utc::now().timestamp() / 60;
+
+    let mut entry = RATE_LIMIT_COUNTER
+        .entry(workspace_id.to_string())
+        .or_insert(RateLimitEntry { count: 0, minute_bucket: current_minute });
+
+    if entry.minute_bucket != current_minute {
+        entry.count = 0;
+        entry.minute_bucket = current_minute;
+    }
+
+    if entry.count >= limit {
+        return Err(Error::Generic(
+            StatusCode::TOO_MANY_REQUESTS,
+            format!(
+                "Rate limit exceeded for public app executions in workspace '{}'. \
+                 Limit: {} per minute per server.",
+                workspace_id, limit
+            ),
+        ));
+    }
+
+    entry.count += 1;
+    Ok(())
+}

backend/windmill-api/src/triggers/filter.rs

@@ -0,0 +1,144 @@
+use serde::{
+    de::{self, MapAccess, Visitor},
+    Deserialize, Deserializer,
+};
+use serde_json::Value;
+use std::fmt;
+
+#[derive(Deserialize)]
+pub struct JsonFilter {
+    pub key: String,
+    pub value: Value,
+}
+
+#[derive(Deserialize)]
+#[serde(untagged)]
+pub enum Filter {
+    JsonFilter(JsonFilter),
+}
+
+struct SupersetVisitor<'a> {
+    key: &'a str,
+    value_to_check: &'a Value,
+}
+
+impl<'de, 'a> Visitor<'de> for SupersetVisitor<'a> {
+    type Value = bool;
+
+    fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
+        formatter.write_str("a JSON object with a specific key at the top level")
+    }
+
+    fn visit_map<V>(self, mut map: V) -> std::result::Result<Self::Value, V::Error>
+    where
+        V: MapAccess<'de>,
+    {
+        let mut result = false;
+        let mut found = false;
+
+        // Must consume entire map to satisfy deserializer contract
+        while let Some(key) = map.next_key::<String>()? {
+            if !found && key == self.key {
+                let json_value: Value = map.next_value()?;
+                result = is_superset(&json_value, self.value_to_check);
+                found = true;
+            } else {
+                // Skip values we don't need (cheaper than full deserialization)
+                let _ = map.next_value::<de::IgnoredAny>()?;
+            }
+        }
+        Ok(result)
+    }
+}
+
+pub fn is_superset(json_value: &Value, value_to_check: &Value) -> bool {
+    match (json_value, value_to_check) {
+        (Value::Object(json_map), Value::Object(check_map)) => {
+            check_map.iter().all(|(k, v)| {
+                json_map
+                    .get(k)
+                    .map_or(false, |json_val| is_superset(json_val, v))
+            })
+        }
+        (Value::Array(json_array), Value::Array(check_array)) => {
+            check_array.iter().all(|check_item| {
+                json_array
+                    .iter()
+                    .any(|json_item| is_superset(json_item, check_item))
+            })
+        }
+        _ => json_value == value_to_check,
+    }
+}
+
+pub fn is_value_superset<'a, 'de, D>(
+    deserializer: D,
+    key: &'a str,
+    value_to_check: &'a Value,
+) -> std::result::Result<bool, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    deserializer.deserialize_map(SupersetVisitor { key, value_to_check })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    #[test]
+    fn test_filter_with_other_top_level_keys() {
+        let payload = r#"{"event_type": "test", "other": "data"}"#;
+        let key = "event_type";
+        let value = json!("test");
+
+        let mut deserializer = serde_json::Deserializer::from_str(payload);
+        let result = is_value_superset(&mut deserializer, key, &value).unwrap();
+        assert!(result, "Should match when key exists with correct value");
+    }
+
+    #[test]
+    fn test_filter_with_key_not_first() {
+        let payload = r#"{"other": "data", "event_type": "test"}"#;
+        let key = "event_type";
+        let value = json!("test");
+
+        let mut deserializer = serde_json::Deserializer::from_str(payload);
+        let result = is_value_superset(&mut deserializer, key, &value).unwrap();
+        assert!(result, "Should match even when key is not first");
+    }
+
+    #[test]
+    fn test_filter_with_nested_object() {
+        let payload = r#"{"data": {"status": "active", "count": 5}, "other": "value"}"#;
+        let key = "data";
+        let value = json!({"status": "active"});
+
+        let mut deserializer = serde_json::Deserializer::from_str(payload);
+        let result = is_value_superset(&mut deserializer, key, &value).unwrap();
+        assert!(result, "Should match when nested object is superset");
+    }
+
+    #[test]
+    fn test_filter_no_match() {
+        let payload = r#"{"event_type": "other", "data": "value"}"#;
+        let key = "event_type";
+        let value = json!("test");
+
+        let mut deserializer = serde_json::Deserializer::from_str(payload);
+        let result = is_value_superset(&mut deserializer, key, &value).unwrap();
+        assert!(!result, "Should not match when value differs");
+    }
+
+    #[test]
+    fn test_filter_key_not_found() {
+        let payload = r#"{"other": "data"}"#;
+        let key = "event_type";
+        let value = json!("test");
+
+        let mut deserializer = serde_json::Deserializer::from_str(payload);
+        let result = is_value_superset(&mut deserializer, key, &value).unwrap();
+        assert!(!result, "Should not match when key doesn't exist");
+    }
+}

backend/windmill-api/src/triggers/http/handler.rs

@@ -963,7 +963,13 @@ async fn route_job(
 
             let s3_object = s3_object.map_err(|err| {
                 tracing::warn!("Error retrieving file from S3: {:?}", err);
-                Error::internal_err(format!("Error retrieving file: {}", err.to_string()))
+                let mut msg = format!("Error retrieving file: {err}");
+                let mut source = std::error::Error::source(&err);
+                while let Some(e) = source {
+                    msg.push_str(&format!("\n  caused by: {e}"));
+                    source = e.source();
+                }
+                Error::internal_err(msg)
             })?;
 
             let mut response_headers = http::HeaderMap::new();

backend/windmill-api/src/triggers/mod.rs

@@ -30,6 +30,7 @@ pub mod sqs;
 #[cfg(feature = "websocket")]
 pub mod websocket;
 
+pub mod filter;
 pub mod global_handler;
 mod handler;
 mod listener;

backend/windmill-api/src/triggers/websocket/listener.rs

@@ -1,5 +1,6 @@
 use super::WebsocketTrigger;
 use crate::triggers::{
+    filter::{is_value_superset, Filter, JsonFilter},
     listener::ListeningTrigger,
     trigger_helpers::{
         trigger_runnable, trigger_runnable_and_wait_for_raw_result,
@@ -13,12 +14,9 @@ use async_trait::async_trait;
 use futures::{stream::SplitSink, SinkExt, StreamExt};
 use http::Response;
 use itertools::Itertools;
-use serde::{
-    de::{self, MapAccess, Visitor},
-    Deserialize, Deserializer,
-};
-use serde_json::{value::RawValue, Value};
-use std::{borrow::Cow, collections::HashMap, fmt, sync::Arc};
+use serde::Deserialize;
+use serde_json::value::RawValue;
+use std::{borrow::Cow, collections::HashMap, sync::Arc};
 use tokio::{net::TcpStream, sync::RwLock};
 use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream};
 use windmill_common::{
@@ -443,18 +441,6 @@ impl Listener for WebsocketTrigger {
     }
 }
 
-#[derive(Deserialize)]
-pub struct JsonFilter {
-    key: String,
-    value: Value,
-}
-
-#[derive(Deserialize)]
-#[serde(untagged)]
-pub enum Filter {
-    JsonFilter(JsonFilter),
-}
-
 pub struct ReturnMessageChannels {
     send_message_tx: tokio::sync::mpsc::Sender<String>,
     killpill_rx: tokio::sync::broadcast::Receiver<()>,
@@ -477,71 +463,6 @@ enum InitialMessage {
     RunnableResult { path: String, args: Box<RawValue>, is_flow: bool },
 }
 
-struct SupersetVisitor<'a> {
-    key: &'a str,
-    value_to_check: &'a Value,
-}
-
-impl<'de, 'a> Visitor<'de> for SupersetVisitor<'a> {
-    type Value = bool;
-
-    fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
-        formatter.write_str("a JSON object with a specific key at the top level")
-    }
-
-    fn visit_map<V>(self, mut map: V) -> std::result::Result<Self::Value, V::Error>
-    where
-        V: MapAccess<'de>,
-    {
-        while let Some(key) = map.next_key::<String>()? {
-            if key == self.key {
-                // Deserialize the value for the key and check if it's a superset
-                let json_value: Value = map.next_value()?;
-                return Ok(is_superset(&json_value, self.value_to_check));
-            } else {
-                // Skip the value if it's not the one we're interested in
-                let _ = map.next_value::<de::IgnoredAny>()?;
-            }
-        }
-        // If the key was not found, return false
-        Ok(false)
-    }
-}
-
-fn is_superset(json_value: &Value, value_to_check: &Value) -> bool {
-    match (json_value, value_to_check) {
-        (Value::Object(json_map), Value::Object(check_map)) => {
-            // Check that all keys and values in check_map exist and match in json_map
-            check_map.iter().all(|(k, v)| {
-                json_map
-                    .get(k)
-                    .map_or(false, |json_val| is_superset(json_val, v))
-            })
-        }
-        (Value::Array(json_array), Value::Array(check_array)) => {
-            // Check that all elements in check_array exist in json_array
-            check_array.iter().all(|check_item| {
-                json_array
-                    .iter()
-                    .any(|json_item| is_superset(json_item, check_item))
-            })
-        }
-        _ => json_value == value_to_check,
-    }
-}
-
-// A function to deserialize and check if the value at the given key is a superset of a passed value
-fn is_value_superset<'a, 'de, D>(
-    deserializer: D,
-    key: &'a str,
-    value_to_check: &'a Value,
-) -> std::result::Result<bool, D::Error>
-where
-    D: Deserializer<'de>,
-{
-    deserializer.deserialize_map(SupersetVisitor { key, value_to_check })
-}
-
 fn raw_value_to_args_hashmap(
     args: Option<&Box<RawValue>>,
 ) -> Result<HashMap<String, Box<RawValue>>> {

backend/windmill-api/src/workspaces.rs

@@ -176,6 +176,7 @@ pub fn workspaced_service() -> Router {
             post(acknowledge_all_critical_alerts),
         )
         .route("/critical_alerts/mute", post(mute_critical_alerts))
+        .route("/public_app_rate_limit", post(edit_public_app_rate_limit))
         .route("/operator_settings", post(update_operator_settings))
         .route(
             "/create_workspace_fork_branch",
@@ -287,6 +288,8 @@ pub struct WorkspaceSettings {
     pub error_handler: Option<serde_json::Value>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub success_handler: Option<serde_json::Value>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub public_app_execution_limit_per_minute: Option<i32>,
 }
 
 /// #[derive(sqlx::Type, Serialize, Deserialize, Debug)]
@@ -625,7 +628,8 @@ async fn get_settings(
             git_app_installations,
             auto_invite,
             error_handler,
-            success_handler
+            success_handler,
+            public_app_execution_limit_per_minute
         FROM
             workspace_settings
         WHERE
@@ -4401,6 +4405,72 @@ pub async fn mute_critical_alerts() -> Error {
     Error::NotFound("Critical Alerts require EE".to_string())
 }
 
+#[derive(Deserialize)]
+pub struct EditPublicAppRateLimitRequest {
+    pub public_app_execution_limit_per_minute: Option<i32>,
+}
+
+async fn edit_public_app_rate_limit(
+    Extension(db): Extension<DB>,
+    Path(w_id): Path<String>,
+    authed: ApiAuthed,
+    Json(req): Json<EditPublicAppRateLimitRequest>,
+) -> Result<String> {
+    require_admin(authed.is_admin, &authed.username)?;
+
+    sqlx::query!(
+        "UPDATE workspace_settings SET public_app_execution_limit_per_minute = $1 WHERE workspace_id = $2",
+        req.public_app_execution_limit_per_minute,
+        &w_id
+    )
+    .execute(&db)
+    .await?;
+
+    // Cache is invalidated via DB trigger -> notify_event -> polling in main.rs
+
+    handle_deployment_metadata(
+        &authed.email,
+        &authed.username,
+        &db,
+        &w_id,
+        DeployedObject::Settings { setting_type: "public_app_rate_limit".to_string() },
+        None,
+        false,
+        None,
+    )
+    .await?;
+
+    Ok(format!(
+        "Updated public app rate limit for workspace: {}",
+        &w_id
+    ))
+}
+
+// 5 minutes fallback TTL (in addition to event-based invalidation)
+const PUBLIC_APP_RATE_LIMIT_CACHE_TTL_SECS: i64 = 300;
+
+pub async fn get_public_app_rate_limit(db: &DB, w_id: &str) -> Result<Option<i32>> {
+    use windmill_common::workspaces::PUBLIC_APP_RATE_LIMIT_CACHE;
+
+    let now = Utc::now().timestamp();
+
+    if let Some((rate_limit, cached_at)) = PUBLIC_APP_RATE_LIMIT_CACHE.get(w_id) {
+        if now - cached_at < PUBLIC_APP_RATE_LIMIT_CACHE_TTL_SECS {
+            return Ok(rate_limit);
+        }
+    }
+
+    let result: Option<Option<i32>> = sqlx::query_scalar(
+        "SELECT public_app_execution_limit_per_minute FROM workspace_settings WHERE workspace_id = $1",
+    )
+    .bind(w_id)
+    .fetch_optional(db)
+    .await?;
+    let rate_limit = result.flatten();
+    PUBLIC_APP_RATE_LIMIT_CACHE.insert(w_id.to_string(), (rate_limit, now));
+    Ok(rate_limit)
+}
+
 #[derive(Deserialize, Serialize)]
 struct ChangeOperatorSettings {
     #[serde(default)]

backend/windmill-autoscaling/Cargo.toml

@@ -10,7 +10,7 @@ path = "./src/lib.rs"
 
 [features]
 enterprise = ["windmill-queue/enterprise", "windmill-common/enterprise"]
-private = ["dep:kube", "dep:k8s-openapi"]
+private = []
 default = []
 
 [dependencies]
@@ -22,8 +22,8 @@ tracing.workspace = true
 windmill-common = { workspace = true, default-features = false }
 windmill-queue.workspace = true
 anyhow.workspace = true
-kube = { workspace = true, optional = true }
-k8s-openapi = { workspace = true, optional = true }
+kube.workspace = true
+k8s-openapi.workspace = true
 tokio.workspace = true
 thiserror.workspace = true
 axum.workspace = true

backend/windmill-common/Cargo.toml

@@ -7,20 +7,20 @@ edition.workspace = true
 [features]
 default = []
 enterprise = []
-private = ["dep:aws-sdk-rds", "dep:systemstat", "dep:aws-config", "dep:aws-credential-types", "dep:aws-smithy-types"]
+private = ["dep:aws-sdk-rds"]
 jemalloc = ["dep:tikv-jemalloc-ctl"]
 tantivy = []
 prometheus = ["dep:prometheus"]
 benchmark = []
-parquet = ["dep:object_store", "dep:aws-sdk-sts", "dep:aws-smithy-types-convert", "dep:datafusion", "dep:aws-config", "dep:aws-credential-types", "dep:aws-smithy-types", "dep:globset"]
-aws_auth = ["dep:aws-sdk-sts", "dep:aws-config", "dep:aws-credential-types", "dep:aws-smithy-types"]
+parquet = ["dep:object_store", "dep:aws-sdk-sts", "dep:aws-smithy-types-convert", "dep:datafusion"]
+aws_auth = ["dep:aws-sdk-sts"]
 otel = ["dep:opentelemetry-semantic-conventions", "dep:opentelemetry-otlp", "dep:opentelemetry_sdk",
     "dep:tracing-opentelemetry", "dep:opentelemetry-appender-tracing", "dep:tonic", "dep:opentelemetry"]
 smtp = ["dep:mail-send"]
 scoped_cache = []
 cloud = []
 openidconnect = ["dep:openidconnect"]
-bedrock = ["dep:aws-sdk-bedrockruntime", "dep:aws-config", "dep:aws-credential-types", "dep:aws-smithy-types"]
+bedrock = ["dep:aws-sdk-bedrockruntime"]
 
 [lib]
 name = "windmill_common"
@@ -63,10 +63,10 @@ cron.workspace = true
 magic-crypt.workspace = true
 object_store = { workspace = true, optional = true }
 prometheus = { workspace = true, optional = true }
-aws-config = { workspace = true, optional = true }
+aws-config.workspace = true
 aws-sdk-sts = { workspace = true, optional = true }
-aws-credential-types = { workspace = true, optional = true }
-aws-smithy-types = { workspace = true, optional = true }
+aws-credential-types.workspace = true
+aws-smithy-types.workspace = true
 aws-sdk-bedrockruntime = { workspace = true, optional = true }
 base64.workspace = true
 bitflags.workspace = true
@@ -100,7 +100,7 @@ url.workspace = true
 urlencoding.workspace = true
 async-recursion.workspace = true
 pep440_rs.workspace = true
-systemstat = { workspace = true, optional = true }
+systemstat.workspace = true
 size.workspace = true
 
 semver.workspace = true
@@ -109,7 +109,7 @@ quick_cache.workspace = true
 pin-project-lite.workspace = true
 futures.workspace = true
 tempfile.workspace = true
-globset = { workspace = true, optional = true }
+globset.workspace = true
 
 opentelemetry-semantic-conventions = { workspace = true, optional = true }
 opentelemetry-otlp = { workspace = true, optional = true }

backend/windmill-common/src/assets.rs

@@ -1,5 +1,6 @@
 use serde::{Deserialize, Serialize};
 use sqlx::PgExecutor;
+use std::collections::BTreeMap;
 
 use crate::{error, scripts::ScriptHash};
 
@@ -37,23 +38,15 @@ pub enum AssetUsageAccessType {
     RW,
 }
 
-pub struct Asset {
-    pub path: String,
-    pub kind: AssetKind,
-}
-
-pub struct AssetUsage {
-    pub path: String,
-    pub kind: AssetUsageKind,
-    pub access_type: AssetUsageAccessType,
-}
-
-#[derive(Serialize, Deserialize, Debug, Clone, Hash, sqlx::Type)]
+#[derive(Serialize, Deserialize, Debug, Clone, Hash)]
 pub struct AssetWithAltAccessType {
     pub path: String,
     pub kind: AssetKind,
     pub access_type: Option<AssetUsageAccessType>,
     pub alt_access_type: Option<AssetUsageAccessType>,
+    /// Map of column name to access type for column-level access tracking
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub columns: Option<BTreeMap<String, AssetUsageAccessType>>,
 }
 
 pub async fn insert_static_asset_usage<'e>(
@@ -63,15 +56,22 @@ pub async fn insert_static_asset_usage<'e>(
     usage_path: &str,
     usage_kind: AssetUsageKind,
 ) -> error::Result<()> {
+    // Convert columns BTreeMap to JSONB format
+    let columns_json = asset
+        .columns
+        .as_ref()
+        .map(|cols| serde_json::to_value(cols).unwrap_or(serde_json::Value::Null));
+
     sqlx::query!(
-        r#"INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind)
-                VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT DO NOTHING"#,
+        r#"INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind, columns)
+                VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT DO NOTHING"#,
         workspace_id,
         asset.path,
         asset.kind as AssetKind,
         (asset.access_type.or(asset.alt_access_type)) as Option<AssetUsageAccessType>,
         usage_path,
-        usage_kind as AssetUsageKind
+        usage_kind as AssetUsageKind,
+        columns_json as Option<serde_json::Value>
     )
     .execute(executor)
     .await?;
@@ -125,6 +125,28 @@ pub fn merge_asset_usage_access_types(
     }
 }
 
+pub fn merge_asset_columns(
+    a: &Option<BTreeMap<String, AssetUsageAccessType>>,
+    b: &Option<BTreeMap<String, AssetUsageAccessType>>,
+) -> Option<BTreeMap<String, AssetUsageAccessType>> {
+    match (a, b) {
+        (None, None) => None,
+        (Some(cols), None) | (None, Some(cols)) => Some(cols.clone()),
+        (Some(cols_a), Some(cols_b)) => {
+            let mut merged = cols_a.clone();
+            for (col, access_b) in cols_b {
+                let access_a = merged.get(col);
+                let merged_access =
+                    merge_asset_usage_access_types(access_a.cloned(), Some(*access_b));
+                if let Some(access) = merged_access {
+                    merged.insert(col.clone(), access);
+                }
+            }
+            Some(merged)
+        }
+    }
+}
+
 impl From<windmill_parser::asset_parser::AssetKind> for AssetKind {
     fn from(parser_kind: windmill_parser::asset_parser::AssetKind) -> Self {
         match parser_kind {

backend/windmill-common/src/auth.rs

@@ -250,11 +250,31 @@ pub async fn fetch_authed_from_permissioned_as(
     w_id: &str,
     db: &DB,
 ) -> Result<Authed> {
-    let super_admin =
-        permissioned_as == SUPERADMIN_SYNC_EMAIL || is_super_admin_email(db, &email).await?;
+    let mut conn = db
+        .acquire()
+        .await
+        .map_err(|e| Error::internal_err(format!("acquiring connection: {e:#}")))?;
+    fetch_authed_from_permissioned_as_conn(&permissioned_as, &email, w_id, &mut conn).await
+}
+
+pub async fn fetch_authed_from_permissioned_as_conn(
+    permissioned_as: &str,
+    email: &str,
+    w_id: &str,
+    conn: &mut sqlx::PgConnection,
+) -> Result<Authed> {
+    let is_super_admin = permissioned_as == SUPERADMIN_SYNC_EMAIL
+        || email == SUPERADMIN_SECRET_EMAIL
+        || email == SUPERADMIN_NOTIFICATION_EMAIL
+        || sqlx::query_scalar!("SELECT super_admin FROM password WHERE email = $1", email)
+            .fetch_optional(&mut *conn)
+            .await
+            .map_err(|e| Error::internal_err(format!("fetching super admin: {e:#}")))?
+            .unwrap_or(false);
+
     if let Some((prefix, name)) = permissioned_as.split_once('/') {
         if prefix == "u" {
-            let (is_admin, is_operator) = if super_admin {
+            let (is_admin, is_operator) = if is_super_admin {
                 (true, false)
             } else {
                 let r = sqlx::query!(
@@ -263,7 +283,7 @@ pub async fn fetch_authed_from_permissioned_as(
                     name,
                     &w_id
                 )
-                .fetch_optional(db)
+                .fetch_optional(&mut *conn)
                 .await?;
                 if let Some(r) = r {
                     (r.is_admin, r.operator)
@@ -274,12 +294,12 @@ pub async fn fetch_authed_from_permissioned_as(
                 }
             };
 
-            let groups = get_groups_for_user(w_id, &name, &email, db).await?;
+            let groups = get_groups_for_user(w_id, &name, email, &mut *conn).await?;
 
-            let folders = get_folders_for_user(w_id, &name, &groups, db).await?;
+            let folders = get_folders_for_user(w_id, &name, &groups, &mut *conn).await?;
 
             Ok(Authed {
-                email,
+                email: email.to_string(),
                 username: name.to_string(),
                 is_admin,
                 is_operator,
@@ -290,9 +310,9 @@ pub async fn fetch_authed_from_permissioned_as(
             })
         } else {
             let groups = vec![name.to_string()];
-            let folders = get_folders_for_user(&w_id, "", &groups, db).await?;
+            let folders = get_folders_for_user(&w_id, "", &groups, &mut *conn).await?;
             Ok(Authed {
-                email,
+                email: email.to_string(),
                 username: format!("group-{name}"),
                 is_admin: false,
                 groups,
@@ -303,26 +323,24 @@ pub async fn fetch_authed_from_permissioned_as(
             })
         }
     } else {
-        let groups = vec![];
-        let folders = vec![];
         Ok(Authed {
-            email,
-            username: permissioned_as,
-            is_admin: super_admin,
+            email: email.to_string(),
+            username: permissioned_as.to_string(),
+            is_admin: is_super_admin,
             is_operator: true,
-            groups,
-            folders,
+            groups: vec![],
+            folders: vec![],
             scopes: None,
             token_prefix: None,
         })
     }
 }
 
-pub async fn get_folders_for_user(
+pub async fn get_folders_for_user<'e, E: sqlx::PgExecutor<'e>>(
     w_id: &str,
     username: &str,
     groups: &[String],
-    db: &DB,
+    db: E,
 ) -> Result<Vec<(String, bool, bool)>> {
     let mut perms = groups
         .into_iter()
@@ -344,11 +362,11 @@ pub async fn get_folders_for_user(
     Ok(folders)
 }
 
-pub async fn get_groups_for_user(
+pub async fn get_groups_for_user<'e, E: sqlx::PgExecutor<'e>>(
     w_id: &str,
     username: &str,
     email: &str,
-    db: &DB,
+    db: E,
 ) -> Result<Vec<String>> {
     let groups = sqlx::query_scalar!(
         "SELECT group_ FROM usr_to_group where usr = $1 AND workspace_id = $2 UNION ALL SELECT igroup FROM email_to_igroup WHERE email = $3",

backend/windmill-common/src/runtime_assets.rs

@@ -1,13 +1,19 @@
-use std::{collections::HashMap, sync::OnceLock};
+use std::{
+    collections::{BTreeMap, HashMap},
+    sync::OnceLock,
+};
 
 use itertools::Itertools;
 use serde_json::value::RawValue;
-use sqlx::{Pool, Postgres, QueryBuilder};
+use sqlx::{types::Json, Pool, Postgres, QueryBuilder};
 use tokio::sync::mpsc;
 use windmill_parser::asset_parser::parse_asset_syntax;
 
 use crate::{
-    assets::{merge_asset_usage_access_types, AssetKind, AssetUsageAccessType, AssetUsageKind},
+    assets::{
+        merge_asset_columns, merge_asset_usage_access_types, AssetKind, AssetUsageAccessType,
+        AssetUsageKind,
+    },
     error,
 };
 
@@ -59,6 +65,7 @@ pub struct InsertRuntimeAssetParams {
     pub job_id: uuid::Uuid,
     pub access_type: Option<AssetUsageAccessType>,
     pub created_at: Option<chrono::DateTime<chrono::Utc>>,
+    pub columns: Option<BTreeMap<String, AssetUsageAccessType>>,
 }
 
 async fn insert_runtime_assets(
@@ -66,13 +73,14 @@ async fn insert_runtime_assets(
     assets: &[InsertRuntimeAssetParams],
 ) -> error::Result<()> {
     for chunk in assets.chunks(1000) {
-        let mut query_builder = QueryBuilder::new("INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind, created_at) ");
+        let mut query_builder = QueryBuilder::new("INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, columns, usage_kind, created_at) ");
         query_builder.push_values(chunk, |mut b, asset| {
             b.push_bind(&asset.workspace_id)
                 .push_bind(&asset.asset_path)
                 .push_bind(&asset.asset_kind)
                 .push_bind(&asset.access_type)
                 .push_bind(asset.job_id.to_string())
+                .push_bind(Json(&asset.columns))
                 .push_bind(&AssetUsageKind::Job)
                 .push_bind(&asset.created_at);
         });
@@ -104,6 +112,7 @@ async fn prune_runtime_assets(
             // Same job used the same asset multiple times
             last_same_job.access_type =
                 merge_asset_usage_access_types(last_same_job.access_type, asset.access_type);
+            last_same_job.columns = merge_asset_columns(&last_same_job.columns, &asset.columns);
         } else if v.len() < max_n {
             v.push(asset);
         }

backend/windmill-common/src/s3_helpers.rs

@@ -793,8 +793,8 @@ pub async fn build_s3_client(s3_resource_ref: &S3Resource) -> error::Result<Arc<
     let store = store_builder.build().map_err(|err| {
         tracing::error!("Error building object store client: {:?}", err);
         error::Error::internal_err(format!(
-            "Error building object store client: {}",
-            err.to_string()
+            "Error building object store client: {:?}",
+            err
         ))
     })?;
 
@@ -860,8 +860,8 @@ fn build_azure_blob_client(
     let store = store_builder.build().map_err(|err| {
         tracing::error!("Error building object store client: {:?}", err);
         error::Error::internal_err(format!(
-            "Error building object store client: {}",
-            err.to_string()
+            "Error building object store client: {:?}",
+            err
         ))
     })?;
 
@@ -900,8 +900,8 @@ async fn build_gcs_client(gcs_resource_ref: &GcsResource) -> error::Result<Arc<d
         .map_err(|err| {
             tracing::error!("Error building GCS object store client: {:?}", err);
             error::Error::internal_err(format!(
-                "Error building GCS object store client: {}",
-                err.to_string()
+                "Error building GCS object store client: {:?}",
+                err
             ))
         })?;
 
@@ -1257,21 +1257,21 @@ pub fn lfs_to_object_store_resource(
     match lfs {
         LargeFileStorage::S3Storage(_) | LargeFileStorage::S3AwsOidc(_) => {
             let s3_resource: S3Resource = serde_json::from_value(resource_value).map_err(|e| {
-                error::Error::internal_err(format!("Error parsing S3 resource: {}", e))
+                error::Error::internal_err(format!("Error parsing S3 resource: {e:?}"))
             })?;
             Ok(ObjectStoreResource::S3(s3_resource))
         }
         LargeFileStorage::AzureBlobStorage(_) | LargeFileStorage::AzureWorkloadIdentity(_) => {
             let azure_blob_resource: AzureBlobResource = serde_json::from_value(resource_value)
                 .map_err(|e| {
-                    error::Error::internal_err(format!("Error parsing Azure Blob resource: {}", e))
+                    error::Error::internal_err(format!("Error parsing Azure Blob resource: {e:?}"))
                 })?;
             Ok(ObjectStoreResource::Azure(azure_blob_resource))
         }
         LargeFileStorage::GoogleCloudStorage(_) => {
             let gcs_resource: GcsResource =
                 serde_json::from_value(resource_value).map_err(|e| {
-                    error::Error::internal_err(format!("Error parsing GCS resource: {}", e))
+                    error::Error::internal_err(format!("Error parsing GCS resource: {e:?}"))
                 })?;
             Ok(ObjectStoreResource::Gcs(gcs_resource))
         }

backend/windmill-common/src/workspaces.rs

@@ -119,6 +119,8 @@ pub struct TeamPlanStatus {
 
 lazy_static::lazy_static! {
     pub static ref TEAM_PLAN_CACHE: Cache<String, TeamPlanStatus> = Cache::new(5000);
+    // Value: (rate_limit, cached_at_timestamp)
+    pub static ref PUBLIC_APP_RATE_LIMIT_CACHE: Cache<String, (Option<i32>, i64)> = Cache::new(1000);
 }
 
 #[cfg(feature = "cloud")]

backend/windmill-indexer/Cargo.toml

@@ -10,9 +10,9 @@ path = "src/lib.rs"
 
 [features]
 default = []
-parquet = ["dep:object_store"]
-private = []
-enterprise = []
+parquet = ["dep:object_store", "windmill-common/parquet"]
+private = ["windmill-common/private"]
+enterprise = ["windmill-common/enterprise"]
 
 [dependencies]
 windmill-common.workspace = true

backend/windmill-queue/src/jobs.rs

@@ -1782,10 +1782,19 @@ pub async fn handle_maybe_scheduled_job<'c>(
     );
 
     if schedule.enabled && script_path == schedule.script_path {
+        let schedule_authed = windmill_common::auth::fetch_authed_from_permissioned_as(
+            windmill_common::users::username_to_permissioned_as(&schedule.edited_by),
+            schedule.email.clone(),
+            w_id,
+            db,
+        )
+        .await
+        .ok();
+
         let push_next_job_future = (|| {
             tokio::time::timeout(std::time::Duration::from_secs(5), async {
                 let mut tx = db.begin().await?;
-                tx = push_scheduled_job(db, tx, &schedule, None, Some(job.scheduled_for)).await?;
+                tx = push_scheduled_job(db, tx, &schedule, schedule_authed.as_ref(), Some(job.scheduled_for)).await?;
                 tx.commit().await?;
                 Ok::<(), Error>(())
             })