> ## Documentation Index
> Fetch the complete documentation index at: https://docs.blockx.chaintable.com/llms.txt
> Use this file to discover all available pages before exploring further.
# Call execution subsystem
> The subsystem inside the Worker that schedules the call list onto the Python Executor pool: DispatcherCore, executor adapter, Executor Pool, subfunction calls, and the UDS protocol
The call execution subsystem lives inside the Worker process and turns the `call list` produced by the Call Builder into aggregated `outputs`. It is the Executor-phase implementation of the [Worker](/en/components/worker) and corresponds to `4.2.4 Call execution subsystem` in the [Architecture overview](/en/architecture/overview). For the Python Executor process itself, see [Python Executor](/en/components/python-executor); this page treats it only as the UDS peer.
## Responsibilities and boundaries
Responsible for:
* Scheduling calls from multiple tasks with one Worker-global `DispatcherCore`, admitting work fairly per task / instance.
* Selecting an Executor for top-level calls via `Filter -> Score`; subfunction calls are always bound back to the Executor running the parent call.
* Call result caching and singleflight deduplication within a task.
* Handling call suspension (IO / subfunction / source fetch) and resumption, and forwarding IO requests to the [IO access subsystem](/en/components/io-subsystem).
* Managing the Python Executor process pool (spawn / restart / sandbox) and maintaining a long-lived UDS connection to each Executor.
* Aggregating the outputs of successful root calls into `outputs []any` and handing them back to `WorkerCore` for the Plugin phase.
Not responsible for:
* Generating tasks, allocating slots, or placing Workers (owned by the [Coordinator](/en/components/coordinator) and the Worker).
* Function source snapshots and auditing themselves (owned by [Function Code View](/en/components/function-code)); this page only consumes `TaskFunctionView.ResolveForDispatch`.
* Real BlockDB / RPC access (owned by the IO subsystem); Executors never connect to external dependencies directly.
Core invariants:
* `DispatcherCore` is a Sans-IO, pure in-memory state machine with a single writer: only the dispatch actor goroutine (`Orchestrator.RunDispatchLoop`) may mutate it; other goroutines can only post a `dispatchEvent`. The lock invariant is enforced by `internal/worker/adapters/actor_invariants_test.go`.
* A call maps to exactly one `DispatchCallContext` for its whole lifetime; retries reuse the same `callID` and only increment `AttemptCount`.
* `DONE` is the only terminal state; `setCallTerminal` is the single accounting point and is idempotent when called repeatedly.
* Subfunction calls have their own `DispatchCallContext` with `MaxAttempts = 1`; they never enter `TotalCount` / `outputs` and only return their result to the parent call through `ResumeCall`.
* Subcall IDs are generated from a process-level monotonic `subcallSeq` (`{parent}/sub-{n}`) and are globally unique within the worker process.
* The Worker side has no per-call timer: the execution budget `CallBudgetMs` is enforced locally by the Executor, and the outer wall-clock protection is the task deadline plus the wedge backstop.
* The adapter reservation is released as soon as the Executor's terminal frame arrives (transport truth), and this is idempotent with the dispatcher's `UnbindCall`.
## Code location
| Path | Purpose |
| -------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------- |
| `cmd/worker/main.go`, `internal/worker/app/app.go` | Assembly: creates `executor.Adapter` and `PoolManager`, starts the dispatch loop and the 500ms tick (snapshot sync, wedge check, heartbeat timeout) |
| `internal/worker/core/dispatcher.go` | `DispatcherCore` event handling, `TryDispatchNext`, instance-fair scheduling, subcall registration/binding |
| `internal/worker/core/dispatcher_internal.go` | Cache keys, `selectExecutor`, retry/backoff, subtree cancellation, `FlushDirtyTasks` convergence epilogue |
| `internal/worker/core/dispatcher_types.go` | `DispatchCallContext`, `TaskDispatchState`, `ExecutorSnapshot`, and every `DispatchCommand` |
| `internal/worker/core/config.go` | `DispatcherConfig` and its defaults |
| `internal/worker/adapters/orchestrator_dispatch.go` | `executeDispatchCommands`: translates core commands into Bind / code resolution / UDS sends; `HandleIO` |
| `internal/worker/adapters/orchestrator_subcall.go` | `HandleSubcall`, `pendingSubcall`, `sendSubcallResume` |
| `internal/worker/adapters/orchestrator_actor.go` | Dispatch actor: `dispatchEvent` types and `applyXxx` handlers |
| `internal/worker/adapters/orchestrator_wedge.go` | `CheckWedgedExecutors` hard-kill backstop |
| `internal/worker/adapters/exec_adapter.go` | `ExecutorAdapter` interface (the orchestrator's dependency surface on the UDS adapter) |
| `internal/worker/adapters/executor/adapter.go` | UDS listener, `executorConn`, reservations (reservations / draining), `Snapshots`, writer loop |
| `internal/worker/adapters/executor/handlers.go` | Executor→Worker message handling: Heartbeat / CallCompleted / CallFailed / CallWaiting |
| `internal/worker/adapters/executor/send.go`, `payload_buffer.go` | Worker→Executor sends, outbox, payload buffer pool |
| `internal/worker/adapters/executor/function_code_source.go` | Source return for `waitKind=function_code` |
| `internal/worker/adapters/executor/pool.go`, `spawner.go`, `sandbox_config.go` | Executor Pool Manager: slot state machine, bare process / containerd sandbox spawn modes |
| `api/uds/types.go`, `api/uds/codec.go` | UDS message types, payload structs, frame encoding/decoding |
| `internal/worker/core/dispatcher*_test.go`, `retry_backoff_test.go` | Core unit tests |
| `internal/worker/adapters/orchestrator_dispatch_test.go`, `orchestrator_subcall_test.go`, `adapters/executor/*_test.go`, `api/uds/*_test.go` | Adapter / protocol unit tests |
| `cmd/worker/worker_process_*_test.go` | Process E2E (real Python Executor) |
| `docs/specs/call-execution-subsystem.md` | Spec for this subsystem (primary) |
| `docs/specs/worker-executor-connection-and-python-sdk-hook.md` | Connection model and UDS contract |
## Core types and interfaces
* `core.DispatcherCore` (`internal/worker/core/dispatcher.go`): the global scheduler. Its input is the `HandleXxx` event methods; its output is `[]DispatchCommand`.
* `core.DispatchCallContext` (`dispatcher_types.go`): the scheduling carrier for a single call, holding `State`, `ExecutorID`, `AttemptCount`, `BudgetMs`, `IsSubcall`, `BatchIdx`, `ArgsJSON`.
* `core.TaskDispatchState` (`dispatcher_types.go`): per-task state, holding `ReadyQueue`, `InflightCalls`, `CallCache`, `Singleflight`, `PendingBindSubcalls`, `ChildSubcalls`, plus the streaming-production fields (`ProductionOpen`, `BatchRemaining`, `PendingSettle`, `CollectedOutputs`).
* `core.ExecutorSnapshot` (`dispatcher_types.go`): the Executor capacity view provided by the adapter, holding `Healthy`, `RunnableCount`, `UsableContexts`, `HeartbeatSeq`, `RunningCallID`, `LastSchedulerActiveAtMs`.
* `core.DispatchCommand` (`dispatcher_types.go`): a sealed interface implemented by `BindCall`, `UnbindCall`, `CancelCallCmd`, `DispatchCallCmd`, `ArmCallRetry`, `ConvergeTaskCalls`, `CompleteCallFromCache`, `BatchSettled`, `StopCallProduction`.
* `core.DispatcherCore.TryDispatchNext / tryDispatchNextFair`: one admission decision; `selectExecutor` performs `Filter -> Score`.
* `core.DispatcherCore.PrepareCallPhaseWithDigests + CommitCallPhase`, `CommitCallStream + PrepareCallBatchWithDigests + CommitCallBatch`: the one-shot and streaming call registration entry points; `Prepare*` has no side effects and can run off-actor.
* `core.DispatcherCore.FlushDirtyTasks` (`dispatcher_internal.go`): the convergence epilogue; grouped by task, it emits `StopCallProduction -> drain -> BatchSettled -> ConvergeTaskCalls`.
* `adapters.ExecutorAdapter` (`exec_adapter.go`): the UDS adapter interface the orchestrator depends on.
```go theme={null}
type ExecutorAdapter interface {
Bind(ctx context.Context, callID, executorID string) error
BindSubcall(ctx context.Context, callID, parentCallID, executorID string) error
Unbind(ctx context.Context, callID, executorID string)
MarkDraining(ctx context.Context, callID, executorID string)
IsDraining(callID, executorID string) bool
SendExecuteCall(ctx context.Context, executorID, callID string, payload uds.ExecuteCallPayload) error
SendResumeCall(ctx context.Context, executorID, callID, requestID string, payload uds.ResumeCallPayload) error
SendCancelCall(ctx context.Context, executorID, callID, reason string, attemptSeq int) error
Snapshots(ctx context.Context) ([]core.ExecutorSnapshot, error)
// SetCallbacks / SetIOHandler / SetSubcallHandler / SetFunctionCodeResolver / SetExecutorLostCallback
}
```
* `adapters.Orchestrator.executeDispatchCommands` (`orchestrator_dispatch.go`): the command executor. `Bind`/`Unbind` run synchronously on the actor; `SendExecuteCall`/`SendResumeCall`/`SendCancelCall` always run in an off-actor goroutine.
* `executor.IOHandler`, `executor.SubcallHandler`, `executor.FunctionCodeResolver` (`adapters/executor/adapter.go`): the three orchestrator entry points the adapter calls back into after receiving `CallWaiting`.
* `executor.Adapter` (`adapter.go`): one `executorConn` per Executor, holding `reservations`, `draining`, `dispatchedAttempts`, `usableContexts`; `Snapshots()` is derived from these.
* `executor.PoolManager`, `PoolConfig`, `SlotState` (`pool.go`): the process slot state machine `RUNNING / WAITING / BACKOFF / STOPPED`; `executorSpawner` abstracts over bare process and sandbox.
* `uds.MessageEnvelope` and the payload structs (`api/uds/types.go`): `ExecuteCallPayload`, `ResumeCallPayload`, `CancelCallPayload`, `CallWaitingPayload`, `CallCompletedPayload`, `CallFailedPayload`, `HeartbeatPayload`.
## Data flow / execution flow
Interaction model: Executor reader goroutines / timers / gRPC post facts as `dispatchEvent`s (`evCallCompleted`, `evCallWaiting`, `evSubcallRequest`, `evExecutorSnapshot`, etc.); the actor calls `DispatcherCore.HandleXxx` to obtain commands; `executeDispatchCommands` executes them; every event ends with `flushDispatchEpilogue`.
```mermaid theme={null}
sequenceDiagram
participant WC as WorkerCore
participant O as Orchestrator actor
participant D as DispatcherCore
participant EA as executor.Adapter
participant EX as Python Executor
participant IO as TaskIOScope
WC->>O: StartCallPhase(taskId, callList)
O->>O: TaskFunctionView resolves each call's digest
O->>D: PrepareCallPhaseWithDigests + CommitCallPhase
loop TryDispatchNext
D-->>O: BindCall(callId, executorId)
O->>EA: Bind (sync, reserves a context)
O->>D: HandleExecutorBound
D-->>O: DispatchCallCmd(functionId, digest, args, budgetMs)
O->>O: ResolveForDispatch(functionId, digest) validation
O->>EA: SendExecuteCall (off-actor)
EA->>EX: ExecuteCall frame
end
EX->>EA: CallWaiting(waitKind=io | subcall | function_code)
EA->>O: onCallWaiting -> HandleCallWaiting (RUNNING->WAITING)
alt waitKind=io
EA->>IO: HandleIO(taskId, callId, req)
IO-->>EA: result
else waitKind=subcall
EA->>O: HandleSubcall -> AddSubcallJSON + TryBindSubcallToParent
O->>EX: ExecuteCall(child call, ParentCallID)
EX->>O: CallCompleted(child call)
end
EA->>EX: ResumeCall(requestId, result, budgetUsedMs)
EA->>O: onCallResumed -> HandleCallResumed (WAITING->RUNNING)
EX->>EA: CallCompleted(output)
EA->>EA: releaseOnTerminal
EA->>O: onCallCompleted -> HandleCallCompleted
O->>D: FlushDirtyTasks
D-->>O: ConvergeTaskCalls(success, outputs)
O->>WC: HandleTaskPhaseFinished(PhaseCalls)
```
Key steps:
`runCallPhase` (`orchestrator_phases.go`) resolves each call's `functionCodeDigest` by `taskCodeEpoch` on the task activation goroutine, calls `PrepareCallPhaseWithDigests`, then posts `evCallPhaseStarted` so the actor runs `CommitCallPhase`. Streaming builders go through `CommitCallStream` plus one `CommitCallBatch` per batch; the producer is backpressured by the `StreamBuildConfig.MaxOutstandingBatches` credit window, and `BatchSettled` returns one credit.
`TryDispatchNext` takes the call at the head of `ReadyQueue`: it first checks `CallCache` (a hit goes straight to `CompleteCallFromCache`), then `Singleflight` (consecutive calls with the same key are registered as waiters in one pass); otherwise `selectExecutor` picks an Executor, the call is registered as the singleflight leader, set to `RUNNING`, `AttemptCount++`, `BudgetMs = CallDeadlineMs`, and `BindCall` is emitted.
`selectExecutor` filters out: `!Healthy`; scheduler stalled for longer than `SchedulerStallTimeoutMs`; `headroom = UsableContexts - localInflight <= 0`; compensated `runnable >= ExecutorMaxRunnable`; `localInflight >= ExecutorMaxInflight`; `MemoryBytes >= ExecutorMemoryHighWatermark`. Scoring: lower `runnable` first, then lower `localInflight`, then higher `headroom`, and finally `ExecutorID` in lexicographic order. When `ExecutorSelectionSampleSize > 0`, only the first N eligible candidates are scanned (a power-of-d approximation).
`BindCall` calls `Adapter.Bind` synchronously on the actor (subcalls go through `BindSubcall`, which passes as long as the parent holds a live reservation); on success, `HandleExecutorBound` emits `DispatchCallCmd`. The adapter uses `resolveFunctionCodeForDispatch` to validate `(functionID, digest)` against the task's pinned view and fetch the `EntrySelector`, assembles `ExecuteCallPayload` (digest only, no source), then in a new goroutine passes the audit gate, performs alive / draining checks, and calls `SendExecuteCall`. A bind failure emits `ArmCallRetry(now)` for a root call without consuming attempt budget; a subcall is parked back into `PendingBindSubcalls`.
The Executor sends `CallWaiting`. `waitKind=function_code` does not change dispatcher state; the adapter returns the source directly via `FunctionCodeResolver`. `io` / `subcall` trigger `HandleCallWaiting`, and the adapter calls `IOHandler.HandleIO` (locating the runtime by the `TaskID` echoed in the payload, without consulting the dispatcher) or `SubcallHandler.HandleSubcall` in a `processWaitRequest` goroutine. The result goes through `SendResumeCall`, which blocks until the frame is actually flushed before `HandleCallResumed` fires; if the send fails, a `resume_send_failed` failure event is synthesized for the parent call (retryable, carrying the attempt sequence number).
When `CallCompleted` / `CallFailed` reaches the adapter, it first releases the reservation via `releaseOnTerminal`, then calls back into `applyCallCompleted` / `applyCallFailed`. The core writes `CallCache`, resolves singleflight waiters, cancels the call's subcall subtree, and calls `markTaskDirty`. `flushDispatchEpilogue` loops over `FlushDirtyTasks` until quiescent: when `TerminalRootCount == TotalCount` (and, for streaming, `ProductionDone`), it emits `ConvergeTaskCalls`, which the adapter translates into `WorkerCore.HandleTaskPhaseFinished(PhaseCalls)`.
### Subfunction call path
1. The SDK inside the Executor sends `CallWaiting(waitKind=subcall, functionId, args, ancestry, grantedBudgetMs)`.
2. `HandleSubcall` posts `evSubcallRequest`; `applySubcallRequest` validates on the actor: depth `len(ancestry) < MaxSubcallDepth`, cycle detection (comparing `CacheKeyForJSON` against ancestor keys), and the requesting executor must equal the executor the parent call is currently bound to (rejecting late requests from an old attempt).
3. `resolveFunctionCodeForDispatch` fetches the subfunction digest; `AddSubcallJSON` creates the child `DispatchCallContext` (inheriting the parent's `BatchIdx`, with `BudgetMs = grantedBudgetMs`), records `pendingSubcalls[childID]`, and `TryBindSubcallToParent` binds it back to the parent Executor. If the parent is unbound or the parent Executor is unhealthy, the subcall is parked in `PendingBindSubcalls` and retried on `HandleExecutorBound(parent)` or `HandleExecutorSnapshotUpdated`.
4. After the child call reaches a terminal state, `sendSubcallResume` returns the result or a `subcallError` (including `retryable`) to the parent Executor via `ResumeCall`; `BudgetUsedMs` lets the parent deduct budget. Subcalls that hit the cache / singleflight go through `CompleteCallFromCache` and resume the parent directly.
5. When the parent call reaches a terminal state (including a failed attempt that is retried), `finalizeSubcallSubtree` recursively cancels the whole subtree and emits `CancelCallCmd`; the adapter marks the reservation as draining and calls `SendCancelCall` best-effort.
### Executor Pool and connections
* `PoolManager.Start` spawns one `python -m blockx_executor --executor-id exec-N --socket-path ... --executor-max-inflight ... --executor-max-runnable ...` per slot. After the process exits, `waitLoop` restarts it after `RestartDelayMs`; more than `MaxRestarts` restarts within `RestartWindowMs` puts the slot into `BACKOFF` for `BackoffMs`. `KillExecutor` is used by the wedge backstop for hard kills.
* With `ExecutorSpawnMode=sandbox`, `sandboxSpawner` starts a containerd container through `github.com/Chaintable/emulator/sandbox`: no network, only the UDS socket directory mounted, only the `childEnvKeys` allowlist of environment variables passed through, and optional CPU / memory limits. See `docs/specs/executor-sandbox-isolation.md` in the blockx repo for the design.
* The Executor actively connects to `Adapter.SocketPath()` (`/blockx-worker-.sock`); the first valid `Heartbeat` completes the attach, and there is no explicit registration message. Each connection has one reader goroutine and one writer goroutine; the writer takes frames from the `outbox` (depth 256), coalescing up to 64 frames per shared `Flush`.
* `ExecuteCall` enqueues without blocking and returns a retryable error when the outbox is full; `ResumeCall` / `CancelCall` block until the flush completes, because dispatcher state progression depends on the frame actually arriving.
* 500ms tick: `SyncExecutorSnapshots` feeds `Adapter.Snapshots()` into `HandleExecutorSnapshotUpdated`; `CheckWedgedExecutors` hard-kills Executors whose `RunningCallID != ""` and whose `LastSchedulerActiveAtMs` has stalled for longer than `SchedulerStallTimeoutMs`; `CheckHeartbeatTimeouts` declares connections with no heartbeat for more than `HeartbeatTimeoutMs` as transport lost, triggering `onExecutorLost` → `applyExecutorLost`, which first refreshes the snapshot and then emits a retryable failure for every inflight call on that Executor.
### UDS protocol
Frame format: a 4-byte big-endian length prefix followed by a JSON `MessageEnvelope`; when a binary sidecar is present, the body starts with the `BXB1` magic plus a 4-byte JSON length, with the raw bytes immediately following the JSON (BlockDB protobuf takes this path to avoid base64). A single frame is capped at 16 MiB. Encoding/decoding uses sonic, and `json.RawMessage` fields are spliced in verbatim.
| Direction | `messageType` | payload | Description |
| --------------- | --------------- | ---------------------- | ------------------------------------------------------------------------------------------------------------------------------------- |
| Worker→Executor | `ExecuteCall` | `ExecuteCallPayload` | `TaskID`, `FunctionID`, `FunctionCodeDigest`, `EntrySelector`, `Args`, `ParentCallID`, `CallBudgetMs`, `AttemptSeq`, optional `Audit` |
| Worker→Executor | `ResumeCall` | `ResumeCallPayload` | `ResumeKind`, `Result` or `Error`, `BudgetUsedMs`; `RequestID` correlates the suspension |
| Worker→Executor | `CancelCall` | `CancelCallPayload` | `Reason`, `AttemptSeq`, best-effort |
| Worker→Executor | `HeartbeatAck` | none | Non-blocking, harmless if lost |
| Executor→Worker | `CallWaiting` | `CallWaitingPayload` | `WaitKind = io / subcall / function_code`, must echo `TaskID`; IO fields or subcall fields |
| Executor→Worker | `CallCompleted` | `CallCompletedPayload` | `Output` (raw JSON), `BudgetUsedMs`, timing fields |
| Executor→Worker | `CallFailed` | `CallFailedPayload` | `ErrorKind`, `Retryable`, `ErrorMessage`, `ErrorStack`, `LocalVars`, etc. |
| Executor→Worker | `Heartbeat` | `HeartbeatPayload` | `RunnableCount`, `InflightCount`, `AvailableContexts`, `MemoryBytes`, `LastSchedulerActiveAtMs`, `RunningCallID`, CPU metrics |
## State and lifecycle
`DispatchCallContext.State` has only four states (`CallReady / CallRunning / CallWaiting / CallDone`). The Executor's internal fine-grained runnable / greenlet states are not surfaced to the Worker.
```mermaid theme={null}
stateDiagram-v2
[*] --> READY: addCall
READY --> RUNNING: BindCall succeeded (AttemptCount++)
READY --> DONE: cache hit / singleflight waiter resolved by leader / drain cancel
RUNNING --> READY: bind failed (attempt rolled back) or retryable failure with MaxAttempts not exhausted
RUNNING --> WAITING: HandleCallWaiting (io / subcall)
WAITING --> RUNNING: HandleCallResumed
WAITING --> READY: retryable failure of a waiting attempt (resume_send_failed / executor lost)
RUNNING --> DONE: CallCompleted or final failure
WAITING --> DONE: final failure or subtree cancelled
DONE --> [*]: removeCall (batch settle / RemoveTask)
```
* Retry and backoff: `handleAttemptFailure` sets `READY` when `retryable && AttemptCount < MaxAttempts && !StopDispatching`; `computeBackoff` emits `ArmCallRetry` using `base * 2^(attempt-1)` plus up to 5% jitter; the adapter uses `time.AfterFunc` to post `evRetryReady` when it fires. The base is taken from the task's `CallRetryPolicy.BaseBackoffMs` when set, otherwise from `DispatcherConfig.BaseBackoffMs`.
* Stale events: `CallFailureStale` drops failure events when the call is `READY && AttemptCount > 0` (backing off) or the `attemptSeq` does not match; a call that is already `DONE` ignores every late terminal event.
* Failure policy: under `CallFailureFastFail`, the final failure of any root call triggers `markStopDispatching`, after which no further calls are dispatched, and `drainCancelledCalls` cancels and converges all remaining calls once inflight drops to zero.
* Budget: each root-call attempt gets `BudgetMs = CallDeadlineMs` (default 5000ms, 0 means unlimited), counting only CPU + IO backend + subcalls + sleep, not queueing time; subcalls receive the `grantedBudgetMs` carved out by the parent.
* Reservation lifecycle (adapter): `Bind` adds to `reservations`; a normal terminal state calls `Unbind`; the cancel path calls `MarkDraining` and waits for the Executor's terminal frame (`releaseOnTerminal`) before releasing; a reconnect era clears all reservations.
* Output retention: when `MaxCollectedOutputBytes > 0` and the accumulated output exceeds the limit, the task stops production with `OUTPUT_BYTES_EXCEEDED` and discards its outputs.
## Configuration
From `internal/worker/app/config.go` (`WorkerFullConfig`) and `internal/worker/core/config.go` (`DispatcherConfig`). See `internal/worker/app/app.go` for environment variable overrides.
| Field | Default | Description |
| ------------------------------------------------------------------------------------------ | -------------------------------------------- | ------------------------------------------------------------------------------------------------------------ |
| `ExecutorCount` (`EXECUTOR_COUNT`) | 8 | Number of Executor processes (`PoolConfig.ExecutorCount`) |
| `ExecutorSpawnMode` (`EXECUTOR_SPAWN_MODE`) | `process` | `process` or `sandbox`; the sandbox additionally needs `EXECUTOR_SANDBOX_IMAGE_REF` etc. |
| `Dispatcher.ExecutorMaxInflight` | 128 | Per-Executor local inflight cap, also passed to Python as `--executor-max-inflight` |
| `Dispatcher.ExecutorMaxRunnable` | 128 | Per-Executor runnable threshold (sampled from heartbeats, compensated by completion count) |
| `Dispatcher.ExecutorSelectionSampleSize` (`EXECUTOR_SELECTION_SAMPLE_SIZE`) | 0 | 0 scans all; N>0 scores only the first N eligible candidates |
| `Dispatcher.ExecutorMemoryHighWatermark` | 0 | Executor RSS high-watermark filter; 0 disables it |
| `Dispatcher.TaskMaxInflightCalls` | 1024 | Per-task admission window |
| `Dispatcher.CallDeadlineMs` (`CALL_DEADLINE_MS`) | 5000 | Execution budget per root-call attempt (the name is legacy; the semantics are now a budget) |
| `Dispatcher.SchedulerStallTimeoutMs` (`SCHEDULER_STALL_TIMEOUT_MS`) | Derived: `CallDeadlineMs + 5000` | Wedge backstop threshold; an explicit value `<= CallDeadlineMs + 1000` is raised to the derived value |
| `Dispatcher.MaxSubcallDepth` | 8 | Maximum subcall depth |
| `Dispatcher.BaseBackoffMs` | 10 | Retry backoff base (when the task does not specify one) |
| `Dispatcher.InstanceFairness` (`DISPATCH_INSTANCE_FAIRNESS`) | false | Enables instance-level max-min fair admission |
| `Dispatcher.InstanceFairnessQuantum` (`DISPATCH_INSTANCE_FAIRNESS_QUANTUM`) | 0 | Number of consecutive binds served to the same instance; 0/1 both mean re-selecting per call; capped at 1024 |
| `Dispatcher.MaxCollectedOutputBytes` (`MAX_COLLECTED_OUTPUT_BYTES`) | 0 | Per-task cap on retained output bytes |
| `Worker.DefaultMaxAttempts` (`CALL_RETRY_MAX_ATTEMPTS`) | 3 | Cap on in-call retry attempts |
| `ExecutorRestartDelayMs / ExecutorMaxRestarts / ExecutorRestartWindow / ExecutorBackoffMs` | 1000 / 10 / 60000 / 30000 | Pool restart throttling |
| `Admission.ExecutorTaskSlots` (`EXECUTOR_TASK_SLOTS`) | 32 | Number of tasks simultaneously in the Executor phase |
| `StreamBuild.MaxOutstandingBatches / BatchMaxRows / BatchMaxArgsBytes` | 2 / 8192 / 32 MiB | Streaming production window and batch sizes (`StreamBuild.Enabled` defaults to false) |
| `executor.Config.HeartbeatTimeoutMs` | 10000 (`stall + 5000` once coupled to stall) | Heartbeat timeout for declaring transport lost |
## Extension points
* Changing the scheduling policy: edit Filter / Score in `selectExecutor` / `betterExecutorCandidate` (`dispatcher_internal.go`); edit the fairness model in `tryDispatchNextFair` (`dispatcher.go`). Run `dispatcher_select_bench_test.go` and `dispatcher_fairness_bench_test.go` to check for regressions.
* Adding a core event or command: add a `DispatchCommand` implementation in `dispatcher_types.go` and a branch in `executeDispatchCommands` in `orchestrator_dispatch.go`; for a new event, add an `evXxx` type and `applyXxx` in `orchestrator_actor.go`, and make sure `dispCore` is only mutated on the actor (`actor_invariants_test.go` will catch violations). Any entry point that can affect convergence must call `markTaskDirty`; convergence only happens inside `FlushDirtyTasks`.
* Adding a UDS message type or payload field: add the constant and struct in `api/uds/types.go`, add routing in `handleMessage` in `adapters/executor/adapter.go`, and add a handler in `handlers.go`; update the Python-side protocol and `api/uds/codec_test.go` in step.
* Adding a `waitKind`: add a branch in `handleCallWaiting` / `processWaitRequest` in `handlers.go`; decide whether it enters the dispatcher's `WAITING` state (`function_code` does not).
* Adding an Executor spawn mode: implement `executorSpawner` and `executorProcess` (`spawner.go`), register it in `newExecutorSpawner`, and add validation in `sandbox_config.go`.
* Adjusting Executor heartbeat fields: `uds.HeartbeatPayload` → `handleHeartbeat` / `heartbeatUpdateExisting` in `handlers.go` → `core.ExecutorSnapshot`.
## Testing
```bash theme={null}
# Core unit tests (pure in-memory, no Python)
go test ./internal/worker/core/ -run 'Dispatcher|Backoff|Subcall' -count=1
# Adapter / protocol unit tests
go test ./internal/worker/adapters/ -run 'Dispatch|Subcall|Actor' -count=1
go test ./internal/worker/adapters/executor/ ./api/uds/ -count=1
# Race check (required after touching the actor / adapter boundary)
go test -race ./internal/worker/...
# Process E2E: starts a real worker + Python executor, requires python/.venv
go test ./cmd/worker/ -run 'WorkerProcess_(Subcall|ExecutorLost|ExecutorPoolRestart|Wedge)' -count=1
```
* Core tests are split by topic: `dispatcher_test.go` (basic scheduling), `dispatcher_subcall_test.go`, `dispatcher_stream_test.go` (streaming production / batch settle / output cap), `dispatcher_instance_fairness_test.go` and `dispatcher_fairness_quantum_test.go`, `retry_backoff_test.go`, `dispatcher_stale_failure_test.go`, `dispatcher_wedge_test.go`.
* Adapter tests inject send failures and similar paths through an `ExecutorAdapter` stub: `orchestrator_dispatch_test.go`, `orchestrator_subcall_test.go`, `orchestrator_send_failure_test.go`, `orchestrator_capacity_failure_test.go`.
* `adapters/executor/` tests start a fake executor over a real UDS socket: `adapter_test.go`, `pool_test.go`, `resume_pool_test.go`, `write_coalesce_test.go`, `payload_buffer_test.go`.
* `api/uds/codec_test.go` and `bench_test.go` cover frame encoding/decoding and the binary sidecar.
* E2E tests live in `cmd/worker/worker_process_*_test.go`; see [Testing](/en/development/testing) and `docs/specs/worker-test-organization.md` in the blockx repo for the test organization conventions.
## Related docs
Specs in the blockx repo:
* `docs/specs/call-execution-subsystem.md`: the design of this subsystem (primary), covering Filter/Score, retries, subcalls, and streaming production.
* `docs/specs/worker-executor-connection-and-python-sdk-hook.md`: connection attach, failure detection, the UDS message contract, and the SDK hook.
* `docs/specs/architecture.md` §4.2.4: where the subsystem sits in the overall architecture.
* `docs/specs/2026-07-23-execute-call-function-code-reference.md`: `ExecuteCall` carries only the digest; `waitKind=function_code` fetches source on demand.
* `docs/specs/2026-07-30-python-executor-cpu-optimization.md`: the lasting impact is that BlockDB requests / responses use the `BXB1` binary sidecar, and the Worker uses the raw bytes as the IO cache key.
* `docs/specs/executor-sandbox-isolation.md`: the containerd sandbox spawn mode.
Related pages on this site:
* [Worker](/en/components/worker), [Python Executor](/en/components/python-executor), [Function Code View](/en/components/function-code), [IO access subsystem](/en/components/io-subsystem), [Plugin system](/en/components/plugins)
* [Task lifecycle](/en/architecture/task-lifecycle), [Protocols and interfaces](/en/architecture/protocols)