This is an automated email from the ASF dual-hosted git repository. AlexStocks pushed a commit to branch fix/issue-3558-time-after-misuse in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
commit 88b788a0d93adcecd2e2d4e2643ddc02acc04268 Author: alexstocks <[email protected]> AuthorDate: Thu Jul 30 08:11:32 2026 +0800 fix: resolve time.After misuse (timer leak + dead drain guard) — #3558 - remoting/zookeeper/listener.go: extract the retry-wait into ZkEventListener.waitForRetry, which uses time.NewTimer + Stop() so the timer and its goroutine are reclaimed promptly when the listener exits (l.exit) instead of lingering until the delay elapses (SA1015). This also removes the duplicated wait block in the two ZK retry loops. - filter/accesslog/filter.go: remove the `default: return` in drainLogs that made the 5s timeout guard a no-op; use time.NewTimer + defer Stop so the guard actually bounds the drain and blocking log writes are not silently ignored. - Add unit tests: TestWaitForRetry_* (exit / timeout / exit-during-wait) and TestDrainLogs_* (returns on closed channel, blocks on empty open channel as a regression for the dead guard, drains buffered data). Refs: #3558 Signed-off-by: alexstocks <[email protected]> --- filter/accesslog/drain_logs_test.go | 104 +++++++++++++++++++++++++++++++ filter/accesslog/filter.go | 7 +-- remoting/zookeeper/listener.go | 38 ++++++----- remoting/zookeeper/listener_wait_test.go | 77 +++++++++++++++++++++++ 4 files changed, 208 insertions(+), 18 deletions(-) diff --git a/filter/accesslog/drain_logs_test.go b/filter/accesslog/drain_logs_test.go new file mode 100644 index 000000000..c3f0f23b4 --- /dev/null +++ b/filter/accesslog/drain_logs_test.go @@ -0,0 +1,104 @@ +/* + * 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 accesslog + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +// TestDrainLogsReturnsAfterChannelClosed verifies that drainLogs does not hang +// when the log channel is already closed (the !ok -> return path). +func TestDrainLogsReturnsAfterChannelClosed(t *testing.T) { + f := &Filter{logChan: make(chan Data), ctx: context.Background()} + close(f.logChan) + + done := make(chan struct{}) + go func() { + f.drainLogs() + close(done) + }() + + select { + case <-done: + case <-time.After(500 * time.Millisecond): + t.Fatal("drainLogs did not return after the channel was closed") + } +} + +// TestDrainLogsBlocksOnEmptyOpenChannel is a regression test for #3558: the +// previous implementation returned immediately on an empty channel because of a +// `default: return` branch, which made the 5s timeout guard dead. After the fix +// drainLogs must block on an empty, open channel until it is closed. +func TestDrainLogsBlocksOnEmptyOpenChannel(t *testing.T) { + f := &Filter{logChan: make(chan Data), ctx: context.Background()} + + done := make(chan struct{}) + go func() { + f.drainLogs() + close(done) + }() + + // It should still be running (blocking) shortly after start. + select { + case <-done: + t.Fatal("drainLogs returned immediately on an empty open channel (dead guard bug)") + case <-time.After(300 * time.Millisecond): + // expected: still blocking, the 5s guard is now live + } + + // Release it and make sure it eventually returns. + close(f.logChan) + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("drainLogs did not return after the channel was closed") + } +} + +// TestDrainLogsDrainsBufferedData verifies that all buffered log entries are +// flushed (and written to the configured file) before drainLogs returns. +func TestDrainLogsDrainsBufferedData(t *testing.T) { + tmp := filepath.Join(t.TempDir(), "access.log") + f := &Filter{ + logChan: make(chan Data, 3), + ctx: context.Background(), + fileCache: make(map[string]*os.File), + } + + for i := 0; i < 3; i++ { + f.logChan <- Data{ + accessLog: tmp, + data: map[string]string{"k": "v"}, + } + } + close(f.logChan) + + f.drainLogs() + + content, err := os.ReadFile(tmp) + assert.NoError(t, err) + assert.Equal(t, 3, strings.Count(string(content), "\n"), + "all buffered log entries should be written") +} diff --git a/filter/accesslog/filter.go b/filter/accesslog/filter.go index 02a1f8e4f..497d536bf 100644 --- a/filter/accesslog/filter.go +++ b/filter/accesslog/filter.go @@ -213,7 +213,8 @@ func (f *Filter) processLogs() { // drainLogs drains remaining log data with timeout protection func (f *Filter) drainLogs() { - timeout := time.After(5 * time.Second) + timer := time.NewTimer(5 * time.Second) + defer timer.Stop() for { select { case accessLogData, ok := <-f.logChan: @@ -221,11 +222,9 @@ func (f *Filter) drainLogs() { return } f.writeLogToFileWithTimeout(accessLogData, 1*time.Second) - case <-timeout: + case <-timer.C: logger.Warn("[Filter][AccessLog] accessLog drain timeout, some logs may be lost") return - default: - return } } } diff --git a/remoting/zookeeper/listener.go b/remoting/zookeeper/listener.go index 1f0de31c0..ad72b6a28 100644 --- a/remoting/zookeeper/listener.go +++ b/remoting/zookeeper/listener.go @@ -65,6 +65,22 @@ func NewZkEventListener(client *gxzookeeper.ZookeeperClient) *ZkEventListener { } } +// waitForRetry waits for the given retry delay, or returns early when the +// listener is signaled to exit via l.exit. It reports whether the exit was +// signaled. The internal timer is always stopped, which avoids the +// timer/goroutine leak that time.After would cause when the exit path is +// taken (see #3558). +func (l *ZkEventListener) waitForRetry(delay time.Duration) bool { + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-timer.C: + return false + case <-l.exit: + return true + } +} + // ListenServiceNodeEvent listen a path node event func (l *ZkEventListener) ListenServiceNodeEvent(zkPath string, listener remoting.DataListener) { l.wg.Add(1) @@ -267,13 +283,10 @@ func (l *ZkEventListener) listenAllDirEvents(conf *common.URL, listener remoting } logger.Errorf("[Remoting][Zookeeper] get children of path {%s} with watcher failed, err=%v", rootPath, err) // Maybe the zookeeper does not ready yet, sleep failTimes * ConnDelay senconds to wait - after := time.After(timeSecondDuration(failTimes * ConnDelay)) - select { - case <-after: - continue - case <-l.exit: - return - } + if l.waitForRetry(timeSecondDuration(failTimes * ConnDelay)) { + return + } + continue } failTimes = 0 if len(children) == 0 { @@ -347,13 +360,10 @@ func (l *ZkEventListener) listenDirEvent(conf *common.URL, zkRootPath string, li logger.Errorf("[Remoting][Zookeeper] get children of path {%s} with watcher failed, err=%v", zkRootPath, err) } // Maybe the provider does not ready yet, sleep failTimes * ConnDelay senconds to wait - after := time.After(timeSecondDuration(failTimes * ConnDelay)) - select { - case <-after: - continue - case <-l.exit: - return - } + if l.waitForRetry(timeSecondDuration(failTimes * ConnDelay)) { + return + } + continue } failTimes = 0 if len(children) == 0 { diff --git a/remoting/zookeeper/listener_wait_test.go b/remoting/zookeeper/listener_wait_test.go new file mode 100644 index 000000000..96acfc0fc --- /dev/null +++ b/remoting/zookeeper/listener_wait_test.go @@ -0,0 +1,77 @@ +/* + * 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 zookeeper + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +// TestWaitForRetry_ExitSignaled verifies that an already-closed exit channel +// makes waitForRetry return true promptly instead of blocking on the timer. +func TestWaitForRetry_ExitSignaled(t *testing.T) { + l := &ZkEventListener{exit: make(chan struct{})} + close(l.exit) + + start := time.Now() + exited := l.waitForRetry(5 * time.Second) + elapsed := time.Since(start) + + assert.True(t, exited, "waitForRetry should report exit when l.exit is closed") + assert.Less(t, elapsed, 500*time.Millisecond, "waitForRetry should return immediately on a closed exit") +} + +// TestWaitForRetry_TimerFires verifies that, without an exit signal, +// waitForRetry waits for the delay and reports false. +func TestWaitForRetry_TimerFires(t *testing.T) { + l := &ZkEventListener{exit: make(chan struct{})} + + start := time.Now() + exited := l.waitForRetry(20 * time.Millisecond) + elapsed := time.Since(start) + + assert.False(t, exited, "waitForRetry should report false when the delay elapses") + assert.GreaterOrEqual(t, elapsed, 20*time.Millisecond, "should wait for the delay") + assert.Less(t, elapsed, 2*time.Second, "should not wait much longer than the delay") +} + +// TestWaitForRetry_ExitDuringWait verifies the exit path taken mid-wait: when +// l.exit is closed while waiting, waitForRetry returns true quickly and the +// internal timer is stopped, avoiding the timer/goroutine leak that time.After +// would cause (see #3558). +func TestWaitForRetry_ExitDuringWait(t *testing.T) { + l := &ZkEventListener{exit: make(chan struct{})} + + done := make(chan bool, 1) + go func() { + done <- l.waitForRetry(10 * time.Second) + }() + + // Let it start waiting on the timer. + time.Sleep(50 * time.Millisecond) + close(l.exit) + + select { + case exited := <-done: + assert.True(t, exited, "waitForRetry should return true once l.exit is closed") + case <-time.After(500 * time.Millisecond): + t.Fatal("waitForRetry did not return promptly after l.exit was closed") + } +}
