Coordinator

​

Responsibilities and boundaries

• Watching Worker registrations and heartbeats in etcd and maintaining a local, disposable WorkerView.
• For each ReserveWorkerSlot, performing validation, the readiness check, candidate filtering and ranking, and then trying a small number of candidates serially.
• Passing the slotId returned by the Worker through to the Client as-is, without interpreting its internal meaning.
• Applying local penalties and short circuit-breaking to Workers based on recent failures.

• Parsing the task payload, generating taskId, or receiving SubmitTask or TaskResult.
• Maintaining a centralized slot ledger, a pending queue, or global taskId deduplication.
• Writing any data to etcd; etcd only carries Worker discovery and heartbeats.

• The Worker’s local slot state machine is the single source of truth; a Coordinator restart only loses the view cache and routing bias, and does not affect correctness.
• CoordinatorCore is a Sans-IO pure decision engine: it never reads the clock or touches the network, and every method receives now (UTC milliseconds) explicitly.
• Before the first full Worker view is established, and after a watch outage exceeds the grace period, ReserveWorkerSlot always returns Unavailable.
• A single request tries candidates serially with no hedged requests; when the outcome is uncertain, it retries only against the same Worker with the same taskId, and never switches Workers.
• The same code runs as two sets of processes with different app.Profile values: cmd/coordinator (block cluster) and cmd/bundle_coordinator (bundle cluster). See Bundle clusters for bundle cluster details.

​

Code location

​

Core types and interfaces

​

core

• core.Config (internal/coordinator/core/coordinator.go): selection and protection parameters; DefaultConfig() provides the defaults, and Validate() requires WatchGracePeriodMs <= HeartbeatTimeoutMs.
• core.CoordinatorCore: holds map[string]*WorkerView, initialized, and watchDisconnectedAtMs. Event entry points: ApplyFullWorkerList, ApplyWorkerUpdate, ApplyWorkerRemoval, MarkWatchDisconnected, MarkWatchReconnected. Decision outputs: IsReady, SelectCandidate. Feedback entry points: RecordSlotSuccess, RecordSlotRejection, RecordSlotBackpressure, RecordSlotTimeout.
• core.WorkerView (internal/coordinator/core/worker_view.go): embeds etcd.WorkerHeartbeat and adds three fields that exist only locally and are never written back to etcd.

type WorkerView struct {
	WorkerHeartbeat

	// Local-only fields — not from etcd, not written back.
	CircuitBreakerDeadlineMs int64
	RecentFailureCount       int
	Penalty                  float64
}

• WorkerView.UsedTotal() = ReservedSlots + RunningTasks; WorkerView.FreeCapacity() = TaskSlots - UsedTotal().
• core.CandidateResult{WorkerAddr, Err} and core.CoordinatorError{Code, Message}: the output of SelectCandidate; Code is a commontypes.ErrorCode.
• core.ValidateReserveRequest(taskID) (internal/coordinator/core/validate.go): only checks that taskId is non-empty.

​

adapters

• adapters.CoordinatorServer (grpc_server.go): implements coordinatorpb.CoordinatorServiceServer; holds core, a SlotReserver, and a sync.RWMutex. Lock/Unlock are exposed so the watcher can write to the core.
• adapters.HandlerConfig{MaxTimeoutRetries}: number of retries against the same Worker when the outcome is uncertain; defaults to 1.
• adapters.SlotReserver interface: RequestTaskSlot(ctx, workerAddr, req) (*commontypes.TaskSnapshot, *SlotError); replaced with a mock in tests.
• adapters.WorkerRPCClient (worker_rpc.go): the gRPC implementation of SlotReserver; caches one grpc.ClientConn per workerAddr. grpcErrToSlotError decides SlotError.MayHaveAllocated.
• adapters.SlotError{Code, Message, MayHaveAllocated}: MayHaveAllocated=true means the request may have reached the Worker, so retries must stay on the same Worker.
• adapters.EtcdWorkerWatcher / EtcdWatcherConfig{Endpoints, KeyPrefixes, DialTimeout, RetryDelay} (etcd_watcher.go): Run(ctx) blocks while running list+watch.
• adapters.BundleCoordinatorServer (bundle_grpc_server.go): a thin translation layer from the bundle proto to CoordinatorServer.

​

app and api

• app.Profile{Deployment, Service, WorkerRegistryPrefixes} (internal/coordinator/app/app.go): the entry point’s identity. Deployment must be "block" or "bundle", and determines which gRPC service is registered and which registry kind is allowed.
• app.Run(p Profile): wires obs, core, WorkerRPCClient, CoordinatorServer, the etcd client, the watcher, and the gRPC server, then blocks until SIGINT/SIGTERM.
• app.CoordinatorRuntimeConfig (config.go): the process-level configuration aggregate; LoadCoordinatorRuntimeConfigFromEnv(p) applies environment variable overrides and calls Validate().
• etcd.WorkerHeartbeat (api/etcd/types.go): the JSON value the Worker writes to etcd, with fields workerAddr, taskSlots, reservedSlots, runningTasks, cpuPercent, memoryPercent, memoryAvailableBytes, memoryTotalBytes, processRssBytes, errorRate, and lastHeartbeatMs.
• etcd.ParseRegistryPrefix and the constants LegacyBlockWorkerRegistryPrefix (/blockx/workers/), LegacyBundleWorkerRegistryPrefix (/blockx/bundle-workers/), ProdDefaultBlockWorkerRegistryPrefix, ProdDefaultBundleWorkerRegistryPrefix, TestDefaultBundleWorkerRegistryPrefix (api/etcd/registry.go).

​

Data flow / execution flow

​

Public error codes

​

State and lifecycle

• initialized: becomes true after the first ApplyFullWorkerList call and never reverts. The watcher calls it only after every configured prefix has completed its initial list.
• watchDisconnectedAtMs: when the watch on any prefix disconnects, MarkWatchDisconnected(now) records the time of the first disconnect; IsReady and SelectCandidate return Unavailable once now - watchDisconnectedAtMs > WatchGracePeriodMs. After every prefix has re-listed successfully and all are connected again, MarkWatchReconnected clears it.

• RecentFailureCount and Penalty: RecordSlotRejection/RecordSlotTimeout each increment by one, and Penalty = float64(RecentFailureCount).
• CircuitBreakerDeadlineMs: set to now + CircuitBreakerCooldownMs (default 10s) when RecentFailureCount >= CircuitBreakerThreshold (default 3); the Worker is hard-filtered for that period.
• RecordSlotSuccess clears all three together; RecordSlotBackpressure leaves all three untouched.
• ApplyFullWorkerList preserves these three fields for surviving Workers; a Worker that reappears after being deleted by ApplyWorkerRemoval starts from zero.

• The slot TTL (SlotTTLMs, default 5s) is the reservation window the Coordinator passes to the Worker in RequestTaskSlot; it is not a task execution timeout, and the Coordinator does not interpret it further.
• A single ReserveWorkerSlot has no separate overall time budget; the upper bound is determined jointly by MaxCandidates × (1 + MaxTimeoutRetries) × WorkerRPCTimeout and the caller’s ctx, roughly 3 × 2 × 3s in the default worst case.
• Heartbeat freshness is computed by subtracting the Worker-written lastHeartbeatMs from the Coordinator’s local clock, so clock skew between machines directly affects the filtering result.

​

Configuration

• cmd/coordinator: /blockx/workers/.
• cmd/bundle_coordinator: /blockx/bundle-workers/ and /blockx/prod/lanes/default/bundle-workers/.
• Test clusters replace them entirely with COORDINATOR_REGISTRY_PREFIXES=/blockx/test/lanes/default/bundle-workers/.

​

Extension points

• Changing the selection policy: touch only SelectCandidate and candidateRanksBefore in internal/coordinator/core/coordinator.go, and add TestSelectCandidate_* cases in coordinator_test.go. The core has no IO, so it can be tested as pure functions. Ranking is currently deterministic; the absolute memory field in the heartbeat (memoryAvailableBytes) is only parsed and takes no part in filtering or ranking.
• Adding a Worker feedback type: add a RecordSlotXxx to the core and dispatch on SlotError.Code in recordSlotOutcome in adapters/grpc_server.go. That is how RecordSlotBackpressure was added.
• Changing Worker error classification: grpcErrToSlotError in adapters/worker_rpc.go and workerCodeToGRPC on the Worker side (internal/worker/adapters/wire.go) must remain inverses of each other; if you change one, check the other (see Worker).
• Adding a deployment form: add a cmd/<name>/main.go that declares an app.Profile, add branches in app.registerCoordinatorService and validateRegistryPrefixes, and allow the new worker kind in ParseRegistryPrefix in api/etcd/registry.go.
• Replacing the Worker RPC implementation: implement the adapters.SlotReserver interface; CoordinatorServer does not depend on a concrete client. The current WorkerRPCClient opens exactly one grpc.ClientConn per workerAddr and does not use the multi-connection balancer from internal/sdk/grpcclient: a point-to-point client targeting a single Worker does not enable a multi-connection strategy. See docs/specs/2026-07-23-grpc-client-balancing.md in the blockx repo for background.
• Adding a configuration item: add the field and its Validate rule in core.Config / adapters.HandlerConfig / adapters.EtcdWatcherConfig, then add environment variable parsing in LoadCoordinatorRuntimeConfigFromEnv in app/config.go, and extend config_test.go.

​

Testing

# core / adapters / app unit tests, plus registry prefix parsing
go test ./internal/coordinator/... ./api/etcd/...

# Process-level tests: build the real binary, start embedded etcd, verify gRPC readiness and error mapping
go test ./cmd/coordinator/... ./cmd/bundle_coordinator/...

# Contract-level E2E (includes coordinator-etcd and coordinator-worker)
go test -v -timeout 300s ./e2e/contract/...

• internal/coordinator/core/coordinator_test.go: every SelectCandidate filter condition and ranking rule, circuit breaker trip/recovery, the grace period semantics of IsReady, ApplyFullWorkerList preserving local fields.
• internal/coordinator/adapters/handler_test.go: covers the orchestration loop with a mock SlotReserver (switching candidates after a timeout, returning Aborted when retries still time out, returning ResourceExhausted when all reject, retrying on an empty SlotID), plus the error codes and compression behavior of the real gRPC server.
• internal/coordinator/adapters/worker_rpc_test.go: MayHaveAllocated classification in grpcErrToSlotError (connection refused vs timeout/cancel/Aborted).
• internal/coordinator/adapters/etcd_watcher_test.go: list+watch on embedded etcd, multi-prefix merge and deduplication, readiness state across disconnect/reconnect, skipping unparseable values.
• internal/coordinator/app/config_test.go, profile_test.go: environment variable parsing and Profile validation.
• e2e/contract/coordinator-etcd/: when etcd starts after the Coordinator, first Unavailable and then recovery; routing still works when the watch recovers within the grace period; becomes Unavailable beyond the grace period.
• e2e/contract/coordinator-worker/: an explicit transport failure switches to the next Worker, a RequestTaskSlot timeout returns SlotUncertain (Aborted), SubmitTask is rejected after the slot expires, an old slotId is invalid after a Worker restart, and rejection when the unified slot pool is full.

​

Related docs

• docs/specs/task-resource-coordinator.md: detailed Coordinator design; protocol, error semantics, selection rules, consistency boundaries.
• docs/specs/architecture.md §4.1: the control plane’s position in the overall architecture and its three boundaries.
• docs/specs/2026-07-28-registry-prefix-migration.md: legacy / v2 registry prefixes, multi-prefix merging, and rollout order.
• docs/specs/2026-07-23-bundle-coordinator-api.md: why the bundle Coordinator has a separate service identity.
• docs/specs/2026-07-23-grpc-client-balancing.md: the scope of the multi-connection balancer.
• docs/specs/worker.md: the authoritative definition of RequestTaskSlot and the slot state machine.

• Worker: the Worker-side implementation of RequestTaskSlot and etcd heartbeat publishing.
• Bundle clusters: bundle clusters and cmd/bundle_coordinator.
• Protocols and interfaces: gRPC / etcd protocol summary.
• Task lifecycle: the full path from ReserveWorkerSlot to SubmitTask.
• Observability: how metrics, logs, and tracing are wired in.