[sergiopeixoto-seequent]

Description

Typed-object creation and replacement uploaded independent tables, attributes, and collections sequentially, making large uploads slow. Some mutations also changed the in-memory object before every upload had succeeded, risking partially updated state on failure.

The branch parallelizes independent uploads with a shared concurrency limit, preserves input order, retries transient upload failures, and commits mutations only after all required uploads succeed. It adds focused tests for concurrency, retries, cancellation, ordering, and mutation atomicity.

Checklist

• I have read the contributing guide and the code of conduct

[daniel-kinney-seequent]

You might want to check if the uploads are being rate limited as much as you wanted, because it looks like each upload_from_cache() allows for ``min(32, os.cpu_count() + 4)` workers.

[sergiopeixoto-seequent]

The semaphore limits eight complete table pipelines, including Parquet writing, but it does not impose an eight-chunk limit. Each active table inherits the common uploader’s min(32, os.cpu_count() + 4) chunk limit. With sufficiently large files, that can allow up to 8 × 32 = 256 chunk transfers to be reading data or awaiting transfer.

I also checked the transport: the default shared AioTransport limits actual HTTP connections to four, so those 256 chunk slots do not mean 256 simultaneous HTTP requests. However, that connection limit is configurable, and chunks can already be buffered while waiting for a connection. It therefore doesn’t eliminate the aggregate memory/concurrency concern.

The intended limit in this PR is eight table pipelines, not eight storage requests. Retry backoff handles transient failures but does not establish a shared chunk budget. I would keep the existing chunk behavior for that documented scope; if we want a strict overall storage-transfer limit, we need to implement that explicitly rather than rely on CPU count or retries.

[daniel-kinney-seequent]

I thought it was a bit confusing to have two semaphores. (It's not a request for change - I just had a quick look out of interest and left my 2c)

[sergiopeixoto-seequent]

They’re alternatives, not two locks acquired together. get_data_client() creates separate client instances, so a per-client semaphore alone would let each client start eight uploads independently.

The context semaphore keeps all nested serialization under one shared eight-table limit.

Outside that serialization scope, the client semaphore bounds directly concurrent upload_table() calls. The or selects which scope applies.

[aminabedi-bentley]

This is a valuable improvement for all compute tasks that upload large data. We observed multi-minute client-side upload spans in conditional simulation, while the corresponding bulk-upload burst in the logs completed in roughly 34 seconds. Investigating the current SDK showed synchronous Parquet/gzip serialization and hashing inside upload_table which blocks the event loop. We cannot attribute every second of the delay to that without telemetry instrumentation - which the SDK (not this PR) currently lacks, but moving this work off-thread directly addresses the bottleneck we have also identified.

My comments fwiw:
👍 Bounding whole serialization/upload pipelines enables CPU/network overlap without allowing unlimited concurrent serialization.
👍 The cancellation handling that waits for background writers is particularly important for task-scoped caches.

❓ Could we make the _MAX_CONCURRENT_UPLOADS configurable, retaining eight as the default and applying the setting consistently to both shared-scope and per-client semaphores? Each slot covers serialization through upload, so suitable concurrency depends on table sizes and available CPU/memory. Configurability would let clients (including compute tasks) tune this accordingly without changing SDK internals.

Out of the scope of this PR, but since I suppose both our teams are looking for the same outcome:
There is a change needed in core-compute-tasks framework to adopt this properly. The framework currently creates a fresh  ObjectAPIClient / ObjectDataClient  for every attribute upload, and the adapter uses ordinary asyncio.gather , outside the new shared upload scope. Consequently, each upload would receive its own per-client semaphore rather than sharing the intended limit.

We should reuse data clients per workspace within the task’s EvoServices lifetime, retaining reference validation and the shared connector/cache. This PR effectively elevates the client from a lightweight per-call wrapper to an owner of concurrency state, so its lifecycle now matters. This is a framework integration follow-up rather than a blocker for this SDK improvement. FYI @BenLewis-Seequent