Repository navigation
Feature/implement details new compute setup - #852
Merged
Edwardvaneechoud merged 616 commits intoOct 28, 2026
Merged
Conversation
…zy match included
…lds, not from the worker's files Why: a fuzzy match on a worker that shares no disk with core failed in core with "No such file or directory". The worker backend built each side with no schema, so checking the join and selecting the side's columns resolved the plan of an input the worker had stored, which opened the worker's file from core. FuzzyJoinPayload carries left_schema and right_schema, optional and None by default. The Fuzzy Match node function fills them from its input engines, which for a stored input hold the schema its result reference carries. In the worker backend _fuzzy_sides builds each side over that schema, so the join is checked and each side selected and renamed over those columns without its plan being resolved; both the /v1 operation route and the task route go through it. LocalBackend takes the fields and does not read them, as it prepares the sides over its own storage. The zero-volume compose test runs the elaborate flow whole on the worker again, its fuzzy match no longer placed in core, and checks that no edge crossed a backend boundary, nothing was written to the exchange store and core cached nothing. CLAUDE.md and flowfile_core/CLAUDE.md say so. Tests: on macOS with Docker Desktop, with the host's AWS configuration hidden. A new case in test_worker_backend_v1.py stores three names on the session worker, fuzzy-joins two leaves of that result and fails if core calls LazyFrame.collect_schema during run_op; it fails without the backend change. test_worker_backend_v1.py and test_worker_backend.py: 167 passed with FLOWFILE_COMPUTE_PROTOCOL=0 and with 1, where they gave 166. test_op_payloads.py and test_unreadable_location.py: 104 passed in each state; the fuzzy cases of test_flowfile.py, test_executor.py and test_local_backend.py: 11 passed in each state, both as before. pytest tests/remote_worker -m docker_integration: 6 passed, 8 deselected in 53 s, the elaborate flow on the worker among them; the services set their own flag, and the six include the flag-off case. ruff format and check are clean on the five Python files. The workflow's "Zero-volume compose" job was not run here; it runs on the pull request.
…red where core cannot read them
…d, not from their stored files
…d, not from their stored files Why: a multi-output node's preview collected the first 100 rows of the output's plan in core. Over a stored result, a random split's other output or a custom node's named outputs, that plan scans the file the backend wrote, which on another host only that worker has, so the preview of every output failed on a missing file, the default one included. `flow_node.py`: with `FLOWFILE_COMPUTE_PROTOCOL` on and a reference held, `get_table_example` takes the default output's rows from the node's cached sample, the sample of the node's own reference, and any other output's, and the default's when none was kept, through `bounded.head_rows` of the output's plan on the backend that holds the node's reference: a deferred submit, one sample of 100 rows and a release, as the formula editor already reads a secondary output's first row. A secondary output's rows are not cached. With the flag off, or for a node that holds no reference, the rows are collected in core as before. The same branch serves every multi-output node, so the split filter's, a Run Flow node's and a kernel custom node's outputs read the same way. `flowfile_core/CLAUDE.md`: the Bounded reads bullet names the preview as a third reader that goes through a backend. Tests: a case in `execution/test_unreadable_location.py` runs a random split and a custom node loaded from a node file over a backend that stores their outputs where core cannot read them, in both flag states, and previews each output: with the flag on the split gives 3 and 3 rows, together the six inputs, the custom node 4 and 2, and the default output gives the same rows again from its plan once its cached sample is dropped; with it off each preview still fails on the missing file. Against the unchanged `flow_node.py` the flag-on cases fail. The module passes 54 in each flag state (50 before); the flow-node and custom-node suites pass 584 in each, three ADLS cases in `test_flowfile.py` failing before and after because no Azurite was running. Not run: the worker's suites (no worker change).
…reached on release
…it held there Why: a flow moved off a worker that cannot be reached could not run until core restarted. Its nodes keep the results and the model they held on that worker across the move, and the next run asked the worker whether each result still existed (with `FLOWFILE_COMPUTE_PROTOCOL` on) or released it there (with it off); either call raised `ComputeUnavailable` and failed the first node, on every run. `flow_node.py`: `release_result` logs a warning naming the node, the backend, the result id and the error when a release cannot reach its backend (`ComputeUnavailable`) and goes on with the next reference; it forgets the references before it releases them, so the result is dropped. Any other release error raises. `builders/ml.py`: Train Model's release of the model it replaces does the same, on the flow log, and the node then holds the new model. `executor.py`: the decision asks whether a held result vanished only of a node that has run with its current setup, the node the "already ran" rule would otherwise skip. A move to another backend resets that, so the first run there never asks the backend the flow left, and the first call to it is the release, which logs and drops. A node that has run and whose own backend is down still fails with `ComputeUnavailable` at that check; an unreachable backend is never read as a missing result. `flowfile_core/CLAUDE.md`: the Execution package bullet says so where it describes the release, Train Model's release and `artifact_is_missing`. Consequences: on the first run after the move a release to a host that does not refuse the connection waits the connect timeout (5 s) once per held reference; a dropped result stays on that worker until its age sweep if the worker comes back; `close_flow` and `delete_node` over a backend that cannot be reached now log per node and finish instead of raising at the first node; a node that has not run with its current setup and holds a result that is gone asks its backend one `exists` fewer and logs its own decision's reason, `NEVER_RAN` or `PERFORMANCE_MODE`, where it logged `CACHE_MISSING` (a node with `cache_results` still logs `CACHE_MISSING`, from its cache check). Tests: a case in `flow_node/test_executor.py` builds manual input → Train Model → Apply Model, gives every node a stored result and Train Model a model on a worker backend at a port nothing listens on, moves the flow to `local` and runs it, in both flag states: the run succeeds, every result and the model are held on `local`, the scores are 5 and 13, and the old worker is asked no `exists` and four releases, three results and the model, each `ComputeUnavailable`. Against the unchanged code both states fail at the first node, the flag-on one at the decision's `exists`, the flag-off one at the release. The executor, unreadable-location and three ML modules pass 230 in each flag state (228 before). Not run: the worker's suites and the conformance suite (no worker or backend change).
…f resolving it in core
…f resolving it in core Why: with the compute protocol on, a cloud read naming a connection the catalog does not hold for its user, or given no user, still fell through to core resolving the connection and putting its credentials in the plan encrypted, where every stored connection's read names it by reference and the executor resolves it. `CloudStorageReader.get_read_credentials` now refuses such a read with the flag on, a `ValueError` naming the connection (`Cloud connection "<name>" is not stored for user <id>`, or `is read with no user`). A read with no saved connection (a cloud change read's `aws-cli` one), `env_vars`, managed identity, a GCS read through gcsfs and the flag-off path are unchanged. The designer's reader, the cloud change read and the catalog target resolve the stored row for their user before the read, and `flowfile_frame` stores a connection built in code for the frame's user (`create_cloud_storage_connection`) before its reader names it, so none of them meets the refusal; only direct callers of `FlowDataEngine.from_cloud_storage_obj` do. `flowfile_core/CLAUDE.md` says so. The tests that call `from_cloud_storage_obj` with a named connection and no user (`test_flow_data_engine_cloud.py`, `test_flow_data_engine_adls.py`, `test_cloud_csv_read_defaults.py`, `test_code_generator.py`, `test_storage_browser_s3.py`) store that connection for user 1 and read as user 1. `test_cloud_plan_secrets.py` adds a frame read of a connection created in code, whose plan carries the reference with the flag on and `$ffsec$` with it off, no secret in either, and a direct read refused with the flag on for a connection never stored and for a stored one read with no user, the flag-off fall-through pinned (the no-user form asserted by the provider's class, since it has no `$ffsec$` marker). Tests: `test_cloud_plan_secrets.py` 101 passed, 1 skipped with the flag on and off (95 and 1 before; the skip is the iam_role reader on the worker with the flag, by design); the five direct-caller modules' cloud reads 53 passed in each state, before and after, ADLS against Azurite; the frame's cloud tests 36 passed in each state. ruff clean on the changed files.
…l is on A GCS connection with an endpoint was read through gcsfs in core: the key went into the pickled filesystem or the plan's options, and a CSV, NDJSON or emulator Delta read put every row in the plan. With the protocol flag on it is now read like every other cloud connection: the plan names it by reference and reaches the endpoint through object_store's gcs_base_url in a service-account key that holds no key, and the backend that runs the read answers its schema. The flag-off read and the writers keep gcsfs.
Why: a GCS connection with a custom endpoint was the one cloud read core still built and ran itself. Core made a gcsfs filesystem with the connection's key and opened the bucket while building the plan, so the key travelled in the pickled filesystem or the plan's options, and a CSV, NDJSON or emulator Delta read put every row in the plan. With the protocol flag on, such a connection is now read like every other cloud connection. `CloudStorageReader.get_read_credentials` no longer makes an exception for it: a `service_account` connection gets a `CredentialRefProvider` and an `env_vars` one none. `get_plan_storage_options` gives it a service-account key with no key material, its `gcs_base_url` the endpoint and OAuth off, which is how object_store reaches a custom GCS endpoint (Polars refuses gcsfs's `token` beside `endpoint_url`, and object_store ignores `endpoint_url`). `FlowDataEngine.from_cloud_storage_obj` takes no gcsfs path and keeps the backend, so parquet, CSV, NDJSON (a file or a directory) and Delta are native scans held on the backend for their schema, as S3 and ADLS reads are. The worker exchanges the key for a bearer as for any GCS read, so a real endpoint now gets a token; an emulator connection of type `service_account` whose key cannot be exchanged now fails at that exchange. A catalog stored on GCS behind an endpoint over a `service_account` connection gets a scan that names the connection, where Polars refused its options before. With the flag off, and for the writers, gcsfs stays. The fake GCS fixture starts the server with `-public-host localhost:4443`, without which it answers 404 to the path-style requests object_store sends; a `test-fake-gcs` container started before this change has to be replaced for the native reads to reach it. Tests: `test_cloud_plan_secrets.py` 117 passed, 1 skipped in each flag state (101 and 1 before). The plan check with no I/O gains a `service_account` and an `env_vars` connection with an endpoint over every format; a run on the worker over fake GCS reads seven readers (parquet, CSV and NDJSON files and directories, Delta) with core's gcsfs, options builder and credential resolution patched to raise, every count 100 000 and no plan core serialised over 10 kB with the flag on, every reader failing in core with it off; a catalog target over a GCS namespace with an endpoint names its connection and carries no key with the flag on, and Polars refuses it with the flag off. The GCS reader tests (`test_flow_data_engine_gcs.py`, `test_flowfile.py -k gcs`) 14 passed and the frame's cloud tests 36 passed, in each flag state. Not run: the rest of the core suite, and a real GCS endpoint, so the key exchange against one is not exercised here (fake GCS accepts any bearer).
A read whose type inference was widened on submit was held under `<hash>_infer<rows>`, which nothing that asks for a node's result under its hash finds: a cached read was read and widened again on every run, its boundary edge had no row count, and the missing-file retry and a formula gate's probe took it as holding nothing. Each backend's submit replaces whatever the id holds, so the resubmit now goes under the hash itself.
… is widened Why: a CSV read whose type inference was widened on submit held its result under `<hash>_infer<rows>`, while everything that asks for a node's result for its current setup looks under the node hash. Such a read with `cache_results` was read and widened again on every run, its boundary edge had no row count to be priced with, and the missing-file retry and a formula gate's probe took it as holding nothing, so they ran on the run's backend instead of the one holding the read. `FlowNode._submit` now submits the widened read again under the node hash. Every backend's submit replaces what an id holds (the worker's `run_submit` releases a held record and overwrites the failed job's entry, over `/v1` and the task routes; `LocalBackend.submit` drops it), and a failed attempt holds nothing, so nothing is released first. `held_result`, the cache decision, `BoundaryEdge.cost`, the retry and the probe are unchanged and now find the widened result. On the worker the job entry under the hash reads `failed` between the two submits and `done` after. The docstrings of `_do_execute` and `_submit`, and the execution-package bullet in `flowfile_core/CLAUDE.md`, say the resubmit goes under the same id. Tests: two cases in `test_mixed_placement.py` over a 120 000-row CSV whose `val` turns to text at row 60 000, read on the worker with `cache_results` (a conflict early in the file is caught by the read's one-row probe in core, which widens the plan before it is submitted). Flow 8518 (MinIO) runs it into a filter three times: the read is submitted twice under its hash and held with its row count; the second run serves it from its cache and prices its edge at 120 000 rows; in the third the first export of it finds it gone, and the retry reruns it on the worker before the filter crosses it again. Flow 8519 puts a formula gate on the worker after it, whose probe reads on the worker. Both were red before the change in each flag state, the read held under `<hash>_infer10000`. The module: 21 passed with the flag off and with it on. The escalation's own tests (`test_flowfile.py -k infer`): 4 passed in each state; the flow-node suite, the unreadable-location cases and the parameter retry: 358 passed in each state, as before the change. The worker's suites, the conformance suite and the frontend were not run: nothing in the worker, the backends or the frontend changed.
…that is no PEM or comes with an http URL
…ent its two refusals
…nd both submit sockets A backends row's CA certificate becomes one SSL context per worker URL, mounted on the worker session as an adapter and handed to the task-route and /v1 submit sockets, so the handshake gate, its probe, the task routes, /v1 and capabilities() all verify that worker against it. The gate's own GET and capabilities() now go through the session around the gate (gated=False) instead of a bare requests.get, so they see the adapter too. Nothing is written to disk; a worker with a public certificate or plain http is reached as before.
… http to another host
… worker and how the worker serves TLS
…ate in core Why: the worker served plain HTTP and core trusted public certificates only, so between core and a worker on another host the token and every plan crossed in clear, and since a plan carries code, the token is code execution on the worker. A worker with a self-signed certificate needed a proxy in front of it. The worker serves TLS itself when `FLOWFILE_WORKER_TLS_CERT` and `FLOWFILE_WORKER_TLS_KEY` name a certificate and its key (`main.py`, uvicorn's `ssl_certfile` and `ssl_keyfile`). A configured worker's row gains an optional `ca_cert` (migration 037), taken by `POST /compute_backends`, which refuses one that is no PEM or comes with an `http://` URL and never answers it. The registry hands it to `WorkerBackend`, which has the transport trust it for the worker's URL (`trust_ca`): one SSL context built from the PEM, nothing on disk, mounted on the worker session as an adapter and handed to the task-route and `/v1` submit sockets, so the handshake gate, its probe plans, the task routes, `/v1` and `capabilities()` all verify that worker against it. A backend made without a CA drops the one an earlier backend on its URL was given, so a worker removed and added again is trusted as its new row says, with its new CA or the public CAs. The gate's `GET /capabilities` and `WorkerBackend.capabilities()` go through the worker session around the gate (`gated=False`) instead of a bare `requests.get`. A worker with a public certificate or plain `http` is reached as before. The Backends page takes the CA as a PEM field and warns while the URL is plain `http` to a host other than loopback; the remote-worker page says the token is code execution on the worker, to keep its port off untrusted networks and serve TLS beyond a trusted one, and how to set it up. Tests: `test_worker_tls.py` starts a worker with a self-signed certificate; a backend naming it as its CA reads its capabilities and submits over `wss` with the REST fallbacks made to fail, and one naming none is refused as unreachable with `CERTIFICATE_VERIFY_FAILED` and nothing held, 6 passed over both flag states. The route, registry and handshake tests 98 passed, among them a worker removed and added again on its URL, first with another CA and then with none, which leaves it on the public CAs; the worker backend suites 167 passed with the flag off and 167 with it on; the migration tests 54 and the worker's import purity 5 passed; vitest 199 files and 2937 tests passed, `vue-tsc` and `eslint` clean; the docs build with its eight standing griffe warnings. Not run: the rest of the core suite, `tests/cloud_e2e`, `tests/remote_worker/` and the worker's suites beyond import purity. Every REST call core makes to a worker now goes through `gated` and every task-route socket through `ssl=`, but for a worker no backend gave a CA the call is the one it was: `gated` is on unless the gate's own GET or `capabilities()` turns it off, `ssl_context_for` answers `None`, websockets' default, and nothing is mounted on the session. Those suites reach their workers that way, over plain `http`, and the worker backend suites drive the session, the gate and both sockets that way in both flag states; the worker's change is two uvicorn options that stay unset in every worker those suites start. Nor was the Backends page opened in a browser: vitest here mounts no component, so the field and the warning are tested through the composable and its predicate.
…storing its outputs fails
…e each task-route split A Run Flow node adopts its child's stored results once per iteration, and an iterate-mode output concatenates every iteration, so the results the node adopted on its previous run are released when the whole run ends rather than after each child run; the nodes downstream, whose plans scanned them, then count as not run. Over the worker's task routes a random split now stores each split as its own task, as the embedded and /v1 backends do, and releases the shuffle once every split is stored. Every store is tracked under the operation's job id, so a cancel reaches whichever store is running. The handshake, split and cancel tests follow both changes, and the frame stubs follow the to_struct signature.
The session results folder is built from the request's flow id, so checking the file against it let that value through as the base. The shared volume comes from the kernel manager, and it is the boundary the kernel can see.
Brings in the dompurify and happy-dom bumps. The frontend lockfile's own version took main's 0.23.3, which matches package.json.
With no sandbox (desktop mode: the folder browser, the project export target and the List Files node) the start path went unchecked. It is now resolved inside one of the filesystem roots, the same bound validate_path_under_cwd and the explorer's own path lookups use, so every path the explorer holds passed the containment check.
…-storage root The kernel route's path check refused every object-storage URI, so a table in a cloud-backed namespace could not be registered. A URI is now accepted when it is the namespace's storage root joined with one directory name, the location the catalog writer gives a table there; local paths are checked as before.
The unreachable-worker test records the model it plants as the Train Model node's own fitted output, since a node now releases only a model it fitted. The filesystem-root test takes the root of the temp folder's own drive, as the Windows runner's working directory is on another drive. The placement fallback spec minimizes the node palette before it hovers the badge, which the palette covered.
The URI branch returned a location built from the request's own directory name, which reached the service's local reads unchecked. The key path is now normalised and required to start with the namespace root's path, and the location is rebuilt from the root and that key; it must still equal the URI that was sent, since an object store keeps '.' and '//' segments literally.
Edwardvaneechoud
marked this pull request as ready for review
October 26, 2026 06:26
|
Claude finished @Edwardvaneechoud's task in 24s —— View job Review of PR #852 (compute backends /
|
Cut the long docstrings of the compute protocol (WorkerBackend, run_op, placement, the embedded worker, the worker's service and routes, the handshake) to what a caller needs: what each thing does, its contracts and the non-obvious why, with wire details left to docs/for-developers/compute-protocol.md. Multi-line comment blocks are one line each. Docstrings that no longer matched the code are corrected: a random split stores each split as its own result, a failed submit-socket connect does not re-arm the handshake, and every operation kind takes its payload under the "payload" key. Only docstring and comment text changes.
|
Claude finished @Edwardvaneechoud's task in 18s —— View job Review of PR #852 (compute backends /
|
…ails-new-compute-setup # Conflicts: # flowfile_frontend/src/renderer/app/components/layout/Header/HeaderButtons.vue
Multi-line # comment blocks in the source files this branch touches, most of them from main, are cut to a single line or removed when they restated the code. Rationale that still needs more than a line moved into a short docstring, a constant's into an attribute docstring. Section banners are gone. A few comments that no longer matched the code are corrected, among them the foreign-key note in KernelManager.delete_kernel and the pool-member drain rule in spawner.handle_task. Only comment and docstring text changes.
|
Claude finished @Edwardvaneechoud's task in 21s —— View job Review of PR #852 (compute backends /
|
…gh one table The worker's materialisation policy reads fanout: a wide plan is stored when it must be, when it is previewed, or when more than one consumer reads it, so the result is written once instead of computed again inside each consumer's plan. Core counts only the consumers the run executes (execute_node(running=)): the plan's nodes less those skipped so far, the fetched node and its inputs in a single-node fetch, every consumer when no run is named. WorkerBackend and EmbeddedWorkerBackend run POST /v1/ops/<kind> through one V1_OPS table (payload, request body, query parameters, held refs, outputs and metadata per kind). The embedded worker calls each service body by the kind's name through validate_call, and a body it refuses is PlanRejected, as the route's 422 is. A ResultRecord whose fields do not fit its state is refused at construction. Review fixes: the embedded ValidationError mapping, V1_OPS as the one table, the executor's set copy and double unpack, the docs and the materialise-rule diagram, the fanout tests above the main guard, a test that every kind names a service body, and the recording backend running one node at a time.
|
Claude finished @Edwardvaneechoud's task in 20s —— View job Review of PR #852 (compute backends /
|
…d markers for deprecated endpoints and code for future deletions
…e unused fixtures, and streamline imports
|
Claude finished @Edwardvaneechoud's task in 22s —— View job Review of PR #852 (compute backends /
|
…o Backends, update descriptions, and streamline worker pool management
…es, and ensure compatibility checks
…d useFlowExecution tests
|
Claude finished @Edwardvaneechoud's task in 16s —— View job Review of PR #852 (compute backends /
|
Edwardvaneechoud
deleted the
feature/implement-details-new-compute-setup
branch
October 28, 2026 18:20
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
No description provided.