Poll M31 assemblies with verified reconnect setup

The M31 driver validates settings, verifies every output mode and pulse
width before online, and establishes an input baseline without events.
Later reads publish state and timed input transitions through the shared
publisher. Offline recovery uses bounded backoff without duplicate
publication mechanics. Publication and cleanup errors return to the
runner, and cancellation closes the owned connection.

Issue: 8fdc39a

diff --git a/README.rst b/README.rst
index 41146534bf3c4e71b56283bf69df5c6aab8059b7..69b14d2f4b2d28da587b04cdc598a197a00b5eb3 100644 (file)
--- a/README.rst
+++ b/README.rst
@@ -5,7 +5,7 @@
 OneHouse is one program that controls a whole house. Linux is supported.
 The command runs an empty driver list and stops on SIGINT or SIGTERM.
 ``doc/driver.rst`` describes the framework and ``doc/mqtt.rst`` defines
 OneHouse is one program that controls a whole house. Linux is supported.
 The command runs an empty driver list and stops on SIGINT or SIGTERM.
 ``doc/driver.rst`` describes the framework and ``doc/mqtt.rst`` defines

-its wire contract.
+its wire contract. ``doc/m31.rst`` describes the M31 integration.

 
 *************
  Development
 
 *************
  Development

diff --git a/doc/m31.rst b/doc/m31.rst
index a1b6eeaa93f9edded98abba95d8e65998e9bed3e..5b374efbf5edd366939a95e174fcfecccc8b016f 100644 (file)
--- a/doc/m31.rst
+++ b/doc/m31.rst
@@ -42,3 +42,33 @@ 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.
 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.

+
+*******************
+ Polling lifecycle
+*******************
+
+``New`` validates ``Settings`` and takes a shared publisher and a
+client. The caller runs both the publisher and the integration through
+``driver.Run``. The command keeps its explicit driver list empty and
+does not open hardware connections.
+
+Every connection writes and reads back all digital output modes. Pulse
+outputs also write and verify their widths before online. The first
+complete read establishes a baseline without events. A later input
+change sends retained state followed by a pressed or released event. All
+events from one input read share its completion time, before coil and
+analog reads. A steady output or analog change sends reported state.
+Pulse coils do not send transient feedback.
+
+A device error publishes offline and retries after one second, doubling
+to thirty seconds until a complete baseline succeeds. Device failures do
+not log inside the integration. Publication and cleanup failures return
+to the runner. Cancellation closes the client without a false offline
+publication. Refresh requests cached snapshots through the shared
+publisher without repeating events or physical work.
+
+Polling can miss an edge that starts and ends between reads. It does not
+expose input counters. A configured poll interval is a target, not a
+guaranteed edge latency. Tests use fake clients and simulated time for
+baseline, transitions, pulse verification, reconnect, refresh, shutdown,
+and error propagation. Physical timing remains unverified.

diff --git a/driver/m31/client.go b/driver/m31/client.go
index 140e13e76abbf07135295fdded541311083a5057..8c65d030354318389641ab392d27592b368d0267 100644 (file)
--- a/driver/m31/client.go
+++ b/driver/m31/client.go
@@ -5,8 +5,8 @@ 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
 // 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.
+// 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
 type Client interface {
        Connect(ctx context.Context, address string, unit byte) error
        Close() error

diff --git a/driver/m31/connection.go b/driver/m31/connection.go
new file mode 100644 (file)
index 0000000..2230f91
--- /dev/null
+++ b/driver/m31/connection.go
@@ -0,0 +1,215 @@
+package m31
+
+import (
+       "context"
+       "encoding/json"
+       "fmt"
+       "strconv"
+       "time"
+
+       "onehouse/protocol"
+)
+
+const (
+       pulseWidthBase = 0x30d4
+       outputModeBase = 0x32c8
+       maximumVoltage = 10000
+)
+
+type connection struct {
+       owner *Driver
+       state snapshot
+}
+
+func (c *connection) start(ctx context.Context) error {
+       err := c.configure(ctx)
+       if err != nil {
+               return fmt.Errorf("configure outputs: %w", err)
+       }
+
+       next, err := c.read(ctx)
+       if err != nil {
+               return fmt.Errorf("read baseline: %w", err)
+       }
+
+       err = c.owner.publisher.Availability(ctx, c.owner.deviceID, protocol.Metadata{}, protocol.AvailabilityPayload{Status: "online"})
+       if err != nil {
+               return fmt.Errorf("publish online: %w", err)
+       }
+
+       err = c.publish(ctx, next, true, nil)
+       if err != nil {
+               return fmt.Errorf("publish baseline: %w", err)
+       }
+
+       return nil
+}
+
+func (c *connection) configure(ctx context.Context) error {
+       for _, output := range c.owner.topology.outputs {
+               mode := uint16(0)
+               if output.pulse != 0 {
+                       err := c.configureRegister(ctx, pulseWidthBase+output.address, output.pulse)
+                       if err != nil {
+                               return fmt.Errorf("set pulse width: %w", err)
+                       }
+
+                       mode = 1
+               }
+
+               err := c.configureRegister(ctx, outputModeBase+output.address, mode)
+               if err != nil {
+                       return fmt.Errorf("set output mode: %w", err)
+               }
+       }
+
+       return nil
+}
+
+func (c *connection) configureRegister(ctx context.Context, address uint16, value uint16) error {
+       err := c.owner.client.WriteRegister(ctx, address, value)
+       if err != nil {
+               return &deviceError{err: fmt.Errorf("write configuration register: %w", err)}
+       }
+
+       values, err := c.owner.client.ReadRegisters(ctx, address, 1)
+       if err != nil {
+               return &deviceError{err: fmt.Errorf("read configuration register: %w", err)}
+       }
+
+       if len(values) != 1 || values[0] != value {
+               return &deviceError{err: fmt.Errorf("configuration register readback differs")}
+       }
+
+       return nil
+}
+
+func (c *connection) read(ctx context.Context) (snapshot, error) {
+       inputs, err := c.owner.client.ReadInputs(ctx, uint16(len(c.owner.topology.inputs)))
+       if err != nil {
+               return snapshot{}, &deviceError{err: fmt.Errorf("read inputs: %w", err)}
+       }
+
+       observedAt := time.Now().UTC()
+       outputs, err := c.owner.client.ReadCoils(ctx, uint16(len(c.owner.topology.outputs)))
+       if err != nil {
+               return snapshot{}, &deviceError{err: fmt.Errorf("read coils: %w", err)}
+       }
+
+       var analog []uint16
+
+       if len(c.owner.topology.analog) != 0 {
+               analog, err = c.owner.client.ReadRegisters(ctx, 0, uint16(len(c.owner.topology.analog)))
+               if err != nil {
+                       return snapshot{}, &deviceError{err: fmt.Errorf("read analog outputs: %w", err)}
+               }
+       }
+
+       if len(inputs) != len(c.owner.topology.inputs) || len(outputs) != len(c.owner.topology.outputs) || len(analog) != len(c.owner.topology.analog) {
+               return snapshot{}, &deviceError{err: fmt.Errorf("incomplete assembly read")}
+       }
+
+       err = validateVoltage(analog)
+       if err != nil {
+               return snapshot{}, &deviceError{err: fmt.Errorf("validate analog outputs: %w", err)}
+       }
+
+       return snapshot{
+               inputs:     inputs,
+               outputs:    outputs,
+               analog:     analog,
+               observedAt: observedAt,
+       }, nil
+}
+
+func (c *connection) publish(ctx context.Context, next snapshot, baseline bool, causes map[string]string) error {
+       for index, point := range c.owner.topology.inputs {
+               if !baseline && next.inputs[index] == c.state.inputs[index] {
+                       continue
+               }
+
+               value := strconv.FormatBool(next.inputs[index])
+               err := c.report(ctx, point.id, value, next.observedAt, "")
+               if err != nil {
+                       return fmt.Errorf("publish input state: %w", err)
+               }
+
+               if !baseline {
+                       action := `"released"`
+                       if next.inputs[index] {
+                               action = `"pressed"`
+                       }
+
+                       err = c.owner.publisher.Event(ctx, c.owner.deviceID, point.id, protocol.Metadata{SentAt: &next.observedAt}, protocol.EventPayload{
+                               "action": json.RawMessage(action),
+                               "value":  json.RawMessage(value),
+                       })
+                       if err != nil {
+                               return fmt.Errorf("publish input transition: %w", err)
+                       }
+               }
+       }
+
+       for index, point := range c.owner.topology.outputs {
+               if point.pulse != 0 || !baseline && next.outputs[index] == c.state.outputs[index] {
+                       continue
+               }
+
+               err := c.report(ctx, point.id, strconv.FormatBool(next.outputs[index]), next.observedAt, causes[point.id])
+               if err != nil {
+                       return fmt.Errorf("publish output state: %w", err)
+               }
+       }
+
+       for index, point := range c.owner.topology.analog {
+               if !baseline && next.analog[index] == c.state.analog[index] {
+                       continue
+               }
+
+               value := strconv.FormatFloat(float64(next.analog[index])/1000, 'f', -1, 64)
+               err := c.report(ctx, point.id, value, next.observedAt, causes[point.id])
+               if err != nil {
+                       return fmt.Errorf("publish analog state: %w", err)
+               }
+       }
+
+       c.state = next
+
+       return nil
+}
+
+func (c *connection) report(ctx context.Context, capability string, value string, observedAt time.Time, cause string) error {
+       err := c.owner.publisher.Reported(ctx, c.owner.deviceID, capability, protocol.Metadata{
+               SentAt:      &observedAt,
+               CausationID: cause,
+       }, protocol.ValuePayload{Value: json.RawMessage(value)})
+       if err != nil {
+               return fmt.Errorf("publish point: %w", err)
+       }
+
+       return nil
+}
+
+func (c *connection) poll(ctx context.Context) error {
+       next, err := c.read(ctx)
+       if err != nil {
+               return fmt.Errorf("read assembly state: %w", err)
+       }
+
+       err = c.publish(ctx, next, false, nil)
+       if err != nil {
+               return fmt.Errorf("publish assembly state: %w", err)
+       }
+
+       return nil
+}
+
+func validateVoltage(values []uint16) error {
+       for _, value := range values {
+               if value > maximumVoltage {
+                       return fmt.Errorf("analog output exceeds ten volts")
+               }
+       }
+
+       return nil
+}

diff --git a/driver/m31/connection_test.go b/driver/m31/connection_test.go
new file mode 100644 (file)
index 0000000..f2e67a6
--- /dev/null
+++ b/driver/m31/connection_test.go
@@ -0,0 +1,132 @@
+package m31
+
+import (
+       "encoding/json"
+       "errors"
+       "testing"
+       "testing/synctest"
+       "time"
+
+       "github.com/stretchr/testify/require"
+
+       "onehouse/protocol"
+)
+
+func TestConnectionBaselineAndInputEvents(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               c, client, transport := testConnection(t)
+               require.Empty(t, transport.ofKind(protocol.Event))
+
+               client.inputs[0], client.inputs[15] = true, true
+               client.inputDelay = time.Millisecond
+               observedAt := time.Now().Add(time.Millisecond).UTC()
+
+               require.NoError(t, c.poll(t.Context()))
+
+               events := transport.ofKind(protocol.Event)
+               require.Len(t, events, 2)
+
+               for _, event := range events {
+                       require.Equal(t, &observedAt, event.Metadata.SentAt)
+                       require.JSONEq(t, `{"action":"pressed","value":true}`, eventJSON(t, event.Payload))
+               }
+
+               require.NoError(t, c.poll(t.Context()))
+               require.Len(t, transport.ofKind(protocol.Event), 2)
+
+               client.inputs[0] = false
+
+               require.NoError(t, c.poll(t.Context()))
+
+               events = transport.ofKind(protocol.Event)
+               require.Len(t, events, 3)
+               require.JSONEq(t, `{"action":"released","value":false}`, eventJSON(t, events[2].Payload))
+
+               client.coils[2] = true
+               transport.messages = nil
+
+               require.NoError(t, c.poll(t.Context()))
+               require.Empty(t, transport.messages)
+       })
+}
+
+func TestConnectionRejectsConfigurationAndReadFailuresBeforeOnline(t *testing.T) {
+       testCases := []struct {
+               name   string
+               change func(*fakeClient)
+       }{
+               {name: "rejects pulse register write failure", change: func(c *fakeClient) {
+                       c.registerErr = errors.New("configuration failed")
+               }},
+               {name: "rejects configuration mismatch", change: func(c *fakeClient) {
+                       c.wrongConfig = true
+               }},
+               {name: "rejects input read failure", change: func(c *fakeClient) {
+                       c.inputErr = errors.New("input read failed")
+               }},
+               {name: "rejects coil read failure", change: func(c *fakeClient) {
+                       c.readErr = errors.New("coil read failed")
+               }},
+               {name: "rejects incomplete read", change: func(c *fakeClient) {
+                       c.shortRead = true
+               }},
+               {name: "rejects unsafe voltage", change: func(c *fakeClient) {
+                       c.analog[0] = 10001
+               }},
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       d, client, transport := testAssembly(t)
+                       tc.change(client)
+
+                       c := &connection{owner: d}
+                       err := c.start(t.Context())
+
+                       var failure *deviceError
+
+                       require.ErrorAs(t, err, &failure)
+                       require.Empty(t, transport.messages)
+               })
+       }
+}
+
+func TestConnectionReturnsPublicationFailures(t *testing.T) {
+       testCases := []struct {
+               name   string
+               kind   protocol.Kind
+               change func(*fakeClient)
+       }{
+               {name: "returns input report failure", kind: protocol.Reported, change: func(c *fakeClient) {
+                       c.inputs[0] = true
+               }},
+               {name: "returns transition failure", kind: protocol.Event, change: func(c *fakeClient) {
+                       c.inputs[0] = true
+               }},
+               {name: "returns output report failure", kind: protocol.Reported, change: func(c *fakeClient) {
+                       c.coils[0] = true
+               }},
+               {name: "returns voltage report failure", kind: protocol.Reported, change: func(c *fakeClient) {
+                       c.analog[0] = 5000
+               }},
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       c, client, transport := testConnection(t)
+                       failure := errors.New("publish failed")
+                       transport.failKind, transport.err = tc.kind, failure
+                       tc.change(client)
+                       require.ErrorIs(t, c.poll(t.Context()), failure)
+               })
+       }
+}
+
+func eventJSON(t *testing.T, value any) string {
+       t.Helper()
+
+       data, err := json.Marshal(value)
+       require.NoError(t, err)
+
+       return string(data)
+}

diff --git a/driver/m31/device_error.go b/driver/m31/device_error.go
new file mode 100644 (file)
index 0000000..1b60654
--- /dev/null
+++ b/driver/m31/device_error.go
@@ -0,0 +1,15 @@
+package m31
+
+// deviceError selects reconnect recovery for a physical transport or feedback
+// failure. Publication and cleanup failures return to the framework runner.
+type deviceError struct {
+       err error
+}
+
+func (f *deviceError) Error() string {
+       return f.err.Error()
+}
+
+func (f *deviceError) Unwrap() error {
+       return f.err
+}

diff --git a/driver/m31/driver.go b/driver/m31/driver.go
new file mode 100644 (file)
index 0000000..9201a59
--- /dev/null
+++ b/driver/m31/driver.go
@@ -0,0 +1,164 @@
+package m31
+
+import (
+       "context"
+       "errors"
+       "fmt"
+       "net"
+       "strconv"
+       "time"
+
+       "onehouse/driver"
+       "onehouse/protocol"
+)
+
+// Driver polls one assembly. Run has one owner and closes the supplied Client
+// on each attempt. Recoverable device failures publish offline and retry with
+// backoff. Publication or cleanup failures return to the framework runner. The
+// shared publisher must run alongside this driver. Refresh is safe during Run.
+// No operation starts a physical connection before Run.
+type Driver struct {
+       deviceID     string
+       address      string
+       unit         byte
+       pollInterval time.Duration
+       topology     topology
+       publisher    *driver.Publisher
+       client       Client
+}
+
+// New validates settings. The caller supplies valid publisher and client
+// dependencies. The driver owns the client until Run returns.
+func New(settings Settings, publisher *driver.Publisher, client Client) (*Driver, error) {
+       if !protocol.ValidID(settings.DeviceID) {
+               return nil, fmt.Errorf("device ID must be a canonical ULID")
+       }
+
+       host, port, err := net.SplitHostPort(settings.Address)
+       if err != nil || host == "" {
+               return nil, fmt.Errorf("address must contain a host and TCP port")
+       }
+
+       number, err := strconv.Atoi(port)
+       if err != nil || number < 1 || number > 65535 {
+               return nil, fmt.Errorf("TCP port is outside its range")
+       }
+
+       if settings.Unit == 0 || settings.Unit > 247 {
+               return nil, fmt.Errorf("unit must be from one through 247")
+       }
+
+       if settings.PollInterval < 10*time.Millisecond || settings.PollInterval > time.Minute {
+               return nil, fmt.Errorf("poll interval must be from ten milliseconds through one minute")
+       }
+
+       points, err := newTopology(settings.Modules)
+       if err != nil {
+               return nil, fmt.Errorf("validate topology: %w", err)
+       }
+
+       return &Driver{
+               deviceID:     settings.DeviceID,
+               address:      settings.Address,
+               unit:         settings.Unit,
+               pollInterval: settings.PollInterval,
+               topology:     points,
+               publisher:    publisher,
+               client:       client,
+       }, nil
+}
+
+// Run configures and verifies output modes before online or polling. A
+// successful first read establishes a baseline without input events. Device
+// failures retry after one second, doubling to thirty seconds until a complete
+// baseline succeeds.
+func (d *Driver) Run(ctx context.Context) error {
+       err := d.publisher.Descriptor(ctx, d.deviceID, protocol.Metadata{}, d.topology.descriptor())
+       if err != nil {
+               return fmt.Errorf("publish descriptor: %w", err)
+       }
+
+       delay := time.Second
+
+       for {
+               healthy, retry, err := d.attempt(ctx)
+               if !retry || ctx.Err() != nil {
+                       return fmt.Errorf("run assembly: %w", err)
+               }
+
+               err = d.publisher.Availability(ctx, d.deviceID, protocol.Metadata{}, protocol.AvailabilityPayload{Status: "offline"})
+               if err != nil {
+                       return fmt.Errorf("publish offline: %w", err)
+               }
+
+               if healthy {
+                       delay = time.Second
+               }
+
+               err = d.wait(ctx, delay)
+               if err != nil {
+                       return fmt.Errorf("wait for reconnect: %w", err)
+               }
+
+               delay = min(delay*2, 30*time.Second)
+       }
+}
+
+// Refresh requests retained snapshots through the shared publisher. It does not
+// repeat input events, physical commands, or pulse operations.
+func (d *Driver) Refresh() {
+       d.publisher.Refresh()
+}
+
+func (d *Driver) attempt(ctx context.Context) (healthy bool, retry bool, err error) {
+       defer func() {
+               var failure *deviceError
+
+               retry = errors.As(err, &failure)
+
+               closeErr := d.client.Close()
+               if closeErr != nil {
+                       retry = false
+                       err = fmt.Errorf("close assembly: %w", errors.Join(err, closeErr))
+               }
+       }()
+
+       err = d.client.Connect(ctx, d.address, d.unit)
+       if err != nil {
+               return false, false, &deviceError{err: fmt.Errorf("connect assembly: %w", err)}
+       }
+
+       connection := &connection{owner: d}
+       err = connection.start(ctx)
+       if err != nil {
+               return false, false, fmt.Errorf("initialize assembly: %w", err)
+       }
+
+       healthy = true
+       ticker := time.NewTicker(d.pollInterval)
+       defer ticker.Stop()
+
+       for {
+               select {
+               case <-ctx.Done():
+                       return healthy, false, ctx.Err()
+               case <-ticker.C:
+                       err = connection.poll(ctx)
+                       if err != nil {
+                               return healthy, false, fmt.Errorf("poll assembly: %w", err)
+                       }
+               }
+       }
+}
+
+func (d *Driver) wait(ctx context.Context, delay time.Duration) error {
+       timer := time.NewTimer(delay)
+       defer timer.Stop()
+
+       select {
+       case <-ctx.Done():
+               return ctx.Err()
+       case <-timer.C:
+               return nil
+       }
+}

diff --git a/driver/m31/driver_test.go b/driver/m31/driver_test.go
new file mode 100644 (file)
index 0000000..1ba6190
--- /dev/null
+++ b/driver/m31/driver_test.go
@@ -0,0 +1,234 @@
+package m31
+
+import (
+       "context"
+       "errors"
+       "testing"
+       "testing/synctest"
+       "time"
+
+       "github.com/stretchr/testify/require"
+
+       "onehouse/protocol"
+)
+
+func TestNewRejectsInvalidAssemblySettings(t *testing.T) {
+       testCases := []struct {
+               name   string
+               change func(*Settings)
+       }{
+               {name: "rejects noncanonical ID", change: func(s *Settings) {
+                       s.DeviceID = "bad"
+               }},
+               {name: "requires host and port", change: func(s *Settings) {
+                       s.Address = "example.com"
+               }},
+               {name: "requires a host", change: func(s *Settings) {
+                       s.Address = ":502"
+               }},
+               {name: "requires numeric port", change: func(s *Settings) {
+                       s.Address = "example.com:bad"
+               }},
+               {name: "rejects port zero", change: func(s *Settings) {
+                       s.Address = "example.com:0"
+               }},
+               {name: "rejects excess port", change: func(s *Settings) {
+                       s.Address = "example.com:65536"
+               }},
+               {name: "rejects broadcast unit", change: func(s *Settings) {
+                       s.Unit = 0
+               }},
+               {name: "rejects reserved unit", change: func(s *Settings) {
+                       s.Unit = 248
+               }},
+               {name: "rejects fast polling", change: func(s *Settings) {
+                       s.PollInterval = time.Millisecond
+               }},
+               {name: "rejects slow polling", change: func(s *Settings) {
+                       s.PollInterval = time.Minute + time.Nanosecond
+               }},
+               {name: "requires valid modules", change: func(s *Settings) {
+                       s.Modules = nil
+               }},
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       d, client, _ := testAssembly(t)
+                       settings := Settings{
+                               DeviceID:     testDeviceID,
+                               Address:      "example.com:502",
+                               Unit:         1,
+                               PollInterval: 50 * time.Millisecond,
+                               Modules:      []Module{{Model: HostModel}},
+                       }
+                       tc.change(&settings)
+
+                       _, err := New(settings, d.publisher, client)
+                       require.Error(t, err)
+               })
+       }
+}
+
+func TestDriverReconnectsWithConfigurationAndFreshBaseline(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               d, client, transport := testAssembly(t)
+               client.connectErrors = []error{errors.New("unavailable")}
+               client.inputs[1] = true
+               ctx, cancel := context.WithCancel(t.Context())
+               defer cancel()
+
+               done := make(chan error, 1)
+
+               go func() {
+                       done <- d.Run(ctx)
+               }()
+
+               synctest.Wait()
+               client.withLock(func(client *fakeClient) {
+                       require.Len(t, client.attempts, 1)
+                       require.Empty(t, client.registerWrites)
+               })
+               require.Empty(t, transport.ofKind(protocol.Event))
+
+               time.Sleep(time.Second)
+               synctest.Wait()
+               client.withLock(func(client *fakeClient) {
+                       require.Len(t, client.attempts, 2)
+                       require.Len(t, client.registerWrites, 33)
+                       require.EqualValues(t, 200, client.registers[pulseWidthBase+2])
+                       require.EqualValues(t, 1, client.registers[outputModeBase+2])
+               })
+               require.Empty(t, transport.ofKind(protocol.Event))
+
+               client.withLock(func(client *fakeClient) {
+                       client.inputs[1] = false
+               })
+
+               time.Sleep(50 * time.Millisecond)
+               synctest.Wait()
+               require.Len(t, transport.ofKind(protocol.Event), 1)
+
+               client.withLock(func(client *fakeClient) {
+                       client.inputErr = errors.New("connection lost")
+               })
+
+               time.Sleep(50 * time.Millisecond)
+               synctest.Wait()
+
+               availability, valid := transport.ofKind(protocol.Availability)[2].Payload.(protocol.AvailabilityPayload)
+               require.True(t, valid)
+               require.Equal(t, "offline", availability.Status)
+
+               client.withLock(func(client *fakeClient) {
+                       client.inputErr = nil
+                       client.inputs[1] = true
+               })
+
+               time.Sleep(time.Second)
+               synctest.Wait()
+               client.withLock(func(client *fakeClient) {
+                       require.Len(t, client.attempts, 3)
+                       require.Len(t, client.registerWrites, 66)
+               })
+               require.Len(t, transport.ofKind(protocol.Event), 1)
+               cancel()
+               require.ErrorIs(t, <-done, context.Canceled)
+               client.withLock(func(client *fakeClient) {
+                       require.Equal(t, 3, client.closes)
+               })
+       })
+}
+
+func TestDriverBoundsReconnectBackoff(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               d, client, _ := testAssembly(t)
+               for range 8 {
+                       client.withLock(func(client *fakeClient) {
+                               client.connectErrors = append(client.connectErrors, errors.New("offline"))
+                       })
+               }
+
+               ctx, cancel := context.WithCancel(t.Context())
+               defer cancel()
+
+               done := make(chan error, 1)
+
+               go func() {
+                       done <- d.Run(ctx)
+               }()
+
+               synctest.Wait()
+
+               for _, delay := range []time.Duration{1, 2, 4, 8, 16, 30, 30} {
+                       time.Sleep(delay * time.Second)
+                       synctest.Wait()
+               }
+
+               client.withLock(func(client *fakeClient) {
+                       require.Len(t, client.attempts, 8)
+                       require.Equal(t, 30*time.Second, client.attempts[7].Sub(client.attempts[6]))
+               })
+               cancel()
+               require.ErrorIs(t, <-done, context.Canceled)
+       })
+}
+
+func TestDriverReturnsPublicationAndCleanupFailures(t *testing.T) {
+       testCases := []struct {
+               name         string
+               kind         protocol.Kind
+               closeFailure bool
+       }{
+               {name: "returns descriptor failure", kind: protocol.Descriptor},
+               {name: "returns online failure", kind: protocol.Availability},
+               {name: "returns baseline failure", kind: protocol.Reported},
+               {name: "returns cleanup failure", closeFailure: true},
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       d, client, transport := testAssembly(t)
+                       failure := errors.New("independent failure")
+                       if tc.closeFailure {
+                               client.closeErr = failure
+                               client.connectErrors = []error{errors.New("device unavailable")}
+                       } else {
+                               transport.failKind, transport.err = tc.kind, failure
+                       }
+
+                       require.ErrorIs(t, d.Run(t.Context()), failure)
+                       require.LessOrEqual(t, len(client.attempts), 1)
+               })
+       }
+}
+
+func TestDriverRefreshDoesNotRepeatCommandsOrEvents(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               d, client, transport := testAssembly(t)
+               ctx, cancel := context.WithCancel(t.Context())
+               defer cancel()
+
+               done := make(chan error, 2)
+
+               go func() {
+                       done <- d.Run(ctx)
+               }()
+               go func() {
+                       done <- d.publisher.Run(ctx)
+               }()
+
+               synctest.Wait()
+
+               count := len(transport.all())
+
+               d.Refresh()
+               synctest.Wait()
+               require.Greater(t, len(transport.all()), count)
+               require.Empty(t, client.writes)
+               require.Empty(t, transport.ofKind(protocol.Event))
+               cancel()
+               require.ErrorIs(t, <-done, context.Canceled)
+               require.ErrorIs(t, <-done, context.Canceled)
+       })
+}

diff --git a/driver/m31/fake_helper_test.go b/driver/m31/fake_helper_test.go
new file mode 100644 (file)
index 0000000..c53dbac
--- /dev/null
+++ b/driver/m31/fake_helper_test.go
@@ -0,0 +1,274 @@
+package m31
+
+import (
+       "context"
+       "errors"
+       "fmt"
+       "slices"
+       "sync"
+       "testing"
+       "time"
+
+       "github.com/stretchr/testify/require"
+
+       "onehouse/driver"
+       "onehouse/protocol"
+)
+
+const testDeviceID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+
+type fakeClient struct {
+       mu             sync.Mutex
+       inputs         []bool
+       coils          []bool
+       analog         []uint16
+       registers      map[uint16]uint16
+       writes         []coilWrite
+       registerWrites []uint16
+       attempts       []time.Time
+       closes         int
+       coilReads      int
+       connectErrors  []error
+       inputErr       error
+       readErr        error
+       writeErr       error
+       registerErr    error
+       closeErr       error
+       wrongConfig    bool
+       shortRead      bool
+       mismatch       int
+       writeDelay     time.Duration
+       inputDelay     time.Duration
+}
+
+func (c *fakeClient) Connect(ctx context.Context, _ string, _ byte) error {
+       c.mu.Lock()
+       defer c.mu.Unlock()
+
+       c.attempts = append(c.attempts, time.Now())
+       if len(c.connectErrors) != 0 {
+               err := c.connectErrors[0]
+               c.connectErrors = c.connectErrors[1:]
+
+               return err
+       }
+
+       return ctx.Err()
+}
+
+func (c *fakeClient) Close() error {
+       c.mu.Lock()
+       defer c.mu.Unlock()
+
+       c.closes++
+
+       return c.closeErr
+}
+
+func (c *fakeClient) ReadInputs(ctx context.Context, _ uint16) ([]bool, error) {
+       c.mu.Lock()
+       defer c.mu.Unlock()
+
+       time.Sleep(c.inputDelay)
+
+       if c.inputErr != nil || ctx.Err() != nil {
+               return nil, errors.Join(c.inputErr, ctx.Err())
+       }
+
+       return slices.Clone(c.inputs), nil
+}
+
+func (c *fakeClient) ReadCoils(ctx context.Context, _ uint16) ([]bool, error) {
+       c.mu.Lock()
+       defer c.mu.Unlock()
+
+       c.coilReads++
+       if c.readErr != nil || ctx.Err() != nil {
+               return nil, errors.Join(c.readErr, ctx.Err())
+       }
+
+       if c.shortRead {
+               return nil, nil
+       }
+
+       return slices.Clone(c.coils), nil
+}
+
+func (c *fakeClient) ReadRegisters(ctx context.Context, address uint16, count uint16) ([]uint16, error) {
+       c.mu.Lock()
+       defer c.mu.Unlock()
+
+       if c.registerErr != nil || ctx.Err() != nil {
+               return nil, errors.Join(c.registerErr, ctx.Err())
+       }
+
+       if address >= pulseWidthBase {
+               value := c.registers[address]
+               if c.wrongConfig {
+                       value++
+               }
+
+               return []uint16{value}, nil
+       }
+
+       return slices.Clone(c.analog[:min(int(count), len(c.analog))]), nil
+}
+
+func (c *fakeClient) WriteCoil(ctx context.Context, address uint16, value bool) error {
+       return c.write(ctx, address, []bool{value}, false)
+}
+
+func (c *fakeClient) WriteCoils(ctx context.Context, address uint16, values []bool) error {
+       return c.write(ctx, address, values, true)
+}
+
+func (c *fakeClient) write(ctx context.Context, address uint16, values []bool, multiple bool) error {
+       c.mu.Lock()
+       defer c.mu.Unlock()
+
+       time.Sleep(c.writeDelay)
+
+       if c.writeErr != nil || ctx.Err() != nil {
+               return errors.Join(c.writeErr, ctx.Err())
+       }
+
+       c.writes = append(c.writes, coilWrite{
+               address:  address,
+               values:   slices.Clone(values),
+               multiple: multiple,
+       })
+
+       for offset, value := range values {
+               if int(address)+offset != c.mismatch {
+                       c.coils[int(address)+offset] = value
+               }
+       }
+
+       return nil
+}
+
+func (c *fakeClient) WriteRegister(ctx context.Context, address uint16, value uint16) error {
+       c.mu.Lock()
+       defer c.mu.Unlock()
+
+       if c.registerErr != nil || ctx.Err() != nil {
+               return errors.Join(c.registerErr, ctx.Err())
+       }
+
+       c.registerWrites = append(c.registerWrites, address)
+       if address >= pulseWidthBase {
+               c.registers[address] = value
+       } else if int(address) != c.mismatch {
+               c.analog[address] = value
+       }
+
+       return nil
+}
+
+func (c *fakeClient) withLock(fn func(*fakeClient)) {
+       c.mu.Lock()
+       defer c.mu.Unlock()
+
+       fn(c)
+}
+
+type coilWrite struct {
+       address  uint16
+       values   []bool
+       multiple bool
+}
+
+type fakeTransport struct {
+       mu       sync.Mutex
+       messages []protocol.Message[any]
+       failKind protocol.Kind
+       err      error
+}
+
+func (f *fakeTransport) Publish(ctx context.Context, publication protocol.Publication) error {
+       f.mu.Lock()
+       defer f.mu.Unlock()
+
+       message, err := protocol.Parse(publication.Topic, publication.Data, publication.Retain)
+       if err != nil {
+               return fmt.Errorf("parse fake publication: %w", err)
+       }
+
+       if message.Address.Kind == f.failKind && f.err != nil {
+               return f.err
+       }
+
+       if ctx.Err() != nil {
+               return ctx.Err()
+       }
+
+       f.messages = append(f.messages, message)
+
+       return nil
+}
+
+func (f *fakeTransport) ofKind(kind protocol.Kind) []protocol.Message[any] {
+       f.mu.Lock()
+       defer f.mu.Unlock()
+
+       var messages []protocol.Message[any]
+
+       for _, message := range f.messages {
+               if message.Address.Kind == kind {
+                       messages = append(messages, message)
+               }
+       }
+
+       return messages
+}
+
+func (f *fakeTransport) all() []protocol.Message[any] {
+       f.mu.Lock()
+       defer f.mu.Unlock()
+
+       return slices.Clone(f.messages)
+}
+
+func testAssembly(t *testing.T) (*Driver, *fakeClient, *fakeTransport) {
+       t.Helper()
+
+       transport := &fakeTransport{}
+       publisher := driver.NewPublisher(transport)
+       client := &fakeClient{
+               inputs:    make([]bool, 16),
+               coils:     make([]bool, 32),
+               analog:    make([]uint16, 8),
+               registers: make(map[uint16]uint16),
+               mismatch:  -1,
+       }
+       settings := Settings{
+               DeviceID:     testDeviceID,
+               Address:      "example.com:502",
+               Unit:         1,
+               PollInterval: 50 * time.Millisecond,
+               Modules: []Module{
+                       {Model: HostModel, PulseWidths: map[int]time.Duration{2: 200 * time.Millisecond}},
+                       {Model: OutputModel},
+                       {Model: AnalogModel},
+                       {Model: MixedModel},
+               },
+       }
+       d, err := New(settings, publisher, client)
+       require.NoError(t, err)
+
+       return d, client, transport
+}
+
+func testConnection(t *testing.T) (*connection, *fakeClient, *fakeTransport) {
+       t.Helper()
+
+       d, client, transport := testAssembly(t)
+       connection := &connection{owner: d}
+       require.NoError(t, connection.start(t.Context()))
+
+       transport.messages = nil
+       client.writes = nil
+       client.coilReads = 0
+
+       return connection, client, transport
+}

diff --git a/driver/m31/module.go b/driver/m31/module.go
index 173aebd6854b453aba5d27276c229261e4738d4e..55b6668c61aca49b6cdd208ef831ac80e286e4ee 100644 (file)
--- a/driver/m31/module.go
+++ b/driver/m31/module.go
@@ -22,8 +22,8 @@ const (
 
 // Module occupies its slice index in Settings.Modules. PulseWidths maps digital
 // output channels to hardware pulse widths. Widths must be whole milliseconds
 
 // Module occupies its slice index in Settings.Modules. PulseWidths maps digital
 // output channels to hardware pulse widths. Widths must be whole milliseconds

-// from 50 through 65535 milliseconds. Other digital outputs use level mode.
-// The constructor copies the topology and does not retain these maps.
+// from 50 through 65535 milliseconds. Other digital outputs use level mode. The
+// constructor copies the topology and does not retain these maps.

 type Module struct {
        Model       string
        PulseWidths map[int]time.Duration
 type Module struct {
        Model       string
        PulseWidths map[int]time.Duration

diff --git a/driver/m31/settings.go b/driver/m31/settings.go
new file mode 100644 (file)
index 0000000..b2692ab
--- /dev/null
+++ b/driver/m31/settings.go
@@ -0,0 +1,16 @@
+package m31
+
+import "time"
+
+// Settings defines one assembly. DeviceID is a canonical ULID. Address contains
+// a host and TCP port. Unit is from one through 247. PollInterval is from ten
+// milliseconds through one minute. Modules has one host followed by at most
+// fifteen expansions in physical order. The constructor retains no slice or map
+// from the caller.
+type Settings struct {
+       DeviceID     string
+       Address      string
+       Unit         byte
+       PollInterval time.Duration
+       Modules      []Module
+}

diff --git a/driver/m31/snapshot.go b/driver/m31/snapshot.go
new file mode 100644 (file)
index 0000000..8c29d12
--- /dev/null
+++ b/driver/m31/snapshot.go
@@ -0,0 +1,10 @@
+package m31
+
+import "time"
+
+type snapshot struct {
+       inputs     []bool
+       outputs    []bool
+       analog     []uint16
+       observedAt time.Time
+}