Keep interrupted Modbus work a context stop

A deadline that only interrupts caller-cancelled transport work returns
the context stop to the runner. Joined independent failures remain
visible instead of being hidden by cancellation. Simulated-time tests
check blocked reads and writes through the runner, preserve an
independent joined failure, and enforce earlier caller deadlines.

Issue: 8fdc39a

diff --git a/doc/m31.rst b/doc/m31.rst
index 698349131607d3a039586ffd3e196f123e945e06..9d9cc42e69cb4171bf54cd2158709cfd2de6a64c 100644 (file)
--- a/doc/m31.rst
+++ b/doc/m31.rst
@@ -116,3 +116,9 @@ cancellation, duplicate replay, and failed feedback or result sends.
 They do not verify electrical simultaneity, contact timing, hardware
 interlocks, or physical pulse width. Local motor protection remains
 necessary when network commands control moving equipment.
+
+A caller cancellation interrupts blocked transport work through its
+deadline. A timeout caused only by that interruption is a context stop
+for the runner. A joined independent transport or cleanup failure stays
+a failure. Operation timeouts while the worker context is active remain
+device failures and enter reconnect recovery.

diff --git a/driver/m31/tcp_client.go b/driver/m31/tcp_client.go
index 6484b101928a4c8a96f8d3bc66256c38d32acff0..972f6984e519180badad7959ada2e1ae16d0bebd 100644 (file)
--- a/driver/m31/tcp_client.go
+++ b/driver/m31/tcp_client.go
@@ -8,6 +8,7 @@ import (
        "fmt"
        "io"
        "net"
+       "os"
        "time"
 )
 
@@ -204,6 +205,10 @@ func (c *tcpClient) exchange(ctx context.Context, function byte, data []byte) (r
                if !stop() {
                        <-interrupted
 
+                       if err == nil || deadlineOnly(err) {
+                               err = ctx.Err()
+                       }
+
                        err = errors.Join(err, interruptErr, ctx.Err())
                }
        }()
@@ -265,3 +270,28 @@ func operationDeadline(ctx context.Context, timeout time.Duration) time.Time {
 
        return deadline
 }
+
+func deadlineOnly(err error) bool {
+       joined, ok := err.(interface{ Unwrap() []error })
+       if ok {
+               found := false
+
+               for _, cause := range joined.Unwrap() {
+                       if cause != nil {
+                               found = true
+                               if !deadlineOnly(cause) {
+                                       return false
+                               }
+                       }
+               }
+
+               return found
+       }
+
+       wrapped, ok := err.(interface{ Unwrap() error })
+       if ok {
+               return deadlineOnly(wrapped.Unwrap())
+       }
+
+       return errors.Is(err, os.ErrDeadlineExceeded)
+}

diff --git a/driver/m31/tcp_client_test.go b/driver/m31/tcp_client_test.go
index befb688f85bf2da9d58e69e1081f7297bbe661ab..6e4979b80ac81d443505bda5f0ca41f5bb2c2e1a 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,149 @@ 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 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([]byte) (int, error) {
+       c.cancel()
+       <-c.interrupted
+
+       return 0, c.failure
+}
+
+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()