From: Anderson A. Giacomolli Date: Tue, 10 Nov 2026 23:13:27 +0000 (-0300) Subject: Keep interrupted Modbus work a context stop X-Git-Url: https://git.giacomolli.uk/onehouse/commitdiff_plain/d1910130620af1355128795e5dbcacab4f476950?ds=sidebyside;hp=ed48ff78adf6b195fc729e6c58772edc3a09def2 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 6983491..9d9cc42 100644 --- 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 6484b10..972f698 100644 --- 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 befb688..6e4979b 100644 --- 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()