You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@seata.apache.org by te...@apache.org on 2026/10/24 08:47:17 UTC
(incubator-seata-go) branch master updated: fix(remoting): give message futures one owner and stop duplicate responses blocking the receive loop (#1163)
This is an automated email from the ASF dual-hosted git repository.
thunguo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-seata-go.git
The following commit(s) were added to refs/heads/master by this push:
new 7881d841 fix(remoting): give message futures one owner and stop duplicate responses blocking the receive loop (#1163)
7881d841 is described below
commit 7881d8410365cfeb03f32cdb5e4b6bcbec990ed8
Author: Gust wang <15...@users.noreply.github.com>
AuthorDate: Sat Oct 24 04:47:12 2026 -0400
fix(remoting): give message futures one owner and stop duplicate responses blocking the receive loop (#1163)
* fix(remoting): give message futures one owner and stop duplicate responses blocking the receive loop
A synchronous request that times out removed its entry from mergeMsgMap, but
the future had been stored in futures, so nothing ever removed it. Every
timeout keeps one future and its request message for the lifetime of the
process. SendAsyncResponse leaves one behind too: it sends with no callback,
and the TC does not reply to a reply, so nothing ever comes back to clear it.
The waiter now owns the cleanup. syncCallback removes the future on both of
its exits, and sendAsync removes it on the paths that end there, which are an
encode failure, a send failure, and a send with no callback to wait for it.
The callback path is left to syncCallback because an asynchronous callback
returns straight away and the wait happens in the goroutine it starts. The
delete is idempotent, so it does not conflict with the response processors
that already remove a future once a reply arrives. The heartbeat transfer
also sends without a callback; its processor already cleared the future when
the pong arrived, so this only makes that unconditional.
Removing the mergeMsgMap delete from syncCallback loses nothing: no
production code stores into that map.
Two related defects on the same paths:
- Encode asserted msg.Body.(proto.Message) without the comma-ok form, so a
body of the wrong type panicked inside sendAsync with the future already
stored, rather than returning an error its only caller already handles.
- gRPC NotifyRpcMessageResponse sent on MessageFuture.Done with no default
case. Done carries a single buffered signal, so a duplicate or late
response blocked the transport receive loop permanently; getty already
guarded this. Both merged-response branches of the client response
processor have the same unguarded send. They are unreachable today, since
nothing populates mergeMsgMap, and are changed for consistency. All four
completion sites now go through MessageFuture.Complete, which records the
response, wakes the waiter and reports whether it signaled.
RpcRequestTimeout keeps its value; the clients now read an unexported
variable initialised from it so tests no longer sit on the real 20s deadline.
Test_syncCallback drops from 20s to milliseconds because of it.
Fixes #1159
* fix(message): make MessageFuture.Complete safe under a concurrent duplicate
Complete wrote Response before attempting the Done send, and a dropped
duplicate (the `default` branch) still ran that write. Both transports
decode each incoming message in its own goroutine, so two responses for
the same ID can call Complete concurrently — not just sequentially, as the
existing "late response" tests assumed. That is an unsynchronized
concurrent write to Response, a real data race, and whichever call's write
happened to land last silently won regardless of which call actually
signaled Done.
Complete now gates on a CompareAndSwap over a new completed field: only the
caller that wins it writes Response, and it does so before the now
unconditional send on Done, which is safe because at most one goroutine
ever passes the gate for a given future's lifetime. A reproduction test that
runs 200 rounds of two goroutines calling Complete concurrently confirms
this under -race; reverting the CAS makes that same test fail with a
reported data race, not just a wrong value.
Adding completed makes MessageFuture contain a lock, which go vet's
copylocks check correctly flags on two existing tests that compared
*messageFuture values by pointer dereference. Both now use assert.Same,
which was already the more accurate assertion — they exist to check that
the same object was stored and retrieved, not that two independently built
values happen to be equal.
Two existing "duplicate response does not block" tests set up their
already-signaled future by writing straight to Done, bypassing Complete
entirely. That desynchronizes completed from Done and would deadlock on the
unconditional send. They now call Complete for setup instead, and the getty
one's assertion is corrected: it expected the dropped duplicate's payload to
have overwritten Response, which was the very inconsistency this commit
removes.
* fix(remoting): skip futures for callback-less messages
Callback-less responses reuse server-generated IDs and must not overwrite an in-flight client request future. Only create and clean up futures when a callback waits for a response, and cover both transports with collision regression tests.
---------
Co-authored-by: ThunGuo <te...@apache.org>
Co-authored-by: Ethan <xi...@apache.org>
---
changes/dev.md | 1 +
changes/dev_zh.md | 1 +
pkg/protocol/message/message_future_test.go | 82 +++++++
pkg/protocol/message/other_message.go | 22 ++
pkg/remoting/getty/futures_ownership_test.go | 209 +++++++++++++++++
pkg/remoting/getty/getty_client.go | 7 +-
pkg/remoting/getty/getty_remoting.go | 22 +-
pkg/remoting/getty/getty_remoting_test.go | 8 +-
pkg/remoting/grpc/futures_ownership_test.go | 248 +++++++++++++++++++++
pkg/remoting/grpc/grpc_client.go | 7 +-
pkg/remoting/grpc/grpc_client_test.go | 3 +
pkg/remoting/grpc/grpc_remoting.go | 28 ++-
pkg/remoting/grpc/grpc_remoting_test.go | 2 +-
pkg/remoting/grpc/readwriter.go | 8 +-
.../client/client_on_response_processor.go | 10 +-
15 files changed, 632 insertions(+), 26 deletions(-)
diff --git a/changes/dev.md b/changes/dev.md
index 197efade..75045569 100755
--- a/changes/dev.md
+++ b/changes/dev.md
@@ -42,6 +42,7 @@
- [[#130](https://github.com/apache/incubator-seata-go/pull/130)] getty session auto close bug
- [[#991](https://github.com/apache/incubator-seata-go/issues/991)] fix connection leaks and prevent nil pointer panic in async worker
- [[#887](https://github.com/apache/incubator-seata-go/issues/887)] make DayValue serialization timezone-stable
+ - [[#1159](https://github.com/apache/incubator-seata-go/issues/1159)] stop leaking message futures when a synchronous request times out or when a message is sent with nobody waiting for a reply, and stop a duplicate response from blocking the transport receive loop
- [[#1153](https://github.com/apache/incubator-seata-go/issues/1153)] fix a done context making the global transaction second phase report a commit that never reached the TC as success, and panic on rollback
- [[#1143](https://github.com/apache/incubator-seata-go/issues/1143)] read the xid of `XA RECOVER` as the `[]byte` the MySQL driver actually returns, and close the result set
- dispatch `ConsistentHashLoadBalance` and `LeastActiveLoadBalance` in `loadbalance.Select()`: both strategies were already implemented and accepted as config values, but missing from the dispatch `switch`, so they silently fell back to random load balancing
diff --git a/changes/dev_zh.md b/changes/dev_zh.md
index db60c4cb..f6137e5d 100644
--- a/changes/dev_zh.md
+++ b/changes/dev_zh.md
@@ -41,6 +41,7 @@ Seata-go 是一款开源的分布式事务解决方案，提供高性能和简
- [[#1201](https://github.com/apache/incubator-seata-go/issues/1201)] TCC 二阶段方法返回 false 时报告可重试的分支失败状态
- [[#904](https://github.com/apache/incubator-seata-go/issues/904)] 修复 XA autoCommit 下 `SELECT ... FOR UPDATE` 后紧接其他语句导致的 "busy buffer" / "driver: bad connection"：将分支提交（XA END + XA PREPARE）延迟到查询结果集关闭之后再执行
- [[#130](https://github.com/apache/incubator-seata-go/pull/130)] 修复getty session自动关闭的bug
+- [[#1159](https://github.com/apache/incubator-seata-go/issues/1159)] 修复同步请求超时、以及无人等待回复的消息发送后 message future 泄漏，并避免重复响应阻塞传输层接收循环
- [[#1153](https://github.com/apache/incubator-seata-go/issues/1153)] 修复 context 已取消时全局事务二阶段的处理：提交请求从未发出却返回成功，回滚则对空响应做类型断言而 panic
- [[#1143](https://github.com/apache/incubator-seata-go/issues/1143)] 按 MySQL 驱动实际返回的 `[]byte` 读取 `XA RECOVER` 的 xid，并关闭结果集
- 修复 `loadbalance.Select()` 在 `load-balance.type` 配置为 `ConsistentHashLoadBalance` / `LeastActiveLoadBalance` 时静默回退为随机负载均衡的问题：两个策略均已实现且配置层已接受，仅分发 `switch` 缺少对应分支
diff --git a/pkg/protocol/message/message_future_test.go b/pkg/protocol/message/message_future_test.go
new file mode 100644
index 00000000..d60102ed
--- /dev/null
+++ b/pkg/protocol/message/message_future_test.go
@@ -0,0 +1,82 @@
+/*
+ * 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 message
+
+import (
+ "sync"
+ "testing"
+ "time"
+)
+
+func TestMessageFutureCompleteSignalsWaiter(t *testing.T) {
+ future := NewMessageFuture(RpcMessage{ID: 1})
+
+ if !future.Complete("ok") {
+ t.Fatal("Complete() = false, want true for the first response")
+ }
+ if future.Response != "ok" {
+ t.Fatalf("Response = %v, want \"ok\"", future.Response)
+ }
+ select {
+ case <-future.Done:
+ default:
+ t.Fatal("Complete() did not signal the waiter")
+ }
+}
+
+// Done holds a single signal. A duplicate or late response must be dropped
+// instead of blocking the transport receive loop.
+func TestMessageFutureCompleteDropsDuplicate(t *testing.T) {
+ future := NewMessageFuture(RpcMessage{ID: 1})
+ future.Complete("first")
+
+ returned := make(chan bool, 1)
+ go func() { returned <- future.Complete("second") }()
+
+ select {
+ case signaled := <-returned:
+ if signaled {
+ t.Fatal("Complete() = true for a duplicate response, want false")
+ }
+ case <-time.After(2 * time.Second):
+ t.Fatal("Complete() blocked on a duplicate response")
+ }
+}
+
+// Both transports decode each incoming message in its own goroutine, so two
+// responses for the same ID can call Complete concurrently. Run under -race:
+// this is what actually proves there is no data race on Response, not just
+// that the logical result looks right.
+func TestMessageFutureCompleteIsRaceFreeUnderConcurrentCalls(t *testing.T) {
+ for i := 0; i < 200; i++ {
+ future := NewMessageFuture(RpcMessage{ID: 1})
+ var wg sync.WaitGroup
+ results := make([]bool, 2)
+ wg.Add(2)
+ go func() { defer wg.Done(); results[0] = future.Complete("first") }()
+ go func() { defer wg.Done(); results[1] = future.Complete("second") }()
+ wg.Wait()
+
+ if results[0] == results[1] {
+ t.Fatalf("round %d: both Complete calls reported %v, want exactly one true", i, results[0])
+ }
+ if future.Response != "first" && future.Response != "second" {
+ t.Fatalf("round %d: Response = %v, want one of the two calls' payloads", i, future.Response)
+ }
+ }
+}
diff --git a/pkg/protocol/message/other_message.go b/pkg/protocol/message/other_message.go
index 6ed55243..b9448fb5 100644
--- a/pkg/protocol/message/other_message.go
+++ b/pkg/protocol/message/other_message.go
@@ -17,6 +17,8 @@
package message
+import "sync/atomic"
+
type RpcMessage struct {
ID int32
Type RequestType
@@ -31,6 +33,12 @@ type MessageFuture struct {
Err error
Response interface{}
Done chan struct{}
+
+ // completed guards Response so at most one caller ever writes it. Complete
+ // can be called from more than one goroutine for the same future: both
+ // transports decode each incoming message in its own goroutine, so two
+ // responses for the same ID race to complete the same future.
+ completed atomic.Bool
}
func NewMessageFuture(message RpcMessage) *MessageFuture {
@@ -40,6 +48,20 @@ func NewMessageFuture(message RpcMessage) *MessageFuture {
}
}
+// Complete records the response and wakes the waiter, reporting whether it
+// signaled. Only the caller that wins the completed flag writes Response,
+// which is what makes a concurrent duplicate call safe: a loser returns
+// false without touching Response, rather than racing the winner's write or
+// clobbering it after the waiter has already read it.
+func (f *MessageFuture) Complete(response interface{}) bool {
+ if !f.completed.CompareAndSwap(false, true) {
+ return false
+ }
+ f.Response = response
+ f.Done <- struct{}{}
+ return true
+}
+
type HeartBeatMessage struct {
Ping bool
}
diff --git a/pkg/remoting/getty/futures_ownership_test.go b/pkg/remoting/getty/futures_ownership_test.go
new file mode 100644
index 00000000..bde0db01
--- /dev/null
+++ b/pkg/remoting/getty/futures_ownership_test.go
@@ -0,0 +1,209 @@
+/*
+ * 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 (
+ "errors"
+ "testing"
+ "time"
+
+ "github.com/golang/mock/gomock"
+ "go.uber.org/atomic"
+
+ "seata.apache.org/seata-go/v2/pkg/protocol/message"
+ "seata.apache.org/seata-go/v2/pkg/remoting/mock"
+)
+
+func newTestClient() *GettyRemotingClient {
+ return &GettyRemotingClient{idGenerator: &atomic.Uint32{}, gettyRemoting: newGettyRemoting()}
+}
+
+// shortenRequestTimeout keeps the tests off the real 20s wait.
+func shortenRequestTimeout(t *testing.T, d time.Duration) {
+ t.Helper()
+ original := rpcRequestTimeout
+ rpcRequestTimeout = d
+ t.Cleanup(func() { rpcRequestTimeout = original })
+}
+
+func countFutures(g *GettyRemoting) int {
+ n := 0
+ g.futures.Range(func(_, _ interface{}) bool { n++; return true })
+ return n
+}
+
+// A timed-out request used to delete from mergeMsgMap, leaving its future in
+// futures for the lifetime of the process.
+func TestSyncCallbackRemovesFutureOnTimeout(t *testing.T) {
+ shortenRequestTimeout(t, 50*time.Millisecond)
+ client := newTestClient()
+ req := message.RpcMessage{ID: 987654321}
+ future := message.NewMessageFuture(req)
+ client.gettyRemoting.futures.Store(req.ID, future)
+
+ if _, err := client.syncCallback(req, future); err == nil {
+ t.Fatal("syncCallback() = nil error, want a timeout")
+ }
+ if client.GetMessageFuture(req.ID) != nil {
+ t.Fatal("the timed-out request is still retained in futures")
+ }
+}
+
+// On the answered path the on-response processor also removes the future, so
+// this is not a leak in production. The waiter still clears it so that cleanup
+// has one owner and does not depend on which processor handled the reply.
+func TestSyncCallbackRemovesFutureOnResponse(t *testing.T) {
+ shortenRequestTimeout(t, 5*time.Second)
+ client := newTestClient()
+ req := message.RpcMessage{ID: 42}
+ future := message.NewMessageFuture(req)
+ client.gettyRemoting.futures.Store(req.ID, future)
+
+ client.gettyRemoting.NotifyRpcMessageResponse(message.RpcMessage{ID: req.ID, Body: "ok"})
+
+ res, err := client.syncCallback(req, future)
+ if err != nil {
+ t.Fatalf("syncCallback() error = %v, want nil", err)
+ }
+ if res != "ok" {
+ t.Fatalf("syncCallback() = %v, want \"ok\"", res)
+ }
+ if client.GetMessageFuture(req.ID) != nil {
+ t.Fatal("the answered request is still retained in futures")
+ }
+}
+
+// SendAsyncResponse sends with no callback, so it must not create a future.
+func TestSendAsyncDoesNotStoreFutureWhenNoCallbackWaits(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+
+ session := mock.NewMockTestSession(ctrl)
+ session.EXPECT().IsClosed().Return(false).AnyTimes()
+ session.EXPECT().WritePkg(gomock.Any(), gomock.Any()).Return(0, 0, nil).AnyTimes()
+
+ remoting := newGettyRemoting()
+ if _, err := remoting.sendAsync(session, message.RpcMessage{ID: 11}, nil); err != nil {
+ t.Fatalf("sendAsync() error = %v, want nil", err)
+ }
+ if got := countFutures(remoting); got != 0 {
+ t.Fatalf("futures holds %d entries after a callback-less send, want 0", got)
+ }
+}
+
+func TestSendAsyncRemovesFutureOnWriteError(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+ session := mock.NewMockTestSession(ctrl)
+ session.EXPECT().IsClosed().Return(false)
+ session.EXPECT().WritePkg(gomock.Any(), gomock.Any()).Return(0, 0, errors.New("write failed"))
+ session.EXPECT().Stat().Return("mock session")
+
+ remoting := newGettyRemoting()
+ callback := func(message.RpcMessage, *message.MessageFuture) (interface{}, error) {
+ t.Fatal("callback should not run after a write failure")
+ return nil, nil
+ }
+ if _, err := remoting.sendAsync(session, message.RpcMessage{ID: 11}, callback); err == nil {
+ t.Fatal("sendAsync() = nil error, want a write failure")
+ }
+ if got := countFutures(remoting); got != 0 {
+ t.Fatalf("futures holds %d entries after a write failure, want 0", got)
+ }
+}
+
+func TestSendAsyncWithoutCallbackPreservesInFlightFuture(t *testing.T) {
+ tests := []struct {
+ name string
+ writeErr error
+ }{
+ {name: "write succeeds"},
+ {name: "write fails", writeErr: errors.New("write failed")},
+ }
+ for _, test := range tests {
+ t.Run(test.name, func(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+ session := mock.NewMockTestSession(ctrl)
+ session.EXPECT().IsClosed().Return(false)
+ session.EXPECT().WritePkg(gomock.Any(), gomock.Any()).Return(0, 0, test.writeErr)
+ if test.writeErr != nil {
+ session.EXPECT().Stat().Return("mock session")
+ }
+
+ remoting := newGettyRemoting()
+ msg := message.RpcMessage{ID: 11}
+ pending := message.NewMessageFuture(msg)
+ remoting.futures.Store(msg.ID, pending)
+
+ _, err := remoting.sendAsync(session, msg, nil)
+ if (err != nil) != (test.writeErr != nil) {
+ t.Fatalf("sendAsync() error = %v, wantErr %t", err, test.writeErr != nil)
+ }
+ if got := remoting.GetMessageFuture(msg.ID); got != pending {
+ t.Fatalf("callback-less send changed the in-flight future: got %p, want %p", got, pending)
+ }
+ })
+ }
+}
+
+func TestRepeatedTimeoutsDoNotGrowFutures(t *testing.T) {
+ shortenRequestTimeout(t, 10*time.Millisecond)
+ client := newTestClient()
+
+ const rounds = 200
+ for i := 0; i < rounds; i++ {
+ id := int32(i + 1)
+ req := message.RpcMessage{ID: id}
+ future := message.NewMessageFuture(req)
+ client.gettyRemoting.futures.Store(id, future)
+ if _, err := client.syncCallback(req, future); err == nil {
+ t.Fatalf("round %d: syncCallback() = nil error, want a timeout", i)
+ }
+ }
+ if got := countFutures(client.gettyRemoting); got != 0 {
+ t.Fatalf("futures holds %d entries after %d timeouts, want 0", got, rounds)
+ }
+}
+
+// The tests above drive syncCallback directly. This one goes through the real
+// entry point so the store and the delete are exercised by the same code path
+// production uses.
+func TestSendSyncTimeoutLeavesNoFuture(t *testing.T) {
+ shortenRequestTimeout(t, 30*time.Millisecond)
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+
+ session := mock.NewMockTestSession(ctrl)
+ session.EXPECT().IsClosed().Return(false).AnyTimes()
+ session.EXPECT().RemoteAddr().Return("127.0.0.1:8091").AnyTimes()
+ session.EXPECT().Stat().Return("mock session").AnyTimes()
+ session.EXPECT().WritePkg(gomock.Any(), gomock.Any()).Return(0, 0, nil).AnyTimes()
+
+ client := newTestClient()
+ const rounds = 50
+ for i := 0; i < rounds; i++ {
+ msg := message.RpcMessage{ID: int32(i + 1)}
+ if _, err := client.gettyRemoting.SendSync(msg, session, client.syncCallback); err == nil {
+ t.Fatalf("round %d: SendSync() = nil error, want a timeout", i)
+ }
+ }
+ if got := countFutures(client.gettyRemoting); got != 0 {
+ t.Fatalf("futures holds %d entries after %d timed-out SendSync calls, want 0", got, rounds)
+ }
+}
diff --git a/pkg/remoting/getty/getty_client.go b/pkg/remoting/getty/getty_client.go
index 4b60a9b5..3164a203 100644
--- a/pkg/remoting/getty/getty_client.go
+++ b/pkg/remoting/getty/getty_client.go
@@ -100,9 +100,12 @@ func (g *GettyRemotingClient) asyncCallback(reqMsg message.RpcMessage, respMsg *
}
func (g *GettyRemotingClient) syncCallback(reqMsg message.RpcMessage, respMsg *message.MessageFuture) (interface{}, error) {
+ // The future lives in futures, not mergeMsgMap, and nothing else removes
+ // it. Once this returns nobody waits for the response any more, on either
+ // branch, so this is the one place that owns the cleanup.
+ defer g.gettyRemoting.RemoveMessageFuture(reqMsg.ID)
select {
- case <-gxtime.GetDefaultTimerWheel().After(RpcRequestTimeout):
- g.gettyRemoting.RemoveMergedMessageFuture(reqMsg.ID)
+ case <-gxtime.GetDefaultTimerWheel().After(rpcRequestTimeout):
log.Errorf("wait resp timeout: %#v", reqMsg)
return nil, fmt.Errorf("wait response timeout, request: %#v", reqMsg)
case <-respMsg.Done:
diff --git a/pkg/remoting/getty/getty_remoting.go b/pkg/remoting/getty/getty_remoting.go
index 4cc565f3..0e228317 100644
--- a/pkg/remoting/getty/getty_remoting.go
+++ b/pkg/remoting/getty/getty_remoting.go
@@ -33,6 +33,10 @@ const (
RpcRequestTimeout = 20 * time.Second
)
+// rpcRequestTimeout is the deadline the client actually waits on. It exists so
+// tests can shorten the wait without changing the exported constant.
+var rpcRequestTimeout = RpcRequestTimeout
+
type (
callbackMethod func(reqMsg message.RpcMessage, respMsg *message.MessageFuture) (interface{}, error)
GettyRemoting struct {
@@ -94,11 +98,18 @@ func (g *GettyRemoting) sendAsync(session getty.Session, msg message.RpcMessage,
log.Warn("sendAsyncRequestWithResponse nothing, caused by null channel.")
return nil, fmt.Errorf("session is closed")
}
- resp := message.NewMessageFuture(msg)
- g.futures.Store(msg.ID, resp)
+ // Only requests with a callback wait for a response. A callback-less response
+ // may reuse the ID of an in-flight request, so it must not touch futures.
+ var resp *message.MessageFuture
+ if callback != nil {
+ resp = message.NewMessageFuture(msg)
+ g.futures.Store(msg.ID, resp)
+ }
_, _, err = session.WritePkg(msg, time.Duration(0))
if err != nil {
- g.futures.Delete(msg.ID)
+ if resp != nil {
+ g.futures.Delete(msg.ID)
+ }
log.Errorf("send message: %#v, session: %s", msg, session.Stat())
return nil, err
}
@@ -133,12 +144,9 @@ func (g *GettyRemoting) GetMergedMessage(msgID int32) *message.MergedWarpMessage
func (g *GettyRemoting) NotifyRpcMessageResponse(rpcMessage message.RpcMessage) {
messageFuture := g.GetMessageFuture(rpcMessage.ID)
if messageFuture != nil {
- messageFuture.Response = rpcMessage.Body
// todo add messageFuture.Err
// messageFuture.Err = rpcMessage.Err
- select {
- case messageFuture.Done <- struct{}{}:
- default:
+ if !messageFuture.Complete(rpcMessage.Body) {
log.Warnf("response notification dropped for msg ID: %d because the future was already signaled", rpcMessage.ID)
}
// client.msgFutures.Delete(rpcMessage.RequestID)
diff --git a/pkg/remoting/getty/getty_remoting_test.go b/pkg/remoting/getty/getty_remoting_test.go
index b23a799c..c4f08b07 100644
--- a/pkg/remoting/getty/getty_remoting_test.go
+++ b/pkg/remoting/getty/getty_remoting_test.go
@@ -54,7 +54,7 @@ func TestGettyRemoting_GetMessageFuture(t *testing.T) {
if test.messageFuture != nil {
gettyRemotingClient.gettyRemoting.futures.Store(test.msgID, test.messageFuture)
messageFuture := gettyRemotingClient.gettyRemoting.GetMessageFuture(test.msgID)
- assert.Equal(t, *test.messageFuture, *messageFuture)
+ assert.Same(t, test.messageFuture, messageFuture)
} else {
messageFuture := gettyRemotingClient.gettyRemoting.GetMessageFuture(test.msgID)
assert.Empty(t, messageFuture)
@@ -212,7 +212,7 @@ func TestGettyRemoting_NotifyRpcMessageResponseDoesNotBlockWhenAlreadySignaled(t
gettyRemoting := newGettyRemoting()
request := message.RpcMessage{ID: 1}
messageFuture := message.NewMessageFuture(request)
- messageFuture.Done <- struct{}{}
+ messageFuture.Complete("first") // fills Done the same way production code does
gettyRemoting.futures.Store(request.ID, messageFuture)
done := make(chan struct{})
@@ -230,6 +230,8 @@ func TestGettyRemoting_NotifyRpcMessageResponseDoesNotBlockWhenAlreadySignaled(t
t.Fatal("NotifyRpcMessageResponse blocked when the future was already signaled")
}
- assert.Equal(t, "late-response", messageFuture.Response)
+ // The dropped duplicate must not overwrite the response the first call
+ // already recorded, or a slow reader could observe either value.
+ assert.Equal(t, "first", messageFuture.Response)
assert.Len(t, messageFuture.Done, 1)
}
diff --git a/pkg/remoting/grpc/futures_ownership_test.go b/pkg/remoting/grpc/futures_ownership_test.go
new file mode 100644
index 00000000..d394f4bf
--- /dev/null
+++ b/pkg/remoting/grpc/futures_ownership_test.go
@@ -0,0 +1,248 @@
+/*
+ * 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 grpc
+
+import (
+ "testing"
+ "time"
+
+ "go.uber.org/atomic"
+
+ "seata.apache.org/seata-go/v2/pkg/protocol/message"
+ "seata.apache.org/seata-go/v2/pkg/remoting/grpc/pb"
+)
+
+func newTestClient() *GrpcRemotingClient {
+ return &GrpcRemotingClient{idGenerator: &atomic.Uint32{}, grpcRemoting: newGrpcRemoting()}
+}
+
+// shortenRequestTimeout keeps the tests off the real 20s wait.
+func shortenRequestTimeout(t *testing.T, d time.Duration) {
+ t.Helper()
+ original := rpcRequestTimeout
+ rpcRequestTimeout = d
+ t.Cleanup(func() { rpcRequestTimeout = original })
+}
+
+func countFutures(g *GrpcRemoting) int {
+ n := 0
+ g.futures.Range(func(_, _ interface{}) bool { n++; return true })
+ return n
+}
+
+// openChannel returns a channel whose Send succeeds, by answering the send
+// tracker the way sendProcessor would.
+func openChannel(t *testing.T) *Channel {
+ t.Helper()
+ ch := &Channel{sendCh: make(chan *pb.GrpcMessageProto, 1), closeCh: make(chan struct{})}
+ stopped := make(chan struct{})
+ go func() {
+ defer close(stopped)
+ for {
+ select {
+ case <-ch.closeCh:
+ return
+ case req := <-ch.sendCh:
+ if tracker, ok := msgSendTrackers.Load(req.Id); ok {
+ tracker.(chan error) <- nil
+ }
+ }
+ }
+ }()
+ t.Cleanup(func() {
+ close(ch.closeCh)
+ <-stopped
+ })
+ return ch
+}
+
+func commitRequest(id int32) message.RpcMessage {
+ return message.RpcMessage{
+ ID: id,
+ HeadMap: map[string]string{},
+ Body: &pb.GlobalCommitRequestProto{
+ AbstractGlobalEndRequest: &pb.AbstractGlobalEndRequestProto{Xid: "xid-1"},
+ },
+ }
+}
+
+// A timed-out request used to delete from mergeMsgMap, leaving its future in
+// futures for the lifetime of the process.
+func TestSyncCallbackRemovesFutureOnTimeout(t *testing.T) {
+ shortenRequestTimeout(t, 50*time.Millisecond)
+ client := newTestClient()
+ req := message.RpcMessage{ID: 987654321}
+ future := message.NewMessageFuture(req)
+ client.grpcRemoting.futures.Store(req.ID, future)
+
+ if _, err := client.syncCallback(req, future); err == nil {
+ t.Fatal("syncCallback() = nil error, want a timeout")
+ }
+ if client.GetMessageFuture(req.ID) != nil {
+ t.Fatal("the timed-out request is still retained in futures")
+ }
+}
+
+// On the answered path the on-response processor also removes the future, so
+// this is not a leak in production. The waiter still clears it so that cleanup
+// has one owner and does not depend on which processor handled the reply.
+func TestSyncCallbackRemovesFutureOnResponse(t *testing.T) {
+ shortenRequestTimeout(t, 5*time.Second)
+ client := newTestClient()
+ req := message.RpcMessage{ID: 42}
+ future := message.NewMessageFuture(req)
+ client.grpcRemoting.futures.Store(req.ID, future)
+
+ client.grpcRemoting.NotifyRpcMessageResponse(message.RpcMessage{ID: req.ID, Body: "ok"})
+
+ res, err := client.syncCallback(req, future)
+ if err != nil {
+ t.Fatalf("syncCallback() error = %v, want nil", err)
+ }
+ if res != "ok" {
+ t.Fatalf("syncCallback() = %v, want \"ok\"", res)
+ }
+ if client.GetMessageFuture(req.ID) != nil {
+ t.Fatal("the answered request is still retained in futures")
+ }
+}
+
+func TestSendAsyncRemovesFutureOnEncodeError(t *testing.T) {
+ remoting := newGrpcRemoting()
+ // A body that is not a proto.Message fails in Encode, before anything is
+ // sent. It used to panic there instead, with the future already stored.
+ msg := message.RpcMessage{ID: 7, HeadMap: map[string]string{}, Body: "not a proto message"}
+
+ callback := func(message.RpcMessage, *message.MessageFuture) (interface{}, error) {
+ t.Fatal("callback should not run after an encode failure")
+ return nil, nil
+ }
+ if _, err := remoting.sendAsync(openChannel(t), msg, callback); err == nil {
+ t.Fatal("sendAsync() = nil error, want an encode failure")
+ }
+ if got := countFutures(remoting); got != 0 {
+ t.Fatalf("futures holds %d entries after an encode failure, want 0", got)
+ }
+}
+
+// SendAsyncResponse sends with no callback, so it must not create a future.
+func TestSendAsyncDoesNotStoreFutureWhenNoCallbackWaits(t *testing.T) {
+ remoting := newGrpcRemoting()
+
+ if _, err := remoting.sendAsync(openChannel(t), commitRequest(11), nil); err != nil {
+ t.Fatalf("sendAsync() error = %v, want nil", err)
+ }
+ if got := countFutures(remoting); got != 0 {
+ t.Fatalf("futures holds %d entries after a callback-less send, want 0", got)
+ }
+}
+
+func TestSendAsyncWithoutCallbackPreservesInFlightFuture(t *testing.T) {
+ tests := []struct {
+ name string
+ msg message.RpcMessage
+ wantErr bool
+ }{
+ {name: "send succeeds", msg: commitRequest(11)},
+ {
+ name: "encode fails",
+ msg: message.RpcMessage{ID: 11, HeadMap: map[string]string{}, Body: "not a proto message"},
+ wantErr: true,
+ },
+ }
+ for _, test := range tests {
+ t.Run(test.name, func(t *testing.T) {
+ remoting := newGrpcRemoting()
+ pending := message.NewMessageFuture(message.RpcMessage{ID: test.msg.ID})
+ remoting.futures.Store(test.msg.ID, pending)
+
+ _, err := remoting.sendAsync(openChannel(t), test.msg, nil)
+ if (err != nil) != test.wantErr {
+ t.Fatalf("sendAsync() error = %v, wantErr %t", err, test.wantErr)
+ }
+ if got := remoting.GetMessageFuture(test.msg.ID); got != pending {
+ t.Fatalf("callback-less send changed the in-flight future: got %p, want %p", got, pending)
+ }
+ })
+ }
+}
+
+func TestNotifyRpcMessageResponseSignalsWaitingFuture(t *testing.T) {
+ remoting := newGrpcRemoting()
+ req := message.RpcMessage{ID: 1}
+ future := message.NewMessageFuture(req)
+ remoting.futures.Store(req.ID, future)
+
+ remoting.NotifyRpcMessageResponse(message.RpcMessage{ID: req.ID, Body: "ok"})
+
+ if future.Response != "ok" {
+ t.Fatalf("Response = %v, want \"ok\"", future.Response)
+ }
+ select {
+ case <-future.Done:
+ default:
+ t.Fatal("expected NotifyRpcMessageResponse to signal the waiting future")
+ }
+}
+
+// Done is buffered for one signal, so a duplicate or late response used to
+// block the receive loop permanently.
+func TestNotifyRpcMessageResponseDoesNotBlockWhenAlreadySignaled(t *testing.T) {
+ remoting := newGrpcRemoting()
+ req := message.RpcMessage{ID: 1}
+ future := message.NewMessageFuture(req)
+ future.Complete("first") // fills Done the same way production code does
+ remoting.futures.Store(req.ID, future)
+
+ returned := make(chan struct{})
+ go func() {
+ remoting.NotifyRpcMessageResponse(message.RpcMessage{ID: req.ID, Body: "late-response"})
+ close(returned)
+ }()
+
+ select {
+ case <-returned:
+ case <-time.After(2 * time.Second):
+ t.Fatal("NotifyRpcMessageResponse blocked when the future was already signaled")
+ }
+
+ // The dropped duplicate must not overwrite the response the first call
+ // already recorded, or a slow reader could observe either value.
+ if future.Response != "first" {
+ t.Fatalf("Response = %v, want \"first\" to survive the dropped duplicate", future.Response)
+ }
+}
+
+func TestRepeatedTimeoutsDoNotGrowFutures(t *testing.T) {
+ shortenRequestTimeout(t, 10*time.Millisecond)
+ client := newTestClient()
+
+ const rounds = 200
+ for i := 0; i < rounds; i++ {
+ id := int32(i + 1)
+ req := message.RpcMessage{ID: id}
+ future := message.NewMessageFuture(req)
+ client.grpcRemoting.futures.Store(id, future)
+ if _, err := client.syncCallback(req, future); err == nil {
+ t.Fatalf("round %d: syncCallback() = nil error, want a timeout", i)
+ }
+ }
+ if got := countFutures(client.grpcRemoting); got != 0 {
+ t.Fatalf("futures holds %d entries after %d timeouts, want 0", got, rounds)
+ }
+}
diff --git a/pkg/remoting/grpc/grpc_client.go b/pkg/remoting/grpc/grpc_client.go
index 3c12a42d..8c7c5aa8 100644
--- a/pkg/remoting/grpc/grpc_client.go
+++ b/pkg/remoting/grpc/grpc_client.go
@@ -110,9 +110,12 @@ func (client *GrpcRemotingClient) asyncCallback(reqMsg message.RpcMessage, respM
}
func (client GrpcRemotingClient) syncCallback(reqMsg message.RpcMessage, respMsg *message.MessageFuture) (interface{}, error) {
+ // The future lives in futures, not mergeMsgMap, and nothing else removes
+ // it. Once this returns nobody waits for the response any more, on either
+ // branch, so this is the one place that owns the cleanup.
+ defer client.grpcRemoting.RemoveMessageFuture(reqMsg.ID)
select {
- case <-gxtime.GetDefaultTimerWheel().After(RpcRequestTimeout):
- client.grpcRemoting.RemoveMergedMessageFuture(reqMsg.ID)
+ case <-gxtime.GetDefaultTimerWheel().After(rpcRequestTimeout):
log.Errorf("wait resp timeout: %#v", reqMsg)
return nil, fmt.Errorf("wait response timeout, request: %#v", reqMsg)
case <-respMsg.Done:
diff --git a/pkg/remoting/grpc/grpc_client_test.go b/pkg/remoting/grpc/grpc_client_test.go
index 153f7967..97c5c8d8 100644
--- a/pkg/remoting/grpc/grpc_client_test.go
+++ b/pkg/remoting/grpc/grpc_client_test.go
@@ -21,6 +21,7 @@ import (
"fmt"
"reflect"
"testing"
+ "time"
"github.com/agiledragon/gomonkey/v2"
"github.com/stretchr/testify/assert"
@@ -103,6 +104,8 @@ func TestGrpcRemotingClient_SendAsyncRequest(t *testing.T) {
func Test_syncCallback(t *testing.T) {
codec.Init()
log.Init()
+ // The timeout case used to sit on the real 20s deadline.
+ shortenRequestTimeout(t, 50*time.Millisecond)
tests := []struct {
name string
respMsg *message.MessageFuture
diff --git a/pkg/remoting/grpc/grpc_remoting.go b/pkg/remoting/grpc/grpc_remoting.go
index 2c7ab18e..da4c07c2 100644
--- a/pkg/remoting/grpc/grpc_remoting.go
+++ b/pkg/remoting/grpc/grpc_remoting.go
@@ -30,6 +30,10 @@ import (
const RpcRequestTimeout = 20 * time.Second
+// rpcRequestTimeout is the deadline the client actually waits on. It exists so
+// tests can shorten the wait without changing the exported constant.
+var rpcRequestTimeout = RpcRequestTimeout
+
type (
callbackMethod func(reqMsg message.RpcMessage, respMsg *message.MessageFuture) (interface{}, error)
@@ -93,15 +97,26 @@ func (g *GrpcRemoting) sendAsync(channel *Channel, msg message.RpcMessage, callb
log.Warn("sendAsyncRequestWithResponse nothing, caused by null channel.")
return nil, fmt.Errorf("stream is closed")
}
- resp := message.NewMessageFuture(msg)
- g.futures.Store(msg.ID, resp)
+ // Only requests with a callback wait for a response. A callback-less response
+ // may reuse the ID of an in-flight request, so it must not touch futures.
+ var resp *message.MessageFuture
+ if callback != nil {
+ resp = message.NewMessageFuture(msg)
+ g.futures.Store(msg.ID, resp)
+ }
request, err := Encode(msg)
if err != nil {
+ if resp != nil {
+ g.futures.Delete(msg.ID)
+ }
+ log.Errorf("encode message: %#v", msg)
return nil, err
}
if err := channel.Send(request); err != nil {
- g.futures.Delete(msg.ID)
+ if resp != nil {
+ g.futures.Delete(msg.ID)
+ }
log.Errorf("send message: %#v", msg)
return nil, err
}
@@ -136,11 +151,12 @@ func (g *GrpcRemoting) GetMergedMessage(msgID int32) *message.MergedWarpMessage
func (g *GrpcRemoting) NotifyRpcMessageResponse(rpcMessage message.RpcMessage) {
messageFuture := g.GetMessageFuture(rpcMessage.ID)
if messageFuture != nil {
- messageFuture.Response = rpcMessage.Body
// todo add messageFuture.Err
// messageFuture.Err = rpcMessage.Err
- messageFuture.Done <- struct{}{}
- // client.msgFutures.Delete(rpcMessage.RequestID)
+ if !messageFuture.Complete(rpcMessage.Body) {
+ log.Warnf("response notification dropped for msg ID: %d because the future was already signaled", rpcMessage.ID)
+ }
+ // The waiter removes the future once it stops waiting.
} else {
log.Infof("msg: {} is not found in msgFutures.", rpcMessage.ID)
}
diff --git a/pkg/remoting/grpc/grpc_remoting_test.go b/pkg/remoting/grpc/grpc_remoting_test.go
index db196c53..ae03d85e 100644
--- a/pkg/remoting/grpc/grpc_remoting_test.go
+++ b/pkg/remoting/grpc/grpc_remoting_test.go
@@ -53,7 +53,7 @@ func TestGrpcRemoting_GetMessageFuture(t *testing.T) {
if test.messageFuture != nil {
grpcRemotingClient.grpcRemoting.futures.Store(test.msgID, test.messageFuture)
messageFuture := grpcRemotingClient.grpcRemoting.GetMessageFuture(test.msgID)
- assert.Equal(t, *test.messageFuture, *messageFuture)
+ assert.Same(t, test.messageFuture, messageFuture)
} else {
messageFuture := grpcRemotingClient.grpcRemoting.GetMessageFuture(test.msgID)
assert.Empty(t, messageFuture)
diff --git a/pkg/remoting/grpc/readwriter.go b/pkg/remoting/grpc/readwriter.go
index 6c957649..dc71bf80 100644
--- a/pkg/remoting/grpc/readwriter.go
+++ b/pkg/remoting/grpc/readwriter.go
@@ -41,7 +41,13 @@ func Encode(msg message.RpcMessage) (*pb.GrpcMessageProto, error) {
}
rpcMessage.HeadMap[string(grpc.CodecType)] = strconv.Itoa(int(msg.Codec))
rpcMessage.HeadMap[string(grpc.CompressType)] = strconv.Itoa(int(msg.Compressor))
- anyMsg, err := anypb.New(msg.Body.(proto.Message))
+ body, ok := msg.Body.(proto.Message)
+ if !ok {
+ // A bare assertion here used to panic inside sendAsync, after the
+ // message future had already been stored.
+ return nil, fmt.Errorf("cannot encode body of type %T, want a proto.Message", msg.Body)
+ }
+ anyMsg, err := anypb.New(body)
if err != nil {
return nil, fmt.Errorf("could not new any msg: %v", err)
}
diff --git a/pkg/remoting/processor/client/client_on_response_processor.go b/pkg/remoting/processor/client/client_on_response_processor.go
index 715d4ad6..590a3b8e 100644
--- a/pkg/remoting/processor/client/client_on_response_processor.go
+++ b/pkg/remoting/processor/client/client_on_response_processor.go
@@ -97,8 +97,9 @@ func (f *clientOnResponseProcessor) handleGrpcOnResponse(rpcMessage message.RpcM
msgID := mergedMessage.MsgIds[i]
response := grpcRemotingClient.GetMessageFuture(msgID)
if response != nil {
- response.Response = mergedResult.Msgs[i]
- response.Done <- struct{}{}
+ if !response.Complete(mergedResult.Msgs[i]) {
+ log.Warnf("merged response dropped for msg ID: %d because the future was already signaled", msgID)
+ }
grpcRemotingClient.RemoveMessageFuture(msgID)
}
}
@@ -129,8 +130,9 @@ func (f *clientOnResponseProcessor) handleSeataOnResponse(rpcMessage message.Rpc
msgID := mergedMessage.MsgIds[i]
response := gettyRemotingClient.GetMessageFuture(msgID)
if response != nil {
- response.Response = mergedResult.Msgs[i]
- response.Done <- struct{}{}
+ if !response.Complete(mergedResult.Msgs[i]) {
+ log.Warnf("merged response dropped for msg ID: %d because the future was already signaled", msgID)
+ }
gettyRemotingClient.RemoveMessageFuture(msgID)
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org