> ## 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. # Worker > How the Worker process manages task slots, drives the three Builder → Calls → Plugin phases, and exposes results The Worker is BlockX's execution-plane node. It maintains a fixed-size pool of task slots on the local machine, accepts tasks submitted by the Coordinator or by a directly connected Client, executes them in the order Builder → Calls → Plugin, and keeps results for a short window so they can be queried and subscribed to. See [Architecture overview](/en/architecture/overview) for where it sits in the overall architecture, and [Task lifecycle](/en/architecture/task-lifecycle) for the end-to-end task path. ## Responsibilities and boundaries The Worker is responsible for: * Exposing the gRPC `WorkerService` (`api/grpc/worker/worker.proto`): `RequestTaskSlot`, `SubmitTask`, `GetTaskResult`, `WatchTasks`. `WatchTasks` is a server-streaming RPC; the Worker listens on a single TCP port. See [Protocols and interfaces](/en/architecture/protocols) for protocol details. * Maintaining the local slot table and task index, with atomic admission and `taskId`-level idempotency. * Pinning one `taskCodeEpoch` per task, running the Call Builder serially, handing the call list to the dispatcher for parallel execution, and finally running the Writer Plugin serially. * Converging the `TaskResult`, pushing it to `WatchTasks` subscribers, and retaining it for `resultRetentionMs`. * Registering with etcd and publishing periodic heartbeats so the Coordinator can select it. The Worker is not responsible for: * Generating `taskId`, building payloads, DAG orchestration, or scheduled triggering (these live upstream). * Cross-Worker deduplication, global result queries, or local recovery after a crash. After a restart, all old slots, old results, and subscriptions are lost. * Call-level scheduling details (dispatcher, executor adapter). See [Call execution subsystem](/en/components/call-execution) for that part. Core invariants: * `WorkerCore` is a pure in-memory, Sans-IO decision engine. It never reads the clock; every method receives `now` explicitly. * The slot table is the single source of truth for capacity and task state transitions; the etcd heartbeat is only a derived view of it. * `slotTable` and `taskIndex` are updated together within the same method call; there is no torn state. * Once a task is successfully activated, every subsequent failure (Builder, call, Plugin, timeout, Watch disconnect) converges into a terminal state in `TaskResult`; no further submission-layer errors are returned. * The adapter does not keep a second copy of slot or phase truth. It only executes the commands the core emits and feeds results back as events. ## Code location | Path | Purpose | | ---------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `cmd/worker/main.go` | Full-profile entry point (`Deployment: "block"`); registers the dbscan / callList / bundleScan and devstub builders plus all Writer Plugins | | `cmd/bundle_worker/main.go` | Bundle-profile entry point (`Deployment: "bundle"`); registers only bundleScan / callList and bundleWrite / tableUpserts, with streaming build enabled by default. See [Bundle clusters](/en/components/bundle) for details | | `internal/worker/app/app.go` | `Profile` definition and `Run(p Profile)`: loads configuration, wires registry / IO / executor / orchestrator / gRPC / etcd, and defines the canonical startup and shutdown order | | `internal/worker/app/config.go` | `WorkerFullConfig`, `DefaultWorkerFullConfig`, `LoadWorkerFullConfigFrom`, `Validate` | | `internal/worker/app/function_code.go` | Wires the Function Code View (devstub / Redis / BlockDB sources) and audit integration | | `internal/worker/app/memory_diagnostics.go` | Low-frequency memory diagnostics logging (Go runtime, RSS, DuckDB counters) | | `internal/worker/core/worker.go` | `WorkerCore` event handlers: slot admission, idempotency, phase progression, terminal-state convergence | | `internal/worker/core/worker_slots.go` | Slot capacity accounting, `HandleSlotLeaseExpired`, `ExpiredSlots` | | `internal/worker/core/types.go` | `SlotState`, `TaskPhase`, `TaskContext`, command and result types | | `internal/worker/core/config.go` | `core.Config`, `DispatcherConfig`, `AdmissionConfig` and their defaults | | `internal/worker/core/dispatcher*.go` | `DispatcherCore`; not covered on this page, see [Call execution subsystem](/en/components/call-execution) | | `internal/worker/adapters/grpc_server.go` | `GRPCServer`: unary handlers for `workerpb.WorkerServiceServer` | | `internal/worker/adapters/stream_subscriber.go` | `SubscriptionManager`: the `WatchTasks` streaming handler and subscription set | | `internal/worker/adapters/orchestrator*.go` | `Orchestrator`: holds the locks and runtimes, executes core commands, runs the three phases, and drives the dispatch actor | | `internal/worker/adapters/admission.go` | Phase-level admission semaphores (builder / executor / plugin / scan) | | `internal/worker/adapters/etcd_publisher.go` | `EtcdPublisher`: registration, lease keep-alive, heartbeat publishing, deregistration | | `internal/worker/adapters/task_runtime.go` and others | `taskRuntime`, `TaskStatistic`, log metadata, resource sampling, pb conversion (`wire.go`), shadow forwarding | | `internal/worker/adapters/exec_adapter.go`, `adapters/executor/` | UDS executor adapter and Pool Manager; belong to the [Call execution subsystem](/en/components/call-execution) | | `internal/worker/devstub/` | Stub builders (`static`, `payload`), stub plugins (`log`, `test`), and a stub backend for dev / e2e | | `api/grpc/worker/worker.proto` | Public gRPC protocol definition | | `internal/worker/**/*_test.go`, `cmd/worker/*_test.go` | core unit / adapter integration / process e2e; see "Testing" below | | `docs/specs/worker.md` | Detailed Worker design spec | ## Core types and interfaces **core (`internal/worker/core/`)** * `WorkerCore` (`worker.go`): a single-threaded state machine. It holds `slots []SlotEntry`, `taskIndex map[string]*TaskIndexEntry`, `tasks map[string]*TaskContext`, `ready`, and `draining`. * Input events (all in `worker.go` unless noted): `HandleRequestTaskSlot(taskID, ttlMs, now)`, `HandleSubmitTask(req, adapterTaskCtx, now)`, `HandleTaskActivationPrepared(taskID, slotID, taskCodeEpoch, now)`, `HandleTaskActivationFailed(taskID, failureCode, now)`, `HandleTaskPhaseFinished(taskID, phase, outcome, now)`, `HandleTaskTimedOut(taskID, now)`, `HandleWatchAttached` / `HandleWatchDetached` / `HandleWatchDisconnected(taskID, now)`, `HandleSlotLeaseExpired(slotID, now)` (`worker_slots.go`), `HandleGetTaskResult(taskID)`. * Timer-scan helpers: `ExpiredSlots(now)`, `TimedOutTasks(now)`, `WatchDisconnectedTasks(now)`, `PurgeExpiredResults(now)`. The adapter calls them on every tick and feeds the results back into the corresponding `Handle*`. * Output commands (`types.go`): `PrepareTaskActivation`, `StartBuilderPhase`, `StartCallPhase{CallList, Streaming}`, `StartWriterPhase{Outputs}`, `ConvergeTask{State, Result}`. They are returned embedded in the result types `RequestTaskSlotResult`, `SubmitTaskResult`, `ActivationResult`, `PhaseTransitionResult`, `LeaseExpiredResult`, and `GetTaskResultOutput`. * `TaskContext` (`types.go`): `TaskID`, `SlotID`, `TaskCodeEpoch`, `Phase`, `DeadlineMs`, `WatchAttached` / `WatchEverAttached` / `WatchDisconnectDeadlineMs`, `CallFailurePolicy`, `CallRetryPolicy`, `AcceptedAtMs`, `AdapterTaskCtx`, plus the `PhaseOutcome` of each of the three phases. * `DeriveHeartbeat(now) etcd.WorkerHeartbeat`: derives `ReservedSlots` / `RunningTasks` from the slot table. ```go theme={null} // internal/worker/core/worker.go func (w *WorkerCore) HandleSubmitTask(req commontypes.SubmitTaskRequest, adapterTaskCtx *commontypes.TaskCtx, now int64) SubmitTaskResult func (w *WorkerCore) HandleTaskPhaseFinished(taskID string, phase TaskPhase, outcome PhaseOutcome, now int64) PhaseTransitionResult ``` **adapters (`internal/worker/adapters/`)** * `GRPCServer` (`grpc_server.go`): implements `RequestTaskSlot` / `SubmitTask` / `GetTaskResult` and delegates `WatchTasks` to `SubscriptionManager`. `workerErrToStatus` (`wire.go`) maps `core.WorkerError` to a gRPC status; business error codes go through `workerCodeToGRPC`: `InvalidArgument → InvalidArgument`, `NoSlot → ResourceExhausted`, `Unavailable → Unavailable`, `InvalidSlot → FailedPrecondition`, `NotFound → NotFound`, `SlotUncertain → Aborted`. * `Orchestrator` (`orchestrator.go`): bridges `WorkerCore` and `DispatcherCore`. Entry methods: `RequestTaskSlot`, `SubmitTask`, `GetTaskResult`, `TickTimers`, `DeriveHeartbeat`, `SetDraining`, `WaitDrain`, `RunDispatchLoop`, `SyncExecutorSnapshots`, `CheckWedgedExecutors`. * `OrchestratorDeps` (`orchestrator.go`): constructor dependencies, including `BuilderRegistry`, `PluginRegistry`, `IOScope`, `Admission`, `FunctionCodeViewProvider`, `EpochResolver`, `ExecAdapter`, `AuditGate`, and `StreamBuild`. * `taskRuntime` (`task_runtime.go`): the adapter-side runtime record for a task. It holds `ioTaskCtx`, `TaskIOScope`, `ctx` / `cancel`, `functionCodeView`, the streaming producer handle, and `TaskStatistic`. It does not hold phase or slot truth. * `SubscriptionManager` (`stream_subscriber.go`): one `streamSubscriber` per `WatchTasks` stream; translates the first attach / last detach into `Orchestrator.HandleWatchAttached` / `HandleWatchDetached`. * `EtcdPublisher` (`etcd_publisher.go`): `Register`, `RunHeartbeatLoop`, `Deregister`. The key is `KeyPrefix + workerAddr`; the lease TTL defaults to 10s. * `ExecutorAdapter` interface (`exec_adapter.go`): the UDS executor surface the Orchestrator depends on; the production implementation is `executor.Adapter`. **app (`internal/worker/app/`)** * `Profile` (`app.go`): `Deployment` (`block` / `bundle`, used only as an observability label), `Service` (tracing / Prometheus service name), `WorkerRegistryPrefix` (etcd registration prefix), `UsageService`, `Builders`, `Plugins`, `TuneDefaults` (only changes built-in defaults). * `Run(p Profile)`: the single wiring and startup/shutdown sequence. ## Data flow / execution flow The Worker follows an "input event → core decision → adapter executes command → event fed back" pattern. `Orchestrator` guards `WorkerCore` with `o.mu` and `runtimes` with `runtimeMu`; `DispatcherCore` is only mutated on the dispatch actor (the `RunDispatchLoop` goroutine), and other goroutines post events to it via `postDispatchEvent`. The lock order and actor invariants are documented in the header comment of `orchestrator_actor.go` and enforced by `actor_invariants_test.go`. ```mermaid theme={null} sequenceDiagram participant C as Client / Coordinator participant G as GRPCServer participant O as Orchestrator participant W as WorkerCore participant A as dispatch actor participant S as SubscriptionManager C->>G: SubmitTask(task, slotId?) G->>O: SubmitTask(ctx, req) O->>W: HandleSubmitTask(req, taskCtx, now) W-->>O: SubmitTaskResult{Prepare} O-->>G: state=RUNNING G-->>C: SubmitTaskResponse Note over O: go activateAndRun O->>O: EpochResolver.CurrentEpoch + FunctionCodeViewProvider.Pin O->>W: HandleTaskActivationPrepared W-->>O: ActivationResult{StartBuilder} O->>O: runBuilderPhase (builderSem, Builder.Build) O->>W: HandleTaskPhaseFinished(Builder) W-->>O: PhaseTransitionResult{StartCalls} O->>A: runCallPhase (executorSem) → evCallPhaseStarted A->>A: DispatcherCore schedules calls, executor calls back evCallCompleted/Failed A->>W: ConvergeTaskCalls → HandleTaskPhaseFinished(Calls) W-->>A: PhaseTransitionResult{StartPlugin} A->>O: go runWriterPhase (pluginSem, Plugin.Execute) O->>W: HandleTaskPhaseFinished(Plugin) W-->>O: PhaseTransitionResult{Converge} O->>A: evTaskTerminal → applyTaskTerminal / executePhaseResult A->>S: OnTaskTerminal → NotifyTerminal S-->>C: WatchTasks stream: TaskUpdate{isTerminal} C->>G: GetTaskResult(taskId) G->>W: HandleGetTaskResult W-->>C: state + TaskResult (within the retention window) ``` A few key points: * `SubmitTask` only goes as far as `HandleSubmitTask` before returning. The core immediately creates a `TaskContext` (`Phase = Preparing`) so that duplicate submissions hit the idempotency path; activation and phase execution happen asynchronously in the `activateAndRun` goroutine. * Activation failures (epoch resolution failure, `Pin` failure, `NewTaskScope` failure) go through `HandleTaskActivationFailed` with failure code `ACTIVATION_FAILED` or `IO_SCOPE_FAILED`. * Builder phase: `runBuilderPhase` first acquires `builderSem`, then looks up `FunctionCallConfig.Type` in `BuilderRegistry`. If it is not found, the task converges directly with `BUILDER_NOT_FOUND`. If the builder implements `StreamingCallBuilder` and `StreamBuild.Enabled` is set, the Builder phase only runs `PrepareStream` and feeds back a `CallStreamMarker`; the core emits `StartCallPhase{Streaming: true}`, and `runCallStreamPhase` then starts the producer goroutine under `scanSem`. * Calls phase: `runCallPhase` acquires `executorSem`, prepares the call context outside the lock with `dispCore.PrepareCallPhaseWithDigests`, then posts `evCallPhaseStarted` to the actor. Call-level scheduling, retries, and subcalls belong to the [Call execution subsystem](/en/components/call-execution). When the actor receives the `ConvergeTaskCalls` command, it calls `HandleTaskPhaseFinished(PhaseCalls)`. * Plugin phase: `runWriterPhase` acquires `pluginSem` and looks up `PluginRegistry` by `ResultHandler.Type`. If no `ResultHandler` is configured, no plugin runs and success is fed back directly. * Terminal state: `executePhaseResult` sends `CancelCall` to any calls still in flight, cleans up dispatcher state, closes the `TaskIOScope`, deletes the `taskRuntime`, writes the `task_finished` log, and notifies subscribers through `OnTaskTerminal`. Slot release has already completed synchronously inside `WorkerCore.convergeTask`. * Timers: `app.Run` calls `orch.TickTimers()` every 500ms. It handles slot lease expiry, task timeouts, Watch grace expiry, and result-cache cleanup in turn, followed by `SyncExecutorSnapshots`, `CheckWedgedExecutors`, and `execAdapter.CheckHeartbeatTimeouts`. * Shadow forwarding: with `ShadowSubmitTargetAddr` configured, `SubmitTaskShadowForwarder` mirrors the request to another Worker at the configured sampling rate after the primary submission succeeds; the receiving side recognizes it via `isShadowSubmit` and strips the `ResultHandler`. `CheckWedgedExecutors` in `orchestrator_wedge.go` is a last-resort safeguard: if an executor is still heartbeating but its `LastSchedulerActiveAtMs` has not advanced for longer than `SchedulerStallTimeoutMs` while it reports a `RunningCallID`, the hook injected via `SetExecutorKiller` hard-kills it so the Pool Manager can spawn a new process. `app.Run` derives `SchedulerStallTimeoutMs` from `CallDeadlineMs` and clamps it to a safe lower bound. ## State and lifecycle ### Slot state machine `SlotState` has only three values (`core/types.go`): `FREE`, `ALLOCATED`, `RUNNING`. The number of slots is fixed at `taskSlots`, and `SlotID` takes the form `slot-N`. ```mermaid theme={null} stateDiagram-v2 [*] --> FREE FREE --> ALLOCATED: HandleRequestTaskSlot succeeds ALLOCATED --> RUNNING: HandleTaskActivationPrepared ALLOCATED --> FREE: HandleSlotLeaseExpired RUNNING --> FREE: convergeTask, any terminal state ``` * `HandleRequestTaskSlot` first performs the idempotency check (if the same `taskId` is still active, it returns a stable snapshot; if it is already terminal and within the retention window, it returns the terminal state), then checks `freeCapacity() = TaskSlots - (ALLOCATED + RUNNING)`, and finally does a linear scan to allocate the first `FREE` slot. * When `SubmitTask` carries no `slotId`, the core runs `HandleRequestTaskSlot` inline with `DefaultTTLMs` as the TTL. * If the lease expires between `SubmitTask` and `HandleTaskActivationPrepared` and the slot is reclaimed, the core converges that task to `FAILED` with `SLOT_LOST`. ### Task-visible states and internal phases Externally visible states (`commontypes.TaskState`): `ALLOCATED`, `RUNNING`, `SUCCEEDED`, `FAILED`. Internal phases (`core.TaskPhase`): `Preparing`, `Builder`, `Calls`, `Plugin`, `Terminal`. All internal phases appear as `RUNNING` externally. Progression rules in `HandleTaskPhaseFinished`: * `Builder` fails → `FAILED`, with the failure code taken from the outcome or `BUILDER_FAILED`. Succeeds → `Calls`. * `Calls` fails → `FAILED`. Succeeds → always emits `StartWriterPhase`. * `Plugin` finishes → terminal state; the failure code is taken from the outcome or `PLUGIN_FAILED`, and `PluginResults` is written into `ExecuteResult`. * Out-of-order phase events (`tc.Phase != phase`) are silently dropped. ### Time semantics | Item | Source | Semantics | | -------------------------------- | ----------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | slot TTL | `RequestTaskSlot.ttl_ms` or `DefaultTTLMs` | Only controls automatic reclamation of `ALLOCATED` slots; not an execution timeout | | task deadline | `EffectiveTaskDeadlineMs`: `now + (task_timeout_ms or TaskDeadlineMs)`, clamped by `MaxTaskTimeoutMs` | On expiry, `TickTimers` posts `HandleTaskTimedOut`, converging to `FAILED / TIMED_OUT / retryable=true`. A negative `task_timeout_ms` or one above the cap returns `InvalidArgument` | | Result retention | `ResultRetentionMs` | After the terminal state, `taskIndex` keeps the `Result`; `PurgeExpiredResults` deletes it on expiry, after which `GetTaskResult` returns `NotFound` | | Watch grace | `WatchDisconnectGraceMs` (0 disables) | For a task that has had a Watch attached at least once, `WatchDisconnectDeadlineMs = now + grace` is set when the last stream disconnects; if nothing reconnects by then, the task converges to `FAILED / WATCH_DISCONNECTED / retryable=false` | | Function Code snapshot retention | `FunctionSnapshotRetentionMs` (0 means `MaxTaskTimeoutMs + 10s`) | `Validate` requires it to be no less than `MaxTaskTimeoutMs`, so that an epoch pinned by a task is not reclaimed | Idempotency semantics: both `RequestTaskSlot` and `SubmitTask` are idempotent by `taskId` within a single Worker. When the same `taskId` is already in `tasks`, `SubmitTask` returns the `RUNNING` snapshot; if it is already terminal and within the retention window, it returns the terminal snapshot without re-executing. Task reclamation reuses the `WatchTasks` stream to detect disconnects; there is no separate owner lease and no Cancel API. The Watch grace row in the table above is exactly this path, and `WatchDisconnectGraceMs` defaults to `0`, so it is off by default. See `docs/specs/2026-07-16-task-owner-lease-and-cancellation.md` in the blockx repo for the design review. ### Startup and shutdown order `Run` proceeds in order: validate the `Profile` → load configuration (built-in defaults → `TuneDefaults` → `WORKER_CONFIG` JSON → environment variables) → initialize tracing / Prometheus / chlog → wire the IO backend, Function Code View, registries, `Orchestrator`, `SubscriptionManager`, and `GRPCServer` → start the UDS server and the executor `PoolManager` → `RunDispatchLoop` → initial `SyncExecutorSnapshots` → 500ms ticker → gRPC `Serve` → only after the first healthy executor appears (`WorkerCore.SetReady`) does it register with etcd and start heartbeats. Phase 1 drain: stop registration → `orch.SetDraining()` (new `RequestTaskSlot` calls and not-yet-activated `SubmitTask` calls return `Unavailable`) → etcd `Deregister` → `WaitDrain` until the active task count reaches 0 or `DrainTimeoutMs` expires. Phase 2 force: cancel the lifecycle ctx → `poolMgr.Stop()` → `execAdapter.Close()` → `CloseIOScope()` → `subMgr.CloseAll()` → `GracefulStop` (then `Stop` after 5s). ## Configuration The configuration struct is `app.WorkerFullConfig` (`internal/worker/app/config.go`). Load order: `DefaultWorkerFullConfig()` → `Profile.TuneDefaults` → the JSON file pointed to by `WORKER_CONFIG` (`LoadWorkerFullConfigFrom`, which only overrides fields present in the file) → environment variables (per-field `envOverride*` in `app.Run`). An explicitly written file / env value always wins, which is also the channel for rolling back a profile's defaults. | Field (JSON path) | Environment variable | Default | Description | | ---------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------- | ------------------------------------------------------------------------- | ------------------------------------------------------------------------------ | | `listenAddr` | `WORKER_LISTEN` | `:8081` | gRPC listen address | | `workerAddr` | `WORKER_ADDR` | Empty; derived at startup from the `listenAddr` port plus the outbound IP | Externally advertised address, also the etcd key suffix | | `etcdEndpoints` | `ETCD_ENDPOINTS` | `localhost:2379` | Set to `none` or empty to skip registration | | `workerRegistryPrefix` | `WORKER_REGISTRY_PREFIX` | Provided by the Profile (`/blockx/workers/` or `/blockx/bundle-workers/`) | Must match `Deployment` | | `executorCount` | `EXECUTOR_COUNT` | `8` | Number of Python executor processes | | `worker.taskSlots` | `TASK_SLOTS` | `50` | Slot pool size | | `worker.defaultTTLMs` | None | `5000` | TTL for inline reservations | | `worker.taskDeadlineMs` | `TASK_DEADLINE_MS` | `300000` | Default task timeout | | `worker.maxTaskTimeoutMs` | `MAX_TASK_TIMEOUT_MS` | `28800000` (8h) | Upper bound for `task_timeout_ms` | | `worker.resultRetentionMs` | `RESULT_RETENTION_MS` | `300000` | Retention window for terminal results | | `worker.watchDisconnectGraceMs` | `WATCH_DISCONNECT_GRACE_MS` | `0` | Watch-disconnect reclamation; 0 disables | | `worker.defaultMaxAttempts` / `defaultBaseBackoffMs` / `defaultRetryJitterPercent` | `CALL_RETRY_MAX_ATTEMPTS` / `CALL_RETRY_BASE_BACKOFF_MS` / `CALL_RETRY_JITTER_PERCENT` | `3` / `10` / `5` | Written into each task's `CallRetryPolicy` | | `admission.builderSlots` / `executorTaskSlots` / `pluginSlots` / `scanSlots` | Only `EXECUTOR_TASK_SLOTS` and `SCAN_SLOTS` | `16` / `32` / `16` / `8` | Phase-level admission | | `streamBuild.enabled` | `STREAM_BUILD_ENABLED` | `false` (`true` in the bundle profile) | Streaming build switch | | `heartbeatIntervalMs` | `WORKER_HEARTBEAT_INTERVAL` (Go duration format, e.g. `2s`) | `200` | Heartbeat publish interval; the code comment recommends 2000 for production | | `drainTimeoutMs` | `DRAIN_TIMEOUT_MS` | `30000` | Upper bound on the drain wait during shutdown | | `phaseIoTimeoutMs` | `PHASE_IO_TIMEOUT_MS` | `10000` | IO timeout for the Builder / Writer phases, independent of the per-call budget | | `grpcKeepaliveTimeMs` / `grpcKeepaliveTimeoutMs` | `WORKER_GRPC_KEEPALIVE_TIME_MS` / `WORKER_GRPC_KEEPALIVE_TIMEOUT_MS` | `30000` / `10000` | Determine the worst-case latency for confirming a Watch disconnect | | `executorRestartEnabled` and related | `EXECUTOR_RESTART_ENABLED` | `true`, delay `1000`, maxRestarts `10`, window `60000`, backoff `30000` | Pool Manager automatic restart | | `functionSnapshotRetentionMs` | `FUNCTION_SNAPSHOT_RETENTION_MS` | `0` (= `maxTaskTimeoutMs + 10s`) | Must be ≥ `maxTaskTimeoutMs` | `core.Config.Validate` requires `taskSlots > 0`, `taskDeadlineMs <= maxTaskTimeoutMs`, and `defaultRetryJitterPercent` between 0 and 5. See [Call execution subsystem](/en/components/call-execution) for the dispatcher-related fields (`dispatcher.*`). ## Extension points * **Adding a Call Builder**: implement `cbtypes.CallBuilder` (optionally `StreamingCallBuilder`) in `internal/plugin/callbuilder/`, define a `CallBuilderName` in `cbtypes`, add a case to the `newBuilderRegistry` switch in `internal/worker/app/app.go`, and add the name to `Builders` in the Profile of every `cmd/*/main.go` that needs it. A builder not listed in the Profile is not registered, and tasks fail with `BUILDER_NOT_FOUND`. See [Plugin system](/en/components/plugins) for details. * **Adding a Writer Plugin**: implement `evtypes.WriterPlugin`, add a case in `newPluginRegistry`, and add it to `Plugins` in the Profile. If the plugin depends on a specific backend address, add validation in `validateProfileRuntimeConfig`. * **Adding an entry point / deployment form**: write only a new `cmd//main.go` that declares an `app.Profile` and calls `app.Run`. Do not hand-roll wiring in the entry point. `Profile.Validate` currently accepts only the `block` and `bundle` `Deployment` values. * **Adding a configuration item**: add the field and default in `WorkerFullConfig`, add an `envOverride*` in `app.Run`, and add a constraint in `Validate` if needed. * **Changing slot / phase / idempotency semantics**: change only `internal/worker/core/worker.go`, and add core unit tests in `internal/worker/core/worker*_test.go`. No new state decisions should appear in the adapter. * **Changing phase execution side effects**: `orchestrator_phases.go`. Any new path that touches `dispCore` must respect the I1–I5 invariants in the header of `orchestrator_actor.go`; `actor_invariants_test.go` catches violations. * **Adding a public RPC**: change `api/grpc/worker/worker.proto`, regenerate `workerpb`, add the handler in `grpc_server.go`, and put the pb conversion in `wire.go`. ## Testing Tests are organized in three layers per `docs/specs/worker-test-organization.md`: * **core unit** (`internal/worker/core/`): `worker_test.go`, `worker_activation_test.go`, and `worker_watch_test.go` cover `WorkerCore`; `dispatcher_*_test.go` covers `DispatcherCore`. No transport or processes are involved. * **adapter integration** (`internal/worker/adapters/`): `orchestrator_*_test.go` (lifecycle, phases, dispatch, stream, budget, deadline\_repro, subcall, send\_failure, and other slices), `grpc_server_test.go`, `stream_subscriber_test.go`, `etcd_publisher_test.go`, `actor_invariants_test.go`. They use a fake executor adapter and do not start Python. `internal/worker/app/*_test.go` covers the Profile, configuration parsing, and etcd registration wiring. * **process e2e** (`cmd/worker/worker_process_*_test.go`, `cmd/bundle_worker/*_test.go`): starts real worker processes and Python executors to verify the public gRPC contract (activation, watch, shutdown, timing, subcall, io, function code, and other slices). ```bash theme={null} # core + adapters + app (recommended with -race) go test -race ./internal/worker/... # process E2E (install the Python dependencies first) uv sync --project python go test -v -timeout 300s ./cmd/worker/... go test -v -timeout 120s ./cmd/bundle_worker/... ``` File naming rule: `_test.go` or `__test.go`; each file belongs to exactly one implementation boundary and one test layer. See [Test organization and commands](/en/development/testing) for more commands. ## Related docs Specs in the blockx repo: * `docs/specs/worker.md`: detailed Worker design. Protocol, state model, execution flow, Executor Pool Manager, heartbeat and failure semantics. * `docs/specs/architecture.md` §4.2.1 / §4.2.2: the Worker's responsibilities in the overall architecture, plus the three layers of concurrency control: phase-level admission, the per-task ramp-up window, and backend admission. * `docs/specs/worker-test-organization.md`: the Worker test organization standard. * `docs/specs/2026-07-16-task-owner-lease-and-cancellation.md`: review record for the Watch-disconnect reclamation design (`watchDisconnectGraceMs`). * `docs/specs/2026-07-28-registry-prefix-migration.md`: format and migration of `workerRegistryPrefix`. Site pages: dispatcher, executor adapter, subcalls, and the wedge safeguard Who calls RequestTaskSlot, and how the Coordinator consumes heartbeats Call Builder and Writer Plugin interfaces Where taskCodeEpoch and code snapshots come from TaskIOScope and backend admission gRPC / etcd protocol details