Repository navigation
Conversation
|
Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA). View this failed invocation of the CLA check for more information. For the most up to date status, view the checks section at the bottom of the pull request. |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a custom libp2p HTTP transport wrapper (newLibp2pHTTPTransport) in internal/node/libp2p_transport.go that ensures underlying libp2p streams are reset when the request context is cancelled. This prevents orphaned streams on remote handlers during long or streaming requests. The new transport is integrated into openai_facade.go and sidecar.go to replace the default libp2phttp.NewTransport. Unit tests are also added in libp2p_transport_test.go to verify stream resetting and persistence behaviors. I have no feedback to provide.
go-libp2p-http uses the request context only to open the stream. When a caller cancels a long or streaming request through the egress proxy, the stream and the remote handler behind it stay open until the remote side finishes on its own. Wrap the host passed to the transport so that each stream opened for a request is reset when that request's context ends, and use it for the egress proxy and the OpenAI facade.
601fd9d to
c9f850f
Compare
go-libp2p-http's RoundTripper uses the request context only to open the stream. After that, cancelling the request does nothing: the stream and the remote handler behind it stay open until the remote side finishes on its own. Through the egress proxy this means a client that disconnects from a long or streaming response (SSE, a streaming A2A task, a streamed completion) leaves the remote peer's handler running and the stream open.
This wraps the host passed to libp2phttp.NewTransport so that NewStream registers context.AfterFunc(ctx, stream.Reset) for the request's context, and stops it when the stream is closed or reset normally. The egress proxy and the OpenAI facade use the wrapped transport.