Call execution subsystem

​

Responsibilities and boundaries

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

• Generating tasks, allocating slots, or placing Workers (owned by the Coordinator and the Worker).
• Function source snapshots and auditing themselves (owned by Function Code View); this page only consumes TaskFunctionView.ResolveForDispatch.
• Real BlockDB / RPC access (owned by the IO subsystem); Executors never connect to external dependencies directly.

• 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

​

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.

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

​

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() (<SocketDir>/blockx-worker-<pid>.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

​

State and lifecycle

• 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

​

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

# 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 and docs/specs/worker-test-organization.md in the blockx repo for the test organization conventions.

​

Related docs

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

• Worker, Python Executor, Function Code View, IO access subsystem, Plugin system
• Task lifecycle, Protocols and interfaces