This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
The following commit(s) were added to refs/heads/develop by this push:
new 1b612551f7 fix(plc4go): stop leaking codec workers on cached reconnect
1b612551f7 is described below
commit 1b612551f7266f9e505b1c609648546190303207
Author: Sebastian Rühl <[email protected]>
AuthorDate: Wed Jun 17 16:44:12 2026 +0200
fix(plc4go): stop leaking codec workers on cached reconnect
The connection cache's returnConnection() reconnected an invalidated
connection via connect(), which overwrites c.connection with a freshly
established one without closing the old one. The stale connection's
DefaultCodec ReceiveWork/ExpireWork goroutines kept running, leaking a pair per
invalidate->reconnect cycle. Against an endpoint that accepts TCP but is
unresponsive at the protocol layer this recurs every poll and grows without
bound (observed ~40k of each worker on a wedged gateway).
Close the stale connection before reconnecting so the codec is
disconnected and its workers terminate.
Also add the missing ctx.Done() case to DefaultCodec.ExpireWork's idle
select. The "no expectations" branch waited only on notifyExpireWorker or a 30s
timer, so the worker ignored context cancellation there (ReceiveWork and
ExpireWork's active branch already handle ctx.Done()).
---
plc4go/pkg/api/cache/connectionContainer.go | 17 +++++++++++++++++
plc4go/spi/default/DefaultCodec.go | 3 +++
2 files changed, 20 insertions(+)
diff --git a/plc4go/pkg/api/cache/connectionContainer.go
b/plc4go/pkg/api/cache/connectionContainer.go
index 6226e14b2e..2a9009f328 100644
--- a/plc4go/pkg/api/cache/connectionContainer.go
+++ b/plc4go/pkg/api/cache/connectionContainer.go
@@ -196,6 +196,23 @@ func (c *connectionContainer) returnConnection(ctx
context.Context, newState cac
Str("connectionString", c.connectionString).
Stringer("newState", newState).
Msg("Client returned a connection, reconnecting.")
+ // Close the stale connection before reconnecting. c.connect()
overwrites
+ // c.connection with a freshly-established one, so without this
the previous
+ // connection's message-codec workers (ReceiveWork/ExpireWork)
keep running,
+ // leaking a pair of goroutines on every invalidate->reconnect
cycle. Against
+ // an endpoint that accepts TCP but is unresponsive at the
protocol layer this
+ // recurs every poll and grows without bound.
+ c.lock.Lock()
+ stale := c.connection
+ c.connection = nil
+ c.lock.Unlock()
+ if stale != nil {
+ if err := stale.Close(); err != nil {
+ c.log.Debug().Err(err).
+ Str("connectionString",
c.connectionString).
+ Msg("Error closing stale connection
before reconnect")
+ }
+ }
c.connect(ctx)
default:
c.log.Debug().Str("connectionString",
c.connectionString).Msg("Client returned valid connection.")
diff --git a/plc4go/spi/default/DefaultCodec.go
b/plc4go/spi/default/DefaultCodec.go
index 729dc3d5a2..aa61c14fc8 100644
--- a/plc4go/spi/default/DefaultCodec.go
+++ b/plc4go/spi/default/DefaultCodec.go
@@ -374,6 +374,9 @@ mainLoop:
select {
case <-m.notifyExpireWorker:
workerLog.Trace().Msg("waking up because of
notification")
+ case <-m.ctx.Done():
+ workerLog.Trace().Msg("context done, exiting
expire work")
+ return
case <-timer.C:
workerLog.Trace().Msg("waking up for next
expire")
}
