Distributed Jobs

ArkFlow Jobs are the new runtime contract for stateful stream processing and exist alongside the existing YAML Stream API. A Job is a graph of operators and edges with stable IDs; submitting one produces an immutable JobVersion and a physical task plan.

Architecture

The Hub is a pure control plane: it persists intent (SQLite), makes placement decisions, and aggregates observations. Agents are the data plane, each running a subgraph of the unified kernel. By default an edge whose endpoints land on different nodes is a placement error: agents then share only object storage (recovery artifacts) and external systems (sources/sinks). A job that opts into placement: split — with every participating node running a shuffle data plane — additionally connects the agents with direct, bounded TCP channels (see Placement modes below).

Control flow runs top-down (write intent → outbox claim-lease dispatch → Agent polls for commands); observation flows bottom-up (heartbeat lease renewal, monotonic reports, generation fencing rejects stale generations).

Event-time semantics

A Job can declare an event-time field, per-partition watermarks, an idle partition timeout, and allowed lateness. The watermark is aggregated from the minimum progress across active partitions; events beyond a window boundary are handled by the drop, route, or update policy.

Event-time fields accept Int64 (milliseconds) and Arrow Timestamp columns in seconds, milliseconds, microseconds, or nanoseconds — all normalized to milliseconds; negative timestamps are rounded with the same div_euclid rule as window boundaries, and overflow returns a field-localized error rather than silently wrapping. Null timestamps are never held indefinitely: when route is configured they follow late events into the side output, otherwise they are dropped and acknowledged.

Within a batch, the watermark advances before the current rows are classified: in a [2100, 100] batch, 100 is immediately handled by the late policy and only genuinely future rows stay held. Window aggregations preserve the input numeric type — a Float64 sum is emitted as Float64 and a Float32 as Float32 with a compatible schema; sum/min/max no longer degrade to integer sentinels (a BREAKING change; integer inputs remain Int64). Non-integral sliding windows (size=5, slide=2) enumerate every window start that contains an event timestamp instead of truncating with integer division.

Triggered windows are retained for their allowed-lateness period: a late Update within that period revises the same (operator, key, window) aggregation and re-emits the full corrected result with an __arkflow_window_update marker column; the buffer is only cleaned up after the period expires.

The keyed aggregate buffer is bounded by max_buffered_keys (default 65536 live (window, key) entries). When the bound is reached — a stalled watermark or extreme key cardinality — the oldest-window aggregates are evicted (announced by a throttled warn log naming the eviction count); the evicted windows emit nothing when their trigger eventually fires.

Embedded state and checkpoints

Hot-path state lives in an embedded KV store local to the Compute node, isolated by job, operator, and key namespace, with TTL, size metering, and a format version. A checkpoint writes a checksum-protected manifest containing the state snapshot, source positions, and watermarks to shared object storage; recovery restores state and source positions before reading any input.

On a single node, command-driven execution and checkpoint flow look like this:

Acknowledged cut

A checkpoint's source positions, watermarks, operator state, and barriers must all come from the same acknowledged boundary: before injecting a barrier, the source chain drains in-flight (non-held) acknowledgements — each ack first commits to the state journal, then advances the WAL cursor, and finally commits the source-side offset — and only then is the immutable cut containing positions and watermarks sealed. Mutations of stateful operators first enter an execution-local journal and are committed to the state backend only after their output has been acknowledged; on output or task failure they roll back, so replays never double-apply. Multi-input chains capture a snapshot of the committed epoch after barrier alignment, before releasing post-barrier data. Acknowledgements held in window buffers do not block barriers: their state stays staged, and recovery rebuilds it by replaying from the source positions.

Kafka checkpoint positions are the highest contiguous acknowledged offset per topic-partition: fan-out branches that complete out of order cannot skip unacknowledged records in a gap. A single source task keeps the connector's full-partition subscription; only multiple physical tasks perform an explicit partition assignment. On recovery the checkpointed positions are merged into the complete configured assignment — partitions without a recorded position keep their configured starting point, and the restored positions become the cursor for the next checkpoint.

WAL-enabled inputs complete their durable flush before read() returns; on ack, the WAL cursor advances first, then the native source offset commits. Recovery folds the WAL contiguous prefix already covered by a checkpoint into the local cursor, so filtered replays leave no acknowledgement gap. Shutdown, reconnects, and the processor concurrency pool all listen for the cancellation signal; on stop, workers and collectors are joined before sources, sinks, and the WAL are closed.

Only a manifest carrying the complete planned task set is sealed as Completed: missing, duplicate, or extra task entries cause rejection, and the last valid recovery point is retained while a node is offline. Upgrading to a newer Job version is allowed when the state format matches (a newer version can restore an older savepoint); downgrades and format changes have no migration path and are rejected on both sides.

Rescaling on recovery

Recovery validates the recovery artifact's recorded task set against the currently compiled plan before restoring state. A mismatch — a changed parallelism or operator topology — fails closed with a configuration error that names the task-set difference (removed/added tasks) and offers two ways out: restore the original parallelism, or reset state with a fresh checkpoint/savepoint. Nothing silently continues with empty state. Stateless jobs do not produce recovery artifacts at all — manifest sealing rejects an empty state snapshot, and local recovery selection applies only to plans with durable state — so they simply compile under the new parallelism.

Declaring rescale: true on the JobSpec opts into redistribution instead — supported for task-set changes that keep every operator represented in the saved state present in the new plan (parallelism changes qualify; removing a stateful operator does not, and fails explicitly):

yaml

jobs:
  - id: orders-rollup
    version: 1
    rescale: true
    parallelism: 4        # was 1 when the checkpoint was written
    max_parallelism: 128  # keep unchanged so key-group ownership stays stable
    operators:
      - id: source
        kind: source
      - id: sink
        kind: sink
    edges:
      - id: e1
        from: source
        to: sink
        partitioned: true
    sources:
      - operator_id: source
        input_type: kafka
        config:
          type: kafka
          brokers: ["localhost:9092"]
          topics: ["orders"]
          consumer_group: orders-rollup
          start_from_latest: false
        codec:
          type: json
        time:
          mode: processing_time
    sinks:
      - operator_id: sink
        output_type: stdout
    state:
      backend: embedded_kv
      format_version: 1
    recovery: latest_checkpoint

With rescale: true, keyed state entries from the artifact are redistributed across the new task set: the routing key is recovered from each entry's encoded key (window entries skip the 8-byte window start; stateful-operator entries strip the type prefix), ownership is computed as key_group_for_key(key, max_parallelism), and the entry is rewritten into the namespace of the task that owns that key group in the new plan. Keys and values are preserved byte-for-byte; an entry whose key encoding cannot be recognized fails recovery explicitly rather than being assigned by guesswork. The same redistribution applies to local Jobs and distributed Hub–Agent deployments: in the distributed path every node reads the whole artifact, redistributes the entries, and restores only those whose new owner is one of its own assignments — each entry lands on exactly one node. Keep max_parallelism unchanged across the rescale so key-group ownership stays stable. Rescaling is a recovery-time operation — it takes effect on the restore path (stop → change parallelism → start), not on a topology change of a running job (use placement rebalance for that).

Control plane and compatibility

The Hub persists jobs, versions, task assignments, and recovery records, and uses generations to keep stale task reports from overwriting newer intent. Agents confirm the job runtime, state backend, and checkpoint protocol versions through capability declarations. The legacy Stream YAML API is neither converted nor removed and keeps running on its original path.

A job's observed status is derived from the aggregation of all expected assignments under the same (generation, action): running/stopped is reported only when every assignment succeeds; anything still pending or retryably degraded keeps the job converging, and one node's transient failure never overrides healthy nodes. Checkpoint commits likewise require the complete set of expected assignments; a partial result from an offline node is never published as a recoverable artifact. Agents use a stable process boot identity to distinguish real restarts and a registered session token to protect requests; each re-registration restarts the report sequence at 0, and late reports from an old session are rejected without rolling back the new session's observation snapshot. Long checkpoints run in the background while heartbeats, reports, and cancellation polls continue; command failures return a terminal Failed result with correlation metadata.

Partial node failure is an incremental re-placement. When one of several dispatched nodes of an automatically placed Job goes offline (or the opt-in rebalance policy evicts it), the reconciler keeps the remembered dispatch order and replaces only the failed slot in place: every surviving node keeps its exact task set — no restart, no re-dispatch for survivors — and a replacement candidate (shuffle-capable for split placements) inherits exactly the failed node's tasks. With no candidate available, a surviving node concentrates the failed slot's tasks instead of reshuffling every task across the reduced node set. Two guards keep this honest: a successful start only satisfies the dispatch skip while its recorded assignment matches the current mapping (a drifted claim is superseded and re-dispatched — this also covers a Hub restart, which re-dispatches once as an assignment confirmation), and a node's failed observation invalidates its successful start so a crashed kernel — for example after a remote edge exhausted its reconnect budget — restarts from a recovery artifact without an operator-driven restart. Under placement: colocated (the default) a surviving node's kernel is untouched by a peer's failure; under placement: split a survivor's remote edges to the failed node still follow the fail-closed wire protocol (bounded transparent reconnect, then failure), after which the guard above restarts it with the stable mapping.

Partition edges select the downstream task by the JobPlan's key-group range rather than by modulo over physical source subtasks, so the same key arriving from different source partitions still lands on the same downstream owner. The legacy YAML Stream tumbling/session buffers keep their "aggregate-then-pipeline" ordering and emit the original schema/rows; the legacy row-count sliding_window is not misread as a time window, and incompatible configurations fail at compile time with a migration hint.

Placement modes

Task placement has two modes, selected by the job spec's placement field.

With the default placement: colocated, an assignment never splits an edge across two nodes, and Hub placement guarantees that adjacent operators sit on the same Compute node: intermediate data never leaves its node, and horizontal scale comes from source-partition splitting (e.g. spreading Kafka partitions across nodes) and independent subtasks. Computations that need a shuffle across the whole stream should chain two jobs through an external system (for example a Kafka topic repartitioned by key) — or opt into split.

With placement: split, the Hub round-robins the plan's physical tasks across the target nodes in deterministic plan order, so subtasks of the same operator can land on different nodes. Edges whose endpoints end up on different nodes are materialized as remote network edges: partitioned edges route records by key-group range to the owning subtask over one bounded TCP channel per (subtask-pair, operator-pair), with the same FIFO, barrier, watermark, and acknowledgement semantics as local bounded channels — an upstream source ack completes only after every downstream replica has acknowledged, and window-held batches stay excluded from barrier drain exactly as locally. Side edges — error sinks and late-event routes — must remain co-located; if a plan would split one across nodes, the entire placement is rejected before dispatch. The Hub dispatches split placements only to nodes that run a data plane (a configured health_check.data_port with a routable health_check.data_host, advertised as the network_shuffle capability); otherwise placement fails closed with no partial dispatch. Deployments that never set these fields keep the colocated behavior unchanged — no extra listener, no capability, byte-identical placement.

Resource-aware placement and rebalancing

When a Job does not pin its target nodes with node_ids, the Hub ranks the eligible nodes by resource headroom before every placement: freshest agent gauges first (most memory available, then most CPU headroom, then node id), and nodes that have not reported gauges last in node-id order. The ordered set feeds the same deterministic round-robin as before, and a retained placement always re-dispatches in its original order, so a Job's task→node mapping never drifts between dispatches.

By default a placement is never moved after it succeeds. Jobs that opt in with a rebalance policy may be relocated when their placed node sustains pressure (memory used or CPU above the thresholds across consecutive agent reports) and the per-Job cooldown has elapsed: the move goes through the same fenced re-placement as a node blip — the abandoned node's start is superseded, it receives a stop command, and exactly one runner remains on the best-ranked remaining nodes. Relocation restores from the latest checkpoint, exactly like any other re-placement; a pinned node_ids set cannot be combined with rebalance: auto.

yaml

streams: []
jobs:
  - id: rebalanced-orders
    version: 1
    placement: split
    rebalance:
      mode: auto            # off (default) | auto
      pressure_streak: 3    # consecutive pressuring reports before a move
      cooldown_ms: 300000   # minimum delay between relocations
    operators:
      - id: source
        kind: source
      - id: sink
        kind: sink
    edges:
      - id: source-sink
        from: source
        to: sink
    sources:
      - operator_id: source
        input_type: memory
        time:
          mode: processing_time
    sinks:
      - operator_id: sink
        output_type: drop

Failure and readiness semantics

Every validation entry point (--validate, the configuration API, local Jobs declared in YAML, and compiled Streams) performs the same side-effect-free deep build used at real startup: unknown components, unsupported state backends, and illegal graph edges fail at validation time rather than at runtime. WALs opened by a dry run are closed before it returns, and the same redb path can be reopened immediately by a real runtime. A runtime that fails a dry run, graph build, or resource connection after entering Starting transitions to Failed before the error is returned; a local Job build failure fails engine startup instead of announcing readiness while broken. Temporary resources, sources, and sinks are connected in dependency order before any task loop starts, and a partial startup closes the connected resources in reverse order.

Network edges: transparent reconnect and fail-closed

Cross-node (network shuffle) edges are fail-closed by contract: a lost connection, a protocol or authentication failure, or a resource-limit overflow fails the affected Job attempt and cleans up both ends of the edge — the attempt is then re-placed under generation fencing, never left half-open to produce or drop data.

Transient network failures are absorbed first. Within a bounded redial budget (5 attempts with backoff by default), the upstream side redials the peer and replays every not-yet-acknowledged data frame; the receiving side de-duplicates by delivery sequence (a replayed frame at or below the highest delivered seq is dropped and re-acked, which is idempotent on both sides), so the job continues without a failure report. Deterministic errors (an illegal frame, failed authentication) are also redialed while budget remains — the replay cannot succeed, so they consume the budget and the edge fails closed once it is exhausted.

Two bounds keep that machinery honest under stress. The replay cache is bounded by both an entry cap and an approximate byte budget (256 MiB by default, estimated from the Arrow batch memory of the unreceipted frames): a send whose replay registration exceeds either bound fails through the normal send-failure path instead of growing the cache without limit. And the receipt-reader idle budget is tiered: an idle connection is declared dead after 30 seconds without inbound frames, but a connection with batches still awaiting receipts gets a 10-minute budget, so a slow downstream withholding receipts under backpressure is never mistaken for a dead peer.

Acknowledgement mirrors across remote edges: a source ack completes only after all downstream copies acknowledge, and a downstream processing failure sends a Failed receipt that aborts the upstream fan-out branch immediately (via the same compensation path as local branches) rather than waiting for a barrier-drain timeout. If a receipt is lost entirely — the downstream node crashes between processing and acknowledging — the pending ack never completes, that checkpoint fails closed, and recovery replays from the last sealed cut (at-least-once).

Stream-stream join

The Job DAG supports a keyed interval join operator. A Join operator declares exactly two inbound edges and names its sides by producer (left_from/right_from upstream operator ids — channel order is a kernel-internal detail). The config carries left_key/right_key, window_ms, and optional left_timestamp/right_timestamp (default __meta_timestamp), ttl_ms, max_per_key, and join_type (default inner; left_outer / right_outer / full_outer keep the outer side's never-matched rows):

yaml

jobs:
  - id: join-orders-profiles
    version: 1
    parallelism: 1
    max_parallelism: 128
    operators:
      - id: orders
        kind: source
      - id: profiles
        kind: source
      - id: join-orders
        kind: join
        config:
          join_type: left_outer
          left_from: orders
          right_from: profiles
          left_key: customer_id
          right_key: customer_id
          window_ms: 5000
      - id: sink
        kind: sink
    edges:
      - { id: left, from: orders, to: join-orders }
      - { id: right, from: profiles, to: join-orders }
      - { id: out, from: join-orders, to: sink }
    sources:
      - operator_id: orders
        input_type: generate
        config:
          type: generate
          context: '{ "customer_id": "c-1", "amount": 42, "ts": 1757000000000 }'
          interval: 500ms
          batch_size: 1
        codec:
          type: json
        time:
          mode: event_time
          timestamp_field: ts
          watermark:
            strategy: bounded_out_of_orderness
            out_of_orderness_ms: 2000
            idle_timeout_ms: 60000
      - operator_id: profiles
        input_type: generate
        config:
          type: generate
          context: '{ "customer_id": "c-1", "tier": "gold", "ts": 1757000000000 }'
          interval: 700ms
          batch_size: 1
        codec:
          type: json
        time:
          mode: event_time
          timestamp_field: ts
          watermark:
            strategy: bounded_out_of_orderness
            out_of_orderness_ms: 2000
            idle_timeout_ms: 60000
    sinks:
      - operator_id: sink
        output_type: stdout

Semantics and boundaries:

• A left row and a right row match when their keys are equal and their event timestamps differ by at most window_ms. Matches emit immediately regardless of join_type (at-least-once — downstream must tolerate replays after recovery).
• Output columns are the original left columns prefixed l_, the right columns prefixed r_, plus join_key. With an outer join_type, the side that can be all-null in unmatched emissions is nullable in the output schema (r_* for left_outer, l_* for right_outer, both for full_outer); inner output schemas are unchanged.
• With left_outer/right_outer/full_outer, a never-matched row on an outer side is emitted once the chain watermark passes timestamp + window_ms + ttl_ms, with the opposite side's columns all null. Rows that matched at least once never re-emit as unmatched. The emission needs the opposite side's schema: while that side has not produced a batch, evicted rows park in a bounded queue (max_per_key per side, oldest dropped) and flush once the schema arrives. Processing-time sources without watermarks never emit unmatched rows — prefer event-time sources for outer joins.
• State is bounded: rows are evicted once the watermark passes timestamp + window_ms + ttl_ms, and each key holds at most max_per_key rows per side (oldest first). Under an outer join_type, capacity eviction also emits the evicted row as unmatched — without a watermark guarantee, so a later match can double-emit (at-least-once artifact). Processing-time sources without watermarks rely on the capacity bound — prefer event-time sources for joins. Capacity evictions are observable: each one emits a throttled warn log (side, key, buffer depth, max_per_key, and the suppressed count within the throttle interval). Inner-side capacity evictions drop the row, so the warn is the only trace of data that fell outside the join — a sustained stream of these warns on a hot key means max_per_key (or the window/ttl bounds) is too small for the skew.
• Recovery rebuilds the buffers (and per-row matched flags) from checkpoint replay; there is no separate join snapshot. The chain watermark is the minimum across both sides, so replay cannot produce a false unmatched row beyond the documented late-data contract.
• Each side's schema must stay stable for the stream's lifetime, and each side's upstream operator must feed the join from a single subtask — a multi-subtask side fails at graph build time with guidance to set the upstream parallelism to 1.
• Not supported: temporal (lookup) joins, non-equi joins, and cross-node shuffle joins (the join operator and both its upstream edges stay co-located on one node).

Legacy Stream join buffers (including window buffers with a legacy join field) still fail compilation, with guidance pointing at the Job DAG join operator.

Operational limits

A single Hub supports a fleet of up to 256 agents (MAX_NODES): node registration beyond the cap is rejected rather than silently degraded, so an over-capacity fleet fails visibly at join time. Sizing deployments beyond 256 nodes per Hub is untested — reach out to maintainers before planning larger fleets (splitting fleets across multiple Hubs with agent multi-Hub discovery is one path).

The ceiling is backed by measured evidence from the production-readiness harness (in-memory store, single host):

• Scale staircase 25 → 64 → 128 → 256 nodes at full fleet: command round p99 stays flat at 1044–1051 ms regardless of fleet size (no knee observed); Hub process RSS at 256 connected nodes: ~441 MB.
• One-hour soak (64 nodes, 30 s session TTL, 10 Hub restarts, full-fleet rebirth): 2859 command rounds, round p99 1055 → 1193 ms (+13%, no drift), RSS growth 38.5 KB/s (below the 50 KB/s gate), and all bounded history tables converged to their retention bounds.

API examples

http

POST /api/v1/jobs/validate
POST /api/v1/jobs
PUT  /api/v1/jobs/{job_id}/desired-state
GET  /api/v1/jobs/{job_id}
GET  /api/v1/jobs/{job_id}/detail
GET  /api/v1/jobs/{job_id}/versions
POST /api/v1/jobs/{job_id}/checkpoints
POST /api/v1/jobs/{job_id}/savepoints
POST /api/v1/jobs/{job_id}/upgrades
POST /api/v1/jobs/{job_id}/upgrades/{upgrade_id}/rollback

Both the workbench and API clients should call validate to check the compiled plan and node capabilities first, then submit with stopped and inspect detail; switch to running only after confirmation. Version upgrades require the job to be stopped and converged, with a completed savepoint whose state format is compatible; when an upgrade fails the job stays stopped and the operator explicitly restores the old version. The checkpoint/savepoint lifecycle is bound to the job version and the state format version.