[Edwardvaneechoud]

Closes #823

Long-lived branch: it collects the work that lets a flow's compute run on a worker on another host, in reviewed pieces that land on it as pull requests (each marked "Part of #823"), and it merges when that work is complete. It stays a draft until then. The roadmap's Remote workers entry describes the direction.

Why

Core never holds a dataset. For each node it builds a Polars query plan and posts it to flowfile_worker, which runs it in a child process, writes the result to its own cache directory and answers with a plan over that file. That works because both services see one disk. Core opens the worker's files itself for previews, the formula editor's first row, pivot values, column statistics, the published data API, exports, catalog previews and the check whether a cached result still exists. The local-or-remote decision lives in globals (OFFLOAD_TO_WORKER, WORKER_URL) and in inline execution_location branches across the engine. A failure between the two services used to degrade silently into in-core execution. And nothing can say that a node cannot run on a given compute.

What this branch delivers

In order, each piece as its own pull request into this branch:

• Loud failures. A typed error family (ComputeError: ComputeUnavailable, VersionMismatch, ResultMissing, PlanRejected, OperationUnsupported, ForeignRef, TaskFailed). An unreachable worker is never read as a cache miss. The in-core continuation after a worker failure and both fallbacks of the published-API runner are gone: a plan the worker cannot load fails the node, and the published API answers 503 when the compute is unreachable. A version handshake on first contact (GET /capabilities: app, Polars and Python versions, platform, plugin versions and plugin library paths, plus a functional probe per expression plugin) refuses a mismatched worker. POST /shutdown answers loopback peers only. A ratchet test pins the number of inline location compares and of modules reading the worker settings; both may only fall.
• One execution seam. flowfile_core/flowfile/execution/: the ComputeBackend interface and its value types, a registry (built-ins plus the flowfile.compute_backends entry-point group; FLOWFILE_COMPUTE_BACKEND, else the offload flag) and an ExecutionContext that a flow derives from its settings and every stage thread and subflow sees. LocalBackend runs in core's process; WorkerBackend is the client of flowfile_worker over its existing routes. Nodes submit their plan with hints, and each backend's materialisation policy reproduces the previous behaviour. Every operation a node runs whole (readers, writers, catalog Delta writes, fuzzy join, random split, custom nodes, ML train and apply, the kernel parquet hand-off, the custom-node dry run) and every bounded read the engine makes (a count, a first row, a null profile, pivot values, schema and column statistics, the Explore Data node's charts) goes through the backend, with no in-core fallback when it fails.
• The worker settings read in one place. OFFLOAD_TO_WORKER and WORKER_URL are read only by the backend registry; the ratchet asserts zero. A conformance suite checks the backend contract over both backends.
• No shared disk, behind FLOWFILE_COMPUTE_PROTOCOL. Worker /v1 routes. A result is a reference with its schema and row count; previews, counts and schemas come over the wire; plans carry credential references, never secrets; catalog tables are addressed by identity. Proof: core and worker started with different storage directories run a flow with previews.
• Remote workers. Capabilities, node requirements and placement, which refuses a flow its backend cannot run before it starts; per-node execution_location; headless runs can name a backend; data crosses between two backends through export and import; a CI job runs core and a worker with no shared volume through a database read, a transform and a catalog write; the flag defaults on and the legacy routes are deprecated; a deployment page and a developer page describe the two-host setup.

Per-node placement across backends with an exchange store, and Polars Cloud as a backend, come after this branch.

Behaviour changes, all deliberate

• An OOM-killed worker child fails the node instead of moving the dataset into core.
• The published API answers 503 instead of collecting in core when the worker is unreachable.
• A node's cached result does not survive a core restart.
• The export route and the kernel parquet hand-off follow the flow's execution location.
• Later, behind the flag: cloud reads resolve aws-cli and iam_role connections with the executor's identity.

Verification

CI runs on every push to this branch and on every pull request into it: core, worker and frame suites, Windows, web and end-to-end. The core conftest restores the default backend after every test and fails the test that changed it. Ratchet today: 1 inline compare and 6 reader modules, down from 19 and 14.

[netlify]

✅ Deploy Preview for flowfile-wasm canceled.

Name	Link
🔨 Latest commit	c3ebc74
🔍 Latest deploy log	https://app.netlify.com/projects/flowfile-wasm/deploys/6ada71b1fa916aa1ec501ea9

[claude]

Claude finished @Edwardvaneechoud's task in 32s —— View job

---

Review of PR #827: compute interface and loud worker failures

I read the CLAUDE.md files, the error family, the handshake, the worker /shutdown and /capabilities changes, and api_runner._materialize. I did not read through the rest of the 79 files line by line, and I ran no tests. The design is coherent and the failure contract is clear. These are the points I'd look at before merging.

Findings

1. Handshake gate is open to other threads while the check runs (flow_data_engine/subprocess_operations/handshake.py, ensure_handshake).

   • _handshake_done = True is set before the checks, so the probe's own calls pass through. The side effect is that any other thread calling the worker during that window skips the gate. This includes parallel stage threads, API requests and a REST-fallback poller.
   • On a mismatched worker, one of those threads can ship its plan before VersionMismatch is raised. The docstring covers the probe's calls but not this case.
   • A re-entrancy marker would keep the guarantee ("nothing goes to a mismatched worker") without losing the probe path. For example, use a thread-local flag or a threading.local set only on the probe thread, plus a lock so concurrent callers wait for the first result.
   • The module global is also unguarded when reset_handshake is called from the session on an outage. A failed check clearing the flag can race with a concurrent success.

2. Handshake probe imports polars_ds inside _probe_plans.

   • If polars_ds is missing or fails to import in core, _probe_plans() raises outside the try/except ComputeError block in probe_plugins. The error would surface as a raw ImportError, not a typed compute error.
   • If polars_ds is a required dependency, say so. Otherwise skip plugins that are absent on both sides.

3. /shutdown loopback gate.

   • It reads request.client.host, which is correct for a non-proxied setup. If the worker runs behind uvicorn with --proxy-headers or forwarded_allow_ips, request.client can be rewritten from X-Forwarded-For, which a remote caller could set. Please confirm the worker's uvicorn config does not enable that.
   • I found no in-repo caller of the worker /shutdown other than the Tauri shell (127.0.0.1), so docker/compose is unaffected. Worth a line in flowfile_worker/CLAUDE.md that a containerised orchestrator can no longer stop the worker this way.

4. api_runner._materialize (flowfile_core/flowfile_core/flowfile/api_runner.py:139).

   • file_ref=f"__api_{api_node.hash}" is the same for concurrent requests with identical parameters on the same node. Two simultaneous requests would share a worker result ref, and one could clear or overwrite the other's result. This may predate the PR, but now that failures are loud it would show up as a spurious ResultMissing or TaskFailed. Adding a per-request uuid to the ref would remove the risk.
   • The final result.collect() is a core-side collect of a FlowDataEngine. It is capped by max_rows and predates the PR, but it is still a core materialisation. Confirm it fits the "core never collects" contract when max_rows is None.

5. errors.worker_failure string handling.

   • ComputeUnavailable.__str__ branches on the lowercase text "connection refused". This is fragile across OS and library versions (Windows reports a different message). Consider matching on the exception type where it is raised and passing a flag.
   • The worker's time-limit kill surfaces as PlanRejected, per your docstring. That is misleading for users, who would read it as a bad plan. A dedicated TaskFailed or a timeout subtype would be clearer. If you keep it, make sure the message text always says "time limit".

6. Deliberate behaviour changes to confirm with release notes.

   • An OOM-killed child now fails the node instead of falling back to core.
   • The published API now answers 503.
   • A node's cache no longer survives a core restart.
   • These are documented in flowfile_core/CLAUDE.md. The cache change is the most user-visible (re-running scheduled flows after a restart), so it deserves a changelog entry.

Tests and conventions

• Test coverage looks thorough: handshake, results-exists-unreachable, plan-rejected, held-results release, and the ratchet test.
• The ratchet test is a good guard. Make sure it fails with a readable message that tells the next contributor what to do.
• Docstrings in handshake.py and errors.py are long relative to the repo's "one short line" comment rule in CLAUDE.md. They carry real why content, so I'd trim rather than remove them.
• No database/models.py changes, so no Alembic migration is needed.
• No public FlowFrame API change that I saw, so no stub regeneration needed.