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 529811d767 fix(plc4go): TCP Write() armed a READ deadline (copy-paste)
- write unbounded, sticky deadline broke late replies
529811d767 is described below
commit 529811d7676a265b1336365a3d1f731f656f2918
Author: Sebastian Rühl <[email protected]>
AuthorDate: Mon Jul 20 11:28:11 2026 +0200
fix(plc4go): TCP Write() armed a READ deadline (copy-paste) - write
unbounded, sticky deadline broke late replies
TransportInstance.Write set tcpConn.SetReadDeadline from the caller's
context deadline (copied from the read path), so the write itself was
never bounded - a peer that stops draining the connection blocked
Write() past any context deadline - and the read deadline stuck to the
connection forever: once expired, every socket read failed instantly
with i/o timeout, so a reply arriving after the request deadline sat
unread in the kernel buffer until the NEXT write re-armed a fresh
deadline. The SetReadDeadline METHOD additionally called
tcpConn.SetDeadline, arming BOTH directions - an (expired) read
deadline made writes fail too.
The accidental sticky read deadline was also the only thing unblocking
the receive worker's bare Peek(1) polls (DefaultCodec drives Receive
with a deadline-less long-lived context), so it cannot simply be
dropped: without a bound, one Receive goroutine per receiveTimeout
cycle piles up on a silent connection, and they race on the shared
bufio.Reader once data finally arrives.
Fix, mirroring the serial transport:
- Write arms SetWriteDeadline from the context deadline (and clears a
leftover one when the context carries no deadline - an expired armed
deadline would fail later unbounded writes instantly).
- A tcp deadlineReader (port of serial's) feeds the bufio.Reader:
an explicit per-operation deadline wins and auto-clears once observed
expired; otherwise a 1s fallback keeps deadline-less polls bounded
(retryable timeout, not fatal, for the codecs' error classification).
- The SetReadDeadline method routes to the deadlineReader instead of
arming conn-wide SetDeadline.
New falsification tests (all fail against the old code): late reply
after write-deadline expiry must be readable, write must honor its
context deadline against a non-draining peer, a read deadline must not
bound writes, an expired write deadline must not stick, and bare polls
must be fallback-bounded. deadlineReader unit tests ported from serial.
---
plc4go/spi/transports/tcp/TransportInstance.go | 34 ++++-
.../tcp/TransportInstanceDeadline_test.go | 169 +++++++++++++++++++++
plc4go/spi/transports/tcp/deadline_reader.go | 86 +++++++++++
plc4go/spi/transports/tcp/deadline_reader_test.go | 113 ++++++++++++++
4 files changed, 398 insertions(+), 4 deletions(-)
diff --git a/plc4go/spi/transports/tcp/TransportInstance.go
b/plc4go/spi/transports/tcp/TransportInstance.go
index e971af2554..c8e62d89f9 100644
--- a/plc4go/spi/transports/tcp/TransportInstance.go
+++ b/plc4go/spi/transports/tcp/TransportInstance.go
@@ -39,6 +39,13 @@ import (
"github.com/apache/plc4x/plc4go/spi/utils"
)
+// defaultReadPollTimeout bounds reads that carry no explicit deadline (the
+// codec's receive worker polls with a deadline-less long-lived context): a
+// silent connection must yield a retryable timeout instead of blocking a
+// worker goroutine forever. Mirrors the serial transport's readTimeout
+// default.
+const defaultReadPollTimeout = time.Second
+
type TransportInstance struct {
transportUtils.DefaultBufferedTransportInstance
@@ -49,6 +56,7 @@ type TransportInstance struct {
transport *Transport
tcpConn net.Conn
+ armer *deadlineReader
reader *bufio.Reader
connected atomic.Bool
@@ -90,7 +98,8 @@ func (m *TransportInstance) Connect(ctx context.Context)
error {
m.LocalAddress = m.tcpConn.LocalAddr().(*net.TCPAddr)
- m.reader = bufio.NewReaderSize(m.tcpConn, 100000)
+ m.armer = newDeadlineReader(m.tcpConn, defaultReadPollTimeout)
+ m.reader = bufio.NewReaderSize(m.armer, 100000)
m.connected.Store(true)
return nil
@@ -137,11 +146,19 @@ func (m *TransportInstance) Write(ctx context.Context,
data []byte) error {
if !m.connected.Load() {
return errors.New("error writing to transport. Not connected")
}
+ // Bound the write itself by the caller's context deadline. This used
to arm
+ // a READ deadline (copy-paste from the read path): the write stayed
+ // unbounded, and the sticky read deadline broke reads of replies
arriving
+ // after it had expired.
if deadline, ok := ctx.Deadline(); ok {
m.log.Trace().Time("deadline", deadline).Msg("deadline set")
- if err := m.tcpConn.SetReadDeadline(deadline); err != nil {
- return errors.Wrap(err, "error setting read deadline")
+ if err := m.tcpConn.SetWriteDeadline(deadline); err != nil {
+ return errors.Wrap(err, "error setting write deadline")
}
+ } else if err := m.tcpConn.SetWriteDeadline(time.Time{}); err != nil {
+ // Clear whatever deadline a previous bounded Write left armed
- an
+ // expired one would fail this (unbounded) write instantly.
+ return errors.Wrap(err, "error clearing write deadline")
}
num, err := m.tcpConn.Write(data)
if err != nil {
@@ -158,7 +175,16 @@ func (m *TransportInstance) GetReader()
transports.ExtendedReader {
}
func (m *TransportInstance) SetReadDeadline(deadline time.Time) error {
- return m.tcpConn.SetDeadline(deadline)
+ // Routed through the deadlineReader (NOT tcpConn.SetDeadline, which
arms
+ // both directions and made read deadlines fail writes): the explicit
+ // deadline wins over the poll fallback for the next read(s) and, once
+ // observed expired, auto-clears instead of sticking to the connection.
+ armer := m.armer
+ if armer == nil {
+ return errors.New("error setting read deadline. Not connected")
+ }
+ armer.setExplicitDeadline(deadline)
+ return nil
}
func (m *TransportInstance) String() string {
diff --git a/plc4go/spi/transports/tcp/TransportInstanceDeadline_test.go
b/plc4go/spi/transports/tcp/TransportInstanceDeadline_test.go
new file mode 100644
index 0000000000..f8277b0188
--- /dev/null
+++ b/plc4go/spi/transports/tcp/TransportInstanceDeadline_test.go
@@ -0,0 +1,169 @@
+/*
+ * 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"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// Write used to arm a READ deadline from the request context (copy-paste from
+// the read path). That deadline was sticky: once it expired, every socket read
+// failed instantly with i/o timeout, so a reply arriving after the request
+// deadline sat unread in the kernel buffer until the NEXT Write re-armed a
+// fresh deadline. A late reply must instead reach the reader - the codec's
+// response matching decides what to do with it, not the transport.
+func TestTransportInstance_WriteMustNotArmReadDeadline(t *testing.T) {
+ ti, serverConn := connectedTestInstance(t)
+
+ writeCtx, cancel := context.WithTimeout(context.Background(),
50*time.Millisecond)
+ defer cancel()
+ require.NoError(t, ti.Write(writeCtx, []byte{0x00, 0x01}))
+
+ // Let the request deadline expire, then deliver the (late) reply.
+ time.Sleep(150 * time.Millisecond)
+ frame := []byte{0x00, 0x01, 0x00, 0x00, 0x00, 0x04, 0x01, 0x03, 0x02,
0x2a}
+ _, err := serverConn.Write(frame)
+ require.NoError(t, err)
+
+ // The reply must still be picked up by the buffered read path.
+ waitBuffered(t, ti, uint32(len(frame)))
+
+ readCtx, cancel := context.WithTimeout(context.Background(),
2*time.Second)
+ defer cancel()
+ got, err := ti.Read(readCtx, uint32(len(frame)))
+ require.NoError(t, err, "a reply arriving after the write context
deadline must still be readable")
+ assert.Equal(t, frame, got)
+}
+
+// Write never armed a WRITE deadline at all, so a wedged peer (nothing
+// draining the connection, kernel buffers full) blocked Write() indefinitely
+// - past any caller context deadline. This is the unbounded half of the
+// copy-paste bug.
+func TestTransportInstance_WriteHonorsContextDeadline(t *testing.T) {
+ ti, serverConn := connectedTestInstance(t)
+
+ // Shrink both kernel buffers as far as the OS allows and never read on
the
+ // server side, so a large enough write must block in the kernel.
+ require.NoError(t, ti.tcpConn.(*net.TCPConn).SetWriteBuffer(1))
+ require.NoError(t, serverConn.(*net.TCPConn).SetReadBuffer(1))
+
+ payload := make([]byte, 64*1024*1024)
+ writeCtx, cancel := context.WithTimeout(context.Background(),
200*time.Millisecond)
+ defer cancel()
+
+ errCh := make(chan error, 1)
+ start := time.Now()
+ go func() {
+ errCh <- ti.Write(writeCtx, payload)
+ }()
+
+ select {
+ case err := <-errCh:
+ require.Error(t, err, "a write the peer never drains must fail
once the context deadline passes")
+ var netErr net.Error
+ require.ErrorAs(t, err, &netErr)
+ assert.True(t, netErr.Timeout(), "expected a timeout error,
got: %v", err)
+ assert.Less(t, time.Since(start), 3*time.Second, "write must
return promptly after the deadline")
+ case <-time.After(5 * time.Second):
+ t.Fatal("Write blocked past the context deadline (no write
deadline armed)")
+ }
+}
+
+// The SetReadDeadline METHOD used to call tcpConn.SetDeadline, arming BOTH
+// directions: a read deadline set by the buffered read path (or a driver)
+// silently bounded - or, once expired, instantly failed - subsequent writes.
+func TestTransportInstance_SetReadDeadlineMustNotBoundWrites(t *testing.T) {
+ ti, serverConn := connectedTestInstance(t)
+
+ // Keep the connection drained so the write below cannot block.
+ go func() {
+ buf := make([]byte, 1024)
+ for {
+ if _, err := serverConn.Read(buf); err != nil {
+ return
+ }
+ }
+ }()
+
+ require.NoError(t, ti.SetReadDeadline(time.Now().Add(-time.Second)))
+
+ err := ti.Write(context.Background(), []byte{0x00, 0x01})
+ require.NoError(t, err, "an (expired) READ deadline must not fail
writes")
+}
+
+// A write deadline armed by one bounded Write must not stick to the
+// connection: a later Write whose context carries NO deadline would otherwise
+// fail instantly once the old deadline expired.
+func TestTransportInstance_ExpiredWriteDeadlineDoesNotStick(t *testing.T) {
+ ti, serverConn := connectedTestInstance(t)
+
+ // Keep the connection drained so writes cannot block.
+ go func() {
+ buf := make([]byte, 1024)
+ for {
+ if _, err := serverConn.Read(buf); err != nil {
+ return
+ }
+ }
+ }()
+
+ // A Write whose context deadline already passed fails - that is the
+ // caller's bounded contract - and leaves an expired deadline armed.
+ expiredCtx, cancel := context.WithDeadline(context.Background(),
time.Now().Add(-time.Second))
+ defer cancel()
+ require.Error(t, ti.Write(expiredCtx, []byte{0x00, 0x01}))
+
+ // A deadline-less Write afterwards must succeed.
+ err := ti.Write(context.Background(), []byte{0x00, 0x01})
+ require.NoError(t, err, "an expired write deadline from a previous
bounded Write must not fail later unbounded writes")
+}
+
+// With the sticky read deadline gone, something else has to keep the receive
+// worker's bare polls bounded: DefaultCodec drives Receive with a
deadline-less
+// long-lived context, and bufio's Peek(1) on a silent connection would
+// otherwise block forever - piling up one abandoned Receive goroutine per
+// receiveTimeout cycle, all racing on the same bufio.Reader once data finally
+// arrives. The fallback read deadline (mirroring the serial transport's
+// deadlineReader) bounds every deadline-less read.
+func TestTransportInstance_BarePollIsBoundedByFallback(t *testing.T) {
+ ti, _ := connectedTestInstance(t)
+
+ done := make(chan error, 1)
+ go func() {
+ _, err := ti.GetReader().Peek(1)
+ done <- err
+ }()
+
+ select {
+ case err := <-done:
+ require.Error(t, err, "Peek on a silent connection must fail
with a timeout, not return data")
+ var netErr net.Error
+ require.ErrorAs(t, err, &netErr)
+ assert.True(t, netErr.Timeout(), "expected a timeout error,
got: %v", err)
+ case <-time.After(5 * time.Second):
+ t.Fatal("bare Peek(1) on a silent connection blocked forever
(no fallback read deadline)")
+ }
+}
diff --git a/plc4go/spi/transports/tcp/deadline_reader.go
b/plc4go/spi/transports/tcp/deadline_reader.go
new file mode 100644
index 0000000000..721ef579db
--- /dev/null
+++ b/plc4go/spi/transports/tcp/deadline_reader.go
@@ -0,0 +1,86 @@
+/*
+ * 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 (
+ "io"
+ "sync/atomic"
+ "time"
+)
+
+// deadlineConn is the subset of net.Conn the deadlineReader needs.
+type deadlineConn interface {
+ io.Reader
+ SetReadDeadline(t time.Time) error
+}
+
+// deadlineReader arms the connection's read deadline before every Read - the
+// TCP analogue of the serial transport's deadlineReader. An explicit deadline
+// set via setExplicitDeadline (the context deadline of the current transport
+// operation) always wins; an explicit deadline observed as already expired is
+// honored for exactly one read - so the bounded caller gets its timeout error
+// - and then auto-clears. Without an active explicit deadline the configured
+// fallback applies (0 = blocking), which keeps bare buffered-layer polls
+// (DefaultCodec drives Receive with a deadline-less long-lived context)
+// bounded instead of hanging forever on a silent connection.
+type deadlineReader struct {
+ conn deadlineConn
+ fallback time.Duration
+ explicit atomic.Value // time.Time; zero value = none
+}
+
+func newDeadlineReader(conn deadlineConn, fallback time.Duration)
*deadlineReader {
+ r := &deadlineReader{conn: conn, fallback: fallback}
+ r.explicit.Store(time.Time{})
+ return r
+}
+
+// setExplicitDeadline sets (or, with a zero time, clears) the explicit
+// read deadline. Safe for concurrent use with Read.
+func (r *deadlineReader) setExplicitDeadline(t time.Time) {
+ r.explicit.Store(t)
+}
+
+func (r *deadlineReader) Read(p []byte) (int, error) {
+ deadline, _ := r.explicit.Load().(time.Time)
+ switch {
+ case !deadline.IsZero():
+ if err := r.conn.SetReadDeadline(deadline); err != nil {
+ return 0, err
+ }
+ if !time.Now().Before(deadline) {
+ // Expired: honored for this read, cleared for the
next. CAS so
+ // a concurrent setExplicitDeadline is never
overwritten. The old
+ // value MUST be the exact Load()ed time.Time (never
reconstructed
+ // or derived): CompareAndSwap uses interface ==, which
includes
+ // time.Time's monotonic-clock reading.
+ r.explicit.CompareAndSwap(deadline, time.Time{})
+ }
+ case r.fallback > 0:
+ if err := r.conn.SetReadDeadline(time.Now().Add(r.fallback));
err != nil {
+ return 0, err
+ }
+ default:
+ if err := r.conn.SetReadDeadline(time.Time{}); err != nil {
+ return 0, err
+ }
+ }
+ return r.conn.Read(p)
+}
diff --git a/plc4go/spi/transports/tcp/deadline_reader_test.go
b/plc4go/spi/transports/tcp/deadline_reader_test.go
new file mode 100644
index 0000000000..a8ff1e5fda
--- /dev/null
+++ b/plc4go/spi/transports/tcp/deadline_reader_test.go
@@ -0,0 +1,113 @@
+/*
+ * 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 (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// recordingConn implements deadlineConn and records SetReadDeadline calls.
+type recordingConn struct {
+ readDeadlines []time.Time
+}
+
+func (r *recordingConn) Read(p []byte) (int, error) {
+ return len(p), nil
+}
+
+func (r *recordingConn) SetReadDeadline(t time.Time) error {
+ r.readDeadlines = append(r.readDeadlines, t)
+ return nil
+}
+
+func TestDeadlineReader_FallbackArmsRelativeDeadline(t *testing.T) {
+ conn := &recordingConn{}
+ reader := newDeadlineReader(conn, 200*time.Millisecond)
+
+ before := time.Now()
+ _, err := reader.Read(make([]byte, 1))
+ require.NoError(t, err)
+
+ require.Len(t, conn.readDeadlines, 1)
+ armed := conn.readDeadlines[0]
+ assert.False(t, armed.Before(before.Add(200*time.Millisecond)),
"deadline must be >= now+fallback")
+ assert.True(t, armed.Before(before.Add(2*time.Second)), "deadline must
be near now+fallback")
+}
+
+func TestDeadlineReader_ZeroFallbackClearsDeadline(t *testing.T) {
+ conn := &recordingConn{}
+ reader := newDeadlineReader(conn, 0)
+
+ _, err := reader.Read(make([]byte, 1))
+ require.NoError(t, err)
+
+ require.Len(t, conn.readDeadlines, 1)
+ assert.True(t, conn.readDeadlines[0].IsZero(), "no fallback means the
deadline is cleared")
+}
+
+func TestDeadlineReader_ExplicitDeadlineWins(t *testing.T) {
+ conn := &recordingConn{}
+ reader := newDeadlineReader(conn, time.Hour)
+
+ explicit := time.Now().Add(30 * time.Millisecond)
+ reader.setExplicitDeadline(explicit)
+ _, err := reader.Read(make([]byte, 1))
+ require.NoError(t, err)
+
+ require.Len(t, conn.readDeadlines, 1)
+ assert.Equal(t, explicit, conn.readDeadlines[0])
+}
+
+func TestDeadlineReader_ExpiredExplicitHonoredOnceThenFallback(t *testing.T) {
+ conn := &recordingConn{}
+ reader := newDeadlineReader(conn, time.Hour)
+
+ expired := time.Now().Add(-time.Second)
+ reader.setExplicitDeadline(expired)
+
+ // First read still sees the expired explicit deadline (the ctx-bounded
+ // caller must get its timeout), ...
+ _, _ = reader.Read(make([]byte, 1))
+ require.Len(t, conn.readDeadlines, 1)
+ assert.Equal(t, expired, conn.readDeadlines[0])
+
+ // ... the next read falls back.
+ _, _ = reader.Read(make([]byte, 1))
+ require.Len(t, conn.readDeadlines, 2)
+ assert.False(t, conn.readDeadlines[1].Equal(expired), "expired explicit
deadline must auto-clear")
+ assert.False(t, conn.readDeadlines[1].IsZero(), "fallback must be
armed")
+}
+
+func TestDeadlineReader_ClearingExplicitRestoresFallback(t *testing.T) {
+ conn := &recordingConn{}
+ reader := newDeadlineReader(conn, time.Minute)
+
+ reader.setExplicitDeadline(time.Now().Add(time.Hour))
+ _, _ = reader.Read(make([]byte, 1))
+ reader.setExplicitDeadline(time.Time{})
+ _, _ = reader.Read(make([]byte, 1))
+
+ require.Len(t, conn.readDeadlines, 2)
+ assert.False(t, conn.readDeadlines[1].Equal(conn.readDeadlines[0]),
"cleared explicit must not persist")
+}
