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

Inizia da clustrix/filesystem.py, in particolare dal client SFTP ClusterFilesystem esistente intorno alla riga 225 e dai metodi cluster_* alle righe 567-660. Implementa e verifica prima le primitive di Phase 1, quindi esegui gli unit test pertinenti per confinamento, modalità, limiti di dimensione e trasferimenti interrotti. Il completamento include round trip reali SLURM e SSH con host GPU, verifica dei byte e prove in docs/evidence/.

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 #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

• #64 (closed; the conflicting design element)
• #20 (Job holds "pointers to the appropriate file paths (e.g., data, results, scratch)")
• #10 (the contrary position on syncing data/results)
• #122 (orphaned modules — file_packaging.py is one of them; this issue would give it a purpose or replace it)