Preserve separate Modbus phase timeouts

[onehouse] / driver / m31 / tcp_client_test.go

diff --git a/driver/m31/tcp_client_test.go b/driver/m31/tcp_client_test.go
index befb688f85bf2da9d58e69e1081f7297bbe661ab..0067566b070cf21907b8b465479db9123d1e432a 100644 (file)
--- a/driver/m31/tcp_client_test.go
+++ b/driver/m31/tcp_client_test.go
@@ -4,13 +4,17 @@ import (
        "context"
        "encoding/binary"
        "errors"
+       "fmt"
        "io"
        "net"
+       "os"
        "testing"
        "testing/synctest"
        "time"
 
        "github.com/stretchr/testify/require"
+
+       "onehouse/driver"
 )
 
 func TestTCPClientEncodesAndChecksModbusOperations(t *testing.T) {
@@ -234,6 +238,206 @@ func TestTCPClientCancellationAndTimeouts(t *testing.T) {
        }
 }
 
+func TestTCPClientCancellationIsAnExpectedRunnerStop(t *testing.T) {
+       testCases := []struct {
+               name    string
+               reading bool
+       }{
+               {name: "stops a blocked write"},
+               {name: "stops a blocked read", reading: true},
+       }
+
+       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, peer.Close())
+                               })
+
+                               ctx, cancel := context.WithCancel(t.Context())
+                               defer cancel()
+
+                               ready := make(chan error, 1)
+                               if tc.reading {
+                                       go func() {
+                                               var request [12]byte
+
+                                               _, err := io.ReadFull(peer, request[:])
+                                               ready <- err
+                                       }()
+                               }
+
+                               done := make(chan error, 1)
+                               go func() {
+                                       done <- driver.Run(ctx, []driver.Driver{&requestDriver{client: c}}, nil)
+                               }()
+
+                               synctest.Wait()
+
+                               if tc.reading {
+                                       require.NoError(t, <-ready)
+                               }
+
+                               cancel()
+                               require.NoError(t, <-done)
+                       })
+               })
+       }
+}
+
+func TestTCPClientUsesCallerDeadlineAndStartsClosed(t *testing.T) {
+       client := NewTCPClient()
+       _, err := client.ReadCoils(t.Context(), 1)
+       require.Error(t, err)
+       require.NoError(t, client.Close())
+       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.WithTimeout(t.Context(), time.Millisecond)
+               defer cancel()
+
+               started := time.Now()
+               _, err := c.ReadCoils(ctx, 1)
+               require.Error(t, err)
+               require.Equal(t, time.Millisecond, time.Since(started))
+       })
+}
+
+func TestTCPClientReadTimeoutStartsAfterRequestWrite(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())
+               })
+
+               done := make(chan error, 1)
+               go func() {
+                       time.Sleep(500 * time.Millisecond)
+
+                       var request [12]byte
+
+                       _, err := io.ReadFull(peer, request[:])
+                       done <- err
+               }()
+
+               started := time.Now()
+               _, err := c.ReadCoils(t.Context(), 1)
+               require.Error(t, err)
+               require.NoError(t, <-done)
+               require.Equal(t, 2500*time.Millisecond, time.Since(started))
+       })
+}
+
+func TestTCPClientStopsAfterCancelledRequestWrite(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               conn, peer := net.Pipe()
+               t.Cleanup(func() {
+                       require.NoError(t, peer.Close())
+               })
+
+               ctx, cancel := context.WithCancel(t.Context())
+               defer cancel()
+
+               transport := &interruptConn{
+                       Conn:        conn,
+                       cancel:      cancel,
+                       interrupted: make(chan struct{}),
+               }
+               c := &tcpClient{conn: transport, unit: 1}
+               t.Cleanup(func() {
+                       require.NoError(t, c.Close())
+               })
+
+               _, err := c.ReadCoils(ctx, 1)
+               require.ErrorIs(t, err, context.Canceled)
+               require.NotErrorIs(t, err, os.ErrDeadlineExceeded)
+       })
+}
+
+func TestTCPClientCancellationPreservesIndependentFailure(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               conn, peer := net.Pipe()
+               t.Cleanup(func() {
+                       require.NoError(t, peer.Close())
+               })
+
+               ctx, cancel := context.WithCancel(t.Context())
+               defer cancel()
+
+               failure := errors.New("independent write failure")
+               transport := &interruptConn{
+                       Conn:        conn,
+                       cancel:      cancel,
+                       failure:     errors.Join(os.ErrDeadlineExceeded, failure),
+                       interrupted: make(chan struct{}),
+               }
+               c := &tcpClient{conn: transport, unit: 1}
+               t.Cleanup(func() {
+                       require.NoError(t, c.Close())
+               })
+
+               _, err := c.ReadCoils(ctx, 1)
+               require.ErrorIs(t, err, failure)
+               require.ErrorIs(t, err, context.Canceled)
+       })
+}
+
+type requestDriver struct {
+       client *tcpClient
+}
+
+func (d *requestDriver) Run(ctx context.Context) (err error) {
+       defer func() {
+               err = errors.Join(err, d.client.Close())
+       }()
+
+       _, err = d.client.ReadCoils(ctx, 1)
+
+       return err
+}
+
+func (d *requestDriver) Refresh() {
+}
+
+type interruptConn struct {
+       net.Conn
+       cancel      context.CancelFunc
+       failure     error
+       interrupted chan struct{}
+}
+
+func (c *interruptConn) Write(data []byte) (int, error) {
+       c.cancel()
+       <-c.interrupted
+
+       if c.failure != nil {
+               return 0, c.failure
+       }
+
+       return len(data), nil
+}
+
+func (c *interruptConn) SetDeadline(deadline time.Time) error {
+       defer close(c.interrupted)
+
+       err := c.Conn.SetDeadline(deadline)
+       if err != nil {
+               return fmt.Errorf("interrupt fake connection: %w", err)
+       }
+
+       return nil
+}
+
 func wireCall(t *testing.T, call func(*tcpClient) error, response func([]byte) []byte) ([]byte, error) {
        t.Helper()