From 14dd88956cf0dd1ed1e7abe77a969d36078eeafe Mon Sep 17 00:00:00 2001 From: Artur Maciag Date: Fri, 13 Nov 2026 20:10:43 +0000 Subject: [PATCH 1/2] node: reset libp2p HTTP streams when the request ends 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. --- internal/node/libp2p_transport.go | 70 +++++++++++++++ internal/node/libp2p_transport_test.go | 116 +++++++++++++++++++++++++ internal/node/openai_facade.go | 3 +- internal/node/sidecar.go | 3 +- 4 files changed, 188 insertions(+), 4 deletions(-) create mode 100644 internal/node/libp2p_transport.go create mode 100644 internal/node/libp2p_transport_test.go diff --git a/internal/node/libp2p_transport.go b/internal/node/libp2p_transport.go new file mode 100644 index 00000000..defcdf0c --- /dev/null +++ b/internal/node/libp2p_transport.go @@ -0,0 +1,70 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package node + +import ( + "context" + + libp2phttp "github.com/libp2p/go-libp2p-http" + "github.com/libp2p/go-libp2p/core/host" + "github.com/libp2p/go-libp2p/core/network" + "github.com/libp2p/go-libp2p/core/peer" + "github.com/libp2p/go-libp2p/core/protocol" +) + +// newLibp2pHTTPTransport returns the go-libp2p-http transport over h, with +// each request's stream reset when the request's context ends. +// +// go-libp2p-http uses the request context only to open the stream. A caller +// that cancels a long or streaming request would otherwise leave the stream, +// and the remote handler behind it, open until the remote side finishes. +func newLibp2pHTTPTransport(h host.Host) *libp2phttp.RoundTripper { + return libp2phttp.NewTransport(cancelableStreamHost{h}) +} + +// cancelableStreamHost resets a stream opened by NewStream when the context +// passed to NewStream ends before the stream is closed. +type cancelableStreamHost struct { + host.Host +} + +func (h cancelableStreamHost) NewStream(ctx context.Context, p peer.ID, pids ...protocol.ID) (network.Stream, error) { + s, err := h.Host.NewStream(ctx, p, pids...) + if err != nil { + return nil, err + } + stop := context.AfterFunc(ctx, func() { _ = s.Reset() }) + return &cancelableStream{Stream: s, stop: stop}, nil +} + +type cancelableStream struct { + network.Stream + stop func() bool +} + +func (s *cancelableStream) Close() error { + s.stop() + return s.Stream.Close() +} + +func (s *cancelableStream) Reset() error { + s.stop() + return s.Stream.Reset() +} + +func (s *cancelableStream) ResetWithError(code network.StreamErrorCode) error { + s.stop() + return s.Stream.ResetWithError(code) +} diff --git a/internal/node/libp2p_transport_test.go b/internal/node/libp2p_transport_test.go new file mode 100644 index 00000000..15bde96e --- /dev/null +++ b/internal/node/libp2p_transport_test.go @@ -0,0 +1,116 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package node + +import ( + "bufio" + "context" + "net/http" + "testing" + "time" + + "github.com/libp2p/go-libp2p" + gostream "github.com/libp2p/go-libp2p-gostream" + libp2phttp "github.com/libp2p/go-libp2p-http" + "github.com/libp2p/go-libp2p/core/host" + "github.com/libp2p/go-libp2p/core/peer" +) + +// streamingPeers starts a server peer whose /stream handler sends one line and +// then waits for its request to end, reporting that on the returned channel. +func streamingPeers(t *testing.T) (client, server host.Host, ended <-chan struct{}) { + t.Helper() + newHost := func() host.Host { + h, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = h.Close() }) + return h + } + server, client = newHost(), newHost() + + listener, err := gostream.Listen(server, libp2phttp.DefaultP2PProtocol) + if err != nil { + t.Fatal(err) + } + done := make(chan struct{}) + mux := http.NewServeMux() + mux.HandleFunc("/stream", func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte("first\n")) + w.(http.Flusher).Flush() + select { + case <-r.Context().Done(): + close(done) + case <-time.After(30 * time.Second): + } + }) + srv := &http.Server{Handler: mux} + go func() { _ = srv.Serve(listener) }() + t.Cleanup(func() { _ = srv.Close() }) + + if err := client.Connect(context.Background(), peer.AddrInfo{ID: server.ID(), Addrs: server.Addrs()}); err != nil { + t.Fatal(err) + } + return client, server, done +} + +func TestLibp2pHTTPTransportResetsStreamOnCancel(t *testing.T) { + client, server, ended := streamingPeers(t) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, "libp2p://"+server.ID().String()+"/stream", nil) + if err != nil { + t.Fatal(err) + } + resp, err := newLibp2pHTTPTransport(client).RoundTrip(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if line, err := bufio.NewReader(resp.Body).ReadString('\n'); err != nil || line != "first\n" { + t.Fatalf("first line = %q, %v", line, err) + } + + cancel() + select { + case <-ended: + case <-time.After(5 * time.Second): + t.Fatal("remote handler still running 5s after the caller cancelled") + } +} + +func TestLibp2pHTTPTransportKeepsStreamUntilCancel(t *testing.T) { + client, server, ended := streamingPeers(t) + + req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, "libp2p://"+server.ID().String()+"/stream", nil) + if err != nil { + t.Fatal(err) + } + resp, err := newLibp2pHTTPTransport(client).RoundTrip(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if line, err := bufio.NewReader(resp.Body).ReadString('\n'); err != nil || line != "first\n" { + t.Fatalf("first line = %q, %v", line, err) + } + select { + case <-ended: + t.Fatal("remote handler ended although the caller kept the request open") + case <-time.After(500 * time.Millisecond): + } +} diff --git a/internal/node/openai_facade.go b/internal/node/openai_facade.go index 52c8da7e..a560f2b1 100644 --- a/internal/node/openai_facade.go +++ b/internal/node/openai_facade.go @@ -29,7 +29,6 @@ import ( "time" "github.com/google/agentmesh/api" - libp2phttp "github.com/libp2p/go-libp2p-http" "github.com/libp2p/go-libp2p/core/network" "github.com/libp2p/go-libp2p/core/peer" ) @@ -103,7 +102,7 @@ type openAIFacade struct { } func newOpenAIFacade(node *AgentMeshNode, egress http.Handler) *openAIFacade { - client := &http.Client{Transport: libp2phttp.NewTransport(node.Host)} + client := &http.Client{Transport: newLibp2pHTTPTransport(node.Host)} return &openAIFacade{ forward: egress, ttl: modelRegistryTTL, diff --git a/internal/node/sidecar.go b/internal/node/sidecar.go index 0bca8f55..5e25a777 100644 --- a/internal/node/sidecar.go +++ b/internal/node/sidecar.go @@ -34,7 +34,6 @@ import ( "time" "github.com/google/agentmesh/api" - libp2phttp "github.com/libp2p/go-libp2p-http" "github.com/libp2p/go-libp2p/core/network" "github.com/libp2p/go-libp2p/core/peer" ) @@ -714,7 +713,7 @@ func (node *AgentMeshNode) prepareEgressPeer(ctx context.Context, peerID string) } func createEgressProxy(node *AgentMeshNode) http.Handler { - transport := libp2phttp.NewTransport(node.Host) + transport := newLibp2pHTTPTransport(node.Host) proxy := &httputil.ReverseProxy{ Rewrite: func(pr *httputil.ProxyRequest) { From c9f850fa28375b1888394da24f44b088d7557c0e Mon Sep 17 00:00:00 2001 From: Artur Maciag Date: Fri, 13 Nov 2026 20:10:43 +0000 Subject: [PATCH 2/2] node: check response body close errors in transport tests --- internal/node/libp2p_transport_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/node/libp2p_transport_test.go b/internal/node/libp2p_transport_test.go index 15bde96e..5dc14a1b 100644 --- a/internal/node/libp2p_transport_test.go +++ b/internal/node/libp2p_transport_test.go @@ -80,7 +80,7 @@ func TestLibp2pHTTPTransportResetsStreamOnCancel(t *testing.T) { if err != nil { t.Fatal(err) } - defer resp.Body.Close() + defer func() { _ = resp.Body.Close() }() if line, err := bufio.NewReader(resp.Body).ReadString('\n'); err != nil || line != "first\n" { t.Fatalf("first line = %q, %v", line, err) } @@ -104,7 +104,7 @@ func TestLibp2pHTTPTransportKeepsStreamUntilCancel(t *testing.T) { if err != nil { t.Fatal(err) } - defer resp.Body.Close() + defer func() { _ = resp.Body.Close() }() if line, err := bufio.NewReader(resp.Body).ReadString('\n'); err != nil || line != "first\n" { t.Fatalf("first line = %q, %v", line, err) }