This is an automated email from the ASF dual-hosted git repository.
AlexStocks pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
The following commit(s) were added to refs/heads/develop by this push:
new 4879a3c41 fix(remoting): clean up pendingResponses on error paths to
prevent map leak (#3440)
4879a3c41 is described below
commit 4879a3c4111cd30f60116baa9607aafc6daabf42
Author: aias00 <[email protected]>
AuthorDate: Sat Jul 11 14:55:48 2026 +0800
fix(remoting): clean up pendingResponses on error paths to prevent map leak
(#3440)
* fix(remoting): clean up pendingResponses on error paths to prevent map
leak
The global pendingResponses sync.Map is cleaned up on the success path
via Response.Handle() -> removePendingResponse(), but entries were never
removed when requests fail:
- ExchangeClient.Request(): AddPendingResponse called before
client.Request(), but on error the entry was left in the map
- ExchangeClient.AsyncRequest(): same pattern, same leak
- getty heartbeat(): timeout branch never called removePendingResponse,
leaving the entry to accumulate indefinitely
Add removePendingResponse calls in all three error/timeout paths. Export
RemovePendingResponse so the getty package (external to remoting) can
call it from its heartbeat timeout handler.
* refactor(remoting): export removePendingResponse directly instead of
wrapper
Address review feedback: the extra RemovePendingResponse wrapper added no
value and was inconsistent with its siblings AddPendingResponse/
GetPendingResponse, which are exported directly. Rename
removePendingResponse
to the exported RemovePendingResponse and update all in-package call sites.
* test(remoting): add regression tests for pendingResponses cleanup on
error paths
Cover the map-leak fix with tests that fail if the cleanup is removed:
- ExchangeClient Request/AsyncRequest remove the pending response when the
underlying client.Request returns an error
- getty heartbeat removes the pending response on read timeout
* test(remoting): use any instead of interface{} in heartbeat test mock
---
remoting/exchange.go | 6 ++--
remoting/exchange_client.go | 2 ++
remoting/exchange_client_test.go | 58 +++++++++++++++++++++++++++++-
remoting/exchange_test.go | 4 +--
remoting/getty/heartbeat_test.go | 76 ++++++++++++++++++++++++++++++++++++++++
remoting/getty/listener.go | 1 +
6 files changed, 141 insertions(+), 6 deletions(-)
diff --git a/remoting/exchange.go b/remoting/exchange.go
index 60dd11b47..ccfcd8c31 100644
--- a/remoting/exchange.go
+++ b/remoting/exchange.go
@@ -93,7 +93,7 @@ func (response *Response) IsHeartbeat() bool {
}
func (response *Response) Handle() {
- pendingResponse := removePendingResponse(SequenceType(response.ID))
+ pendingResponse := RemovePendingResponse(SequenceType(response.ID))
if pendingResponse == nil {
logger.Errorf("[Remoting] failed to get pending response
context for response package %s", *response)
return
@@ -172,8 +172,8 @@ func AddPendingResponse(pr *PendingResponse) {
pendingResponses.Store(SequenceType(pr.seq), pr)
}
-// get and remove response
-func removePendingResponse(seq SequenceType) *PendingResponse {
+// RemovePendingResponse gets and removes the pending response for the given
sequence ID.
+func RemovePendingResponse(seq SequenceType) *PendingResponse {
if pendingResponses == nil {
return nil
}
diff --git a/remoting/exchange_client.go b/remoting/exchange_client.go
index a7cb53cf2..6de4833e3 100644
--- a/remoting/exchange_client.go
+++ b/remoting/exchange_client.go
@@ -132,6 +132,7 @@ func (client *ExchangeClient) Request(invocation
*base.Invocation, url *common.U
err := client.client.Request(request, timeout, rsp)
// request error
if err != nil {
+ RemovePendingResponse(SequenceType(request.ID))
res.Err = err
return err
}
@@ -165,6 +166,7 @@ func (client *ExchangeClient) AsyncRequest(invocation
*base.Invocation, url *com
err := client.client.Request(request, timeout, rsp)
if err != nil {
+ RemovePendingResponse(SequenceType(request.ID))
result.Err = err
return err
}
diff --git a/remoting/exchange_client_test.go b/remoting/exchange_client_test.go
index 08245e9ab..4a829c5d9 100644
--- a/remoting/exchange_client_test.go
+++ b/remoting/exchange_client_test.go
@@ -26,10 +26,14 @@ import (
import (
"github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
)
import (
"dubbo.apache.org/dubbo-go/v3/common"
+ "dubbo.apache.org/dubbo-go/v3/protocol/base"
+ "dubbo.apache.org/dubbo-go/v3/protocol/invocation"
+ "dubbo.apache.org/dubbo-go/v3/protocol/result"
)
type mockClient struct {
@@ -37,6 +41,7 @@ type mockClient struct {
available bool
connectErr error
connCount int
+ requestErr error
}
func (m *mockClient) SetExchangeClient(client *ExchangeClient) {}
@@ -44,7 +49,9 @@ func (m *mockClient) Close()
{}
func (m *mockClient) IsAvailable() bool { m.mu.Lock();
defer m.mu.Unlock(); return m.available }
func (m *mockClient) Request(request *Request, timeout time.Duration, response
*PendingResponse) error {
- return nil
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ return m.requestErr
}
func (m *mockClient) Connect(url *common.URL) error {
@@ -104,3 +111,52 @@ func TestExchangeClientIsAvailable(t *testing.T) {
m.mu.Unlock()
assert.False(t, ec.IsAvailable())
}
+
+// newTestInvocation builds a *base.Invocation usable by
ExchangeClient.Request/AsyncRequest.
+func newTestInvocation() *base.Invocation {
+ var inv base.Invocation = invocation.NewRPCInvocation("test", nil, nil)
+ return &inv
+}
+
+// TestExchangeClientRequestErrorCleanup is a regression test for the map leak
fix:
+// when the underlying client.Request returns an error, the pending response
for the
+// request ID must be removed from pendingResponses so the global map does not
leak.
+func TestExchangeClientRequestErrorCleanup(t *testing.T) {
+ m := &mockClient{available: true, requestErr: errors.New("request
failed")}
+ ec := NewExchangeClient(testURL(), m, 5*time.Second, true)
+
+ before := countPendingResponses()
+ res := &result.RPCResult{}
+ err := ec.Request(newTestInvocation(), testURL(), time.Second, res)
+
+ require.Error(t, err)
+ assert.Equal(t, err, res.Err)
+ // No pending response should be left behind on the error path.
+ assert.Equal(t, before, countPendingResponses(), "pendingResponses
leaked on Request error path")
+}
+
+// TestExchangeClientAsyncRequestErrorCleanup mirrors the sync case for
AsyncRequest.
+func TestExchangeClientAsyncRequestErrorCleanup(t *testing.T) {
+ m := &mockClient{available: true, requestErr: errors.New("request
failed")}
+ ec := NewExchangeClient(testURL(), m, 5*time.Second, true)
+
+ before := countPendingResponses()
+ res := &result.RPCResult{}
+ cb := func(response common.CallbackResponse) {}
+ err := ec.AsyncRequest(newTestInvocation(), testURL(), time.Second, cb,
res)
+
+ require.Error(t, err)
+ assert.Equal(t, err, res.Err)
+ assert.Equal(t, before, countPendingResponses(), "pendingResponses
leaked on AsyncRequest error path")
+}
+
+// countPendingResponses returns the number of entries currently held in the
global
+// pendingResponses map.
+func countPendingResponses() int {
+ n := 0
+ pendingResponses.Range(func(_, _ any) bool {
+ n++
+ return true
+ })
+ return n
+}
diff --git a/remoting/exchange_test.go b/remoting/exchange_test.go
index 92d7b6746..e12e0e8ca 100644
--- a/remoting/exchange_test.go
+++ b/remoting/exchange_test.go
@@ -106,8 +106,8 @@ func TestAddGetRemovePendingResponse(t *testing.T) {
pr := NewPendingResponse(999)
AddPendingResponse(pr)
assert.Equal(t, pr, GetPendingResponse(SequenceType(999)))
- assert.Equal(t, pr, removePendingResponse(SequenceType(999)))
- assert.Nil(t, removePendingResponse(SequenceType(999)))
+ assert.Equal(t, pr, RemovePendingResponse(SequenceType(999)))
+ assert.Nil(t, RemovePendingResponse(SequenceType(999)))
}
func TestResponseHandle(t *testing.T) {
diff --git a/remoting/getty/heartbeat_test.go b/remoting/getty/heartbeat_test.go
new file mode 100644
index 000000000..6e76b3e83
--- /dev/null
+++ b/remoting/getty/heartbeat_test.go
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 getty
+
+import (
+ "testing"
+ "time"
+)
+
+import (
+ getty "github.com/apache/dubbo-getty"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/remoting"
+)
+
+// mockHeartbeatSession embeds getty.Session so it satisfies the interface, and
+// overrides only WritePkg. It captures the ID of the heartbeat request that
was
+// written so the test can assert the pending response is cleaned up
afterwards.
+type mockHeartbeatSession struct {
+ getty.Session
+ writtenID int64
+}
+
+func (m *mockHeartbeatSession) WritePkg(pkg any, _ time.Duration) (int, int,
error) {
+ if req, ok := pkg.(*remoting.Request); ok {
+ m.writtenID = req.ID
+ }
+ // Return sendLen == 0 so heartbeat does not try to close the session.
+ return 0, 0, nil
+}
+
+// TestHeartbeatTimeoutCleanup is a regression test for the map leak fix: when
a
+// heartbeat times out (no response arrives), the pending response registered
for
+// the heartbeat request ID must be removed from remoting's pendingResponses
so the
+// global map does not leak.
+func TestHeartbeatTimeoutCleanup(t *testing.T) {
+ sess := &mockHeartbeatSession{}
+ done := make(chan error, 1)
+
+ err := heartbeat(sess, 10*time.Millisecond, func(err error) { done <-
err })
+ require.NoError(t, err)
+ require.NotZero(t, sess.writtenID, "heartbeat request should have been
written")
+
+ select {
+ case cbErr := <-done:
+ // The timeout branch is what performs the cleanup.
+ assert.Equal(t, errHeartbeatReadTimeout, cbErr)
+ case <-time.After(2 * time.Second):
+ t.Fatal("heartbeat callback was not invoked")
+ }
+
+ // After the timeout branch runs (which happens before the callback),
the pending
+ // response for the heartbeat request must have been removed.
+ assert.Nil(t,
remoting.GetPendingResponse(remoting.SequenceType(sess.writtenID)),
+ "pendingResponses leaked on heartbeat timeout path")
+}
diff --git a/remoting/getty/listener.go b/remoting/getty/listener.go
index ed3a69114..4964463aa 100644
--- a/remoting/getty/listener.go
+++ b/remoting/getty/listener.go
@@ -354,6 +354,7 @@ func heartbeat(session getty.Session, timeout
time.Duration, callBack func(err e
select {
case <-gxtime.After(timeout):
err1 = errHeartbeatReadTimeout
+
remoting.RemovePendingResponse(remoting.SequenceType(req.ID))
case <-resp.Done:
err1 = resp.Err
}
