feat: resumable network volume upload and download

docs/cli/references/command_line_interface.md

@@ -23,6 +23,10 @@ rp ssh POD_ID       # open a terminal on a pod
 rp pod list
 rp pod create
 rp pod connect POD_ID
+
+# Network volume files
+rp volume upload VOLUME ./dataset datasets/dataset
+rp volume download VOLUME outputs/ ./outputs
 ```
 
 ## Overview
@@ -59,3 +63,38 @@ entrypoint or a resource function; interrupt discovery if an import blocks.
 On Ctrl-C, a dev session waits for its running local entrypoint to finish before
 deleting endpoints. Blocking entrypoint code can delay shutdown. Cleanup failures
 exit with an error and report the endpoint IDs that still need deletion.
+
+### Network volume files
+
+```bash
+rp volume upload VOLUME ./model.safetensors models/
+rp volume upload VOLUME ./dataset datasets/dataset
+rp volume download VOLUME models/model.safetensors
+rp volume download VOLUME outputs/ ./outputs
+```
+
+Transfers use the volume's S3-compatible API, so they need S3 API keys
+(separate from the API key): set `RUNPOD_S3_ACCESS_KEY` and
+`RUNPOD_S3_SECRET_KEY`, pass `--s3-access-key` / `--s3-secret-key`, or save
+`s3_access_key` / `s3_secret_key` in `~/.runpod/config.toml`. The CLI prompts
+for missing keys and offers to save them. You can omit `VOLUME` in an
+interactive session to pick a volume from a list. The
+volume must be in a [datacenter with the S3 API](https://docs.runpod.io/storage/s3-api).
+`VOLUME` is the volume's ID or name, looked up with your API key.
+
+To resume an interrupted transfer, run the same command again. Files that
+are unchanged (same size, destination not older) are skipped, a large upload
+reuses the parts it already sent, and a large download keeps its finished
+chunks in `<name>.rpdownload`. `--force` transfers every file. Parts of an
+upload that is never resumed stay in the volume's hidden `.s3compat_uploads`
+folder and count toward its size.
+
+The same transfers are available from Python:
+
+```python
+from runpod import NetworkVolume
+
+volume = NetworkVolume("my-volume")
+volume.upload("./dataset", "datasets/dataset")
+volume.download("outputs/", "./outputs")
+```

runpod/_volume_transfer_utils.py

@@ -0,0 +1,160 @@
+"""Filesystem and checksum helpers for resumable transfers.
+
+This private module keeps disk state and path safety details out of the S3
+request orchestration. The public error type remains in network_volume_transfer.
+"""
+
+import hashlib
+import json
+import os
+import unicodedata
+from pathlib import Path, PurePosixPath
+from typing import Any, Callable, Dict, List, Optional, Tuple
+
+MIB = 1024 * 1024
+
+
+def _read_json(path: Path) -> Optional[Dict[str, Any]]:
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return None
+    return data if isinstance(data, dict) else None
+
+
+def _write_json(path: Path, data: Dict[str, Any], sync_dir: Callable[[Path], None]) -> None:
+    """replace path atomically, synced, so a crash leaves the old state or
+    the new one."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_name(path.name + ".tmp")
+    with open(tmp, "w", encoding="utf-8") as handle:
+        json.dump(data, handle)
+        handle.flush()
+        os.fsync(handle.fileno())
+    os.replace(tmp, path)
+    sync_dir(path.parent)
+
+
+def _sync_dir(path: Path) -> None:
+    """make a rename in path survive a power loss. best effort: windows
+    cannot open a directory."""
+    try:
+        fd = os.open(path, os.O_RDONLY)
+    except OSError:
+        return
+    try:
+        os.fsync(fd)
+    except OSError:
+        pass
+    finally:
+        os.close(fd)
+
+
+def _md5_hex(path: Path, offset: int, length: int) -> str:
+    digest = hashlib.md5()  # noqa: S324 - etag comparison, not security
+    with open(path, "rb") as handle:
+        handle.seek(offset)
+        while length > 0 and (block := handle.read(min(8 * MIB, length))):
+            digest.update(block)
+            length -= len(block)
+    return digest.hexdigest()
+
+
+def _etag(value: Any) -> str:
+    return str(value or "").strip('"')
+
+
+def _is_md5(etag: str) -> bool:
+    return len(etag) == 32 and all(c in "0123456789abcdef" for c in etag.lower())
+
+
+def _multipart_etag(parts: List[Dict[str, Any]]) -> str:
+    """the etag s3 (and runpod, verified) gives an object assembled from
+    parts: md5 of the parts' binary md5s, then "-<count>". "" when a part
+    etag is not an md5."""
+    etags = [_etag(part["ETag"]) for part in parts]
+    if not all(_is_md5(etag) for etag in etags):
+        return ""
+    digest = hashlib.md5(b"".join(bytes.fromhex(etag) for etag in etags))  # noqa: S324
+    return f"{digest.hexdigest()}-{len(parts)}"
+
+
+def _same_version(a: os.stat_result, b: os.stat_result) -> bool:
+    return a.st_size == b.st_size and a.st_mtime_ns == b.st_mtime_ns
+
+
+def _not_older(a: float, b: float) -> bool:
+    """a >= b in whole seconds; s3 dates carry no finer precision."""
+    return int(a) >= int(b)
+
+
+def _check_remote_path(remote_path: Optional[str]) -> None:
+    if remote_path and ".." in remote_path.replace("\\", "/").split("/"):
+        raise ValueError(f"remote path {remote_path!r} must not contain '..'")
+
+
+def _check_local_names(
+    targets: List[Path], root: Path, suffixes: Tuple[str, ...], error_type: type[Exception]
+) -> None:
+    """refuse targets that would share a path on disk: two that are one name
+    on a case-insensitive or normalizing filesystem ("Model.bin" and
+    "model.bin"), a file whose name is also a folder another file lives in
+    ("a" beside "A/b"), or a file whose partial, sidecar or sidecar temp name
+    is also a target or a folder one lives in ("x" beside "x.rpdownload/y").
+    names are compared nfc-normalized (apfs ignores normalization) and case
+    folded, which can refuse a listing a case-sensitive filesystem would
+    take: the safe direction."""
+
+    def fold(path: PurePosixPath) -> str:
+        return unicodedata.normalize("NFC", str(path)).casefold()
+
+    rels = [PurePosixPath(target.relative_to(root).as_posix()) for target in targets]
+    files: Dict[str, PurePosixPath] = {}
+    dirs: Dict[str, PurePosixPath] = {}
+    for rel in rels:
+        if fold(rel) in files:
+            raise error_type(
+                f"cannot download both {files[fold(rel)]} and {rel}: they would be one "
+                "file on a case-insensitive or normalizing filesystem"
+            )
+        files[fold(rel)] = rel
+        dirs.update({fold(parent): rel for parent in rel.parents if str(parent) != "."})
+    for rel in rels:
+        if fold(rel) in dirs:
+            raise error_type(
+                f"cannot download both {rel} and {dirs[fold(rel)]}: "
+                "the first would have to be a file and a folder"
+            )
+        for suffix in suffixes:
+            twin = fold(rel.with_name(rel.name + suffix))
+            other = files.get(twin) or dirs.get(twin)
+            if other is not None:
+                raise error_type(
+                    f"cannot download both {rel} and {other}: "
+                    f"{rel.name + suffix} is used for the first's partial download"
+                )
+
+
+def _safe_target(root: Path, rel: str, error_type: type[Exception]) -> Path:
+    """local path for a remote relative key, refusing anything outside root.
+    "." and empty segments are refused too: "a/./b" would alias "a/b"."""
+    segments = rel.split("/")
+    if (
+        rel.startswith("/")
+        or "\\" in rel
+        or any(seg in ("..", ".", "") for seg in segments)
+        or ":" in segments[0]
+    ):
+        raise error_type(f"refusing unsafe remote path: {rel!r}")
+    target = root.joinpath(*segments)
+    if root.resolve() not in target.resolve().parents:
+        raise error_type(f"refusing unsafe remote path: {rel!r}")
+    # a symlinked folder would let two remote paths share one local file,
+    # which the name collision check cannot see
+    for depth in range(1, len(segments)):
+        if root.joinpath(*segments[:depth]).is_symlink():
+            raise error_type(
+                f"{root.joinpath(*segments[:depth])} is a symlink; "
+                "download into a folder without symlinked subfolders"
+            )
+    return target

runpod/apps/volume.py

@@ -84,6 +84,75 @@ def name(self) -> str:
     def __repr__(self) -> str:
         return f"<NetworkVolume {self.name!r} size={self.size}GB>"
 
+    def upload(
+        self,
+        local_path,
+        remote_path: Optional[str] = None,
+        *,
+        access_key: Optional[str] = None,
+        secret_key: Optional[str] = None,
+        concurrency: int = 4,
+        progress=None,
+        force: bool = False,
+        endpoint_url: Optional[str] = None,
+    ) -> Dict[str, int]:
+        """upload a file or folder to this volume through its s3 api.
+
+        the volume must exist; it is found by id or name with the runpod
+        api key, which also gives its datacenter. `remote_path` defaults to
+        the local basename. a folder's contents go under `remote_path/`. for
+        a file, a `remote_path` ending in "/" keeps the local file name.
+        unchanged files (same size, remote not older, in whole seconds) are
+        skipped unless `force`, and an interrupted multipart upload resumes
+        on re-run.
+
+        s3 keys come from the arguments, RUNPOD_S3_ACCESS_KEY and
+        RUNPOD_S3_SECRET_KEY, or the saved s3_access_key/s3_secret_key in
+        ~/.runpod/config.toml. the CLI prompts for missing keys and can pick
+        a volume from a list when VOLUME is omitted.
+        `progress(event, path, nbytes)` receives
+        "start", "advance", "reuse", "skip", and "done" events.
+
+        returns {files_transferred, files_skipped, bytes_transferred}.
+        """
+        from runpod import network_volume_transfer as nvt
+
+        transfer = nvt._make_transfer(
+            self.reference, access_key, secret_key, endpoint_url, concurrency, progress, force
+        )
+        return transfer.upload(Path(local_path), remote_path)
+
+    def download(
+        self,
+        remote_path: str,
+        local_path=None,
+        *,
+        access_key: Optional[str] = None,
+        secret_key: Optional[str] = None,
+        concurrency: int = 4,
+        progress=None,
+        force: bool = False,
+        endpoint_url: Optional[str] = None,
+    ) -> Dict[str, int]:
+        """download a file or folder from this volume through its s3 api.
+
+        a `remote_path` ending in "/" (or one that only exists as a prefix)
+        is a folder. `local_path` defaults to the remote basename; one
+        ending in "/" is a folder to download into. local files of the same
+        size and not older than the remote copy are skipped (downloads stamp
+        the remote time on the file) unless `force`, and partial downloads
+        resume on re-run. keys and `progress` work as in `upload`.
+
+        returns {files_transferred, files_skipped, bytes_transferred}.
+        """
+        from runpod import network_volume_transfer as nvt
+
+        transfer = nvt._make_transfer(
+            self.reference, access_key, secret_key, endpoint_url, concurrency, progress, force
+        )
+        # passed through as given: Path() would drop a trailing "/"
+        return transfer.download(remote_path, local_path)
+
 
 class GlobalVolume(Volume):
     """a global volume resolved by name or id without datacenter placement."""

runpod/cli/groups/config/functions.py

@@ -29,29 +29,57 @@ def set_credentials(api_key: str, profile: str = "default", overwrite=False) ->
     [default]
     api_key = "RUNPOD_API_KEY"
     """
-    cred_dir = os.path.dirname(CREDENTIAL_FILE)
-    os.makedirs(cred_dir, exist_ok=True)
-    Path(CREDENTIAL_FILE).touch(exist_ok=True)
+    config = _load_config()
+    if not overwrite and profile in config:
+        raise ValueError(
+            "Profile already exists. Use set_credentials(overwrite=True) to update."
+        )
+    _set_profile_values(config, profile, {"api_key": api_key})
+    _write_config(config)
+
+
+def set_s3_keys(access_key: str, secret_key: str, profile: str = "default") -> None:
+    """
+    Saves S3 API keys for network volume transfers in ~/.runpod/config.toml,
+    next to the profile's api_key.
+
+    --- File Structure ---
+
+    [default]
+    api_key = "RUNPOD_API_KEY"
+    s3_access_key = "..."
+    s3_secret_key = "..."
+    """
+    config = _load_config()
+    _set_profile_values(
+        config, profile, {"s3_access_key": access_key, "s3_secret_key": secret_key}
+    )
+    _write_config(config)
+
 
+def _load_config() -> tomlkit.TOMLDocument:
+    """the config file as an editable document, created empty if missing."""
+    os.makedirs(os.path.dirname(CREDENTIAL_FILE), exist_ok=True)
+    Path(CREDENTIAL_FILE).touch(exist_ok=True)
     with open(CREDENTIAL_FILE, "r", encoding="UTF-8") as cred_file:
-        try:
-            content = cred_file.read()
-            config = (
-                tomlkit.parse(content)
-                if content.strip()
-                else tomlkit.document()
-            )
-        except tomlkit.exceptions.ParseError as exc:
-            raise ValueError("~/.runpod/config.toml is not a valid TOML file.") from exc
+        content = cred_file.read()
+    try:
+        return tomlkit.parse(content) if content.strip() else tomlkit.document()
+    except tomlkit.exceptions.ParseError as exc:
+        raise ValueError("~/.runpod/config.toml is not a valid TOML file.") from exc
 
-    if not overwrite:
-        if profile in config:
-            raise ValueError(
-                "Profile already exists. Use set_credentials(overwrite=True) to update."
-            )
 
-    config[profile] = {"api_key": api_key}
+def _set_profile_values(config: tomlkit.TOMLDocument, profile: str, values: dict) -> None:
+    """set values in a profile, keeping its other keys."""
+    if profile in config:
+        config[profile].update(values)
+    else:
+        config[profile] = values
 
+
+def _write_config(config: tomlkit.TOMLDocument) -> None:
+    """replace the config file atomically; mkstemp keeps it owner-only."""
+    cred_dir = os.path.dirname(CREDENTIAL_FILE)
     fd, tmp_path = tempfile.mkstemp(dir=cred_dir, suffix=".toml")
     try:
         with os.fdopen(fd, "w", encoding="UTF-8") as tmp_file: