fix(cli): report upload progress in bytes transferred (#1171)

* fix(cli): report upload progress in bytes transferred

`fal files upload` sized its progress task in parts, so a 100 MB file - ten
10 MB parts uploaded by ten concurrent workers - had nothing to advance until
those parts all landed at the end. The bar sat near 0% for the whole transfer
and then jumped to done. A full-file MD5 pass ran before the first byte went
out, adding a second phase with no advancement at all.

Progress is now sized in bytes and advanced as each part body is handed to the
transport, and the digest is folded into the reader thread the upload already
runs, so verifying the etag costs no extra read and no silent wait. The small
(non-multipart) branch advances the same way while its request body is written.

`BaseMultipartUpload.upload_file` gains an `on_bytes_uploaded` callback and a
`content_md5` property; `on_part_complete` and existing callers are unchanged.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(cli): make upload integrity and retry safety explicit

Three follow-ups to the byte-progress change.

Retry safety no longer rests on the HTTP client rewinding a consumed body.
`_request` takes an optional `files_factory` and rebuilds the payload on every
attempt, so a part that fails and retries resends its bytes by construction
rather than because httpx happens to seek the stream back to zero. The raw
`bytes` body used when no progress callback is supplied is unchanged.

The part count is fixed from the size sampled before the read, so a file that
shrank underneath the reader would upload a prefix and report success. The
reader now counts the bytes it consumed and the upload fails if that does not
match the sampled size. This restores the end-to-end guarantee that was lost
when the digest stopped being computed from an independent pass over the file.

Computing the digest is now opt-in via `compute_md5`. Only the `/data` path
verifies the etag; app file sync was paying for a hash over every uploaded
file and discarding it.

Each of the three is covered by a test that fails when the change is reverted.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(cli): settle the multipart bar on the uploaded size

Part callbacks compute their running total under the tracker lock but are
invoked outside it, so the last callback to arrive is not necessarily the
highest. The multipart branch took whatever the final callback reported, unlike
the small-file branch which already writes the full size at the end, so a
successful upload could leave the bar short of 100%.

Found by an end-to-end sweep against live storage: a backwards delivery is rare
(one in ~255,000 updates) but nothing made the final value correct, and no test
covered it -- the monotonicity tests run at concurrency 1 and the concurrent one
asserts only the maximum, never the last value.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(cli): show a bracketed filename verbatim in the progress bar

A Rich task description is parsed as markup, so any bracketed span that looks
like a tag is dropped from the label: `x[bold]y.bin` rendered as `xy.bin`, and
`x[not a tag]y.bin` as `xy.bin`. Both filenames are legal. Escaping the
basename keeps the label matching what is on disk.

Pre-existing on both progress paths, not introduced by the byte-progress
change. A closing tag is not reachable here -- it needs `/`, which is the path
separator, so os.path.basename removes everything before it -- which means this
mangles the display and cannot raise.

Also corrects the digest comment: the byte-count check catches an edit that
changes the file's length, but an edit that preserves it is not detected, since
the digest is taken from the same bytes that were sent.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(cli): satisfy mypy on the per-attempt body factory

`on_progress` is Optional, and narrowing it with `if on_progress is None`
does not carry into the nested `build_files`, since a closure could run after
the name was rebound. Bind the narrowed callback to a non-Optional local that
the closure captures instead.

Caught by the pre-commit mypy hook (v1.3.0, --check-untyped-defs), which I had
not run on the earlier commit.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: jim-fal

projects/fal/src/fal/files.py

@@ -1,4 +1,3 @@
-import hashlib
 import os
 import posixpath
 from functools import cached_property
@@ -12,20 +11,13 @@
     MULTIPART_MAX_CONCURRENCY,
     MULTIPART_THRESHOLD,
     DataFileMultipartUpload,
+    ProgressFileReader,
 )
 
 if TYPE_CHECKING:
     import httpx
 
 
-def _compute_md5(lpath, chunk_size=8192):
-    hasher = hashlib.md5()
-    with open(lpath, "rb") as fobj:
-        for chunk in iter(lambda: fobj.read(chunk_size), b""):
-            hasher.update(chunk)
-    return hasher.hexdigest()
-
-
 class FalFileSystem(AbstractFileSystem):
     def __init__(
         self,
@@ -153,13 +145,15 @@ def get_file(self, rpath, lpath, **kwargs):
             fobj.write(response.content)
 
     def _put_file_multipart(self, lpath, rpath, size, progress):
-        num_parts = max(1, (size + MULTIPART_CHUNK_SIZE - 1) // MULTIPART_CHUNK_SIZE)
-        task = progress.add_task("Calculating checksum...", total=num_parts)
-        md5 = _compute_md5(lpath)
-        progress.update(task, description=f"Uploading {os.path.basename(lpath)}")
+        from rich.markup import escape
 
-        def on_part_complete(part_number: int):
-            progress.advance(task)
+        # A task description is parsed as Rich markup, so a name containing
+        # tag-like brackets would be mangled or raise.
+        name = escape(os.path.basename(lpath))
+        task = progress.add_task(f"Uploading {name}", total=size)
+
+        def on_bytes_uploaded(uploaded: int):
+            progress.update(task, completed=uploaded)
 
         multipart = DataFileMultipartUpload(
             client=self._client,
@@ -168,15 +162,34 @@ def on_part_complete(part_number: int):
             max_concurrency=MULTIPART_MAX_CONCURRENCY,
         )
 
-        etag = multipart.upload_file(lpath, on_part_complete=on_part_complete)
+        etag = multipart.upload_file(
+            lpath, on_bytes_uploaded=on_bytes_uploaded, compute_md5=True
+        )
 
+        # Concurrent parts report their running totals outside the tracker lock,
+        # so the last callback to arrive is not necessarily the highest. Settle
+        # the bar on the size that was actually uploaded.
+        progress.update(task, completed=size)
+
+        # The digest is taken from the bytes that were read and sent, so this
+        # compares the stored object against the upload, not against the file on
+        # disk. An edit that changes the file's length fails the byte-count check
+        # in upload_file; one that keeps it identical is not detected here.
+        md5 = multipart.content_md5
         if etag and etag != md5:
             raise RuntimeError(
                 f"MD5 mismatch on {rpath}: {etag} != {md5}, please contact support"
             )
 
     def put_file(self, lpath, rpath, mode="overwrite", **kwargs):
-        from rich.progress import BarColumn, Progress, SpinnerColumn, TextColumn
+        from rich.markup import escape
+        from rich.progress import (
+            BarColumn,
+            DownloadColumn,
+            Progress,
+            SpinnerColumn,
+            TextColumn,
+        )
 
         if os.path.isdir(lpath):
             return
@@ -189,18 +202,25 @@ def put_file(self, lpath, rpath, mode="overwrite", **kwargs):
             TextColumn("[progress.description]{task.description}"),
             BarColumn(),
             TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
+            DownloadColumn(),
         ) as progress:
             if size > MULTIPART_THRESHOLD:
                 self._put_file_multipart(lpath, abs_rpath, size, progress)
             else:
-                task = progress.add_task(f"{os.path.basename(lpath)}", total=1)
+                # A zero total renders as an indeterminate pulse forever.
+                total = size or 1
+                task = progress.add_task(escape(os.path.basename(lpath)), total=total)
                 with open(lpath, "rb") as fobj:
+                    reader = ProgressFileReader(
+                        fobj,
+                        lambda uploaded: progress.update(task, completed=uploaded),
+                    )
                     self._request(
                         "POST",
                         f"/files/file/local/{abs_rpath}",
-                        files={"file_upload": (posixpath.basename(lpath), fobj)},
+                        files={"file_upload": (posixpath.basename(lpath), reader)},
                     )
-                progress.advance(task)
+                progress.update(task, completed=total)
         self.dircache.clear()
 
     def put_file_from_url(self, url, rpath, mode="overwrite", **kwargs):

projects/fal/src/fal/upload.py

@@ -1,11 +1,13 @@
 import concurrent.futures
+import hashlib
+import io
 import logging
 import math
 import os
 import queue
 import time
 from threading import Lock, Thread
-from typing import Any, Callable, Dict, List, Optional, Tuple, cast
+from typing import Any, BinaryIO, Callable, Dict, List, Optional, Tuple, cast
 
 import httpx
 
@@ -18,6 +20,55 @@
 MULTIPART_THRESHOLD = 10 * 1024 * 1024  # 10MB
 
 
+class ProgressFileReader:
+    """Read-through view of a binary stream that reports the absolute offset.
+
+    Reporting the offset rather than a delta keeps the count honest across a
+    retry, which rewinds the body and resends it from the start.
+    """
+
+    def __init__(
+        self,
+        fobj: BinaryIO,
+        on_progress: Callable[[int], None],
+    ) -> None:
+        self._fobj = fobj
+        self._on_progress = on_progress
+
+    def read(self, size: int = -1) -> bytes:
+        chunk = self._fobj.read(size)
+        self._on_progress(self._fobj.tell())
+        return chunk
+
+    def seek(self, offset: int, whence: int = os.SEEK_SET) -> int:
+        return self._fobj.seek(offset, whence)
+
+    def tell(self) -> int:
+        return self._fobj.tell()
+
+
+class _BytesUploadedTracker:
+    """Folds concurrent per-part offsets into one cumulative byte count.
+
+    The total is recomputed from the latest offset of every part rather than
+    accumulated, so a rewound part cannot inflate it.
+    """
+
+    def __init__(self, on_bytes_uploaded: Callable[[int], None]) -> None:
+        self._on_bytes_uploaded = on_bytes_uploaded
+        self._offsets: Dict[int, int] = {}
+        self._lock = Lock()
+
+    def for_part(self, part_number: int) -> Callable[[int], None]:
+        def report(offset: int) -> None:
+            with self._lock:
+                self._offsets[part_number] = offset
+                total = sum(self._offsets.values())
+            self._on_bytes_uploaded(total)
+
+        return report
+
+
 class BaseMultipartUpload:
     def __init__(
         self,
@@ -31,6 +82,16 @@ def __init__(
         self._upload_id: Optional[str] = None
         self._parts: List[Dict[str, object]] = []
         self._parts_lock = Lock()
+        self._content_md5: Optional[str] = None
+
+    @property
+    def content_md5(self) -> Optional[str]:
+        """MD5 of the bytes sent by the last successful `upload_file` call.
+
+        `None` until a call has read the file through, so callers must treat a
+        missing digest as "not verified" rather than "verified".
+        """
+        return self._content_md5
 
     @property
     def upload_id(self) -> str:
@@ -65,11 +126,16 @@ def _request(
         method: str,
         path: str,
         max_retries: int = 3,
+        files_factory: Optional[Callable[[], Dict[str, Any]]] = None,
         **kwargs,
     ) -> httpx.Response:
         last_exception = None
 
         for attempt in range(max_retries):
+            # A single-use body must be rebuilt per attempt: a stream left at
+            # EOF by a failed attempt would otherwise resend nothing.
+            if files_factory is not None:
+                kwargs["files"] = files_factory()
             try:
                 response = self.client.request(method, path, **kwargs)
 
@@ -139,14 +205,33 @@ def initiate(self) -> str:
         return self.upload_id
 
     def _upload_part(
-        self, part_number: int, data: bytes, filename: str = ""
+        self,
+        part_number: int,
+        data: bytes,
+        filename: str = "",
+        on_progress: Optional[Callable[[int], None]] = None,
     ) -> Dict[str, object]:
         file_name = filename or "chunk"
-        response = self._request(
-            "PUT",
-            f"{self.part_url}/{part_number}",
-            files={"file_upload": (file_name, data, "application/octet-stream")},
-        )
+        if on_progress is None:
+            response = self._request(
+                "PUT",
+                f"{self.part_url}/{part_number}",
+                files={"file_upload": (file_name, data, "application/octet-stream")},
+            )
+        else:
+            # Bound outside the closure: narrowing does not carry into a nested
+            # function, which could be called after the name was rebound.
+            report: Callable[[int], None] = on_progress
+
+            def build_files() -> Dict[str, Any]:
+                reader = ProgressFileReader(io.BytesIO(data), report)
+                return {"file_upload": (file_name, reader, "application/octet-stream")}
+
+            response = self._request(
+                "PUT",
+                f"{self.part_url}/{part_number}",
+                files_factory=build_files,
+            )
         result = response.json()
         part_info = {
             "part_number": result["part_number"],
@@ -177,9 +262,25 @@ def upload_file(
         self,
         file_path: str,
         on_part_complete: Optional[Callable[[int], None]] = None,
+        on_bytes_uploaded: Optional[Callable[[int], None]] = None,
+        compute_md5: bool = False,
     ) -> str:
+        """Upload `file_path` and return the server etag.
+
+        `on_part_complete` fires once per finished part. `on_bytes_uploaded`
+        fires as the body of each part is handed to the transport, with the
+        running total of payload bytes sent across all parts. `compute_md5`
+        populates `content_md5`; it costs a hash over the whole file, so it is
+        opt-in for callers that verify the etag.
+        """
         size = os.path.getsize(file_path)
 
+        tracker = (
+            _BytesUploadedTracker(on_bytes_uploaded)
+            if on_bytes_uploaded is not None
+            else None
+        )
+
         # Handle empty files specially - upload single empty part
         if size == 0:
             try:
@@ -191,6 +292,10 @@ def upload_file(
                 self._upload_part(1, b"")
                 if on_part_complete:
                     on_part_complete(1)
+                if on_bytes_uploaded:
+                    on_bytes_uploaded(0)
+                if compute_md5:
+                    self._content_md5 = hashlib.md5(b"").hexdigest()
                 return self.complete()
             except FileExistsError:
                 return ""
@@ -209,6 +314,8 @@ def upload_file(
             maxsize=self.max_concurrency * 2
         )
         read_error: List[Exception] = []
+        hasher = hashlib.md5() if compute_md5 else None
+        bytes_read: List[int] = [0]
 
         def reader_thread():
             """Reads file chunks and puts them in bounded queue"""
@@ -217,6 +324,11 @@ def reader_thread():
                     for part_number in range(1, num_parts + 1):
                         chunk = f.read(self.chunk_size)
                         if chunk:
+                            # Hashing here rides along with the read the upload
+                            # already needs, instead of a second full-file pass.
+                            if hasher is not None:
+                                hasher.update(chunk)
+                            bytes_read[0] += len(chunk)
                             chunk_queue.put((part_number, chunk))
                 # Sentinel to signal completion
                 chunk_queue.put(None)
@@ -243,6 +355,9 @@ def reader_thread():
                         self._upload_part,
                         part_number,
                         chunk,
+                        on_progress=(
+                            tracker.for_part(part_number) if tracker else None
+                        ),
                     )
                     futures.append((part_number, future))
 
@@ -257,6 +372,17 @@ def reader_thread():
             if read_error:
                 raise read_error[0]
 
+            # The part count is fixed from the size sampled before the read, so
+            # a file that changes underneath us would otherwise upload a prefix
+            # and report success.
+            if bytes_read[0] != size:
+                raise RuntimeError(
+                    f"{file_path} changed while uploading: read "
+                    f"{bytes_read[0]} bytes, expected {size}"
+                )
+
+            if hasher is not None:
+                self._content_md5 = hasher.hexdigest()
             return self.complete()
         except FileExistsError:
             return ""