Make Clustrix a data mover: stage declared inputs and outputs (reverses the "not a data mover" limitation)

[jeremymanning]

Summary

docs/source/introduction.rst:115 currently declares:

It is not a data mover. Clustrix ships your code and arguments, not your dataset. If your function needs a 200 GB file, that file has to already be reachable from the worker. The filesystem utilities help you inspect and locate remote data, but they are not a transfer service for bulk inputs.

That is an accurate description of current behaviour and an inaccurate description of intent. The record shows data movement was an architectural goal, was designed, was partially built, and was abandoned mid-wiring — after which the docs described the abandonment as a deliberate scope boundary.

Full design: notes/data-mover-design.md. This issue carries enough to act on without it.

---

1. The conflicting design element (found, quoted)

Issue #64 (CLOSED), "Core Architecture: Function Serialization and Dependency Management for Remote Execution", lists under Current Approach Limitations — i.e. as a defect to fix:

3. No local file support - cannot access local modules, data files, or custom code

and under Proposed Solution Direction:

Dependency Packaging Approach

1. Dependency Detection: Analyze function for local imports, file references, and dependencies
2. File Packaging: Create zip archive of all local dependencies
3. Remote Deployment: SCP and unpack files on cluster with unique identifier (MD5 hash)
4. Environment Recreation: Patch import paths and execute function in recreated context

It even names the hard parts: "How to maintain security while allowing file transfer?", "Network transfer overhead for large dependencies", "File system security considerations".

#64 was closed on the strength of docs/function_serialization_technical_design.md, whose final revision (commit 3417b84, "Update technical design document - IMPLEMENTATION COMPLETE") claims:

✅ Phase 2: Dependency Packaging Core (COMPLETE)

• ✅ Create FilePackager for selective file collection
• ✅ Build package deployment system for remote transfer

✅ Phase 4: Integration & Testing (COMPLETE)

• ✅ Integrate with existing ClusterExecutor

That last bullet is false — see §2. The doc was deleted in fb373d2.

Supporting element, issue #20 ("Proposed classes"), the original Job spec:

• stores a function, arguments, and pointers to the appropriate file paths (e.g., data, results, scratch)

The part of the record that cuts the other way — report honestly

Issue #10, comment by @jeremymanning:

Syncing code seems like a great idea-- we want to be able to ensure that the user is running what they think they're running, and managing file transfers would be fantastic. Syncing results or data seems like a bad idea; I'm imagining that could eat up some serious space on a laptop hard drive that might not have enough room.

So the original position was asymmetric: upload was wanted ("managing file transfers would be fantastic"); download of data/results was explicitly rejected on grounds of local disk and transfer time. #64 later widened "code" to "local modules, data files, or custom code" without revisiting #10.

Synthesis this plan adopts: the upload direction was always in scope and was never delivered; the download direction was deliberately out of scope and that objection is still valid. The feature is therefore not symmetric. Outputs come back by explicit request only, never by inference.

---

2. What already exists (file:line)

Transport exists and is wired — but only for the payload:

• clustrix/executor_connections.py:146 upload_file (sftp.put), :156 download_file (sftp.get). No chunking, progress, or resume.
• Callers: executor_schedulers.py:99 (function_data.pkl up), executor_core.py:197 (result.pkl down), executor_scheduler_status.py:473 (error pickle down).

Detection exists and is completely orphaned:

• clustrix/dependency_analysis.py:249 _analyze_file_references finds paths three ways: string-literal first arg to a call in self.file_operations = {"open","read","write","load","dump","save"} (line 140); any read/write/readline/writelines method call, recorded as path="<unknown>"; and any string constant containing a separator and ending in one of 17 hardcoded extensions. add_file_references (line 110) promotes every relative hit into data_files (line 99).
• clustrix/file_packaging.py:319 _add_data_files zips them into data/<relpath>.
• Nothing calls it. grep -rn "FilePackager\|package_function" clustrix/decorator.py clustrix/executor_core.py clustrix/executor_connections.py clustrix/utils.py → empty. The only importers of file_packaging.py are clustrix/__init__.py:39 and tests/test_file_packaging.py. Same class of finding as Delete ~5,100 lines of orphaned modules (13% of the package has zero importers) #122.

The cluster_* API is read-only: clustrix/filesystem.py:567-660 defines exactly cluster_ls/find/stat/exists/isdir/isfile/glob/du/count_files. No cluster_put, no cluster_get. ClusterFilesystem holds an SFTP client (filesystem.py:225) used only for stat-style reads.

Shared-filesystem detection already exists: filesystem.py:113 _auto_detect_cluster_location, requiring same_host and shared_filesystem. Its comment documents a real past bug (a laptop on the VPN judged to be the cluster) — worth preserving as a warning.

Cleanup: config.py:86 cleanup_on_success=True; executor_core.py:214 runs rm -rf {remote_dir}. It knows only about the job directory.

Inbound-byte security: executor_core.py:131 _verify_result_signature reads result.pkl.hmac (line 160); executor_core.py:204 calls it before dill.loads at line 211.

Prior art for out-of-band staging: clustrix/hf_jobs.py:352-375 already stages oversized payloads to a private HF dataset repo, docstring: "HuggingFace rejects very large environment variables, which capped a job's arguments at a few hundred kilobytes -- fine for a function, useless for data."

Gap table

Capability	State
SFTP put/get	Exists, wired
Detect referenced files	Exists, orphaned, heuristic
Package data files	Exists, orphaned
Shared-FS detection	Exists, cluster_* only
Explicit put/get API	Missing
Content-addressed dedup / manifest	Missing
Size policy (refuse/warn/delegate)	Missing
Output return path	Missing
Resume / integrity on partial transfer	Missing
Staged-input lifecycle vs cleanup_on_success	Missing

---

3. Design decisions

Semantics — explicit declaration, never inference. The existing detector's third rule is "any string ending in one of 17 extensions", so automatic staging would upload on a literal like "s3://bucket/notes.log". Silent inferred I/O is the worst failure mode here. API:

@cluster(cores=8,
         inputs=["data/subjects.h5", "config/model.yaml"],
         outputs=["results/*.npz"])
def fit(subject): ...

Inputs are visible to the function at the same relative path they had locally, so the function is unchanged between local and remote runs. Outputs are fetched only on success, only when declared. Primitives underneath: cluster_put(local, remote, config) / cluster_get(remote, local, config) in filesystem.py.

Size — three bands, refuse past the top. stage_warn_bytes (default 100 MB) → silent; below stage_max_bytes (default 5 GB) → transfer with a warning naming file, size, throughput, ETA; at/above → raise, naming the file, the threshold, the config key, and the two remedies (shared storage; stage_backend="rsync"). A refusal beats a silent multi-hour SFTP that looks like a hang. stage_backend ∈ {"sftp","rsync"}, opt-in only — auto-selecting by size makes the transport depend on the data.

Idempotence — content hash with an mtime+size fast path. (size, mtime_ns, inode) hit in the local cache reuses the cached BLAKE2b-256 digest; otherwise stream-hash. Blobs live content-addressed at {remote_work_dir}/_stage/<d[:2]>/<digest>, outside any job dir; job dirs get symlinks (→ hardlink → copy fallback). Remote manifest _stage/manifest.json (flock + atomic rename; digest -> {size, first_seen, last_used, refcount}) is authoritative. Local ~/.clustrix/stage-cache/<host>.json is a pure optimisation — deleting it costs time, never correctness.

Shared filesystems — per-path probe, not per-host. New shared_filesystem_roots config; if realpath(local) is under one and a one-shot remote stat matches size+mtime, stage nothing and point the job at the absolute path. The remote stat is mandatory — filesystem.py:113's comment records what happens when this is answered by names.

Cleanup — three distinct lifetimes. Job dir: cleanup_on_success, unchanged. Staged blobs: survive the job, reclaimed by a submission-time reaper via stage_cache_ttl_days (7) / stage_cache_max_bytes, plus explicit clustrix.clear_stage_cache(config) and a CLI verb. Fetched outputs: never touched by clustrix.

Security — downloaded data files do NOT need an HMAC, and here is why. result.pkl is signed because clustrix itself calls dill.loads on it; the RCE path is created by our deserialization, not by the transfer. A staged output is written to disk and handed over as a path — clustrix never interprets it, and the attacker who could tamper with it already controls the remote account. What we owe instead: (1) integrity — digest every staged input, re-verify on the worker; digest outputs on the worker, verify after fetch; (2) path confinement — output globs realpath-confined to the job dir, escapes rejected not clamped (Zip-Slip class); (3) 0600/0700 modes, matching the care in executor_connections.py::create_remote_file; (4) refuse credential-shaped paths (~/.ssh/*, *.pem, .env) without an explicit override; (5) the manifest is remote-origin data — fixed-schema json.load, never pickle/eval.

Failure modes: upload to <digest>.partial + fsync + size check + atomic rename, never register an unverified digest; quota exceeded → reap, retry once, then raise naming the store and clear_stage_cache (never silently run the job without the file); permission denied → probe writability once per session and fail before any bytes move; file changed mid-upload → re-stat after transfer and raise if (size, mtime_ns) moved; missing declared input → hard failure at submission; missing declared output → warn (or error under require_outputs=True); concurrent stagers → per-digest lock, loser waits for the rename; no symlink support on the worker → hardlink → copy, logged (a copy doubles quota).

Out of scope (this is what keeps it from becoming a workflow engine): no sync/mirror/watch, no DAG or data-derived ordering, no cluster→cluster transfer, no S3/GCS backends in the transfer path, no compression/format conversion, no provenance DB, and no automatic inference of inputs from source in any phase.

---

4. Phases

Verification targets are the backends this project has actually proven: real SLURM, real SSH GPU host, HF Jobs. Evidence lands in docs/evidence/.

Phase 1 — Primitives: cluster_put / cluster_get

Add both to filesystem.py alongside the existing nine, over the existing SFTP client (filesystem.py:225) / shutil locally. Streamed BLAKE2b, atomic .partial+rename, 0600/0700, path confinement, size bands with stage_warn_bytes/stage_max_bytes on ClusterConfig. No manifest, no caching, no decorator changes.

DoD: a file cluster_put to the real SLURM host and the real SSH GPU host, digest verified there by remote command; cluster_get round-trips byte-identical; a file over stage_max_bytes raises with the documented message; a killed transfer leaves only a .partial with nothing registered; unit tests for confinement and mode; evidence committed.

Phase 2 — Content-addressed store, manifest, dedup

_stage/ layout, locked+atomic remote manifest, local (path,size,mtime,inode)→digest cache, skip-if-present, symlink/hardlink/copy materialisation, per-digest lock.

DoD: on real SLURM, staging the same ~1 GB file twice transfers bytes exactly once (measured); deleting the local cache still yields zero re-upload; two concurrent processes produce one copy and one manifest entry; a corrupted manifest is detected and rebuilt from store contents rather than crashing.

Phase 3 — @cluster(inputs=..., outputs=...)

Resolve declarations at submission, stage, materialise at original relative paths, run, collect declared outputs on success, fetch with digest verification. require_outputs.

DoD: an unmodified function opening "data/x.csv" relatively runs identically locally and on real SLURM with inputs=["data/x.csv"]; glob outputs return and verify; a failed job fetches nothing; an output pattern escaping the job dir is rejected on the worker; cleanup_on_success removes the job dir while _stage/ survives — proven by a second run reusing the blob.

Phase 4 — Shared-FS elision and quota safety

shared_filesystem_roots, per-path remote stat probe, reaper (stage_cache_ttl_days, stage_cache_max_bytes), clear_stage_cache, quota retry-once, CLI verb.

DoD: on real SLURM a file already on the site's shared root stages with zero bytes transferred and the job reads the original path; the negative case (same filename, different content, not actually shared) is correctly not elided; a store over stage_cache_max_bytes is reaped below it on next submission with the eviction logged; an induced quota failure produces the documented error, not a hang.

Phase 5 — stage_backend="rsync" (only if 1-4 hit a wall)

rsync -az --partial --info=progress2 over the existing SSH identity, same manifest, same digests, strictly opt-in.

DoD: a multi-GB stage to real SLURM completes via rsync with the same digest the SFTP path would produce, resumes after an induced interruption, and a host without rsync errors clearly rather than falling back.

---

5. Documentation

docs/source/introduction.rst:115 is owned by a separate docs sweep — do not edit it here.

Interim wording:

It does not yet move your data. Clustrix ships your code and arguments, not your dataset. If your function needs a 200 GB file, that file has to already be reachable from the worker today. The filesystem utilities help you inspect and locate remote data, but they are not yet a transfer service for bulk inputs. Staging declared inputs and outputs is planned — see this issue.

After Phases 1-4, it leaves the "what it is not" list and becomes:

It moves the data you declare. @cluster(inputs=[...], outputs=[...]) stages the files your function names, content-addressed and deduplicated, so re-running does not re-upload unchanged inputs and a file already on shared storage is not uploaded at all. It is not a sync tool: nothing moves that you did not name, and nothing comes back that you did not ask for.

The "not a workflow engine" and "not a low-latency dispatcher" bullets stay; the out-of-scope list above is what keeps them true.

---

6. Decision needed before any code is written

Is the stable path contract "the same relative path the file had locally", or "an opaque staged path handed to the function"?

Relative-path is what lets a function run unmodified in both places — the entire value proposition — but it asserts control over the worker's working-directory layout, breaks for absolute-path inputs, and forces the symlink/hardlink/copy machinery in Phase 2. Opaque-path is far simpler and impossible to get wrong, but every user rewrites their function to take a path argument, at which point they may reasonably ask what this bought them over scp.

This plan assumes relative-path. If that is wrong, Phases 2 and 3 both shrink substantially and must be re-scoped first.

Related

• Core Architecture: Function Serialization and Dependency Management for Remote Execution #64 (closed; the conflicting design element)
• Proposed classes #20 (Job holds "pointers to the appropriate file paths (e.g., data, results, scratch)")
• Things we'd like to see in a cluster tools package #10 (the contrary position on syncing data/results)
• Delete ~5,100 lines of orphaned modules (13% of the package has zero importers) #122 (orphaned modules — file_packaging.py is one of them; this issue would give it a purpose or replace it)

[jeremymanning]

We could potentially use huggingface buckets and/or the streaming architecture to facilitate this and solve the "asymmetry" issue.

I'm imagining a multi-step process:

1. Prepare a package: configure HF credentials using the cluster config machinery and then provide a function for feeding in data or file paths and returning a new object that can be passed to cluster-decorated functions (either pass one, or pass a list of such objects). the objects should include a local path (where the file(s) exist locally) along with a private remote path linking to the HF bucket/path with the dataset(s). OR if the dataset is small (say, less than some default (configurable) threshold, or if the data package machinery is called with a force_local flag set to True) just package the dataset inside of that same object.
2. then, when the cluster command is called, the object needs to be sent (via rcp, scp, or similar) to the configured cluster, and then referenced as needed to access the required data on demand.

re: streaming: i'm not 100% sure how to leverage this properly. if there is a way we can systematically figure out which part(s) of the dataset are needed at a given time, then this would be incredibly powerful because we could (for example) run an analysis on a machine that couldn't actually fit the full dataset-- along the lines of how HF streaming enables inference on models when the machine executing the inference can't fit the full dataset. but generalizing this may be more complicated. perhaps it's possible by doing something like running a mock simulation that wraps every function recursively to either execute the function (if it takes less than some very small threshold amount of time) OR simply track which parts of the dataset are accessed inside that piece. i'm imagining this would use threading and timers in some way, and then interrupt function calls if they took too long. and then we'd have to do some sort of fancy bookkeeping to tag different parts of the execution and figure out which parts of the dataset go with which tags, and then organize the to-be-streamed dataset so that parts could be copied over on demand. unless i'm missing an obvious solution, this is likely more complex than should be attempted in this initial "support data sharing" implementation. if so, we should open a new issue to address the streaming functionality separately, and defer this part of this issue accordingly.

[jeremymanning]

Requirement added: deletion through the wrapped object

From the repository owner, recorded here so it survives the session it arrived in:

"there should also be a way to delete data (clean it up) via the wrapped object"

The data package owns a lifecycle, and the handle that created it has to be able to end it. Points that need deciding rather than bolting a delete() on the end:

• What it deletes, precisely. The remote bucket object, yes. Anything already staged onto the cluster, probably. The user's own local source files, never — deleting someone's data because they cleaned up a transfer is not a defensible behaviour.
• Idempotence. Deleting twice is not an error; deleting something already gone is not an error. But deleting something that exists and failing must raise. Silent failure paths are the defect class this project has spent Pre-push hook cannot block a push: run_real_world_tests.py exits 0 on failure #147, Security: unauthenticated pickle of remote results (RCE) and disabled SSH host key verification #121 and Command injection in filesystem.py: nine shell commands interpolate user paths, zero shlex.quote #154 unwinding.
• Shared blobs. If two packages reference the same content-addressed blob, deleting one must not pull the data out from under the other — refcount it, or scope each package's remote path so they cannot collide.
• Partial state. A package whose upload was interrupted still owns a .partial or a half-registered manifest entry. Deletion has to reclaim that too, or the failure path leaks storage in the owner's HF account indefinitely.
• Discoverability. A user who has lost the object still needs a way out, because the data is sitting in their bucket costing them storage. The object must not be the only key to its own deletion — there should be a way to list what clustrix has put there, and probably a module-level sweep.
• Context managers. If the object supports with, whether __exit__ deletes is a deliberate choice, not an incidental one. Defaulting to not deleting seems right (a package is usually reused across runs), with an explicit ephemeral=True-style construction for the other case — but the call site must make it obvious which is happening.

Verification, on this project's usual terms: create a package with a few KB, confirm it exists by asking the remote, delete it, confirm it is gone by asking again. Not by trusting a return value.

[jeremymanning]

Simplification from the owner: no automatic cleanup, no refcounting, pickle is the persistence story

"the deletion is handled from the data object, which is created outside of the cluster call. data is never automatically cleaned up. so the collision case isn't necessary to handle. if the dataset needs to remain, just pickle the data object, save locally, and load it later as needed. only clean it up if and only if it's not needed any more -- which is a determination to be made by the user, not automatically."

This overrides three things above, including one I raised in the previous comment.

Refcounting is out of scope. I flagged the shared-blob case; it is ruled out. A data object owns its remote data. No manifest refcount, and no machinery to detect two objects pointing at the same blob.

Nothing is ever cleaned up automatically. That kills the Phase 4 reaper outright — no stage_cache_ttl_days, no stage_cache_max_bytes, no submission-time eviction, no deletion when a job finishes. cleanup_on_success continues to govern the job directory and must not reach staged data. It also settles the context-manager question from the previous comment in the negative: __exit__ does not delete, and the ephemeral=True mode I floated is contrary to the model and will not be built.

Persistence is pickle. The documented way to keep a dataset across sessions is to pickle the data object, save it locally, and load it when needed. Two consequences for the implementation:

• The object must survive a pickle round trip into a fresh interpreter and still both resolve its data for a @cluster call and delete its remote data.
• Therefore no live HF client, socket, SFTP channel or file handle may be pickled into it — carry the remote path, digest, size and metadata, and re-establish the connection lazily. Credentials must not be pickled in either: a saved package should not be a token on disk, and should re-authenticate from the ordinary configuration path on load.

Deletion stays explicit and stays loud. Never touches local source files. Idempotent when the data is already gone. Raises when a deletion that should have succeeded did not.

One thing this makes more important rather than less: with automatic reclamation off the table, a user who loses the object has nothing that will ever clean up after them. A way to list what clustrix has placed in their bucket is part of the feature, not a convenience.

[jeremymanning]

Round one implemented, and two requirements that arrived in chat (recording them so they are not lost)

Implemented in clustrix/staging.py per @jeremymanning's comment above (HF buckets, not SFTP
staging into _stage/). Streaming deferred to #155. Full write-up in
notes/data-package-implementation.md.

Two requirements arrived by chat during implementation and are not in the issue body. Verbatim:

there should also be a way to delete data (clean it up) via the wrapped object

the deletion is handled from the data object, which is created outside of the cluster call.
data is never automatically cleaned up. so the collision case isn't necessary to handle. if the
dataset needs to remain, just pickle the data object, save locally, and load it later as needed.
only clean it up if and only if it's not needed any more -- which is a determination to be made
by the user, not automatically.

Consequences, so the next session does not re-derive the old plan from §3/§4 above:

• The Phase 4 reaper is dead. No TTL, no stage_cache_ttl_days, no stage_cache_max_bytes,
  no eviction at submission, no deletion on job completion, no deletion in __exit__.
  cleanup_on_success still governs the job directory only, and must not touch staged data.
• No refcounting and no content-addressed dedup. Each package gets its own remote folder keyed
  by a fresh uuid, so two packages can never share a blob. Cost: identical content packaged twice
  is stored twice.
• The pickled object is the persistence story. DataPackage carries only plain data — no HF
  client, no socket, no credential — so pickle.dump it, load it in a fresh interpreter later, and
  it still resolves its data and still deletes it. Tested in a real subprocess.
• Deletion is explicit, idempotent, and honest. pkg.delete() removes the remote folder (which
  also disposes of a half-finished upload) and the copy clustrix wrote into its own cache. It never
  touches the files you packaged. Already gone is not an error; present-and-will-not-delete raises.
  list_data_packages() / delete_data_package(id) are the way out for a user who lost the object.

Also, §6's open question ("relative path or opaque path?") is sidestepped rather than answered:
paths are relative inside the package, so pkg.path("data/x.csv") and pkg.materialize()
preserve the layout without clustrix asserting control over the worker's working directory.

file_packaging.py was not adopted: _add_data_files consumes dependency_analysis.py's
heuristic data_files, which is exactly the source-inference this design rejects. It stays orphaned
and stays #122's problem.

[jeremymanning]

Tracked as a sub-issue of #160 (deferred). Consolidated after a three-agent audit that verified every open issue against the code; see #160 for the plan and the ordering.