feat(py-client): Implement "many" api for batch requests - #546
feat(py-client): Implement "many" api for batch requests#546matt-codecov wants to merge 1 commit into
Conversation
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #546 +/- ##
==========================================
+ Coverage 87.99% 88.23% +0.24%
==========================================
Files 96 98 +2
Lines 15956 16537 +581
==========================================
+ Hits 14041 14592 +551
- Misses 1915 1945 +30
☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
Please see also the now closed #419. It's gone out of sync with main, so it's better to carry on here. The prior PR focused on streaming as much as possible and had gone through a round of feedback with the Python SDK maintainers for the public API. Also, there's configuration for concurrency that uses an optional thread pool executor. I hope there are some parts we could leverage from that. |
|
bugbot run |
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
![sentry[bot]](https://avatars.githubusercontent.com/in/12637?s=60&v=4){kind=small}
| if key is None: | ||
| if is_error: | ||
| key = "<unknown>" |
There was a problem hiding this comment.
Bug: A successful keyless insert is incorrectly reported as an ErrorResult if the server returns a success status but omits the x-sn-batch-operation-key header.
Severity: MEDIUM
Suggested Fix
Make the client code more robust to server contract violations. For a successful response (2xx status) that is missing the x-sn-batch-operation-key header, consider returning a PutResult with a sentinel or None key instead of an ErrorResult. This would correctly represent the operation's success while indicating the key is unknown, preventing a successful write from being reported as a failure.
Prompt for AI Agent
Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.
Location: clients/python/src/objectstore_client/many.py#L525-L527
Potential issue: The client-side parsing logic for batch operations assumes that a
successful response for a keyless insert will always contain the
`x-sn-batch-operation-key` header with the server-generated key. If the server returns a
success status (e.g., 2xx) but omits this header due to a bug, proxy issue, or future
change, the client code incorrectly interprets this as a failure. Instead of returning a
`PutResult`, it constructs and returns an `ErrorResult` with a "missing header" message.
This misrepresents a successful data insertion as an error to the caller, creating a
discrepancy between the client's state and the actual state on the server.
Did we get this right? 👍 / 👎 to inform future reviews.
9ac7967 to
c481164
Compare
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 2 potential issues.
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Want reviews to match your repository better? Bugbot Learning can learn team-specific rules from PR activity. A team admin can enable Learning in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit c481164. Configure here.
|
|
||
| def _make_batch_url(self) -> str: | ||
| relative_path = f"/v1/objects:batch/{self._usecase.name}/{self._scope}/" | ||
| return self._base_path.rstrip("/") + relative_path |
There was a problem hiding this comment.
Batch URL skips path encoding
Medium Severity
_make_batch_url builds the path without utils.encode_path, unlike _make_url and _make_multipart_url. Usecase names or base paths with spaces or other non-safe characters produce an invalid batch URL while single-object calls still work, so session.many() fails for scopes that the rest of the client already supports.
Reviewed by Cursor Bugbot for commit c481164. Configure here.
| # clean everything up. | ||
| cancelled.set() | ||
| if own_executor: | ||
| pool.shutdown(wait=False, cancel_futures=True) |
There was a problem hiding this comment.
Abandoned executor work still runs
Medium Severity
On abandon, cancelled is set but pending futures are only cancelled when session.many() owns the executor. With a caller-supplied executor, submitted work that has not started still runs to completion (including puts and deletes) even though docs say undispatched operations are cancelled and results are never delivered.
Reviewed by Cursor Bugbot for commit c481164. Configure here.
| concurrently and each operation's relative order is undefined. To minimize the | ||
| likelihood of racing operations on the same key, the client will separate | ||
| same-key operations into different batches if at least one of the operations is | ||
| a write or delete. However, with a `concurrency` value larger than `1`, it is |
There was a problem hiding this comment.
I don't think there's a need to separate operations on the same key into different batches, as the result will be non-deterministic anyways when concurrency > 1.
There was a problem hiding this comment.
this idea was taken from #419, except that PR only applied it when concurrency=1. this PR applies it for all values of concurrency. if the splitting logic is written anyway, why not apply it on all paths to at least make the race less likely?
as for whether it's necessary: tbh i'm ambivalent. if i were using the objectstore client, i'd probably take care to keep only the last PUT/DELETE operation (last write wins) and drop any post-write GET/HEAD operations because we already know what they'll return (unless clients expect an out-of-band racing write operation lol). but there is a case one may want to express with a single .many() today that would need this fix: a GET /foo/bar to read the current value before a PUT /foo/bar to replace the value. a non-atomic swap.
this fix and concurrency=1 are needed for that case to work right with a single session.many() call. but as a workaround you can do a session.many() with all of your GETs and then a second session.many() with all the PUTs. that eliminates any racing (as far as this client goes) and doesn't even require concurrency=1. it might be better to do it that way anyway.
so i think i've talked myself into agreeing with you. @jan-auer is there anything else to consider about this or shall i remove this behavior?
lcian
changed the title
| concurrency: The maximum number of requests in flight. Defaults to | ||
| ``1``, which runs everything sequentially on the calling thread, | ||
| without a thread pool. Raising it is sufficient to send requests | ||
| concurrently, but each request opens/closes its own connection | ||
| without additional configuration on the :class:`Client` (a | ||
| ``maxsize`` key in the ``connection_kwargs`` dict to control | ||
| ``urllib3`` connection pool size). |
There was a problem hiding this comment.
In the Rust client, by default, we send the requests with some degree of concurrency.
I would probably do that here too (so, default to the internal threadpool and concurrency > 1), probably with the same defaults as Rust.
Unless you have strong reasons to avoid that, it seems better UX to just handle this internally in the client rather than having the user necessarily think about this.
| @@ -0,0 +1,971 @@ | |||
| """ | |||
| Batch operations API for executing multiple get/put/delete operations. | |||
There was a problem hiding this comment.
| Batch operations API for executing multiple get/put/delete operations. | |
| Batch operations API for executing multiple operations. |
This should/will support also HEAD and whatever other ops we come up with in the future
| # pool's `maxsize` still go out (urllib3 opens a connection on demand and caps | ||
| # only how many it keeps idle) but every connection over that limit is closed |
There was a problem hiding this comment.
urllib3 opens a connection on demand and caps only how many it keeps idle
This is true unless you configure the pool with block=True (https://urllib3.readthedocs.io/en/stable/advanced-usage.html).
Maybe we should avoid going so much into such details in these comments, I think what we have in the README and docstring on the API is enough as it also talks about this.
|
|
||
| A port of the ``ZSTD_COMPRESSBOUND`` macro definition in ``zstd.h``. | ||
| """ | ||
| margin = ((128 << 10) - size) >> 11 if size < (128 << 10) else 0 |
There was a problem hiding this comment.
https://github.com/facebook/zstd/blob/82d322c4973d9e2968d94047a40892bc6d9a9bdf/lib/zstd.h#L249 is the source. it is a bit of a mess to parse but here it is with some extra indentation for clarity:
#define ZSTD_COMPRESSBOUND(srcSize) \
( \
((size_t)(srcSize) >= ZSTD_MAX_INPUT_SIZE) \
? 0 \
: (srcSize) + ((srcSize)>>8) + ( /* size + (size >> 8) + */ \
((srcSize) < (128<<10)) /* if size < (128 << 10) */ \
? (((128<<10) - (srcSize)) >> 11) /* then ((128 << 10) - size) >> 11 */ \
: 0 /* else 0 */ \
) \
)once upon a time writing convoluted preprocessor macros was my job
| # If this op `_conflicts()` with an op in the current batch, or we've | ||
| # hit a batch size/length limit, cut the batch off here and yield it. | ||
| if batch and ( | ||
| _conflicts(op, keys) |
There was a problem hiding this comment.
As I said previously, I would just avoid this logic and complexity, if the user requests conflicting operations it's their fault, our contract is clear on the fact that the ops can be executed in arbitrary order.
| if concurrency == 1 and executor is None: | ||
| for item in work: | ||
| yield from _run_work(session, item) | ||
| else: | ||
| yield from _execute_concurrent(session, work, concurrency, executor) |
There was a problem hiding this comment.
Do we really need the special case for concurrency=1 or could we just send the ops into _execute_concurrent with a special/single-threaded executor and that's it?
| # part-way through the batch response. | ||
| work_results.close() | ||
| except Exception as error: | ||
| put(ErrorResult(None, error)) |
There was a problem hiding this comment.
My clanker says:
- [P2] Preserve known indexes for malformed status headers — /Users/lorenzotmp/repos/objectstore/clients/python/src/objectstore_client/many.py:876-880
When a response part has a valid operation-index header but a missing or malformed status header, this returns position=None and ErrorResult(index=None) even though
the operation is known. The position is therefore not added to seen, and _unanswered later emits a second failure for the same operation, violating the one-result-per-
operation contract and discarding useful failure context.
This seems legit. If you get a 500 for an intermediate proxy for instance, this would happen.
We can probably avoid this by doing a better check on the response headers.
| # Measures batch processing latency. Note that this includes the time the | ||
| # caller spends consuming the results iterator and not just client/server | ||
| # latency. | ||
| with measure_storage_operation( |
There was a problem hiding this comment.
This is a bit weird for a user because a user doesn't think in terms of "batch" calls but rather in terms of "many" calls and/or individual ops.
But I also don't have any better approach to suggest here.
| Passes a result through the results queue out to the caller. Returns | ||
| ``False`` if we were told to exit. | ||
| """ | ||
| while not (cancelled.is_set() or _SHUTTING_DOWN.is_set()): |
There was a problem hiding this comment.
Is this the only place where we need to check _SHUTTING_DOWN?
(I would be happy if we could avoid that altogether but apparently not using it could cause the process to hang)
3 participants
#277 / #478 implemented the
manyAPI in the Rust client which uses Objectstore's batch endpoint. This PR ports it over to Python.Incorporates direction from #419.
Closes FS-330
Some notes:
ThreadPoolExecutor'smax_workersto allow for results to be streamed from individual batch requests without having to buffer the whole response. Otherwise a batch of 1000 5MB GET results would eat 5GB RAM and delay yielding anything to the caller.Get/Putoperation type classes copy the arg list of theget()andput()methods onsession, the decompression code inget()is copied... but I didn't want to touch existing code much to reorganize in this PR.1because that's the default urllib3 connection pool size. You can still send concurrent requests with a connection pool size of 1, it just opens/closes a connection per request and logs a warning about it instead of actually pooling.