diff --git a/.cursor/rules/code-comments.mdc b/.cursor/rules/code-comments.mdc new file mode 100644 index 00000000..fabc2ea6 --- /dev/null +++ b/.cursor/rules/code-comments.mdc @@ -0,0 +1,33 @@ +--- +description: Code comment style and when to comment +globs: **/*.{ts,tsx,mjs,js,py} +alwaysApply: true +--- + +# Code comments + +## Style + +- Explain **why**, not **what**. The code already shows what; comments justify a decision, flag a constraint, or warn about a non-obvious interaction. +- Plain prose, sentence case, periods. No banner art or decorative dashes. +- Reference symbols and paths in backticks (`` `useChat` ``, `lib/db/...`). +- Active voice ("Returns null on..." not "Null is returned on..."). +- Reference constants by name; don't restate their value. + +## Comment when + +A future reader would otherwise have to spelunk to answer "why does it do this?": + +- Architectural choices and invariants the type system can't express. +- Workarounds for bugs / browser quirks / library limits — link the issue and note when it can be removed. +- Non-obvious tradeoffs ("we re-render on every keystroke because the sidebar live-mirrors the value"). + +## Don't comment + +- Restating the line below (`// increment the counter`). +- Standard library / framework behavior (`autoFocus = true` doesn't need a paragraph explaining what `autoFocus` does). +- TODOs without an owner or ticket — open an issue instead. + +## Length + +Inline comments should be **3 lines or fewer**. File headers cap at **4–5 lines**. Go longer only when a reader would genuinely be lost without it; otherwise trim. If a long comment only restates what the code does, delete it. diff --git a/.cursor/rules/run-checks.mdc b/.cursor/rules/run-checks.mdc index ec5e0c2c..88339816 100644 --- a/.cursor/rules/run-checks.mdc +++ b/.cursor/rules/run-checks.mdc @@ -1,5 +1,6 @@ --- description: Ensure that all code changes are checked for formatting, lint, and type errors by running `make check-all` after every update. +globs: **/*.{ts,tsx,mjs,js,py} alwaysApply: true --- diff --git a/lambda/src/data_hub_lambda/archive_builder.py b/lambda/src/data_hub_lambda/archive_builder.py index cc349def..42828e08 100644 --- a/lambda/src/data_hub_lambda/archive_builder.py +++ b/lambda/src/data_hub_lambda/archive_builder.py @@ -28,7 +28,10 @@ """ from __future__ import annotations +import io import logging +from collections.abc import Iterator +from concurrent.futures import Future, ThreadPoolExecutor from dataclasses import dataclass from typing import Any @@ -61,6 +64,20 @@ # case before it reaches S3. _MAX_TOTAL_BYTES = _PART_SIZE_BYTES * _MAX_PARTS +# The bottleneck for runs with thousands of tiny files is per-object +# ``GetObject`` latency, not bandwidth, so overlapping fetches collapses what +# was a serial chain of round-trips. +_PREFETCH_CONCURRENCY = 16 + +# Files at or below this size are prefetched into memory; larger and +# unknown-size files stream inline so peak memory stays bounded regardless of +# total archive size (a 200 GB single-file run must still fit the Lambda). +_PREFETCH_MAX_FILE_BYTES = 16 * 1024 * 1024 + +# Bounds peak memory from the look-ahead window (many ``_PREFETCH_MAX_FILE_BYTES`` +# files queued ahead of a slow writer) independently of ``_PREFETCH_CONCURRENCY``. +_PREFETCH_MAX_INFLIGHT_BYTES = 256 * 1024 * 1024 + # --------------------------------------------------------------------------- # Multipart upload file-like wrapper @@ -212,11 +229,17 @@ class ArchiveFile: ``source_bucket`` is per-file so the builder can zip across the raw and processed buckets in a single archive — the web app sends each file's bucket alongside its key. + + ``size_bytes`` is an optional hint used solely to decide whether a file is + small enough to prefetch into memory concurrently (see ``build_run_archive``). + It does not affect correctness: an unknown size is treated as "too large to + buffer" and the file is streamed inline. Never trusted for allocation. """ key: str name: str source_bucket: str + size_bytes: int | None = None @dataclass @@ -299,7 +322,15 @@ def parse_build_request( raise ValueError(f"File key '{key}' does not belong to run '{expected_prefix}'") if "/" in name or name in ("", ".", ".."): raise ValueError(f"Invalid archive entry name: {name!r}") - parsed_files.append(ArchiveFile(key=key, name=name, source_bucket=bucket)) + + # Optional prefetch hint, coerced leniently: anything but a + # non-negative int becomes ``None`` ("unknown" size), so a malformed + # value streams the file inline rather than failing the build. + raw_size = entry.get("size_bytes") + size_bytes = raw_size if isinstance(raw_size, int) and raw_size >= 0 else None + parsed_files.append( + ArchiveFile(key=key, name=name, source_bucket=bucket, size_bytes=size_bytes) + ) return BuildArchiveRequest( instrument_id=instrument_id, @@ -321,6 +352,14 @@ def build_run_archive( in-memory buffer. Returns the destination location and total bytes uploaded on success. Aborts the multipart upload on any failure so partial objects don't leak. + + Source objects that fit under ``_PREFETCH_MAX_FILE_BYTES`` are fetched + concurrently a window ahead of the writer (see ``_iter_archive_readers``), + which collapses the per-object GetObject latency that otherwise dominates + runs with thousands of tiny files. The zip is still written single-threaded + and in request order, so entry ordering and the multipart stream are + untouched. Larger (or unknown-size) files stream inline to keep memory + bounded. """ import zipfile @@ -333,6 +372,10 @@ def build_run_archive( part_size=_PART_SIZE_BYTES, ) + # Managed by hand rather than ``with`` so the error path can cancel + # queued-but-unstarted prefetches via ``cancel_futures=True``, instead of + # draining doomed ``GetObject`` work before the exception surfaces. + executor = ThreadPoolExecutor(max_workers=_PREFETCH_CONCURRENCY) try: # ZipFile.write() requires a real file path on disk; .open() with # mode="w" returns a writable handle we can stream into. Both rely @@ -343,8 +386,8 @@ def build_run_archive( compression=zipfile.ZIP_STORED, allowZip64=True, ) as zf: - for file in request.files: - _append_file_to_zip(s3, file, zf) + for file, reader in _iter_archive_readers(s3, request.files, executor): + _write_reader_to_zip(reader, file.name, zf) if stream.tell() > _MAX_TOTAL_BYTES: raise ValueError( f"Archive exceeded the {_MAX_TOTAL_BYTES}-byte cap " @@ -354,6 +397,8 @@ def build_run_archive( except Exception: stream.abort() raise + finally: + executor.shutdown(cancel_futures=True) stream.close() @@ -364,17 +409,96 @@ def build_run_archive( ) -def _append_file_to_zip(s3_client: Any, file: ArchiveFile, zf: Any) -> None: +def _is_prefetch_eligible(file: ArchiveFile) -> bool: + # Unknown size → not eligible: we can't admit it to the byte budget, and + # the safe assumption is that it might be huge, so it streams inline. + return file.size_bytes is not None and file.size_bytes <= _PREFETCH_MAX_FILE_BYTES + + +def _fetch_to_buffer(s3_client: Any, file: ArchiveFile) -> io.BytesIO: + """Read a small source object fully into memory, on a worker thread. + + Only ever called for ``_is_prefetch_eligible`` files, so the buffer is + bounded by ``_PREFETCH_MAX_FILE_BYTES``. + """ obj = s3_client.get_object(Bucket=file.source_bucket, Key=file.key) body = obj["Body"] + try: + return io.BytesIO(body.read()) + finally: + body.close() + + +def _iter_archive_readers( + s3_client: Any, + files: list[ArchiveFile], + executor: ThreadPoolExecutor, +) -> Iterator[tuple[ArchiveFile, Any]]: + """Yield ``(file, reader)`` pairs in request order, prefetching small files. + + A sliding window of up to ``_PREFETCH_CONCURRENCY`` eligible files (and at + most ``_PREFETCH_MAX_INFLIGHT_BYTES`` of buffered data) is fetched ahead of + the consumer on worker threads. ``reader`` is an in-memory ``BytesIO`` for + prefetched files, or the live ``StreamingBody`` for inline (large/unknown) + files fetched lazily when the consumer reaches them. Either way the reader + exposes ``read(n)``/``close()`` so the writer treats them identically. + + The submission cursor advances past inline files without buffering them, so + a single large file in the middle of a run doesn't stall prefetching of the + small files after it. + """ + n = len(files) + submit_idx = 0 + inflight_bytes = 0 + futures: dict[int, Future[io.BytesIO]] = {} + + def _pump() -> None: + nonlocal submit_idx, inflight_bytes + while submit_idx < n and len(futures) < _PREFETCH_CONCURRENCY: + file = files[submit_idx] + if not _is_prefetch_eligible(file): + # Streamed inline at consume time; advance so the window can + # keep reaching the small files beyond it. + submit_idx += 1 + continue + size = file.size_bytes or 0 + # Always allow at least one in-flight fetch; otherwise honor the + # byte budget so the look-ahead window can't balloon memory. + if futures and inflight_bytes + size > _PREFETCH_MAX_INFLIGHT_BYTES: + break + futures[submit_idx] = executor.submit(_fetch_to_buffer, s3_client, file) + inflight_bytes += size + submit_idx += 1 + + try: + for idx in range(n): + file = files[idx] + _pump() + future = futures.pop(idx, None) + if future is not None: + buffer = future.result() + inflight_bytes -= file.size_bytes or 0 + yield file, buffer + else: + # Large/unknown-size file: stream it inline, lazily, now. + obj = s3_client.get_object(Bucket=file.source_bucket, Key=file.key) + yield file, obj["Body"] + finally: + # On early exit (writer error mid-stream) cancel prefetches we never + # consumed so the executor's shutdown doesn't block on doomed work. + for future in futures.values(): + future.cancel() + + +def _write_reader_to_zip(reader: Any, name: str, zf: Any) -> None: try: # force_zip64=True makes the per-entry header ZIP64-capable so files # ≥4 GB don't blow up the writer mid-stream. - with zf.open(file.name, mode="w", force_zip64=True) as entry: + with zf.open(name, mode="w", force_zip64=True) as entry: while True: - chunk = body.read(_COPY_BLOCK_SIZE_BYTES) + chunk = reader.read(_COPY_BLOCK_SIZE_BYTES) if not chunk: break entry.write(chunk) finally: - body.close() + reader.close() diff --git a/lambda/tests/archive_builder/test_archive_builder.py b/lambda/tests/archive_builder/test_archive_builder.py index 5f28705c..1303f454 100644 --- a/lambda/tests/archive_builder/test_archive_builder.py +++ b/lambda/tests/archive_builder/test_archive_builder.py @@ -9,6 +9,7 @@ from __future__ import annotations import io +import threading import zipfile from typing import Any @@ -50,14 +51,36 @@ def __init__(self, source_objects: dict[tuple[str, str], bytes] | None = None) - self.fail_part_at: int | None = None self.calls: list[tuple[str, dict[str, Any]]] = [] + # Concurrency instrumentation for the prefetch tests. `get_object` can + # now run on worker threads, so guard shared state with a lock and + # track peak in-flight GETs + the thread each key was fetched on. + self._lock = threading.Lock() + self._concurrent_get = 0 + self.max_concurrent_get = 0 + self.get_threads: dict[str, int] = {} + # When set, every get_object rendezvouses here before returning, so a + # test can assert N fetches genuinely overlap (a serial implementation + # would deadlock and time out). + self.get_barrier: threading.Barrier | None = None + # -- source side -- def get_object(self, *, Bucket: str, Key: str) -> dict[str, Any]: - self.calls.append(("get_object", {"Bucket": Bucket, "Key": Key})) + with self._lock: + self.calls.append(("get_object", {"Bucket": Bucket, "Key": Key})) + self.get_threads[Key] = threading.get_ident() + self._concurrent_get += 1 + self.max_concurrent_get = max(self.max_concurrent_get, self._concurrent_get) try: - data = self.source_objects[(Bucket, Key)] - except KeyError as exc: - raise FileNotFoundError(f"missing fixture s3://{Bucket}/{Key}") from exc - return {"Body": _StubS3Body(data)} + if self.get_barrier is not None: + self.get_barrier.wait(timeout=5) + try: + data = self.source_objects[(Bucket, Key)] + except KeyError as exc: + raise FileNotFoundError(f"missing fixture s3://{Bucket}/{Key}") from exc + return {"Body": _StubS3Body(data)} + finally: + with self._lock: + self._concurrent_get -= 1 # -- destination side -- def create_multipart_upload(self, *, Bucket: str, Key: str) -> dict[str, Any]: @@ -133,8 +156,13 @@ def _make_request( ) -def _file(key: str, name: str, source_bucket: str = "raw-bucket") -> ArchiveFile: - return ArchiveFile(key=key, name=name, source_bucket=source_bucket) +def _file( + key: str, + name: str, + source_bucket: str = "raw-bucket", + size_bytes: int | None = None, +) -> ArchiveFile: + return ArchiveFile(key=key, name=name, source_bucket=source_bucket, size_bytes=size_bytes) # --------------------------------------------------------------------------- @@ -251,6 +279,128 @@ def test_large_file_spans_multiple_parts(self, monkeypatch: pytest.MonkeyPatch) assert zf.read("big.bin") == body +# --------------------------------------------------------------------------- +# Prefetch / concurrency behavior +# --------------------------------------------------------------------------- + + +class TestBuildRunArchivePrefetch: + def test_many_small_files_preserve_order_and_contents(self) -> None: + # The motivating case: thousands of tiny files. Sizes are supplied so + # every file is prefetch-eligible. Order and contents must survive the + # out-of-order concurrent fetches. + count = 50 + objects: dict[tuple[str, str], bytes] = {} + files: list[ArchiveFile] = [] + for i in range(count): + key = f"akta-fplc/RUN001/f{i:04d}.csv" + body = f"row-{i}\n".encode() * (i + 1) + objects[("raw-bucket", key)] = body + files.append(_file(key, f"f{i:04d}.csv", size_bytes=len(body))) + s3 = StubS3Client(objects) + request = _make_request(files) + + build_run_archive(request, s3_client=s3) + + zipped = s3.uploaded_objects[("archives-bucket", "runs/akta-fplc/RUN001/abc.zip")] + with zipfile.ZipFile(io.BytesIO(zipped)) as zf: + assert zf.namelist() == [f"f{i:04d}.csv" for i in range(count)] + for i in range(count): + assert zf.read(f"f{i:04d}.csv") == f"row-{i}\n".encode() * (i + 1) + + def test_small_files_are_fetched_concurrently(self) -> None: + # A barrier sized to the file count forces every get_object to overlap; + # a serial fetch loop would never release the barrier and the wait() + # would raise BrokenBarrierError / time out. + count = 4 + objects: dict[tuple[str, str], bytes] = {} + files: list[ArchiveFile] = [] + for i in range(count): + key = f"akta-fplc/RUN001/f{i}.csv" + body = f"data-{i}".encode() + objects[("raw-bucket", key)] = body + files.append(_file(key, f"f{i}.csv", size_bytes=len(body))) + s3 = StubS3Client(objects) + s3.get_barrier = threading.Barrier(count) + request = _make_request(files) + + build_run_archive(request, s3_client=s3) + + assert s3.max_concurrent_get == count + + def test_large_file_streams_inline_on_main_thread( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + # Drop the prefetch threshold so the "big" file is ineligible. Inline + # files are fetched lazily on the calling thread (never buffered whole + # on a worker); eligible files are fetched on a pool thread. + import data_hub_lambda.archive_builder as ab + + monkeypatch.setattr(ab, "_PREFETCH_MAX_FILE_BYTES", 16) + s3 = StubS3Client( + { + ("raw-bucket", "akta-fplc/RUN001/big.bin"): b"x" * 4096, + ("raw-bucket", "akta-fplc/RUN001/small.csv"): b"tiny", + } + ) + request = _make_request( + [ + _file("akta-fplc/RUN001/big.bin", "big.bin", size_bytes=4096), + _file("akta-fplc/RUN001/small.csv", "small.csv", size_bytes=4), + ] + ) + main_ident = threading.get_ident() + + build_run_archive(request, s3_client=s3) + + # Big file streamed inline on the main thread; small file prefetched on + # a worker thread. + assert s3.get_threads["akta-fplc/RUN001/big.bin"] == main_ident + assert s3.get_threads["akta-fplc/RUN001/small.csv"] != main_ident + zipped = s3.uploaded_objects[("archives-bucket", "runs/akta-fplc/RUN001/abc.zip")] + with zipfile.ZipFile(io.BytesIO(zipped)) as zf: + assert zf.read("big.bin") == b"x" * 4096 + assert zf.read("small.csv") == b"tiny" + + def test_unknown_size_streams_inline(self) -> None: + # No size hint → treated as "might be huge" → inline on the main thread. + s3 = StubS3Client({("raw-bucket", "akta-fplc/RUN001/unknown.bin"): b"payload"}) + request = _make_request([_file("akta-fplc/RUN001/unknown.bin", "unknown.bin")]) + main_ident = threading.get_ident() + + build_run_archive(request, s3_client=s3) + + assert s3.get_threads["akta-fplc/RUN001/unknown.bin"] == main_ident + + def test_mixed_small_and_large_preserve_order(self, monkeypatch: pytest.MonkeyPatch) -> None: + import data_hub_lambda.archive_builder as ab + + monkeypatch.setattr(ab, "_PREFETCH_MAX_FILE_BYTES", 16) + s3 = StubS3Client( + { + ("raw-bucket", "akta-fplc/RUN001/a.csv"): b"alpha", + ("raw-bucket", "akta-fplc/RUN001/big.bin"): b"B" * 100, + ("raw-bucket", "akta-fplc/RUN001/c.csv"): b"charlie", + } + ) + request = _make_request( + [ + _file("akta-fplc/RUN001/a.csv", "a.csv", size_bytes=5), + _file("akta-fplc/RUN001/big.bin", "big.bin", size_bytes=100), + _file("akta-fplc/RUN001/c.csv", "c.csv", size_bytes=7), + ] + ) + + build_run_archive(request, s3_client=s3) + + zipped = s3.uploaded_objects[("archives-bucket", "runs/akta-fplc/RUN001/abc.zip")] + with zipfile.ZipFile(io.BytesIO(zipped)) as zf: + assert zf.namelist() == ["a.csv", "big.bin", "c.csv"] + assert zf.read("a.csv") == b"alpha" + assert zf.read("big.bin") == b"B" * 100 + assert zf.read("c.csv") == b"charlie" + + # --------------------------------------------------------------------------- # Failure-path tests # --------------------------------------------------------------------------- @@ -272,6 +422,27 @@ def test_aborts_multipart_on_upload_part_failure(self, monkeypatch: pytest.Monke assert s3.aborted, "expected abort_multipart_upload to be called" assert ("archives-bucket", "runs/akta-fplc/RUN001/abc.zip") not in s3.uploaded_objects + def test_aborts_when_prefetch_get_object_fails(self) -> None: + # An eligible (sized) file whose source object is missing fails inside + # the worker; the future result re-raises in the main thread, which + # must abort the multipart upload rather than leak a partial object. + s3 = StubS3Client( + {("raw-bucket", "akta-fplc/RUN001/present.csv"): b"here"} + # "missing.csv" intentionally absent. + ) + request = _make_request( + [ + _file("akta-fplc/RUN001/present.csv", "present.csv", size_bytes=4), + _file("akta-fplc/RUN001/missing.csv", "missing.csv", size_bytes=4), + ] + ) + + with pytest.raises(FileNotFoundError): + build_run_archive(request, s3_client=s3) + + assert s3.aborted, "expected abort_multipart_upload to be called" + assert ("archives-bucket", "runs/akta-fplc/RUN001/abc.zip") not in s3.uploaded_objects + def test_refuses_to_exceed_max_parts(self, monkeypatch: pytest.MonkeyPatch) -> None: # With a 256-byte part size and a 4-part cap, an 8 KB file forces the # writer past the cap mid-stream. We expect a clear ValueError before @@ -437,3 +608,24 @@ def test_rejects_name_with_path_traversal(self) -> None: ] with pytest.raises(ValueError, match="Invalid archive entry name"): parse_build_request(payload) + + def test_parses_valid_size_bytes(self) -> None: + payload = self._base_payload() + payload["files"][0]["size_bytes"] = 2048 + request = parse_build_request(payload) + assert request.files[0].size_bytes == 2048 + + def test_size_bytes_defaults_to_none_when_absent(self) -> None: + # The base payload omits size_bytes (legacy callers); it must parse as + # None so the file streams inline rather than crashing. + request = parse_build_request(self._base_payload()) + assert request.files[0].size_bytes is None + + @pytest.mark.parametrize("bad_size", [-1, "1024", 1.5, None]) + def test_invalid_size_bytes_coerced_to_none(self, bad_size: Any) -> None: + # Lenient coercion: a malformed hint never rejects the build, it just + # falls back to inline streaming. + payload = self._base_payload() + payload["files"][0]["size_bytes"] = bad_size + request = parse_build_request(payload) + assert request.files[0].size_bytes is None diff --git a/web/lib/api/archive-builder.ts b/web/lib/api/archive-builder.ts index da7821eb..f33f8936 100644 --- a/web/lib/api/archive-builder.ts +++ b/web/lib/api/archive-builder.ts @@ -88,7 +88,16 @@ export type InvokeBuildArchiveInput = { // produced processed CSV). The Lambda allow-lists each bucket against its // own configured raw + processed env vars, so this is not a pivot point // for a caller with `lambda:InvokeFunctionUrl` to read arbitrary S3. - files: { s3Key: string; filename: string; sourceBucket: string }[]; + // + // `sizeBytes` is an optional hint the builder uses to decide whether a file + // is small enough to prefetch concurrently; it never affects correctness, so + // it's fine to omit (the builder streams unknown-size files inline). + files: { + s3Key: string; + filename: string; + sourceBucket: string; + sizeBytes?: number | null; + }[]; }; export type InvokeBuildArchiveResult = @@ -128,6 +137,9 @@ export async function invokeBuildArchive( key: f.s3Key, name: f.filename, source_bucket: f.sourceBucket, + // Snake-cased to match the Lambda payload contract. Omitted when null so + // the builder treats it as "unknown" and streams the file inline. + ...(f.sizeBytes == null ? {} : { size_bytes: f.sizeBytes }), })), }; if (input.jobId) payload.job_id = input.jobId; diff --git a/web/lib/api/run-archive.ts b/web/lib/api/run-archive.ts index 8b62b4a6..bfd49371 100644 --- a/web/lib/api/run-archive.ts +++ b/web/lib/api/run-archive.ts @@ -19,17 +19,52 @@ import { import { and, eq, inArray, isNull } from "drizzle-orm"; import { after } from "next/server"; -// Hint to caller (LLM, polling UI) for how long to wait before checking -// again. Sized to be longer than a small archive's typical build (which -// the Lambda finishes in 1-3s for a few-MB run) but short enough that the -// chat experience doesn't stall. +// Floor for the build-time retry hint handed to the caller (LLM, polling UI). +// Even a trivial archive pays Lambda cold-start + invoke + presign overhead, +// so polling sooner than this just wastes round-trips. export const ARCHIVE_BUILD_RETRY_AFTER_SECONDS = 5; +// Ceiling for the hint, keeping the chat interactive: better to have a caller +// poll a couple extra times on a genuinely huge run than stall on one long +// wait when the build may well finish early. +export const ARCHIVE_BUILD_RETRY_AFTER_MAX_SECONDS = 30; + +// Coefficients for the retry-hint estimate. Build time is dominated by either +// per-object `GetObject` latency (many tiny files) or throughput (a few large +// files), so the estimate sums both terms. They are rough first guesses; the +// floor/cap bound the error, and they can be retuned against real telemetry. +const ARCHIVE_BUILD_BASE_SECONDS = 3; +const ARCHIVE_BUILD_SECONDS_PER_FILE = 0.005; +const ARCHIVE_BUILD_BYTES_PER_SECOND = 200 * 1024 * 1024; + +// Estimates how long the caller should wait before polling again, from the +// shape of the run. `totalBytes` sums only known file sizes (NULL sizes add +// nothing to the throughput term but still count toward the per-file term). +// Clamped to [`ARCHIVE_BUILD_RETRY_AFTER_SECONDS`, `ARCHIVE_BUILD_RETRY_AFTER_MAX_SECONDS`]. +export function estimateRetryAfterSeconds(input: { + fileCount: number; + totalBytes: number; +}): number { + const estimate = + ARCHIVE_BUILD_BASE_SECONDS + + input.fileCount * ARCHIVE_BUILD_SECONDS_PER_FILE + + input.totalBytes / ARCHIVE_BUILD_BYTES_PER_SECOND; + const rounded = Math.ceil(estimate); + return Math.min( + ARCHIVE_BUILD_RETRY_AFTER_MAX_SECONDS, + Math.max(ARCHIVE_BUILD_RETRY_AFTER_SECONDS, rounded) + ); +} + export type DownloadableFile = { id: number; filename: string; s3Bucket: string; s3Key: string; + // Nullable: older rows and not-yet-sized uploads can lack a size. Used only + // to estimate the build-time retry hint and to let the builder decide + // prefetch eligibility — never required for correctness. + sizeBytes: number | null; }; export type PrepareRunArchiveInput = { @@ -197,17 +232,25 @@ export async function prepareRunArchive( s3Key: f.s3Key, filename: f.filename, sourceBucket: f.s3Bucket, + sizeBytes: f.sizeBytes, })), }; after(() => runArchiveBuildInBackground(job.id, buildInput, fingerprint)); } + const totalBytes = downloadable.reduce( + (sum, f) => sum + (f.sizeBytes ?? 0), + 0 + ); return { ok: true, status: "building", jobId: job.id, ownsBuild, - retryAfterSeconds: ARCHIVE_BUILD_RETRY_AFTER_SECONDS, + retryAfterSeconds: estimateRetryAfterSeconds({ + fileCount: downloadable.length, + totalBytes, + }), }; } @@ -233,6 +276,7 @@ async function loadDownloadableFiles( filename: files.filename, s3Bucket: files.s3Bucket, s3Key: files.s3Key, + sizeBytes: files.sizeBytes, }) .from(files) .where(and(...conditions)); @@ -246,6 +290,7 @@ async function loadDownloadableFiles( filename: f.filename, s3Bucket: f.s3Bucket, s3Key: f.s3Key, + sizeBytes: f.sizeBytes, })); } diff --git a/web/package-lock.json b/web/package-lock.json index 700725a9..15876ab9 100644 --- a/web/package-lock.json +++ b/web/package-lock.json @@ -3461,9 +3461,9 @@ } }, "node_modules/@next/env": { - "version": "16.2.6", - "resolved": "https://registry.npmjs.org/@next/env/-/env-16.2.6.tgz", - "integrity": "sha512-gd8HoHN4ufj73WmR3JmVolrpJR47ILK6LouP5xElPglaVxir6e1a7VzvTvDWkOoPXT9rkkTzyCxBu4yeZfZwcw==", + "version": "16.2.7", + "resolved": "https://registry.npmjs.org/@next/env/-/env-16.2.7.tgz", + "integrity": "sha512-tMJizPlj6ZYpBMMdK8S0LJufrP4QTdR6pcv9KQ/bVETPAmg0j1mlHE9G2c38UyGHxoBapgwuj7XjbGJ2RcDFOg==", "license": "MIT" }, "node_modules/@next/eslint-plugin-next": { @@ -3477,9 +3477,9 @@ } }, "node_modules/@next/swc-darwin-arm64": { - "version": "16.2.6", - "resolved": "https://registry.npmjs.org/@next/swc-darwin-arm64/-/swc-darwin-arm64-16.2.6.tgz", - "integrity": "sha512-ZJGkkcNfYgrrMkqOdZ7zoLa1TOy0qpcMfk/z4Mh/FKUz40gVO+HNQWqmLxf67Z5WB64DRp0dhEbyHfel+6sJUg==", + "version": "16.2.7", + "resolved": "https://registry.npmjs.org/@next/swc-darwin-arm64/-/swc-darwin-arm64-16.2.7.tgz", + "integrity": "sha512-vm1EDI/pVaBNNiychmxk3fft+OhQPVD9cIM/tReLZIQ3TfQ4kqI9DwKk00dzuS1ulC7icbrzCFrmRRlk9PfNdw==", "cpu": [ "arm64" ], @@ -3493,9 +3493,9 @@ } }, "node_modules/@next/swc-darwin-x64": { - "version": "16.2.6", - "resolved": "https://registry.npmjs.org/@next/swc-darwin-x64/-/swc-darwin-x64-16.2.6.tgz", - "integrity": "sha512-v/YLBHIY132Ced3puBJ7YJKw1lqsCrgcNo2aRJlCEyQrrCeRJlvGlnmxhPxNQI3KE3N1DN5r9TPNPvka3nq5RQ==", + "version": "16.2.7", + "resolved": "https://registry.npmjs.org/@next/swc-darwin-x64/-/swc-darwin-x64-16.2.7.tgz", + "integrity": "sha512-O3IRSv1ZBL1zs0WrIgefTEcTKFVn+ryxBNe54erJ6KsD+2f/Mmt7g2jOYh8PSBdUwPtKQJuCsTMlZ7tIu2AcsQ==", "cpu": [ "x64" ], @@ -3509,12 +3509,15 @@ } }, "node_modules/@next/swc-linux-arm64-gnu": { - "version": "16.2.6", - "resolved": "https://registry.npmjs.org/@next/swc-linux-arm64-gnu/-/swc-linux-arm64-gnu-16.2.6.tgz", - "integrity": "sha512-RPOvqlYBbcQjkz9VQQDZ2T2bARIjXZV1KFlt+V2Mr6SW/e4I9fcKsaA0hdyf2FHoTlsV2xnBd5Y912rP/1Ce6w==", + "version": "16.2.7", + "resolved": "https://registry.npmjs.org/@next/swc-linux-arm64-gnu/-/swc-linux-arm64-gnu-16.2.7.tgz", + "integrity": "sha512-Re6PZtjBDd0aMU+VcZcC/PrIvj4WhrjDYtMhhCVQamWN4L90EVP0pcEOBQD25prSlw7OzNw5QpHLWMilRLsRNw==", "cpu": [ "arm64" ], + "libc": [ + "glibc" + ], "license": "MIT", "optional": true, "os": [ @@ -3525,12 +3528,15 @@ } }, "node_modules/@next/swc-linux-arm64-musl": { - "version": "16.2.6", - "resolved": "https://registry.npmjs.org/@next/swc-linux-arm64-musl/-/swc-linux-arm64-musl-16.2.6.tgz", - "integrity": "sha512-URUTu1+dMkxJsPFgm+OeEvq9wf5sujw0EvgYy80TDGHTSLTnIHeqb0Eu8A3sC95IRgjejQL+kC4mw+4yPxiAXA==", + "version": "16.2.7", + "resolved": "https://registry.npmjs.org/@next/swc-linux-arm64-musl/-/swc-linux-arm64-musl-16.2.7.tgz", + "integrity": "sha512-qyogG9QtBzWxgJfeGBvOEHI3851gTfCF3wLZ5RDLTBJGAmE9p1qDwKCOdrBrvBzRvYDT+gUDp72pzlSEfAXgNA==", "cpu": [ "arm64" ], + "libc": [ + "musl" + ], "license": "MIT", "optional": true, "os": [ @@ -3541,12 +3547,15 @@ } }, "node_modules/@next/swc-linux-x64-gnu": { - "version": "16.2.6", - "resolved": "https://registry.npmjs.org/@next/swc-linux-x64-gnu/-/swc-linux-x64-gnu-16.2.6.tgz", - "integrity": "sha512-DOj182mPV8G3UkrayLoREM5YEYI+Dk5wv7Ox9xl1fFibAELEsFD0lDPfHIeILlutMMfdyhlzYPELG3peuKaurw==", + "version": "16.2.7", + "resolved": "https://registry.npmjs.org/@next/swc-linux-x64-gnu/-/swc-linux-x64-gnu-16.2.7.tgz", + "integrity": "sha512-Vhe4ZDuBpmMogrGi5D4R2Kq4JAQlj6+wvgaFYy31zfES0zPmt6TLA+cuYpM/OLrPZjo2MYQTHVqNUSCR6+fDZQ==", "cpu": [ "x64" ], + "libc": [ + "glibc" + ], "license": "MIT", "optional": true, "os": [ @@ -3557,12 +3566,15 @@ } }, "node_modules/@next/swc-linux-x64-musl": { - "version": "16.2.6", - "resolved": "https://registry.npmjs.org/@next/swc-linux-x64-musl/-/swc-linux-x64-musl-16.2.6.tgz", - "integrity": "sha512-HKQ5SP/V/ub73UvF7n/zeJlxk2kLmtL7Wzrg4WfmkjmNos5onJ2tKu7yZOPdL18A6Svfn3max29ym+ry7NkK4g==", + "version": "16.2.7", + "resolved": "https://registry.npmjs.org/@next/swc-linux-x64-musl/-/swc-linux-x64-musl-16.2.7.tgz", + "integrity": "sha512-srvian89JahFLw1YLBEuhvPJ0DO5lpUeJQMXy4xYo7g628ZlNgXdNkqoxSAv9OYrBfByh6vxISMwW/mRbzCY+g==", "cpu": [ "x64" ], + "libc": [ + "musl" + ], "license": "MIT", "optional": true, "os": [ @@ -3573,9 +3585,9 @@ } }, "node_modules/@next/swc-win32-arm64-msvc": { - "version": "16.2.6", - "resolved": "https://registry.npmjs.org/@next/swc-win32-arm64-msvc/-/swc-win32-arm64-msvc-16.2.6.tgz", - "integrity": "sha512-LZXpTlPyS5v7HhSmnvsLGP3iIYgYOBnc8r8ArlT55sGHV89bR2HlDdBjWQ+PY6SJMmk8TuVGFuxalnP3k/0Dwg==", + "version": "16.2.7", + "resolved": "https://registry.npmjs.org/@next/swc-win32-arm64-msvc/-/swc-win32-arm64-msvc-16.2.7.tgz", + "integrity": "sha512-GX3wvLpULFuRFJzwHaKfm7QZJ18F4ZSuxlPJ96BoBglCzBmdSjyeBKF+ZhWhvL/ckxNfLnNa7bsObO2ipYpszw==", "cpu": [ "arm64" ], @@ -3589,9 +3601,9 @@ } }, "node_modules/@next/swc-win32-x64-msvc": { - "version": "16.2.6", - "resolved": "https://registry.npmjs.org/@next/swc-win32-x64-msvc/-/swc-win32-x64-msvc-16.2.6.tgz", - "integrity": "sha512-F0+4i0h9J6C4eE3EAPWsoCk7UW/dbzOjyzxY0qnDUOYFu6FFmdZ6l97/XdV3/Nz3VYyO7UWjyEJUXkGqcoXfMA==", + "version": "16.2.7", + "resolved": "https://registry.npmjs.org/@next/swc-win32-x64-msvc/-/swc-win32-x64-msvc-16.2.7.tgz", + "integrity": "sha512-J4WlM72NMk076Qsg0jTdK3SNXatlSdnjW7L7oNGLst1tAGjHrJh/FYi+pw9wyIjEtGRKDNzD0zuiY16oWYWVaw==", "cpu": [ "x64" ], @@ -11220,9 +11232,9 @@ } }, "node_modules/hono": { - "version": "4.12.18", - "resolved": "https://registry.npmjs.org/hono/-/hono-4.12.18.tgz", - "integrity": "sha512-RWzP96k/yv0PQfyXnWjs6zot20TqfpfsNXhOnev8d1InAxubW93L11/oNUc3tQqn2G0bSdAOBpX+2uDFHV7kdQ==", + "version": "4.12.25", + "resolved": "https://registry.npmjs.org/hono/-/hono-4.12.25.tgz", + "integrity": "sha512-2NFaIyNVgJmBs/ecmtGzlmluTFs5cHEWGTdu0t1HBwYzoGXOL5nUQBRMXsXWla5i4KkG//QMzVP88m1+I3fdAQ==", "license": "MIT", "engines": { "node": ">=16.9.0" @@ -13760,12 +13772,12 @@ } }, "node_modules/next": { - "version": "16.2.6", - "resolved": "https://registry.npmjs.org/next/-/next-16.2.6.tgz", - "integrity": "sha512-qOVgKJg1+At15NpeUP+eJgCHvTCgXsogweq87Ri/Ix7PkqQHg4sdaXmSFqKlgaIXE4kW0g25LE68W87UANlHtw==", + "version": "16.2.7", + "resolved": "https://registry.npmjs.org/next/-/next-16.2.7.tgz", + "integrity": "sha512-eMJxgjRzBaj3olkP4cBamHDXL79A8FC6u1GcsO1D1Tsx8bw/LLXUJCaoajVxtnhD3A1IJqIT8IcRJjgBIPJq4w==", "license": "MIT", "dependencies": { - "@next/env": "16.2.6", + "@next/env": "16.2.7", "@swc/helpers": "0.5.15", "baseline-browser-mapping": "^2.9.19", "caniuse-lite": "^1.0.30001579", @@ -13779,14 +13791,14 @@ "node": ">=20.9.0" }, "optionalDependencies": { - "@next/swc-darwin-arm64": "16.2.6", - "@next/swc-darwin-x64": "16.2.6", - "@next/swc-linux-arm64-gnu": "16.2.6", - "@next/swc-linux-arm64-musl": "16.2.6", - "@next/swc-linux-x64-gnu": "16.2.6", - "@next/swc-linux-x64-musl": "16.2.6", - "@next/swc-win32-arm64-msvc": "16.2.6", - "@next/swc-win32-x64-msvc": "16.2.6", + "@next/swc-darwin-arm64": "16.2.7", + "@next/swc-darwin-x64": "16.2.7", + "@next/swc-linux-arm64-gnu": "16.2.7", + "@next/swc-linux-arm64-musl": "16.2.7", + "@next/swc-linux-x64-gnu": "16.2.7", + "@next/swc-linux-x64-musl": "16.2.7", + "@next/swc-win32-arm64-msvc": "16.2.7", + "@next/swc-win32-x64-msvc": "16.2.7", "sharp": "^0.34.5" }, "peerDependencies": { diff --git a/web/tests/unit/estimate-retry-after.test.ts b/web/tests/unit/estimate-retry-after.test.ts new file mode 100644 index 00000000..16cbbe77 --- /dev/null +++ b/web/tests/unit/estimate-retry-after.test.ts @@ -0,0 +1,57 @@ +import { + ARCHIVE_BUILD_RETRY_AFTER_MAX_SECONDS, + ARCHIVE_BUILD_RETRY_AFTER_SECONDS, + estimateRetryAfterSeconds, +} from "@/lib/api/run-archive"; +import { describe, expect, it } from "vitest"; + +describe("estimateRetryAfterSeconds", () => { + it("returns the floor for a tiny single-file run", () => { + expect(estimateRetryAfterSeconds({ fileCount: 1, totalBytes: 1024 })).toBe( + ARCHIVE_BUILD_RETRY_AFTER_SECONDS + ); + }); + + it("scales with file count for many small files", () => { + // 3000 tiny files: per-file term dominates (3 + 3000*0.005 = 18s); the + // ~6 MB byte term adds a fraction that rounds up to 19s. + const seconds = estimateRetryAfterSeconds({ + fileCount: 3000, + totalBytes: 3000 * 2 * 1024, + }); + expect(seconds).toBe(19); + }); + + it("scales with total bytes for a few large files", () => { + // ~10 GB across 3 files: throughput term (~51s) dominates and clamps to + // the cap. + const seconds = estimateRetryAfterSeconds({ + fileCount: 3, + totalBytes: 10 * 1024 * 1024 * 1024, + }); + expect(seconds).toBe(ARCHIVE_BUILD_RETRY_AFTER_MAX_SECONDS); + }); + + it("clamps to the ceiling for enormous runs", () => { + expect( + estimateRetryAfterSeconds({ + fileCount: 1_000_000, + totalBytes: Number.MAX_SAFE_INTEGER, + }) + ).toBe(ARCHIVE_BUILD_RETRY_AFTER_MAX_SECONDS); + }); + + it("treats zero bytes (all-NULL sizes) as a count-only estimate", () => { + // The bytes term vanishes; the per-file term still applies. 2000 files: + // 3 + 2000*0.005 = 13s. + expect(estimateRetryAfterSeconds({ fileCount: 2000, totalBytes: 0 })).toBe( + 13 + ); + }); + + it("never returns a value below the floor", () => { + expect( + estimateRetryAfterSeconds({ fileCount: 0, totalBytes: 0 }) + ).toBeGreaterThanOrEqual(ARCHIVE_BUILD_RETRY_AFTER_SECONDS); + }); +});