Exchange bounded M31 Modbus TCP requests

A serialized client implements the six Modbus functions that the M31
assembly needs. Transaction and response checks reject malformed or
unrelated replies before they become physical feedback. Caller context
cancellation and independent transport deadlines bound every exchange.
In-memory stream tests exercise the wire without a device or broker.

Issue: 8fdc39a

diff --git a/doc/m31.rst b/doc/m31.rst
index b9b714f314b8bbe391bcba878adc6c9b021273ba..79f874e3cfb5027e259c903e9e74082319b627a6 100644 (file)
--- a/doc/m31.rst
+++ b/doc/m31.rst
@@ -27,3 +27,20 @@ capabilities are read-only.
 Tests verify topology and descriptor values without a physical module.
 They do not confirm physical expansion limits or relay timing under
 load. The conservative sixteen-module limit includes the host.
 Tests verify topology and descriptor values without a physical module.
 They do not confirm physical expansion limits or relay timing under
 load. The conservative sixteen-module limit includes the host.

+
+***************
+ Modbus client
+***************
+
+``NewTCPClient`` supplies a caller-owned, serialized connection through
+``Client``. Connect selects the configured unit and has a two-second
+timeout. A request has a one-second write deadline and a two-second
+response deadline, both measured from its start. Caller deadlines and
+cancellation can stop it earlier. Each response must match the
+transaction, unit, function, length, and write acknowledgement.
+Supported functions are 01, 02, 03, 05, 15, and 16. No external Modbus
+package is required.
+
+In-memory stream tests check request bytes, response validation,
+exceptions, truncation, cancellation, and timeouts. They do not verify a
+physical host or TCP network behavior.

diff --git a/driver/m31/client.go b/driver/m31/client.go
new file mode 100644 (file)
index 0000000..140e13e
--- /dev/null
+++ b/driver/m31/client.go
@@ -0,0 +1,19 @@
+package m31
+
+import "context"
+
+// Client owns one serialized Modbus connection. Connect replaces a closed
+// connection and selects a unit from one through 247. Reads must return the
+// exact requested count. Writes validate their acknowledgements. All operations
+// honor caller cancellation. The driver closes each connection before retry.
+// A caller must not use the client concurrently.
+type Client interface {
+       Connect(ctx context.Context, address string, unit byte) error
+       Close() error
+       ReadInputs(ctx context.Context, count uint16) ([]bool, error)
+       ReadCoils(ctx context.Context, count uint16) ([]bool, error)
+       ReadRegisters(ctx context.Context, address uint16, count uint16) ([]uint16, error)
+       WriteCoil(ctx context.Context, address uint16, value bool) error
+       WriteCoils(ctx context.Context, address uint16, values []bool) error
+       WriteRegister(ctx context.Context, address uint16, value uint16) error
+}

diff --git a/driver/m31/tcp_client.go b/driver/m31/tcp_client.go
new file mode 100644 (file)
index 0000000..6484b10
--- /dev/null
+++ b/driver/m31/tcp_client.go
@@ -0,0 +1,267 @@
+package m31
+
+import (
+       "bytes"
+       "context"
+       "encoding/binary"
+       "errors"
+       "fmt"
+       "io"
+       "net"
+       "time"
+)
+
+const (
+       connectTimeout = 2 * time.Second
+       writeTimeout   = time.Second
+       readTimeout    = 2 * time.Second
+)
+
+type tcpClient struct {
+       conn        net.Conn
+       unit        byte
+       transaction uint16
+}
+
+// NewTCPClient supplies Modbus TCP functions 01, 02, 03, 05, 15, and 16 without
+// opening a connection. Connect has a two-second timeout. Writes have a
+// one-second timeout, and reads have a two-second timeout. Earlier caller
+// deadlines apply to each operation. The caller owns Close.
+func NewTCPClient() Client {
+       return &tcpClient{}
+}
+
+func (c *tcpClient) Connect(ctx context.Context, address string, unit byte) error {
+       if c.conn != nil {
+               return fmt.Errorf("modbus connection is already open")
+       }
+
+       if unit == 0 || unit > 247 {
+               return fmt.Errorf("modbus unit must be from one through 247")
+       }
+
+       ctx, cancel := context.WithTimeout(ctx, connectTimeout)
+       defer cancel()
+
+       conn, err := (&net.Dialer{}).DialContext(ctx, "tcp", address)
+       if err != nil {
+               return fmt.Errorf("connect Modbus transport: %w", err)
+       }
+
+       c.conn, c.unit = conn, unit
+
+       return nil
+}
+
+func (c *tcpClient) Close() error {
+       if c.conn == nil {
+               return nil
+       }
+
+       err := c.conn.Close()
+       c.conn = nil
+       if err != nil {
+               return fmt.Errorf("close Modbus transport: %w", err)
+       }
+
+       return nil
+}
+
+func (c *tcpClient) ReadInputs(ctx context.Context, count uint16) ([]bool, error) {
+       return c.readBits(ctx, 2, count)
+}
+
+func (c *tcpClient) ReadCoils(ctx context.Context, count uint16) ([]bool, error) {
+       return c.readBits(ctx, 1, count)
+}
+
+func (c *tcpClient) readBits(ctx context.Context, function byte, count uint16) ([]bool, error) {
+       if count == 0 || count > 2000 {
+               return nil, fmt.Errorf("bit read count must be from one through 2000")
+       }
+
+       data, err := c.exchange(ctx, function, words(0, count))
+       if err != nil {
+               return nil, fmt.Errorf("read Modbus bits: %w", err)
+       }
+
+       size := (int(count) + 7) / 8
+       if len(data) != size+1 || int(data[0]) != size {
+               return nil, fmt.Errorf("modbus bit response has an invalid length")
+       }
+
+       values := make([]bool, count)
+
+       for index := range values {
+               values[index] = data[1+index/8]&(1<<uint(index%8)) != 0
+       }
+
+       return values, nil
+}
+
+func (c *tcpClient) ReadRegisters(ctx context.Context, address uint16, count uint16) ([]uint16, error) {
+       if count == 0 || count > 125 || uint32(address)+uint32(count) > 65536 {
+               return nil, fmt.Errorf("invalid register read range")
+       }
+
+       data, err := c.exchange(ctx, 3, words(address, count))
+       if err != nil {
+               return nil, fmt.Errorf("read Modbus registers: %w", err)
+       }
+
+       if len(data) != int(count)*2+1 || int(data[0]) != int(count)*2 {
+               return nil, fmt.Errorf("modbus register response has an invalid length")
+       }
+
+       values := make([]uint16, count)
+
+       for index := range values {
+               values[index] = binary.BigEndian.Uint16(data[1+2*index:])
+       }
+
+       return values, nil
+}
+
+func (c *tcpClient) WriteCoil(ctx context.Context, address uint16, value bool) error {
+       var encoded uint16
+
+       if value {
+               encoded = 0xff00
+       }
+
+       return c.write(ctx, 5, words(address, encoded), words(address, encoded))
+}
+
+func (c *tcpClient) WriteCoils(ctx context.Context, address uint16, values []bool) error {
+       if len(values) == 0 || len(values) > 1968 || uint32(address)+uint32(len(values)) > 65536 {
+               return fmt.Errorf("invalid coil write range")
+       }
+
+       data := append(words(address, uint16(len(values))), byte((len(values)+7)/8))
+       data = append(data, make([]byte, (len(values)+7)/8)...)
+
+       for index, value := range values {
+               if value {
+                       data[5+index/8] |= 1 << uint(index%8)
+               }
+       }
+
+       return c.write(ctx, 15, data, words(address, uint16(len(values))))
+}
+
+func (c *tcpClient) WriteRegister(ctx context.Context, address uint16, value uint16) error {
+       data := append(words(address, 1), 2)
+       data = append(data, words(value)...)
+
+       return c.write(ctx, 16, data, words(address, 1))
+}
+
+func (c *tcpClient) write(ctx context.Context, function byte, request []byte, expected []byte) error {
+       data, err := c.exchange(ctx, function, request)
+       if err != nil {
+               return fmt.Errorf("write Modbus value: %w", err)
+       }
+
+       if !bytes.Equal(data, expected) {
+               return fmt.Errorf("modbus write acknowledgement differs from request")
+       }
+
+       return nil
+}
+
+func (c *tcpClient) exchange(ctx context.Context, function byte, data []byte) (response []byte, err error) {
+       if ctx.Err() != nil {
+               return nil, ctx.Err()
+       }
+
+       if c.conn == nil {
+               return nil, fmt.Errorf("modbus transport is closed")
+       }
+
+       c.transaction++
+       request := append(words(c.transaction, 0, uint16(len(data)+2)), c.unit, function)
+       request = append(request, data...)
+       err = c.conn.SetWriteDeadline(operationDeadline(ctx, writeTimeout))
+       if err != nil {
+               return nil, fmt.Errorf("set Modbus write deadline: %w", err)
+       }
+
+       err = c.conn.SetReadDeadline(operationDeadline(ctx, readTimeout))
+       if err != nil {
+               return nil, fmt.Errorf("set Modbus read deadline: %w", err)
+       }
+
+       interrupted := make(chan struct{})
+
+       var interruptErr error
+
+       stop := context.AfterFunc(ctx, func() {
+               interruptErr = c.conn.SetDeadline(time.Now())
+
+               close(interrupted)
+       })
+       defer func() {
+               if !stop() {
+                       <-interrupted
+
+                       err = errors.Join(err, interruptErr, ctx.Err())
+               }
+       }()
+
+       count, err := c.conn.Write(request)
+       if err != nil {
+               return nil, fmt.Errorf("write Modbus request: %w", err)
+       }
+
+       if count != len(request) {
+               return nil, fmt.Errorf("write Modbus request: %w", io.ErrShortWrite)
+       }
+
+       var header [7]byte
+
+       _, err = io.ReadFull(c.conn, header[:])
+       if err != nil {
+               return nil, fmt.Errorf("read Modbus header: %w", err)
+       }
+
+       length := binary.BigEndian.Uint16(header[4:6])
+       if binary.BigEndian.Uint16(header[:2]) != c.transaction || binary.BigEndian.Uint16(header[2:4]) != 0 || header[6] != c.unit || length < 2 || length > 254 {
+               return nil, fmt.Errorf("invalid Modbus response header")
+       }
+
+       body := make([]byte, length-1)
+       _, err = io.ReadFull(c.conn, body)
+       if err != nil {
+               return nil, fmt.Errorf("read Modbus body: %w", err)
+       }
+
+       if body[0] == function|0x80 && len(body) == 2 {
+               return nil, fmt.Errorf("modbus exception %d", body[1])
+       }
+
+       if body[0] != function {
+               return nil, fmt.Errorf("invalid Modbus response function")
+       }
+
+       return body[1:], nil
+}
+
+func words(values ...uint16) []byte {
+       data := make([]byte, 2*len(values))
+
+       for index, value := range values {
+               binary.BigEndian.PutUint16(data[2*index:], value)
+       }
+
+       return data
+}
+
+func operationDeadline(ctx context.Context, timeout time.Duration) time.Time {
+       deadline := time.Now().Add(timeout)
+       caller, exists := ctx.Deadline()
+       if exists && caller.Before(deadline) {
+               return caller
+       }
+
+       return deadline
+}

diff --git a/driver/m31/tcp_client_test.go b/driver/m31/tcp_client_test.go
new file mode 100644 (file)
index 0000000..befb688
--- /dev/null
+++ b/driver/m31/tcp_client_test.go
@@ -0,0 +1,279 @@
+package m31
+
+import (
+       "context"
+       "encoding/binary"
+       "errors"
+       "io"
+       "net"
+       "testing"
+       "testing/synctest"
+       "time"
+
+       "github.com/stretchr/testify/require"
+)
+
+func TestTCPClientEncodesAndChecksModbusOperations(t *testing.T) {
+       testCases := []struct {
+               name    string
+               call    func(*tcpClient) error
+               request []byte
+               reply   []byte
+       }{
+               {
+                       name: "reads low bit first across byte boundaries",
+                       call: func(c *tcpClient) error {
+                               value, err := c.ReadInputs(t.Context(), 9)
+                               require.Equal(t, []bool{true, false, true, false, false, false, false, false, true}, value)
+
+                               return err
+                       },
+                       request: []byte{2, 0, 0, 0, 9},
+                       reply:   []byte{2, 2, 5, 1},
+               },
+               {
+                       name: "reads coil state",
+                       call: func(c *tcpClient) error {
+                               value, err := c.ReadCoils(t.Context(), 1)
+                               require.Equal(t, []bool{true}, value)
+
+                               return err
+                       },
+                       request: []byte{1, 0, 0, 0, 1},
+                       reply:   []byte{1, 1, 1},
+               },
+               {
+                       name: "reads full sixteen bit registers",
+                       call: func(c *tcpClient) error {
+                               value, err := c.ReadRegisters(t.Context(), 8, 2)
+                               require.Equal(t, []uint16{10000, 65535}, value)
+
+                               return err
+                       },
+                       request: []byte{3, 0, 8, 0, 2},
+                       reply:   []byte{3, 4, 0x27, 0x10, 0xff, 0xff},
+               },
+               {
+                       name: "writes true with function five",
+                       call: func(c *tcpClient) error {
+                               return c.WriteCoil(t.Context(), 7, true)
+                       },
+                       request: []byte{5, 0, 7, 0xff, 0},
+                       reply:   []byte{5, 0, 7, 0xff, 0},
+               },
+               {
+                       name: "writes false with function five",
+                       call: func(c *tcpClient) error {
+                               return c.WriteCoil(t.Context(), 7, false)
+                       },
+                       request: []byte{5, 0, 7, 0, 0},
+                       reply:   []byte{5, 0, 7, 0, 0},
+               },
+               {
+                       name: "writes a packed contiguous coil range",
+                       call: func(c *tcpClient) error {
+                               return c.WriteCoils(t.Context(), 6, []bool{true, false, true, false, false, false, false, false, true})
+                       },
+                       request: []byte{15, 0, 6, 0, 9, 2, 5, 1},
+                       reply:   []byte{15, 0, 6, 0, 9},
+               },
+               {
+                       name: "writes a holding register with function sixteen",
+                       call: func(c *tcpClient) error {
+                               return c.WriteRegister(t.Context(), 0x30d4, 200)
+                       },
+                       request: []byte{16, 0x30, 0xd4, 0, 1, 2, 0, 200},
+                       reply:   []byte{16, 0x30, 0xd4, 0, 1},
+               },
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       request, err := wireCall(t, tc.call, func(header []byte) []byte {
+                               return append(append(header[:4:4], words(uint16(len(tc.reply)+1))...), append([]byte{7}, tc.reply...)...)
+                       })
+                       require.NoError(t, err)
+                       require.Equal(t, append([]byte{0, 1, 0, 0, 0, byte(len(tc.request) + 1), 7}, tc.request...), request)
+               })
+       }
+}
+
+func TestTCPClientRejectsInvalidAndTruncatedResponses(t *testing.T) {
+       testCases := []struct {
+               name  string
+               reply []byte
+       }{
+               {name: "rejects another transaction", reply: []byte{0, 2, 0, 0, 0, 4, 7, 1, 1, 0}},
+               {name: "rejects another protocol", reply: []byte{0, 1, 0, 1, 0, 4, 7, 1, 1, 0}},
+               {name: "rejects another unit", reply: []byte{0, 1, 0, 0, 0, 4, 8, 1, 1, 0}},
+               {name: "rejects empty PDU", reply: []byte{0, 1, 0, 0, 0, 1, 7}},
+               {name: "rejects excessive PDU", reply: []byte{0, 1, 0, 0, 0, 255, 7}},
+               {name: "rejects truncated header", reply: []byte{0, 1}},
+               {name: "rejects truncated body", reply: []byte{0, 1, 0, 0, 0, 4, 7, 1}},
+               {name: "rejects wrong function", reply: []byte{0, 1, 0, 0, 0, 4, 7, 2, 1, 0}},
+               {name: "rejects exception", reply: []byte{0, 1, 0, 0, 0, 3, 7, 0x81, 2}},
+               {name: "rejects wrong byte count", reply: []byte{0, 1, 0, 0, 0, 4, 7, 1, 2, 0}},
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       _, err := wireCall(t, func(c *tcpClient) error {
+                               _, err := c.ReadCoils(t.Context(), 1)
+                               return err
+                       }, func([]byte) []byte {
+                               return tc.reply
+                       })
+                       require.Error(t, err)
+               })
+       }
+}
+
+func TestTCPClientRejectsInvalidRangesAndAcknowledgements(t *testing.T) {
+       c := &tcpClient{}
+       _, err := c.ReadInputs(t.Context(), 0)
+       require.Error(t, err)
+
+       _, err = c.ReadCoils(t.Context(), 2001)
+       require.Error(t, err)
+
+       _, err = c.ReadRegisters(t.Context(), 0, 0)
+       require.Error(t, err)
+
+       _, err = c.ReadRegisters(t.Context(), 0, 126)
+       require.Error(t, err)
+
+       _, err = c.ReadRegisters(t.Context(), 65535, 2)
+       require.Error(t, err)
+       require.Error(t, c.WriteCoils(t.Context(), 0, nil))
+       require.Error(t, c.WriteCoils(t.Context(), 65535, []bool{true, true}))
+       require.Error(t, c.WriteCoils(t.Context(), 0, make([]bool, 1969)))
+       require.Error(t, c.WriteCoil(t.Context(), 0, true))
+       require.NoError(t, c.Close())
+       require.Error(t, c.Connect(t.Context(), "", 0))
+       require.Error(t, c.Connect(t.Context(), "", 248))
+       require.Error(t, c.Connect(t.Context(), "", 1))
+
+       _, err = wireCall(t, func(c *tcpClient) error {
+               return c.WriteRegister(t.Context(), 0, 20)
+       }, func(header []byte) []byte {
+               return append(header[:7:7], 16, 0, 0, 0, 2)
+       })
+       require.Error(t, err)
+
+       _, err = wireCall(t, func(c *tcpClient) error {
+               _, readErr := c.ReadRegisters(t.Context(), 0, 1)
+               return readErr
+       }, func(header []byte) []byte {
+               return append(header[:7:7], 3, 3, 0, 0, 0)
+       })
+       require.Error(t, err)
+}
+
+func TestTCPClientCancellationAndTimeouts(t *testing.T) {
+       testCases := []struct {
+               name   string
+               cancel bool
+               write  bool
+       }{
+               {name: "cancels a blocked write", cancel: true, write: true},
+               {name: "cancels a blocked read", cancel: true},
+               {name: "limits a blocked write", write: true},
+               {name: "limits a blocked read"},
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       synctest.Test(t, func(t *testing.T) {
+                               conn, peer := net.Pipe()
+                               c := &tcpClient{conn: conn, unit: 1}
+                               t.Cleanup(func() {
+                                       require.NoError(t, c.Close())
+                                       require.NoError(t, peer.Close())
+                               })
+
+                               ctx, cancel := context.WithCancel(t.Context())
+                               defer cancel()
+
+                               result := make(chan error, 1)
+
+                               if !tc.write {
+                                       go func() {
+                                               var request [12]byte
+
+                                               _, err := io.ReadFull(peer, request[:])
+                                               result <- err
+                                       }()
+                               }
+
+                               done := make(chan error, 1)
+                               go func() {
+                                       _, err := c.ReadCoils(ctx, 1)
+                                       done <- err
+                               }()
+
+                               synctest.Wait()
+
+                               if !tc.write {
+                                       require.NoError(t, <-result)
+                               }
+
+                               if tc.cancel {
+                                       cancel()
+                               } else {
+                                       time.Sleep(2 * time.Second)
+                               }
+
+                               err := <-done
+                               require.Error(t, err)
+
+                               if tc.cancel {
+                                       require.ErrorIs(t, err, context.Canceled)
+                               }
+                       })
+               })
+       }
+}
+
+func wireCall(t *testing.T, call func(*tcpClient) error, response func([]byte) []byte) ([]byte, error) {
+       t.Helper()
+
+       conn, peer := net.Pipe()
+       c := &tcpClient{conn: conn, unit: 7}
+       t.Cleanup(func() {
+               require.NoError(t, c.Close())
+               require.NoError(t, peer.Close())
+       })
+
+       requests := make(chan []byte, 1)
+       done := make(chan error, 1)
+
+       go func() {
+               var header [7]byte
+
+               _, err := io.ReadFull(peer, header[:])
+               if err != nil {
+                       done <- err
+                       return
+               }
+
+               body := make([]byte, int(binary.BigEndian.Uint16(header[4:6]))-1)
+               _, err = io.ReadFull(peer, body)
+               requests <- append(header[:7:7], body...)
+               if err == nil {
+                       _, err = peer.Write(response(header[:]))
+               }
+
+               if errors.Is(err, io.ErrClosedPipe) {
+                       err = nil
+               }
+
+               err = errors.Join(err, peer.Close())
+               done <- err
+       }()
+
+       err := call(c)
+       require.NoError(t, c.Close())
+       require.NoError(t, <-done)
+
+       return <-requests, err
+}