Worker

​

Responsibilities and boundaries

• 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 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.

• 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 for that part.

• 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

​

Core types and interfaces

• 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.

// 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

• 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.

• 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

• 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. 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.

​

State and lifecycle

​

Slot state machine

• 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

• 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

​

Startup and shutdown order

​

Configuration

​

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 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/<name>/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

• 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).

# 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/...

​

Related docs

• 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.