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

[jim-fal]

Symptom

Uploading a large file with fal files upload looks hung: the progress bar sits at or near 0% for the entire transfer, then jumps straight to done. Nothing is actually wrong with the upload — the bar just has no way to move.

Why the bar could not move

Two things compounded in projects/fal/src/fal/files.py:

1. Progress was counted in parts, not bytes. _put_file_multipart created the task with total=num_parts and advanced one unit per completed part. With MULTIPART_CHUNK_SIZE = 10 MB and MULTIPART_MAX_CONCURRENCY = 10, a 100 MB file is exactly 10 parts uploaded by 10 concurrent workers. All ten are in flight at once sharing the same link, so they finish at roughly the same moment — the bar has nothing to report until the transfer is essentially over.
2. A full-file MD5 pass ran before the first byte went out. _compute_md5(lpath) read the whole file while the task displayed "Calculating checksum..." with completed=0. On a large file or a slow disk that is a second silent, apparently-stalled phase before the silent upload phase.

The non-multipart branch (files under the 10 MB threshold) had the same shape: total=1, advanced once after the whole blocking request returned.

What changed

projects/fal/src/fal/upload.py

• ProgressFileReader — a read-through view of a binary stream that reports its absolute offset after each read. httpx rewinds a multipart body before resending it, so reporting an offset rather than a delta keeps a retried part from double-counting.
• _BytesUploadedTracker — folds concurrent per-part offsets into one cumulative total, recomputed from each part's latest offset rather than accumulated, for the same reason.
• BaseMultipartUpload.upload_file gains on_bytes_uploaded(total_bytes), fired as each part's body is handed to the transport. on_part_complete is untouched, and both are optional.
• The MD5 is now folded into the reader thread that upload_file already runs, and exposed as BaseMultipartUpload.content_md5. The file is read once instead of twice, and the hashing overlaps the upload rather than preceding it.

projects/fal/src/fal/files.py

• _put_file_multipart sizes the task in bytes (total=size) and drives it from on_bytes_uploaded. The "Calculating checksum..." phase is gone because there is no longer a separate checksum pass.
• The etag/MD5 integrity check is kept, reading the digest from multipart.content_md5. content_md5 is None until a call has read the file through, so a set etag with no digest raises rather than silently skipping verification (covered by a test).
• The non-multipart branch wraps the open file in ProgressFileReader, so the bar advances while the request body is written.
• Added DownloadColumn so movement is visible even when the percentage rounds to the same integer.

Trade-off worth knowing

upload_file now always computes the MD5, including for AppFileMultipartUpload (app file sync), which previously did not. It rides along with a read the upload has to do anyway and overlaps the network, so the cost is small; the alternative was a flag that every caller would have to thread through.

How to test

Reproduce the symptom and verify the fix

tools/-free repro used here: a fake backend over a single shared 25 MB/s pipe drained at 64 KiB granularity, so ten concurrent parts progress together and finish together — the real-world shape. It drives FalFileSystem._put_file_multipart with a real rich.progress.Progress and samples the bar every 250 ms. No network, no credentials.

demo_progress.py

"""Drive FalFileSystem._put_file_multipart against a rate-limited fake backend.

Simulates a 100 MB upload over a shared 25 MB/s pipe and samples the progress
bar every 250 ms, printing what the user would actually see.
"""

import hashlib
import os
import tempfile
import threading
import time
from types import SimpleNamespace

import httpx
from rich.progress import BarColumn, Progress, SpinnerColumn, TextColumn

from fal.files import FalFileSystem

SIZE = 100 * 1024 * 1024
RATE = 25 * 1024 * 1024  # bytes/sec, shared across all workers

STATE = {"etag": ""}
SLICE = 64 * 1024


class SharedPipe:
    """One shared link: every worker drains it at 64 KiB granularity, so ten
    concurrent parts finish at roughly the same moment, as they do in real life.
    """

    def __init__(self, rate):
        self.rate = rate
        self.lock = threading.Lock()
        self.start = time.monotonic()
        self.sent = 0

    def consume(self, n):
        with self.lock:
            self.sent += n
            due = self.start + self.sent / self.rate
        delay = due - time.monotonic()
        if delay > 0:
            time.sleep(delay)


PIPE = SharedPipe(RATE)


class RateLimitedTransport(httpx.BaseTransport):
    def handle_request(self, request):
        path = request.url.path
        if path.endswith("/initiate"):
            request.read()
            return httpx.Response(200, json={"upload_id": "u1"})
        if path.endswith("/complete"):
            request.read()
            return httpx.Response(200, json={"etag": STATE["etag"]})
        for chunk in request.stream:
            for offset in range(0, len(chunk), SLICE):
                PIPE.consume(min(SLICE, len(chunk) - offset))
        part_number = int(path.rsplit("/", 1)[-1])
        return httpx.Response(
            200, json={"part_number": part_number, "etag": "e%d" % part_number}
        )


def main():
    with tempfile.TemporaryDirectory() as tmp:
        lpath = os.path.join(tmp, "weights.bin")
        payload = os.urandom(SIZE)
        with open(lpath, "wb") as f:
            f.write(payload)
        STATE["etag"] = hashlib.md5(payload).hexdigest()
        del payload

        client = httpx.Client(
            base_url="http://testserver", transport=RateLimitedTransport()
        )

        progress = Progress(
            SpinnerColumn(),
            TextColumn("[progress.description]{task.description}"),
            BarColumn(),
            TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
            disable=True,
        )

        PIPE.start = start = time.monotonic()
        PIPE.sent = 0
        stop = threading.Event()

        def sample():
            while not stop.is_set():
                if progress.tasks:
                    t = progress.tasks[0]
                    print(
                        "t=%5.1fs  %-22s %5.1f%%"
                        % (time.monotonic() - start, t.description, t.percentage),
                        flush=True,
                    )
                else:
                    print(
                        "t=%5.1fs  %-22s %6s"
                        % (time.monotonic() - start, "(no task yet)", "-"),
                        flush=True,
                    )
                stop.wait(0.25)

        sampler = threading.Thread(target=sample, daemon=True)
        sampler.start()
        try:
            FalFileSystem._put_file_multipart(
                SimpleNamespace(_client=client),
                lpath,
                "/data/weights.bin",
                SIZE,
                progress,
            )
        finally:
            stop.set()
            sampler.join()
        print("t=%5.1fs  upload returned, md5 verified" % (time.monotonic() - start))


if __name__ == "__main__":
    main()

python demo_progress.py

Before (main, commit 335261eb) — 0% for three of the four seconds, reaching 10% before the upload completes:

t=  0.0s  (no task yet)               -
t=  0.3s  Uploading weights.bin    0.0%
t=  0.5s  Uploading weights.bin    0.0%
t=  0.8s  Uploading weights.bin    0.0%
t=  1.0s  Uploading weights.bin    0.0%
t=  1.3s  Uploading weights.bin    0.0%
t=  1.5s  Uploading weights.bin    0.0%
t=  1.8s  Uploading weights.bin    0.0%
t=  2.0s  Uploading weights.bin    0.0%
t=  2.3s  Uploading weights.bin    0.0%
t=  2.5s  Uploading weights.bin    0.0%
t=  2.8s  Uploading weights.bin    0.0%
t=  3.0s  Uploading weights.bin   10.0%
t=  3.3s  Uploading weights.bin   10.0%
t=  3.6s  Uploading weights.bin   10.0%
t=  3.8s  Uploading weights.bin   10.0%
t=  4.0s  upload returned, md5 verified

After (this branch) — tracks the transfer, same 4.0 s wall time:

t=  0.0s  (no task yet)               -
t=  0.3s  Uploading weights.bin    6.9%
t=  0.5s  Uploading weights.bin   13.3%
t=  0.8s  Uploading weights.bin   19.7%
t=  1.0s  Uploading weights.bin   26.1%
t=  1.3s  Uploading weights.bin   32.4%
t=  1.5s  Uploading weights.bin   38.7%
t=  1.8s  Uploading weights.bin   44.9%
t=  2.0s  Uploading weights.bin   51.2%
t=  2.3s  Uploading weights.bin   57.6%
t=  2.5s  Uploading weights.bin   64.0%
t=  2.8s  Uploading weights.bin   70.3%
t=  3.0s  Uploading weights.bin   76.7%
t=  3.3s  Uploading weights.bin   83.0%
t=  3.6s  Uploading weights.bin   89.4%
t=  3.8s  Uploading weights.bin   95.8%
t=  4.0s  upload returned, md5 verified

New tests

pytest -v projects/fal/tests/unit/test_upload_progress.py projects/fal/tests/unit/test_files_upload_progress.py

============================= test session starts ==============================
collecting ... collected 13 items

projects/fal/tests/unit/test_upload_progress.py::test_progress_reader_reports_absolute_offset PASSED [  7%]
projects/fal/tests/unit/test_upload_progress.py::test_byte_progress_is_monotonic_and_ends_at_file_size PASSED [ 15%]
projects/fal/tests/unit/test_upload_progress.py::test_byte_progress_never_exceeds_file_size_under_concurrency PASSED [ 23%]
projects/fal/tests/unit/test_upload_progress.py::test_retried_part_does_not_inflate_byte_progress PASSED [ 30%]
projects/fal/tests/unit/test_upload_progress.py::test_content_md5_matches_file_without_a_separate_read PASSED [ 38%]
projects/fal/tests/unit/test_upload_progress.py::test_empty_file_still_reports_a_digest PASSED [ 46%]
projects/fal/tests/unit/test_upload_progress.py::test_part_callback_still_fires_for_every_part PASSED [ 53%]
projects/fal/tests/unit/test_upload_progress.py::test_reader_failure_propagates_and_leaves_no_digest PASSED [ 61%]
projects/fal/tests/unit/test_files_upload_progress.py::test_multipart_task_is_sized_in_bytes_and_tracks_the_transfer PASSED [ 69%]
projects/fal/tests/unit/test_files_upload_progress.py::test_multipart_raises_when_the_server_etag_differs_from_the_local_digest PASSED [ 76%]
projects/fal/tests/unit/test_files_upload_progress.py::test_multipart_raises_when_no_digest_was_produced PASSED [ 84%]
projects/fal/tests/unit/test_files_upload_progress.py::test_small_file_upload_advances_while_the_body_is_written PASSED [ 92%]
projects/fal/tests/unit/test_files_upload_progress.py::test_empty_file_upload_finishes_the_bar PASSED [100%]

============================== 13 passed in 0.04s ==============================

test_upload_progress.py runs the real BaseMultipartUpload against an httpx.MockTransport, which drains the request body, so the byte callbacks fire from the actual multipart encoder rather than from a stub. It covers monotonicity, the exact final total, behaviour under concurrency, a retried part not inflating the count, the digest matching hashlib.md5 of the file, the empty-file path, on_part_complete back-compat, and reader-thread error propagation leaving no digest behind.

test_files_upload_progress.py drives FalFileSystem against a real rich.progress.Progress subclass that records every reported position, so the assertions are against rich's own task state rather than a mock's call log.

This replaces test_files_checksum_progress.py, which asserted the old parts-based task and the "Calculating checksum..." phase that no longer exists.

Full unit suite

pytest -n auto -q projects/fal/tests/unit

1247 passed, 3 skipped, 56 warnings in 5.73s

Lint / types

pre-commit run --files projects/fal/src/fal/files.py projects/fal/src/fal/upload.py \
  projects/fal/tests/unit/test_upload_progress.py projects/fal/tests/unit/test_files_upload_progress.py

ruff-format..............................................................Passed
ruff.....................................................................Passed
ban-lazy-imports-serialized..........................(no files to check)Skipped
check for added large files..............................................Passed
check for merge conflicts................................................Passed
check vcs permalinks.....................................................Passed
debug statements (python)................................................Passed
mypy.....................................................................Passed

Not verified

• No end-to-end fal files upload against a real backend. It needs credentials and a live storage backend, and uploading a 100 MB file to production storage purely to watch a progress bar is not a reasonable test. Every result above comes from fake transports. Worth a manual fal files upload of a >10 MB file on a real account before release.
• Terminal rendering was not eyeballed. The demo reads task.percentage from a Progress constructed with disable=True rather than capturing rendered frames, so the new DownloadColumn layout has not been seen in a terminal.
• AppFileMultipartUpload (app file sync) was only checked by the existing unit tests. It picks up the always-on MD5 computation and was not exercised against a real deploy.
• Retry behaviour is covered only by a synthetic 500. A real mid-stream network failure and reconnect was not reproduced.
• Local pytest ran on Python 3.12 only; CI covers 3.10 through 3.14 plus Windows.

🤖 Generated with Claude Code