Distributed runtime and the cluster control plane

The control plane is a Coordinator process plus one or more Worker processes that talk a small binary protocol over TCP to deploy, run, checkpoint and recover jobs.

Overview

A clink cluster has exactly one active Coordinator (coordinator) and any number of Workers (workers). The coordinator is the single source of truth for deployment: it accepts job submissions from clients, plans them into per-subtask deployment tasks, hands those tasks to workers, tracks liveness and checkpoint progress, and reports completion. Each worker is a worker that registers with the coordinator, receives deployment tasks, runs them, and reports back. All three roles (coordinator, worker, client) speak one binary, length-prefixed framing defined in include/clink/cluster/protocol.hpp. This control plane is qualified at width by QUAL-06: a 147-operator SQL graph deployed as 292 network-bridged subtasks, exactly once under sustained fault injection, recovering a worker kill at that width in 28 seconds.

Jobs are not built into the cluster binaries. A job is a compiled shared library that the client ships with the submission; the coordinator caches it and re-ships it inside every deployment, and each worker dlopens it before running tasks. This is the job-as-plugin model (CLINK_REGISTER_JOB). An out-of-tree module builds with clink_add_job_module() from the installed CMake package (docs/consumer-examples/CMakeLists.txt shows the shape), the same recipe the repository's own job modules and the ABI gate's fixtures use; see design record 010 for the compatibility contract it carries.

Plugin bytes are content-addressed and ship at most once per receiver (item 30). A submission sends hash-only references first (PluginBinary with empty bytes); the coordinator resolves them from its cache and answers missing_plugin_hashes on the ack for anything it lacks, which the submitter uploads in one retry on the same connection - so every submit after a module's first carries no plugin bytes at all. On the deploy side the coordinator remembers, per worker connection, which hashes it has shipped and sends references thereafter; a re-registered worker starts from an empty set and receives full bytes again, matching its pid-keyed cache starting empty. A reference a receiver cannot resolve fails the submit or deploy loudly - it is a claim about the peer's cache, never permission to run without the module. clink_coordinator_plugin_bytes_shipped_total, ..._plugin_ships_deduped_total and ..._submit_plugin_cache_hits_total make the traffic visible.

On ELF the repository's own job modules are split-debug stripped after linking (cmake/ClinkPluginDebugSplit.cmake): the DWARF - which measured ~102 MB of a ~104 MB module against 2.6 MB of code - moves whole into <module>.so.debug beside the artefact, and the shipped module (~5 MB) carries a .gnu_debuglink naming it, so gdb and llvm-symbolizer resolve symbols from the same directory. The dynamic symbol table survives stripping, so the clink_plugin_* handshake, CLINK_REGISTER_JOB entry points and the ABI-hash check are unaffected; a submit measures ~0.45 s against ~2.7 s fat. The job_plugin_ships_stripped test pins the property (no .debug_info section, hard size budget). Consumers building their own modules against an installed clink get the same win from the same two commands: objcopy --only-keep-debug job.so job.so.debug && objcopy --strip-unneeded --add-gnu-debuglink=job.so.debug job.so. Keep the .so.debug from the build that shipped - crash addresses from any cached copy of the module symbolise against it. macOS modules are left alone: their DWARF never links into the module in the first place.

The single binary tools/clink_node.cpp runs as either a coordinator (--role=coordinator) or a worker (--role=worker); clients submit through the clink::application::JobSubmitter library rather than through clink_node.

Task placement co-locates a pipeline instance

assign_task_placement (include/clink/cluster/coordinator.hpp, implemented in src/cluster/coordinator.cpp) assigns a worker to every planned task, grouping the tasks that share a subtask index and placing each group entirely on one worker.

Tasks sharing a subtask index are joined by FORWARD edges: subtask i of the source feeds subtask i of the bridge feeds subtask i of the projection. The data plane hands a batch across a forward edge as a pointer when both ends live in one process (LocalDataPlane) and serialises it over a socket when they do not, so placement alone decides whether a forward edge costs a pointer copy or a network round trip.

It used to place one task at a time, greedy first-fit, in plan order. Plan order is operator-major - every subtask of the source, then every subtask of the bridge - so filling one worker's slots before touching the next scattered each instance across hosts. Measured on a 3-worker rig at parallelism 12, nexmark q0 (four operators, forward edges only, no shuffle at all) sent 16 of its 24 data-plane edges over TCP:

before   local=8   socket=16   67% of edges serialised
after    local=24  socket=0     0%

and its CPU-per-event was 2.05x worse at parallelism 12 than at 4, where the same sweep on a single host was flat to within 5%.

Two properties worth knowing:

• A hash-shuffled edge is unaffected. Subtask i sends to every downstream subtask regardless of placement, so co-location can only convert forward edges. Measured over loopback, a forward-edge-only query (q0) gained about 20% CPU efficiency while a shuffle-dominated one (q12) was neutral. Over a real network the forward-edge gain is larger than the loopback figure, because that is the cost being avoided.
• Placement is deterministic. Workers are visited in sorted worker-id order rather than in the coordinator registry's order, which is an unordered_map and neither sorted nor stable across processes. Placement was previously unrepeatable across deploys of the same plan, which also made any measurement of it unrepeatable.

An instance larger than any single worker's free capacity is split across whatever is free: splitting is worse than co-locating and much better than refusing to deploy. Running out of capacity entirely is a reported failure, not a silent partial placement.

tests/test_task_placement.cpp pins the contract, including the split and out-of-capacity paths. It was verified against a faithful reproduction of the old behaviour - and note that a first attempt at that reproduction passed, because iterating group-by-group co-locates by accident; it is the operator-major PLAN order that scatters an instance.

Where it lives

Concern	File(s)
Wire protocol, message kinds, framing	include/clink/cluster/protocol.hpp, encode/decode in include/clink/cluster/messages.hpp
Coordinator (accept, watchdog, dispatch, recovery)	include/clink/cluster/coordinator.hpp, src/cluster/coordinator.cpp
Worker (register, deploy, run, heartbeat)	include/clink/cluster/worker.hpp, src/cluster/worker.cpp
HA leader election (file coordinator)	include/clink/cluster/ha_coordinator.hpp, src/cluster/ha_coordinator.cpp
HA leader election (optional etcd coordinator)	impls/etcd/include/clink/etcd/etcd_ha_coordinator.hpp
coordinator endpoint discovery for workers	include/clink/cluster/service_discovery.hpp
Client submission library	include/clink/application/job_submitter.hpp, src/application/job_submitter.cpp
Plugin contract and registry	include/clink/plugin/plugin.hpp, include/clink/plugin/plugin_impl.hpp
Job-as-plugin macro	include/clink/job/register_job.hpp
Plugin loader and ABI gate	include/clink/cluster/plugin_loader.hpp, src/cluster/plugin_loader.cpp, generated include/clink/plugin/abi_version.hpp (from .in)
Process entry point	tools/clink_node.cpp

How it works

The wire protocol

Every message on a control connection is a length-prefixed frame: a 4-byte big-endian length followed by a payload, and the payload's first byte is a MessageKind (include/clink/cluster/protocol.hpp). String fields inside a payload are themselves [u32 length BE][bytes], and all multi-byte integers are big-endian. The framing is built by MessageBuilder and read by MessageReader in the same header; encode_frame(kind, msg) (in include/clink/cluster/messages.hpp) writes the kind byte, then the body, then prepends the length.

MessageReader owns its payload by value and bounds-checks every read (read_string, consume_byte_ throw on truncation). Many decoders treat trailing fields as optional by guarding on r.eof() so an older peer that sends a shorter frame is handled gracefully. For example decode_register reads slot_count and http_port only if more bytes remain.

MessageKind values are grouped by direction:

sequenceDiagram
  participant C as Client
  participant coordinator as Coordinator
  participant worker as Worker
  worker->>coordinator: Register
  coordinator->>worker: RegisterAck
  C->>coordinator: HelloClient
  C->>coordinator: SubmitJob (graph+plugins)
  coordinator->>worker: Deploy (tasks+plugins)
  worker->>coordinator: SubtaskListening
  coordinator->>worker: PeerUpdate
  coordinator->>C: SubmitJobAck
  Note over worker: subtasks run
  worker->>coordinator: SubtaskFinished
  coordinator->>C: JobCompleted

(*RescaleOperator = 12 and Savepoint = 13 are distinct values: the client-loop dispatch matches on the MessageKind byte, so every client -> coordinator request kind must be unique. CancelJob = 103 is deliberately overloaded for both the coordinator -> worker cancel broadcast and the client -> coordinator cancel request; it is disambiguated by connection direction, not by a shared value with another client request.)

Connection routing: the first frame

The coordinator's accept loop (accept_loop_ in src/cluster/coordinator.cpp) wraps each accepted file descriptor into a network::Connection via an injected AcceptFactory (plain TCP by default; a TLS factory when configured). It never parks in a blocking accept(): it waits in NetworkSocket::wait_for_connection, a poll() on the listener and its AcceptWake, and the factory's accept then runs on a non-blocking listener. A custom factory therefore must not assume a blocking listener or a blocking accepted socket: on Darwin an accepted socket inherits O_NONBLOCK, and NetworkSocket::accept_one is the accept that clears it (the contract is on set_accept_factory in include/clink/cluster/coordinator.hpp). stop() wakes the loop, joins it, and only then closes the listener. The wake never closes anything: on Linux it shuts the listener down (shutdown(SHUT_RD) wakes a poll() or accept() there and costs no descriptor), and on Darwin, which ignores shutdown() on a listener, it writes to a self-pipe. Closing the listener to wake the thread is unsafe on both platforms: on Darwin a close() that lands while the thread is entering accept() can miss it, and both then block until a connection arrives (the occasional ShutdownLeaks timeout under load); on Linux close() does not wake accept() at all, and the freed descriptor can be reissued to another listener in the process before the thread reaches it.

The accept thread only accepts. Everything a new connection does before it is admitted, its handshake and its first frame, runs on an admission thread of its own (begin_admission_ and admit_connection_), so a client that stalls either holds up no other connection. With TLS on, the factory clink_node installs runs the handshake non-blocking on that thread (handshake_accepted_tls_connection, which waits in NetworkSocket::wait_ready on the socket and the accept wake together), bounded by Config::handshake_timeout (default 5000 ms, --handshake-timeout-ms). A handshake that fails or misses that deadline closes the connection, logs one connection rejected line and increments clink_coordinator_handshake_failures_total. Each step of a server handshake (an SSL_accept call, where its CPU goes) takes one of the server context's handshake slots, half the hardware threads by default (TlsServerContext::set_max_concurrent_handshake_steps); the slot is never held while waiting on the client, so a flood of handshakes is bounded in the CPU it can take and a stalled handshake holds nothing. A step that cannot get a slot before its handshake deadline fails the handshake. Both TLS contexts refuse renegotiation (SSL_OP_NO_RENEGOTIATION): a TLS 1.2 renegotiation is a full handshake run inside SSL_read, on an admission thread reading a first frame or a session's reader, outside any slot, and OpenSSL 1.1.1 allows a client's by default. Nothing that abandons a handshake (a shutdown of its socket, the stop wake) can reach a thread waiting for a slot, so the wait runs in 10 ms slices and checks between them, and once more with the slot held, before the step computes: a handshake evicted or stopped while it waited gives up its thread within a slice and never spends a slot's CPU (on Linux a ClientHello that arrived before the shutdown is still readable, so the step would otherwise run the full key exchange for a dropped connection). The first frame is then read with heartbeat_timeout as a per-read bound, which drops a client that sends nothing, and under kMaxFirstFrameBytes (64 KiB, frame_io.hpp), since a Register or HelloClient is a few hundred bytes and nothing has been authenticated yet; the 256 MiB cap applies only to the frames an established connection sends later.

Per-read bounds do not stop a client that sends a byte at a time, each inside the bound. So the whole of admission has a deadline too: a connection that has not finished its handshake and sent its whole first frame within handshake_timeout + heartbeat_timeout of being accepted is closed. The accept thread keeps the deadlines, waking for the earliest one, and closes an overdue connection by shutting down a duplicate of its socket (or, for a listener-form factory's connection, with Connection::shutdown_read), which fails whatever its admission thread is waiting in. stop() does the same to every connection still being admitted before it joins the accept thread and the admission threads, including those whose first frame is in (their duplicate is kept until handle_first_frame_ returns), so it never waits on a handshake, a first frame, or a reply to a peer that does not read.

handle_first_frame_ answers some peers before they are admitted: a refusal (an incompatible protocol version, a full worker or client cap, a stale or premature re-registration) or the RegisterAck itself. It does so on the admission thread, under the first-frame lock, and for the RegisterAck under mu_, so a peer that never reads must not be able to hold that send. Each such reply goes through send_admission_reply_, which bounds every send() by Coordinator::kAdmissionReplySendBound (250 ms, or heartbeat_timeout if shorter; Connection::set_send_timeout, SO_SNDTIMEO), and a Register's worker_id and data_host are refused at decode beyond kMaxRegisterStringBytes (255 bytes, protocol.hpp), so no reply that echoes them is larger than a fresh socket's send buffer. A send() waits only while that buffer is full, which a peer waiting for its reply never brings about, so the bound is short. A peer that does not read can still fill it: over TLS, OpenSSL writes from inside the first frame's reads (an alert for each renegotiation it refuses, a reply to a TLS 1.3 KeyUpdate), so the bound is how long such a peer holds the first-frame lock, and mu_ for a RegisterAck. Registration refusals are decided under mu_ and sent after it is released. Deadlines saturate: a timeout too large to add to the clock is no deadline, never one that has already passed.

At most Config::max_pending_connections (default 64, --max-pending-connections) connections are admitted at once, counting each from its accept until handle_first_frame_ has taken it, so the threads whose first frame is in and that wait to be registered are bounded too. A connection beyond the cap evicts one or is refused (choose_admission_to_evict in include/clink/cluster/admission_policy.hpp). The peer address holding the most admissions gives way first (an IPv6 peer counts by its /64), and within it the connection that has got least far: accepted only, then handshake completed, then first frame begun. A connection whose first frame has arrived is never evicted. A newcomer that would itself be the least advanced in the busiest address is refused instead, so a worker whose TLS handshake is done (under mTLS, authenticated) never gives way to a stranger that has sent nothing. The bound this gives: a flood from one host evicts that host's own connections, and a worker from another host is evicted only when every address holds at most one admission and the worker is among the least advanced and oldest. The only admission of an address is not evicted on a tie while another address holds more: those are then all first frames waiting on registration, and evicting for them would let one host's queue push workers from other hosts out in turn. When nothing can be evicted (every slot is a first frame waiting on a slow registration, or further along than the newcomer), a newcomer from an address that holds no admission is still admitted, beyond the cap, so one host's queued first frames do not turn away a worker from another. So is a newcomer whose address's admissions all have their whole first frame in: they wait on registration alone, so a burst of workers starting together on one host is admitted while registration works through it, up to the ceiling below, where the listen backlog used to hold it. What the cap still limits for one address is how many of its connections may be mid-handshake or mid-first-frame at once: past it the oldest of those is evicted, or the newcomer refused when they are all further along. A WorkerSupervisor retries either, so set max_pending_connections above the number of workers that start together behind one address, especially with TLS, whose handshakes take long enough for a whole simultaneous start to be under way at once. Refusals count in clink_coordinator_handshake_failures_total. Each refusal, eviction and other drop before admission is logged, but a refusal costs a client only a connect, so the lines are rate-limited (AdmissionLogLimiter): five in full per ten seconds, then one summary line with the number left out and the address that sent the most. The counters, not the log, are the precise record.

Eviction interrupts an admission; it does not end its thread, which lets go of its socket only once it notices. So the cap is not the bound on threads and descriptors: every admission still tracked, interrupted ones included, counts towards a ceiling of twice max_pending_connections, each towards its own address. At the ceiling nothing more is evicted, and a new connection is admitted only if its address holds nothing, or less than its fair share of the ceiling (the ceiling divided by the addresses holding admissions, its own included). One host reconnecting fast evicts its own admissions below the ceiling, and if their threads are slow to let go (a TLS handshake waiting for a slot notices within a 10 ms slice) it fills the ceiling with its own and is refused there, while a worker from another host still gets in. At three times max_pending_connections every new connection is refused, whatever the existing admission threads are blocked in: that hard limit is the bound on admission threads and their descriptors, and a flood from that many distinct addresses can still reach it until their deadlines pass. An interrupted admission's duplicate descriptor is closed at the interrupt, so each one waiting to let go holds the one socket its thread owns; a live admission holds two (its socket and the duplicate), so admissions can take up to six times max_pending_connections descriptors on top of the worker and client connections. When accept() fails for want of descriptors or memory (EMFILE, ENFILE, ENOBUFS, ENOMEM), the connection stays in the listen backlog and the listener stays readable, so the accept thread waits on the stop wake alone before trying again, from 10 ms doubling to a second, and logs the first failure of a run and the count once accepting resumes, rather than spinning on a core until a descriptor frees. A connection that accept() takes with no descriptor left for its duplicate is closed rather than left in the backlog, and counts in the same run; its worker's supervisor retries it. A listener-form factory reports a failed accept by throwing, so after a throw the accept thread checks whether a descriptor is free and, when none is, backs off the same way. The accepted socket itself is owned by the accept thread until the admission thread starts, so a throw in the bookkeeping between them closes it rather than leaking it.

Isolation between hosts rests on the coordinator seeing each peer's own address. Behind a hop that rewrites it (an L4 load balancer or HA virtual IP that SNATs, a NAT gateway), or with several workers on one host, every peer behind it is one source, and among them only progress separates connections: one that has sent nothing can evict a worker from the same address whose TLS handshake is still under way. A listener-form factory's connections all share the empty source, and arrive with their handshake done, so among them a newcomer evicts the oldest admission that has got no further than its handshake: without mTLS, a client that completes handshakes and then sends nothing can evict workers there, the weaker guarantee of that form. Keep SNAT out of the path to the control port, or require mTLS, under which a worker past its handshake is never evicted for a connection that has not got as far.

Admission threads hand their connections to handle_first_frame_ one at a time, under one lock, but not in accept order: two connections' first frames can be handled in either order. So each registration carries its admission id (rising in accept order), and a registration from a connection accepted before the one the worker's current session registered on is refused. Without that, a worker whose first attempt timed out waiting for its RegisterAck, and which reconnected and registered again, could have the abandoned attempt handled second and replace the live session. Nothing an admission thread does can end the process: an exception on it (a std::bad_alloc while a frame grows, a throw from a custom Connection) drops that connection, logs it and counts it as a malformed frame.

Before this, the accept thread ran every handshake as a blocking SSL_accept with no deadline, and read every first frame itself: one client that connected and never finished its handshake (a port scanner, a plain-TCP client on the TLS port) stopped every later worker registering and left stop() joining the thread for ever, and a queue of silent clients cost each later connection a full timeout per client. A custom factory installed in the AcceptRequest form of set_accept_factory is handed the accepted socket, the handshake deadline and the wake, runs on the admission thread, and must be safe to call from several threads at once. One installed in the older listener form does the accept itself, so it and any handshake it does run on the accept thread, and must bound that handshake. The coordinator then dispatches on the frame's kind in handle_first_frame_:

• Register -> the connection becomes a long-lived worker connection. The coordinator records the WorkerConnection, replies RegisterAck, and starts three threads for it (start_reader_for_): a writer, which alone writes everything the connection is sent after its RegisterAck (see Outbound frames to a worker), a reader, and a dispatch thread that handles the connection's frames in the order the reader read them.
• HelloClient -> the connection becomes a client connection, owned as a shared_ptr and serviced by a per-client thread (handle_client_loop_) that reads SubmitJob, ListJobs, CancelJob, RescaleJob, RescaleOperator and Savepoint frames.
• Anything else is a protocol violation and the connection is dropped.

client                         Coordinator                        Worker
  |                                |                                  |
  |                                |<----- Register ------------------|
  |                                |------ RegisterAck -------------->|
  |--- HelloClient --------------->|                                  |
  |--- SubmitJob (graph+plugins)-->|                                  |
  |                                |--- Deploy (tasks+plugins) ------>|
  |                                |<---- SubtaskListening -----------|
  |                                |---- PeerUpdate ----------------->|
  |<-- SubmitJobAck ---------------|        (subtasks run)            |
  |                                |<---- SubtaskFinished ------------|
  |<-- JobCompleted ---------------|                                  |

Worker registration and lifecycle

Worker::connect_to_coordinator (src/cluster/worker.cpp) opens the coordinator connection (plain TCP by default, via an injectable ConnectFactory), sends Register (carrying worker_id, data_host, slot_count, and an advertised HTTP port), and waits for RegisterAck. It then starts a reader thread (reader_loop_) and, if heartbeat_interval > 0, a heartbeat thread (heartbeat_loop_) that sends a Heartbeat frame on each tick (default interval 500 ms, Worker::Config::heartbeat_interval).

reader_loop_ dispatches inbound coordinator frames: Deploy, PeerUpdate, CancelJob, StartJob, TriggerCheckpoint, CommitCheckpoint, AbortCheckpoint, FinalCheckpointAssigned, BeginRescale, CutoverPeerUpdate and CutoverRebind (the last three are the hot-rescale control surface: an arming BeginRescale names the cutover checkpoint and the worker replies with BeginRescaleAck reporting what it armed; CutoverRebind binds new inbound listeners for the post-cutover subtasks; CutoverPeerUpdate carries per-feeding-task endpoint lists that swap the held output groups onto the new peers - see ./fault-tolerance-and-rescale.md). When the coordinator connection drops (read_frame returns no value), the loop sets disconnected_, wakes any subtasks blocked waiting for a PeerUpdate, and returns.

On Deploy (handle_deploy_), the worker allocates (or reuses) a per-job JobBundle, loads any plugin libraries shipped with the deploy into that bundle's registry, registers a pending-task record per subtask, stashes the job's checkpoint config, and spawns one task thread per DeploymentTask. For the built-in generic subtask role (kGenericSubtaskRole, the string "__clink_subtask" in include/clink/cluster/job_planner.hpp), each task thread runs run_generic_subtask_, which:

1. parses the OperatorChainSpec out of DeploymentTask::extra_config,
2. binds one inbound network bridge per input edge,
3. sends SubtaskListening reporting the bound port(s) (sources send an empty edge_ports list, which the coordinator uses purely as a "ready" tick),
4. waits for the coordinator's PeerUpdate (bounded by peer_update_timeout, default 30 s), then
5. builds and runs the operator chain through the local executor.

A CancelJob reaches a task wherever it is in those steps. A task waiting for its PeerUpdate stops waiting and exits. Past that wait, every task registers a cancel token before step 5, and CancelJob flips the tokens registered at that moment: a task that runs one operator through its registered runner registers it just before the runner starts, and a chain of operators (or one with a fused source or sink), like an operator run by a built-in dispatch arm, registers it before it builds anything and hands it to its executor as JobConfig::external_cancel_token. Such a task still being set up when the cancel arrives finds its job in the worker's cancelled set (cancelled_jobs_, cleared by the job's next Deploy) and is not run, so it builds no operator at all. A cancel that arrives once the operators are being built waits for that build: a factory cannot be interrupted, and the executor reads the token only once it runs, cancelling every runner as soon as it starts. Before chains registered a token, a chain stopped only when its network peers closed, and one whose upstream chains failed while building, after the PeerUpdate, waited in its bridges for good: the job's restart drain waited for it until the watchdog failed the job, and the chain outlived the coordinator. ErrorCancelConvergence.AChainWhoseUpstreamNeverConnectedIsCancelledWithItsJob holds that shape. A cancelled task reports the operator errors its executor caught before the token was set (LocalExecutor::operator_errors_before_cancel()) and drops the rest, which are its teardown's: a closed channel, a refused send. An operator that failed on its own and was cancelled only afterwards, with its peers, therefore still names its own failure.

The two-phase listen/peer-update handshake is how the coordinator resolves the actual host:port each upstream subtask should connect to: subtasks bind ephemeral ports first, report them, and the coordinator fans the resolved peer addresses back out once every subtask of the job has reported. See ./network-stack.md for the data-plane bridges and ./task-lifecycle.md for what runs inside a subtask.

Port discovery and peer resolution

The coordinator tracks per-job port state in JobState. Each SubtaskListening (handle_subtask_listening_) records the listening port keyed by the four-tuple (downstream_role, downstream_subtask, upstream_role, upstream_subtask) so that a multi-input subtask (union or join) which binds several inbound listeners can be matched to the correct upstream for each. When received_listenings == expected_listenings, the coordinator resolves every task's peers[] against that port map and broadcasts PeerUpdate per worker (send_peer_updates_locked_). Subtasks with no peers still receive an empty PeerUpdate as a "go" signal.

Heartbeats, the watchdog and lost-worker detection

Each worker's reader thread on the coordinator side stamps last_seen on every frame it reads (start_reader_for_), so any message, not just Heartbeat, counts as liveness. The reader answers each Heartbeat itself, posting the fenced HeartbeatAck to the connection's outbox rather than sending it, and queues every other frame for the connection's FIFO dispatch thread (worker_dispatch_loop_), which runs the handlers one at a time. A handler that waits on the store - the COMPLETED and CONFIRMED marker writes, or the job-completion records written under the coordinator's lock - therefore delays that worker's later frames but never its heartbeats or their answers, and a worker that stops reading cannot hold its reader in a send either. When the reader also ran the handlers, a marker write on a slow object store that outlasted heartbeat_timeout got a live worker declared lost and turned a completed checkpoint into a restart, and a completion record held under the lock stopped every worker's heartbeats at once, long enough to end every v2 worker's session. Liveness now means the worker's socket delivers frames: a handler stalled for ever no longer gets its worker declared lost, and one stalled past restart_drain_timeout during a restart keeps that restart's drain open until the watchdog fails the job, as it does for any survivor that heartbeats but does not drain. Frames of a session declared lost are dropped. The dispatch thread checks before it hands a frame to its handler, and every handler checks again in its first hold of the coordinator's lock: a frame is stamped as proof of life when it is read and can be handled any time later, so the watchdog can declare the session lost in between, and a SubtaskFinished handled after the loss fold was counted a second time (a job completing with a subtask still running elsewhere) or reported a drain of a subtask the loss had killed. A dedicated watchdog thread (watchdog_loop_) wakes every Config::watchdog_interval (default 100 ms) and declares a worker lost when now - last_seen > Config::heartbeat_timeout (default 2000 ms). Because a healthy worker heartbeats every 500 ms, several missed heartbeats are needed before a false positive.

Silence is not the only verdict. A worker can go on heartbeating while it reads nothing, its reader held in a long dispatch such as the load of a fresh plugin, which also pauses the worker's own lease (see Worker control-session recovery). On the same tick the watchdog therefore reads each registered worker's outbox (see Outbound frames to a worker). A worker whose frames have been pending for longer than Config::outbound_stall_timeout (default 10000 ms), with none of them making progress, is declared lost: the stall verdict. Progress is the socket taking a chunk of up to 64 KiB, so a worker that reads a large Deploy slowly but steadily is never stalled, however long the whole transfer takes. Zero or less turns the stall verdict off, and so does a value too large for the clock: the watchdog compares both timeouts in milliseconds, so milliseconds::max() is a timeout that never passes, for heartbeat_timeout as well. A worker whose outbox has stopped by itself because a post overflowed one of its bounds or could not allocate is declared lost at the watchdog's next tick rather than heartbeat_timeout later, whatever the stall timeout: the stop has already shut its connection down, and the worker may still be heartbeating. This is the overflow verdict. A transport that failed under the writer is not one: the connection is gone both ways, so the worker's heartbeats stop with it and heartbeat_timeout declares the loss, as it did before the outbox. A worker that retired the session itself, as it does when its lease runs out across a coordinator pause, re-registers within that time, and the re-registration retires the old session. An outbox the coordinator stopped itself, by an abort or by the flush at stop(), is never a verdict. Neither verdict needs anything from the worker's protocol, so both cover v1 workers, which keep no lease and get no heartbeat answers.

The default stall timeout is five times heartbeat_timeout and above the worker's 3 s lease, which a dispatch held in a fresh plugin's load has been seen to outlast. It is well below restart_drain_timeout (120 s), so a survivor whose frames stall during a restart drain is declared lost, and its subtasks join the restart, long before the drain deadline would fail the job.

A sweep that is itself late judges nothing. When the coordinator was suspended or starved for longer than watchdog_interval plus the shorter of the two timeouts (the stall timeout only while it is on), no frame could be read or written in that gap, so the sweep stamps every worker's last_seen and restarts every outbox's stall clock, and judges them at the next tick. Bounding it by heartbeat_timeout alone took a pause shorter than that but longer than the stall timeout for a stall of every worker with frames pending, a worker part-way through a large Deploy included.

Every loss is logged with its cause, as worker lost: <id> (<cause>) on coordinator.watchdog, and the same cause goes into the coordinator.worker_lost event, the abort of the session's outbox, and the error each pending subtask gets when its job cannot restart. The cause is heartbeat timeout for a silent worker, no outbound progress for <N> ms with <M> frames pending for a stall, and outbound queue stopped (<reason>): <detail> for an outbox that stopped by itself, with the outbox's own reason (overflow or allocation) and detail.

When a worker is marked lost (mark_worker_lost_locked_):

• it is added to lost_worker_ids_, and its session's outbox is aborted: what is queued for it is discarded and counted, its writer ends, and its socket is shut down both ways, so a worker that is in fact alive meets the end of the stream right after the frames already written to it, rather than a connection that stays open and silent until its own lease runs out,
• for each job that had pending tasks on the lost worker, the coordinator either folds those tasks into a restart or, if no restart is possible, synthesises a worker lost (<cause>) error per pending subtask and counts them as completed; a session still retiring its predecessor folds only the subtasks placed on it (see the re-registration paragraph under HA below),
• surviving workers of a touched job are sent CancelJob so their subtasks wind down cooperatively (role handlers poll was_cancelled(); the local executor's cancel token is flipped).

It is a known property of the loss path that a worker declared lost, if it is still running, can act on frames that were written to its socket before the declaration, a CommitCheckpoint among them, until it notices that its session has ended.

Whether a restart is attempted is gated by effective_max_restarts(job.checkpoint) (include/clink/cluster/protocol.hpp): with a checkpoint directory set and no explicit override, the default is kDefaultSelfHealRestarts (10) self-heal attempts; without checkpointing it is 0 (fail-fast, since there is nothing to restore from). An explicit max_restarts_on_worker_loss of 0 forces fail-fast even when checkpointing is on; an explicit N caps the attempts. The watchdog also enforces a restart_drain_timeout (default 120 s): if a job sits in awaiting_restart while a survivor neither drains nor dies, the watchdog fails the job rather than wedge or risk double-running a slow-but-alive subtask. The failure names each subtask that did not drain, with the worker holding it, and none that did (RestartDrainReadiness.ADrainTimeoutNamesOnlyTheSubtasksThatDidNotDrain). A survivor whose frames stall past outbound_stall_timeout before then is declared lost instead, and its subtasks join the restart. The full restart-from-checkpoint mechanism, including the second-worker-loss-during-drain handling, is documented in ./fault-tolerance-and-rescale.md.

Outbound frames to a worker

Every frame the coordinator sends a worker after its RegisterAck is posted to that session's outbox (WorkerConnection::outbox, src/cluster/outbox.hpp). The connection's writer thread writes what is posted, and nothing else writes the socket. A post queues the encoded frame under the outbox's own lock, a leaf that nothing waits under, and returns at once, so it is safe under the coordinator's lock and never waits on the peer. No thread that decides a frame (the watchdog, a dispatch thread, the checkpoint trigger loop, a submitter, a reader answering a heartbeat) can be held by a worker that has stopped reading.

Each frame used to be sent by the thread that decided it. A worker whose socket buffers had filled, behind a large Deploy it was not reading, held that thread in send() and every later sender behind the connection's send lock. When the sender held the coordinator's lock, as the dispatch thread sending the PeerUpdate after a job's last SubtaskListening did, the watchdog stopped too, and no worker could be declared lost until the send ended. Only the replies to a peer not yet admitted, a refusal or the RegisterAck, are still sent directly, each send() bounded as described under Connection routing. The RegisterAck goes out in the hold that publishes the session, before the session's writer starts, so it is always the first frame on the connection.

The writer is the first of the session's three threads to start. It holds the outbox and never the session record, so it can never be the thread that drops the record and joins itself. If it cannot be started, the outbox is given up before the reader starts: the reader meets the end of the stream at once, and the worker is lost by the usual path.

Frames reach a worker in the order they were posted to its connection. Each post is made where the frame used to be sent, so the frames one thread posts arrive in its order, and frames posted in one hold of the coordinator's lock arrive in the order posted. A broadcast is encoded once and the same frame posted to each connection.

A post that returns Queued means only that the frame will be written later or discarded when the outbox stops, and nothing in the coordinator treats a queued frame as received. A post is refused once the outbox has stopped, or while it flushes at shutdown. A refusal is logged and the caller goes on, the submit included. A submit fails only before it publishes its job, in the hold that checks every worker it placed tasks on still has the session they were placed on. Once the job is published, a Deploy refused because the session's outbox has stopped (the session was lost, replaced or overflowed in between) is logged, the remaining Deploys are still posted, and the submit reports the job: whatever stopped the outbox folds the job's subtasks on that session through the loss path, the retirement or the overflow verdict. A submit that threw there told its caller a published job had not been submitted, which a resubmit then ran a second time. A frame for a worker that has just gone away is usually queued and then discarded, and the loss is left to the watchdog, as it was for a frame that reached the socket buffer of a worker that then died.

By default a session holds at most 10000 queued control frames and 64 MiB of them. Deploys are counted apart, as bulk frames, up to 1 GiB (four frames of kMaxFrameBytes), so a burst of them never pushes out the protocol traffic queued beside it. Every Deploy is bulk, not only one that ships plugin bytes: it also carries the job's UDF modules and a chain spec per task, so one that refers to every plugin by hash can still be megabytes. The frame being written never counts. The bounds are set by Config::max_outbound_control_frames, max_outbound_control_bytes and max_outbound_bulk_bytes, whose defaults are the outbox's own (OutboxLimits), and every worker session's outbox is built with them. Zero refuses every frame of its class, so every worker would be lost at its first such frame. A post that finds its class at a bound stops the outbox (reason overflow, with a detail naming the bound) and shuts the socket down both ways, and the watchdog declares the worker lost at its next tick, the overflow its cause (see Heartbeats, the watchdog and lost-worker detection).

An outbox stops when a post overflows it or cannot allocate, when the transport fails under the writer, when the shutdown flush runs out of time, or when the coordinator aborts it. A stop of either of the first two kinds loses the worker by the overflow verdict; a failed transport ends the session, and the heartbeat timeout or a re-registration settles the worker. The coordinator aborts a lost session's outbox (above), and a replaced session's in the hold that sets it aside at a same-id re-registration (see Job persistence and recovery). Stopping discards what is queued and shuts the socket down both ways, which wakes a writer blocked in send() or SSL_write. The frames discarded are counted by class (control or bulk) and reason (overflow, allocation, write_failed, flush_deadline, aborted) in clink_coordinator_outbound_frames_discarded_total, with one coordinator.outbox log line per connection. A discarded frame was given up; if it was being written at the time, some or all of it may still have reached the peer.

A session is reaped once its reader and its writer have both finished. The reaper joins both and drops the transport, the outbox's reference included, so the descriptor closes; the record and its outbox stay. A worker that half-closes its connection ends its session's reader at once, but the session is not reaped while its writer runs, so what is queued for a worker that still reads is written until the worker is declared lost.

Client connections are unchanged: a client session's replies, and a job's JobCompleted, are still sent by the thread that produced them.

Checkpoint coordination

The coordinator owns a checkpoint_trigger_loop_ thread. For each job with a checkpoint directory and a positive interval_ms, it allocates the next checkpoint id, seeds a pending-ack set from the live task set, and broadcasts TriggerCheckpoint to every worker hosting the job. Each worker injects a barrier into its source subtasks, holding it for any source that has not registered its barrier injectors yet so that every source of the job emits the same barriers; as subtasks snapshot they reply SubtaskCheckpointed. When the pending set for an id empties, the coordinator writes a COMPLETED-<id> marker under the checkpoint dir and advances latest_completed_checkpoint_id. Sinks implementing two-phase commit then receive CommitCheckpoint (or AbortCheckpoint if any subtask failed its snapshot). Bounded sources at clean end-of-stream use RequestFinalCheckpoint/FinalCheckpointAssigned to obtain one job-wide final checkpoint id so the post-last-checkpoint tail is durably committed before completion. The barrier mechanics and 2PC sink protocol are covered in ./checkpointing.md.

HA leader election

For a single active coordinator, clink supports standby coordinators that hold the control port closed until they win an election. The abstraction is HaCoordinator (include/clink/cluster/ha_coordinator.hpp): start() spawns a poll thread that tries to acquire leadership, is_leader() and an on_become_leader callback report transitions, and current_leader_endpoint() lets a worker discover where the active coordinator is.

Two implementations exist:

• File coordinator (default, src/cluster/ha_coordinator.cpp). make_file_ha_coordinator(ha_dir) takes an exclusive fcntl write lock on <ha_dir>/leader.lock (non-blocking F_SETLK) and, on success, bumps a monotonic epoch and atomically writes <ha_dir>/active-leader.json (write to .tmp, then rename) with the leader's advertised host, port, epoch and timestamp. The OS releases the lock when the holding process exits or crashes, which lets a standby acquire it. The poll interval defaults to 200 ms. This is intended for a shared filesystem on a single machine or a cluster with shared storage.

The HA directory must be on a filesystem that honours POSIX write locks. The whole fencing scheme reduces to that one primitive: if a second process is granted the same lock, every coordinator sharing the directory believes it leads, each announces an epoch above the last, and nothing reports an error. Bind mounts (including Docker Desktop's), 9p and virtiofs shares, and NFS exports mounted nolock are the usual offenders, and the symptom is two live leaders rather than a failure.

On start() the coordinator therefore proves the primitive instead of assuming it: it holds the lock and forks a child that asks for the same one. The fork is necessary because POSIX record locks are per-process, so a second descriptor in the same process reports success on any filesystem. A directory that grants both makes the coordinator refuse to stand for leadership, with an error naming the likely cause; --ha-allow-unsafe-locks overrides that and warns instead. A probe that cannot run (another coordinator holds the probe file, which is normal when two start together) is inconclusive and does not condemn the directory.

The check covers exclusion between processes ON ONE HOST at startup. It cannot detect a mount that honours locks locally but not between hosts - the NFS behaviour that matters most for a multi-node deployment - so a coordinator on such a mount passes the probe and can still split brain against one on another host. Use the etcd coordinator where leadership must hold across hosts. - etcd coordinator (optional, impls/etcd/include/clink/etcd/etcd_ha_coordinator.hpp). make_etcd_ha_coordinator(EtcdHaConfig) performs etcd v3 leader election: grant a lease (default TTL 10 s), keep it alive on a background thread, atomic-put the leader key, and have standbys watch the key. Lease loss is the failure-detection primitive, so a frozen leader loses leadership after at most the lease TTL. The leader key is namespaced by cluster_name. The factory is only compiled when the cluster is built with the etcd impl (CLINK_WITH_ETCD / linked as clink_etcd); tools/clink_node.cpp guards the call with #ifdef CLINK_LINKED_ETCD and prints a loud error if --etcd-endpoints is given without it.

In clink_node, passing --ha-dir (or --etcd-endpoints) puts the coordinator in HA mode: coordinator.start() is deferred until on_become_leader fires, at which point the coordinator binds the control port and calls recover_persisted_jobs(). A standby coordinator just sits on the coordinator poll thread.

The takeover path is integration-tested end to end in tests/integration/test_coordinator_ha_failover.cpp: a coordinator-only loss (standby recovers the persisted job and drives new checkpoints) and the compound loss where the leader and a worker are SIGKILLed together with no window between them. The compound test asserts the property a consumer actually depends on - the standby restores from the last COMPLETED-N marker (a non-zero restore_from_ckpt, checked against the recovery log line, since a from-scratch re-run regenerates the same checkpoint ids and atomically overwrites the same committed filenames) and the committed output stays each-record-exactly-once.

Worker control-session recovery

clink_node --role=worker is a long-lived process containing a WorkerSupervisor. A Worker object is deliberately a single-use control and task session: it owns task threads, per-job registries, checkpoint waits and network endpoints that have a clean reset boundary only at destruction. When its coordinator connection ends, the supervisor removes the session from the HTTP surface, immediately cancels and fences all of its subtasks, joins every task and control thread, destroys the session, rediscovers the coordinator, and registers a fresh Worker under the same stable worker id. The next session is never published before the old one has fully drained, so restored subtasks cannot overlap work from the superseded session.

Discovery and transport failures retry indefinitely with bounded exponential backoff and full jitter (--reconnect-initial-backoff-ms, default 100; --reconnect-max-backoff-ms, default 5000). --reconnect-discovery-timeout-ms bounds each discovery pass. A temporary coordinator connection-cap refusal is retryable; an incompatible protocol, malformed handshake or non-retryable registration refusal stops the process with code 2 because another retry cannot repair it. The supervisor remembers the highest coordinator epoch it has admitted and refuses a discovered or handshaking endpoint below it, including an unfenced epoch 0 downgrade.

With TLS on (--tls-ca), each connect to the coordinator is bounded as a whole. connect_tls_connection (TlsSocket::connect) runs the TCP connect non-blocking (NetworkSocket::connect_within: wait for writability, then read SO_ERROR) and the client handshake as a non-blocking SSL_connect loop, both waiting in NetworkSocket::wait_ready against one deadline, TlsConnectOptions::connect_timeout (default 5000 ms, kDefaultTlsConnectTimeout; the worker's --handshake-timeout-ms, 1 to 86400000). The socket is made blocking again before it is handed over, since every reader expects one. A connect that misses the deadline throws, naming the stage (the TCP connect, or the handshake, noting when the server sent nothing in reply to the ClientHello), and the supervisor retries it with backoff like any other transport failure. The deadline starts once the host has resolved: getaddrinfo has resolver timeouts of its own (five seconds a nameserver under glibc's defaults), and counting them would let a dead first nameserver spend the whole bound before a packet was sent. Before this, the connect was a blocking TCP connect and a blocking SSL_connect with no bound, and a coordinator whose TCP connection completed and that never answered the ClientHello held the attempt indefinitely: the worker neither retried nor honoured shutdown. The kernel completes a connection into the listen backlog whatever the process is doing, so a frozen coordinator, an older one whose accept thread a stalled peer was holding, or a stalled middlebox all did that. (The current plaintext coordinator is not one of them: it reads the ClientHello as a first frame far over its size cap and closes the connection, which fails the handshake at once.) A caller that passes no options still gets the default deadline, so an unbounded client handshake is no longer reachable by accident; zero or less in TlsConnectOptions means none, for a caller that owns the server too. The Kafka transactional-resume dialer bounds its TLS connects the same way, at 5000 ms, the bound on each of its reads. A plain-TCP connect (connect_plain) is unchanged: it has no handshake to stall, and its TCP connect is bounded by the kernel's own connect timeout.

The handshake also checks that the coordinator's certificate names the host the worker connected by, not only that the --tls-ca CA signed it; without that check any certificate the CA signed would be accepted for any host. The host is the one the connect was given: --coordinator-host, or under HA the leader's advertised host (--advertise-host, which defaults to its bind host). TlsSocket::connect sets the check on each session (SSL_get0_param), not on the shared context. A DNS name is matched against the certificate's subjectAltName DNS entries, or its subject CN when it has none, with partial wildcards off (X509_CHECK_FLAG_NO_PARTIAL_WILDCARDS: * only as a whole left-most label). An IPv4 or IPv6 literal (recognised by inet_pton, an IPv6 zone stripped) is matched against the certificate's IP entries (X509_VERIFY_PARAM_set1_ip_asc), never its DNS names or CN, set explicitly because SSL_set1_host does not match an address on every OpenSSL release the engine builds against (1.1.1 takes it as a name). A certificate that does not name the host fails the handshake with the server's certificate does not match the host <host> (it names DNS:..., IP:...), which says how to fix it and names the opt-out, and the supervisor retries it with backoff like any other transport failure, so a misissued certificate shows as a worker that never registers, with the reason in its log and in GET /api/v1/worker's last_error. --tls-no-verify-hostname (TlsConnectOptions::verify_hostname = false) turns the name check off and leaves the chain check on; it is for a deployment whose certificates cannot name the address workers connect by, and is refused without --tls-ca. Every other TlsSocket::connect caller gets the check by default too; one whose opt-out is not this flag names its own in TlsConnectOptions::verify_hostname_opt_out, as the Kafka resume dialer does (see the Kafka connector).

EOF is not sufficient failure detection: a one-way partition can leave a TCP socket apparently writable. Protocol v2 therefore sequences worker heartbeats and makes a v2 coordinator return a fenced HeartbeatAck. A v2 worker ends its session after --coordinator-heartbeat-timeout-ms without any coordinator frame (default 3000 ms). The acknowledgement is capability-gated for rolling upgrades: a v2 coordinator never sends it to a v1 worker, and a v2 worker connected to a v1 coordinator retains EOF-based detection.

The lease's health depends on the reader thread never blocking, so the worker executes sink commit callbacks on a dedicated FIFO dispatch thread rather than on the reader. A 2PC commit blocks on the external system - a Kafka commit_transaction against an unreachable broker holds for the full transaction timeout - and when that ran on the reader, no frame was read and the coordinator-contact clock froze for the duration: a broker outage then read as coordinator silence, and perfectly healthy workers ended their own sessions mid-outage, converting a data-plane fault into control-plane worker loss and restart storms. The single dispatch consumer preserves per-worker commit order across checkpoints, which the transactional sinks' open-transaction sequencing relies on; retirement gates refuse dispatches that outlive their runner exactly as before.

The worker HTTP listener belongs to the supervisor rather than a session and keeps the same port across recovery. /api/v1/health and session-backed routes return 503 while discovery, backoff or draining is active; /api/v1/config remains available. Recovery is observable through clink_worker_control_connection_attempts_total, clink_worker_control_disconnects_total, clink_worker_control_reconnects_total and clink_worker_control_connected. An external process supervisor remains the final boundary for a worker binary crash, host loss, fatal handshake, or an operator that cannot honour cancellation. It is no longer the normal response to coordinator failover.

Shutdown

clink_node installs a SIGTERM/SIGINT handler and calls Worker::stop() or Coordinator::stop() on the way out.

Worker::stop() flips every registered per-subtask cancel token before joining the task threads - the same flip CancelJob performs. That matters because a running subtask's LocalExecutor watches JobConfig::external_cancel_token and nothing else; it has no view of the Worker. Setting the worker's own stop_ flag reaches the pending-task waiters and the EOS waits, but not a runner that is mid-stream. Until 2026-08-03 stop() did only the latter, so SIGTERM exited an idle worker promptly and hung a busy one indefinitely - the reverse of what a container grace period needs.

Cancelling rather than draining is deliberate. The subtasks are going away with the process either way, and a cancelled subtask runs its normal teardown, so a 2PC sink aborts its pending transaction instead of leaving it for recovery. What clink does NOT do on SIGTERM is take a final checkpoint or drain the sources: records since the last checkpoint are replayed on restart. There is no "stop with savepoint" path.

The joins are unbounded on purpose. Once cancellation is signalled, a subtask that will not exit is a bug in that operator, and detaching a thread still touching the Worker's members during destruction trades a diagnosable hang for a use-after-free.

Coordinator::stop() first joins every thread of its own that posts to a worker: the readers, and with them the dispatch threads, the watchdog, the checkpoint trigger loop, the recovery and in-doubt threads, the autoscalers, the admission threads and the client sessions. None of them can be held by a peer, since a post never waits on one. It then flushes every worker connection against one deadline, Config::outbound_flush_timeout (default 1000 ms) away: each writer writes what is still queued for its worker, and whatever a worker has not taken by the deadline is discarded and counted (reason flush_deadline), the connection closed, and the writer, woken by the close if it was blocked in send(), joined. A worker that has stopped reading therefore holds stop() that long at most, and the frames queued at shutdown are written or counted, never dropped unseen. The readers are woken by shutting the read side of their sockets down, which leaves what is queued alone; on Darwin, though, a socket that receives data once its read side is shut down resets the connection, so a worker that sends something, a heartbeat say, during the flush loses what was still queued for it there, counted as write_failed.

Limits on what a peer can make a node do

Frame IO is one shared implementation (include/clink/cluster/frame_io.hpp), and it does not trust the peer. A frame longer than kMaxFrameBytes (256 MiB) is refused on the header alone, and the body is read in 64 KiB chunks so memory tracks the bytes that have actually arrived rather than the number the header claimed. Both matter: a cap alone still lets four bytes cause a 256 MiB allocation.

MessageReader::read_count bounds every container length by the bytes remaining in the frame. An element cannot cost less than a byte on the wire, so a larger count is malformed by construction - no arbitrary limit is involved.

Each of the frame-handling loops - the coordinator's accept loop, its per-client readers, its per-worker readers and dispatch threads, and Worker::reader_loop_ - wraps a single frame in a try. A worker frame that does not decode is caught on the dispatch thread, which gives the connection up; the reader then stops reading, so last_seen stops moving and the watchdog declares the worker lost. A frame that does not decode costs the peer its connection, is logged, and increments clink_malformed_frames_total. This is load-bearing rather than defensive: the decoders throw by design on a malformed payload, and an exception leaving a thread function is std::terminate, so before this a single malformed frame killed the process.

A worker connection's dispatch backlog is bounded as well. The reader keeps reading while a handler waits, so frames that used to stay in the socket buffer, pushing back on the peer, now queue in memory. Config::max_worker_dispatch_backlog_frames (default 10000) and Config::max_worker_dispatch_backlog_bytes (default 64 MiB) cap the queue; a single frame is always admitted to an empty queue, whatever its size. Exceeding either is handled like a malformed frame: it is logged, counts clink_malformed_frames_total, the queued frames are discarded and the worker loses its connection. A well-behaved worker stays far below both, because it acks only barriers the coordinator triggered and no barrier is triggered before its id is on record in the checkpoint store, so a stalled store stops new acks at the source.

In the other direction, what the coordinator has queued for a worker is bounded per connection too (see Outbound frames to a worker), so a worker that stops reading costs at most those bounds in memory before its connection is given up, and holds none of the coordinator's threads meanwhile.

None of that bounds the number of connections, the rate of frames, or aggregate memory across peers. See docs/history/production-hardening-2026-08.md, W14.

Protocol version negotiation

Every node declares two numbers at the handshake: protocol_version, the wire protocol it speaks, and min_compatible_protocol_version, the oldest peer it will talk to. check_protocol_compatibility (include/clink/cluster/protocol.hpp) accepts only when each side falls inside the other's range, and the coordinator applies it at both Register and HelloClient while the worker applies it to the RegisterAck. The check is symmetric on purpose: "the coordinator can read the worker" does not imply the reverse, and a one-sided handshake admits a pairing that half-works.

This is deliberately NOT the mechanism that handles a message merely gaining a field. That is the additive-tail idiom - encoders append at the end, decoders read r.eof() ? default : read() - and it must keep working without negotiation, because it is what lets a cluster be rolled one node at a time. kClusterProtocolVersion is bumped for a change additive tails cannot express: a field changing meaning or width, a MessageKind repurposed, a semantic contract changed under an unchanged encoding, or a capability whose new frame kind must be gated to peers that understand it. Bumping it for an additive tail alone would refuse the very upgrade the idiom exists to allow.

A peer built before versioning sends no fields, which decode as 0 and are read as version 1. Refusals name both versions and say which end to upgrade, and increment clink_protocol_mismatches_total. A refused client receives a SubmitJobAck nack; CLI tools other than clink submit surface its message through protocol_rejection_message rather than reporting an unexpected frame kind.

Raising kMinCompatibleClusterProtocolVersion is an end-of-support decision, not a routine bump: it strands every peer below it, loudly and at the handshake.

The version a worker registered at is also what the coordinator reads to admit the Row sidecar's second layout (typed timestamp columns between workers, version 3). Each whole-job deploy (submit, restart, the replan rescale, HA recovery, and the embedded engine's deploy, which goes through submit) computes admit_row_layout over the workers its placement uses and the layout-bearing operators its tasks name (count_layout_bearing_ops), and stores the answer on the job: layout 2 only when there is at least one such operator and every one of those workers registered at v3 or later. The coordinator then stamps row_layout=2 onto the layout-bearing operators of each Deploy frame (stamp_row_layout, include/clink/cluster/row_layout_admission.hpp); the job's task records and its persisted manifest keep the unstamped spec, so the next whole-job deploy decides again. A partial deploy (a hot cutover's new subtasks) carries the job's current layout unchanged. Each decision is logged as job_id=<id> row_layout=<n>, with the reason when the answer is 1 (no layout-bearing operator, or the worker that held it back), and the number of operators stamped. CLINK_TEST_REGISTER_PROTOCOL_VERSION, read by the worker at registration, makes a worker declare an older version so a test can build a mixed-version cluster from one build; it is a test hook and nothing else reads it.

Fencing: a coordinator that has lost leadership cannot act

Losing leadership is not something a leader can detect on its own. A coordinator partitioned from the coordination store, or paused past its lease, keeps every worker connection open, keeps its in-memory job state, and keeps its checkpoint timer running. Fencing is what stops it acting on that stale belief.

Every coordinator-to-worker control frame carries the sending coordinator's epoch: RegisterAck, Deploy, PeerUpdate, CancelJob, TriggerCheckpoint, CommitCheckpoint, AbortCheckpoint, FinalCheckpointAssigned, BeginRescale, CutoverPeerUpdate and CutoverRebind. clink_node calls Coordinator::set_epoch from the become-leader callback before the listener opens, and Coordinator::fenced_frame_ stamps the epoch as it encodes, so no send site can forget it.

A worker binds the epoch carried by the RegisterAck that admitted it. Worker::accept_epoch_ then drops any later frame carrying a lower epoch, incrementing clink_worker_fenced_frames_total and logging at error - a non-zero count is a split-brain signal. A higher epoch re-binds the worker rather than being refused: a failover that keeps the connection up presents that way, and refusing it would fence the worker off from the legitimate new leader.

Epoch 0 means "unfenced". A non-HA coordinator never sets one, so a single-coordinator cluster behaves exactly as it did before fencing existed. The field is appended at the tail of each message body and read with the additive idiom the rest of the protocol uses (r.eof() ? 0 : r.read_u64_be()), so a node running a pre-fencing build decodes as 0 and a cluster keeps working while it is rolled.

Control-plane metadata is fenced too: the job manifest and the history record each carry "coordinator_epoch", the active-leader file carries "epoch", and a write from a lower epoch than the one already stored is refused. The check is a genuine compare-and-set (fenced_metadata_cas_write): the epoch read, the metadata_write_allowed rule and the fsync-plus-rename all run inside one per-target critical section, held as flock(LOCK_EX) on a <path>.wlock beside the record. Two racing writers can no longer both pass the check - the second one's read happens only after the first one's rename is complete, so the interleave where a superseded coordinator read a stale epoch and then renamed its record over the real leader's cannot occur. The lock is the same primitive family as the leader lock itself, released by the kernel when its holder dies, so the fenced path adds no dependency and no stale-lock recovery problem; the lock file is created once and never unlinked, because unlink-and-recreate lets a later writer lock a fresh inode while an earlier one still holds the old. The race is pinned by CoordinatorFencing.TwoWritersCannotStraddleTheCasWindow, which stretches the historic window with an in-section fault delay and proves a concurrent writer waits instead of slipping inside.

Job persistence and recovery

When HA is active the coordinator persists each submitted job. set_ha_dir creates <ha_dir>/jobs/ and <ha_dir>/history/. On submit, the coordinator writes <ha_dir>/jobs/<job_id>/manifest.json plus the plugin .so bytes (persist_job_manifest_). On takeover, recover_persisted_jobs() first lets worker registrations settle - it waits until the registered set has been stable for Config::recovery_worker_settle (default 1 s, 0 disables), bounded by recovery_worker_settle_deadline (default 5 s) - because the supervisors of a running cluster reconnect within moments of a new leader binding but not simultaneously, and a redeploy fired at the first registration schedules every task onto that one worker and never rebalances back (a QUAL-01 recovery ran its remaining minutes with all sixteen tasks on one worker for exactly this reason). It then scans <ha_dir>/jobs/, re-reads each manifest, reloads its plugins into a fresh JobBundle, and re-submits the job with restore_from set to the latest COMPLETED-N marker found on disk for that job (a whole marker name only: the temp file of a publish the dead leader did not finish is not a marker, since that checkpoint's completion was never durable), named by the job's own marker directory (<checkpoint_dir>/_jobs/<job id>) so that other jobs sharing the checkpoint directory never make the restore ambiguous; the marker says where that checkpoint's state is. A job that restores from its newest completed checkpoint passes over a marker its newest deploy restored below. Every deploy that continues a run (each restart, a restore, a takeover's own deploy) puts <checkpoint_dir>/_jobs/<job id>/RESTORED before any of its frames leave, naming the checkpoint it restored from and the first id its run numbers, and a COMPLETED marker strictly between the two is a put that was still out when the job restored: a restart redeployed while it was in flight, or a superseded coordinator's put landed after the takeover read the disk. The run that restored below such a checkpoint has rolled its transactions back at open, as the PostgreSQL two-phase sink does with every prepared transaction its restore does not hold, so restoring from it would resume the source past an interval no store holds; the exactly-once specification found the case. The takeover reads the record before the markers, and an embedded run resuming its own checkpoints applies it the same way. The record is fenced by the coordinator's epoch, so a superseded coordinator cannot put an older range over the leader's; a restart that cannot put it waits like a restart without capacity, retried every watchdog tick and failed only at the capacity deadline, and one the fence refuses deploys nothing. A record that does not parse fails the job's recovery rather than restore past it. A tracked job restores from its CONFIRMED markers and its walk reads every COMPLETED one, as before. A job that has not completed a checkpoint of its own yet keeps the restore point it was submitted with (a savepoint, or a predecessor's checkpoint), or starts fresh when it was submitted without one, which is the rule a restart applies. It used to take its own id 0 as the restore point; the deploy lint refuses a restore directory with id 0, so a takeover before a job's first checkpoint dropped the job. The re-submitted job numbers its new checkpoints above every id already recorded on disk, including every id the dead leader claimed on the job's checkpoint-id record, and never merely above the restore point; a job with no completed checkpoint of its own gets the same floor (see the checkpointing page). It is idempotent (already-running ids are skipped) and pins each recovered job's state-backend URI to the one it originally ran with (pin_recovered_state_backend) so a cluster default configured after the job was submitted cannot silently rebind it. A recovery that fails ONLY for capacity - the takeover raced the workers' new control sessions, so no slots had registered within submit_wait_for_slots - is parked rather than dropped: the id joins pending_recovery_ids_ and a retry thread re-runs recover_one_persisted_job_ from the manifest whenever a worker registers (the same already-running guard makes the retry idempotent). The retry is the same takeover continuing: it records nothing again and does not walk again, since what the takeover's walk refused it would only probe a second time, and it deploys the restore point the takeover decided (kept per job beside the takeover record) rather than reading the markers again, which a superseded coordinator's put landing during the park would have moved past what the takeover reported. The takeover reads the job's markers once, before it loads the plugins, and its restore point, its reported completed point and its walk all follow that read: the walk runs to the completed point the takeover read and reported, not to a fresh read, which a stale COMPLETED put landing in between would put above the reported range. The recovered job also keeps that completed point (CheckpointConfig::resumed_completed, set as well by a tracked run's resume), so a restart before its first new checkpoint completes still holds for in-doubt resolution over a gap the takeover's walk left open (a refusal, a broker outage) and walks it again, as a restart of the run that completed it would. Every other recovery failure drops the job, since re-running it would fail the same way: it is logged at error level, says the job is not running and needs a resubmit, and counts clink_ha_recovery_failed_total. Terminal-state records are also persisted to <ha_dir>/history/ and reloaded into a bounded in-memory ring (reload_history_from_disk_).

A --ha-dir worker rediscovers the leader from active-leader.json (or etcd) before every session; a non-HA worker reconnects to its configured address. On the Coordinator side, a worker re-registering under an existing stable id atomically replaces its old session. The hold that sets the old session aside also aborts its outbox, once the new session's RegisterAck has gone (a registration that fails there restores the old session untouched): nothing more is sent on the old connection, what was still queued for it is discarded and counted, and its socket is shut down both ways, which wakes its reader and its writer. The new session's reader starts at once, so its heartbeats are answered from the first. Its dispatch thread first retires the old session (retire_replaced_session_): it joins the old reader, which joins the old dispatch thread once that has handled every frame the old session read, since those frames were sent before the re-registration, and the old writer, which the abort has ended even if a worker that had stopped reading held it in send(); it drops the old session's transport, as the reaper does for a lost one, so a submit, a held final-checkpoint request or a staged Deploy that still holds the old session or its outbox does not keep the descriptor open; it then folds the old session's still-in-flight subtasks into checkpoint recovery and moves the capacity gauges, and only then handles a frame of the new session. A subtask whose clean SubtaskFinished was still queued is therefore finished, not folded as lost, and neither the accept thread nor the new session waits on a store write the old session's frames make. The new session is schedulable from its RegisterAck, before the old one is retired, so a worker id alone does not say which session holds a subtask: every placement stamps its subtasks with the session its Deploy went to (JobState::pending_session, WorkerConnection::session), and the retirement folds only subtasks stamped with an earlier session. A job placed on the new session while the old session's frames drain keeps running, and if a job has subtasks on both, the new session's are drained as survivors of the restart. A slot is released against the session it was charged to, so nothing the old session does frees slots the new session's placements hold: a subtask's finish frees the reporting session's slot, and a cutover's teardown or abort frees the slot of the session stamped on the subtask, which frees nothing once that session is superseded, because its retirement takes its slots off the books. A per-subtask retry decided on a superseded session's error moves the subtask, its stamp and its slot, to the session the retry is sent to. A submit places its tasks in one hold of the coordinator's lock and publishes the job in a later one, after the store reads and writes between them; the publishing hold checks that each worker still has the session the tasks were placed on. A worker that re-registered in between, whose old session may already be retired without the job to fold, has the tasks deployed on its new session, which takes over their slots if it still has room for them; a new session that another submit has filled meanwhile, or a worker with no live session left, fails the submit retryably, and the slots go back. Every free-slot count treats a session charged past its capacity (a retry moved onto a full successor) as having none, never as an unsigned wrap-around. A new session lost while it is still retiring its predecessor folds only the subtasks stamped with its own session: the predecessor's are left to the predecessor's queued frames and then its retirement, so a SubtaskFinished the old session sent is never counted against a run the loss redeployed. They count as draining in whatever restart the loss starts or joins, including the replan a hot cutover aborted by the loss stages, and in any drain staged while the retirement is still out. A worker that registers again while an earlier session of it is still being retired is refused with a retryable RegisterAck until the retirement ends; the worker supervisor backs off and registers again. At most one superseded session is therefore draining per worker id, and its threads, which are not counted against max_worker_connections, are bounded by the dispatch backlog caps. Workers embedded outside clink_node can also discover the coordinator endpoint through the simpler ServiceDiscovery abstraction (include/clink/cluster/service_discovery.hpp): static config, environment variables, or a file containing host:port.

The client submission path

A client does not run clink_node. It links clink::application::JobSubmitter (include/clink/application/job_submitter.hpp) and calls submit(graph_json, plugin_paths, opts). The submitter (src/application/job_submitter.cpp):

1. reads each plugin file into memory and content-hashes it,
2. opens a TCP connection to the coordinator,
3. sends HelloClient then SubmitJob (graph JSON + plugin binaries + CheckpointConfig),
4. waits for SubmitJobAck (bounded by ack_timeout, default 10 s),
5. if wait_for_completion, waits for JobCompleted (bounded by wait_timeout, default 60 s),
6. closes the connection and returns a SubmitResult.

If the client connection drops while the job is still running, the coordinator clears the job's notify_client_conn pointer and the job continues; only the ability to push JobCompleted back is lost. list_jobs() uses the same HelloClient-then-request pattern and returns both running and recently-completed jobs (completion_signalled distinguishes them).

On the coordinator side, handle_submit_ decodes the message, allocates a per-job JobBundle, writes each plugin to the coordinator's local cache and dlopens it into the bundle (so the planner can see plugin-defined op types), optionally rejects on a state schema-evolution incompatibility, then calls submit_job(...) which plans the graph and dispatches Deploy messages. Job planning and slot assignment are covered in ./jobs-and-scheduling.md.

submit_job gates every submission before planning or allocation. After configuration coherence comes the connector-availability gate (check_connector_availability in src/cluster/connector_availability.cpp): an op type no registry knows, whose name matches clink's compiled connector vocabulary, is refused with the connector named, the connectors this binary does have (read from the capability registry, never a hardcoded list) and the CLINK_WITH_* flag that adds the missing one. Availability is decided by the registries the job would deploy against - the bundle's when the job carries one, so plugin registrations count - and the gate runs in whichever process owns the submission: a local clink run submits through an in-process coordinator and checks the local binary, while a distributed submission is gated on the target cluster's coordinator, never the submitting CLI's binary. Op types matching no connector vocabulary pass through for deploy to resolve, so plugin and inline operators are unaffected. Then the delivery-guarantee gate (check_delivery_guarantee, src/cluster/guarantee_gate.cpp) rejects a job that ASKED for a guarantee the pipeline cannot provide, and the graph-level lint rejects structural errors. A job is asked for the strongest guarantee any of its ops declares in delivery_guarantee. When its sinks declare different ones, as a statement set's tables can, each sink is checked instead for the one it is held to, through the sources that feed it: its own declaration, or the job's strongest when it declares none. The job is still reported as what its weakest sink gives, and guarantee_reasons carries a line per sink. One gate runs after planning, because it needs the plan: a submission that restores (restore_from_dir with restore_from_checkpoint_id, including an HA recovery) is refused before any slot is waited for when the checkpoint's COMPLETED marker records a different subtask index set from the one the plan deploys (check_restore_layout, src/cluster/restore_compat_gate.cpp), since every task restores from its own index and the operators would otherwise receive each other's state. The rule and its limits are in ./fault-tolerance-and-rescale.md.

The job-as-plugin model

A clink job is a single shared library. The contract (include/clink/job/register_job.hpp) is declared with one macro at file scope:

void define_job(clink::api::Pipeline& env) {
    env.from_elements<int64_t>({1, 2, 3, 4, 5})
       .map<int64_t>([](int64_t v) { return v * 2; })
       .sink(/* ... */);
}
CLINK_REGISTER_JOB("my-job", "1.0", "demo", define_job);

CLINK_REGISTER_JOB expands CLINK_DECLARE_PLUGIN (the ABI handshake getters) and emits the job exports: clink_plugin_register, clink_job_build (returns the captured JobGraphSpec JSON), and clink_job_check_restore_compatibility (the state schema-evolution pre-deploy check). Graph capture runs once per mapped module, while registration is deliberately repeatable:

• in the submitter process, clink_job_build returns the once-captured JSON the submitter uploads alongside the .so;
• in the coordinator and each worker process, every clink_plugin_register call runs build_fn against that job's fresh JobBundle registry, so inline operator types such as _inline_map_<n> resolve identically on every side even when the process reuses an existing module mapping.

The same .so is therefore dlopen'd on every process that touches the job: the submitter, the coordinator (for planning and validation), and each worker (to run it). Cross-process matching of inline operator types relies on build_fn registering operators in a deterministic order. Every registration gets a fresh Pipeline environment, whose inline-name counter starts at zero and re-mints the same names the submitter's graph JSON references.

PluginLoader::load_into maps a source path once and invokes its registration hook for every target bundle. Successful C++ modules remain mapped for the process lifetime. This is intentional: type-erased factories, module statics, TLS and failure paths that partially populate a registry make dlclose unsafe to prove in the general case, and Linux can defer some C++ module teardown until process exit even after an apparent close. The loader's path/content-hash cache bounds this to one mapping per admitted plugin rather than one per job or worker control session; the operating system reclaims the mappings when the process exits. A failed ABI or target gate is still closed because registration has not run and no module-owned closure can have escaped.

Plugin loading and the ABI gate

PluginLoader (src/cluster/plugin_loader.cpp) dlopens a .so with RTLD_NOW | RTLD_LOCAL, resolves the extern "C" handshake symbols (clink_plugin_abi_fingerprint, clink_plugin_abi_version, clink_plugin_abi_hash, clink_plugin_target_triple, clink_plugin_toolchain, clink_plugin_abi_manifest, clink_plugin_metadata, clink_plugin_register), and gates the load on three checks, all folded into the pure check_plugin_abi() for testing (design record 010):

• ABI compatibility. The default gate compares clink_plugin_abi_fingerprint() to cluster_abi_fingerprint() - both return kAbiFingerprint, a structural fingerprint computed at clink configure time (CMakeLists.txt) as SHA256 over the content of the declared extension surface: the headers named by the tracked manifest scripts/plugin-abi-surface.txt, generated by scripts/gen-plugin-abi-surface.py as the transitive include closure of the plugin authoring entry points (include/clink/{api,job,plugin,operators,connectors}/), plus the conditional-compilation options that surface uses (the manifest's [options] section), the pinned Arrow version and the manual CLINK_ABI_VERSION - baked into the generated include/clink/plugin/abi_version.hpp. Equal fingerprints load; a difference is a hard refusal, and because every plugin also carries its per-header manifest (clink_plugin_abi_manifest), the refusal names the headers that differ (or blames the options material when the manifests are identical). It rotates when the surface a plugin can reach across the dlopen boundary actually changes - a boundary-type layout, a virtual on Operator/Source/Sink, an inline/template body, a toggled ABI option - but not on host-side header edits (sql/, embed/, queryable_state/, ...) or .cpp / test / doc / build-only commits. A header entering or leaving the closure is a reviewed manifest diff, gated by --check in CI and the pre-commit hook, and the generator fails if any first-party plugin source includes a header outside the declared surface. Two fallbacks preserve safety: a plugin built before the fingerprint symbol existed and strict mode (CLINK_STRICT_PLUGIN_ABI=1) both revert to the historic exact commit-hash comparison (kAbiHash, from git rev-parse HEAD). The commit hash is otherwise informational (reported in logs, LoadedPlugin, and clink_node --version).
• Target triple. The plugin's target triple (darwin-arm64 / linux-x86_64 / linux-arm64) must match the cluster's, or the load is refused. This gate is unconditional - it is a genuine binary-compat axis independent of the ABI fingerprint.
• Toolchain identity. kToolchain (in the generated header) is composed by the preprocessor per translation unit - which standard library, the libstdc++ dual-ABI choice, _GLIBCXX_DEBUG, and ASan/TSan/MSan instrumentation - so a module baked under one toolchain cannot load into a cluster built under another (a TSan cluster refuses an uninstrumented plugin, a libc++ host refuses a libstdc++ module). Standard-library versions are deliberately not part of the string: both mainstream libraries hold their ABI stable across releases. A plugin without the symbol predates the fingerprint that introduced it and is already refused by the ABI gate, so the absent-symbol case waves nothing through.

Deploy-time refusals by any of the three gates are fatal (SubtaskFinishedMsg.fatal): they are deterministic for the binaries involved, so the coordinator fails the job with the cause instead of consuming restart budget on redeploy-and-refuse loops. They are also reachable only on a mixed-version cluster - submit-time checks make same-version refusals happen earlier:

• Submit preflight. The submitter reads each plugin's identity via its own dlopen and advertises it inside SubmitJob (PluginAbiAdvert, an eof-guarded tail). The coordinator checks adverts against its own identity before resolving references or requesting bytes, so an incompatible plugin is refused on the references-only first exchange with zero plugin bytes shipped; the ack carries the cluster's identity and (on refusal) its manifest, and the client names the differing headers against its local copy. Adverts are best-effort claims - the load-time gate remains the authority.

RTLD_LOCAL keeps the plugin's symbols out of the global namespace. Because clink_core is statically linked into both the host and the .so, each side has its own copy of any process-wide singleton, so plugin registrations must be routed through the PluginRegistry/JobBundle view passed into clink_plugin_register rather than a Registry::default_instance() resolved inside the .so.

Security: the "safe to expose" baseline

The default posture is loopback + plain TCP + no auth - correct for a trusted single host, unsafe on a shared or public network. The baseline that makes a cluster safe to expose:

• Frame caps. Every wire frame is length-prefixed by an attacker-controllable u32. The readers cap it at kMaxFrameBytes (256 MiB, network/wire.hpp) and drop the connection on anything larger, closing the memory-amplification DoS where a peer that claims 4 GiB makes the reader allocate 4 GiB and OOM the process. A connection's first frame, read before the peer has identified itself, is capped at kMaxFirstFrameBytes (64 KiB), so the connections being admitted at once cannot be made to buffer 256 MiB each, and the strings a Register carries at kMaxRegisterStringBytes (255 bytes), so a reply that names the worker stays small. Always on.
• Token auth on the HTTP control plane. Set CLINK_AUTH_TOKEN on the node (clink_node reads it for both the coordinator and worker HTTP servers) and every request must carry Authorization: Bearer <token> or gets 401 before its handler runs - the console, the /api/v1 routes, and SQL submission over HTTP (POST /api/v1/jobs/spec). Clients present it automatically from the same env var (clink run / the SQL submitter and the queryable-state reader call HttpClient::set_bearer_token). Unset leaves auth off (backward compatible). The token rides an env var, not a flag, so it does not leak in ps; a CORS preflight (OPTIONS) is allowed through so a browser can present credentials on the real request.
• Control-plane TLS/mTLS. The coordinator/worker control connections run through injectable accept/connect factories; a TLS factory (build-gated, clink::tls) encrypts and can mutually authenticate the control plane. Pair it with bind_host=0.0.0.0 for multi-host. A worker verifies the coordinator's certificate against --tls-ca and checks that it names the host the worker connected by (a DNS name against its DNS entries, an IP address against its IP entries), so issue each coordinator a certificate whose subjectAltName carries every name and address workers use for it, the advertised host under HA included; --tls-no-verify-hostname turns the name check off (see Worker control-session recovery). Each connection is admitted on its own thread under a deadline that covers its handshake and first frame, and the number admitted at once is capped (see Connection routing), so neither a client that stalls its handshake or first frame nor a stream of them can hold up worker registration or stop(). In the other direction, a worker's TLS connect to the coordinator, TCP connect and handshake together, has a deadline of its own (see Worker control-session recovery), so a coordinator that never answers cannot hold a worker either.
• Secret indirection (env://). A connector option may reference a secret as env://VAR instead of embedding it, so a job spec or persisted catalog stores a reference, not a plaintext password or key. BuildContext::param_or resolves it from the environment at deploy time; an unset variable yields empty (a clear failure, never a leak). password='env://PGPASSWORD' is the shape.

Remaining hardening, still trusted-network today: the inter-operator data plane reuses the same TLS-capable connection factories as the control plane but is not TLS by default, so run it on a trusted network segment until data-plane TLS is wired on by default. A concrete safe-to-expose recipe today: CLINK_AUTH_TOKEN set, control plane on TLS, connector secrets via env://, data plane on a private network.

Key types and APIs

Type / function	Responsibility
MessageKind, MessageBuilder, MessageReader (protocol.hpp)	The wire vocabulary and the length-prefixed, big-endian framing primitives
encode_frame / decode_* (messages.hpp)	Serialise/parse each message body around a leading kind byte
Coordinator (coordinator.hpp)	Accept connections, plan and deploy jobs, run the watchdog and checkpoint trigger, recover persisted jobs
Coordinator::Config	watchdog_interval, heartbeat_timeout, handshake_timeout, max_pending_connections, bind_host, advertise_host, outbound_flush_timeout, outbound_stall_timeout, the outbound bounds (max_outbound_control_frames, max_outbound_control_bytes, max_outbound_bulk_bytes), restart and slot-wait policy, optional autoscaler
Coordinator::JobState	Per-job in-memory state: tasks by worker, port map, checkpoint acks, restart bookkeeping, commit groups, rescale coordinator
Worker (worker.hpp)	Register, run deployed tasks via role handlers, heartbeat, dispatch checkpoint/commit/rescale frames
RoleHandler / kGenericSubtaskRole	Per-role task entry point; the built-in generic role runs any planned operator chain
DeploymentTask, PeerAddress (protocol.hpp)	One subtask's placement, bind port, peers, restore directives and key-group range
CheckpointConfig, effective_max_restarts (protocol.hpp)	Per-job checkpoint/restore/restart and state-backend settings, and the restart-policy resolution
HaCoordinator, make_file_ha_coordinator, make_etcd_ha_coordinator	Leader election and leader-endpoint discovery
Coordinator::set_epoch / epoch(), Worker::bound_epoch() / fenced_frame_count()	Fencing epoch: stamped on every control frame, enforced worker-side
kClusterProtocolVersion, check_protocol_compatibility	Wire-protocol version negotiation at the handshake
read_frame, kMaxFrameBytes, MessageReader::read_count	Bounded frame and container decoding; refuses a peer-chosen allocation
ServiceDiscovery and subclasses	worker-side discovery of the coordinator endpoint (static, env var, file)
JobSubmitter (job_submitter.hpp)	Programmatic client: connect, HelloClient + SubmitJob, await ack/completion
PluginRegistry (plugin.hpp)	Registration sink for a job/plugin's types, sources, operators, sinks, selectors and key extractors
CLINK_REGISTER_JOB, CLINK_DECLARE_PLUGIN	Emit the job/plugin C-ABI exports
PluginLoader (plugin_loader.hpp)	dlopen, verify the ABI gate, run the register hook

Configuration and knobs

Coordinator (Coordinator::Config, defaults from include/clink/cluster/coordinator.hpp):

• watchdog_interval = 100 ms; how often worker liveness is re-evaluated.
• heartbeat_timeout = 2000 ms; a worker is lost after this with no message. A value too large for the clock, up to milliseconds::max(), never passes.
• handshake_timeout = 5000 ms; the most one accepted connection's TLS handshake may take before it is closed and counted (clink_node --handshake-timeout-ms, 0 to 86400000). A connection must also finish its handshake and first frame within handshake_timeout + heartbeat_timeout of its accept, or it is closed. Zero or less leaves the handshake bounded only by that admission deadline; a value too large for the clock saturates to no deadline.
• max_pending_connections = 64; most connections being admitted (from the accept until the first frame is handled) at once; beyond it a connection from the busiest peer address is evicted, or the newcomer refused (clink_node --max-pending-connections, 0 or more; zero is treated as one). A newcomer from an address holding nothing, or whose admissions all wait only on registration, is admitted past it. At twice this, counting evicted admissions whose threads have not yet let go, only an address below its fair share is admitted; three times this is the hard bound on admission threads, each holding up to two descriptors. Set it above the number of workers that start together behind one address.
• bind_host = 127.0.0.1; control-plane bind address (set 0.0.0.0 for multi-host, and pair with TLS).
• advertise_host = empty -> defaults to bind_host; the host published in resolved peer addresses.
• outbound_flush_timeout = 1000 ms; how long stop() lets the worker connections write what is queued for them, against one deadline for all, before it discards and counts what is left and closes them (see Shutdown). Zero or less discards whatever is queued at once; a value too large for the clock saturates to no deadline. clink_node --outbound-flush-timeout-ms.
• outbound_stall_timeout = 10000 ms; how long frames may be pending for a worker with none of them making progress before the watchdog declares it lost (see Heartbeats, the watchdog and lost-worker detection). Zero or less turns the stall verdict off, and so does a value too large for the clock (up to milliseconds::max()), which never passes; a worker whose outbox stops by itself is still declared lost. clink_node --outbound-stall-timeout-ms.
• max_outbound_control_frames = 10000, max_outbound_control_bytes = 64 MiB, max_outbound_bulk_bytes = 1 GiB; the most a worker session's outbox queues: control frames by count and by bytes, Deploys by bytes (see Outbound frames to a worker). A post past a bound stops the outbox, and the watchdog declares the worker lost at its next tick. Zero refuses every frame of its class, so clink_node refuses to start with one (--max-outbound-control-frames, --max-outbound-control-bytes, --max-outbound-bulk-bytes, 0 or more).
• max_restarts = 0; per-subtask retry attempts for non-checkpointed jobs.
• restart_drain_timeout = 120000 ms; bound on a survivor drain before the job is failed.
• restart_capacity_timeout = 180000 ms; how long a restart whose drain is covered waits for free slots to redeploy before the job is failed. A worker registering restarts the clock. A cancel that lands while a restart waits, for capacity or for in-doubt resolution, ends the job on the watchdog's next tick once the drain is covered and no resolution is running: the restart is never re-fired for a cancelled job, so nothing else would end it.
• submit_wait_for_slots = 0 ms; how long submit waits for spare slots (0 = reject immediately).
• default_state_backend_uri = empty; cluster-wide default backend for jobs that chose none.
• Default control port kDefaultCoordinatorPort = 6123; history ring kCoordinatorHistoryCap = 128. The cap bounds BOTH surfaces of a terminal job: the public CompletedJobRecord in the ring and the internal JobState behind it are evicted together (oldest first), so the coordinator's per-job memory does not grow with jobs ever run. A running job is never evicted, job_errors answers from the ring after the state is gone, and snapshot_job returns nothing for a job older than the ring - the same answer it gives for an unknown id.

Worker (Worker::Config):

• heartbeat_interval = 500 ms.
• slot_count = 1; concurrent tasks this worker can host.
• peer_update_timeout = 30000 ms; max wait for PeerUpdate before aborting a task.
• http_port = 0; advertised read-API port (0 = no HTTP, coordinator proxy skips it).
• checkpoint_num_retained = 1 (clamped to >= 1); completed checkpoints kept per job.

Per-job CheckpointConfig (protocol.hpp): checkpoint_dir (empty disables checkpointing), interval_ms (0 disables periodic triggers), restore_from_dir (a checkpoint directory, or a job's marker directory in one) + restore_from_checkpoint_id, max_restarts_on_worker_loss (kRestartAuto resolves to self-heal when checkpointing is on, else fail-fast), alignment (default Aligned), and state_backend_uri (the base the job's own state root, <base>/_state/<job id>, is composed under; empty means checkpoint_dir).

OTLP export (OpenTelemetry)

clink_node coordinator|worker --otlp-endpoint=host[:port] (default port 4318) ships the full MetricsRegistry snapshot and the engine's lifecycle spans to an OpenTelemetry collector's OTLP/HTTP JSON endpoints (/v1/metrics, /v1/traces) every --otlp-interval-ms (default 10000). Off unless the flag is given: no exporter thread runs and the span sites are no-ops. The wire encoding is hand-rolled protobuf-JSON (include/clink/metrics/otlp_export.hpp, no opentelemetry-cpp dependency); a non-finite double, such as the sum of a histogram fed NaN, takes the mapping's own spelling, the strings "NaN", "Infinity" and "-Infinity", so the body always parses; transport is plain HTTP, so a TLS hop belongs to a collector agent next to the process. Spans are coarse engine transitions, never per-record: clink.checkpoint (trigger to completion, with job and checkpoint ids), clink.submit (gates to deployed, with job id, task count and the restore point when there is one), clink.recovery (one per HA-recovered job, with outcome recovered/parked/failed - parked is the capacity retry, not an error), and clink.rescale (request to rescaled-deploys for the drain and replan modes; arm to completion with mode=hot_cutover for the in-place path). The exporter's own health is visible in clink_otlp_exports_total, clink_otlp_export_failures_total and clink_otlp_spans_dropped_total (bounded span buffer, drop-oldest). The Prometheus /metrics endpoint is unaffected and remains the primary scrape surface.

Checking a configuration before deploying it

clink lint (tools/clink_lint.cpp) applies the same checks a submission is gated on, without contacting a cluster. It takes the same flags as clink run and clink_node, so a command line can be pasted in as-is, and reports every setting that would be accepted and then ignored (--checkpoint-interval-ms with no --checkpoint-dir, a half-set restore pair, --capture-records with no --capture-dir) along with combinations that contradict each other.

With --graph-json=<file> it also lints the operators (lint_job_graph in config_lint.hpp), which is a different question from whether recovery will happen: whether the operators as declared can do what they say.

• A keyed operator above kNumKeyGroups (128) is an error. key_group_range_for_subtask divides 128 groups among the subtasks, so above 128 the surplus ones get an empty range: no records, a slot each, and any keyed state written through them dropped by the range filter at restore. That is the shape of F38, and it is a property of the graph, checkable before a record moves. The coordinator refuses such a submission.
• Rescale bounds the runtime cannot reach are a warning. request_operator_rescale validates a target through rescale_parent_mapping, which takes integer multiples and divisors only, so a range whose ends are not integer factors of the current parallelism is a range every request against which is refused. The warning names the values that ARE reachable.
• Rescale bounds with no periodic checkpointing are a warning. A rescale restores the new subtasks from a completed checkpoint, so without one the request is refused and the autoscaler cannot act either.
• A graph larger than the cluster is an error, when the caller passes --available-slots. Only reported when asked: the coordinator has its own capacity check at submission and duplicating it would give two messages for one fact.

The coordinator's own settings are linted when at least one of its flags is given, with clink_node's defaults for the rest; clink_node runs the same checks at startup and logs them. The liveness checks are lint_liveness_config. The outbound checks (lint_outbound_config):

• An outbound bound of zero is an error, and clink_node refuses to start with one. Outbox::post refuses a frame whose class is at its bound, and the refusal costs the worker its connection, so every worker would be declared lost at its first such frame.
• A bulk bound below one largest frame (kMaxFrameBytes, 256 MiB) is a warning. A Deploy carries the bytes of every plugin it ships, so two sent back to back to a healthy worker, the second queued behind the first, can reach the bound and cost it its connection.
• A stall timeout at or below heartbeat_timeout is a warning. The stall verdict is meant for a worker that keeps heartbeating but has stopped reading; that short, it also takes a worker whose reader is held up for less time than the heartbeat tolerates.
• A stall timeout at or above restart_drain_timeout is a warning. A survivor that stops reading during a restart's drain is declared lost and folded into the restart by the stall verdict; the drain deadline would otherwise fail the job first.
• A watchdog period longer than the stall timeout is a warning, as for the heartbeat: the watchdog sweep runs the check, so the effective stall timeout is its period.

With the stall verdict off (zero or less) only the bounds are checked.

Exit codes rather than text are the interface: 0 clean, 1 at least one error, 2 bad usage. Warnings print but do not fail the command - a gate that refuses legitimate deployments gets switched off.

The flag parsing and CheckpointConfig assembly live in tools/cli_config_args.hpp, shared with clink run. That sharing is load-bearing: a linter that parsed flags its own way could reach a different verdict from the gate it claims to preview, and a clean lint would stop meaning the submission will be accepted.

clink_node (tools/clink_node.cpp) flags: --role={coordinator|worker}, --id (worker), --coordinator-host/--coordinator-port, --ha-dir, --etcd-endpoints/--etcd-cluster/--etcd-lease-ttl-s, --http-port, --slots, --sql-catalog-dir (coordinator, see below), --handshake-timeout-ms (both roles: the coordinator's deadline for each accepted connection's handshake, a worker's for each TLS connect to the coordinator), the coordinator's outbound settings (--outbound-stall-timeout-ms, --outbound-flush-timeout-ms, --max-outbound-control-frames, --max-outbound-control-bytes, --max-outbound-bulk-bytes; see Configuration and knobs), and the TLS flags: the coordinator's --tls-cert/--tls-key and --tls-client-ca (mTLS), and the worker's --tls-ca, --tls-client-cert/--tls-client-key and --tls-no-verify-hostname. An incoherent combination of TLS flags is refused at startup (tools/cli_tls_args.hpp). GET /api/v1/config reports the coordinator's settings in effect, these included. The etcd path additionally requires building with the etcd impl (CLINK_WITH_ETCD, linked as clink_etcd).

SQL over HTTP

Serving a console same-origin. The coordinator's HTTP server carries no embedded UI (the ops console lives in its own repository, clink-fe); clink_node --http-static-dir=<dir> mounts a built console bundle at / beside the JSON API, so one port serves both and the console needs no CORS setup and no separate web server. Serving rules (include/clink/http/static_files.hpp, unit-tested): files resolve under a canonicalised root with content types by extension; an extensionless miss falls back to index.html (browser-history deep links land on the SPA router); a missed asset with an extension is a real 404; anything escaping the root - .. segments, symlinks pointing out - is refused. The catch-all registers after every API route, and handlers match in registration order, so it can never shadow /api/v1 or /metrics. Without the flag, / answers with a JSON signpost to the API. Wildcard routes are a first-class HttpServer feature now: a trailing * in a route pattern matches any tail and surfaces it as path_params["*"].

When the coordinator is built with the SQL frontend linked (CLINK_LINKED_SQL, the default), three HTTP endpoints let a client compile and run SQL without the clink_submit_sql CLI:

• POST /api/v1/jobs/sql?mode=explain|compile|submit[&parallelism=N][&name=foo] - the request body is raw SQL text (one or more statements; no JSON wrapper, so SQL quoting is untouched). DDL (CREATE TABLE/VIEW, ALTER, RENAME, DROP) is applied to a coordinator-held session catalog so it is visible to later statements and later requests. For each INSERT / CREATE MATERIALIZED VIEW (and an explicit EXPLAIN) the mode decides the action: explain returns the LogicalPlan tree text, compile returns the compiled JobGraphSpec JSON as a string, submit runs it and returns the job id. Errors return 400 with {ok:false,error,position} (1-based byte offset). ANALYZE is rejected over HTTP (it runs a local scan; use the CLI).
• GET /api/v1/catalog - the session catalog: every registered table / view / materialized view with its columns, kind, connector and primary key.
• GET /api/v1/connectors - the SQL connector vocabulary (the WITH (connector='...') values), with best-effort source/sink flags and a category, and per entry what this binary declares: available (the connector is linked in), the capability records (variants: name, source or sink, delivery, transactional, commit-recoverable, replayable, offset model, checkpoint integration, TLS, SQL availability) grouped under the entry by the longest-prefix rule the guarantee gate uses (kafka_2pc under kafka, s3_parquet_2pc under s3_parquet), and the roll-ups sink_delivery (the strongest among its sink records), source_replayable and tls. The vocabulary is static; the records and availability are read from the capability registry, so a console shows the truth of the running cluster.

Every submit route reads the job's recovery configuration from the query string, through one parser, checkpoint_config_from_query (include/clink/cluster/submit_query_config.hpp): POST /api/v1/jobs (the multipart plugin upload), POST /api/v1/jobs/spec and POST /api/v1/jobs/sql?mode=submit take checkpoint_dir, checkpoint_interval_ms, state_backend, max_restarts_on_worker_loss, alignment=aligned|unaligned|adaptive, restore_from_dir and restore_from_checkpoint_id. An unset parameter keeps the coordinator's default; a malformed one (a non-integer or negative number, an unknown alignment) is refused with a 400 that names the parameter, never defaulted silently. Without a checkpoint_dir the job never checkpoints, so its exactly-once sinks never commit and a worker loss is not recoverable; GET /api/v1/jobs/:id reports what the job is running with.

GET /api/v1/jobs and GET /api/v1/jobs/:id carry the delivery guarantee the gate computed at submit: delivery_guarantee (an EndToEndGuarantee level) and guarantee_limiting_factor (the sink or property that caps it, empty when nothing does), and the detail adds the analyser's guarantee_reasons and guarantee_warnings. The verdict is stored on the job when the submit deploys it, so a job recovered by an HA takeover (which resubmits through the same path) carries it too, and the log line at submit is no longer the only record of it.

The session catalog is in-memory by default (lost on coordinator restart). Passing --sql-catalog-dir <dir> loads any persisted table definitions at startup and auto-saves subsequent DDL there. The endpoints reuse the same clink::sql entry points as the CLI (parse -> Binder -> optimize -> PhysicalPlanner), so the compiled spec is identical; submission is the same coordinator.submit_job path as POST /api/v1/jobs/spec. POST /api/v1/jobs/spec?name=<job>&state_backend=<uri> takes a JobGraphSpec JSON body; the optional state_backend query (percent-encoded, so a URI carrying its own ?... query round-trips) sets the job's CheckpointConfig.state_backend_uri, else the cluster --default-state-backend applies. clink_submit_sql --state-backend <uri> is the CLI front for it.

Guarantees and caveats

• Single active coordinator. There is one coordinator at a time. HA provides standby failover via leader election, not active-active. The file coordinator requires a shared filesystem and relies on OS lock release on process death; the header notes that pathological cases (for example an NFS hang) can leak stale ownership.
• etcd is optional and build-gated. The etcd coordinator only exists when the cluster is built with the etcd impl; otherwise --etcd-endpoints is rejected at startup.
• Fencing is coordinator-to-worker only. Worker-to-coordinator frames carry no epoch, so a superseded coordinator can still receive and act on status from workers it no longer owns. It cannot make them do anything, which is the dangerous direction, but its own view is not fenced.
• A worker is fenced from a superseded coordinator only once it re-registers with the new leader. It binds the leader's epoch at that RegisterAck, and the leader recovers its jobs once registrations have settled, not once every worker has bound the new epoch. A worker still on a superseded coordinator's connection, while that coordinator's trigger loop runs on, therefore accepts its frames after the takeover, checkpoint barriers included. The checkpoint ids cannot collide: each one is claimed on the job's checkpoint-id record before it is allocated, a claim lands only above what the record holds, and a coordinator whose claim is refused numbers above the record (see the checkpointing page). Trace validation checks the same of every recorded run, and diverges at a Trigger whose id the other coordinator has triggered. That keeps the two coordinators' checkpoints apart; it does not stop the superseded coordinator's other frames reaching such a worker.
• The metadata guard depends on the HA directory honouring file locks. The compare-and-set (see the fencing section above) serialises writers with flock, the same family of primitive as the leader lock; on a filesystem that grants everyone the lock (bind mounts, nolock NFS), neither the leadership fence nor the metadata CAS excludes anything - which is exactly what the start-time lock probe refuses to run on.
• Transport security. The control plane defaults to loopback and plain TCP. TLS is wired through injectable accept/connect factories in clink_node (and is itself build-gated); any deployment beyond a trusted local network should enable it. A worker checks the coordinator certificate's name against the host it connected by; a deployment that connects by an address its certificates do not carry must reissue them or pass --tls-no-verify-hostname, which leaves only the chain check.
• ABI gate is a structural fingerprint over a declared surface. By default a plugin and the cluster must agree on a SHA-256 over the declared extension surface (the headers in scripts/plugin-abi-surface.txt plus the build options that surface uses, the pinned Arrow version and CLINK_ABI_VERSION), on the target triple, and on the toolchain identity (stdlib, dual-ABI choice, sanitizer instrumentation) - so a cluster rebuild keeps existing plugin binaries deployable unless the contract genuinely moved, and a refusal names the differing headers. Incompatible submits are refused on the references-only exchange before any bytes ship; deploy-time gate refusals are fatal rather than retried. CLINK_STRICT_PLUGIN_ABI=1 restores the legacy exact-commit-hash comparison. There is no sandbox: a crash in plugin code terminates the worker process, after which the coordinator restarts the job per its restart policy (the contract is "trust your own plugins"). See design record 010.
• Admitted C++ plugins remain mapped for the process lifetime. dlclose cannot prove that module-owned factories, TLS, statics or partially registered closures are gone, and Linux teardown exposed that assumption as a process-exit crash. The loader reuses one mapping per source path/content-hash and reruns registration into each isolated job bundle, so reconnects and repeated jobs do not grow mappings. Failed pre-registration ABI/target checks are still closed safely.
• Exactly-once depends on the source and sink. worker-loss recovery rolls the whole job back to its last completed checkpoint and replays; whether that is exactly-once end to end depends on the source's replay support and on sinks implementing two-phase commit. See ./checkpointing.md and the connector docs at ../connectors/README.md.
• Restart bounds. Self-heal is bounded (default 10 attempts in a burst) so a persistently failing job stops looping; an explicit max_restarts_on_worker_loss overrides this. A checkpoint completing after a clean run (restart_budget_reset_after, default 10 minutes, since the last restart) forgives the spent budget, so the bound applies to a burst of failures rather than a job's lifetime, and deliberate rescales spend none. A survivor that heartbeats but neither drains nor dies is escalated to a job failure on restart_drain_timeout rather than force-restarted, unless the frames queued for it stall past outbound_stall_timeout first, which declares it lost.
• Determinism requirement for jobs. Inline operator type names are minted per environment, so cross-process matching relies on build_fn registering operators in a stable order.

Related

• ./architecture.md - where the control plane sits in the component stack.
• ./jobs-and-scheduling.md - graph planning, slot assignment and how a SubmitJob becomes DeploymentTasks.
• ./task-lifecycle.md - what runs inside a deployed subtask.
• ./network-stack.md - the data-plane bridges set up via SubtaskListening/PeerUpdate.
• ./checkpointing.md - barriers, TriggerCheckpoint/CommitCheckpoint and the 2PC sink protocol.
• ./fault-tolerance-and-rescale.md - the restart-from-checkpoint path, rescale choreography and schema evolution.
• ./state-and-backends.md - state-backend URIs and restore semantics referenced by CheckpointConfig.
• ../connectors/README.md - source and sink connectors and their replay/commit guarantees.