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 88.77% 88.96% +0.19%
==========================================
Files 108 110 +2
Lines 18116 18692 +576
==========================================
+ Hits 16082 16630 +548
- Misses 2034 2062 +28
☔ 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
lcian
changed the title
|
|
||
| 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 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?
There was a problem hiding this comment.
the special case here skips any executor/threading and just runs it on the caller thread. i don't think there is a "same thread executor" ready-made we could use here. but it probably wouldn't be hard to write
| # 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.
c481164 to
ff03dcd
Compare
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 3 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 ff03dcd. Configure here.
| ) | ||
| completed = False | ||
| seen: set[int] = set() | ||
| try: |
There was a problem hiding this comment.
Batch request errors drop operations
High Severity
session._pool.request sits outside the try that fans failures through _unanswered. A connection error, timeout, or other exception from request therefore never produces per-operation results. With concurrency=1 that exception escapes and aborts the rest of many(). Concurrently, run records a single ErrorResult with index=None, so every operation in that batch is dropped. Callers comparing result count to the input, or using raise_for_failures(), miss those failures.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit ff03dcd. Configure here.
| return None, ErrorResult( | ||
| None, | ||
| _malformed(f"missing or invalid {HEADER_BATCH_OPERATION_STATUS} header"), | ||
| ) |
There was a problem hiding this comment.
Malformed status double-reports operations
Medium Severity
When a response part has a valid x-sn-batch-operation-index but a missing or malformed status header, _result_from_part returns position=None and ErrorResult(index=None) even though the operation is already known. That position is not added to seen, so _unanswered later emits a second failure for the same operation. Result count no longer matches the input, and raise_for_failures() surfaces the same failure twice.
Reviewed by Cursor Bugbot for commit ff03dcd. Configure here.
| self._reader: ZstdCompressionReader | None = compressor.stream_reader( | ||
| source, | ||
| closefd=False, # Not our stream to close | ||
| ) |
There was a problem hiding this comment.
Compressors allocated for every part
Medium Severity
_ZstdBody constructs a ZstdCompressor and stream_reader in __init__, and _insert_part builds every part before the request is written. A batch of zstd puts therefore holds one compression context per part at once. The class later drops the reader to avoid that cost, but the contexts already exist. A full 1000-operation batch can retain on the order of a gigabyte of compressor state until send starts.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit ff03dcd. Configure here.
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.