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 162260187a feat(plc4go): exempt subscription-carrying connections from
cache idle TTL
162260187a is described below
commit 162260187a33a319fc6123fd46581bf0b6dccaa7
Author: Sebastian Rühl <[email protected]>
AuthorDate: Wed Jul 15 13:57:43 2026 +0200
feat(plc4go): exempt subscription-carrying connections from cache idle TTL
WithMaxIdleTime reaps connections that sat idle in the cache past the
configured age. For connections carrying stateful subscriptions (BACnet
COV: server-side registrations plus client-side lifetime/2 refresh
timers) a reap silently destroys that state - passive updates stop until
the client happens to re-subscribe, which may be hours away.
The cache now consults an optional capability interface
(ActiveSubscriptionCount) before declaring an idle connection expired:
connections reporting active subscription handles are kept regardless of
age. The BACnet connection implements the capability by summing live
handles across its subscribers (newly mutex-guarded, as the cache calls
in from outside the connect path). Drivers without the capability keep
the existing TTL behavior unchanged.
---
plc4go/internal/bacnetip/Connection.go | 19 ++++
.../bacnetip/ConnectionSubscriptionCount_test.go | 48 ++++++++++
plc4go/internal/bacnetip/Subscriber.go | 8 ++
plc4go/pkg/api/cache/PlcConnectionCache.go | 5 +
plc4go/pkg/api/cache/connectionContainer.go | 24 ++++-
.../cache/connectionIdleTTLSubscription_test.go | 101 +++++++++++++++++++++
6 files changed, 204 insertions(+), 1 deletion(-)
diff --git a/plc4go/internal/bacnetip/Connection.go
b/plc4go/internal/bacnetip/Connection.go
index 2500ae2f61..227e30ae6c 100644
--- a/plc4go/internal/bacnetip/Connection.go
+++ b/plc4go/internal/bacnetip/Connection.go
@@ -49,6 +49,7 @@ type Connection struct {
configuration Configuration
driverContext DriverContext
subscribers []*Subscriber
+ subscribersMu sync.Mutex
tm transactions.RequestTransactionManager
connectionId string
@@ -252,6 +253,8 @@ func (c *Connection) UnsubscriptionRequestBuilder()
apiModel.PlcUnsubscriptionRe
}
func (c *Connection) addSubscriber(subscriber *Subscriber) {
+ c.subscribersMu.Lock()
+ defer c.subscribersMu.Unlock()
if slices.Contains(c.subscribers, subscriber) {
c.log.Debug().Interface("subscriber",
subscriber).Msg("Subscriber already added")
return
@@ -259,6 +262,22 @@ func (c *Connection) addSubscriber(subscriber *Subscriber)
{
c.subscribers = append(c.subscribers, subscriber)
}
+// ActiveSubscriptionCount reports how many COV subscription handles are
+// currently registered across this connection's subscribers. The connection
+// cache consults it (as an optional capability) to exempt subscription-
+// carrying connections from idle reaping: their server-side COV
+// registrations and refresh timers live on this connection and would be
+// destroyed by a reap.
+func (c *Connection) ActiveSubscriptionCount() int {
+ c.subscribersMu.Lock()
+ defer c.subscribersMu.Unlock()
+ count := 0
+ for _, s := range c.subscribers {
+ count += s.activeHandleCount()
+ }
+ return count
+}
+
func (c *Connection) String() string {
return fmt.Sprintf("bacnetip.Connection")
}
diff --git a/plc4go/internal/bacnetip/ConnectionSubscriptionCount_test.go
b/plc4go/internal/bacnetip/ConnectionSubscriptionCount_test.go
new file mode 100644
index 0000000000..92620d827e
--- /dev/null
+++ b/plc4go/internal/bacnetip/ConnectionSubscriptionCount_test.go
@@ -0,0 +1,48 @@
+/*
+ * 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 bacnetip
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+)
+
+// ActiveSubscriptionCount feeds the connection cache's subscription-aware
+// idle reaping: it must reflect the live COV handle count across all
+// subscribers of the connection.
+func TestConnection_ActiveSubscriptionCount(t *testing.T) {
+ conn := &Connection{}
+ assert.Zero(t, conn.ActiveSubscriptionCount(), "fresh connection has no
subscriptions")
+
+ s1 := NewSubscriber(conn)
+ s2 := NewSubscriber(conn)
+ conn.addSubscriber(s1)
+ conn.addSubscriber(s2)
+ assert.Zero(t, conn.ActiveSubscriptionCount(), "subscribers without
handles don't count")
+
+ s1.storeHandle(&SubscriptionHandle{subscriberProcessId: 1})
+ s1.storeHandle(&SubscriptionHandle{subscriberProcessId: 2})
+ s2.storeHandle(&SubscriptionHandle{subscriberProcessId: 7})
+ assert.Equal(t, 3, conn.ActiveSubscriptionCount(), "handles sum across
subscribers")
+
+ s1.removeHandle(2)
+ assert.Equal(t, 2, conn.ActiveSubscriptionCount(), "removed handles
drop out of the count")
+}
diff --git a/plc4go/internal/bacnetip/Subscriber.go
b/plc4go/internal/bacnetip/Subscriber.go
index 1bf90f11ba..ee7ff04c47 100644
--- a/plc4go/internal/bacnetip/Subscriber.go
+++ b/plc4go/internal/bacnetip/Subscriber.go
@@ -354,6 +354,14 @@ func (m *Subscriber) lookupHandle(processId uint32)
*SubscriptionHandle {
return m.handles[processId]
}
+// activeHandleCount reports the number of currently registered subscription
+// handles (feeds Connection.ActiveSubscriptionCount).
+func (m *Subscriber) activeHandleCount() int {
+ m.handlesMu.RLock()
+ defer m.handlesMu.RUnlock()
+ return len(m.handles)
+}
+
// findHandle locates the SubscriptionHandle that owns the given
// apiModel.PlcSubscriptionHandle (returned earlier in the Subscribe response).
// Used by Unsubscribe which only has the api-level handle in hand.
diff --git a/plc4go/pkg/api/cache/PlcConnectionCache.go
b/plc4go/pkg/api/cache/PlcConnectionCache.go
index d0aaf4d316..3f19b54b2f 100644
--- a/plc4go/pkg/api/cache/PlcConnectionCache.go
+++ b/plc4go/pkg/api/cache/PlcConnectionCache.go
@@ -84,6 +84,11 @@ func WithMaxWaitTime(maxWaitTime time.Duration)
WithConnectionCacheOption {
// 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.
+//
+// Connections that report active subscription handles (see the drivers'
+// subscription support, e.g. BACnet COV) are exempt from the TTL: their
+// subscription state lives on the connection and would be silently destroyed
+// by a reap, cutting off passive updates until the client re-subscribes.
func WithMaxIdleTime(maxIdleTime time.Duration) WithConnectionCacheOption {
return func(plcConnectionCache *plcConnectionCache) {
plcConnectionCache.maxIdleTime = maxIdleTime
diff --git a/plc4go/pkg/api/cache/connectionContainer.go
b/plc4go/pkg/api/cache/connectionContainer.go
index 4938a0c531..5162be623d 100644
--- a/plc4go/pkg/api/cache/connectionContainer.go
+++ b/plc4go/pkg/api/cache/connectionContainer.go
@@ -357,10 +357,32 @@ func (c *connectionContainer) returnConnection(ctx
context.Context, newState cac
return nil
}
+// subscriptionAware is an optional capability of cached connections: drivers
+// whose connections carry stateful subscriptions (e.g. BACnet COV, with its
+// server-side registrations and client-side refresh timers) implement it so
+// the cache can tell "parked but working" apart from "abandoned".
+type subscriptionAware interface {
+ // ActiveSubscriptionCount reports the number of currently active
+ // subscription handles on the connection.
+ ActiveSubscriptionCount() int
+}
+
// idleExpired reports whether the connection outstayed the configured max idle
// time. Must be called with c.lock held.
+//
+// A connection with active subscription handles never expires: its
+// subscription state lives on the connection, so reaping it would silently
+// destroy server-side registrations and refresh timers, cutting off passive
+// updates until the client happens to re-subscribe. Such a connection is not
+// "idle" in any meaningful sense even when no lease touched it for a while.
func (c *connectionContainer) idleExpired() bool {
- return c.maxIdleTime > 0 && time.Since(c.idleSince) > c.maxIdleTime
+ if c.maxIdleTime <= 0 || time.Since(c.idleSince) <= c.maxIdleTime {
+ return false
+ }
+ if sa, ok := c.connection.(subscriptionAware); ok &&
sa.ActiveSubscriptionCount() > 0 {
+ return false
+ }
+ return true
}
func (c *connectionContainer) String() string {
diff --git a/plc4go/pkg/api/cache/connectionIdleTTLSubscription_test.go
b/plc4go/pkg/api/cache/connectionIdleTTLSubscription_test.go
new file mode 100644
index 0000000000..971f9d1eb6
--- /dev/null
+++ b/plc4go/pkg/api/cache/connectionIdleTTLSubscription_test.go
@@ -0,0 +1,101 @@
+/*
+ * 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"
+)
+
+// stubSubscribedConnection implements tracedPlcConnection via interface
+// embedding (unimplemented methods panic if called) and reports a fixed
+// number of active subscription handles.
+type stubSubscribedConnection struct {
+ tracedPlcConnection
+ activeSubscriptions int
+}
+
+func (s *stubSubscribedConnection) ActiveSubscriptionCount() int { return
s.activeSubscriptions }
+func (s *stubSubscribedConnection) IsConnected() bool { return true
}
+func (s *stubSubscribedConnection) Close() error { return nil }
+func (s *stubSubscribedConnection) IsTraceEnabled() bool { return
false }
+
+// A connection carrying active subscription handles must never be reaped by
+// the idle TTL: its subscription state (e.g. BACnet COV refresh timers) lives
+// on the connection and would be silently destroyed, cutting off passive
+// updates until the client happens to re-subscribe.
+func Test_connectionContainer_idleExpiredSkipsSubscribedConnection(t
*testing.T) {
+ c := newConnectionContainer(testutils.ProduceTestingLogger(t),
testDriverManager(t), "simulated://1.2.3.4:42")
+ c.maxIdleTime = time.Minute
+
+ stub := &stubSubscribedConnection{activeSubscriptions: 1}
+ c.lock.Lock()
+ c.connection = stub
+ c.state = StateIdle
+ c.idleSince = time.Now().Add(-2 * time.Minute) // aged past the TTL
+ c.lock.Unlock()
+
+ c.lock.RLock()
+ expired := c.idleExpired()
+ c.lock.RUnlock()
+ assert.False(t, expired, "aged connection with active subscriptions
must not be considered expired")
+
+ // The same aged connection without subscriptions expires as usual.
+ stub.activeSubscriptions = 0
+ c.lock.RLock()
+ expired = c.idleExpired()
+ c.lock.RUnlock()
+ assert.True(t, expired, "aged connection without subscriptions must
expire")
+}
+
+// Lease-level behavior: an aged-but-subscribed connection is handed out as-is
+// instead of being replaced.
+func Test_connectionContainer_leaseKeepsAgedSubscribedConnection(t *testing.T)
{
+ c := newConnectionContainer(testutils.ProduceTestingLogger(t),
testDriverManager(t), "simulated://1.2.3.4:42")
+ c.maxIdleTime = time.Minute
+
+ stub := &stubSubscribedConnection{activeSubscriptions: 3}
+ c.lock.Lock()
+ c.connection = stub
+ c.state = StateIdle
+ c.idleSince = time.Now().Add(-2 * time.Minute)
+ c.lock.Unlock()
+
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ connChan, errChan := c.lease(ctx)
+ select {
+ case lease := <-connChan:
+ require.NotNil(t, lease)
+ c.lock.RLock()
+ assert.Same(t, stub, c.connection, "subscribed connection must
be kept across the lease")
+ c.lock.RUnlock()
+ case err := <-errChan:
+ t.Fatalf("expected a lease, got error: %v", err)
+ case <-ctx.Done():
+ t.Fatal("timed out waiting for lease")
+ }
+}
