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 ea6686861d fix(plc4go): connection cache no longer bricks containers
or hands out dead connections
ea6686861d is described below
commit ea6686861d11c1de2c385e3dbdb6447c4a6dacb0
Author: Sebastian Rühl <[email protected]>
AuthorDate: Thu Jul 2 07:32:04 2026 +0200
fix(plc4go): connection cache no longer bricks containers or hands out dead
connections
The connection cache had four related defects that permanently broke a
connection string after transient endpoint failures (observed in the field
against connection-limited Modbus TCP gateways):
1. A failed initial connect left the container in StateInitialized forever.
connect() is only invoked at container creation and on lease return, so
subsequent lease requests queued with no reconnect ever attempted - every
caller timed out until process restart. connect() now marks the container
StateInvalid on failure.
2. lease() on a StateInvalid container sent nothing on either channel, so
callers silently timed out forever. It now queues the request and kicks
off a reconnect attempt, delivering either a fresh lease or the connect
error.
3. lease() handed out idle cached connections without any liveness check.
Remotes that drop idle connections (idle timeouts, connection-limited
gateways) caused the next client to fail on first use. Idle connections
are now checked with IsConnected() and replaced when dead.
4. Leases delivered to callers that had already given up (context cancel /
maxWaitTime) parked the connection in StateInUse with nobody left to
return it. Queued waiters with cancelled contexts are now skipped and
failed explicitly, and GetConnection drains abandoned lease channels on
the context-cancel path (previously only on the maxWaitTime path).
Also fixes a double-lease when returning an invalid connection with waiters
queued (both connect() and returnConnection() handed out the same fresh
connection) and a Debug log in onConnectionEvent that was missing its Msg()
call and never emitted.
---
plc4go/pkg/api/cache/PlcConnectionCache.go | 38 ++--
plc4go/pkg/api/cache/connectionContainer.go | 104 +++++++++--
plc4go/pkg/api/cache/connectionRecovery_test.go | 227 ++++++++++++++++++++++++
3 files changed, 347 insertions(+), 22 deletions(-)
diff --git a/plc4go/pkg/api/cache/PlcConnectionCache.go
b/plc4go/pkg/api/cache/PlcConnectionCache.go
index 85d0e3e7ec..4f39de4547 100644
--- a/plc4go/pkg/api/cache/PlcConnectionCache.go
+++ b/plc4go/pkg/api/cache/PlcConnectionCache.go
@@ -124,7 +124,10 @@ func (c *plcConnectionCache) onConnectionEvent(event
connectionEvent) {
if c.tracer != nil {
c.tracer.AddTrace("destroy-connection",
errorEvent.getError().Error())
}
- c.log.Debug().Str("connectionString",
connectionContainerInstance.connectionString)
+ c.log.Debug().
+ Str("connectionString",
connectionContainerInstance.connectionString).
+ Err(errorEvent.getError()).
+ Msg("Connection reported an error event")
}
}
@@ -180,10 +183,8 @@ func (c *plcConnectionCache) GetConnection(ctx
context.Context, connectionString
}
connChan, errChan := connection.lease(ctx)
maximumWaitTimeout := time.NewTimer(c.maxWaitTime)
+ defer maximumWaitTimeout.Stop()
select {
- case <-ctx.Done(): // abort on context cancel
- return nil, ctx.Err()
-
case conn := <-connChan: // Wait till we get a lease.
c.log.Debug().
Str("connectionString", connectionString).
@@ -201,18 +202,16 @@ func (c *plcConnectionCache) GetConnection(ctx
context.Context, connectionString
case err := <-errChan:
return nil, errors.Wrap(err, "error while trying to get lease
on connection")
- case <-ctx.Done():
+ case <-ctx.Done(): // abort on context cancel
+ // Drain the channels in the background: a lease may still be
delivered to
+ // the (buffered) channel after we've given up. Without
returning it, the
+ // connection would stay leased-to-nobody (StateInUse) forever.
+ c.drainAbandonedLease(connection, connChan, errChan)
return nil, ctx.Err()
case <-maximumWaitTimeout.C: // Timeout after the maximum waiting time.
// In this case we need to drain the chan and return it
immediate
- c.wg.Go(func() {
- select {
- case <-connChan:
- case <-errChan:
- }
- _ = connection.returnConnection(ctx, StateIdle)
- })
+ c.drainAbandonedLease(connection, connChan, errChan)
if c.tracer != nil {
c.tracer.AddTransactionalTrace(txId, "get-connection",
"timeout")
}
@@ -221,6 +220,21 @@ func (c *plcConnectionCache) GetConnection(ctx
context.Context, connectionString
}
}
+// drainAbandonedLease waits in the background for the outcome of a lease
request
+// whose caller has given up, and returns a delivered lease straight back to
the
+// container. Errors just get discarded. The container guarantees every queued
+// request eventually receives either a lease or an error (even on failed
+// connects), so this goroutine always terminates.
+func (c *plcConnectionCache) drainAbandonedLease(connection
*connectionContainer, connChan chan *plcConnectionLease, errChan chan error) {
+ c.wg.Go(func() {
+ select {
+ case <-connChan:
+ _ = connection.returnConnection(context.Background(),
StateIdle)
+ case <-errChan:
+ }
+ })
+}
+
func (c *plcConnectionCache) Close() error {
ctx := context.TODO()
c.log.Debug().Msg("Closing connection cache started.")
diff --git a/plc4go/pkg/api/cache/connectionContainer.go
b/plc4go/pkg/api/cache/connectionContainer.go
index 2a9009f328..99a3bef7a9 100644
--- a/plc4go/pkg/api/cache/connectionContainer.go
+++ b/plc4go/pkg/api/cache/connectionContainer.go
@@ -113,6 +113,12 @@ func (c *connectionContainer) connect(ctx context.Context)
{
} else {
c.log.Trace().Msg("no waiting clients")
}
+ // Mark the container as invalid so the next lease request
knows it has to
+ // re-attempt the connection. Previously the state was left
untouched
+ // (StateInitialized on the initial connect), which parked the
container in a
+ // state where lease() queued requests forever and no reconnect
was ever
+ // attempted - permanently breaking the connection string until
restart.
+ c.state = StateInvalid
return
}
@@ -127,13 +133,11 @@ func (c *connectionContainer) connect(ctx
context.Context) {
// Mark the connection as idle for now.
c.state = StateIdle
// If there is a request in the queue, hand out the connection to that.
- if waitingClientsLen := len(c.queue); waitingClientsLen > 0 {
- c.log.Trace().Int("waitingClientsLen",
waitingClientsLen).Msg("notifies waiting clients of connection")
- // Get the first in the queue.
- queueHead := c.queue[0]
- c.queue = c.queue[1:]
+ if queueHead := c.nextWaiter(); queueHead != nil {
+ c.log.Trace().Int("waitingClientsLen",
len(c.queue)).Msg("notifies waiting clients of connection")
// Mark the connection as being used.
c.state = StateInUse
+ c.leaseCounter++
// Return the lease to the caller.
connection := newPlcConnectionLease(c, c.leaseCounter,
c.connection)
// In this case we don'c need to check for blocks
@@ -145,6 +149,49 @@ func (c *connectionContainer) connect(ctx context.Context)
{
}
}
+// nextWaiter pops the next queued lease request whose context is still live.
+// Requests whose caller has already given up waiting are failed explicitly
(their
+// error channel is buffered) instead of being handed a lease: sending a lease
to
+// an abandoned request would park the connection in StateInUse forever, as
nobody
+// is left to return it. Must be called with c.lock held.
+func (c *connectionContainer) nextWaiter() *connectionRequest {
+ for len(c.queue) > 0 {
+ head := c.queue[0]
+ c.queue = c.queue[1:]
+ if head.ctx.Err() != nil {
+ c.log.Debug().Str("connectionString",
c.connectionString).
+ Msg("Skipping lease request with cancelled
context")
+ select {
+ case head.errChan <- head.ctx.Err():
+ default:
+ }
+ continue
+ }
+ return &head
+ }
+ return nil
+}
+
+// startReconnect discards any stale connection and re-runs connect
asynchronously.
+// State is set to StateInitialized so concurrent lease requests queue up
instead of
+// spawning additional connect attempts. Must be called with c.lock held.
+func (c *connectionContainer) startReconnect(ctx context.Context) {
+ stale := c.connection
+ c.connection = nil
+ c.state = StateInitialized
+ go func() {
+ if stale != nil {
+ // Close the stale connection so its message-codec
workers don't leak.
+ 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)
+ }()
+}
+
func (c *connectionContainer) addListener(listener connectionListener) {
// Get the lock.
c.lock.Lock()
@@ -162,6 +209,22 @@ func (c *connectionContainer) lease(ctx context.Context)
(chan *plcConnectionLea
// Check if the connection is available.
switch c.state {
case StateIdle:
+ // Verify the cached connection is still alive before handing
it out. The
+ // 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() {
+ 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.")
+ c.queue = append(c.queue, connectionRequest{ctx: ctx,
connChan: connectionChan, errChan: errorChan})
+ c.startReconnect(ctx)
+ break
+ }
c.leaseCounter++
connection := newPlcConnectionLease(c, c.leaseCounter,
c.connection)
c.state = StateInUse
@@ -179,7 +242,18 @@ func (c *connectionContainer) lease(ctx context.Context)
(chan *plcConnectionLea
Int("waiting-queue-size", len(c.queue)).
Msg("Added lease-request to queue.")
case StateInvalid:
- c.log.Debug().Str("connectionString",
c.connectionString).Msg("No lease because invalid")
+ // A previous connect or reconnect failed. Previously the
request was simply
+ // ignored here (nothing was ever sent on either channel), so
every caller
+ // blocked until timeout and the container stayed broken
forever. Instead,
+ // queue the request and attempt to re-establish the connection.
+ if c.closed {
+ errorChan <- errors.New("connection container is
closed")
+ break
+ }
+ c.log.Debug().Str("connectionString", c.connectionString).
+ Msg("Connection is invalid - attempting reconnect.")
+ c.queue = append(c.queue, connectionRequest{ctx: ctx, connChan:
connectionChan, errChan: errorChan})
+ c.startReconnect(ctx)
}
return connectionChan, errorChan
}
@@ -214,6 +288,17 @@ func (c *connectionContainer) returnConnection(ctx
context.Context, newState cac
}
}
c.connect(ctx)
+
+ c.lock.Lock()
+ defer c.lock.Unlock()
+ if c.connection == nil {
+ c.state = StateInvalid
+ return errors.New("Can'c 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
+ // another lease here would lease the same connection twice.
+ return nil
default:
c.log.Debug().Str("connectionString",
c.connectionString).Msg("Client returned valid connection.")
}
@@ -225,11 +310,10 @@ func (c *connectionContainer) returnConnection(ctx
context.Context, newState cac
}
// Check how many others are waiting for this connection.
- if waitingClientsLen := len(c.queue); waitingClientsLen > 0 {
- c.log.Trace().Int("waitingClientsLen",
waitingClientsLen).Msg("notifies waiting clients of connection return")
+ if next := c.nextWaiter(); next != nil {
+ c.log.Trace().Int("waitingClientsLen",
len(c.queue)).Msg("notifies waiting clients of connection return")
// There are waiting clients, give the connection to the next
client in the line.
- next := c.queue[0]
- c.queue = c.queue[1:]
+ c.state = StateInUse
c.leaseCounter++
connection := newPlcConnectionLease(c, c.leaseCounter,
c.connection)
diff --git a/plc4go/pkg/api/cache/connectionRecovery_test.go
b/plc4go/pkg/api/cache/connectionRecovery_test.go
new file mode 100644
index 0000000000..c21c00e072
--- /dev/null
+++ b/plc4go/pkg/api/cache/connectionRecovery_test.go
@@ -0,0 +1,227 @@
+/*
+ * 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"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/plc4x/plc4go/internal/simulated"
+ plc4go "github.com/apache/plc4x/plc4go/pkg/api"
+ "github.com/apache/plc4x/plc4go/pkg/api/config"
+ "github.com/apache/plc4x/plc4go/spi/options"
+ "github.com/apache/plc4x/plc4go/spi/testutils"
+)
+
+func testDriverManager(t *testing.T) plc4go.PlcDriverManager {
+ logger := testutils.ProduceTestingLogger(t)
+ driverManager :=
plc4go.NewPlcDriverManager(config.WithCustomLogger(logger))
+ t.Cleanup(func() {
+ assert.NoError(t, driverManager.Close())
+ })
+
driverManager.RegisterDriver(simulated.NewDriver(options.WithCustomLogger(logger)))
+ return driverManager
+}
+
+// A container whose initial connect failed must not ignore later lease
requests:
+// it has to re-attempt the connection (and deliver an error if that fails
again)
+// instead of leaving the caller to time out with nothing on either channel.
+func Test_connectionContainer_leaseAfterFailedConnectDeliversError(t
*testing.T) {
+ c := newConnectionContainer(testutils.ProduceTestingLogger(t),
testDriverManager(t), "simulated://1.2.3.4:42?connectionError=nope")
+
+ // Initial connect fails and must leave the container in StateInvalid.
+ c.connect(context.Background())
+ c.lock.RLock()
+ state := c.state
+ c.lock.RUnlock()
+ assert.Equal(t, StateInvalid, state)
+
+ // A subsequent lease request must trigger a reconnect attempt and get
the
+ // (failing) result delivered - not silence.
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ connChan, errChan := c.lease(ctx)
+ select {
+ case err := <-errChan:
+ assert.Error(t, err)
+ case conn := <-connChan:
+ t.Fatalf("expected an error, got a lease: %v", conn)
+ case <-ctx.Done():
+ t.Fatal("lease request on invalid container was silently
ignored (pre-fix behavior)")
+ }
+}
+
+// A container in StateInvalid whose endpoint has recovered must reconnect and
+// hand out a fresh lease on the next request.
+func Test_connectionContainer_leaseOnInvalidReconnects(t *testing.T) {
+ c := newConnectionContainer(testutils.ProduceTestingLogger(t),
testDriverManager(t), "simulated://1.2.3.4:42")
+ c.lock.Lock()
+ c.state = StateInvalid // simulate a previously failed connect/reconnect
+ c.lock.Unlock()
+
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ connChan, errChan := c.lease(ctx)
+ select {
+ case conn := <-connChan:
+ require.NotNil(t, conn)
+ assert.True(t, conn.IsConnected())
+ assert.NoError(t, conn.Close())
+ case err := <-errChan:
+ t.Fatalf("expected a lease, got error: %v", err)
+ case <-ctx.Done():
+ t.Fatal("lease request on invalid container was silently
ignored (pre-fix behavior)")
+ }
+}
+
+// An idle cached connection that died while parked in the cache must not be
+// handed out; the container has to establish a replacement.
+func Test_connectionContainer_deadIdleConnectionReplaced(t *testing.T) {
+ c := newConnectionContainer(testutils.ProduceTestingLogger(t),
testDriverManager(t), "simulated://1.2.3.4:42")
+ c.connect(context.Background())
+ c.lock.RLock()
+ state, firstConnection := c.state, c.connection
+ c.lock.RUnlock()
+ require.Equal(t, StateIdle, state)
+ require.NotNil(t, firstConnection)
+
+ // Simulate the remote dropping the idle connection.
+ firstConnection.Invalidate()
+ require.False(t, firstConnection.IsConnected())
+
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ connChan, errChan := c.lease(ctx)
+ select {
+ case conn := <-connChan:
+ require.NotNil(t, conn)
+ assert.True(t, conn.IsConnected(), "handed-out lease must be
alive")
+ assert.NoError(t, conn.Close())
+ case err := <-errChan:
+ t.Fatalf("expected a lease, got error: %v", err)
+ case <-ctx.Done():
+ t.Fatal("timed out waiting for a replacement connection")
+ }
+
+ c.lock.RLock()
+ secondConnection := c.connection
+ c.lock.RUnlock()
+ assert.NotSame(t, firstConnection, secondConnection, "dead connection
must have been replaced")
+}
+
+// Queued lease requests whose caller already gave up must be skipped when a
+// connection is handed out - otherwise the connection is leased to nobody and
+// stays StateInUse forever.
+func Test_connectionContainer_cancelledWaiterSkipped(t *testing.T) {
+ c := newConnectionContainer(testutils.ProduceTestingLogger(t),
testDriverManager(t), "simulated://1.2.3.4:42")
+ c.connect(context.Background())
+
+ // Take the connection so follow-up requests queue.
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ connChan, _ := c.lease(ctx)
+ lease := <-connChan
+
+ // Queue a request and abandon it.
+ abandonedCtx, abandonedCancel :=
context.WithCancel(context.Background())
+ abandonedConnChan, abandonedErrChan := c.lease(abandonedCtx)
+ abandonedCancel()
+
+ // Queue a live request behind it.
+ liveConnChan, liveErrChan := c.lease(ctx)
+
+ // Returning the lease must skip the abandoned waiter (failing it
explicitly)
+ // and serve the live one.
+ require.NoError(t, lease.Close())
+
+ select {
+ case conn := <-liveConnChan:
+ require.NotNil(t, conn)
+ assert.NoError(t, conn.Close())
+ case err := <-liveErrChan:
+ t.Fatalf("live waiter got error: %v", err)
+ case <-time.After(5 * time.Second):
+ t.Fatal("live waiter was starved - connection was likely leased
to the abandoned waiter")
+ }
+
+ // The abandoned waiter must have received an explicit error, not a
lease.
+ select {
+ case err := <-abandonedErrChan:
+ assert.Error(t, err)
+ case conn := <-abandonedConnChan:
+ t.Fatalf("abandoned waiter got a lease: %v", conn)
+ case <-time.After(5 * time.Second):
+ t.Fatal("abandoned waiter was never notified")
+ }
+}
+
+// End-to-end through the cache: a connection string whose first connect fails
+// must keep yielding prompt errors on every attempt - not one error and then
+// silent timeouts (pre-fix behavior: the container stayed StateInitialized
with
+// requests queueing forever, or StateInvalid with requests ignored).
+func Test_plcConnectionCache_failedFirstConnectDoesNotBrick(t *testing.T) {
+ cache := NewPlcConnectionCache(testDriverManager(t))
+ t.Cleanup(func() {
+ assert.NoError(t, cache.Close())
+ })
+
+ for i := range 3 {
+ ctx, cancel := context.WithTimeout(context.Background(),
5*time.Second)
+ start := time.Now()
+ _, err := cache.GetConnection(ctx,
"simulated://1.2.3.4:42?connectionError=nope")
+ cancel()
+ assert.Error(t, err, "attempt %d", i+1)
+ assert.NotErrorIs(t, err, context.DeadlineExceeded, "attempt %d
must fail fast with the connect error, not time out", i+1)
+ assert.Less(t, time.Since(start), 3*time.Second, "attempt %d
must not run into the timeout", i+1)
+ }
+}
+
+// Concurrent lease requests against an invalid container must result in
exactly
+// one reconnect attempt with everyone served (leases handed out sequentially).
+func Test_connectionContainer_concurrentLeaseOnInvalid(t *testing.T) {
+ c := newConnectionContainer(testutils.ProduceTestingLogger(t),
testDriverManager(t), "simulated://1.2.3.4:42")
+ c.lock.Lock()
+ c.state = StateInvalid
+ c.lock.Unlock()
+
+ const clients = 5
+ var wg sync.WaitGroup
+ for range clients {
+ wg.Go(func() {
+ ctx, cancel :=
context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ connChan, errChan := c.lease(ctx)
+ select {
+ case conn := <-connChan:
+ assert.NoError(t, conn.Close())
+ case err := <-errChan:
+ t.Errorf("expected a lease, got error: %v", err)
+ case <-ctx.Done():
+ t.Error("timed out waiting for lease")
+ }
+ })
+ }
+ wg.Wait()
+}
