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 24daaca625 test(plc4go): pin BACnet write SimpleAck handling across
codec, driver, and UDP transport
24daaca625 is described below
commit 24daaca625e85d7df4cf6d28d27918f42c2b5397
Author: Sebastian Rühl <[email protected]>
AuthorDate: Tue Jul 14 14:57:16 2026 +0200
test(plc4go): pin BACnet write SimpleAck handling across codec, driver, and
UDP transport
Regression tests from a ugagent write-hang investigation (three rounds) that
exonerated plc4go at every layer:
- WriteRoundtrip_test.go: codec-level — a wire-shaped SimpleAck (matching
invoke id + service choice) resolves WriteProperty Execute with OK; wrong
invoke id does not resolve; Error-PDU maps to ACCESS_DENIED.
- IdleWriteRoundtrip_test.go: driver-level — writes resolve after idle gaps
and with delayed replies (cache idle-reconnect included).
- DelayedReply_test.go: UDP transport — the per-poll receive deadline is
re-armed across arbitrary idle periods; delayed datagrams still reach the
codec.
Left for a future change (identified, not fixed here): TransportInstance
Reset() runs unlocked on every lease and can data-race the receive worker.
---
.../internal/bacnetip/IdleWriteRoundtrip_test.go | 204 +++++++++++++++
plc4go/internal/bacnetip/WriteRoundtrip_test.go | 281 +++++++++++++++++++++
plc4go/spi/transports/udp/DelayedReply_test.go | 112 ++++++++
3 files changed, 597 insertions(+)
diff --git a/plc4go/internal/bacnetip/IdleWriteRoundtrip_test.go
b/plc4go/internal/bacnetip/IdleWriteRoundtrip_test.go
new file mode 100644
index 0000000000..07f8d8007a
--- /dev/null
+++ b/plc4go/internal/bacnetip/IdleWriteRoundtrip_test.go
@@ -0,0 +1,204 @@
+/*
+ * 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"
+ "net"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/rs/zerolog"
+ "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"
+ model
"github.com/apache/plc4x/plc4go/protocols/bacnetip/readwrite/model"
+ "github.com/apache/plc4x/plc4go/spi/options"
+ "github.com/apache/plc4x/plc4go/spi/testutils"
+)
+
+// delayedWriteDevice answers WriteProperty with a SimpleAck after a fixed
delay,
+// mimicking the ~300ms round-trip observed in the field (instant replies in
the
+// other fakes cannot exercise the delayed-arrival path).
+type delayedWriteDevice struct {
+ conn *net.UDPConn
+ wg sync.WaitGroup
+ log zerolog.Logger
+ delay time.Duration
+ ctx context.Context
+}
+
+func startDelayedWriteDevice(t *testing.T, log zerolog.Logger, delay
time.Duration) *delayedWriteDevice {
+ t.Helper()
+ conn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0,
1), Port: 0})
+ require.NoError(t, err)
+ d := &delayedWriteDevice{conn: conn, log: log, delay: delay, ctx:
t.Context()}
+ d.wg.Add(1)
+ go d.serve()
+ return d
+}
+
+func (d *delayedWriteDevice) port() int { return
d.conn.LocalAddr().(*net.UDPAddr).Port }
+func (d *delayedWriteDevice) stop() { _ = d.conn.Close(); d.wg.Wait() }
+
+func (d *delayedWriteDevice) serve() {
+ defer d.wg.Done()
+ buf := make([]byte, 4096)
+ for {
+ n, src, err := d.conn.ReadFromUDP(buf)
+ if err != nil {
+ return
+ }
+ data := make([]byte, n)
+ copy(data, buf[:n])
+ bvlc, err := model.BVLCParse[model.BVLC](d.ctx, data)
+ if err != nil {
+ continue
+ }
+ npduRetriever, ok := bvlc.(interface{ GetNpdu() model.NPDU })
+ if !ok {
+ continue
+ }
+ cr, ok :=
npduRetriever.GetNpdu().GetApdu().(model.APDUConfirmedRequest)
+ if !ok {
+ continue
+ }
+ invokeId := cr.GetInvokeId()
+ choice := cr.GetServiceRequest().GetServiceChoice()
+ go func() {
+ time.Sleep(d.delay)
+ var resp model.BVLC
+ if choice ==
model.BACnetConfirmedServiceChoice_READ_PROPERTY {
+ resp = buildReadAckFor(invokeId)
+ } else {
+ resp =
wrapAPDU(model.NewAPDUSimpleAck(invokeId, choice), false)
+ }
+ theBytes, err := resp.Serialize()
+ if err != nil {
+ return
+ }
+ _, _ = d.conn.WriteToUDP(theBytes, src)
+ }()
+ }
+}
+
+// TestNativeBacnetWrite_DelayedReplyAfterIdle combines both field conditions:
a
+// multi-second idle gap AND a ~400ms device reply delay on the write.
+func TestNativeBacnetWrite_DelayedReplyAfterIdle(t *testing.T) {
+ log := testutils.ProduceTestingLogger(t)
+ device := startDelayedWriteDevice(t, log, 400*time.Millisecond)
+ t.Cleanup(device.stop)
+
+ conn := newWriteTestConnection(t, log, device.port())
+
+ code, val, ok := readPresentValue(t, conn)
+ require.True(t, ok, "priming read hung")
+ assert.Equal(t, apiModel.PlcResponseCode_OK, code)
+ assert.InDelta(t, 23.5, val, 0.001)
+
+ time.Sleep(3 * time.Second)
+
+ code, ok = writePresentValue(t, conn)
+ require.True(t, ok, "delayed write after idle gap did not resolve")
+ assert.Equal(t, apiModel.PlcResponseCode_OK, code)
+}
+
+// TestNativeBacnetWrite_AfterIdleGap reproduces the real field ordering: a
read
+// succeeds, then the connection sits idle for several seconds (the receive
loop
+// keeps polling with expiring read deadlines), then a WriteProperty is issued.
+// The SimpleAck must still resolve the Execute channel.
+func TestNativeBacnetWrite_AfterIdleGap(t *testing.T) {
+ log := testutils.ProduceTestingLogger(t)
+ device := startFakeWriteDevice(t, log)
+ t.Cleanup(device.stop)
+
+ conn := newWriteTestConnection(t, log, device.port())
+
+ // One read to prime the connection (mirrors discovery/poll).
+ code, val, ok := readPresentValue(t, conn)
+ require.True(t, ok, "priming read hung")
+ assert.Equal(t, apiModel.PlcResponseCode_OK, code)
+ assert.InDelta(t, 23.5, val, 0.001)
+
+ // Idle gap well beyond the 10ms per-poll read deadline.
+ time.Sleep(3 * time.Second)
+
+ code, ok = writePresentValue(t, conn)
+ t.Logf("device received %d request(s)", device.requestCount())
+ require.True(t, ok, "write after idle gap did not complete (Execute
never resolved)")
+ assert.Equal(t, apiModel.PlcResponseCode_OK, code)
+}
+
+// TestNativeBacnetWrite_AfterIdleViaCache mirrors the real ugagent path: the
+// connection is leased from a PlcConnectionCache with a short max-idle-time, a
+// read runs, the lease is returned and sits idle past maxIdleTime (so the
cache
+// proactively reconnects on the next lease, opening a NEW ephemeral socket),
+// then a write is issued. The SimpleAck must still resolve.
+func TestNativeBacnetWrite_AfterIdleViaCache(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)
+ connCache := cache.NewPlcConnectionCache(dm,
+ cache.WithCustomLogger(log),
+ cache.WithMaxIdleTime(500*time.Millisecond),
+ )
+ 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: a read.
+ {
+ ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+ defer cancel()
+ conn, err := connCache.GetConnection(ctx, connStr)
+ 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()) // return lease to cache ->
becomes idle
+ }
+
+ // Idle past maxIdleTime so the next lease triggers a reconnect (new
socket).
+ time.Sleep(2 * time.Second)
+
+ // Lease #2: a write on the (reconnected) connection.
+ {
+ ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+ defer cancel()
+ conn, err := connCache.GetConnection(ctx, connStr)
+ require.NoError(t, err)
+ code, ok := writePresentValue(t, conn)
+ t.Logf("device received %d request(s)", device.requestCount())
+ require.NoError(t, conn.Close())
+ require.True(t, ok, "write after idle-reconnect via cache did
not complete (Execute never resolved)")
+ assert.Equal(t, apiModel.PlcResponseCode_OK, code)
+ }
+}
diff --git a/plc4go/internal/bacnetip/WriteRoundtrip_test.go
b/plc4go/internal/bacnetip/WriteRoundtrip_test.go
new file mode 100644
index 0000000000..07c4828a3d
--- /dev/null
+++ b/plc4go/internal/bacnetip/WriteRoundtrip_test.go
@@ -0,0 +1,281 @@
+/*
+ * 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"
+ "net"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/rs/zerolog"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ plc4go "github.com/apache/plc4x/plc4go/pkg/api"
+ apiModel "github.com/apache/plc4x/plc4go/pkg/api/model"
+ apiTransports "github.com/apache/plc4x/plc4go/pkg/api/transports"
+ model
"github.com/apache/plc4x/plc4go/protocols/bacnetip/readwrite/model"
+ "github.com/apache/plc4x/plc4go/spi/options"
+ "github.com/apache/plc4x/plc4go/spi/testutils"
+)
+
+// fakeWriteDevice is a minimal UDP BACnet/IP device that answers every
+// confirmed WriteProperty / WritePropertyMultiple request with the
spec-correct
+// success response: a SimpleAck (clause 15.9) echoing the request's invoke id
+// and service choice.
+type fakeWriteDevice struct {
+ conn *net.UDPConn
+ wg sync.WaitGroup
+ log zerolog.Logger
+ gotReqs int
+ ctx context.Context
+ mu sync.Mutex
+
+ // wrongInvokeId, when true, makes the device reply with an off-by-one
+ // invoke id so we can assert the client does NOT accept the mismatch.
+ wrongInvokeId bool
+ // replyError, when true, makes the device reply with a BACnet Error-PDU
+ // (WRITE_ACCESS_DENIED) instead of a SimpleAck.
+ replyError bool
+}
+
+func startFakeWriteDevice(t *testing.T, log zerolog.Logger) *fakeWriteDevice {
+ t.Helper()
+ addr := &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0}
+ conn, err := net.ListenUDP("udp4", addr)
+ require.NoError(t, err)
+ d := &fakeWriteDevice{conn: conn, log: log, ctx: t.Context()}
+ d.wg.Add(1)
+ go d.serve()
+ return d
+}
+
+func (d *fakeWriteDevice) port() int { return
d.conn.LocalAddr().(*net.UDPAddr).Port }
+
+func (d *fakeWriteDevice) requestCount() int {
+ d.mu.Lock()
+ defer d.mu.Unlock()
+ return d.gotReqs
+}
+
+func (d *fakeWriteDevice) serve() {
+ defer d.wg.Done()
+ buf := make([]byte, 4096)
+ for {
+ n, src, err := d.conn.ReadFromUDP(buf)
+ if err != nil {
+ return // socket closed
+ }
+ data := make([]byte, n)
+ copy(data, buf[:n])
+
+ invokeId, choice, ok := d.extractInvokeIdAndChoice(data)
+ if !ok {
+ d.log.Warn().Msg("fake write device: could not extract
invoke id; ignoring")
+ continue
+ }
+ d.mu.Lock()
+ d.gotReqs++
+ d.mu.Unlock()
+
+ replyInvokeId := invokeId
+ if d.wrongInvokeId {
+ replyInvokeId = invokeId + 1
+ }
+
+ var resp model.BVLC
+ switch {
+ case choice == model.BACnetConfirmedServiceChoice_READ_PROPERTY:
+ resp = buildReadAckFor(replyInvokeId)
+ case d.replyError:
+ resp = d.buildErrorReply(replyInvokeId, choice)
+ default:
+ resp = d.buildSimpleAck(replyInvokeId, choice)
+ }
+ theBytes, err := resp.Serialize()
+ if err != nil {
+ d.log.Error().Err(err).Msg("fake write device:
serialize response")
+ continue
+ }
+ if _, err := d.conn.WriteToUDP(theBytes, src); err != nil {
+ d.log.Error().Err(err).Msg("fake write device: write
response")
+ continue
+ }
+ d.log.Info().Uint8("invokeId", replyInvokeId).Int("bytes",
len(theBytes)).Stringer("dst", src).Msg("fake write device sent reply")
+ }
+}
+
+func (d *fakeWriteDevice) extractInvokeIdAndChoice(data []byte) (uint8,
model.BACnetConfirmedServiceChoice, bool) {
+ bvlc, err := model.BVLCParse[model.BVLC](d.ctx, data)
+ if err != nil {
+ d.log.Error().Err(err).Msg("fake write device: parse BVLC")
+ return 0, 0, false
+ }
+ npduRetriever, ok := bvlc.(interface{ GetNpdu() model.NPDU })
+ if !ok {
+ return 0, 0, false
+ }
+ apdu := npduRetriever.GetNpdu().GetApdu()
+ cr, ok := apdu.(model.APDUConfirmedRequest)
+ if !ok {
+ d.log.Warn().Msgf("fake write device: not a confirmed request:
%T", apdu)
+ return 0, 0, false
+ }
+ return cr.GetInvokeId(), cr.GetServiceRequest().GetServiceChoice(), true
+}
+
+func (d *fakeWriteDevice) buildSimpleAck(invokeId uint8, choice
model.BACnetConfirmedServiceChoice) model.BVLC {
+ apdu := model.NewAPDUSimpleAck(invokeId, choice)
+ return wrapAPDU(apdu, false)
+}
+
+// buildReadAckFor mirrors fakeBacnetDevice.buildReadPropertyAck: a
ReadProperty
+// ComplexAck for ANALOG_INPUT,1/PRESENT_VALUE echoing the invoke id.
+func buildReadAckFor(invokeId uint8) model.BVLC {
+ serviceAck := model.NewBACnetServiceAckReadProperty(
+ 0,
+ model.CreateBACnetContextTagObjectIdentifier(0,
uint16(model.BACnetObjectType_ANALOG_INPUT), 1),
+ model.CreateBACnetPropertyIdentifierTagged(1,
uint32(model.BACnetPropertyIdentifier_PRESENT_VALUE)),
+ nil,
+
constructedDataFromTag(model.CreateBACnetApplicationTagReal(23.5)),
+ )
+ apdu := model.NewAPDUComplexAck(false, false, invokeId, nil, nil,
serviceAck, nil, nil)
+ return wrapAPDU(apdu, false)
+}
+
+func (d *fakeWriteDevice) buildErrorReply(invokeId uint8, choice
model.BACnetConfirmedServiceChoice) model.BVLC {
+ base := buildErrorAPDU(model.ErrorClass_PROPERTY,
model.ErrorCode_WRITE_ACCESS_DENIED)
+ apdu := model.NewAPDUError(invokeId, choice, base.GetError())
+ return wrapAPDU(apdu, false)
+}
+
+func (d *fakeWriteDevice) stop() {
+ _ = d.conn.Close()
+ d.wg.Wait()
+}
+
+// writePresentValue issues a single WriteRequest for
ANALOG_VALUE,1/PRESENT_VALUE
+// against the given connection and returns the response code. The bool reports
+// whether the Execute channel resolved at all (false == timed out / hung).
+func writePresentValue(t *testing.T, conn plc4go.PlcConnection)
(apiModel.PlcResponseCode, bool) {
+ t.Helper()
+ wr, err := conn.WriteRequestBuilder().
+ AddTagAddress("pv", "ANALOG_VALUE,1/PRESENT_VALUE",
float32(75.0)).
+ Build()
+ require.NoError(t, err)
+ ctx, cancel := context.WithTimeout(t.Context(), 6*time.Second)
+ t.Cleanup(cancel)
+ select {
+ case <-ctx.Done():
+ return 0, false
+ case res := <-wr.Execute(ctx):
+ if res.GetErr() != nil {
+ t.Logf("write error: %v", res.GetErr())
+ return 0, false
+ }
+ return res.GetResponse().GetResponseCode("pv"), true
+ }
+}
+
+func newWriteTestConnection(t *testing.T, log zerolog.Logger, port int)
plc4go.PlcConnection {
+ t.Helper()
+ dm := plc4go.NewPlcDriverManager(options.WithCustomLogger(log))
+ dm.RegisterDriver(NewDriver(options.WithCustomLogger(log)))
+ apiTransports.RegisterUdpTransport(dm)
+
+ connStr :=
fmt.Sprintf("bacnet-ip:udp://127.0.0.1:%d?local-port=0&ApduTimeoutMs=3000",
port)
+ ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+ t.Cleanup(cancel)
+ conn, err := dm.GetConnection(ctx, connStr)
+ require.NoError(t, err)
+ t.Cleanup(func() { assert.NoError(t, conn.Close()) })
+ return conn
+}
+
+// TestNativeBacnetWrite_SimpleAckResolves is the core regression: a
WriteProperty
+// answered by a spec-correct SimpleAck must resolve the Execute channel with
OK.
+func TestNativeBacnetWrite_SimpleAckResolves(t *testing.T) {
+ log := testutils.ProduceTestingLogger(t)
+ device := startFakeWriteDevice(t, log)
+ t.Cleanup(device.stop)
+
+ conn := newWriteTestConnection(t, log, device.port())
+ code, ok := writePresentValue(t, conn)
+ t.Logf("device received %d request(s)", device.requestCount())
+ require.True(t, ok, "write did not complete (Execute never resolved —
hung on SimpleAck)")
+ assert.Equal(t, apiModel.PlcResponseCode_OK, code)
+}
+
+// TestNativeBacnetWrite_WrongInvokeIdDoesNotResolve confirms a SimpleAck
bearing
+// the wrong invoke id is NOT accepted (the request must hang / time out rather
+// than falsely report success).
+func TestNativeBacnetWrite_WrongInvokeIdDoesNotResolve(t *testing.T) {
+ log := testutils.ProduceTestingLogger(t)
+ device := startFakeWriteDevice(t, log)
+ device.wrongInvokeId = true
+ t.Cleanup(device.stop)
+
+ conn := newWriteTestConnection(t, log, device.port())
+ _, ok := writePresentValue(t, conn)
+ assert.False(t, ok, "a SimpleAck with the wrong invoke id must NOT
resolve the write")
+}
+
+// TestNativeBacnetWrite_AfterReads reproduces the real E2E ordering: the same
+// connection is used for several ReadProperty requests first (advancing the
+// shared invoke-id generator so the write lands on a non-zero invoke id, and
+// exercising the idle period between requests), then a WriteProperty is issued
+// on that same connection. The write must resolve with OK.
+func TestNativeBacnetWrite_AfterReads(t *testing.T) {
+ log := testutils.ProduceTestingLogger(t)
+ device := startFakeWriteDevice(t, log)
+ t.Cleanup(device.stop)
+
+ conn := newWriteTestConnection(t, log, device.port())
+
+ // Several reads first (mirrors discovery) so the write uses a later
invoke id.
+ for i := 0; i < 6; i++ {
+ code, val, ok := readPresentValue(t, conn)
+ require.True(t, ok, "read %d hung", i)
+ assert.Equal(t, apiModel.PlcResponseCode_OK, code)
+ assert.InDelta(t, 23.5, val, 0.001)
+ }
+
+ code, ok := writePresentValue(t, conn)
+ t.Logf("device received %d request(s)", device.requestCount())
+ require.True(t, ok, "write after reads did not complete (Execute never
resolved)")
+ assert.Equal(t, apiModel.PlcResponseCode_OK, code)
+}
+
+// TestNativeBacnetWrite_ErrorPduResolvesWithCode confirms a BACnet Error-PDU
for
+// the write resolves the request with the mapped error code.
+func TestNativeBacnetWrite_ErrorPduResolvesWithCode(t *testing.T) {
+ log := testutils.ProduceTestingLogger(t)
+ device := startFakeWriteDevice(t, log)
+ device.replyError = true
+ t.Cleanup(device.stop)
+
+ conn := newWriteTestConnection(t, log, device.port())
+ code, ok := writePresentValue(t, conn)
+ require.True(t, ok, "an Error-PDU must resolve the write (not hang)")
+ assert.Equal(t, apiModel.PlcResponseCode_ACCESS_DENIED, code)
+}
diff --git a/plc4go/spi/transports/udp/DelayedReply_test.go
b/plc4go/spi/transports/udp/DelayedReply_test.go
new file mode 100644
index 0000000000..1413ee07d5
--- /dev/null
+++ b/plc4go/spi/transports/udp/DelayedReply_test.go
@@ -0,0 +1,112 @@
+/*
+ * 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 udp
+
+import (
+ "context"
+ "net"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+)
+
+// ackBytes is the exact 9-byte BACnet/IP SimpleAck observed in the field
+// (81 0a 00 09 ...). Bytes [2],[3] are the BVLC length = 0x0009.
+var ackBytes = []byte{0x81, 0x0a, 0x00, 0x09, 0x01, 0x00, 0x20, 0x06, 0x0f}
+
+// pollOneLikeCodec drives the transport exactly like
bacnetip.MessageCodec.Receive:
+// GetNumBytesAvailableInBuffer (which arms a fresh short read deadline), then
+// PeekReadableBytes/Read with a *deadline-less* background context (the codec
+// polls with m.ctx, which is context.Background()-derived). It loops on the
+// same ~10ms cadence as the default codec ReceiveWork loop.
+func pollOneLikeCodec(t *testing.T, ti *TransportInstance, overall
time.Duration) ([]byte, bool) {
+ t.Helper()
+ deadline := time.Now().Add(overall)
+ for time.Now().Before(deadline) {
+ num, err := ti.GetNumBytesAvailableInBuffer()
+ if err == nil && num >= 4 {
+ hdr, err := ti.PeekReadableBytes(context.Background(),
4)
+ if err != nil {
+ time.Sleep(10 * time.Millisecond)
+ continue
+ }
+ size := uint32(uint16(hdr[2])<<8 | uint16(hdr[3]))
+ if num < size {
+ time.Sleep(10 * time.Millisecond)
+ continue
+ }
+ data, err := ti.Read(context.Background(), size)
+ if err != nil {
+ time.Sleep(10 * time.Millisecond)
+ continue
+ }
+ return data, true
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+ return nil, false
+}
+
+// TestTransportInstance_DelayedReplyAfterIdle reproduces the field scenario:
+// a request/reply works, the connection then sits idle well past the per-poll
+// read deadline, and a *delayed* reply arrives. It must still reach the codec.
+func TestTransportInstance_DelayedReplyAfterIdle(t *testing.T) {
+ for _, idle := range []time.Duration{0, 500 * time.Millisecond, 2 *
time.Second} {
+ t.Run(idle.String(), func(t *testing.T) {
+ peer, err := net.ListenUDP("udp4", &net.UDPAddr{IP:
net.IPv4(127, 0, 0, 1), Port: 0})
+ require.NoError(t, err)
+ t.Cleanup(func() { _ = peer.Close() })
+
+ ti := NewTransportInstance(nil,
peer.LocalAddr().(*net.UDPAddr), false, nil)
+ require.NoError(t, ti.Connect(t.Context()))
+ t.Cleanup(func() { _ = ti.Close() })
+
+ // First exchange: client writes so the peer learns the
client addr.
+ require.NoError(t, ti.Write(context.Background(),
ackBytes))
+ buf := make([]byte, 4096)
+ require.NoError(t,
peer.SetReadDeadline(time.Now().Add(2*time.Second)))
+ _, clientAddr, err := peer.ReadFromUDP(buf)
+ require.NoError(t, err)
+
+ // Immediate reply must arrive.
+ _, err = peer.WriteToUDP(ackBytes, clientAddr)
+ require.NoError(t, err)
+ _, ok := pollOneLikeCodec(t, ti, 2*time.Second)
+ require.True(t, ok, "immediate reply lost")
+
+ // Idle gap: the receive loop keeps polling (arming
10ms deadlines that
+ // keep expiring) with no traffic.
+ if idle > 0 {
+ stop := time.Now().Add(idle)
+ for time.Now().Before(stop) {
+ _, _ = ti.GetNumBytesAvailableInBuffer()
+ time.Sleep(10 * time.Millisecond)
+ }
+ }
+
+ // Delayed reply after the idle gap.
+ _, err = peer.WriteToUDP(ackBytes, clientAddr)
+ require.NoError(t, err)
+ _, ok = pollOneLikeCodec(t, ti, 2*time.Second)
+ require.True(t, ok, "delayed reply after %s idle gap
was lost", idle)
+ })
+ }
+}
