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 ca4c256436 feat(plc4go): add WithMaxIdleTime to the connection cache
ca4c256436 is described below
commit ca4c25643618cb85a7aa21e7a5ce7ba6c92eb64e
Author: Sebastian Rühl <[email protected]>
AuthorDate: Thu Jul 2 17:36:21 2026 +0200
feat(plc4go): add WithMaxIdleTime to the connection cache
Remotes routinely reap idle TCP connections without a FIN/RST ever
reaching the client (half-open TCP) - a death that no liveness flag can
detect, as it only surfaces when the first write hits the dead socket.
Field data against Modbus TCP gateways shows pooled connections dying
this way between poll cycles, costing retry attempts on every affected
poll.
WithMaxIdleTime(d) discards cached connections that sat idle for longer
than d and re-establishes them transparently on the next lease (0 = keep
forever, the default - previous behavior is unchanged). The idle clock
starts when a connection becomes idle (after connect and on lease
return). As a side effect, connection slots on connection-limited
remotes are freed between bursts instead of being parked indefinitely.
---
plc4go/pkg/api/cache/PlcConnectionCache.go | 17 +++
plc4go/pkg/api/cache/connectionContainer.go | 40 +++++-
plc4go/pkg/api/cache/connectionIdleTTL_test.go | 192 +++++++++++++++++++++++++
3 files changed, 244 insertions(+), 5 deletions(-)
diff --git a/plc4go/pkg/api/cache/PlcConnectionCache.go
b/plc4go/pkg/api/cache/PlcConnectionCache.go
index 4f39de4547..d0aaf4d316 100644
--- a/plc4go/pkg/api/cache/PlcConnectionCache.go
+++ b/plc4go/pkg/api/cache/PlcConnectionCache.go
@@ -77,6 +77,19 @@ func WithMaxWaitTime(maxWaitTime time.Duration)
WithConnectionCacheOption {
}
}
+// WithMaxIdleTime discards cached connections that sat idle for longer than
+// the given duration and re-establishes them on the next lease (0 = keep
+// forever, the default). Use this against remotes that silently reap idle
+// connections (half-open TCP): such a death is undetectable until the first
+// write fails, so connections past this age are replaced proactively. As a
+// side effect, connection slots on connection-limited remotes are freed
+// between bursts instead of being parked indefinitely.
+func WithMaxIdleTime(maxIdleTime time.Duration) WithConnectionCacheOption {
+ return func(plcConnectionCache *plcConnectionCache) {
+ plcConnectionCache.maxIdleTime = maxIdleTime
+ }
+}
+
func WithTracer() WithConnectionCacheOption {
return func(plcConnectionCache *plcConnectionCache) {
plcConnectionCache.EnableTracer()
@@ -107,6 +120,9 @@ type plcConnectionCache struct {
// If the connection is used for a longer time, it is forcefully
removed from the client.
maxLeaseTime time.Duration
maxWaitTime time.Duration
+ // Maximum duration a connection may sit idle before being replaced on
the
+ // next lease (0 = keep forever). See WithMaxIdleTime.
+ maxIdleTime time.Duration
cacheLock *sync.RWMutex
connections map[string]*connectionContainer
@@ -160,6 +176,7 @@ func (c *plcConnectionCache) GetConnection(ctx
context.Context, connectionString
c.log.Debug().Str("connectionString",
connectionString).Msg("Create new cached connection")
// Create a new connection container.
cc := newConnectionContainer(c.log, c.driverManager,
connectionString)
+ cc.maxIdleTime = c.maxIdleTime
// Register for connection events (Like connection closed or
error).
cc.addListener(c)
// Store the new connection container in the cache of
connections.
diff --git a/plc4go/pkg/api/cache/connectionContainer.go
b/plc4go/pkg/api/cache/connectionContainer.go
index 99a3bef7a9..4938a0c531 100644
--- a/plc4go/pkg/api/cache/connectionContainer.go
+++ b/plc4go/pkg/api/cache/connectionContainer.go
@@ -23,6 +23,7 @@ import (
"context"
"fmt"
"sync"
+ "time"
"github.com/rs/zerolog"
@@ -44,6 +45,16 @@ type connectionContainer struct {
queue []connectionRequest
// Listeners for connection events.
listeners []connectionListener
+ // Maximum duration a connection may sit idle in the cache before it is
+ // discarded and re-established on the next lease (0 = keep forever).
+ // Remotes routinely reap idle connections without a FIN/RST reaching us
+ // (half-open TCP), which no liveness flag can detect - the death only
+ // surfaces on the first write. Refusing to hand out connections that
+ // exceeded this age avoids that failure mode and frees connection slots
+ // on connection-limited remotes between bursts.
+ maxIdleTime time.Duration
+ // idleSince records when the connection last became idle.
+ idleSince time.Time
log zerolog.Logger
}
@@ -132,6 +143,7 @@ func (c *connectionContainer) connect(ctx context.Context) {
c.tracerEnabled = c.connection.IsTraceEnabled()
// Mark the connection as idle for now.
c.state = StateIdle
+ c.idleSince = time.Now()
// If there is a request in the queue, hand out the connection to that.
if queueHead := c.nextWaiter(); queueHead != nil {
c.log.Trace().Int("waitingClientsLen",
len(c.queue)).Msg("notifies waiting clients of connection")
@@ -213,14 +225,25 @@ func (c *connectionContainer) lease(ctx context.Context)
(chan *plcConnectionLea
// remote may have dropped it while it sat idle in the cache
(idle timeouts
// and connection-limited gateways do this routinely); without
this check
// the client receives a dead connection and fails on first use.
- if c.connection == nil || !c.connection.IsConnected() {
+ // A connection that exceeded the configured max idle time is
treated the
+ // same way: remotes that silently reap idle connections leave
a half-open
+ // socket that IsConnected() cannot detect, so past that age
the connection
+ // is not to be trusted and gets replaced proactively.
+ if c.connection == nil || !c.connection.IsConnected() ||
c.idleExpired() {
if c.closed {
// The cache is shutting down - don't
reconnect, just report.
errorChan <- errors.New("connection container
is closed")
break
}
- c.log.Debug().Str("connectionString",
c.connectionString).
- Msg("Cached idle connection is no longer alive
- reconnecting.")
+ if c.idleExpired() {
+ c.log.Debug().Str("connectionString",
c.connectionString).
+ Dur("maxIdleTime", c.maxIdleTime).
+ Time("idleSince", c.idleSince).
+ Msg("Cached idle connection exceeded
max idle time - reconnecting.")
+ } else {
+ c.log.Debug().Str("connectionString",
c.connectionString).
+ Msg("Cached idle connection is no
longer alive - reconnecting.")
+ }
c.queue = append(c.queue, connectionRequest{ctx: ctx,
connChan: connectionChan, errChan: errorChan})
c.startReconnect(ctx)
break
@@ -293,7 +316,7 @@ func (c *connectionContainer) returnConnection(ctx
context.Context, newState cac
defer c.lock.Unlock()
if c.connection == nil {
c.state = StateInvalid
- return errors.New("Can'c return a broken connection")
+ return errors.New("Can't return a broken connection")
}
// connect() already dealt with the queue: it either handed the
fresh
// connection to the next waiter (StateInUse) or marked it
idle. Handing out
@@ -306,7 +329,7 @@ func (c *connectionContainer) returnConnection(ctx
context.Context, newState cac
defer c.lock.Unlock()
if c.connection == nil {
c.state = StateInvalid
- return errors.New("Can'c return a broken connection")
+ return errors.New("Can't return a broken connection")
}
// Check how many others are waiting for this connection.
@@ -329,10 +352,17 @@ func (c *connectionContainer) returnConnection(ctx
context.Context, newState cac
c.log.Debug().Str("connectionString", c.connectionString).
Msg("Connection set to 'idle'.")
c.state = StateIdle
+ c.idleSince = time.Now()
}
return nil
}
+// idleExpired reports whether the connection outstayed the configured max idle
+// time. Must be called with c.lock held.
+func (c *connectionContainer) idleExpired() bool {
+ return c.maxIdleTime > 0 && time.Since(c.idleSince) > c.maxIdleTime
+}
+
func (c *connectionContainer) String() string {
return fmt.Sprintf("connectionContainer{%s:%s, leaseCounter: %d,
closed: %t, state: %s}", c.connectionString, c.connection, c.leaseCounter,
c.closed, c.state)
}
diff --git a/plc4go/pkg/api/cache/connectionIdleTTL_test.go
b/plc4go/pkg/api/cache/connectionIdleTTL_test.go
new file mode 100644
index 0000000000..47b30e130b
--- /dev/null
+++ b/plc4go/pkg/api/cache/connectionIdleTTL_test.go
@@ -0,0 +1,192 @@
+/*
+ * 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
+ *
+ * https://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 cache
+
+import (
+ "context"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/plc4x/plc4go/spi/testutils"
+)
+
+// leaseWithTimeout requests a lease and waits for the outcome.
+func leaseWithTimeout(t *testing.T, c *connectionContainer)
*plcConnectionLease {
+ t.Helper()
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ connChan, errChan := c.lease(ctx)
+ select {
+ case conn := <-connChan:
+ return conn
+ case err := <-errChan:
+ t.Fatalf("expected a lease, got error: %v", err)
+ case <-ctx.Done():
+ t.Fatal("timed out waiting for lease")
+ }
+ return nil
+}
+
+// An idle connection older than maxIdleTime must not be handed out - the
+// container has to discard it and establish a replacement. Remotes that
+// silently reap idle connections leave a half-open socket that IsConnected()
+// cannot detect; past the configured age the connection is not to be trusted.
+func Test_connectionContainer_idleExpiredConnectionReplaced(t *testing.T) {
+ c := newConnectionContainer(testutils.ProduceTestingLogger(t),
testDriverManager(t), "simulated://1.2.3.4:42")
+ c.maxIdleTime = time.Minute
+ c.connect(context.Background())
+ c.lock.RLock()
+ firstConnection := c.connection
+ c.lock.RUnlock()
+ require.NotNil(t, firstConnection)
+
+ // Age the idle connection past the TTL.
+ c.lock.Lock()
+ c.idleSince = time.Now().Add(-2 * time.Minute)
+ c.lock.Unlock()
+
+ lease := leaseWithTimeout(t, c)
+ require.NotNil(t, lease)
+ assert.True(t, lease.IsConnected(), "handed-out lease must be alive")
+ assert.NoError(t, lease.Close())
+
+ c.lock.RLock()
+ secondConnection := c.connection
+ c.lock.RUnlock()
+ assert.NotSame(t, firstConnection, secondConnection, "expired idle
connection must have been replaced")
+}
+
+// A connection younger than maxIdleTime is reused as-is.
+func Test_connectionContainer_idleNotExpiredKeepsConnection(t *testing.T) {
+ c := newConnectionContainer(testutils.ProduceTestingLogger(t),
testDriverManager(t), "simulated://1.2.3.4:42")
+ c.maxIdleTime = time.Hour
+ c.connect(context.Background())
+ c.lock.RLock()
+ firstConnection := c.connection
+ c.lock.RUnlock()
+ require.NotNil(t, firstConnection)
+
+ lease := leaseWithTimeout(t, c)
+ require.NotNil(t, lease)
+ assert.NoError(t, lease.Close())
+
+ c.lock.RLock()
+ secondConnection := c.connection
+ c.lock.RUnlock()
+ assert.Same(t, firstConnection, secondConnection, "young idle
connection must be reused")
+}
+
+// maxIdleTime = 0 (the default) disables the TTL entirely - connections are
+// kept forever, preserving the previous behavior.
+func Test_connectionContainer_maxIdleTimeDisabledByDefault(t *testing.T) {
+ c := newConnectionContainer(testutils.ProduceTestingLogger(t),
testDriverManager(t), "simulated://1.2.3.4:42")
+ c.connect(context.Background())
+ c.lock.RLock()
+ firstConnection := c.connection
+ c.lock.RUnlock()
+ require.NotNil(t, firstConnection)
+
+ // Even an ancient idle timestamp must not trigger a replacement.
+ c.lock.Lock()
+ c.idleSince = time.Now().Add(-24 * time.Hour)
+ c.lock.Unlock()
+
+ lease := leaseWithTimeout(t, c)
+ require.NotNil(t, lease)
+ assert.NoError(t, lease.Close())
+
+ c.lock.RLock()
+ secondConnection := c.connection
+ c.lock.RUnlock()
+ assert.Same(t, firstConnection, secondConnection, "TTL disabled:
connection must be kept")
+}
+
+// Returning a lease must restart the idle clock.
+func Test_connectionContainer_returnResetsIdleClock(t *testing.T) {
+ c := newConnectionContainer(testutils.ProduceTestingLogger(t),
testDriverManager(t), "simulated://1.2.3.4:42")
+ c.maxIdleTime = time.Minute
+ c.connect(context.Background())
+
+ // Age the container, then take and return a lease - the return must
+ // refresh idleSince so the connection is trusted again.
+ c.lock.Lock()
+ c.idleSince = time.Now().Add(-2 * time.Minute)
+ c.lock.Unlock()
+
+ lease := leaseWithTimeout(t, c) // triggers replacement (expired)
+ require.NotNil(t, lease)
+ require.NoError(t, lease.Close())
+
+ c.lock.RLock()
+ idleSince := c.idleSince
+ connection := c.connection
+ c.lock.RUnlock()
+ assert.WithinDuration(t, time.Now(), idleSince, 10*time.Second, "idle
clock must restart on return")
+
+ // An immediate follow-up lease reuses the same connection.
+ lease2 := leaseWithTimeout(t, c)
+ require.NotNil(t, lease2)
+ assert.NoError(t, lease2.Close())
+ c.lock.RLock()
+ assert.Same(t, connection, c.connection)
+ c.lock.RUnlock()
+}
+
+// End-to-end through the cache: WithMaxIdleTime propagates to containers and
+// an aged connection is transparently replaced on the next GetConnection.
+func Test_plcConnectionCache_withMaxIdleTime(t *testing.T) {
+ cache := NewPlcConnectionCache(testDriverManager(t),
WithMaxIdleTime(time.Minute)).(*plcConnectionCache)
+ t.Cleanup(func() {
+ assert.NoError(t, cache.Close())
+ })
+
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ conn, err := cache.GetConnection(ctx, "simulated://1.2.3.4:42")
+ require.NoError(t, err)
+ require.NoError(t, conn.Close())
+
+ cache.cacheLock.RLock()
+ container := cache.connections["simulated://1.2.3.4:42"]
+ cache.cacheLock.RUnlock()
+ require.NotNil(t, container)
+ assert.Equal(t, time.Minute, container.maxIdleTime, "option must
propagate to containers")
+
+ // Age the pooled connection and lease again - must succeed on a fresh
one.
+ container.lock.Lock()
+ firstConnection := container.connection
+ container.idleSince = time.Now().Add(-2 * time.Minute)
+ container.lock.Unlock()
+
+ ctx2, cancel2 := context.WithTimeout(context.Background(),
5*time.Second)
+ defer cancel2()
+ conn2, err := cache.GetConnection(ctx2, "simulated://1.2.3.4:42")
+ require.NoError(t, err)
+ assert.True(t, conn2.IsConnected())
+ require.NoError(t, conn2.Close())
+
+ container.lock.RLock()
+ secondConnection := container.connection
+ container.lock.RUnlock()
+ assert.NotSame(t, firstConnection, secondConnection, "aged connection
must have been replaced")
+}
