MultipartUploader#

class nominal.experimental.ingest.MultipartUploader(timeout: float, max_part_retries: int)#

Bases: object

Uploads many files via driver-per-file structured concurrency over three pools.

Enqueue files with enqueue_file; track completion via the returned futures. Each file’s whole lifecycle is collapsed behind a single Future[str] that resolves to the object’s storage location or raises.

Work is split across a small pool (one single-shot upload_file task per small-route file), a driver pool (one sequential driver per multipart file, which is also the in-flight bound on open multipart uploads), and a part pool (one sign+PUT task per part, submitted only by running drivers). Neither route can starve the other’s workers in any batch shape.

Wait DAG (deadlock-freedom by inspection): small tasks and part tasks wait only on the gate; drivers wait on part futures and the gate; the gate waits only on time.

small_file_route_max_bytes (EXPERIMENTAL) routes files at/below that size through the backend single-shot upload_file endpoint (one request) rather than multipart, collapsing the per-file metadata that rate-limits many-small-file batches. Larger files use multipart.

Tuning. The knobs compose around one budget and one pipe. Every Nominal API request (initiate/sign/complete/upload_file) competes for the server’s per-caller admission budget through the max_nominal_concurrency lane, while file bytes flow directly to storage over max_storage_workers concurrent PUT streams. Which of the two binds depends on the batch shape:

  • Many small files — the admission budget binds long before bandwidth does. The single-shot route is the big lever: one request per file instead of multipart’s three. Widening the lane does NOT finish the batch sooner once the budget is the wall — it only provokes refusals, and a refused single-shot upload re-sends its whole body, so wire traffic grows with no time won. The offered request rate is roughly lane width divided by request latency; if uploads show heavy throttling (long per-request tails), narrow the lane rather than widening it.

  • A few huge files — bandwidth binds and the lane sits idle. Aggregate throughput is roughly streams x per-stream throughput, but only until the network path saturates; past saturation, extra streams degrade each other and aggregate can FALL. Find the knee empirically: measure at the default max_storage_workers, then halve and double it and keep whichever is faster. part_size (on enqueue_file) sets how many streams one file can occupy — keep files at roughly 3+ parts so streams stay fed near file boundaries — and max_files_in_flight trades a flatter finishing tail (lower) against more interleaving (higher).

  • Memory — each PUT stream holds one full part in memory, so peak buffered memory is roughly max_storage_workers x part_size (~640 MiB at the defaults). Shrink either to fit a constrained host.

  • Unreliable networks — file_retry_timeout (default one hour) rides out outages by retrying whole files whose failures look like network weather; timeout and max_part_retries bound each hung PUT attempt; a smaller part_size makes every retry cheaper at the cost of more requests.

When in doubt, keep the defaults — they are benchmark-tuned, and the classic mistake is raising everything at once, which slows BOTH regimes: streams past saturation contend with each other, and lane width past the admission budget only converts useful bandwidth into refused-and-resent requests.

timeout: float#
max_part_retries: int#
close(*, cancel_pending: bool = False) → None#

Shut the uploader down; afterwards enqueue_file raises.

Either way, every future enqueue_file has returned is settled by the time this returns — dropped files with CancelledError — and is safe to pass to concurrent.futures.wait / as_completed from any thread.

Parameters:

cancel_pending – If false (the default), every enqueued file runs to completion before this returns. If true, queued files are dropped and in-flight multipart files are cut short at their next part boundary, so the wait is bounded by one round of in-flight uploads plus the capped abort pass — not by the rest of the batch.

classmethod create(client: NominalClient, *, max_storage_workers: int | None = None, max_small_file_workers: int | None = None, max_files_in_flight: int | None = None, small_file_route_max_bytes: int | None = DEFAULT_SMALL_FILE_ROUTE_MAX_BYTES, max_nominal_concurrency: int = NOMINAL_MAX_CONCURRENCY, timeout: float = 30.0, max_part_retries: int = 3, per_request_retry_timeout: float = DEFAULT_THROTTLE_DEADLINE_S, max_backoff_duration: float = DEFAULT_MAX_BACKOFF_DURATION_S, file_retry_timeout: float | None = DEFAULT_FILE_RETRY_TIMEOUT_S) → Self#

Create a MultipartUploader sized for one batch of uploads.

Parameters:

  • client – The NominalClient to upload with. Uploads land in the client’s configured workspace, and requests ride the client’s shared transport, whose short jittered throttle retries are the local layer of the two-layer backoff. The uploader never closes anything it borrows from the client.

  • max_storage_workers – Part-pool size — the number of concurrent direct-to-storage PUT streams. Defaults to DEFAULT_MAX_STORAGE_WORKERS (benchmark-tuned). Raising it may degrade throughput rather than improve it: past the network path’s saturation point, streams contend with each other. Raise it when the path to storage is measured to reward more concurrent streams — and tune it together with enqueue_file’s part_size: each stream holds one full part in memory, so peak buffered upload memory is roughly max_storage_workers × part_size (~640 MiB at the defaults; 40 workers at the default part size is ~2.5 GiB).

  • max_small_file_workers – Small-pool size. Defaults to max_nominal_concurrency — small files spend their whole task inside the nominal lane, so extra threads beyond the lane width would only queue at its entrance.

  • max_files_in_flight – Driver-pool size, which is the bound on concurrently open multipart uploads. Defaults to half of max_storage_workers (rounded up), which keeps the part pool fed without piling up a hard-to-drain tail of half-done files; there is no unbounded mode.

  • small_file_route_max_bytes – EXPERIMENTAL. Files whose size is <= this many bytes are uploaded single-shot via the backend upload_file endpoint (one request) instead of multipart — avoiding the per-file metadata burst that rate-limits many-small-file batches. Defaults to DEFAULT_SMALL_FILE_ROUTE_MAX_BYTES; guarded to MAX_SMALL_FILE_ROUTE_BYTES. Pass None to disable the route and send every file through multipart.

  • max_nominal_concurrency – Width of the nominal lane — the maximum number of Nominal API requests in flight at once, across every pool. Benchmark-tuned to fill the server’s admission budget without provoking refusals; see NOMINAL_MAX_CONCURRENCY.

  • timeout – Per-request timeout for direct-to-storage part PUTs.

  • max_part_retries – Attempts per part before the file fails.

  • per_request_retry_timeout – Wall-clock budget, per request, for retrying while the server is throttling. Throttling is not a terminal error inside that budget; exceeding it raises NominalRequestThrottledError. Non-throttle errors always fail immediately.

  • max_backoff_duration – Cap on the shared storm damper’s delay. Under sustained throttling the damper doubles toward this cap and every retry sleeps a jittered fraction of it; successes decay it back to zero.

  • file_retry_timeout – Wall-clock budget, per file, for retrying TRANSIENT failures — connection errors, timeouts, throttle-budget exhaustion, HTTP 408/429/5xx — by re-running the file’s whole upload after a jittered exponential backoff. Sized for network weather: a wifi drop mid-batch pauses the affected files rather than failing them. Permanent failures (a broken file, other 4xx, unknown errors) never retry and surface immediately regardless of this value. Pass None to disable file-level retry.

Returns:

An uploader ready to accept files. Close it (or use it as a context manager) to release its threads and sessions.

Raises:

ValueError – A pool size is not positive, or small_file_route_max_bytes is outside (0, MAX_SMALL_FILE_ROUTE_BYTES].

enqueue_file(path: Path, *, file_type: FileType | None = None, name: str | None = None, part_size: int = DEFAULT_CHUNK_SIZE) → Future[str]#

Schedule a file upload and return a future for its storage location.

Never blocks: the file is validated on the calling thread and then queued on its route’s pool. The returned future is the executor’s own, so cancelling a file that has not started yet genuinely dequeues it (and it never initiates anything); cancelling a running file returns False. The future stays safe to wait on across a cancelling close: close settles every future it drops (see close).

Parameters:

  • path – File to upload.

  • file_type – Override the file type inferred from path.

  • name – Override the filename derived from path. An existing extension is preserved; only names without one use the default extension from file_type.

  • part_size – Bytes per multipart part. Ignored on the small-file route. Each in-flight part is held fully in memory, so peak buffered memory across the uploader is roughly max_storage_workers × part_size — tune the two together.

Returns:

A future resolving to the uploaded object’s location.

Raises:

  • RuntimeError – The uploader is closed.

  • FileNotFoundError – path does not exist.

  • ValueError – An applied default extension is malformed, the upload name is unsafe for storage, or — multipart route only — part_size is not positive or would need more parts than the storage provider allows. A multipart part_size below the provider’s non-final-part minimum is permitted (deliberately small parts can suit a lossy network) but logged as a warning, since the provider may reject the upload at completion.