diff --git a/CHANGELOG.md b/CHANGELOG.md index 530daa4c5f..b4e5239223 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ### Highlights +- Tasks can declare a slot cost, so a task that needs more memory or CPU consumes more than one worker slot. See [Task Slot Cost](https://docs.hatchet.run/v1/advanced-assignment/slot-cost). - Fixes timeouts in `UserSession` cleanup jobs by adding an index via a new `hatchet-migrate` migration. ## [0.90.13] - 2026-06-29 diff --git a/examples/go/slot-cost/main.go b/examples/go/slot-cost/main.go new file mode 100644 index 0000000000..f487c70cbb --- /dev/null +++ b/examples/go/slot-cost/main.go @@ -0,0 +1,42 @@ +package main + +import ( + "log" + + "github.com/hatchet-dev/hatchet/pkg/cmdutils" + hatchet "github.com/hatchet-dev/hatchet/sdks/go" +) + +func main() { + client, err := hatchet.NewClient() + if err != nil { + log.Fatalf("failed to create hatchet client: %v", err) + } + + // > Slot cost + omega := client.NewStandaloneTask("omega", func(ctx hatchet.Context, input any) (any, error) { + log.Println("heavy work") + return nil, nil + }, hatchet.WithSlotCost(5)) + + weenie := client.NewStandaloneTask("weenie", func(ctx hatchet.Context, input any) (any, error) { + log.Println("light work") + return nil, nil + }, hatchet.WithSlotCost(1)) + + worker, err := client.NewWorker("slot-cost-worker", + hatchet.WithWorkflows(omega, weenie), + hatchet.WithSlots(10), + ) + if err != nil { + log.Fatalf("failed to create worker: %v", err) + } + + interruptCtx, cancel := cmdutils.NewInterruptContext() + defer cancel() + + log.Println("Starting slot cost worker...") + if err := worker.StartBlocking(interruptCtx); err != nil { + log.Fatalf("failed to start worker: %v", err) + } +} diff --git a/examples/python/slot_cost/worker.py b/examples/python/slot_cost/worker.py new file mode 100644 index 0000000000..0abb42789b --- /dev/null +++ b/examples/python/slot_cost/worker.py @@ -0,0 +1,26 @@ +from hatchet_sdk import Context, EmptyModel, Hatchet + +hatchet = Hatchet() + +# > Slot cost + + +@hatchet.task(slot_cost=5) +def omega(input: EmptyModel, ctx: Context) -> None: + print("heavy work") + + +@hatchet.task(slot_cost=1) +def weenie(input: EmptyModel, ctx: Context) -> None: + print("light work") + + + + +def main() -> None: + worker = hatchet.worker("slot-cost-worker", workflows=[omega, weenie]) + worker.start() + + +if __name__ == "__main__": + main() diff --git a/examples/typescript/slot_cost/workflow.ts b/examples/typescript/slot_cost/workflow.ts new file mode 100644 index 0000000000..b528c7b327 --- /dev/null +++ b/examples/typescript/slot_cost/workflow.ts @@ -0,0 +1,19 @@ +// > Slot cost +import { hatchet } from '../hatchet-client'; + +export const omega = hatchet.task({ + name: 'omega', + slotCost: 5, + fn: async () => { + console.log('heavy work'); + }, +}); + +export const weenie = hatchet.task({ + name: 'weenie', + slotCost: 1, + fn: async () => { + console.log('light work'); + }, +}); + diff --git a/frontend/docs/pages/reference/changelog/python.mdx b/frontend/docs/pages/reference/changelog/python.mdx index f8eaae9865..cc4d119fb0 100644 --- a/frontend/docs/pages/reference/changelog/python.mdx +++ b/frontend/docs/pages/reference/changelog/python.mdx @@ -1,5 +1,11 @@ {/* AUTOGENERATED — do not edit. Run `task sync-changelog` to regenerate from sdks/python/CHANGELOG.md */} +## v1.34.0 - 2026-07-09 + +### Added + +- Added `slot_cost` to the `hatchet.task` and `workflow.task` decorators, so a task that needs more memory or CPU can consume more than one worker slot and a worker runs fewer of them at once. On older engines it has no effect. See [Task Slot Cost](https://docs.hatchet.run/v1/advanced-assignment/slot-cost). + ## v1.33.18 - 2026-07-08 ### Fixed diff --git a/frontend/docs/pages/reference/changelog/typescript.mdx b/frontend/docs/pages/reference/changelog/typescript.mdx index d6c8ac4ee8..c3b0edb47e 100644 --- a/frontend/docs/pages/reference/changelog/typescript.mdx +++ b/frontend/docs/pages/reference/changelog/typescript.mdx @@ -1,5 +1,11 @@ {/* AUTOGENERATED — do not edit. Run `task sync-changelog` to regenerate from sdks/typescript/CHANGELOG.md */} +## v1.25.0 - 2026-07-09 + +### Added + +- Added `slotCost` to task options, so a task that needs more memory or CPU can consume more than one worker slot and a worker runs fewer of them at once. Durable tasks do not accept it, and on older engines it has no effect. See [Task Slot Cost](https://docs.hatchet.run/v1/advanced-assignment/slot-cost). + ## v1.24.3 - 2026-06-17 ### Removed diff --git a/frontend/docs/pages/v1/advanced-assignment/_meta.js b/frontend/docs/pages/v1/advanced-assignment/_meta.js index 6f279b34c7..21b8e16f07 100644 --- a/frontend/docs/pages/v1/advanced-assignment/_meta.js +++ b/frontend/docs/pages/v1/advanced-assignment/_meta.js @@ -2,5 +2,6 @@ export default { index: { display: "hidden" }, "sticky-assignment": "Sticky Assignment", "worker-affinity": "Worker Affinity", + "slot-cost": "Task Slot Cost", "manual-slot-release": "Manual Slot Release", }; diff --git a/frontend/docs/pages/v1/advanced-assignment/index.mdx b/frontend/docs/pages/v1/advanced-assignment/index.mdx index 736a40af64..93a90065f0 100644 --- a/frontend/docs/pages/v1/advanced-assignment/index.mdx +++ b/frontend/docs/pages/v1/advanced-assignment/index.mdx @@ -1,6 +1,6 @@ --- title: Advanced Assignment -description: Sticky assignment, worker affinity, and manual slot release. +description: Sticky assignment, worker affinity, task slot cost, and manual slot release. --- # Advanced Assignment @@ -9,4 +9,5 @@ Fine-tune how tasks are assigned to workers. - [Sticky Assignment](/v1/advanced-assignment/sticky-assignment) — Pin tasks to specific workers - [Worker Affinity](/v1/advanced-assignment/worker-affinity) — Prefer specific workers +- [Task Slot Cost](/v1/advanced-assignment/slot-cost) — Set how many worker slots a task consumes - [Manual Slot Release](/v1/advanced-assignment/manual-slot-release) — Control when slots free up diff --git a/frontend/docs/pages/v1/advanced-assignment/slot-cost.mdx b/frontend/docs/pages/v1/advanced-assignment/slot-cost.mdx new file mode 100644 index 0000000000..a0bac39b4e --- /dev/null +++ b/frontend/docs/pages/v1/advanced-assignment/slot-cost.mdx @@ -0,0 +1,44 @@ +import { snippets } from "@/lib/generated/snippets"; +import { Snippet } from "@/components/code"; +import { Callout, Tabs } from "nextra/components"; +import UniversalTabs from "@/components/UniversalTabs"; + +# Task Slot Cost + +Every worker has a fixed number of slots that limit how many tasks it runs at once, set with the `slots` option on the worker and defaulting to 100. By default a task consumes one slot while it runs. Slot cost lets a task consume more than one slot, so a task that needs more memory or CPU takes up more of a worker's capacity and the worker runs fewer of them at the same time. + +## Setting slot cost + +Set the slot cost when you define a task. A cost of 5 means the task reserves five slots while it runs. + + + + + + + + + + + + + +On a worker with 100 slots, the omega task at cost 5 and the weenie task at cost 1 draw from the same 100 slots. The worker runs at most 20 omega tasks, or 100 weenie tasks, or any mix whose costs sum to 100. + +## How the reservation works + +Slot capacity is local to a single worker. A task's slot cost is charged against the one worker that runs it, and a reservation cannot span two workers. A task with cost 5 needs a single worker with 5 free slots. It cannot take 3 slots from one worker and 2 from another, even when the worker pool has capacity in total. + +Set the slot count on each worker to at least the largest slot cost you use. If a task's cost is greater than every worker's slot count, the task can never be scheduled. It waits in the queue until its [schedule timeout](../timeouts.mdx) and is then cancelled. + + + Changing a task's slot cost changes its workflow version, so the next + registration creates a new version. Leaving the cost unset, or setting it to + 1, keeps the task at one slot and does not change the version. + + +## Slot cost and concurrency limits + +Slot cost and [concurrency limits](../concurrency.mdx) solve different problems. A concurrency limit caps how many runs of a key execute at once, and with a static key the limit applies across the whole worker pool. Slot cost does not limit the number of runs. It changes how much of a worker one run consumes. + +If you want at most five heavy runs at once, use a concurrency limit. If you want each heavy run to take five times the worker capacity of a light run, use slot cost. The two can be combined on the same task. diff --git a/frontend/docs/pages/v1/concurrency.mdx b/frontend/docs/pages/v1/concurrency.mdx index f3dd0aa862..f3d02f854f 100644 --- a/frontend/docs/pages/v1/concurrency.mdx +++ b/frontend/docs/pages/v1/concurrency.mdx @@ -23,6 +23,13 @@ You should primarily use concurrency control when you need to ensure fair access Concurrency control also lets you limit the number of runs for a workflow globally, if you use a static CEL expression, such as `'global'`. This and [rate limiting](./rate-limits.mdx) are the recommended mechanisms for setting per-workflow throughput limits. + + Concurrency limits how many runs happen at once. To instead control how much + of a worker's capacity a single run consumes, so a heavy task takes up more + slots than a light one, see [Task Slot + Cost](./advanced-assignment/slot-cost.mdx). + + ## Available Strategies: - [**Group Round Robin**](#group-round-robin) queues incoming task and workflow runs and only dispatches them to workers and triggers them once an available slot is open. diff --git a/frontend/docs/pages/v1/faq.mdx b/frontend/docs/pages/v1/faq.mdx index 4bb9bc82bd..58d6cd2cb4 100644 --- a/frontend/docs/pages/v1/faq.mdx +++ b/frontend/docs/pages/v1/faq.mdx @@ -16,6 +16,8 @@ If your workers are resource starved, there are basically two options: 1. Reduce the slot count, so the worker runs less work concurrently. This is a blunt instrument, in the sense that it doesn't let you _tune_ resources to the needs of the workload running on the worker. For instance, if you're using 100% of your memory but only 10% of your CPU, reducing the slot count will likely help the worker stay online, but you'll be significantly under-utilizing CPU. In this case, you can: 2. Reconfigure the specs of the machine the worker is running on. For instance, in the example above, you might be able to migrate from a CPU-optimized machine to a memory-optimized one, which will give you more efficient resource utilization across the board. +If only some of your tasks are heavy, [task slot cost](/v1/advanced-assignment/slot-cost) lets those tasks consume more than one slot, so the slot count can stay tuned for the light ones. + On the other hand, if your workers are underutilizing resources, your options are: 1. Increase the number of slots on them so they can pick up more work. This is especially helpful for heavily I/O bound tasks, which generally are spending most of their time waiting. diff --git a/frontend/docs/pages/v1/troubleshooting/index.mdx b/frontend/docs/pages/v1/troubleshooting/index.mdx index 55f0a9292f..6a8005577a 100644 --- a/frontend/docs/pages/v1/troubleshooting/index.mdx +++ b/frontend/docs/pages/v1/troubleshooting/index.mdx @@ -40,7 +40,7 @@ If tasks remain in the `QUEUED` state and never move to `RUNNING`: 1. **No workers registered for the task** — check the Workers tab in the dashboard and confirm a worker is registered that handles the task name. If you recently renamed a task, make sure the worker has been restarted with the updated code. -2. **All worker slots are full** — if every slot is occupied by other tasks, new tasks will wait in the queue. Check worker utilization in the dashboard or increase the [slot count](/v1/workers#slots). +2. **All worker slots are full** — if every slot is occupied by other tasks, new tasks will wait in the queue. Check worker utilization in the dashboard or increase the [slot count](/v1/workers#slots). A task with a [slot cost](/v1/advanced-assignment/slot-cost) higher than every worker's slot count can never be scheduled, and is cancelled when its schedule timeout is reached. 3. **Concurrency or rate limit is blocking** — if you've configured [concurrency limits](/v1/concurrency) or [rate limits](/v1/rate-limits), tasks may be held back intentionally. Review your configuration. diff --git a/frontend/docs/pages/v1/workers.mdx b/frontend/docs/pages/v1/workers.mdx index 92ac89ab03..eb89658e9c 100644 --- a/frontend/docs/pages/v1/workers.mdx +++ b/frontend/docs/pages/v1/workers.mdx @@ -144,4 +144,6 @@ To set the slot count, pass the `slots` option when declaring the worker: The default slot count for workers in Hatchet is 100. In many cases, leaving the default as-is will be perfectly fine, especially when first getting set up with Hatchet. +By default each running task consumes one slot. A task can be configured to consume more than one, so a task that needs more memory or CPU takes up more of a worker's capacity. See [Task Slot Cost](./advanced-assignment/slot-cost.mdx). + diff --git a/pkg/client/create/task.go b/pkg/client/create/task.go index 7c05ca96b6..4e2c7ca519 100644 --- a/pkg/client/create/task.go +++ b/pkg/client/create/task.go @@ -65,6 +65,10 @@ type WorkflowTask[I, O any] struct { Parents []NamedTask DefaultPriority *int32 + + // (optional) SlotCost is the number of default worker slots this task consumes. Defaults to one. + // Durable tasks ignore it. + SlotCost *int32 } type WorkflowOnFailureTask[I, O any] struct { diff --git a/pkg/repository/slot_cost_writepath_test.go b/pkg/repository/slot_cost_writepath_test.go new file mode 100644 index 0000000000..c1f0f19495 --- /dev/null +++ b/pkg/repository/slot_cost_writepath_test.go @@ -0,0 +1,84 @@ +//go:build !e2e && !load && !rampup && !integration + +package repository + +import ( + "context" + "testing" + + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// The trigger v1_step_slot_request_insert_trigger inserts a one-slot row ({default: 1}, or +// {durable: 1} for a durable step) on Step insert. This checks that a registered slot cost +// overwrites that row and persists. +func TestSlotCostPersistsThroughWritePath(t *testing.T) { + pool, cleanup := setupPostgresWithMigration(t) + defer cleanup() + + ctx := context.Background() + repo := newWorkflowTestRepository(pool) + + // The overwrite only matters while this trigger exists, so require it to be present. + requireTriggerExists(ctx, t, pool, "v1_step_slot_request_insert_trigger") + + desc := "slot-cost-writepath" + opts := &CreateWorkflowVersionOpts{ + Name: "slot-cost-writepath", + Description: &desc, + Tasks: []CreateStepOpts{ + {ReadableId: "heavy", Action: "integration:heavy", SlotRequests: map[string]int32{SlotTypeDefault: 5}}, + {ReadableId: "light", Action: "integration:light", SlotRequests: map[string]int32{SlotTypeDefault: 1}}, + {ReadableId: "plain", Action: "integration:plain"}, + {ReadableId: "dur", Action: "integration:dur", IsDurable: true}, + }, + } + + _, err := repo.PutWorkflowVersion(ctx, internalTenantId, opts) + require.NoError(t, err) + + got := readSlotRequestsByStep(ctx, t, pool) + + assert.Equal(t, map[string]int32{SlotTypeDefault: 5}, got["heavy"], "explicit slot cost 5 should persist") + assert.Equal(t, map[string]int32{SlotTypeDefault: 1}, got["light"], "explicit slot cost 1 should persist") + assert.Equal(t, map[string]int32{SlotTypeDefault: 1}, got["plain"], "no explicit cost stays at one default slot") + assert.Equal(t, map[string]int32{SlotTypeDurable: 1}, got["dur"], "durable default is unchanged") +} + +func requireTriggerExists(ctx context.Context, t *testing.T, pool *pgxpool.Pool, name string) { + t.Helper() + + var exists bool + err := pool.QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM pg_trigger WHERE tgname = $1)`, name).Scan(&exists) + require.NoError(t, err) + require.True(t, exists, "test premise requires the compatibility trigger %q to be present", name) +} + +func readSlotRequestsByStep(ctx context.Context, t *testing.T, pool *pgxpool.Pool) map[string]map[string]int32 { + t.Helper() + + rows, err := pool.Query(ctx, ` + SELECT s."readableId", r.slot_type, r.units + FROM v1_step_slot_request r + JOIN "Step" s ON s.id = r.step_id + WHERE r.tenant_id = $1 + `, internalTenantId) + require.NoError(t, err) + defer rows.Close() + + out := map[string]map[string]int32{} + for rows.Next() { + var readableID, slotType string + var units int32 + require.NoError(t, rows.Scan(&readableID, &slotType, &units)) + if out[readableID] == nil { + out[readableID] = map[string]int32{} + } + out[readableID][slotType] = units + } + require.NoError(t, rows.Err()) + + return out +} diff --git a/pkg/repository/sqlcv1/workflows.sql b/pkg/repository/sqlcv1/workflows.sql index a0bbd6cd44..cee49819a6 100644 --- a/pkg/repository/sqlcv1/workflows.sql +++ b/pkg/repository/sqlcv1/workflows.sql @@ -322,8 +322,11 @@ SELECT unnest(@units::integer[]), CURRENT_TIMESTAMP, CURRENT_TIMESTAMP --- NOTE: ON CONFLICT can be removed after the 0_76_d migration is run to remove insert triggers added in 0_76 -ON CONFLICT (tenant_id, step_id, slot_type) DO NOTHING; +-- The trigger v1_step_slot_request_insert_trigger writes a {default: 1} (or {durable: 1}) row on +-- Step insert, so DO UPDATE overwrites it with the requested units instead of leaving the default. +-- The conflict handling is only here because of that trigger. +ON CONFLICT (tenant_id, step_id, slot_type) DO UPDATE + SET units = EXCLUDED.units, updated_at = CURRENT_TIMESTAMP; -- name: AddStepParents :exec INSERT INTO "_StepOrder" ("A", "B") diff --git a/pkg/repository/sqlcv1/workflows.sql.go b/pkg/repository/sqlcv1/workflows.sql.go index eeec44eaaa..9d7c1e50ed 100644 --- a/pkg/repository/sqlcv1/workflows.sql.go +++ b/pkg/repository/sqlcv1/workflows.sql.go @@ -465,7 +465,8 @@ SELECT unnest($4::integer[]), CURRENT_TIMESTAMP, CURRENT_TIMESTAMP -ON CONFLICT (tenant_id, step_id, slot_type) DO NOTHING +ON CONFLICT (tenant_id, step_id, slot_type) DO UPDATE + SET units = EXCLUDED.units, updated_at = CURRENT_TIMESTAMP ` type CreateStepSlotRequestsParams struct { @@ -475,7 +476,9 @@ type CreateStepSlotRequestsParams struct { Units []int32 `json:"units"` } -// NOTE: ON CONFLICT can be removed after the 0_76_d migration is run to remove insert triggers added in 0_76 +// The trigger v1_step_slot_request_insert_trigger writes a {default: 1} (or {durable: 1}) row on +// Step insert, so DO UPDATE overwrites it with the requested units instead of leaving the default. +// The conflict handling is only here because of that trigger. func (q *Queries) CreateStepSlotRequests(ctx context.Context, db DBTX, arg CreateStepSlotRequestsParams) error { _, err := db.Exec(ctx, createStepSlotRequests, arg.Tenantid, diff --git a/pkg/scheduling/v1/slot_cost_test.go b/pkg/scheduling/v1/slot_cost_test.go new file mode 100644 index 0000000000..730dd882bc --- /dev/null +++ b/pkg/scheduling/v1/slot_cost_test.go @@ -0,0 +1,154 @@ +//go:build !e2e && !load && !rampup && !integration + +package v1 + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" + + repo "github.com/hatchet-dev/hatchet/pkg/repository" + "github.com/hatchet-dev/hatchet/pkg/repository/sqlcv1" +) + +// These tests use the scheduler's internal slot request map ({default: N}) directly, not the +// public SDK parameter. + +func defaultSlots(w *worker, n int) []*slot { + slots := make([]*slot, n) + for i := range slots { + slots[i] = newSlot(w, newSlotMeta([]string{"A"}, repo.SlotTypeDefault)) + } + return slots +} + +func TestSlotCost_MixedHeavyAndLightShareDefaultPool(t *testing.T) { + workerId := uuid.New() + w := &worker{ListActiveWorkersResult: testWorker(workerId)} + + a, err := actionWithSlots("A", defaultSlots(w, 6)...) + require.NoError(t, err) + + heavy := findAssignableSlots(a.slots, a, map[string]int32{repo.SlotTypeDefault: 5}, nil, nil) + require.NotNil(t, heavy) + require.Len(t, heavy.slots, 5) + + light := findAssignableSlots(a.slots, a, map[string]int32{repo.SlotTypeDefault: 1}, nil, nil) + require.NotNil(t, light) + require.Len(t, light.slots, 1) + + none := findAssignableSlots(a.slots, a, map[string]int32{repo.SlotTypeDefault: 1}, nil, nil) + require.Nil(t, none) +} + +func TestSlotCost_ReservationMustFitOnOneWorker(t *testing.T) { + w1 := &worker{ListActiveWorkersResult: testWorker(uuid.New())} + w2 := &worker{ListActiveWorkersResult: testWorker(uuid.New())} + + all := append(defaultSlots(w1, 4), defaultSlots(w2, 4)...) + a, err := actionWithSlots("A", all...) + require.NoError(t, err) + + none := findAssignableSlots(a.slots, a, map[string]int32{repo.SlotTypeDefault: 5}, nil, nil) + require.Nil(t, none) + + fits := findAssignableSlots(a.slots, a, map[string]int32{repo.SlotTypeDefault: 4}, nil, nil) + require.NotNil(t, fits) + require.Len(t, fits.slots, 4) +} + +// An over-capacity task is unassigned only while inside its schedule timeout. Past the timeout the +// scheduler routes it to schedulingTimedOut, which the engine cancels with reason +// SCHEDULING_TIMED_OUT, so the wait is bounded. +func TestSlotCost_OverCapacityWaitsThenSchedulingTimesOut(t *testing.T) { + tenantId := uuid.New() + workerId := uuid.New() + + s := newTestScheduler(t, tenantId, &mockAssignmentRepo{}) + w := &worker{ListActiveWorkersResult: testWorker(workerId)} + + a, err := actionWithSlots("A", defaultSlots(w, 4)...) + require.NoError(t, err) + s.actions["A"] = a + + waiting := testQI(tenantId, "A", 1) + waiting.ScheduleTimeoutAt = ts(time.Now().UTC().Add(5 * time.Minute)) + + expired := testQI(tenantId, "A", 2) + expired.ScheduleTimeoutAt = ts(time.Now().UTC().Add(-1 * time.Second)) + + stepRequests := map[uuid.UUID]map[string]int32{ + waiting.StepID: {repo.SlotTypeDefault: 5}, + expired.StepID: {repo.SlotTypeDefault: 5}, + } + + ch := s.tryAssign( + context.Background(), + []*sqlcv1.V1QueueItem{waiting, expired}, + map[uuid.UUID][]*sqlcv1.GetDesiredLabelsRow{}, + stepRequests, + nil, + nil, + ) + + assigned := map[int64]bool{} + unassigned := map[int64]bool{} + timedOut := map[int64]bool{} + + for r := range ch { + for _, as := range r.assigned { + assigned[as.QueueItem.TaskID] = true + } + for _, u := range r.unassigned { + unassigned[u.TaskID] = true + } + for _, to := range r.schedulingTimedOut { + timedOut[to.TaskID] = true + } + } + + require.Empty(t, assigned) + require.True(t, unassigned[waiting.TaskID]) + require.False(t, timedOut[waiting.TaskID]) + require.True(t, timedOut[expired.TaskID]) +} + +func TestSlotCost_ExplicitDefaultCostBlocksProportionally(t *testing.T) { + tenantId := uuid.New() + workerId := uuid.New() + + s := newTestScheduler(t, tenantId, &mockAssignmentRepo{}) + w := &worker{ListActiveWorkersResult: testWorker(workerId)} + + a, err := actionWithSlots("A", defaultSlots(w, 2)...) + require.NoError(t, err) + s.actions["A"] = a + + qi1 := testQI(tenantId, "A", 1) + qi2 := testQI(tenantId, "A", 2) + qis := []*sqlcv1.V1QueueItem{qi1, qi2} + + stepRequests := map[uuid.UUID]map[string]int32{ + qi1.StepID: {repo.SlotTypeDefault: 2}, + qi2.StepID: {repo.SlotTypeDefault: 2}, + } + + res, _, err := s.tryAssignBatch(context.Background(), "A", qis, 0, + map[uuid.UUID][]*sqlcv1.GetDesiredLabelsRow{}, stepRequests, nil, nil) + require.NoError(t, err) + + assigned, noSlots := 0, 0 + for _, r := range res { + if r.succeeded { + assigned++ + } + if r.noSlots { + noSlots++ + } + } + require.Equal(t, 1, assigned) + require.Equal(t, 1, noSlots) +} diff --git a/sdks/go/examples/slot-cost/main.go b/sdks/go/examples/slot-cost/main.go new file mode 100644 index 0000000000..ccac0b18dc --- /dev/null +++ b/sdks/go/examples/slot-cost/main.go @@ -0,0 +1,43 @@ +package main + +import ( + "log" + + "github.com/hatchet-dev/hatchet/pkg/cmdutils" + hatchet "github.com/hatchet-dev/hatchet/sdks/go" +) + +func main() { + client, err := hatchet.NewClient() + if err != nil { + log.Fatalf("failed to create hatchet client: %v", err) + } + + // > Slot cost + omega := client.NewStandaloneTask("omega", func(ctx hatchet.Context, input any) (any, error) { + log.Println("heavy work") + return nil, nil + }, hatchet.WithSlotCost(5)) + + weenie := client.NewStandaloneTask("weenie", func(ctx hatchet.Context, input any) (any, error) { + log.Println("light work") + return nil, nil + }, hatchet.WithSlotCost(1)) + // !! + + worker, err := client.NewWorker("slot-cost-worker", + hatchet.WithWorkflows(omega, weenie), + hatchet.WithSlots(10), + ) + if err != nil { + log.Fatalf("failed to create worker: %v", err) + } + + interruptCtx, cancel := cmdutils.NewInterruptContext() + defer cancel() + + log.Println("Starting slot cost worker...") + if err := worker.StartBlocking(interruptCtx); err != nil { + log.Fatalf("failed to start worker: %v", err) + } +} diff --git a/sdks/go/internal/declaration.go b/sdks/go/internal/declaration.go index 0665298bcf..cbe20771c3 100644 --- a/sdks/go/internal/declaration.go +++ b/sdks/go/internal/declaration.go @@ -316,6 +316,7 @@ func (w *workflowDeclarationImpl[I, O]) Task(opts create.WorkflowTask[I, O], fn RateLimits: opts.RateLimits, WorkerLabels: opts.WorkerLabels, Concurrency: opts.Concurrency, + SlotCost: opts.SlotCost, }, } diff --git a/sdks/go/internal/task/task.go b/sdks/go/internal/task/task.go index 852cac4a58..d264e6b741 100644 --- a/sdks/go/internal/task/task.go +++ b/sdks/go/internal/task/task.go @@ -56,6 +56,10 @@ type TaskShared struct { // Concurrency defines constraints on how many instances of this task can run simultaneously Concurrency []*types.Concurrency + // SlotCost is the number of default worker slots a non-durable task consumes. Defaults to one. + // Durable tasks ignore it. + SlotCost *int32 + // The function to execute when the task runs // must be a function that takes an input and a Hatchet context and returns an output and an error Fn interface{} @@ -249,7 +253,11 @@ func (t *TaskDeclaration[I]) Dump(workflowName string, taskDefaults *create.Task base.Action = getActionID(workflowName, t.Name) base.IsDurable = false if base.SlotRequests == nil { - base.SlotRequests = map[string]int32{slotTypeDefault: 1} + units := int32(1) + if t.SlotCost != nil { + units = *t.SlotCost + } + base.SlotRequests = map[string]int32{slotTypeDefault: units} } base.Parents = make([]string, len(t.Parents)) copy(base.Parents, t.Parents) diff --git a/sdks/go/slot_cost_test.go b/sdks/go/slot_cost_test.go new file mode 100644 index 0000000000..8fa4c09455 --- /dev/null +++ b/sdks/go/slot_cost_test.go @@ -0,0 +1,63 @@ +//go:build !e2e && !load && !rampup && !integration + +package hatchet + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// These offline tests dump the registration request to check what WithSlotCost puts in it. + +func TestWithSlotCost_MapsToDefaultPool(t *testing.T) { + c := newTestClient() + task := c.NewStandaloneTask("heavy", sampleTaskFn, WithSlotCost(5)) + + req, _, _, _ := task.Dump() + + require.Len(t, req.Tasks, 1) + assert.Equal(t, map[string]int32{"default": 5}, req.Tasks[0].SlotRequests) +} + +func TestWithSlotCost_OmittedKeepsOneDefaultSlot(t *testing.T) { + c := newTestClient() + task := c.NewStandaloneTask("plain", sampleTaskFn) + + req, _, _, _ := task.Dump() + + require.Len(t, req.Tasks, 1) + assert.Equal(t, map[string]int32{"default": 1}, req.Tasks[0].SlotRequests) +} + +func TestWithSlotCost_OneIsAccepted(t *testing.T) { + c := newTestClient() + task := c.NewStandaloneTask("one", sampleTaskFn, WithSlotCost(1)) + + req, _, _, _ := task.Dump() + + require.Len(t, req.Tasks, 1) + assert.Equal(t, map[string]int32{"default": 1}, req.Tasks[0].SlotRequests) +} + +func TestWithSlotCost_RejectsNonPositive(t *testing.T) { + c := newTestClient() + + assert.Panics(t, func() { + c.NewStandaloneTask("zero", sampleTaskFn, WithSlotCost(0)) + }) + assert.Panics(t, func() { + c.NewStandaloneTask("negative", sampleTaskFn, WithSlotCost(-2)) + }) +} + +func TestWithSlotCost_DurableTaskUnchanged(t *testing.T) { + c := newTestClient() + task := c.NewStandaloneDurableTask("durable", sampleDurableFn) + + req, _, _, _ := task.Dump() + + require.Len(t, req.Tasks, 1) + assert.Equal(t, map[string]int32{"durable": 1}, req.Tasks[0].SlotRequests) +} diff --git a/sdks/go/workflow.go b/sdks/go/workflow.go index c018f95195..8d253de5fc 100644 --- a/sdks/go/workflow.go +++ b/sdks/go/workflow.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "math" "reflect" "sync" "time" @@ -294,6 +295,7 @@ type taskConfig struct { skipIf condition.Condition description string evictionPolicy *EvictionPolicy + slotCost *int32 } // WithRetries sets the number of retry attempts for failed tasks. @@ -311,6 +313,22 @@ func WithRetryBackoff(factor float32, maxBackoffSeconds int) TaskOption { } } +// WithSlotCost sets the number of default worker slots this task consumes. A normal task consumes +// one. Set it higher for a task that needs more memory or CPU, so a worker runs fewer of them at +// once. A single worker must have that many free slots to run it. Durable tasks ignore it. Panics +// if cost is not positive. +func WithSlotCost(cost int) TaskOption { + if cost <= 0 || cost > math.MaxInt32 { + panic("slot cost must be a positive integer") + } + + c := int32(cost) + + return func(config *taskConfig) { + config.slotCost = &c + } +} + // WithScheduleTimeout sets the maximum time a task can wait to be scheduled. func WithScheduleTimeout(timeout time.Duration) TaskOption { return func(config *taskConfig) { @@ -511,6 +529,7 @@ func (w *Workflow) NewTask(name string, fn any, options ...TaskOption) *Task { Parents: config.parents, WaitFor: config.waitFor, SkipIf: config.skipIf, + SlotCost: config.slotCost, } if config.isDurable { diff --git a/sdks/python/CHANGELOG.md b/sdks/python/CHANGELOG.md index cb00f1173d..0fc0c342f7 100644 --- a/sdks/python/CHANGELOG.md +++ b/sdks/python/CHANGELOG.md @@ -5,6 +5,12 @@ All notable changes to Hatchet's Python SDK will be documented in this changelog The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [1.34.0] - 2026-07-09 + +### Added + +- Added `slot_cost` to the `hatchet.task` and `workflow.task` decorators, so a task that needs more memory or CPU can consume more than one worker slot and a worker runs fewer of them at once. On older engines it has no effect. See [Task Slot Cost](https://docs.hatchet.run/v1/advanced-assignment/slot-cost). + ## [1.33.18] - 2026-07-08 ### Fixed diff --git a/sdks/python/examples/slot_cost/worker.py b/sdks/python/examples/slot_cost/worker.py new file mode 100644 index 0000000000..b2a2ef8708 --- /dev/null +++ b/sdks/python/examples/slot_cost/worker.py @@ -0,0 +1,27 @@ +from hatchet_sdk import Context, EmptyModel, Hatchet + +hatchet = Hatchet() + +# > Slot cost + + +@hatchet.task(slot_cost=5) +def omega(input: EmptyModel, ctx: Context) -> None: + print("heavy work") + + +@hatchet.task(slot_cost=1) +def weenie(input: EmptyModel, ctx: Context) -> None: + print("light work") + + +# !! + + +def main() -> None: + worker = hatchet.worker("slot-cost-worker", workflows=[omega, weenie]) + worker.start() + + +if __name__ == "__main__": + main() diff --git a/sdks/python/hatchet_sdk/hatchet.py b/sdks/python/hatchet_sdk/hatchet.py index fd67366af0..b7ef44436a 100644 --- a/sdks/python/hatchet_sdk/hatchet.py +++ b/sdks/python/hatchet_sdk/hatchet.py @@ -394,6 +394,7 @@ def task( backoff_max_seconds: int | None = None, default_filters: list[DefaultFilter] | None = None, default_additional_metadata: JSONSerializableMapping | None = None, + slot_cost: int | None = None, ) -> Callable[ [Callable[Concatenate[EmptyModel, Context, P], R | CoroutineLike[R]]], Standalone[EmptyModel, R], @@ -425,6 +426,7 @@ def task( backoff_max_seconds: int | None = None, default_filters: list[DefaultFilter] | None = None, default_additional_metadata: JSONSerializableMapping | None = None, + slot_cost: int | None = None, ) -> Callable[ [Callable[Concatenate[TWorkflowInput, Context, P], R | CoroutineLike[R]]], Standalone[TWorkflowInput, R], @@ -455,6 +457,7 @@ def task( backoff_max_seconds: int | None = None, default_filters: list[DefaultFilter] | None = None, default_additional_metadata: JSONSerializableMapping | None = None, + slot_cost: int | None = None, ) -> ( Callable[ [Callable[Concatenate[EmptyModel, Context, P], R | CoroutineLike[R]]], @@ -504,6 +507,8 @@ def task( :param default_additional_metadata: A dictionary of additional metadata to attach to each run of this task by default. + :param slot_cost: The number of default worker slots this task consumes. A normal task consumes one. Set it higher for a task that needs more memory or CPU, so a worker runs fewer of them at once. A single worker must have that many free slots to run it. + :returns: A decorator which creates a `Standalone` task object. """ @@ -552,6 +557,7 @@ def inner( backoff_factor=backoff_factor, backoff_max_seconds=backoff_max_seconds, concurrency=_concurrency, + slot_cost=slot_cost, ) created_task = task_wrapper(func) diff --git a/sdks/python/hatchet_sdk/runnables/workflow.py b/sdks/python/hatchet_sdk/runnables/workflow.py index 56966f5faf..7d2fc0b27a 100644 --- a/sdks/python/hatchet_sdk/runnables/workflow.py +++ b/sdks/python/hatchet_sdk/runnables/workflow.py @@ -1351,6 +1351,7 @@ def task( wait_for: list[Condition | OrGroup] | None = None, skip_if: list[Condition | OrGroup] | None = None, cancel_if: list[Condition | OrGroup] | None = None, + slot_cost: int | None = None, ) -> Callable[ [Callable[Concatenate[TWorkflowInput, Context, P], R | CoroutineLike[R]]], Task[TWorkflowInput, R], @@ -1384,11 +1385,20 @@ def task( :param cancel_if: A list of conditions that, if met, will cause the task to be canceled. + :param slot_cost: The number of default worker slots this task consumes. A normal task consumes one. Set it higher for a task that needs more memory or CPU, so a worker runs fewer of them at once. A single worker must have that many free slots to run it. + :returns: A decorator which creates a `Task` object. + + :raises ValueError: If `slot_cost` is not positive. """ _warn_if_str_duration(schedule_timeout, execution_timeout) + if slot_cost is not None and slot_cost <= 0: + raise ValueError("slot_cost must be a positive integer") + + slot_requests = {"default": slot_cost} if slot_cost is not None else None + computed_params = ComputedTaskParameters( schedule_timeout=schedule_timeout, execution_timeout=execution_timeout, @@ -1431,6 +1441,7 @@ def inner( wait_for=wait_for, skip_if=skip_if, cancel_if=cancel_if, + slot_requests=slot_requests, ) self._default_tasks.append(task) diff --git a/sdks/python/pyproject.toml b/sdks/python/pyproject.toml index 1c75e34c7d..885c15bf22 100644 --- a/sdks/python/pyproject.toml +++ b/sdks/python/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "hatchet-sdk" -version = "1.33.18" +version = "1.34.0" description = "This is the official Python SDK for Hatchet, a distributed, fault-tolerant task queue. The SDK allows you to easily integrate Hatchet's task scheduling and workflow orchestration capabilities into your Python applications." readme = "README.md" license = { text = "MIT" } diff --git a/sdks/python/tests/test_slot_cost.py b/sdks/python/tests/test_slot_cost.py new file mode 100644 index 0000000000..7e4b208618 --- /dev/null +++ b/sdks/python/tests/test_slot_cost.py @@ -0,0 +1,88 @@ +"""Unit tests for the public slot_cost task parameter.""" + +import base64 +import json +from typing import Any + +import pytest + +from hatchet_sdk import Context, DurableContext, EmptyModel, Hatchet + + +@pytest.fixture(scope="session", autouse=True) +def worker() -> Any: + # These tests run offline, so override conftest's engine-backed worker fixture. + yield None + + +def _offline_token() -> str: + # A well-formed JWT so ClientConfig can construct offline. The client reads its claims and does + # not check the signature. + def segment(data: dict[str, str]) -> str: + return base64.urlsafe_b64encode(json.dumps(data).encode()).rstrip(b"=").decode() + + header = segment({"alg": "none", "typ": "JWT"}) + payload = segment( + { + "sub": "tenant-offline", + "server_url": "https://localhost", + "grpc_broadcast_address": "localhost:7070", + } + ) + return f"{header}.{payload}.signature" + + +def dummy(input: EmptyModel, ctx: Context) -> dict[str, str]: + return {"foo": "bar"} + + +async def dummy_durable(input: EmptyModel, ctx: DurableContext) -> dict[str, str]: + return {"foo": "bar"} + + +@pytest.fixture +def hatchet(monkeypatch: pytest.MonkeyPatch) -> Hatchet: + monkeypatch.setenv("HATCHET_CLIENT_TOKEN", _offline_token()) + return Hatchet() + + +def test_slot_cost_maps_to_default_pool(hatchet: Hatchet) -> None: + wf = hatchet.workflow(name="slot-cost-wf") + t = wf.task(slot_cost=5)(dummy) + + assert t.to_proto("svc").slot_requests == {"default": 5} + + +def test_omitting_slot_cost_keeps_one_default_slot(hatchet: Hatchet) -> None: + wf = hatchet.workflow(name="slot-cost-wf") + t = wf.task()(dummy) + + assert t.to_proto("svc").slot_requests == {"default": 1} + + +def test_slot_cost_one_is_accepted(hatchet: Hatchet) -> None: + wf = hatchet.workflow(name="slot-cost-wf") + t = wf.task(slot_cost=1)(dummy) + + assert t.to_proto("svc").slot_requests == {"default": 1} + + +@pytest.mark.parametrize("bad", [0, -1, -5]) +def test_non_positive_slot_cost_is_rejected(hatchet: Hatchet, bad: int) -> None: + wf = hatchet.workflow(name="slot-cost-wf") + + with pytest.raises(ValueError): + wf.task(slot_cost=bad)(dummy) + + +def test_standalone_task_slot_cost(hatchet: Hatchet) -> None: + standalone = hatchet.task(name="standalone-heavy", slot_cost=5)(dummy) + + assert standalone._task.to_proto("svc").slot_requests == {"default": 5} + + +def test_durable_task_is_unchanged(hatchet: Hatchet) -> None: + wf = hatchet.workflow(name="slot-cost-wf") + t = wf.durable_task()(dummy_durable) + + assert t.to_proto("svc").slot_requests == {"durable": 1} diff --git a/sdks/typescript/CHANGELOG.md b/sdks/typescript/CHANGELOG.md index 55d479d983..399d34664a 100644 --- a/sdks/typescript/CHANGELOG.md +++ b/sdks/typescript/CHANGELOG.md @@ -5,6 +5,12 @@ All notable changes to Hatchet's TypeScript SDK will be documented in this chang The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [1.25.0] - 2026-07-09 + +### Added + +- Added `slotCost` to task options, so a task that needs more memory or CPU can consume more than one worker slot and a worker runs fewer of them at once. Durable tasks do not accept it, and on older engines it has no effect. See [Task Slot Cost](https://docs.hatchet.run/v1/advanced-assignment/slot-cost). + ## [1.24.3] - 2026-06-17 ### Removed diff --git a/sdks/typescript/package.json b/sdks/typescript/package.json index e970dd0944..c4a93fba25 100644 --- a/sdks/typescript/package.json +++ b/sdks/typescript/package.json @@ -1,6 +1,6 @@ { "name": "@hatchet-dev/typescript-sdk", - "version": "1.24.3", + "version": "1.25.0", "description": "Background task orchestration & visibility for developers", "types": "dist/index.d.ts", "files": [ diff --git a/sdks/typescript/src/v1/client/worker/slot-cost.test.ts b/sdks/typescript/src/v1/client/worker/slot-cost.test.ts new file mode 100644 index 0000000000..104ae63df3 --- /dev/null +++ b/sdks/typescript/src/v1/client/worker/slot-cost.test.ts @@ -0,0 +1,46 @@ +import { WorkflowDeclaration } from '../../declaration'; +import { mapSlotRequestsPb } from './worker-internal'; + +// Never called. It exists so tsc checks that durable task options reject slotCost; if the +// omission regresses, the @ts-expect-error goes unused and tsc fails. +export function durableTaskRejectsSlotCost() { + const wf = new WorkflowDeclaration({ name: 'slot-cost-type-check' }); + wf.durableTask({ + name: 'd', + // @ts-expect-error slotCost is not available on durable tasks + slotCost: 1, + fn: async () => undefined, + }); +} + +describe('mapSlotRequestsPb', () => { + it('maps a slotCost to a request against the default pool', () => { + expect(mapSlotRequestsPb({ slotCost: 5 }, false)).toEqual({ default: 5 }); + }); + + it('defaults to one default slot when slotCost is omitted', () => { + expect(mapSlotRequestsPb({}, false)).toEqual({ default: 1 }); + }); + + it('accepts a slotCost of 1', () => { + expect(mapSlotRequestsPb({ slotCost: 1 }, false)).toEqual({ default: 1 }); + }); + + it('rejects a slotCost of 0 or a negative slotCost', () => { + expect(() => mapSlotRequestsPb({ slotCost: 0 }, false)).toThrow(/positive integer/); + expect(() => mapSlotRequestsPb({ slotCost: -3 }, false)).toThrow(/positive integer/); + }); + + it('rejects a non-integer slotCost', () => { + expect(() => mapSlotRequestsPb({ slotCost: 2.5 }, false)).toThrow(/positive integer/); + }); + + it('keeps durable tasks on the durable pool and does not apply slotCost', () => { + expect(mapSlotRequestsPb({}, true)).toEqual({ durable: 1 }); + expect(mapSlotRequestsPb({ slotCost: 5 }, true)).toEqual({ durable: 1 }); + }); + + it('honors an explicit internal slotRequests map when present', () => { + expect(mapSlotRequestsPb({ slotRequests: { default: 3 } }, false)).toEqual({ default: 3 }); + }); +}); diff --git a/sdks/typescript/src/v1/client/worker/worker-internal.ts b/sdks/typescript/src/v1/client/worker/worker-internal.ts index ffa409fc51..3217e411f8 100644 --- a/sdks/typescript/src/v1/client/worker/worker-internal.ts +++ b/sdks/typescript/src/v1/client/worker/worker-internal.ts @@ -277,7 +277,7 @@ export class InternalWorker { backoffMaxSeconds: onFailure.backoff?.maxSeconds || workflow.taskDefaults?.backoff?.maxSeconds, isDurable: false, - slotRequests: { default: 1 }, + slotRequests: mapSlotRequestsPb(onFailure, false), }; } @@ -402,8 +402,7 @@ export class InternalWorker { backoffMaxSeconds: task.backoff?.maxSeconds || workflow.taskDefaults?.backoff?.maxSeconds, conditions: taskConditionsToPb(task, this.client.config.namespace), isDurable: durableTaskSet.has(task), - slotRequests: - task.slotRequests || (durableTaskSet.has(task) ? { durable: 1 } : { default: 1 }), + slotRequests: mapSlotRequestsPb(task, durableTaskSet.has(task)), concurrency: task.concurrency ? Array.isArray(task.concurrency) ? task.concurrency @@ -1077,6 +1076,30 @@ function isLeafTask(task: LeafableTask, allTasks: LeafableTask[]): boolean { return !allTasks.some((t) => t.parents?.some((p) => p.name === task.name)); } +/** Durable tasks stay on the durable pool; slotCost applies only to the default pool. */ +export function mapSlotRequestsPb( + task: { slotRequests?: Record; slotCost?: number }, + isDurable: boolean +): Record { + if (task.slotRequests) { + return task.slotRequests; + } + + if (isDurable) { + return { durable: 1 }; + } + + if (task.slotCost !== undefined) { + if (!Number.isInteger(task.slotCost) || task.slotCost <= 0) { + throw new Error(`slotCost must be a positive integer, got: ${task.slotCost}`); + } + + return { default: task.slotCost }; + } + + return { default: 1 }; +} + export function mapRateLimitPb( limits: CreateWorkflowTaskOpts['rateLimits'] ): CreateStepRateLimit[] { diff --git a/sdks/typescript/src/v1/declaration.ts b/sdks/typescript/src/v1/declaration.ts index 24d19d507e..b705290e93 100644 --- a/sdks/typescript/src/v1/declaration.ts +++ b/sdks/typescript/src/v1/declaration.ts @@ -1032,7 +1032,7 @@ export class WorkflowDeclaration< ? FnReturn : never, >( - options: Omit, 'fn'> & { + options: Omit, 'fn' | 'slotCost'> & { name: Name; fn: Fn; } diff --git a/sdks/typescript/src/v1/examples/slot_cost/workflow.ts b/sdks/typescript/src/v1/examples/slot_cost/workflow.ts new file mode 100644 index 0000000000..9f915a86d0 --- /dev/null +++ b/sdks/typescript/src/v1/examples/slot_cost/workflow.ts @@ -0,0 +1,20 @@ +// > Slot cost +import { hatchet } from '../hatchet-client'; + +export const omega = hatchet.task({ + name: 'omega', + slotCost: 5, + fn: async () => { + console.log('heavy work'); + }, +}); + +export const weenie = hatchet.task({ + name: 'weenie', + slotCost: 1, + fn: async () => { + console.log('light work'); + }, +}); + +// !! diff --git a/sdks/typescript/src/v1/task.ts b/sdks/typescript/src/v1/task.ts index 86c7695ebc..1920665d5f 100644 --- a/sdks/typescript/src/v1/task.ts +++ b/sdks/typescript/src/v1/task.ts @@ -159,6 +159,17 @@ export type CreateBaseTaskOpts< */ concurrency?: Concurrency | Concurrency[]; + /** + * (optional) the number of default worker slots this task consumes. + * + * A worker has a fixed number of slots (default 100), and a normal task consumes one. Set slotCost + * higher for a task that needs more memory or CPU, so a worker runs fewer of them at once. A single + * worker must have that many free slots to run it. Not available on durable tasks. + * + * default: 1 + */ + slotCost?: number; + /** @internal */ slotRequests?: Record; }; @@ -239,7 +250,7 @@ export type CreateWorkflowDurableTaskOpts< I extends InputType = UnknownInputType, O extends OutputType = void, C extends DurableTaskFn = DurableTaskFn, -> = CreateWorkflowTaskOpts & { +> = Omit, 'slotCost'> & { /** * Eviction policy for the durable task. Controls TTL-based eviction and capacity-based eviction. * Defaults to the built-in eviction policy when omitted or `undefined`.