feat(capabilities)!: Add streaming to http capabilities#2251
Conversation
bwoebi
changed the title
📚 Documentation Check Results📦
|
🎉 All green!🧪 All tests passed 🎯 Code Coverage (details) 🔗 Commit SHA: aa3098b | Docs | Datadog PR Page | Give us feedback! |
🔒 Cargo Deny Results📦
|
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 503b5ad27e
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| let result = tokio::time::timeout(Duration::from_millis(self.timeout_ms), async { | ||
| let response = future.await??; |
There was a problem hiding this comment.
When the endpoint stalls after the body is sent, this timeout returns an error but the inner value being timed out is a JoinHandle; dropping a Tokio JoinHandle detaches the task rather than cancelling it. In that scenario each timed-out debugger upload leaves the spawned HTTP request running in the background, holding its task/connection until the server eventually responds, so repeated timeouts can accumulate stuck work. Abort the handle on timeout or avoid spawning so the request future is actually cancelled.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
I think the Codex assessment is correct, even if the issue hasn't been introduced by the PR but seems to be pre-existing. I suppose we'd have to call abort on timeout to make sure the task doesn't run in the background.
There was a problem hiding this comment.
Yes, indeed. Had fixed it locally after codex complained, but forgot to commit.
| pub type ResponseFuture = | ||
| Pin<Box<dyn Future<Output = Result<http::Response<bytes::Bytes>, HttpError>> + Send>>; | ||
|
|
||
| pub type BodySender = Box<dyn StreamingBodySender + Send>; |
There was a problem hiding this comment.
MaybeSend for streaming capabilities
On wasm clients the HTTP capability is expected to wrap JS futures that are not Send, but these new public streaming aliases require both the response future and body sender to be Send. That means any wasm WasmCapabilities implementation cannot use or override the new streaming API despite MaybeSend existing specifically to keep these capability traits portable; use the conditional MaybeSend bound here as with request().
Useful? React with 👍 / 👎.
de3528f to
da204ee
Compare
BenchmarksComparisonBenchmark execution time: 2026-07-20 13:13:22 Comparing candidate commit aa3098b in PR branch Found 3 performance improvements and 10 performance regressions! Performance is the same for 129 metrics, 0 unstable metrics.
|
Artifact Size Benchmark Reportaarch64-alpine-linux-musl
aarch64-unknown-linux-gnu
libdatadog-x64-windows
libdatadog-x86-windows
x86_64-alpine-linux-musl
x86_64-unknown-linux-gnu
|
This allows moving live-debugger sender impl to use http-capabilities. Signed-off-by: Bob Weinand <bob.weinand@datadoghq.com>
da204ee to
bf5b07c
Compare
yannham
left a comment
yannham
left a comment
There was a problem hiding this comment.
I initially thought it could be separated out of the capabilities impl, but indeed implementing efficient streaming requires http-backend-specific tools. Plus the default impl makes it ok for WASM that doesn't have to implement it for now 👍
| Other(anyhow::Error), | ||
| } | ||
|
|
||
| pub type ChunkFuture<'a> = Pin<Box<dyn MaybeSendFuture<Result<(), HttpError>> + 'a>>; |
There was a problem hiding this comment.
Do we really need to spell this type out manually? I think this would be a use-case for async_trait that we use elsewhere in libdatadog (IIRC, the runtime result will be the same, but you just wouldn't have to write those types like ChunkFuture manually)
There was a problem hiding this comment.
I can't get the MaybeSend into the async_trait though, right? async_trait forces Pin<Box<dyn Future<T> + Send + 'a>>
There was a problem hiding this comment.
In fact async trait is probably not even needed. What I mean instead of defining ChunkFuture<'a> explicitly, couldn't you just define fn send_chunk(&mut self, data: bytes::Bytes) -> impl MaybeSendFuture<Result<(), HttpError>> + 'a? Maybe that doesn't work for some reason, but just trying to understand if spelling this type out is really needed.
There was a problem hiding this comment.
No, that's why async_trait exists in the first place, you can't do that on trait functions like that.
There was a problem hiding this comment.
Ah, because StreamingBodySender needs to be returned and thus put in a box as a dyn StreamingBodySender, ack 👍 (because you can definitely use an impl in return position in a trait without async_trait, that's what HttpClientCapability does for request, but then you can't use as a dyn Trait anymore...)
| .await | ||
| .map_err(|e| HttpClientError::IoError(e.to_string()))? | ||
| .to_bytes(); | ||
| let (status, headers, body_bytes) = tokio::time::timeout(timeout, async { |
There was a problem hiding this comment.
This change looks unrelated (or at least separate), since virtually no code is using libdd-http-client yet (beside Python is starting to do so). Is it to align with the behavior of NativeHttpClient? Also, shouldn't we also adapt the reqwest backend to behave the same?
There was a problem hiding this comment.
The reqwest backend already behaves this way, just that timeout is native to reqwest:
if let Some(timeout) = request.timeout {
builder = builder.timeout(timeout);
}
This aligns it with reqwest, where the timeout is end-to-end.
| let result = tokio::time::timeout(Duration::from_millis(self.timeout_ms), async { | ||
| let response = future.await??; |
There was a problem hiding this comment.
I think the Codex assessment is correct, even if the issue hasn't been introduced by the PR but seems to be pre-existing. I suppose we'd have to call abort on timeout to make sure the task doesn't run in the background.
| Other(anyhow::Error), | ||
| } | ||
|
|
||
| pub type ChunkFuture<'a> = Pin<Box<dyn MaybeSendFuture<Result<(), HttpError>> + 'a>>; |
There was a problem hiding this comment.
That's essentially a MaybeSendBoxFuture, right ? Do you think it'd be useful to have it in maybe_send.rs as MaybeSend and MaybeSendFuture ?
There was a problem hiding this comment.
I don't know... Let's add it once we have more usages? E.g. ChunkFuture and ResponseFuture are subtly different that the one has a lifetime and the other doesn't.
bwoebi
changed the title
Signed-off-by: Bob Weinand <bob.weinand@datadoghq.com>
3 participants
This allows moving live-debugger sender impl to use http-capabilities.