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 5238b33c61 fix(plc4go): make UDP TransportInstance.Reset a no-op
(lease-time Reset raced the receive worker and ate inbound datagrams)
5238b33c61 is described below
commit 5238b33c611e751f0e50e02f83a8db001fdfe282
Author: Sebastian Rühl <[email protected]>
AuthorDate: Wed Jul 15 14:55:15 2026 +0200
fix(plc4go): make UDP TransportInstance.Reset a no-op (lease-time Reset
raced the receive worker and ate inbound datagrams)
Reset() poked the read deadline, drained one datagram into a discard
buffer, and swapped m.reader - all unsynchronized, and all invoked from
outside the receive worker: the connection cache runs Reset on every
successful lease, and DefaultCodec runs it on retryable errors. The
receive worker concurrently drives GetNumBytesAvailableInBuffer/Peek on
the same socket and bufio.Reader, so the drain could eat a fresh
datagram off the socket and the reader swap could throw away bytes the
worker had just buffered.
Field evidence (BACnet): a WriteProperty SimpleAck reached the host
interface with a good checksum 0.8ms after the request, was never
surfaced by the codec, and the write hung until its expectation
expired. The new roundtrip tests reproduce the exact production shape -
a no-TTL connection cache re-leasing the same live connection (Reset on
a live socket under an active 10ms receive poll) - and fail under
-race against the old Reset.
For UDP nothing needs resetting: datagram framing means stale buffered
data is just a complete late reply, which expectation matching already
drops. Reconnection semantics stay in Close/Connect. The TCP transport
has the same unlocked pattern but stream framing makes a plain no-op
wrong there; that remains open.
---
.../bacnetip/LeaseResetWriteRoundtrip_test.go | 137 +++++++++++++++++++++
plc4go/spi/transports/udp/TransportInstance.go | 25 ++--
2 files changed, 154 insertions(+), 8 deletions(-)
diff --git a/plc4go/internal/bacnetip/LeaseResetWriteRoundtrip_test.go
b/plc4go/internal/bacnetip/LeaseResetWriteRoundtrip_test.go
new file mode 100644
index 0000000000..c1cce86325
--- /dev/null
+++ b/plc4go/internal/bacnetip/LeaseResetWriteRoundtrip_test.go
@@ -0,0 +1,137 @@
+/*
+ * 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 (
+ "context"
+ "fmt"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ plc4go "github.com/apache/plc4x/plc4go/pkg/api"
+ "github.com/apache/plc4x/plc4go/pkg/api/cache"
+ apiModel "github.com/apache/plc4x/plc4go/pkg/api/model"
+ apiTransports "github.com/apache/plc4x/plc4go/pkg/api/transports"
+ "github.com/apache/plc4x/plc4go/spi/options"
+ "github.com/apache/plc4x/plc4go/spi/testutils"
+)
+
+// TestNativeBacnetWrite_LeaseResetOnLiveSocket covers the one cache path the
+// earlier roundtrip tests missed: a cache WITHOUT maxIdleTime (the production
+// default) re-leases the SAME live connection, and every successful lease runs
+// TransportInstance.Reset() — deadline poke, drain read, bufio.Reader swap —
+// concurrently with the codec's 10ms receive-poll worker on the same socket.
+// The write's SimpleAck (arriving ~1ms after TX, like the field pcap) must
+// still resolve Execute on every iteration.
+//
+// Field shape (CI run 29414550340): read lease, return, idle gap, write lease
+// (Reset on live socket), TX +300ms, ack +0.8ms — ack visible on the host
+// interface with a good checksum but never surfaced by the codec.
+func TestNativeBacnetWrite_LeaseResetOnLiveSocket(t *testing.T) {
+ log := testutils.ProduceTestingLogger(t)
+ device := startFakeWriteDevice(t, log)
+ t.Cleanup(device.stop)
+
+ dm := plc4go.NewPlcDriverManager(options.WithCustomLogger(log))
+ dm.RegisterDriver(NewDriver(options.WithCustomLogger(log)))
+ apiTransports.RegisterUdpTransport(dm)
+ // No WithMaxIdleTime: the connection is never recycled, so every lease
+ // after the first resets the LIVE transport instead of reconnecting.
+ connCache := cache.NewPlcConnectionCache(dm,
cache.WithCustomLogger(log))
+ t.Cleanup(func() { assert.NoError(t, connCache.Close()) })
+
+ connStr :=
fmt.Sprintf("bacnet-ip:udp://127.0.0.1:%d?local-port=0&ApduTimeoutMs=3000",
device.port())
+
+ // Lease #1: priming read (creates the connection; scan analogue).
+ {
+ ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+ conn, err := connCache.GetConnection(ctx, connStr)
+ cancel()
+ require.NoError(t, err)
+ code, val, ok := readPresentValue(t, conn)
+ require.True(t, ok, "priming read via cache hung")
+ assert.Equal(t, apiModel.PlcResponseCode_OK, code)
+ assert.InDelta(t, 23.5, val, 0.001)
+ require.NoError(t, conn.Close())
+ }
+
+ // Short idle, then many write leases. Each lease runs Reset() against
the
+ // live socket while the receive worker polls — every iteration is one
+ // spin of the race wheel.
+ time.Sleep(500 * time.Millisecond)
+ for i := 0; i < 50; i++ {
+ ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+ conn, err := connCache.GetConnection(ctx, connStr)
+ cancel()
+ require.NoError(t, err, "lease %d failed", i)
+ code, ok := writePresentValue(t, conn)
+ require.NoError(t, conn.Close())
+ require.True(t, ok, "iteration %d: write did not complete
(Execute never resolved) — SimpleAck lost around lease-time Reset", i)
+ assert.Equal(t, apiModel.PlcResponseCode_OK, code, "iteration
%d", i)
+ }
+}
+
+// TestNativeBacnetWrite_LeaseResetWithIdleGap is the exact field ordering with
+// a real idle gap per iteration: read, return, idle >1s, write lease (Reset on
+// live socket), delayed device reply. Slower per iteration but closest to the
+// failing CI run's shape.
+func TestNativeBacnetWrite_LeaseResetWithIdleGap(t *testing.T) {
+ if testing.Short() {
+ t.Skip("multi-second idle gaps")
+ }
+ log := testutils.ProduceTestingLogger(t)
+ device := startDelayedWriteDevice(t, log, 300*time.Millisecond)
+ t.Cleanup(device.stop)
+
+ dm := plc4go.NewPlcDriverManager(options.WithCustomLogger(log))
+ dm.RegisterDriver(NewDriver(options.WithCustomLogger(log)))
+ apiTransports.RegisterUdpTransport(dm)
+ connCache := cache.NewPlcConnectionCache(dm,
cache.WithCustomLogger(log))
+ t.Cleanup(func() { assert.NoError(t, connCache.Close()) })
+
+ connStr :=
fmt.Sprintf("bacnet-ip:udp://127.0.0.1:%d?local-port=0&ApduTimeoutMs=5000",
device.port())
+
+ {
+ ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+ conn, err := connCache.GetConnection(ctx, connStr)
+ cancel()
+ require.NoError(t, err)
+ code, val, ok := readPresentValue(t, conn)
+ require.True(t, ok, "priming read via cache hung")
+ assert.Equal(t, apiModel.PlcResponseCode_OK, code)
+ assert.InDelta(t, 23.5, val, 0.001)
+ require.NoError(t, conn.Close())
+ }
+
+ for i := 0; i < 3; i++ {
+ time.Sleep(1500 * time.Millisecond) // idle gap on the returned
lease
+ ctx, cancel := context.WithTimeout(t.Context(), 15*time.Second)
+ conn, err := connCache.GetConnection(ctx, connStr)
+ cancel()
+ require.NoError(t, err, "lease %d failed", i)
+ code, ok := writePresentValue(t, conn)
+ require.NoError(t, conn.Close())
+ require.True(t, ok, "iteration %d: write after idle gap did not
complete", i)
+ assert.Equal(t, apiModel.PlcResponseCode_OK, code, "iteration
%d", i)
+ }
+}
diff --git a/plc4go/spi/transports/udp/TransportInstance.go
b/plc4go/spi/transports/udp/TransportInstance.go
index f6a0f594da..4634d6daa7 100644
--- a/plc4go/spi/transports/udp/TransportInstance.go
+++ b/plc4go/spi/transports/udp/TransportInstance.go
@@ -123,15 +123,24 @@ func (m *TransportInstance) Connect(ctx context.Context)
error {
return nil
}
+// Reset is deliberately a no-op for UDP. It used to poke the read deadline,
+// drain one datagram into a discard buffer, and swap m.reader — but it is
+// called from OUTSIDE the receive worker (connection-cache lease grants,
+// DefaultCodec retryable-error handling) while the worker concurrently drives
+// GetNumBytesAvailableInBuffer/Peek on the very same socket and reader. Every
+// one of those mutations races the worker and can silently destroy a FRESH
+// inbound datagram: the drain read eats it off the socket, and the reader
+// swap throws away bytes the worker had just buffered (observed in the field
+// as a BACnet WriteProperty SimpleAck that reached the host interface with a
+// good checksum but never surfaced; reproduced by
+// bacnetip.TestNativeBacnetWrite_LeaseResetOnLiveSocket).
+//
+// Nothing here needs cleaning anyway: UDP is datagram-framed, so "stale"
+// buffered data is just a complete late reply, which the codec's expectation
+// matching already drops without harm. Reconnection semantics live in
+// Close/Connect.
func (m *TransportInstance) Reset() {
- if m.udpConn == nil {
- m.log.Trace().Msg("No connection to reset")
- return
- }
- _ = m.udpConn.SetReadDeadline(time.Now().Add(1))
- _, _, _ = m.udpConn.ReadFromUDP(make([]byte, 4096))
- m.reader = bufio.NewReader(m.udpConn)
- m.log.Trace().Msg("Connection reset")
+ m.log.Trace().Msg("Reset is a no-op for UDP (see doc comment)")
}
func (m *TransportInstance) Close() error {
