feat: multiple slot types (#2927)

* feat: adds support for multiple slot types, primarily motivated by durable slots

---------

Co-authored-by: mrkaye97 <mrkaye97@gmail.com>

Author: grutt

.github/workflows/release.yaml

@@ -9,7 +9,7 @@ on:
 name: Release
 jobs:
   load:
-    runs-on: ubicloud-standard-4
+    runs-on: ubicloud-standard-8
     timeout-minutes: 30
     strategy:
       matrix:

.github/workflows/test.yml

@@ -54,7 +54,7 @@ jobs:
         run: docker compose down
 
   unit:
-    runs-on: ubicloud-standard-4
+    runs-on: ubicloud-standard-8
     steps:
       - uses: actions/checkout@v6
       - name: Setup Go
@@ -72,7 +72,7 @@ jobs:
         run: go test $(go list ./... | grep -v "quickstart") -v -failfast
 
   integration:
-    runs-on: ubicloud-standard-4
+    runs-on: ubicloud-standard-8
     env:
       DATABASE_URL: postgresql://hatchet:hatchet@127.0.0.1:5431/hatchet?sslmode=disable
 
@@ -117,7 +117,7 @@ jobs:
         run: docker compose down
 
   e2e:
-    runs-on: ubicloud-standard-4
+    runs-on: ubicloud-standard-8
     timeout-minutes: 30
     env:
       DATABASE_URL: postgresql://hatchet:hatchet@127.0.0.1:5431/hatchet?sslmode=disable
@@ -206,7 +206,7 @@ jobs:
         run: docker compose down
 
   e2e-pgmq:
-    runs-on: ubicloud-standard-4
+    runs-on: ubicloud-standard-8
     timeout-minutes: 30
     env:
       DATABASE_URL: postgresql://hatchet:hatchet@127.0.0.1:5431/hatchet?sslmode=disable
@@ -297,7 +297,7 @@ jobs:
         run: docker compose down
 
   load:
-    runs-on: ubicloud-standard-4
+    runs-on: ubicloud-standard-8
     timeout-minutes: 30
     strategy:
       matrix:
@@ -339,8 +339,213 @@ jobs:
           TESTING_MATRIX_PG_VERSION: ${{ matrix.pg-version }}
           TESTING_MATRIX_OPTIMISTIC_SCHEDULING: ${{ matrix.optimistic-scheduling }}
 
+  load-online-migrate:
+    runs-on: ubicloud-standard-8
+    timeout-minutes: 30
+    env:
+      DATABASE_URL: postgresql://hatchet:hatchet@127.0.0.1:5431/hatchet?sslmode=disable
+
+    steps:
+      - uses: actions/checkout@v6
+        with:
+          fetch-depth: 0
+          fetch-tags: true
+
+      - name: Setup Go
+        uses: actions/setup-go@v6
+        with:
+          go-version: "1.25"
+
+      - name: Compose
+        run: docker compose up -d
+
+      - name: Determine latest stable release tag
+        run: |
+          LATEST_TAG=$(git tag --sort=-v:refname | grep -E '^v[0-9]+\.[0-9]+\.[0-9]+$' | head -1)
+          if [ -z "$LATEST_TAG" ]; then
+            echo "ERROR: No stable release tag found"
+            exit 1
+          fi
+          echo "Latest stable tag: $LATEST_TAG"
+          echo "LATEST_TAG=$LATEST_TAG" >> $GITHUB_ENV
+
+      - name: Pull old release images
+        run: |
+          docker pull ghcr.io/hatchet-dev/hatchet/hatchet-migrate:${{ env.LATEST_TAG }}
+          docker pull ghcr.io/hatchet-dev/hatchet/hatchet-admin:${{ env.LATEST_TAG }}
+          docker pull ghcr.io/hatchet-dev/hatchet/hatchet-engine:${{ env.LATEST_TAG }}
+          docker pull ghcr.io/hatchet-dev/hatchet/hatchet-loadtest:${{ env.LATEST_TAG }}
+
+      - name: Run old migrations
+        run: |
+          docker run --rm --network host \
+            -e DATABASE_URL="${{ env.DATABASE_URL }}" \
+            ghcr.io/hatchet-dev/hatchet/hatchet-migrate:${{ env.LATEST_TAG }}
+
+      - name: Setup config and seed database
+        run: |
+          mkdir -p generated
+          docker run --rm --network host \
+            -v ${{ github.workspace }}/generated:/hatchet/generated \
+            -e DATABASE_URL="${{ env.DATABASE_URL }}" \
+            -e SERVER_GRPC_PORT=7077 \
+            -e SERVER_GRPC_BROADCAST_ADDRESS=localhost:7077 \
+            -e SERVER_GRPC_INSECURE=true \
+            -e SERVER_AUTH_COOKIE_DOMAIN=localhost \
+            -e SERVER_AUTH_COOKIE_INSECURE=true \
+            ghcr.io/hatchet-dev/hatchet/hatchet-admin:${{ env.LATEST_TAG }} \
+            /hatchet/hatchet-admin quickstart --skip certs --generated-config-dir /hatchet/generated
+
+      - name: Generate API token
+        run: |
+          TOKEN=$(docker run --rm --network host \
+            -v ${{ github.workspace }}/generated:/hatchet/generated \
+            -e DATABASE_URL="${{ env.DATABASE_URL }}" \
+            -e SERVER_GRPC_PORT=7077 \
+            -e SERVER_GRPC_BROADCAST_ADDRESS=localhost:7077 \
+            -e SERVER_GRPC_INSECURE=true \
+            -e SERVER_AUTH_COOKIE_DOMAIN=localhost \
+            -e SERVER_AUTH_COOKIE_INSECURE=true \
+            ghcr.io/hatchet-dev/hatchet/hatchet-admin:${{ env.LATEST_TAG }} \
+            /hatchet/hatchet-admin token create --config /hatchet/generated)
+          echo "HATCHET_CLIENT_TOKEN=$TOKEN" >> $GITHUB_ENV
+
+      - name: Start old engine
+        run: |
+          docker run -d --name hatchet-engine --network host \
+            -v ${{ github.workspace }}/generated:/hatchet/generated \
+            -e DATABASE_URL="${{ env.DATABASE_URL }}" \
+            -e SERVER_GRPC_PORT=7077 \
+            -e SERVER_GRPC_BROADCAST_ADDRESS=localhost:7077 \
+            -e SERVER_GRPC_INSECURE=true \
+            -e SERVER_AUTH_COOKIE_DOMAIN=localhost \
+            -e SERVER_AUTH_COOKIE_INSECURE=true \
+            -e SERVER_MSGQUEUE_KIND=postgres \
+            -e SERVER_LOGGER_LEVEL=warn \
+            -e SERVER_LOGGER_FORMAT=console \
+            -e DATABASE_LOGGER_LEVEL=warn \
+            -e DATABASE_LOGGER_FORMAT=console \
+            ghcr.io/hatchet-dev/hatchet/hatchet-engine:${{ env.LATEST_TAG }} \
+            /hatchet/hatchet-engine --config /hatchet/generated
+          echo "Waiting 30s for engine to start..."
+          sleep 30
+
+      - name: Start old load test
+        run: |
+          docker run -d --name hatchet-loadtest --network host \
+            -e HATCHET_CLIENT_TOKEN="${{ env.HATCHET_CLIENT_TOKEN }}" \
+            -e HATCHET_CLIENT_TLS_STRATEGY=none \
+            -e HATCHET_CLIENT_HOST_PORT=localhost:7077 \
+            ghcr.io/hatchet-dev/hatchet/hatchet-loadtest:${{ env.LATEST_TAG }} \
+            /hatchet/hatchet-load-test loadtest -e 10 -d 240s -w 60s -s 100
+
+      - name: Wait then apply new migrations
+        run: |
+          echo "Waiting 30s for load test to get started..."
+          sleep 30
+          echo "Applying new migrations from current branch..."
+          go run ./cmd/hatchet-migrate
+          echo "New migrations applied successfully"
+
+      - name: Wait for load test to complete
+        run: |
+          echo "Waiting for load test container to finish..."
+          docker wait hatchet-loadtest
+          EXIT_CODE=$(docker inspect hatchet-loadtest --format='{{.State.ExitCode}}')
+          echo "Load test exited with code: $EXIT_CODE"
+          if [ "$EXIT_CODE" != "0" ]; then
+            echo "=== Load test logs ==="
+            docker logs hatchet-loadtest
+            echo "=== Engine logs ==="
+            docker logs hatchet-engine
+            exit 1
+          fi
+          echo "Load test passed"
+
+      - name: Teardown
+        if: always()
+        run: |
+          docker rm -f hatchet-loadtest hatchet-engine 2>/dev/null || true
+          docker compose down
+
+  load-deadlock:
+    runs-on: ubicloud-standard-8
+    timeout-minutes: 30
+    strategy:
+      matrix:
+        migrate-strategy: ["latest"]
+        rabbitmq-enabled: ["true"]
+        pg-version: ["17-alpine"]
+        optimistic-scheduling: ["true", "false"]
+
+    steps:
+      - uses: actions/checkout@v6
+
+      - name: Install Task
+        uses: arduino/setup-task@v2
+        with:
+          repo-token: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Setup Go
+        uses: actions/setup-go@v6
+        with:
+          go-version: "1.25"
+
+      - name: Setup pnpm
+        uses: pnpm/action-setup@v4
+        with:
+          version: 10.16.1
+          run_install: false
+
+      - name: Go deps
+        run: go mod download
+
+      - name: Add go-deadlock dependency
+        run: go get github.com/sasha-s/go-deadlock@v0.3.6
+
+      - name: Patch sync imports to use go-deadlock (sed)
+        shell: bash
+        run: |
+          set -euo pipefail
+
+          # Replace ONLY the stdlib "sync" import with an alias that preserves `sync.X` call sites.
+          # - `import "sync"` -> `import sync "github.com/sasha-s/go-deadlock"`
+          # - within import blocks: `"sync"` -> `sync "github.com/sasha-s/go-deadlock"`
+          # NOTE: use `-i''` (no backup) for portability across GNU/BSD sed.
+          find . -name '*.go' -not -path './vendor/*' -print0 | xargs -0 sed -i'' -E \
+            -e 's/^([[:space:]]*)import[[:space:]]+"sync"[[:space:]]*$/\1import sync "github.com\/sasha-s\/go-deadlock"/' \
+            -e 's/^([[:space:]]*)"sync"[[:space:]]*$/\1sync "github.com\/sasha-s\/go-deadlock"/'
+
+          # Keep formatting/import grouping consistent after rewriting.
+          find . -name '*.go' -not -path './vendor/*' -print0 | xargs -0 gofmt -w
+
+          # Evidence in CI logs that rewriting happened (or not).
+          echo "Changed Go files (after patch):"
+          git diff --name-only -- '*.go' || true
+
+          echo ""
+          echo "Contents of pkg/scheduling/v1/scheduler.go after patch:"
+          echo "----"
+          cat pkg/scheduling/v1/scheduler.go
+          echo "----"
+
+      - name: Test (deadlock-instrumented)
+        run: |
+          # Disable gzip compression for load tests to reduce CPU overhead
+          # Compression adds overhead without benefit for 0kb payloads
+          HATCHET_CLIENT_DISABLE_GZIP_COMPRESSION=true go test -tags load ./... -p 5 -v -race -failfast -timeout 20m
+        env:
+          # This job adds go-deadlock + -race overhead; relax perf threshold to avoid flakes.
+          HATCHET_LOADTEST_AVERAGE_DURATION_THRESHOLD: 1s
+          # Give the engine a bit more time to come up under instrumentation.
+          HATCHET_LOADTEST_STARTUP_SLEEP: 30s
+          TESTING_MATRIX_MIGRATE: ${{ matrix.migrate-strategy }}
+          TESTING_MATRIX_RABBITMQ_ENABLED: ${{ matrix.rabbitmq-enabled }}
+          TESTING_MATRIX_PG_VERSION: ${{ matrix.pg-version }}
+          TESTING_MATRIX_OPTIMISTIC_SCHEDULING: ${{ matrix.optimistic-scheduling }}
+
   rampup:
-    runs-on: ubicloud-standard-4
+    runs-on: ubicloud-standard-8
     timeout-minutes: 30
     strategy:
       matrix:

api-contracts/dispatcher/dispatcher.proto

@@ -33,6 +33,11 @@ service Dispatcher {
     rpc ReleaseSlot(ReleaseSlotRequest) returns (ReleaseSlotResponse) {}
 
     rpc UpsertWorkerLabels(UpsertWorkerLabelsRequest) returns (UpsertWorkerLabelsResponse) {}
+
+    // GetVersion returns the dispatcher protocol version as a simple integer.
+    // SDKs use this to determine feature support (e.g. slot_config registration).
+    // Old engines that do not implement this RPC will return UNIMPLEMENTED.
+    rpc GetVersion(GetVersionRequest) returns (GetVersionResponse) {}
 }
 
 message WorkerLabels {
@@ -67,7 +72,8 @@ message WorkerRegisterRequest {
     // (optional) the services for this worker
     repeated string services = 3;
 
-    // (optional) the number of slots this worker can handle
+    // (optional) the number of default slots this worker can handle
+    // deprecated: use slot_config instead
     optional int32 slots = 4;
 
     // (optional) worker labels (i.e. state or other metadata)
@@ -79,6 +85,9 @@ message WorkerRegisterRequest {
     // (optional) information regarding the runtime environment of the worker
     optional RuntimeInfo runtime_info = 7;
 
+    // (optional) slot config for this worker (slot_type -> units)
+    map<string, int32> slot_config = 9;
+
 }
 
 message WorkerRegisterResponse {
@@ -403,3 +412,9 @@ message ReleaseSlotRequest {
 }
 
 message ReleaseSlotResponse {}
+
+message GetVersionRequest {}
+
+message GetVersionResponse {
+    string version = 1;
+}

api-contracts/openapi/components/schemas/worker.yaml

@@ -76,6 +76,19 @@ WorkerType:
     - MANAGED
     - WEBHOOK
 
+WorkerSlotConfig:
+  type: object
+  description: Slot availability and limits for a slot type.
+  properties:
+    available:
+      type: integer
+      description: The number of available units for this slot type.
+    limit:
+      type: integer
+      description: The maximum number of units for this slot type.
+  required:
+    - limit
+
 RegisteredWorkflow:
   type: object
   properties:
@@ -136,12 +149,11 @@ Worker:
         - ACTIVE
         - INACTIVE
         - PAUSED
-    maxRuns:
-      type: integer
-      description: The maximum number of runs this worker can execute concurrently.
-    availableRuns:
-      type: integer
-      description: The number of runs this worker can execute concurrently.
+    slotConfig:
+      type: object
+      description: Slot availability and limits for this worker (slot_type -> { available, limit }).
+      additionalProperties:
+        $ref: "#/WorkerSlotConfig"
     dispatcherId:
       type: string
       description: "the id of the assigned dispatcher, in UUID format"

api-contracts/openapi/components/schemas/workflow.yaml

@@ -281,6 +281,14 @@ Step:
     timeout:
       type: string
       description: The timeout of the step.
+    isDurable:
+      type: boolean
+      description: Whether the step is durable.
+    slotRequests:
+      type: object
+      description: Slot requests for the step (slot_type -> units).
+      additionalProperties:
+        type: integer
     children:
       type: array
       items:

api-contracts/v1/workflows.proto

@@ -168,6 +168,8 @@ message CreateTaskOpts {
     repeated Concurrency concurrency = 11; // (optional) the task concurrency options
     optional TaskConditions conditions = 12; // (optional) the task conditions for creating the task
     optional string schedule_timeout = 13; // (optional) the timeout for the schedule
+    bool is_durable = 14; // (optional) whether the task is durable
+    map<string, int32> slot_requests = 15; // (optional) slot requests (slot_type -> units)
 }
 
 message CreateTaskRateLimit {

api/v1/server/handlers/v1/filters/create.go

@@ -5,12 +5,13 @@ import (
 	"fmt"
 
 	"github.com/google/uuid"
+	"github.com/labstack/echo/v4"
+
 	"github.com/hatchet-dev/hatchet/api/v1/server/oas/apierrors"
 	"github.com/hatchet-dev/hatchet/api/v1/server/oas/gen"
 	"github.com/hatchet-dev/hatchet/api/v1/server/oas/transformers/v1"
 	v1 "github.com/hatchet-dev/hatchet/pkg/repository"
 	"github.com/hatchet-dev/hatchet/pkg/repository/sqlcv1"
-	"github.com/labstack/echo/v4"
 )
 
 func (t *V1FiltersService) V1FilterCreate(ctx echo.Context, request gen.V1FilterCreateRequestObject) (gen.V1FilterCreateResponseObject, error) {