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 40c4267248 fix(plc4go): make TCP TransportInstance.Reset a no-op
(raced the receive worker, could desync stream framing)
40c4267248 is described below
commit 40c42672489d8e89eafd089d2e9f3d14ad5f6b84
Author: Sebastian Rühl <[email protected]>
AuthorDate: Thu Jul 16 08:49:28 2026 +0200
fix(plc4go): make TCP TransportInstance.Reset a no-op (raced the receive
worker, could desync stream framing)
Same defect class as the UDP transport fix: Reset() poked the read
deadline, drained one socket read into a discard buffer, and swapped
m.reader - unsynchronized, and invoked from outside the receive worker
(connection-cache lease grants, DefaultCodec retryable-error handling)
while the worker concurrently drives Peek/Read on the same conn and
reader. On a stream the damage is worse than on datagrams: the drain can
consume a fresh reply, and discarding bytes the reader had already
buffered desyncs the protocol framing for everything that follows. The
swap also silently shrank the reader from the 100k buffer Connect
creates to bufio's 4k default.
Serial's Reset was already a no-op; TCP now matches UDP and serial.
Stale complete frames from abandoned exchanges are handled by the
codec's response matching, and reconnection semantics live in
Close/Connect.
New tests: a deterministic buffered-frame-survives-Reset check and a
worker-vs-Reset storm that fails under -race against the old code.
Full modbus/eip/s7/cbus/ads/codec/cache suites green under -race.
---
plc4go/spi/transports/tcp/TransportInstance.go | 24 ++--
.../transports/tcp/TransportInstanceReset_test.go | 151 +++++++++++++++++++++
2 files changed, 167 insertions(+), 8 deletions(-)
diff --git a/plc4go/spi/transports/tcp/TransportInstance.go
b/plc4go/spi/transports/tcp/TransportInstance.go
index ebd014f213..e971af2554 100644
--- a/plc4go/spi/transports/tcp/TransportInstance.go
+++ b/plc4go/spi/transports/tcp/TransportInstance.go
@@ -96,15 +96,23 @@ func (m *TransportInstance) Connect(ctx context.Context)
error {
return nil
}
+// Reset is deliberately a no-op for TCP, matching the UDP and serial
+// transports. It used to poke the read deadline, drain one socket read 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/Read on the same conn and reader. Every
+// one of those mutations races the worker, and on a STREAM they are doubly
+// destructive: the drain can consume a fresh reply, and dropping bytes the
+// reader had already buffered desyncs the protocol framing for everything
+// that follows (the swap also silently shrank the reader from the 100k
+// buffer Connect creates to bufio's 4k default). Stale complete frames from
+// an abandoned exchange are already handled by the codec's response
+// matching; reconnection semantics live in Close/Connect.
+// See TestTransportInstance_ResetKeepsBufferedBytes /
+// TestTransportInstance_ResetDoesNotRaceReceiveWorker.
func (m *TransportInstance) Reset() {
- if m.tcpConn == nil {
- m.log.Trace().Msg("No connection to reset")
- return
- }
- _ = m.tcpConn.SetReadDeadline(time.Now().Add(1))
- _, _ = m.tcpConn.Read(make([]byte, 4096))
- m.reader = bufio.NewReader(m.tcpConn)
- m.log.Trace().Msg("Connection reset")
+ m.log.Trace().Msg("Reset is a no-op for TCP (see doc comment)")
}
func (m *TransportInstance) Close() error {
diff --git a/plc4go/spi/transports/tcp/TransportInstanceReset_test.go
b/plc4go/spi/transports/tcp/TransportInstanceReset_test.go
new file mode 100644
index 0000000000..d4487f6670
--- /dev/null
+++ b/plc4go/spi/transports/tcp/TransportInstanceReset_test.go
@@ -0,0 +1,151 @@
+/*
+ * 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 tcp
+
+import (
+ "context"
+ "net"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/plc4x/plc4go/spi/testutils"
+)
+
+// connectedTestInstance dials a local listener and returns the transport
+// instance plus the server-side conn for injecting bytes.
+func connectedTestInstance(t *testing.T) (*TransportInstance, net.Conn) {
+ t.Helper()
+ listener, err := net.Listen("tcp", "127.0.0.1:0")
+ require.NoError(t, err)
+ t.Cleanup(func() { _ = listener.Close() })
+
+ serverConnCh := make(chan net.Conn, 1)
+ go func() {
+ conn, err := listener.Accept()
+ if err != nil {
+ return
+ }
+ serverConnCh <- conn
+ }()
+
+ addr := listener.Addr().(*net.TCPAddr)
+ ti := NewTcpTransportInstance(addr, 1, nil,
testutils.EnrichOptionsWithOptionsForTesting(t)...)
+ require.NoError(t, ti.Connect(context.Background()))
+ t.Cleanup(func() { _ = ti.Close() })
+
+ serverConn := <-serverConnCh
+ t.Cleanup(func() { _ = serverConn.Close() })
+ return ti, serverConn
+}
+
+// waitBuffered polls until the instance's reader has buffered n bytes.
+func waitBuffered(t *testing.T, ti *TransportInstance, n uint32) {
+ t.Helper()
+ deadline := time.Now().Add(5 * time.Second)
+ for time.Now().Before(deadline) {
+ avail, err := ti.GetNumBytesAvailableInBuffer()
+ require.NoError(t, err)
+ if avail >= n {
+ return
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+ t.Fatalf("timed out waiting for %d buffered bytes", n)
+}
+
+// A frame the peer already delivered - and that the receive path already
+// buffered - must survive a lease-time Reset. The old Reset swapped m.reader,
+// silently discarding buffered bytes: on a stream that both loses the reply
+// and desyncs the framing for everything after it (the TCP analogue of the
+// BACnet SimpleAck loss fixed in the UDP transport).
+func TestTransportInstance_ResetKeepsBufferedBytes(t *testing.T) {
+ ti, serverConn := connectedTestInstance(t)
+
+ frame := []byte{0x00, 0x01, 0x00, 0x00, 0x00, 0x04, 0x01, 0x03, 0x02,
0x2a}
+ _, err := serverConn.Write(frame)
+ require.NoError(t, err)
+ waitBuffered(t, ti, uint32(len(frame)))
+
+ ti.Reset() // cache-lease analogue
+
+ ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+ defer cancel()
+ got, err := ti.Read(ctx, uint32(len(frame)))
+ require.NoError(t, err, "buffered frame must survive Reset")
+ assert.Equal(t, frame, got, "frame bytes must be intact after Reset")
+}
+
+// Reset must not race the receive worker. The worker side is modelled the way
+// DefaultCodec drives it: a tight GetNumBytesAvailableInBuffer/Read poll loop,
+// while another goroutine (the connection cache granting a lease) calls
+// Reset(). Run with -race.
+func TestTransportInstance_ResetDoesNotRaceReceiveWorker(t *testing.T) {
+ ti, serverConn := connectedTestInstance(t)
+
+ stop := make(chan struct{})
+ var wg sync.WaitGroup
+
+ // Peer keeps trickling bytes so the worker's reads have real traffic.
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ payload := []byte{0xAA, 0xBB, 0xCC, 0xDD}
+ for {
+ select {
+ case <-stop:
+ return
+ default:
+ _, _ = serverConn.Write(payload)
+ time.Sleep(time.Millisecond)
+ }
+ }
+ }()
+
+ // Receive worker analogue.
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ for {
+ select {
+ case <-stop:
+ return
+ default:
+ if avail, err :=
ti.GetNumBytesAvailableInBuffer(); err == nil && avail > 0 {
+ ctx, cancel :=
context.WithTimeout(context.Background(), 50*time.Millisecond)
+ _, _ = ti.Read(ctx, avail)
+ cancel()
+ }
+ }
+ }
+ }()
+
+ // Cache-lease analogue: Reset storm from a foreign goroutine.
+ for i := 0; i < 200; i++ {
+ ti.Reset()
+ time.Sleep(500 * time.Microsecond)
+ }
+
+ close(stop)
+ wg.Wait()
+}
