diff --git a/.changeset/chat-agent-preload-oom-recovery.md b/.changeset/chat-agent-preload-oom-recovery.md new file mode 100644 index 00000000000..7885a3530ab --- /dev/null +++ b/.changeset/chat-agent-preload-oom-recovery.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/sdk": patch +--- + +Fix a preloaded `chat.agent` run dropping an in-flight message when it retries after an out-of-memory error. The message being processed when the run hit the OOM is now recovered and re-run on the retry, instead of being skipped while the run waited for a new message. diff --git a/.changeset/chat-handover-cursor-off-by-one.md b/.changeset/chat-handover-cursor-off-by-one.md new file mode 100644 index 00000000000..6829d113d7b --- /dev/null +++ b/.changeset/chat-handover-cursor-off-by-one.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/core": patch +--- + +Fix a chunk occasionally dropped when a chat.agent run takes over from the warm first turn. The realtime stream writer now reports the inclusive last-written position as the resume cursor, so the agent's first record after the handover is no longer skipped. diff --git a/.changeset/chat-session-caught-up-resume.md b/.changeset/chat-session-caught-up-resume.md new file mode 100644 index 00000000000..b405bf60472 --- /dev/null +++ b/.changeset/chat-session-caught-up-resume.md @@ -0,0 +1,7 @@ +--- +"@trigger.dev/core": patch +"@trigger.dev/sdk": patch +"trigger.dev": patch +--- + +`AgentChat.reconnect()` now settles promptly when reconnecting to an idle chat instead of holding the connection open for the full long-poll window. Also upgrades the S2 streamstore client to 0.25 and moves realtime streams to S2's current hosts. diff --git a/.changeset/friendly-keys-batch-tokens.md b/.changeset/friendly-keys-batch-tokens.md new file mode 100644 index 00000000000..90506d48d28 --- /dev/null +++ b/.changeset/friendly-keys-batch-tokens.md @@ -0,0 +1,6 @@ +--- +"@trigger.dev/core": patch +"@trigger.dev/sdk": patch +--- + +Allow task-scoped environment API keys to run batch operations for their permitted tasks. The SDK declares the batch's task set before creation, and `@trigger.dev/core/v3/apiKeys` now exports the additional-key format helper. diff --git a/.changeset/mcp-trigger-task-region.md b/.changeset/mcp-trigger-task-region.md new file mode 100644 index 00000000000..56e362eb1fb --- /dev/null +++ b/.changeset/mcp-trigger-task-region.md @@ -0,0 +1,5 @@ +--- +"trigger.dev": patch +--- + +You can now choose the region a run executes in when triggering a task through the MCP server. diff --git a/.changeset/retry-run-start-on-connection-error.md b/.changeset/retry-run-start-on-connection-error.md new file mode 100644 index 00000000000..3624ee1b7c5 --- /dev/null +++ b/.changeset/retry-run-start-on-connection-error.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/core": patch +--- + +Transient connection errors when a run starts are now retried for longer, so a brief connectivity blip no longer sends the run back through the queue and delays its first attempt. diff --git a/.github/VOUCHED.td b/.github/VOUCHED.td index 7237a5369ae..a2ebac99dd8 100644 --- a/.github/VOUCHED.td +++ b/.github/VOUCHED.td @@ -24,4 +24,5 @@ jrossi ThullyoCunha ConProgramming saasjesus -brentshulman-silkline \ No newline at end of file +brentshulman-silkline +Leafgard \ No newline at end of file diff --git a/.github/workflows/check-review-md.yml b/.github/workflows/check-review-md.yml index 06f0d2d82c4..79b9d7eacd5 100644 --- a/.github/workflows/check-review-md.yml +++ b/.github/workflows/check-review-md.yml @@ -38,7 +38,7 @@ jobs: with: anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }} use_sticky_comment: true - allowed_bots: "devin-ai-integration[bot]" + allowed_bots: "devin-ai-integration[bot],claude[bot]" claude_args: | --max-turns 30 diff --git a/.github/workflows/claude-md-audit.yml b/.github/workflows/claude-md-audit.yml index aa62444c9df..d1b1bbdb45a 100644 --- a/.github/workflows/claude-md-audit.yml +++ b/.github/workflows/claude-md-audit.yml @@ -40,7 +40,7 @@ jobs: with: anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }} use_sticky_comment: true - allowed_bots: "devin-ai-integration[bot]" + allowed_bots: "devin-ai-integration[bot],claude[bot]" claude_args: | --max-turns 25 diff --git a/.github/workflows/dependabot-critical-alerts.yml b/.github/workflows/dependabot-critical-alerts.yml deleted file mode 100644 index 62d68c30023..00000000000 --- a/.github/workflows/dependabot-critical-alerts.yml +++ /dev/null @@ -1,87 +0,0 @@ -name: Dependabot Critical Alerts - -on: - schedule: - - cron: "0 8 * * *" # Daily 08:00 UTC - workflow_dispatch: - inputs: - severity: - description: "Severity to alert on" - type: choice - options: - - critical - - high - - medium - - low - default: critical - -concurrency: - group: ${{ github.workflow }} - cancel-in-progress: false - -permissions: - contents: read - -jobs: - alert: - name: Post critical alerts - # Set the ENABLE_DEPENDABOT_ALERTS repository variable to 'false' to turn off - # the Dependabot alert/summary notifiers — e.g. forks/mirrors that lack the - # DEPENDABOT_ALERTS_TOKEN / SLACK_BOT_TOKEN secrets. Defaults to enabled. - if: ${{ vars.ENABLE_DEPENDABOT_ALERTS != 'false' }} - runs-on: warp-ubuntu-latest-x64-2x - environment: dependabot-summary - env: - SEVERITY: ${{ inputs.severity || 'critical' }} - steps: - - name: Fetch alerts - id: alerts - env: - GH_TOKEN: ${{ secrets.DEPENDABOT_ALERTS_TOKEN }} - REPO: ${{ github.repository }} - run: | - set -euo pipefail - gh api -X GET "/repos/$REPO/dependabot/alerts" \ - -F state=open -F severity="$SEVERITY" --paginate > pages.json - jq -s 'add' pages.json > alerts.json - TOTAL=$(jq 'length' alerts.json) - echo "total=$TOTAL" >> "$GITHUB_OUTPUT" - if [ "$TOTAL" = "0" ]; then - exit 0 - fi - LIST=$(jq -r ' - map("• <\(.html_url)|#\(.number)> *\(.dependency.package.name)* - \(.security_advisory.summary)") - | join("\n") - ' alerts.json) - { - echo "list<> "$GITHUB_OUTPUT" - - - name: Build Slack payload - if: steps.alerts.outputs.total != '0' - env: - REPO: ${{ github.repository }} - CHANNEL: ${{ vars.SLACK_CHANNEL_ID }} - TOTAL: ${{ steps.alerts.outputs.total }} - LIST: ${{ steps.alerts.outputs.list }} - run: | - jq -n \ - --arg channel "$CHANNEL" \ - --arg repo "$REPO" \ - --arg total "$TOTAL" \ - --arg list "$LIST" \ - --arg severity "$SEVERITY" \ - '{ - channel: $channel, - text: ":bufo-alarma: `\($repo)` - *\($total) open \($severity) alert(s)*\n\($list)\n\n" - }' > payload.json - - - name: Post Slack alert - if: steps.alerts.outputs.total != '0' - uses: slackapi/slack-github-action@45a88b9581bfab2566dc881e2cd66d334e621e2c # v3.0.3 - with: - method: chat.postMessage - token: ${{ secrets.SLACK_BOT_TOKEN }} - payload-file-path: payload.json diff --git a/.github/workflows/dependabot-weekly-summary.yml b/.github/workflows/dependabot-weekly-summary.yml deleted file mode 100644 index fc9eaebcb94..00000000000 --- a/.github/workflows/dependabot-weekly-summary.yml +++ /dev/null @@ -1,210 +0,0 @@ -name: Dependabot Weekly Summary - -on: - schedule: - - cron: "0 8 * * 1" # Mon 08:00 UTC - workflow_dispatch: - -# Single-purpose monitoring workflow; serialise on workflow name only - we never -# want two concurrent summary runs racing to post the same digest. -concurrency: - group: ${{ github.workflow }} - cancel-in-progress: false - -permissions: - contents: read # gh CLI baseline - pull-requests: read # gh pr list (open dependabot PRs) - actions: read # gh run list / view (parse latest dependabot run logs) - -jobs: - summary: - name: Post weekly Dependabot summary - # Set the ENABLE_DEPENDABOT_ALERTS repository variable to 'false' to turn off - # the Dependabot alert/summary notifiers — e.g. forks/mirrors that lack the - # DEPENDABOT_ALERTS_TOKEN / SLACK_BOT_TOKEN secrets. Defaults to enabled. - if: ${{ vars.ENABLE_DEPENDABOT_ALERTS != 'false' }} - runs-on: warp-ubuntu-latest-x64-2x - environment: dependabot-summary - env: - # Severities surface in the actions list when their remaining TTR drops - # below this many days. Override via repo/env var ACTION_THRESHOLD_DAYS. - THRESHOLD_DAYS: ${{ vars.ACTION_THRESHOLD_DAYS || '7' }} - steps: - - name: Fetch alerts and compute summaries - id: alerts - env: - GH_TOKEN: ${{ secrets.DEPENDABOT_ALERTS_TOKEN }} - REPO: ${{ github.repository }} - run: | - if ! gh api -X GET "/repos/$REPO/dependabot/alerts" --paginate > pages.json 2> err.txt; then - echo "total=?" >> "$GITHUB_OUTPUT" - ERR=$(head -c 200 err.txt | tr '\n' ' ') - echo "by_severity=:x: _failed to fetch alerts: ${ERR}_" >> "$GITHUB_OUTPUT" - echo "actions=:x: _alerts unavailable_" >> "$GITHUB_OUTPUT" - exit 0 - fi - jq -s '[.[][] | select(.state == "open")]' pages.json > open.json - - TOTAL=$(jq 'length' open.json) - echo "total=$TOTAL" >> "$GITHUB_OUTPUT" - - if [ "$TOTAL" = "0" ]; then - echo "by_severity=:white_check_mark: No open alerts." >> "$GITHUB_OUTPUT" - echo "actions=_None_" >> "$GITHUB_OUTPUT" - exit 0 - fi - - # Severity breakdown - real newlines so jq --arg in the payload - # builder encodes them as proper \n in JSON (Slack renders as breaks). - BY_SEV=$(jq -r ' - group_by(.security_advisory.severity) - | map({sev: .[0].security_advisory.severity, - count: length, - weight: ({"critical":0,"high":1,"medium":2,"low":3}[.[0].security_advisory.severity])}) - | sort_by(.weight) - | map("• *\(.count)* \(.sev)") - | join("\n") - ' open.json) - { - echo "by_severity<> "$GITHUB_OUTPUT" - - # Actions: alerts within THRESHOLD_DAYS of their TTR (P0=7d, P1=30d, P2=90d, P3=no deadline) - # Grouped by (package, severity); shows earliest deadline per group. - ACTIONS=$(jq -r --argjson threshold "$THRESHOLD_DAYS" ' - [.[] - | (.security_advisory.severity) as $sev - | ({"critical":7,"high":30,"medium":90,"low":null}[$sev]) as $ttr - | select($ttr != null) - | ((now - (.created_at | fromdateiso8601)) / 86400 | floor) as $age - | {pkg: .dependency.package.name, sev: $sev, remaining: ($ttr - $age)} - ] - | group_by([.pkg, .sev]) - | map({pkg: .[0].pkg, sev: .[0].sev, count: length, min_remaining: ([.[].remaining] | min)}) - | map(select(.min_remaining < $threshold)) - | sort_by(.min_remaining) - | if length == 0 then "_None_" - else (map( - "• *\(.pkg)* (\(.sev))" + - (if .count > 1 then " ×\(.count)" else "" end) + " - " + - (if .min_remaining < 0 then "*OVERDUE* by \(-.min_remaining)d" - else "\(.min_remaining)d remaining" end) - ) | join("\n")) - end - ' open.json) - { - echo "actions<> "$GITHUB_OUTPUT" - - - name: Fetch open dependabot PRs - id: prs - env: - GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} - REPO: ${{ github.repository }} - REPO_URL: https://github.com/${{ github.repository }} - run: | - if ! PR_JSON=$(gh pr list --repo "$REPO" --state open --author "app/dependabot" --json number,title 2> err.txt); then - ERR=$(head -c 200 err.txt | tr '\n' ' ') - echo "list=:x: _failed to fetch PRs: ${ERR}_" >> "$GITHUB_OUTPUT" - exit 0 - fi - LIST=$(echo "$PR_JSON" | jq -r --arg url "$REPO_URL" ' - if length == 0 then "_None_" - else (map("• <\($url)/pull/\(.number)|#\(.number)> \(.title)") | join("\n")) - end - ') - { - echo "list<> "$GITHUB_OUTPUT" - - - name: Find latest npm dependabot run - id: latest - env: - GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} - REPO: ${{ github.repository }} - run: | - # Repos without a dependabot.yml have no "Dependabot Updates" workflow; - # treat the lookup failure as "no recent run found" rather than failing. - if ! RUN_ID=$(gh run list --repo "$REPO" --workflow "Dependabot Updates" --status success --limit 30 --json databaseId,name --jq 'first(.[] | select(.name | startswith("npm_and_yarn")) | .databaseId) // empty' 2>/dev/null); then - RUN_ID="" - fi - echo "run_id=$RUN_ID" >> "$GITHUB_OUTPUT" - - - name: Extract stuck deps (only if actions pending) - id: stuck - env: - GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} - REPO: ${{ github.repository }} - RUN_ID: ${{ steps.latest.outputs.run_id }} - ACTIONS: ${{ steps.alerts.outputs.actions }} - run: | - # Skip the stuck section entirely when nothing in the actions list - # - keeps the digest tidy when there's nothing to actually act on. - if [ "$ACTIONS" = "_None_" ]; then - echo "section=" >> "$GITHUB_OUTPUT" - exit 0 - fi - HEADER=$'\n\n*Couldn\'t auto-fix (need manual `pnpm.overrides`):*\n' - if [ -z "$RUN_ID" ]; then - { - echo "section<> "$GITHUB_OUTPUT" - exit 0 - fi - gh run view "$RUN_ID" --repo "$REPO" --log > log.txt 2>&1 || true - STUCK=$(grep -oE "No update possible for [^[:space:]]+ [0-9][^[:space:]]*" log.txt | sed 's/No update possible for //' | sort -u || true) - if [ -z "$STUCK" ]; then - { - echo "section<> "$GITHUB_OUTPUT" - exit 0 - fi - LIST=$(echo "$STUCK" | awk 'NR>1{printf "\n"} {printf "• *%s* %s", $1, $2}') - { - echo "section<> "$GITHUB_OUTPUT" - - - name: Build Slack payload - env: - REPO: ${{ github.repository }} - CHANNEL: ${{ vars.SLACK_CHANNEL_ID }} - TOTAL: ${{ steps.alerts.outputs.total }} - BY_SEVERITY: ${{ steps.alerts.outputs.by_severity }} - PRS_LIST: ${{ steps.prs.outputs.list }} - ACTIONS: ${{ steps.alerts.outputs.actions }} - STUCK: ${{ steps.stuck.outputs.section }} - run: | - # Build payload via jq so PR titles or error strings containing - # quotes/backslashes/newlines can't break the JSON. - jq -n \ - --arg channel "$CHANNEL" \ - --arg repo "$REPO" \ - --arg total "$TOTAL" \ - --arg by_severity "$BY_SEVERITY" \ - --arg prs_list "$PRS_LIST" \ - --arg actions "$ACTIONS" \ - --arg stuck "$STUCK" \ - --arg threshold "$THRESHOLD_DAYS" \ - '{ - channel: $channel, - text: ":calendar: *Weekly Dependabot summary* - `\($repo)`\n\n*Open alerts (\($total)):*\n\($by_severity)\n\n*Open Dependabot PRs:*\n\($prs_list)\n\n*Actions needed (<\($threshold)d remaining):*\n\($actions)\($stuck)\n\n" - }' > payload.json - - - name: Post Slack summary - uses: slackapi/slack-github-action@45a88b9581bfab2566dc881e2cd66d334e621e2c # v3.0.3 - with: - method: chat.postMessage - token: ${{ secrets.SLACK_BOT_TOKEN }} - payload-file-path: payload.json diff --git a/.github/workflows/e2e-webapp.yml b/.github/workflows/e2e-webapp.yml index 1931cfdd336..1c1525f31bc 100644 --- a/.github/workflows/e2e-webapp.yml +++ b/.github/workflows/e2e-webapp.yml @@ -15,7 +15,7 @@ jobs: e2eTests: name: "🧪 E2E Tests: Webapp" runs-on: warp-ubuntu-latest-x64-16x - timeout-minutes: 20 + timeout-minutes: 30 env: DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }} steps: @@ -80,6 +80,8 @@ jobs: docker pull postgres:14 docker pull redis:7.2 docker pull testcontainers/ryuk:0.11.0 + docker pull ghcr.io/s2-streamstore/s2:0.40.0@sha256:b26249e2ede0949755f5af8028185dc2bcfc3aa2db21eb9610543d144eb6ee9d + docker pull minio/minio:latest echo "Image pre-pull complete" - name: 📥 Download deps @@ -91,6 +93,9 @@ jobs: - name: 🏗️ Build Webapp run: pnpm run build --filter webapp + - name: 🎭 Install Playwright Chromium + run: cd apps/webapp && pnpm exec playwright install chromium + - name: 🧪 Run Webapp E2E Tests run: cd apps/webapp && pnpm exec vitest run --config vitest.e2e.config.ts --reporter=default env: diff --git a/.github/workflows/typecheck.yml b/.github/workflows/typecheck.yml index d8521f3936a..950d5a5e981 100644 --- a/.github/workflows/typecheck.yml +++ b/.github/workflows/typecheck.yml @@ -23,11 +23,19 @@ jobs: version: 10.33.2 - name: ⎔ Setup node - uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v6.4.0 + uses: WarpBuilds/setup-node@bc639b444d583175926b588962199c247d23e8d3 # v6 with: node-version: 24.18.0 cache: "pnpm" + - name: Restore Turbo cache + uses: WarpBuilds/cache@40f3443ae7b70e568d6e2070ea897f3df94d7553 # v1 + with: + path: node_modules/.cache/turbo + key: turbo-typecheck-${{ runner.os }}-${{ hashFiles('pnpm-lock.yaml') }}-${{ github.sha }} + restore-keys: | + turbo-typecheck-${{ runner.os }}-${{ hashFiles('pnpm-lock.yaml') }}- + - name: 📥 Download deps run: pnpm install --frozen-lockfile diff --git a/.gitignore b/.gitignore index 1ee5643d57d..7d9dc169042 100644 --- a/.gitignore +++ b/.gitignore @@ -79,3 +79,6 @@ ailogger-output.log # local git worktree checkouts (not source) — keeps oxfmt/oxlint from descending into them .worktrees/ + +# local planning/design docs, not committed +**/docs/superpowers/ diff --git a/.server-changes/backpressure-hold-last-verdict.md b/.server-changes/backpressure-hold-last-verdict.md new file mode 100644 index 00000000000..18f3ca7fc50 --- /dev/null +++ b/.server-changes/backpressure-hold-last-verdict.md @@ -0,0 +1,6 @@ +--- +area: supervisor +type: fix +--- + +When the capacity signal drops out, the last decision is held for a grace period rather than released. diff --git a/.server-changes/batch-item-streaming-rate-limit.md b/.server-changes/batch-item-streaming-rate-limit.md new file mode 100644 index 00000000000..f817195e9b1 --- /dev/null +++ b/.server-changes/batch-item-streaming-rate-limit.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Batch triggers no longer fail to start their runs when an environment is under heavy API load. If a batch still can't finish being created, `batchTriggerAndWait` now fails with an error instead of leaving the parent run waiting forever, and the batches page says so rather than reporting that it resumed. diff --git a/.server-changes/collapse-trigger-queued-snapshot.md b/.server-changes/collapse-trigger-queued-snapshot.md new file mode 100644 index 00000000000..dffc9671e6b --- /dev/null +++ b/.server-changes/collapse-trigger-queued-snapshot.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: improvement +--- + +Triggering a task now does one fewer database write, so runs reach the queue slightly faster. diff --git a/.server-changes/default-billing-alerts.md b/.server-changes/default-billing-alerts.md deleted file mode 100644 index e828c506050..00000000000 --- a/.server-changes/default-billing-alerts.md +++ /dev/null @@ -1,6 +0,0 @@ ---- -area: webapp -type: improvement ---- - -Organizations without billing alerts now get default spend alert thresholds, so you're notified before usage grows unexpectedly. The billing limit page no longer pre-selects an option before you've set a limit and prompts you to configure one. Alert previews now update immediately after you change your billing limit. diff --git a/.server-changes/fix-runs-list-batch-drop-on-unparseable-json.md b/.server-changes/fix-runs-list-batch-drop-on-unparseable-json.md new file mode 100644 index 00000000000..fe56e609510 --- /dev/null +++ b/.server-changes/fix-runs-list-batch-drop-on-unparseable-json.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Fixed a rare case where a single run or span carrying data that could not be ingested would make other runs or trace events in the same batch go missing from the runs list, traces, and logs. Now the rest of the batch is always kept: an affected run still appears with its status (only its un-ingestable output is dropped), and an affected trace event or payload is skipped instead of taking down everything around it. diff --git a/.server-changes/fix-sidebar-feedback.md b/.server-changes/fix-sidebar-feedback.md new file mode 100644 index 00000000000..066332c0314 --- /dev/null +++ b/.server-changes/fix-sidebar-feedback.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Fix the sidebar "Help & Feedback → Contact us" form sometimes not sending your message. It now sends reliably every time. diff --git a/.server-changes/impersonation-consent-and-view-as-user.md b/.server-changes/impersonation-consent-and-view-as-user.md new file mode 100644 index 00000000000..232e97e8b90 --- /dev/null +++ b/.server-changes/impersonation-consent-and-view-as-user.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: improvement +--- + +Admins opening an impersonation link from outside the dashboard now get a confirmation page naming the organization and destination instead of being bounced back, and while impersonating they can switch to "View as user" to see the dashboard exactly as that user sees it, with the admin-only UI and the impersonation highlight both hidden. Stopping impersonation is still one click away in the account menu. diff --git a/.server-changes/light-theme.md b/.server-changes/light-theme.md new file mode 100644 index 00000000000..39d4f5892d2 --- /dev/null +++ b/.server-changes/light-theme.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: feature +--- + +The dashboard gets a refreshed Dark theme as its new default look. On the account page you can switch to Light, System preference (follows your OS), or the original Classic look, and fine-tune the new themes with a contrast slider. diff --git a/.server-changes/multi-task-scope-enforcement.md b/.server-changes/multi-task-scope-enforcement.md new file mode 100644 index 00000000000..27e31cf5ec4 --- /dev/null +++ b/.server-changes/multi-task-scope-enforcement.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Requests spanning multiple tasks now require permission for every requested task instead of accepting permission for only one task. diff --git a/.server-changes/plain-graphql-migration.md b/.server-changes/plain-graphql-migration.md new file mode 100644 index 00000000000..81468a91444 --- /dev/null +++ b/.server-changes/plain-graphql-migration.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: improvement +--- + +In-app feedback and add-on/quota requests are now recorded with more account context for the support team. diff --git a/.server-changes/pod-count-exact-list.md b/.server-changes/pod-count-exact-list.md new file mode 100644 index 00000000000..d739f8d1209 --- /dev/null +++ b/.server-changes/pod-count-exact-list.md @@ -0,0 +1,6 @@ +--- +area: supervisor +type: improvement +--- + +Self-hosted Kubernetes deployments now measure running-task count more accurately when deciding whether to pause pulling new work, so the safeguard engages closer to its configured thresholds. diff --git a/.server-changes/queue-metrics-remember-period.md b/.server-changes/queue-metrics-remember-period.md new file mode 100644 index 00000000000..339d3139c8d --- /dev/null +++ b/.server-changes/queue-metrics-remember-period.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: improvement +--- + +The Queues pages now open on the last hour instead of the last day, and remember the time period you picked when you navigate between queues or reload the page. diff --git a/.server-changes/queues-page-environment-wide-charts.md b/.server-changes/queues-page-environment-wide-charts.md new file mode 100644 index 00000000000..bb3d8ea5da8 --- /dev/null +++ b/.server-changes/queues-page-environment-wide-charts.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +The four charts at the top of the Queues page now always cover the whole environment, so paging through or re-sorting your queues no longer changes them. The scheduling delay chart also leaves a gap where no runs started, instead of dropping to zero. diff --git a/.server-changes/realtime-emission-fanout-metrics.md b/.server-changes/realtime-emission-fanout-metrics.md deleted file mode 100644 index bd9ef83afbc..00000000000 --- a/.server-changes/realtime-emission-fanout-metrics.md +++ /dev/null @@ -1,6 +0,0 @@ ---- -area: webapp -type: improvement ---- - -Add metrics to the realtime backend that measure how often a single changed run is served to multiple subscriptions in one batch. diff --git a/.server-changes/report-messages-not-treeshaken.md b/.server-changes/report-messages-not-treeshaken.md new file mode 100644 index 00000000000..b599e84f1ce --- /dev/null +++ b/.server-changes/report-messages-not-treeshaken.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Fix the health report failing with an internal error when requested through the API. diff --git a/.server-changes/scoped-environment-api-key-auth.md b/.server-changes/scoped-environment-api-key-auth.md new file mode 100644 index 00000000000..948f44d1977 --- /dev/null +++ b/.server-changes/scoped-environment-api-key-auth.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: feature +--- + +Additional environment API keys can authenticate API requests using their configured permissions, with revoked and expired keys rejected. Batch responses use server-issued public access tokens so additional keys never need the environment signing secret. diff --git a/.server-changes/sso-directory-sync-dev-environments.md b/.server-changes/sso-directory-sync-dev-environments.md new file mode 100644 index 00000000000..072482b8b19 --- /dev/null +++ b/.server-changes/sso-directory-sync-dev-environments.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Team members who join an organization through single sign-on or directory sync now get their own development environment for every project, so the dashboard's dev view and `trigger dev` become available to them without an admin having to set anything up. Members who were already missing one are repaired the next time their membership is synced from your identity provider. diff --git a/.server-changes/supervisor-run-pod-tolerations.md b/.server-changes/supervisor-run-pod-tolerations.md new file mode 100644 index 00000000000..885b2b596da --- /dev/null +++ b/.server-changes/supervisor-run-pod-tolerations.md @@ -0,0 +1,6 @@ +--- +area: supervisor +type: feature +--- + +Self-hosted Kubernetes deployments can now add tolerations to run pods, so runs are allowed onto tainted nodes. An invalid toleration now stops the supervisor at startup instead of failing every run pod, so check existing values before upgrading. diff --git a/.server-changes/worker-deployment-lookup-ordering.md b/.server-changes/worker-deployment-lookup-ordering.md new file mode 100644 index 00000000000..1dbfbdd253a --- /dev/null +++ b/.server-changes/worker-deployment-lookup-ordering.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: improvement +--- + +Speeds up resolving the latest worker version and deployment for an environment, removing an occasional stall when triggering runs in projects that have accumulated many deployed versions. diff --git a/apps/supervisor/src/backpressure/backpressureMetrics.ts b/apps/supervisor/src/backpressure/backpressureMetrics.ts index ffe57628548..9b622357a38 100644 --- a/apps/supervisor/src/backpressure/backpressureMetrics.ts +++ b/apps/supervisor/src/backpressure/backpressureMetrics.ts @@ -8,6 +8,8 @@ export class BackpressureMetrics { readonly dryRun: Gauge; /** Dequeue attempts the gate skipped - or would have, in dry-run (labelled). */ readonly skipsTotal: Counter; + /** Verdict source reads that failed (threw). */ + readonly readFailuresTotal: Counter; constructor(opts: { register: Registry; prefix?: string }) { const prefix = opts.prefix ?? "supervisor_backpressure"; @@ -30,5 +32,11 @@ export class BackpressureMetrics { labelNames: ["dry_run"], registers: [opts.register], }); + + this.readFailuresTotal = new Counter({ + name: `${prefix}_read_failures_total`, + help: "Verdict source reads that threw", + registers: [opts.register], + }); } } diff --git a/apps/supervisor/src/backpressure/backpressureMonitor.test.ts b/apps/supervisor/src/backpressure/backpressureMonitor.test.ts index 7af28ffc9f5..e6c00394bbb 100644 --- a/apps/supervisor/src/backpressure/backpressureMonitor.test.ts +++ b/apps/supervisor/src/backpressure/backpressureMonitor.test.ts @@ -89,6 +89,60 @@ describe("BackpressureMonitor", () => { monitor.stop(); }); + it("holds an engaged verdict while reads fail, then releases past the max age", async () => { + let call = 0; + const source: BackpressureSignalSource = { + read: async () => { + call++; + if (call === 1) { + return { engaged: true, ts: Date.now() }; + } + throw new Error("signal source unreachable"); + }, + }; + const monitor = new BackpressureMonitor({ + enabled: true, + source, + refreshIntervalMs: 1000, + maxVerdictAgeMs: 15_000, + }); + + monitor.start(); + await vi.advanceTimersByTimeAsync(0); + expect(monitor.shouldSkipDequeue()).toBe(true); + + await vi.advanceTimersByTimeAsync(5000); + expect(monitor.shouldSkipDequeue()).toBe(true); // read failing, verdict held + + await vi.advanceTimersByTimeAsync(11_000); + expect(monitor.shouldSkipDequeue()).toBe(false); // past max age, released + + monitor.stop(); + }); + + it("releases immediately on an explicit null even when a grace window is configured", async () => { + let engaged: boolean | null = true; + const source: BackpressureSignalSource = { + read: async () => (engaged === null ? null : { engaged, ts: Date.now() }), + }; + const monitor = new BackpressureMonitor({ + enabled: true, + source, + refreshIntervalMs: 1000, + maxVerdictAgeMs: 15_000, + }); + + monitor.start(); + await vi.advanceTimersByTimeAsync(0); + expect(monitor.shouldSkipDequeue()).toBe(true); + + engaged = null; + await vi.advanceTimersByTimeAsync(1000); + expect(monitor.shouldSkipDequeue()).toBe(false); // null is an answer, not a failure + + monitor.stop(); + }); + it("fails open when the source reports unknown (null)", async () => { const { source } = countingSource(null); const monitor = new BackpressureMonitor({ enabled: true, source, refreshIntervalMs: 1000 }); @@ -292,6 +346,7 @@ describe("BackpressureMonitor", () => { const logs: Array<{ message: string; meta?: Record }> = []; const logger = { info: (message: string, meta?: Record) => logs.push({ message, meta }), + error: (message: string, meta?: Record) => logs.push({ message, meta }), }; const monitor = new BackpressureMonitor({ enabled: true, diff --git a/apps/supervisor/src/backpressure/backpressureMonitor.ts b/apps/supervisor/src/backpressure/backpressureMonitor.ts index 6b4170697e5..aa16fdeaa60 100644 --- a/apps/supervisor/src/backpressure/backpressureMonitor.ts +++ b/apps/supervisor/src/backpressure/backpressureMonitor.ts @@ -2,6 +2,7 @@ import type { BackpressureMetrics } from "./backpressureMetrics.js"; export interface BackpressureLogger { info(message: string, meta?: Record): void; + error(message: string, meta?: Record): void; } export type BackpressureVerdict = { @@ -11,9 +12,10 @@ export type BackpressureVerdict = { }; /** - * Source of the current backpressure verdict. `read()` returns `null` when the - * verdict is unknown (missing/unreadable) - the monitor treats unknown as - * "not engaged" (fail-open). + * Source of the current backpressure verdict. `read()` returns `null` when the source + * answered but there is no verdict - the monitor treats that as "not engaged" + * (fail-open). A thrown error is different: the read itself failed, so the monitor + * keeps the previous verdict until it ages past `maxVerdictAgeMs`. */ export interface BackpressureSignalSource { read(): Promise; @@ -24,8 +26,9 @@ export type BackpressureMonitorOptions = { source: BackpressureSignalSource; refreshIntervalMs?: number; /** - * If set, a cached verdict older than this is treated as unknown (fail-open). - * Guards against the source silently going stale (e.g. hanging reads). + * If set, an engaged verdict older than this is released (fail-open), bounding how + * long a dead source can hold the brake. Reads that fail keep the last verdict, so + * this doubles as the grace window for riding out a transient source outage. */ maxVerdictAgeMs?: number; /** @@ -54,6 +57,7 @@ export class BackpressureMonitor { private refreshInFlight = false; private wasEngaged = false; private releasedAt?: number; + private readFailing = false; constructor(private readonly opts: BackpressureMonitorOptions) { this.opts.metrics?.dryRun.set(this.opts.dryRun ? 1 : 0); @@ -152,12 +156,31 @@ export class BackpressureMonitor { } private async refresh(): Promise { + let next: BackpressureVerdict | null = null; + let readError: unknown; try { - this.verdict = await this.opts.source.read(); - } catch { - // Fail-open: a dead/unreachable source must never pin the brake. Treat as - // unknown (no verdict) so dequeue resumes as if backpressure were off. - this.verdict = null; + next = await this.opts.source.read(); + } catch (error) { + readError = error; + } + + if (readError === undefined) { + this.verdict = next; // an explicit null means "no pressure", so honour it + this.readFailing = false; + } else { + const held = this.opts.maxVerdictAgeMs !== undefined; + if (!held) { + this.verdict = null; // unbounded hold could pin the brake forever + } + this.opts.metrics?.readFailuresTotal.inc(); + if (!this.readFailing) { + this.readFailing = true; // log once per outage, not once per tick + this.opts.logger?.error("backpressure read failed", { + reason: String(readError), + heldPreviousVerdict: held, + engaged: this.computeEngaged(), + }); + } } // Track the engaged→released transition to anchor the resume ramp. Use the diff --git a/apps/supervisor/src/backpressure/k8sPodCountSignalSource.test.ts b/apps/supervisor/src/backpressure/k8sPodCountSignalSource.test.ts index 8c7104cfb14..12cd92e2d88 100644 --- a/apps/supervisor/src/backpressure/k8sPodCountSignalSource.test.ts +++ b/apps/supervisor/src/backpressure/k8sPodCountSignalSource.test.ts @@ -1,52 +1,49 @@ import { describe, it, expect } from "vitest"; -import { parsePodCount, K8sPodCountSignalSource } from "./k8sPodCountSignalSource.js"; +import { K8sPodCountSignalSource } from "./k8sPodCountSignalSource.js"; +import { podCountFromList, withTimeout } from "../clients/kubernetes.js"; -describe("parsePodCount", () => { - it("reads the pods object count", () => { - const text = [ - "# HELP apiserver_storage_objects Number of stored objects", - "# TYPE apiserver_storage_objects gauge", - 'apiserver_storage_objects{resource="pods"} 8421', - 'apiserver_storage_objects{resource="configmaps"} 17', - ].join("\n"); - expect(parsePodCount(text)).toBe(8421); +describe("podCountFromList", () => { + it("returns items.length when the list is not truncated", () => { + expect(podCountFromList({ items: [{}], metadata: {} })).toBe(1); }); - it("is tolerant of extra labels in any order", () => { - const text = 'apiserver_storage_objects{group="",resource="pods",extra="x"} 12'; - expect(parsePodCount(text)).toBe(12); + it("returns zero for an empty namespace", () => { + expect(podCountFromList({ items: [], metadata: {} })).toBe(0); }); - it("parses scientific notation", () => { - const text = 'apiserver_storage_objects{resource="pods"} 1.2e+04'; - expect(parsePodCount(text)).toBe(12000); + it("adds remainingItemCount when the list is truncated", () => { + const list = { items: [{}], metadata: { _continue: "tok", remainingItemCount: 24492 } }; + expect(podCountFromList(list)).toBe(24493); }); - it("throws when the pods metric is absent", () => { - const text = 'apiserver_storage_objects{resource="configmaps"} 17'; - expect(() => parsePodCount(text)).toThrow(/not found/); + it("throws when truncated but remainingItemCount is absent", () => { + const list = { items: [{}], metadata: { _continue: "tok" } }; + expect(() => podCountFromList(list)).toThrow(/remainingItemCount/); }); - it("throws on a non-finite value (e.g. 1e999)", () => { - const text = 'apiserver_storage_objects{resource="pods"} 1e999'; - expect(() => parsePodCount(text)).toThrow(); + it("throws when truncated but remainingItemCount is negative", () => { + const list = { items: [{}], metadata: { _continue: "tok", remainingItemCount: -1 } }; + expect(() => podCountFromList(list)).toThrow(/remainingItemCount/); }); +}); - it("throws on a negative value", () => { - const text = 'apiserver_storage_objects{resource="pods"} -5'; - expect(() => parsePodCount(text)).toThrow(); +describe("withTimeout", () => { + it("rejects once the deadline passes", async () => { + await expect(withTimeout(new Promise(() => {}), 10, "pod count list")).rejects.toThrow( + /timed out/ + ); }); -}); -function metrics(count: number): string { - return `apiserver_storage_objects{resource="pods"} ${count}`; -} + it("passes a value through when it settles first", async () => { + await expect(withTimeout(Promise.resolve(7), 1000, "pod count list")).resolves.toBe(7); + }); +}); describe("K8sPodCountSignalSource", () => { it("engages at the engage threshold and reports the count", async () => { const counts: number[] = []; const source = new K8sPodCountSignalSource({ - fetchMetrics: async () => metrics(10000), + fetchPodCount: async () => 10000, engageThreshold: 10000, releaseThreshold: 5000, reportPodCount: (c) => counts.push(c), @@ -59,7 +56,7 @@ describe("K8sPodCountSignalSource", () => { it("does not engage below the engage threshold", async () => { const source = new K8sPodCountSignalSource({ - fetchMetrics: async () => metrics(9999), + fetchPodCount: async () => 9999, engageThreshold: 10000, releaseThreshold: 5000, }); @@ -69,7 +66,7 @@ describe("K8sPodCountSignalSource", () => { it("stays engaged in the hysteresis band, releases only below release threshold", async () => { let count = 10000; const source = new K8sPodCountSignalSource({ - fetchMetrics: async () => metrics(count), + fetchPodCount: async () => count, engageThreshold: 10000, releaseThreshold: 5000, }); @@ -82,9 +79,9 @@ describe("K8sPodCountSignalSource", () => { expect((await source.read()).engaged).toBe(false); // band again -> stays off }); - it("propagates scrape failures (monitor fails open on throw)", async () => { + it("propagates fetch failures (monitor fails open on throw)", async () => { const source = new K8sPodCountSignalSource({ - fetchMetrics: async () => { + fetchPodCount: async () => { throw new Error("connection refused"); }, engageThreshold: 10000, diff --git a/apps/supervisor/src/backpressure/k8sPodCountSignalSource.ts b/apps/supervisor/src/backpressure/k8sPodCountSignalSource.ts index e1fa0b78b11..949ebbfa025 100644 --- a/apps/supervisor/src/backpressure/k8sPodCountSignalSource.ts +++ b/apps/supervisor/src/backpressure/k8sPodCountSignalSource.ts @@ -1,22 +1,7 @@ import type { BackpressureSignalSource, BackpressureVerdict } from "./backpressureMonitor.js"; -// Reads the apiserver's stored-pod-object count from a Prometheus /metrics scrape. -const POD_COUNT_RE = /^apiserver_storage_objects\{[^}]*resource="pods"[^}]*\}\s+([0-9.eE+]+)/m; - -export function parsePodCount(metricsText: string): number { - const match = metricsText.match(POD_COUNT_RE); - if (!match) { - throw new Error('apiserver_storage_objects{resource="pods"} not found in metrics'); - } - const value = Number(match[1]); - if (!Number.isFinite(value)) { - throw new Error(`unparseable pod count: ${match[1]}`); - } - return value; -} - export type K8sPodCountSignalSourceOptions = { - fetchMetrics: () => Promise; + fetchPodCount: () => Promise; engageThreshold: number; releaseThreshold: number; reportPodCount?: (count: number) => void; @@ -29,8 +14,7 @@ export class K8sPodCountSignalSource implements BackpressureSignalSource { constructor(private readonly opts: K8sPodCountSignalSourceOptions) {} async read(): Promise { - const text = await this.opts.fetchMetrics(); - const count = parsePodCount(text); + const count = await this.opts.fetchPodCount(); this.opts.reportPodCount?.(count); if (this.engaged) { diff --git a/apps/supervisor/src/clients/kubernetes.ts b/apps/supervisor/src/clients/kubernetes.ts index 9ab8e5bb3c1..1e511a68e6c 100644 --- a/apps/supervisor/src/clients/kubernetes.ts +++ b/apps/supervisor/src/clients/kubernetes.ts @@ -2,7 +2,6 @@ import * as k8s from "@kubernetes/client-node"; import type { Informer, KubernetesObject, ListPromise } from "@kubernetes/client-node"; import { assertExhaustive } from "@trigger.dev/core/utils"; import { SimpleStructuredLogger } from "@trigger.dev/core/v3/utils/structuredLogger"; -import * as https from "node:https"; export const RUNTIME_ENV = process.env.KUBERNETES_PORT ? "kubernetes" : "local"; @@ -54,55 +53,90 @@ function getKubeConfig() { export { k8s }; /** - * Builds a function that scrapes the apiserver's Prometheus /metrics endpoint. - * One lightweight aggregate read - not a pod listing. Requires the service - * account to be granted GET on the /metrics non-resource URL. + * createPodCountFetcher sizes a namespace's pod collection with a single `limit=1` + * list: one pod transferred, no informer, no watch cache. + * + * This is an ESTIMATE, not an exact count. Kubernetes documents `remainingItemCount` + * as intended for estimating collection size and reserves the right not to set it or + * make it exact. Counting exactly would mean paginating the whole collection, which is + * what this deliberately avoids. Treat the value as a tight estimate from a quorum read + * at request time, and set thresholds with that in mind. + * + * Two request-shape constraints, both load-bearing. A label or field selector makes + * the apiserver omit `remainingItemCount` entirely, and setting `resourceVersion` + * serves a cached count instead of a quorum read - so neither is passed. */ -export function createApiserverMetricsFetcher(timeoutMs: number): () => Promise { - const kubeConfig = getKubeConfig(); +export function createPodCountFetcher( + api: K8sApi, + namespace: string, + timeoutMs: number +): () => Promise { + const serverTimeoutSeconds = Math.max(1, Math.floor(timeoutMs / 1000)); + let pending: Promise | undefined; return async () => { - const cluster = kubeConfig.getCurrentCluster(); - if (!cluster) { - throw new Error("no current cluster in kubeconfig"); + if (pending) { + throw new Error("pod count list still in flight from a previous tick"); } - const url = new URL(`${cluster.server}/metrics`); - const opts: https.RequestOptions = { - method: "GET", - protocol: url.protocol, - hostname: url.hostname, - port: url.port, - path: url.pathname, - }; - // applyToHTTPSOptions sets the cluster CA, client cert/key, and auth headers - // (incl. exec plugins) on the request - so TLS verifies against the cluster - // CA, not the system store. The fetch-options path attaches the CA as an - // https.Agent, which global fetch (undici) ignores. - await kubeConfig.applyToHTTPSOptions(opts); - - return new Promise((resolve, reject) => { - const req = https.request(opts, (res) => { - const status = res.statusCode ?? 0; - let body = ""; - res.setEncoding("utf8"); - res.on("data", (chunk) => { - body += chunk; - }); - res.on("end", () => { - if (status >= 200 && status < 300) { - resolve(body); - } else { - reject(new Error(`apiserver /metrics scrape failed: ${status}`)); - } - }); - }); - // Without this a hung connect/TLS/read never settles, and the monitor's - // refreshInFlight guard would freeze the source (silent fail-open). - req.setTimeout(timeoutMs, () => { - req.destroy(new Error(`apiserver /metrics scrape timed out after ${timeoutMs}ms`)); - }); - req.on("error", reject); - req.end(); + + const request = api.core.listNamespacedPod({ + namespace, + limit: 1, + timeoutSeconds: serverTimeoutSeconds, }); + + pending = request + .catch(() => {}) + .finally(() => { + pending = undefined; + }); + + return podCountFromList(await withTimeout(request, timeoutMs, "pod count list")); }; } + +/** + * podCountFromList turns a `limit=1` pod list into a population estimate. + * + * `remainingItemCount` is only set when the list is truncated, so `_continue` is the + * truncation signal: absent means the returned page is the whole collection and its + * length is exact. When truncated the total leans on `remainingItemCount`, which is + * documented as an estimate - so the result is an estimate too. Truncated without a + * usable count is unknowable, so it throws rather than returning a low number the + * caller would act on. + */ +export function podCountFromList(list: { + items: unknown[]; + metadata?: { _continue?: string; remainingItemCount?: number }; +}): number { + if (!list.metadata?._continue) { + return list.items.length; + } + + const remaining = list.metadata.remainingItemCount; + if (typeof remaining !== "number" || !Number.isFinite(remaining) || remaining < 0) { + throw new Error("pod list truncated but remainingItemCount absent or invalid"); + } + + return list.items.length + remaining; +} + +/** + * withTimeout rejects if `promise` outlives `timeoutMs`, so a hung request cannot + * freeze the caller. It cannot cancel: the k8s client threads no AbortSignal through to + * fetch, so an abandoned request keeps running. Callers must therefore also bound the + * request server-side (`timeoutSeconds`) and refuse to start a second one while the + * first is pending, or a blackholed connection accumulates one socket per attempt. + */ +export function withTimeout(promise: Promise, timeoutMs: number, what: string): Promise { + let timer: NodeJS.Timeout; + const deadline = new Promise((_resolve, reject) => { + timer = setTimeout( + () => reject(new Error(`${what} timed out after ${timeoutMs}ms`)), + timeoutMs + ); + timer.unref(); + }); + + return Promise.race([promise, deadline]).finally(() => clearTimeout(timer)); +} diff --git a/apps/supervisor/src/env.ts b/apps/supervisor/src/env.ts index 6c954b4e76a..670a00ab582 100644 --- a/apps/supervisor/src/env.ts +++ b/apps/supervisor/src/env.ts @@ -1,7 +1,7 @@ import { randomUUID } from "crypto"; import { env as stdEnv } from "std-env"; import { z } from "zod"; -import { AdditionalEnvVars, BoolEnv } from "./envUtil.js"; +import { AdditionalEnvVars, BoolEnv, NodeLabelValue, Tolerations } from "./envUtil.js"; export const Env = z .object({ @@ -79,7 +79,7 @@ export const Env = z .number() .int() .positive() - .default(15_000), // Stale verdict → fail-open (treat as not engaged) + .default(120_000), // Grace window: held verdict older than this → fail-open TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_HOST: z.string().optional(), TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_PORT: z.coerce.number().int().optional(), TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_USERNAME: z.string().optional(), @@ -112,6 +112,7 @@ export const Env = z // Optional services TRIGGER_WARM_START_URL: z.string().optional(), + TRIGGER_WARM_START_DISPATCH_URL: z.string().optional(), TRIGGER_CHECKPOINT_URL: z.string().optional(), TRIGGER_METADATA_URL: z.string().optional(), @@ -172,7 +173,7 @@ export const Env = z // Kubernetes settings KUBERNETES_FORCE_ENABLED: BoolEnv.default(false), KUBERNETES_NAMESPACE: z.string().default("default"), - KUBERNETES_WORKER_NODETYPE_LABEL: z.string().default("v4-worker"), + KUBERNETES_WORKER_NODETYPE_LABEL: NodeLabelValue.default("v4-worker"), KUBERNETES_IMAGE_PULL_SECRETS: z.string().optional(), // csv KUBERNETES_EPHEMERAL_STORAGE_SIZE_LIMIT: z.string().default("10Gi"), KUBERNETES_EPHEMERAL_STORAGE_SIZE_REQUEST: z.string().default("2Gi"), @@ -255,65 +256,8 @@ export const Env = z .max(100) .default(20), - // Schedule toleration settings - scheduled runs tolerate taints on the dedicated pool - // Comma-separated list of tolerations in the format: key=value:effect - // For Exists operator (no value): key:effect - KUBERNETES_SCHEDULED_RUN_TOLERATIONS: z - .string() - .transform((val, ctx) => { - const tolerations = val - .split(",") - .map((entry) => entry.trim()) - .filter((entry) => entry.length > 0) - .map((entry) => { - const colonIdx = entry.lastIndexOf(":"); - if (colonIdx === -1) { - ctx.addIssue({ - code: z.ZodIssueCode.custom, - message: `Invalid toleration format (missing effect): "${entry}"`, - }); - return z.NEVER; - } - - const effect = entry.slice(colonIdx + 1); - const validEffects = ["NoSchedule", "NoExecute", "PreferNoSchedule"]; - if (!validEffects.includes(effect)) { - ctx.addIssue({ - code: z.ZodIssueCode.custom, - message: `Invalid toleration effect "${effect}" in "${entry}". Must be one of: ${validEffects.join( - ", " - )}`, - }); - return z.NEVER; - } - - const keyValue = entry.slice(0, colonIdx); - const eqIdx = keyValue.indexOf("="); - const key = eqIdx === -1 ? keyValue : keyValue.slice(0, eqIdx); - - if (!key) { - ctx.addIssue({ - code: z.ZodIssueCode.custom, - message: `Invalid toleration format (empty key): "${entry}"`, - }); - return z.NEVER; - } - - if (eqIdx === -1) { - return { key, operator: "Exists" as const, effect }; - } - - return { - key, - operator: "Equal" as const, - value: keyValue.slice(eqIdx + 1), - effect, - }; - }); - - return tolerations; - }) - .optional(), + KUBERNETES_RUNNER_TOLERATIONS: Tolerations.optional(), // every run pod + KUBERNETES_SCHEDULED_RUN_TOLERATIONS: Tolerations.optional(), // schedule-tree runs only // Placement tags settings PLACEMENT_TAGS_ENABLED: BoolEnv.default(false), diff --git a/apps/supervisor/src/envUtil.test.ts b/apps/supervisor/src/envUtil.test.ts index c3d35758f16..378830f8ab0 100644 --- a/apps/supervisor/src/envUtil.test.ts +++ b/apps/supervisor/src/envUtil.test.ts @@ -1,5 +1,5 @@ import { describe, it, expect } from "vitest"; -import { BoolEnv, AdditionalEnvVars } from "./envUtil.js"; +import { BoolEnv, AdditionalEnvVars, NodeLabelValue, Tolerations } from "./envUtil.js"; describe("BoolEnv", () => { it("should parse string 'true' as true", () => { @@ -78,3 +78,128 @@ describe("AdditionalEnvVars", () => { }); }); }); + +describe("NodeLabelValue", () => { + it("should keep a clean value untouched", () => { + expect(NodeLabelValue.parse("v4-worker")).toBe("v4-worker"); + }); + + it("should trim surrounding whitespace, which Kubernetes would reject", () => { + expect(NodeLabelValue.parse(" v4-worker ")).toBe("v4-worker"); + expect(NodeLabelValue.parse("\tv4-worker\n")).toBe("v4-worker"); + }); + + it("should treat a whitespace-only value as the empty off-switch", () => { + expect(NodeLabelValue.parse("")).toBe(""); + expect(NodeLabelValue.parse(" ")).toBe(""); + }); + + it("should still apply a default only when unset", () => { + const withDefault = NodeLabelValue.default("v4-worker"); + expect(withDefault.parse(undefined)).toBe("v4-worker"); + expect(withDefault.parse("")).toBe(""); + }); + + it("should reject a value Kubernetes would reject, rather than 422 every pod create", () => { + for (const invalid of ["my worker", "-bad-", "bad.", "a".repeat(64)]) { + expect(NodeLabelValue.safeParse(invalid).success).toBe(false); + } + }); +}); + +describe("Tolerations", () => { + it("should parse key=value entries as Equal", () => { + expect(Tolerations.parse("dedicated=runs:NoSchedule")).toEqual([ + { key: "dedicated", operator: "Equal", value: "runs", effect: "NoSchedule" }, + ]); + }); + + it("should parse entries without a value as Exists", () => { + expect(Tolerations.parse("scheduled-runs:NoExecute")).toEqual([ + { key: "scheduled-runs", operator: "Exists", effect: "NoExecute" }, + ]); + }); + + it("should keep an empty value as an exact match for a valueless taint", () => { + expect(Tolerations.parse("dedicated=:NoSchedule")).toEqual([ + { key: "dedicated", operator: "Equal", value: "", effect: "NoSchedule" }, + ]); + + expect(Tolerations.parse("dedicated:NoSchedule")).toEqual([ + { key: "dedicated", operator: "Exists", effect: "NoSchedule" }, + ]); + }); + + it("should parse an empty string as no tolerations", () => { + expect(Tolerations.parse("")).toEqual([]); + expect(Tolerations.parse(" ")).toEqual([]); + }); + + it("should skip blank entries and trim whitespace", () => { + expect(Tolerations.parse(" a=b:NoSchedule , ,")).toEqual([ + { key: "a", operator: "Equal", value: "b", effect: "NoSchedule" }, + ]); + }); + + it("should reject a missing effect, an unknown effect, and an empty key", () => { + for (const invalid of ["dedicated=runs", "dedicated=runs:Nope", "=runs:NoSchedule"]) { + expect(Tolerations.safeParse(invalid).success).toBe(false); + } + }); + + it("should accept a hyphenated key, a digit-suffixed key, and every effect", () => { + expect( + Tolerations.parse("capacity-1=true:PreferNoSchedule,spot:NoExecute,gpu=a10:NoSchedule") + ).toEqual([ + { key: "capacity-1", operator: "Equal", value: "true", effect: "PreferNoSchedule" }, + { key: "spot", operator: "Exists", effect: "NoExecute" }, + { key: "gpu", operator: "Equal", value: "a10", effect: "NoSchedule" }, + ]); + }); + + it("should accept a DNS-subdomain prefixed key", () => { + expect( + Tolerations.parse("node.cluster.x-k8s.io/machinepool=scheduled-runs:NoSchedule") + ).toEqual([ + { + key: "node.cluster.x-k8s.io/machinepool", + operator: "Equal", + value: "scheduled-runs", + effect: "NoSchedule", + }, + ]); + }); + + it("should reject a key or value that Kubernetes would reject at pod create", () => { + for (const invalid of [ + "dedicated=prod runs:NoSchedule", + "ded icated=runs:NoSchedule", + "dedicated=-runs:NoSchedule", + `dedicated=${"r".repeat(64)}:NoSchedule`, + `${"a".repeat(64)}=runs:NoSchedule`, + `example.com/${"a".repeat(64)}=runs:NoSchedule`, + "a/b/c=runs:NoSchedule", + "Example.com/pool=runs:NoSchedule", + ]) { + expect(Tolerations.safeParse(invalid).success).toBe(false); + } + }); + + it("should bound the prefix and the name separately, as Kubernetes does", () => { + const longestPrefix = `${"a".repeat(63)}.${"b".repeat(63)}.${"c".repeat(63)}.${"d".repeat(61)}`; + expect(longestPrefix.length).toBe(253); + + expect(Tolerations.parse(`${longestPrefix}/${"n".repeat(63)}=runs:NoSchedule`)).toHaveLength(1); + expect(Tolerations.safeParse(`${longestPrefix}a/pool=runs:NoSchedule`).success).toBe(false); + }); + + it("should tolerate whitespace around the separators", () => { + expect(Tolerations.parse("dedicated = runs : NoSchedule")).toEqual([ + { key: "dedicated", operator: "Equal", value: "runs", effect: "NoSchedule" }, + ]); + }); + + it("should reject a stray extra effect instead of folding it into the value", () => { + expect(Tolerations.safeParse("dedicated=runs:NoSchedule:NoExecute").success).toBe(false); + }); +}); diff --git a/apps/supervisor/src/envUtil.ts b/apps/supervisor/src/envUtil.ts index 917f984cc37..67811f76fcb 100644 --- a/apps/supervisor/src/envUtil.ts +++ b/apps/supervisor/src/envUtil.ts @@ -16,6 +16,136 @@ export const BoolEnv = baseBoolEnv as Omit & { default: (value: boolean) => z.ZodDefault; }; +const QUALIFIED_NAME = /^[A-Za-z0-9]([-A-Za-z0-9_.]*[A-Za-z0-9])?$/; +const DNS_SUBDOMAIN = /^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$/; +const LABEL_VALUE = /^(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])?$/; +const QUALIFIED_NAME_MAX = 63; +const DNS_SUBDOMAIN_MAX = 253; +const LABEL_VALUE_MAX = 63; + +/** + * isLabelValue mirrors the Kubernetes label value rules. Empty is valid upstream. + */ +function isLabelValue(value: string): boolean { + return value.length <= LABEL_VALUE_MAX && LABEL_VALUE.test(value); +} + +/** + * isQualifiedName mirrors the Kubernetes qualified name rules used for taint and + * label keys: an optional DNS subdomain prefix before the slash, then the name. + * The two halves have different length limits and different case rules, so a + * single pattern with one overall bound gets both ends wrong. + */ +function isQualifiedName(key: string): boolean { + const slashIdx = key.indexOf("/"); + + if (slashIdx === -1) { + return key.length <= QUALIFIED_NAME_MAX && QUALIFIED_NAME.test(key); + } + + const prefix = key.slice(0, slashIdx); + const name = key.slice(slashIdx + 1); + + return ( + prefix.length <= DNS_SUBDOMAIN_MAX && + DNS_SUBDOMAIN.test(prefix) && + name.length <= QUALIFIED_NAME_MAX && + QUALIFIED_NAME.test(name) + ); +} + +/** + * A node label value. Trimmed because Kubernetes rejects surrounding whitespace + * outright, so a padded value fails every pod create. Deliberately no `min(1)`: + * empty is the off-switch, and the Helm chart ships empty by default. + */ +export const NodeLabelValue = z.string().trim().refine(isLabelValue, { + message: + "Must be a Kubernetes label value: alphanumeric, with dashes, underscores and dots inside, at most 63 characters", +}); + +/** + * Comma-separated pod tolerations in the format `key=value:effect`, or `key:effect` + * for the Exists operator. Keys and values are checked against the Kubernetes + * naming rules here so a typo fails at startup, rather than 422ing every single + * pod create with the cause buried in an API server message. + */ +export const Tolerations = z.string().transform((val, ctx) => { + return val + .split(",") + .map((entry) => entry.trim()) + .filter((entry) => entry.length > 0) + .map((entry) => { + const colonIdx = entry.lastIndexOf(":"); + if (colonIdx === -1) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: `Invalid toleration format (missing effect): "${entry}"`, + }); + return z.NEVER; + } + + const effect = entry.slice(colonIdx + 1).trim(); + const validEffects = ["NoSchedule", "NoExecute", "PreferNoSchedule"]; + if (!validEffects.includes(effect)) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: `Invalid toleration effect "${effect}" in "${entry}". Must be one of: ${validEffects.join( + ", " + )}`, + }); + return z.NEVER; + } + + const keyValue = entry.slice(0, colonIdx); + const eqIdx = keyValue.indexOf("="); + const key = (eqIdx === -1 ? keyValue : keyValue.slice(0, eqIdx)).trim(); + + if (!key) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: `Invalid toleration format (empty key): "${entry}"`, + }); + return z.NEVER; + } + + if (!isQualifiedName(key)) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: `Invalid toleration key "${key}" in "${entry}". Must be a Kubernetes taint key, optionally prefixed with a DNS subdomain.`, + }); + return z.NEVER; + } + + if (eqIdx === -1) { + return { key, operator: "Exists" as const, effect }; + } + + const value = keyValue.slice(eqIdx + 1).trim(); + if (!value) { + logger.warn( + 'Toleration has an empty value, so it matches only a taint whose value is also empty. Drop the "=" to tolerate any value of this key.', + { entry, key } + ); + } + + if (!isLabelValue(value)) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: `Invalid toleration value "${value}" in "${entry}". Must be a Kubernetes label value: alphanumeric, with dashes, underscores and dots inside.`, + }); + return z.NEVER; + } + + return { + key, + operator: "Equal" as const, + value, + effect, + }; + }); +}); + export const AdditionalEnvVars = z.preprocess((val) => { if (typeof val !== "string") { return val; diff --git a/apps/supervisor/src/index.ts b/apps/supervisor/src/index.ts index cf73be90eea..542841bd6e5 100644 --- a/apps/supervisor/src/index.ts +++ b/apps/supervisor/src/index.ts @@ -21,8 +21,8 @@ import { CheckpointClient, isKubernetesEnvironment, } from "@trigger.dev/core/v3/serverOnly"; -import { createK8sApi, createApiserverMetricsFetcher } from "./clients/kubernetes.js"; -import { collectDefaultMetrics, Gauge, Histogram } from "prom-client"; +import { createK8sApi, createPodCountFetcher } from "./clients/kubernetes.js"; +import { collectDefaultMetrics, Counter, Gauge, Histogram } from "prom-client"; import { register } from "./metrics.js"; import { PodCleaner } from "./services/podCleaner.js"; import { FailedPodHandler } from "./services/failedPodHandler.js"; @@ -60,6 +60,21 @@ const workloadCreateDuration = new Histogram({ registers: [register], }); +const outboundRequestsTotal = new Counter({ + name: "supervisor_outbound_request_total", + help: "Count of outbound HTTP requests from the supervisor, by target name, method, response status, and outcome (ok, http_error, invalid_response, network_error).", + labelNames: ["name", "method", "status", "outcome"], + registers: [register], +}); + +const outboundRequestDuration = new Histogram({ + name: "supervisor_outbound_request_duration_seconds", + help: "Duration of outbound HTTP requests from the supervisor, by target name and outcome. Includes the HTTP client's internal retries and backoff.", + labelNames: ["name", "outcome"], + buckets: [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2, 5, 10, 11, 12.5, 15, 20, 30, 60], + registers: [register], +}); + class ManagedSupervisor { private readonly workerSession: SupervisorSession; private readonly metricsServer?: HttpServer; @@ -80,6 +95,8 @@ class ManagedSupervisor { private readonly isKubernetes = isKubernetesEnvironment(env.KUBERNETES_FORCE_ENABLED); private readonly warmStartUrl = env.TRIGGER_WARM_START_URL; + private readonly warmStartDispatchUrl = + env.TRIGGER_WARM_START_DISPATCH_URL ?? env.TRIGGER_WARM_START_URL; private readonly wideEventOpts: WideEventOptions = { service: "supervisor", @@ -259,14 +276,16 @@ class ManagedSupervisor { // RELEASE < ENGAGE is enforced in env.ts (superRefine), so it's valid here. const podCountGauge = new Gauge({ name: "supervisor_cluster_pod_count", - help: "Total pod objects stored in the cluster, scraped for backpressure", + help: "Pod objects in the workload namespace, counted for backpressure", registers: [register], }); this.backpressureMonitors.push( new BackpressureMonitor({ enabled: true, source: new K8sPodCountSignalSource({ - fetchMetrics: createApiserverMetricsFetcher( + fetchPodCount: createPodCountFetcher( + createK8sApi(), + env.KUBERNETES_NAMESPACE, env.TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_SCRAPE_TIMEOUT_MS ), engageThreshold: env.TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_ENGAGE, @@ -322,6 +341,10 @@ class ManagedSupervisor { runNotificationsEnabled: env.TRIGGER_WORKLOAD_API_ENABLED, heartbeatIntervalSeconds: env.TRIGGER_WORKER_HEARTBEAT_INTERVAL_SECONDS, sendRunDebugLogs: env.SEND_RUN_DEBUG_LOGS, + onHttpRequestComplete: ({ name, method, status, outcome, durationMs }) => { + outboundRequestsTotal.inc({ name, method, status, outcome }); + outboundRequestDuration.observe({ name, outcome }, durationMs / 1000); + }, preDequeue: async () => { // Synchronous, hot-path-safe cached read; false when no monitors are active. const skipForBackpressure = this.backpressureMonitors.some((m) => m.shouldSkipDequeue()); @@ -679,7 +702,7 @@ class ManagedSupervisor { return false; } - const warmStartUrlWithPath = new URL("/warm-start", this.warmStartUrl); + const warmStartUrlWithPath = new URL("/warm-start", this.warmStartDispatchUrl); const headers: Record = { "Content-Type": "application/json", @@ -692,6 +715,18 @@ class ManagedSupervisor { headers.traceparent = traceparent; } + const requestStart = performance.now(); + const record = ( + status: string, + outcome: "ok" | "http_error" | "invalid_response" | "network_error" + ) => { + outboundRequestsTotal.inc({ name: "warm_start", method: "POST", status, outcome }); + outboundRequestDuration.observe( + { name: "warm_start", outcome }, + (performance.now() - requestStart) / 1000 + ); + }; + try { const res = await fetch(warmStartUrlWithPath.href, { method: "POST", @@ -700,8 +735,10 @@ class ManagedSupervisor { }); if (!res.ok) { + record(String(res.status), "http_error"); this.logger.error("Warm start failed", { runId: dequeuedMessage.run.id, + statusCode: res.status, }); return false; } @@ -710,6 +747,7 @@ class ManagedSupervisor { const parsedData = z.object({ didWarmStart: z.boolean() }).safeParse(data); if (!parsedData.success) { + record(String(res.status), "invalid_response"); this.logger.error("Warm start response invalid", { runId: dequeuedMessage.run.id, data, @@ -717,8 +755,11 @@ class ManagedSupervisor { return false; } + record(String(res.status), "ok"); + return parsedData.data.didWarmStart; } catch (error) { + record("none", "network_error"); this.logger.error("Warm start error", { runId: dequeuedMessage.run.id, error, diff --git a/apps/supervisor/src/workloadManager/kubernetes.test.ts b/apps/supervisor/src/workloadManager/kubernetes.test.ts index 85ad3cbebff..bb15c23e9f4 100644 --- a/apps/supervisor/src/workloadManager/kubernetes.test.ts +++ b/apps/supervisor/src/workloadManager/kubernetes.test.ts @@ -1,6 +1,8 @@ import { describe, expect, it } from "vitest"; import { BLOCK_IO_URING_SECCOMP_PROFILE, + nodetypeNodeSelector, + runPodTolerations, withBlockIoUringSeccompProfile, } from "./kubernetesPodSpec.js"; @@ -14,6 +16,46 @@ const basePodSpec = { }, }; +describe("nodetypeNodeSelector", () => { + it("omits the nodeSelector entirely when the label is empty or unset", () => { + for (const label of ["", undefined]) { + expect(nodetypeNodeSelector(label)).toEqual({}); + } + }); + + it("pins to nodetype=